~~~~~~~~~~~~~~~~~~~~~~~

Entry points for Kallithea hooking into Mercurial and Git.

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Aug 6, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import os

import sys

import mercurial.scmutil

import paste.deploy

import kallithea

import kallithea.config.application

from kallithea.lib import hooks, webutils

from kallithea.lib.utils2 import HookEnvironmentError, ascii_str, get_hook_environment, safe_bytes, safe_str

from kallithea.lib.vcs.backends.base import EmptyChangeset

from kallithea.lib.vcs.utils.helpers import get_scm_size

from kallithea.model import db

def repo_size(ui, repo, hooktype=None, **kwargs):

    """Show size of Mercurial repository.

    Called as Mercurial hook changegroup.repo_size after push.

    """

    size_hg, size_root = get_scm_size('.hg', safe_str(repo.root))

    last_cs = repo[len(repo) - 1]

    msg = ('Repository size .hg: %s Checkout: %s Total: %s\n'

           'Last revision is now r%s:%s\n') % (

        webutils.format_byte_size(size_hg),

        webutils.format_byte_size(size_root),

        webutils.format_byte_size(size_hg + size_root),

        last_cs.rev(),

        ascii_str(last_cs.hex())[:12],

    )

    ui.status(safe_bytes(msg))

def pull_action(ui, repo, **kwargs):

    """Logs user pull action

    Called as Mercurial hook outgoing.kallithea_pull_action.

    """

    hooks.log_pull_action()

def push_action(ui, repo, node, node_last, **kwargs):

    """

    Register that changes have been added to the repo - log the action *and* invalidate caches.

    Note: This hook is not only logging, but also the side effect invalidating

        real_path = os.path.normpath(os.path.realpath(path))

        if real_path != os.path.normpath(path):

            log.warning('Using normalized path %s instead of %s', real_path, path)

        return real_path

    def create_settings(self, repo_root_path):

        ui_config = [

            ('paths', '/', repo_root_path, True),

            #('phases', 'publish', 'false', False)

            ('extensions', 'largefiles', '', True),

            ('largefiles', 'usercache', os.path.join(repo_root_path, '.cache', 'largefiles'), True),

            ('extensions', 'hggit', '', False),

        ]

        for ui_section, ui_key, ui_value, ui_active in ui_config:

            ui_conf = db.Ui(

                ui_section=ui_section,

                ui_key=ui_key,

                ui_value=ui_value,

                ui_active=ui_active)

            self.sa.add(ui_conf)

    # force set push_ssl requirement to False, Kallithea handles that

    baseui.setconfig(b'web', b'push_ssl', False)

    baseui.setconfig(b'web', b'allow_push', b'*')

    # prevent interactive questions for ssh password / passphrase

    ssh = baseui.config(b'ui', b'ssh', default=b'ssh')

    baseui.setconfig(b'ui', b'ssh', b'%s -oBatchMode=yes -oIdentitiesOnly=yes' % ssh)

    # push / pull hooks

    baseui.setconfig(b'hooks', b'changegroup.kallithea_push_action', b'python:kallithea.bin.vcs_hooks.push_action')

    baseui.setconfig(b'hooks', b'outgoing.kallithea_pull_action', b'python:kallithea.bin.vcs_hooks.pull_action')

    if baseui.config(b'hooks', ascii_bytes(db.Ui.HOOK_REPO_SIZE)):  # ignore actual value

        baseui.setconfig(b'hooks', ascii_bytes(db.Ui.HOOK_REPO_SIZE), b'python:kallithea.bin.vcs_hooks.repo_size')

    if repo_path is not None:

        # Note: MercurialRepository / mercurial.localrepo.instance will do this too, so it will always be possible to override db settings or what is hardcoded above

        baseui.readconfig(repo_path)

    assert baseui.plain()  # set by hgcompat.monkey_do (invoked from import of vcs.backends.hg) to minimize potential impact of loading config files

    return baseui

def set_app_settings(config):

    """

    Updates app config with new settings from database

        response.mustcontain(no=['cd %s2' % base.TESTS_TMP_PATH])

    def test_add_existing_builtin_hook(self):

        self.log_user()

        response = self.app.post(base.url('admin_settings_hooks'),

                                params=dict(new_hook_ui_key='changegroup.update',

                                            new_hook_ui_value='attempted_new_value',

                                            _session_csrf_secret_token=self.session_csrf_secret_token()))

        self.checkSessionFlash(response, 'Builtin hooks are read-only')

        response = response.follow()

        response.mustcontain('changegroup.update')

    def test_index_search(self):

        self.log_user()

        response = self.app.get(base.url('admin_settings_search'))

    def test_index_system(self):

        self.log_user()

        response = self.app.get(base.url('admin_settings_system'))

    def test_ga_code_active(self):

        self.log_user()

        old_title = 'Kallithea'