pytest >= 4.6.6, < 5.5

pytest-sugar >= 0.9.2, < 0.10

pytest-benchmark >= 3.2.2, < 3.3

pytest-localserver >= 0.5.0, < 0.6

mock >= 3.0.0, < 4.1

Sphinx >= 1.8.0, < 3.1

WebTest >= 2.0.6, < 2.1

pyflakes == 2.2.0

    'DBSession': kallithea.model.meta.Session,

    # Configure App without an authentication backend.

    'auth_backend': None,

    # Use custom error page for these errors. By default, Turbogears2 does not add

    # 400 in this list.

    # Explicitly listing all is considered more robust than appending to defaults,

    # in light of possible future framework changes.

    'errorpage.status_codes': [400, 401, 403, 404],

    # Disable transaction manager -- currently Kallithea takes care of transactions itself

    'tm.enabled': False,

    # Set the default i18n source language so TG doesn't search beyond 'en' in Accept-Language.

    'i18n.lang': 'en',

})

# DebugBar, a debug toolbar for TurboGears2.

# (https://github.com/TurboGears/tgext.debugbar)

# To enable it, install 'tgext.debugbar' and 'kajiki', and run Kallithea with

# 'debug = true' (not in production!)

# See the Kallithea documentation for more information.

try:

    from tgext.debugbar import enable_debugbar

    assert kajiki

except ImportError:

    pass

else:

    base_config['renderers'].append('kajiki')

    enable_debugbar(base_config)

def setup_configuration(app):

    config = app.config

    if not kallithea.lib.locale.current_locale_is_valid():

        log.error("Terminating ...")

        sys.exit(1)

    # Mercurial sets encoding at module import time, so we have to monkey patch it

    hgencoding = config.get('hgencoding')

    if hgencoding:

        mercurial.encoding.encoding = hgencoding

    if config.get('ignore_alembic_revision', False):

        log.warning('database alembic revision checking is disabled')

    else:

        dbconf = config['sqlalchemy.url']

            ancestors = [ascii_str(hgrepo[ancestor].hex()) for ancestor in

                         hgrepo.revs(b"id(%s) & ::id(%s)", ascii_bytes(other_rev), ascii_bytes(org_rev))]

            if ancestors:

                log.debug("shortcut found: %s is already an ancestor of %s", other_rev, org_rev)

            else:

                log.debug("no shortcut found: %s is not an ancestor of %s", other_rev, org_rev)

                ancestors = [ascii_str(hgrepo[ancestor].hex()) for ancestor in

                             hgrepo.revs(b"heads(::id(%s) & ::id(%s))", ascii_bytes(org_rev), ascii_bytes(other_rev))] # FIXME: expensive!

            other_changesets = [

                other_repo.get_changeset(rev)

                for rev in hgrepo.revs(

                    b"ancestors(id(%s)) and not ancestors(id(%s)) and not id(%s)",

                    ascii_bytes(other_rev), ascii_bytes(org_rev), ascii_bytes(org_rev))

            ]

            org_changesets = [

                org_repo.get_changeset(ascii_str(hgrepo[rev].hex()))

                for rev in hgrepo.revs(

                    b"ancestors(id(%s)) and not ancestors(id(%s)) and not id(%s)",

                    ascii_bytes(org_rev), ascii_bytes(other_rev), ascii_bytes(other_rev))

            ]

        elif alias == 'git':

            if org_repo != other_repo:

                from dulwich.repo import Repo

                gitrepo = Repo(org_repo.path)

                SubprocessGitClient(thin_packs=False).fetch(other_repo.path, gitrepo)

                gitrepo_remote = Repo(other_repo.path)

                SubprocessGitClient(thin_packs=False).fetch(org_repo.path, gitrepo_remote)

                revs = [

                    ascii_str(x.commit.id)

                    for x in gitrepo_remote.get_walker(include=[ascii_bytes(other_rev)],

                                                       exclude=[ascii_bytes(org_rev)])

                ]

                other_changesets = [other_repo.get_changeset(rev) for rev in reversed(revs)]

                if other_changesets:

                    ancestors = [other_changesets[0].parents[0].raw_id]

                else:

                    # no changesets from other repo, ancestor is the other_rev

                    ancestors = [other_rev]

                gitrepo.close()

                gitrepo_remote.close()

            else:

                so = org_repo.run_git_command(

              "html:\n%s\n"

              % (' '.join(recipients), headers, subject, body, html_body))

    if mail_server:

        log.debug("Sending e-mail. " + logmsg)

    else:

        log.error("SMTP mail server not configured - cannot send e-mail.")

        log.warning(logmsg)

        return False

    try:

        m = SmtpMailer(envelope_from, user, passwd, mail_server, smtp_auth,

                       mail_port, ssl, tls, debug=debug)

        m.send(recipients, subject, body, html_body, headers=headers)

    except:

        log.error('Mail sending failed')

        log.error(traceback.format_exc())

        return False

    return True

@celerylib.task

@celerylib.dbsession

def create_repo(form_data, cur_user):

    from kallithea.model.repo import RepoModel

    DBS = celerylib.get_session()

    cur_user = User.guess_instance(cur_user)

    owner = cur_user

    repo_name = form_data['repo_name']

    repo_name_full = form_data['repo_name_full']

    repo_type = form_data['repo_type']

    description = form_data['repo_description']

    private = form_data['repo_private']

    clone_uri = form_data.get('clone_uri')

    repo_group = form_data['repo_group']

    landing_rev = form_data['repo_landing_rev']

    copy_fork_permissions = form_data.get('copy_permissions')

    copy_group_permissions = form_data.get('repo_copy_permissions')

    fork_of = form_data.get('fork_parent_id')

    state = form_data.get('repo_state', Repository.STATE_PENDING)

    # repo creation defaults, private and repo_type are filled in form

    defs = Setting.get_default_repo_settings(strip_prefix=True)

    enable_statistics = defs.get('repo_enable_statistics')

    enable_downloads = defs.get('repo_enable_downloads')

                confirm = getpass.getpass('Confirm password:')

                if password != confirm:

                    log.error('passwords mismatch')

                    return False

                if len(password) < 6:

                    log.error('password is to short use at least 6 characters')

                    return False

                return password

            if username is None:

                username = input('Specify admin username:')

            if password is None:

                password = get_password()

                if not password:

                    # second try

                    password = get_password()

                    if not password:

                        sys.exit()

            if email is None:

                email = input('Specify admin email:')

            self.create_user(username, password, email, True)

        else:

            log.info('creating admin and regular test users')

            self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,

                             TEST_USER_ADMIN_EMAIL, True)

            self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,

                             TEST_USER_REGULAR_EMAIL, False)

            self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS,

                             TEST_USER_REGULAR2_EMAIL, False)

    def create_auth_plugin_options(self, skip_existing=False):

        """

        Create default auth plugin settings, and make it active

        :param skip_existing:

        """

        for k, v, t in [('auth_plugins', 'kallithea.lib.auth_modules.auth_internal', 'list'),

                        ('auth_internal_enabled', 'True', 'bool')]:

            if skip_existing and Setting.get_by_name(k) is not None:

                log.debug('Skipping option %s', k)

                continue

            setting = Setting(k, v, t)

            self.sa.add(setting)

def email_or_none(author):

    """Try to match email part of VCS committer string with a local user.

    Return primary email of user, email part of the specified author name, or None."""

    if not author:

        return None

    email = user_attr_or_none(author, 'email')

    if email is not None:

        return email # always use user's main email address - not necessarily the one used to find user

    # extract email from the commit string

    email = author_email(author)

    if email:

        return email

    # No valid email, not a valid user in the system, none!

    return None

def person(author, show_attr="username"):

    """Find the user identified by 'author', return one of the users attributes,

    default to the username attribute, None if there is no user"""

    from kallithea.model.db import User

    # if author is already an instance use it for extraction

    if isinstance(author, User):

        return getattr(author, show_attr)

    value = user_attr_or_none(author, show_attr)

    if value is not None:

        return value

    # Still nothing?  Just pass back the author name if any, else the email

    return author_name(author) or email(author)

def person_by_id(id_, show_attr="username"):

    from kallithea.model.db import User

    # maybe it's an ID ?

    if str(id_).isdigit() or isinstance(id_, int):

        id_ = int(id_)

        user = User.get(id_)

        if user is not None:

            return getattr(user, show_attr)

    return id_

def boolicon(value):

    """Returns boolean value of a value, represented as small html image of true/false

    icons

    :param value: value

    """

    if value:

        return HTML.tag('i', class_="icon-ok")

    else:

        return HTML.tag('i', class_="icon-minus-circled")

def action_parser(user_log, feed=False, parse_cs=False):

    """

    """

    from tg import tmpl_context as c

    if not c.visual.use_gravatar:

        return ''

    src = gravatar_url(email_address, size * 2)

    if src:

        # here it makes sense to use style="width: ..." (instead of, say, a

        # stylesheet) because we using this to generate a high-res (retina) size

        html = ('<i class="icon-gravatar {cls}"'

                ' style="font-size: {size}px;background-size: {size}px;background-image: url(\'{src}\')"'

                '></i>').format(cls=cls, size=size, src=src)

    else:

        # if src is empty then there was no gravatar, so we use a font icon

        html = ("""<i class="icon-user {cls}" style="font-size: {size}px;"></i>"""

            .format(cls=cls, size=size))

    return literal(html)

def gravatar_url(email_address, size=30, default=''):

    from tg import tmpl_context as c

    if not c.visual.use_gravatar:

        return ""

    _def = 'anonymous@kallithea-scm.org'  # default gravatar

    email_address = email_address or _def

    if email_address == _def:

        return default

    parsed_url = urllib.parse.urlparse(url.current(qualified=True))

    url = (c.visual.gravatar_url or User.DEFAULT_GRAVATAR_URL) \

               .replace('{email}', email_address) \

               .replace('{md5email}', hashlib.md5(safe_bytes(email_address).lower()).hexdigest()) \

               .replace('{netloc}', parsed_url.netloc) \

               .replace('{scheme}', parsed_url.scheme) \

               .replace('{size}', str(size))

    return url

def changed_tooltip(nodes):

    """

    Generates a html string for changed nodes in changeset page.

    It limits the output to 30 entries

    :param nodes: LazyNodesGenerator

    """

    if nodes:

        pref = ': <br/> '

        suf = ''

        if len(nodes) > 30:

            suf = '<br/>' + _(' and %s more') % (len(nodes) - 30)

        return literal(pref + '<br/> '.join([x.path

                                             for x in nodes[:30]]) + suf)

    else:

        return ': ' + _('No files')

def fancy_file_stats(stats):

    """

    Displays a fancy two colored bar for number of added/deleted

    lines of code on file

    :param stats: two element list of added/deleted lines of code

    """

    a, d = stats['added'], stats['deleted']

    width = 100

    if stats['binary']:

        # binary mode

        lbl = ''

        bin_op = 1

        if BIN_FILENODE in stats['ops']:

            lbl = 'bin+'

        if NEW_FILENODE in stats['ops']:

            lbl += _('new file')

            bin_op = NEW_FILENODE

        elif MOD_FILENODE in stats['ops']:

            lbl += _('mod')

            bin_op = MOD_FILENODE

        elif DEL_FILENODE in stats['ops']:

            lbl += _('del')

            bin_op = DEL_FILENODE

        elif RENAMED_FILENODE in stats['ops']:

            lbl += _('rename')

            bin_op = RENAMED_FILENODE

     'full_name',

     'active',

     'password',

     'emails',

    """

    from kallithea import EXTENSIONS

    callback = getattr(EXTENSIONS, 'DELETE_USER_HOOK', None)

    if callable(callback):

        return callback(deleted_by=deleted_by, **user_dict)

    return 0

def _hook_environment(repo_path):

    """

    Create a light-weight environment for stand-alone scripts and return an UI and the

    db repository.

    Git hooks are executed as subprocess of Git while Kallithea is waiting, and

    they thus need enough info to be able to create an app environment and

    connect to the database.

    """

    import paste.deploy

    import kallithea.config.application

    extras = get_hook_environment()

    path_to_ini_file = extras['config']

    kallithea.CONFIG = paste.deploy.appconfig('config:' + path_to_ini_file)

    #logging.config.fileConfig(ini_file_path) # Note: we are in a different process - don't use configured logging

    kallithea.config.application.make_app(kallithea.CONFIG.global_conf, **kallithea.CONFIG.local_conf)

    # fix if it's not a bare repo

    if repo_path.endswith(os.sep + '.git'):

        repo_path = repo_path[:-5]

    repo = Repository.get_by_full_path(repo_path)

    if not repo:

        raise OSError('Repository %s not found in database' % repo_path)

    baseui = make_ui()

    return baseui, repo

def handle_git_pre_receive(repo_path, git_stdin_lines):

    """Called from Git pre-receive hook"""

    # Currently unused. TODO: remove?

            inputstream = FileWrapper(environ['wsgi.input'],

                                      req.content_length)

        else:

            inputstream = environ['wsgi.input']

        gitenv = dict(os.environ)

        # forget all configs

        gitenv['GIT_CONFIG_NOGLOBAL'] = '1'

        cmd = [_git_path, git_command[4:], '--stateless-rpc', self.content_path]

        log.debug('handling cmd %s', cmd)

        try:

            out = subprocessio.SubprocessIOChunker(

                cmd,

                inputstream=inputstream,

                env=gitenv,

                cwd=self.content_path,

            )

        except EnvironmentError as e:

            log.error(traceback.format_exc())

            raise exc.HTTPExpectationFailed()

        if git_command in ['git-receive-pack']:

            # updating refs manually after each push.

            # Needed for pre-1.7.0.4 git clients using regular HTTP mode.

            from kallithea.lib.vcs import get_repo

            repo = get_repo(self.content_path)

            if repo:

                update_server_info(repo._repo)

        resp = Response()

        resp.content_type = 'application/x-%s-result' % git_command.encode('utf-8')

        resp.charset = None

        resp.app_iter = out

        return resp

    def __call__(self, environ, start_response):

        req = Request(environ)

        _path = self._get_fixedpath(req.path_info)

        if _path.startswith('info/refs'):

            app = self.inforefs

        elif req.accept.acceptable_offers(self.valid_accepts):

            app = self.backend

        try:

            resp = app(req, environ)

        except exc.HTTPException as e:

            resp = e

            log.error(traceback.format_exc())

        except Exception as e:

            log.error(traceback.format_exc())

    def __init__(self, repo_root, repo_name):

        repo_location = os.path.join(repo_root, repo_name)

        if not os.path.isdir(repo_location):

            raise OSError(repo_location)

        self.content_path = repo_location

        self.repo_name = repo_name

        self.repo_location = repo_location

    def __call__(self, environ, start_response):

        content_path = self.content_path

        try:

            app = GitRepository(self.repo_name, content_path)

        except (AssertionError, OSError):

            content_path = os.path.join(content_path, '.git')

            if os.path.isdir(content_path):

                app = GitRepository(self.repo_name, content_path)

            else:

                return exc.HTTPNotFound()(environ, start_response)

        return app(environ, start_response)

def make_wsgi_app(repo_name, repo_root):

    app = GitDirectory(repo_root, repo_name)

    return GunzipFilter(LimitedInputFilter(app))

        'prefix': prefix, # undocumented, empty or starting with /

        'repo': repo_name,

        'repoid': str(repo_id),

        'system_user': system_user,

        'hostname': parsed_url.hostname,

    }

    url = re.sub('{([^{}]+)}', lambda m: args.get(m.group(1), m.group(0)), clone_uri_tmpl)

    # remove leading @ sign if it's present. Case of empty user

    url_obj = urlobject.URLObject(url)

    if not url_obj.username:

        url_obj = url_obj.with_username(None)

    return str(url_obj)

def get_changeset_safe(repo, rev):

    """

    Safe version of get_changeset if this changeset doesn't exists for a

    repo it returns a Dummy one instead

    :param repo:

    :param rev:

    """

    from kallithea.lib.vcs.exceptions import RepositoryError

    if not isinstance(repo, BaseRepository):

        raise Exception('You must pass an Repository '

                        'object as first argument got %s' % type(repo))

    try:

        cs = repo.get_changeset(rev)

    except (RepositoryError, LookupError):

        cs = EmptyChangeset(requested_revision=rev)

    return cs

def datetime_to_time(dt):

    if dt:

        return time.mktime(dt.timetuple())

def time_to_datetime(tm):

    if tm:

        if isinstance(tm, str):

            try:

                tm = float(tm)

            except ValueError:

                return

        return datetime.datetime.fromtimestamp(tm)

            result.append(key)

            continue

        except RepositoryError:

            # Wrong backend

            pass

        except VCSError:

            # No backend at all

            pass

    return result

def get_highlighted_code(name, code, type='terminal'):

    """

    If pygments are available on the system

    then returned output is colored. Otherwise

    unchanged content is returned.

    """

    import logging

    try:

        import pygments

        pygments

    except ImportError:

        return code

    from pygments import highlight

    from pygments.formatters import TerminalFormatter

    try:

        lexer = guess_lexer_for_filename(name, code)

        formatter = TerminalFormatter()

        content = highlight(code, lexer, formatter)

    except ClassNotFound:

        logging.debug("Couldn't guess Lexer, will not use pygments.")

        content = code

    return content

def parse_changesets(text):

    """

    Returns dictionary with *start*, *main* and *end* ids.

    Examples::

        >>> parse_changesets('aaabbb')

        {'start': None, 'main': 'aaabbb', 'end': None}

        >>> parse_changesets('aaabbb..cccddd')

        {'start': 'aaabbb', 'end': 'cccddd', 'main': None}

    """

    text = text.strip()

    @classmethod

    def get_auth_settings(cls):

        ret = cls.query() \

                .filter(cls.app_settings_name.startswith('auth_')).all()

        fd = {}

        for row in ret:

            fd[row.app_settings_name] = row.app_settings_value

        return fd

    @classmethod

    def get_default_repo_settings(cls, strip_prefix=False):

        ret = cls.query() \

                .filter(cls.app_settings_name.startswith('default_')).all()

        fd = {}

        for row in ret:

            key = row.app_settings_name

            if strip_prefix:

                key = remove_prefix(key, prefix='default_')

            fd.update({key: row.app_settings_value})

        return fd

    @classmethod

    def get_server_info(cls):

        import pkg_resources

        from kallithea.lib.utils import check_git_version

        mods = [(p.project_name, p.version) for p in pkg_resources.working_set]

        info = {

            'modules': sorted(mods, key=lambda k: k[0].lower()),

            'py_version': platform.python_version(),

            'platform': platform.platform(),

            'kallithea_version': kallithea.__version__,

            'git_version': str(check_git_version()),

            'git_path': kallithea.CONFIG.get('git_path')

        }

        return info

class Ui(Base, BaseDbModel):

    __tablename__ = 'ui'

    __table_args__ = (

        Index('ui_ui_section_ui_key_idx', 'ui_section', 'ui_key'),

        UniqueConstraint('ui_section', 'ui_key'),

        _table_args_default_dict,

    )

    HOOK_UPDATE = 'changegroup.update'

    HOOK_REPO_SIZE = 'changegroup.repo_size'

        if res is None or not res.active or res.is_default_user:

            return None

        return res

    @classmethod

    def get_by_email(cls, email, cache=False):

        q = cls.query().filter(sqlalchemy.func.lower(cls.email) == sqlalchemy.func.lower(email))

        ret = q.scalar()

        if ret is None:

            q = UserEmailMap.query()

            # try fetching in alternate email map

            q = q.filter(sqlalchemy.func.lower(UserEmailMap.email) == sqlalchemy.func.lower(email))

            q = q.options(joinedload(UserEmailMap.user))

            ret = getattr(q.scalar(), 'user', None)

        return ret

    @classmethod

    def get_from_cs_author(cls, author):

        """

        Tries to get User objects out of commit author string

        :param author:

        """

        # Valid email in the attribute passed, see if they're in the system

        _email = email(author)

        if _email:

            user = cls.get_by_email(_email)

            if user is not None:

                return user

        # Maybe we can match by username?

        _author = author_name(author)

        user = cls.get_by_username(_author, case_insensitive=True)

        if user is not None:

            return user

    def update_lastlogin(self):

        """Update user lastlogin"""

        self.last_login = datetime.datetime.now()

        log.debug('updated user %s lastlogin', self.username)

    @classmethod

    def get_first_admin(cls):

        user = User.query().filter(User.admin == True).first()

        if user is None:

            raise Exception('Missing administrative account!')

        return user

    def get_repo(self, repository):

        return Repository.guess_instance(repository)

    def get_by_repo_name(self, repo_name):

        repo = Repository.query() \

            .filter(Repository.repo_name == repo_name)

        return repo.scalar()

    def get_all_user_repos(self, user):

        """

        Gets all repositories that user have at least read access

        :param user:

        """

        from kallithea.lib.auth import AuthUser

        auth_user = AuthUser(dbuser=User.guess_instance(user))

        repos = [repo_name

            for repo_name, perm in auth_user.permissions['repositories'].items()

            if perm in ['repository.read', 'repository.write', 'repository.admin']

            ]

        return Repository.query().filter(Repository.repo_name.in_(repos))

    @classmethod

    def _render_datatable(cls, tmpl, *args, **kwargs):

        from tg.i18n import ugettext as _

        _tmpl_lookup = app_globals.mako_lookup

        template = _tmpl_lookup.get_template('data_table/_dt_elements.html')

        tmpl = template.get_def(tmpl)

        kwargs.update(dict(_=_, h=h, c=c, request=request))

        return tmpl.render_unicode(*args, **kwargs)

    def get_repos_as_dict(self, repos_list, repo_groups_list=None,

                          admin=False,

                          short_name=False):

        """Return repository list for use by DataTable.

        repos_list: list of repositories - but will be filtered for read permission.

        repo_groups_list: added at top of list without permission check.

        admin: return data for action column.

        """

        _render = self._render_datatable

        from kallithea.model.scm import ScmModel

        def repo_lnk(name, rtype, rstate, private, fork_of):

            return _render('repo_name', name, rtype, rstate, private, fork_of,

                           short_name=short_name)

        def following(repo_id, is_following):

            return _render('following', repo_id, is_following)

        def last_change(last_change):

            return _render("last_change", last_change)

        def rss_lnk(repo_name):

            return _render("rss", repo_name)

        def atom_lnk(repo_name):

            return _render("atom", repo_name)

        def last_rev(repo_name, cs_cache):

            return _render('revision', repo_name, cs_cache.get('revision'),

                           cs_cache.get('raw_id'), cs_cache.get('author'),

                           cs_cache.get('message'))

        def desc(desc):

                    # this group, if he is not super admin he get's admin

                    # permission set above

                    if perm.user != owner or owner.is_admin:

                        UserRepoGroupToPerm.create(perm.user, new_repo_group, perm.permission)

                for perm in group_perms:

                    UserGroupRepoGroupToPerm.create(perm.users_group, new_repo_group, perm.permission)

            else:

                self._create_default_perms(new_repo_group)

            if not just_db:

                # we need to flush here, in order to check if database won't

                # throw any exceptions, create filesystem dirs at the very end

                Session().flush()

                self._create_group(new_repo_group.group_name)

            return new_repo_group

        except Exception: