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

    '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():

                    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)])

                ]

                       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')

            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):

        """

    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

                ' 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):

    """

                                             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+'

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]

                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)

        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))

    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 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):

                .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

            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

        :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)

                # 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:

            log.error(traceback.format_exc())

            raise

    def _update_permissions(self, repo_group, perms_new=None,

                            perms_updates=None, recursive=None,

                            check_perms=True):

        from kallithea.model.repo import RepoModel

        if not perms_new:

            perms_new = []

        if not perms_updates:

            perms_updates = []

        def _set_perm_user(obj, user, perm):

            if isinstance(obj, RepoGroup):

                self.grant_user_permission(repo_group=obj, user=user, perm=perm)

            elif isinstance(obj, Repository):

                user = User.guess_instance(user)

    def get(self, user_id):

        user = User.query()

        return user.get(user_id)

    def get_user(self, user):

        return User.guess_instance(user)

    def create(self, form_data, cur_user=None):

        if not cur_user:

            cur_user = getattr(get_current_authuser(), 'username', None)

        _fd = form_data

        user_data = {

            'username': _fd['username'],

            'password': _fd['password'],

            'email': _fd['email'],

            'firstname': _fd['firstname'],

            'lastname': _fd['lastname'],

            'active': _fd['active'],

            'admin': False

        }

        # raises UserCreationError if it's not allowed

        check_allowed_create_user(user_data, cur_user)

        :param active:

        :param firstname:

        :param lastname:

        :param active:

        :param admin:

        :param extern_name:

        :param extern_type:

        :param cur_user:

        """

        if not cur_user:

            cur_user = getattr(get_current_authuser(), 'username', None)

        user_data = {

            'username': username, 'password': password,

            'email': email, 'firstname': firstname, 'lastname': lastname,

            'active': active, 'admin': admin

        }

        # raises UserCreationError if it's not allowed

        check_allowed_create_user(user_data, cur_user)

        log.debug('Checking for %s account in Kallithea database', username)

        user = User.get_by_username(username, case_insensitive=True)

        if user is None:

            log.debug('creating new user %s', username)

                Session().add(new_user)

                Session().flush() # make database assign new_user.user_id

            if not edit:

                log_create_user(new_user.get_dict(), cur_user)

            return new_user

        except (DatabaseError,):

            log.error(traceback.format_exc())

            raise

    def create_registration(self, form_data):

        from kallithea.model.notification import NotificationModel

        form_data['admin'] = False

        form_data['extern_type'] = User.DEFAULT_AUTH_TYPE

        form_data['extern_name'] = ''

        new_user = self.create(form_data)

        # notification to admins

        subject = _('New user registration')

        body = (

            'New user registration\n'

            '---------------------\n'

            '- Username: {user.username}\n'

            msg='\0'.join([session_id, str(user.user_id), user.email, str(timestamp)]).encode('utf-8'),

            digestmod=hashlib.sha1,

        ).hexdigest()

    def send_reset_password_email(self, data):

        """

        Sends email with a password reset token and link to the password

        reset confirmation page with all information (including the token)

        pre-filled. Also returns URL of that page, only without the token,

        allowing users to copy-paste or manually enter the token from the

        email.

        """

        from kallithea.lib.celerylib import tasks

        from kallithea.model.notification import EmailNotificationModel

        user_email = data['email']

        user = User.get_by_email(user_email)

        timestamp = int(time.time())

        if user is not None:

            if self.can_change_password(user):

                log.debug('password reset user %s found', user)

                token = self.get_reset_password_token(user,

                                                      timestamp,

                                                      h.session_csrf_secret_token())

                # URL must be fully qualified; but since the token is locked to

                # the current browser session, we must provide a URL with the

        if token_age > UserModel.password_reset_token_lifetime:

            log.debug('password reset token expired')

            return False

        expected_token = self.get_reset_password_token(user,

                                                       timestamp,

                                                       h.session_csrf_secret_token())

        log.debug('computed password reset token: %s', expected_token)

        log.debug('received password reset token: %s', token)

        return expected_token == token

    def reset_password(self, user_email, new_passwd):

        from kallithea.lib.celerylib import tasks

        user = User.get_by_email(user_email)

        if user is not None:

            if not self.can_change_password(user):

                raise Exception('trying to change password for external user')

            user.password = auth.get_crypt_password(new_passwd)

            Session().commit()

            log.info('change password for %s', user_email)

        if new_passwd is None:

            raise Exception('unable to set new password')

        tasks.send_email([user_email],

                 _('Password reset notification'),

    def test_Optional_object(self):

        from kallithea.controllers.api.api import Optional

        option1 = Optional(None)

        assert '<Optional:%s>' % None == repr(option1)

        assert option1() is None

        assert 1 == Optional.extract(Optional(1))

        assert 'trololo' == Optional.extract('trololo')

    def test_Optional_OAttr(self):

        option1 = Optional(OAttr('apiuser'))

        assert 'apiuser' == Optional.extract(option1)

    def test_OAttr_object(self):

        from kallithea.controllers.api.api import OAttr

        oattr1 = OAttr('apiuser')

        assert '<OptionalAttr:apiuser>' == repr(oattr1)

        assert oattr1() == oattr1

    def test_api_wrong_key(self):

        (dict(hours= -24), '1 day ago'),

        (dict(hours= -24 * 5), '5 days ago'),

        (dict(months= -1), '1 month ago'),

        (dict(months= -1, days= -2), '1 month and 2 days ago'),

        (dict(months= -1, days= -20), '1 month and 19 days ago'),

        (dict(years= -1, months= -1), '1 year and 1 month ago'),

        (dict(years= -1, months= -10), '1 year and 10 months ago'),

        (dict(years= -2, months= -4), '2 years and 4 months ago'),

        (dict(years= -2, months= -11), '2 years and 11 months ago'),

        (dict(years= -3, months= -2), '3 years and 2 months ago'),

    ])

    def test_age(self, age_args, expected):

        from kallithea.lib.utils2 import age

        with test_context(self.app):

            n = datetime.datetime(year=2012, month=5, day=17)

            delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

            assert age(n + delt(**age_args), now=n) == expected

    @base.parametrize('age_args,expected', [

        (dict(), 'just now'),

        (dict(seconds= -1), '1 second ago'),

        (dict(seconds= -60 * 2), '2 minutes ago'),

        (dict(hours= -1), '1 hour ago'),

        (dict(hours= -24), '1 day ago'),

        (dict(hours= -24 * 5), '5 days ago'),

        (dict(months= -1), '1 month ago'),

        (dict(months= -1, days= -2), '1 month ago'),

        (dict(months= -1, days= -20), '1 month ago'),

        (dict(years= -1, months= -1), '13 months ago'),

        (dict(years= -1, months= -10), '22 months ago'),

        (dict(years= -2, months= -4), '2 years ago'),

        (dict(years= -2, months= -11), '3 years ago'),

        (dict(years= -3, months= -2), '3 years ago'),

        (dict(years= -4, months= -8), '5 years ago'),

    ])

    def test_age_short(self, age_args, expected):

        from kallithea.lib.utils2 import age

        with test_context(self.app):

            n = datetime.datetime(year=2012, month=5, day=17)

            delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

            assert age(n + delt(**age_args), show_short_version=True, now=n) == expected

    @base.parametrize('age_args,expected', [

        (dict(), 'just now'),

        (dict(seconds=1), 'in 1 second'),

        (dict(seconds=60 * 2), 'in 2 minutes'),

        (dict(hours=1), 'in 1 hour'),

        (dict(hours=24), 'in 1 day'),

        (dict(hours=24 * 5), 'in 5 days'),

        (dict(months=1), 'in 1 month'),

        (dict(months=1, days=1), 'in 1 month and 1 day'),

        (dict(years=1, months=1), 'in 1 year and 1 month')

    ])

    def test_age_in_future(self, age_args, expected):

        from kallithea.lib.utils2 import age

        with test_context(self.app):

            n = datetime.datetime(year=2012, month=5, day=17)

            delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

            assert age(n + delt(**age_args), now=n) == expected

    def test_tag_extractor(self):

        sample = (

            "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"

            "[requires] [stale] [see<>=>] [see => http://example.com]"

            "[requires => url] [lang => python] [just a tag]"

            "[,d] [ => ULR ] [obsolete] [desc]]"

        )

        from kallithea.lib.utils2 import get_clone_url

        clone_url = get_clone_url(clone_uri_tmpl=clone_uri_tmpl, prefix_url='http://vps1:8000' + prefix,

                                  repo_name=repo_name, repo_id=23, username=username)

        assert clone_url == expected

    def _quick_url(self, text, tmpl="""<a class="changeset_hash" href="%s">%s</a>""", url_=None):

        """

        Changes `some text url[foo]` => `some text <a href="/">foo</a>

        :param text:

        """

        import re

        # quickly change expected url[] into a link

        url_pattern = re.compile(r'(?:url\[)(.+?)(?:\])')

        def url_func(match_obj):

            _url = match_obj.groups()[0]

            return tmpl % (url_ or '/repo_name/changeset/%s' % _url, _url)

        return url_pattern.sub(url_func, text)

    @base.parametrize('sample,expected', [

      ("",

       ""),

      ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",

        assert replaced == expected, 'url:%s, got:`%s` expected: `%s`' % (test, replaced, expected)

    @base.parametrize('canonical,test,expected', [

        ('http://www.example.org/', '/abc/xyz', 'http://www.example.org/abc/xyz'),

        ('http://www.example.org', '/abc/xyz', 'http://www.example.org/abc/xyz'),

        ('http://www.example.org', '/abc/xyz/', 'http://www.example.org/abc/xyz/'),

        ('http://www.example.org', 'abc/xyz/', 'http://www.example.org/abc/xyz/'),

        ('http://www.example.org', 'about', 'http://www.example.org/about-page'),

        ('http://www.example.org/repos/', 'abc/xyz/', 'http://www.example.org/repos/abc/xyz/'),

        ('http://www.example.org/kallithea/repos/', 'abc/xyz/', 'http://www.example.org/kallithea/repos/abc/xyz/'),

    ])

    def test_canonical_url(self, canonical, test, expected):

        from tg import request

        m = routes.Mapper()

        m.connect('about', '/about-page')

        url = routes.URLGenerator(m, {'HTTP_HOST': 'http_host.example.org'})

        config_mock = {

            'canonical_url': canonical,

        }

        with test_context(self.app):

            request.environ['routes.url'] = url

            with mock.patch('kallithea.CONFIG', config_mock):

                assert canonical_url(test) == expected

    @base.parametrize('canonical,expected', [

        ('http://www.example.org', 'www.example.org'),

        ('http://www.example.org/repos/', 'www.example.org'),

        ('http://www.example.org/kallithea/repos/', 'www.example.org'),

    ])

    def test_canonical_hostname(self, canonical, expected):

        from tg import request