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.

    """

    extras = get_hook_environment()

    path_to_ini_file = extras['config']

    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(config.global_conf, **config.local_conf)

    # fix if it's not a bare repo

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

        repo_path = repo_path[:-5]

    repo = db.Repository.get_by_full_path(repo_path)

    if not repo:

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

    return repo

def post_receive(repo_path, git_stdin_lines):

    """Called from Git post-receive hook.

    The returned value is used as hook exit code and must be 0.

    """

    try:

        repo = _git_hook_environment(repo_path)

    except HookEnvironmentError as e:

        sys.stderr.write("Skipping Kallithea Git post-receive hook %r.\nGit was apparently not invoked by Kallithea: %s\n" % (sys.argv[0], e))

        return 0

    # the post push hook should never use the cached instance

    scm_repo = repo.scm_instance_no_cache()

    rev_data = []

    for l in git_stdin_lines:

        old_rev, new_rev, ref = l.strip().split(' ')

        _ref_data = ref.split('/')

        if _ref_data[1] in ['tags', 'heads']:

            rev_data.append({'old_rev': old_rev,

                             'new_rev': new_rev,

                             'ref': ref,

                             'type': _ref_data[1],

                             'name': '/'.join(_ref_data[2:])})

        # example, when using uwsgi, it will point at the uwsgi program itself.

        # FIXME This may not work on Windows and may need a shell wrapper script.

        return (kallithea.CONFIG.get('git_hook_interpreter')

                or sys.executable

                or '/usr/bin/env python3')

    def install_git_hooks(self, repo, force=False):

        """

        Creates a kallithea hook inside a git repository

        :param repo: Instance of VCS repo

        :param force: Overwrite existing non-Kallithea hooks

        """

        hooks_path = os.path.join(repo.path, 'hooks')

        if not repo.bare:

            hooks_path = os.path.join(repo.path, '.git', 'hooks')

        if not os.path.isdir(hooks_path):

            os.makedirs(hooks_path)

        tmpl_post = b"#!%s\n" % safe_bytes(self._get_git_hook_interpreter())

        tmpl_post += pkg_resources.resource_string(

            'kallithea', os.path.join('templates', 'py', 'git_post_receive_hook.py')

        )

            hook_file = os.path.join(hooks_path, h_type)

            other_hook = False

            log.debug('Installing git hook %s in repo %s', h_type, repo.path)

            if os.path.islink(hook_file):

                log.debug("Found symlink hook at %s", hook_file)

                other_hook = True

            elif os.path.isfile(hook_file):

                log.debug('hook file %s exists, checking if it is from kallithea', hook_file)

                with open(hook_file, 'rb') as f:

                    data = f.read()

                    matches = re.search(br'^KALLITHEA_HOOK_VER\s*=\s*(.*)$', data, flags=re.MULTILINE)

                    if matches:

                        ver = safe_str(matches.group(1))

                        log.debug('Found Kallithea hook - it has KALLITHEA_HOOK_VER %s', ver)

                    else:

                        log.debug('Found non-Kallithea hook at %s', hook_file)

                        other_hook = True

            elif os.path.exists(hook_file):

                log.debug("Found hook that isn't a regular file at %s", hook_file)

                other_hook = True

            if other_hook and not force:

                log.warning('skipping overwriting hook file %s', hook_file)

                log.debug('writing hook file %s', hook_file)

                if other_hook:

                    backup_file = hook_file + '.bak'

                    log.warning('moving existing hook to %s', backup_file)

                    os.rename(hook_file, backup_file)

                try:

                    fh, fn = tempfile.mkstemp(prefix=hook_file + '.tmp.')

                    os.write(fh, tmpl.replace(b'_TMPL_', safe_bytes(kallithea.__version__)))

                    os.close(fh)

                    os.chmod(fn, 0o777 & ~umask)

                    os.rename(fn, hook_file)

                except (OSError, IOError) as e:

                    log.error('error writing hook %s: %s', hook_file, e)

def AvailableRepoGroupChoices(repo_group_perm_level, extras=()):

    """Return group_id,string tuples with choices for all the repo groups where

    the user has the necessary permissions.

    Top level is -1.

    """

    groups = db.RepoGroup.query().all()

    if HasPermissionAny('hg.admin')('available repo groups'):

        groups.append(None)

    else:

        groups = list(RepoGroupList(groups, perm_level=repo_group_perm_level))

        if HasPermissionAny('hg.create.repository')('available repo groups'):

            groups.append(None)

        for extra in extras:

            if not any(rg == extra for rg in groups):

                groups.append(extra)

    return db.RepoGroup.groups_choices(groups=groups)

    def test_similar_paths(self):

        cs = self.repo.get_changeset()

        paths = lambda *n: [x.path for x in n]

        assert paths(*cs.get_nodes('bot')) == ['bot/build', 'bot/templates', 'bot/__init__.py']

        assert paths(*cs.get_nodes('bot/build')) == ['bot/build/migrations', 'bot/build/static', 'bot/build/templates']

        assert paths(*cs.get_nodes('bot/build/static')) == ['bot/build/static/templates']

        # this get_nodes below causes troubles !

        assert paths(*cs.get_nodes('bot/build/static/templates')) == ['bot/build/static/templates/f.html', 'bot/build/static/templates/f1.html']

        assert paths(*cs.get_nodes('bot/build/templates')) == ['bot/build/templates/err.html', 'bot/build/templates/err2.html']

        assert paths(*cs.get_nodes('bot/templates/')) == ['bot/templates/404.html', 'bot/templates/500.html']

class TestGitHooks(object):

    """

    Tests related to hook functionality of Git repositories.

    """

    def setup_method(self):

        # For each run we want a fresh repo.

        self.repo_directory = get_new_dir("githookrepo")

        self.repo = GitRepository(self.repo_directory, create=True)

        # Create a dictionary where keys are hook names, and values are paths to

        # them in the non-bare repo. Deduplicates code in tests a bit.

        self.kallithea_hooks = {

        }

    def test_hooks_created_if_missing(self):

        """

        Tests if hooks are installed in repository if they are missing.

        """

        for hook, hook_path in self.kallithea_hooks.items():

            if os.path.exists(hook_path):

                os.remove(hook_path)

        ScmModel().install_git_hooks(self.repo)

    def test_kallithea_hooks_updated(self):

        """

        Tests if hooks are updated if they are Kallithea hooks already.

        """

        for hook, hook_path in self.kallithea_hooks.items():

            with open(hook_path, "w") as f:

                f.write("KALLITHEA_HOOK_VER=0.0.0\nJUST_BOGUS")

        ScmModel().install_git_hooks(self.repo)

    def test_custom_hooks_untouched(self):

        """

        Tests if hooks are left untouched if they are not Kallithea hooks.

        """

        for hook, hook_path in self.kallithea_hooks.items():

            with open(hook_path, "w") as f:

                f.write("#!/bin/bash\n#CUSTOM_HOOK")

        ScmModel().install_git_hooks(self.repo)

        for hook, hook_path in self.kallithea_hooks.items():

            with open(hook_path) as f:

                assert "CUSTOM_HOOK" in f.read()

    def test_custom_hooks_forced_update(self):

        """

        Tests if hooks are forcefully updated even though they are custom hooks.

        """

        for hook, hook_path in self.kallithea_hooks.items():

            with open(hook_path, "w") as f:

                f.write("#!/bin/bash\n#CUSTOM_HOOK")

        ScmModel().install_git_hooks(self.repo, force=True)