args = {

        'scheme': parsed_url.scheme,

        'user': urllib.parse.quote(username or ''),

        'netloc': parsed_url.netloc + prefix,  # like "hostname:port/prefix" (with optional ":port" and "/prefix")

        '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 short_ref_name(ref_type, ref_name):

    """Return short description of PR ref - revs will be truncated"""

    if ref_type == 'rev':

        return ref_name[:12]

    return ref_name

def link_to_ref(repo_name, ref_type, ref_name, rev=None):

    """

    Return full markup for a PR ref to changeset_home for a changeset.

    If ref_type is 'branch', it will link to changelog.

    ref_name is shortened if ref_type is 'rev'.

    if rev is specified, show it too, explicitly linking to that revision.

    """

    txt = short_ref_name(ref_type, ref_name)

    if ref_type == 'branch':

        u = webutils.url('changelog_home', repo_name=repo_name, branch=ref_name)

    else:

        u = webutils.url('changeset_home', repo_name=repo_name, revision=ref_name)

    l = webutils.link_to(repo_name + '#' + txt, u)

    if rev and ref_type != 'rev':

        l = webutils.literal('%s (%s)' % (l, webutils.link_to(rev[:12], webutils.url('changeset_home', repo_name=repo_name, revision=rev))))

    return l

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:

    """

    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)

class AttributeDict(dict):

    def __getattr__(self, attr):

        return self.get(attr, None)

    __setattr__ = dict.__setitem__

    __delattr__ = dict.__delitem__

def obfuscate_url_pw(engine):

    try:

        _url = sa_url.make_url(engine or '')

    except ArgumentError:

        return engine

    if _url.password:

        _url.password = 'XXXXX'

    return str(_url)

def get_hook_environment():

    """

    Get hook context by deserializing the global KALLITHEA_EXTRAS environment

    variable.

    Called early in Git out-of-process hooks to get .ini config path so the

    basic environment can be configured properly. Also used in all hooks to get

    information about the action that triggered it.

    """

    try:

        kallithea_extras = os.environ['KALLITHEA_EXTRAS']

    except KeyError:

        raise HookEnvironmentError("Environment variable KALLITHEA_EXTRAS not found")

    extras = json.loads(kallithea_extras)

    for k in ['username', 'repository', 'scm', 'action', 'ip', 'config']:

        try:

            extras[k]

        except KeyError:

            raise HookEnvironmentError('Missing key %s in KALLITHEA_EXTRAS %s' % (k, extras))

    return AttributeDict(extras)

def set_hook_environment(username, ip_addr, repo_name, repo_alias, action=None):

    """Prepare global context for running hooks by serializing data in the

    global KALLITHEA_EXTRAS environment variable.

    Most importantly, this allow Git hooks to do proper logging and updating of

    caches after pushes.

    Must always be called before anything with hooks are invoked.

    """

    extras = {

        'ip': ip_addr, # used in action_logger

        'username': username,

        'action': action or 'push_local', # used in process_pushed_raw_ids action_logger

        'repository': repo_name,

        'scm': repo_alias,

        'config': kallithea.CONFIG['__file__'], # used by git hook to read config

    }

    os.environ['KALLITHEA_EXTRAS'] = json.dumps(extras)

def get_current_authuser():

    """

    Gets kallithea user from threadlocal tmpl_context variable if it's

    defined, else returns None.

    """

    try:

        return getattr(tmpl_context, 'authuser', None)

    except TypeError:  # No object (name: context) has been registered for this thread

        return None

def urlreadable(s, _cleanstringsub=re.compile('[^-a-zA-Z0-9./]+').sub):

    return _cleanstringsub('_', s).rstrip('_')

def recursive_replace(str_, replace=' '):

    """

    Recursive replace of given sign to just one instance

    :param str_: given string

    :param replace: char to find and replace multiple instances

    Examples::

    >>> recursive_replace("Mighty---Mighty-Bo--sstones",'-')

    'Mighty-Mighty-Bo-sstones'

    """

    if str_.find(replace * 2) == -1:

        return str_

    else:

        str_ = str_.replace(replace * 2, replace)

        return recursive_replace(str_, replace)

def repo_name_slug(value):

    """

    Return slug of name of repository

    This function is called on each creation/modification

    of repository to prevent bad names in repo

    """

    slug = remove_formatting(value)

    slug = strip_tags(slug)

    for c in r"""`?=[]\;'"<>,/~!@#$%^&*()+{}|: """:

        slug = slug.replace(c, '-')

    slug = recursive_replace(slug, '-')

    slug = collapse(slug, '-')

    return slug

        dest_dir = _get_tmp_dir()

        clone_url = vt.repo_url_param(webserver, vt.repo_name)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(

            webserver, vt, dest_dir, clone_url='http://%s:%s/tmp' % (

                webserver.server_address[0], webserver.server_address[1]),

            ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'not found' in stderr

        elif vt.repo_type == 'hg':

            assert 'HTTP Error 404: Not Found' in stderr

    @parametrize_vcs_test

    def test_ip_restriction(self, webserver, vt):

        user_model = UserModel()

        try:

            # Add IP constraint that excludes the test context:

            user_model.add_extra_ip(base.TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

            meta.Session().commit()

            # IP permissions are cached, need to wait for the cache in the server process to expire

            time.sleep(1.5)

            clone_url = vt.repo_url_param(webserver, vt.repo_name)

            stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

            if vt.repo_type == 'git':

                # The message apparently changed in Git 1.8.3, so match it loosely.

                assert re.search(r'\b403\b', stderr) or 'abort: User test_admin from 127.0.0.127 cannot be authorized' in stderr

            elif vt.repo_type == 'hg':

                assert 'abort: HTTP Error 403: Forbidden' in stderr or 'remote: abort: User test_admin from 127.0.0.127 cannot be authorized' in stdout + stderr

        finally:

            # release IP restrictions

            for ip in db.UserIpMap.query():

                db.UserIpMap.delete(ip.ip_id)

            meta.Session().commit()

            # IP permissions are cached, need to wait for the cache in the server process to expire

            time.sleep(1.5)

        clone_url = vt.repo_url_param(webserver, vt.repo_name)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir())

        if vt.repo_type == 'git':

            assert 'Cloning into' in stdout + stderr

            assert stderr == '' or stdout == ''

        elif vt.repo_type == 'hg':

            assert 'requesting all changes' in stdout

            assert 'adding changesets' in stdout

            assert 'adding manifests' in stdout

            assert 'adding file changes' in stdout

            assert stderr == ''

    @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks

    def test_custom_hooks_preoutgoing(self, testhook_cleanup, webserver, testfork, vt):

        # set prechangegroup to failing hook (returns True)

        db.Ui.create_or_update_hook('preoutgoing.testhook', 'python:kallithea.tests.fixture.failing_test_hook')

        meta.Session().commit()

        # clone repo

        clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type], username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS)

        dest_dir = _get_tmp_dir()

        stdout, stderr = Command(base.TESTS_TMP_PATH) \

            .execute(vt.repo_type, 'clone', clone_url, dest_dir, ignoreReturnCode=True)

        if vt.repo_type == 'hg':

            assert 'preoutgoing.testhook hook failed' in stdout + stderr

        elif vt.repo_type == 'git':

            assert 'error: 406' in stderr

    @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks

    def test_custom_hooks_prechangegroup(self, testhook_cleanup, webserver, testfork, vt):

        # set prechangegroup to failing hook (returns exit code 1)

        db.Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.failing_test_hook')

        meta.Session().commit()

        # clone repo

        clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type], username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS)

        dest_dir = _get_tmp_dir()

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, clone_url,

                                             ignoreReturnCode=True)

        assert 'failing_test_hook failed' in stdout + stderr

        assert 'Traceback' not in stdout + stderr

        assert 'prechangegroup.testhook hook failed' in stdout + stderr

        # there are still outgoing changesets

        stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)

        assert stdout != ''

        # set prechangegroup hook to exception throwing method

        db.Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.exception_test_hook')

        meta.Session().commit()

        # re-try to push

        stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True)

        if vt is HgHttpVcsTest:

            # like with 'hg serve...' 'HTTP Error 500: INTERNAL SERVER ERROR' should be returned

            assert 'HTTP Error 500: INTERNAL SERVER ERROR' in stderr

        elif vt is HgSshVcsTest:

            assert 'remote: Exception: exception_test_hook threw an exception' in stdout

        # there are still outgoing changesets

        stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)

        assert stdout != ''

        # set prechangegroup hook to method that returns False

        db.Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.passing_test_hook')

        meta.Session().commit()

        # re-try to push

        stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True)

        assert 'passing_test_hook succeeded' in stdout + stderr

        assert 'Traceback' not in stdout + stderr

        assert 'prechangegroup.testhook hook failed' not in stdout + stderr

        # no more outgoing changesets

        stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)

        assert stdout == ''

        assert stderr == ''

    def test_add_submodule_git(self, webserver, testfork):

        dest_dir = _get_tmp_dir()

        clone_url = GitHttpVcsTest.repo_url_param(webserver, base.GIT_REPO)

        fork_url = GitHttpVcsTest.repo_url_param(webserver, testfork['git'])

        # add submodule

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute('git clone', fork_url, dest_dir)

        stdout, stderr = Command(dest_dir).execute('git submodule add', clone_url, 'testsubmodule')

        stdout, stderr = Command(dest_dir).execute('git commit -am "added testsubmodule pointing to', clone_url, '"', EMAIL=base.TEST_USER_ADMIN_EMAIL)

        stdout, stderr = Command(dest_dir).execute('git push', fork_url, 'master')

        # check for testsubmodule link in files page

        self.log_user()

        response = self.app.get(base.url(controller='files', action='index',

                                    repo_name=testfork['git'],

                                    revision='tip',

                                    f_path='/'))

        # check _repo_files_url that will be used to reload as AJAX

        response.mustcontain('var _repo_files_url = ("/%s/files/");' % testfork['git'])

        response.mustcontain('<a class="submodule-dir" href="%s" target="_blank"><i class="icon-file-submodule"></i><span>testsubmodule @ ' % clone_url)

        # check that following a submodule link actually works - and redirects

        response = self.app.get(base.url(controller='files', action='index',

                                    repo_name=testfork['git'],

                                    revision='tip',

                                    f_path='/testsubmodule'),

                                status=302)

        assert response.location == clone_url