return {}

        bt = OrderedDict()

        for bn, _heads, node, isclosed in sorted(self._repo.branchmap().iterbranches()):

            if isclosed:

                if closed:

                    bt[safe_str(bn)] = ascii_str(mercurial.node.hex(node))

            else:

                if normal:

                    bt[safe_str(bn)] = ascii_str(mercurial.node.hex(node))

        return bt

    @LazyProperty

    def tags(self):

        """

        Gets tags for this repository

        """

        return self._get_tags()

    def _get_tags(self):

        if self._empty:

            return {}

        return OrderedDict(sorted(

            ((safe_str(n), ascii_str(mercurial.node.hex(h))) for n, h in self._repo.tags().items()),

            reverse=True,

            key=lambda x: x[0],  # sort by name

        ))

    def tag(self, name, user, revision=None, message=None, date=None,

            **kwargs):

        """

        Creates and returns a tag for the given ``revision``.

        :param name: name for new tag

        :param user: full username, i.e.: "Joe Doe <joe.doe@example.com>"

        :param revision: changeset id for which new tag would be created

        :param message: message of the tag's commit

        :param date: date of tag's commit

        :raises TagAlreadyExistError: if tag with same name already exists

        """

        if name in self.tags:

            raise TagAlreadyExistError("Tag %s already exists" % name)

        changeset = self.get_changeset(revision)

        local = kwargs.setdefault('local', False)

        if message is None:

            message = "Added tag %s for changeset %s" % (name,

                changeset.short_id)

        if date is None:

            date = safe_bytes(datetime.datetime.now().strftime('%a, %d %b %Y %H:%M:%S'))

        try:

            mercurial.tags.tag(self._repo, safe_bytes(name), changeset._ctx.node(), safe_bytes(message), local, safe_bytes(user), date)

        except mercurial.error.Abort as e:

            raise RepositoryError(e.args[0])

        # Reinitialize tags

        self.tags = self._get_tags()

        tag_id = self.tags[name]

        return self.get_changeset(revision=tag_id)

    def remove_tag(self, name, user, message=None, date=None):

        """

        Removes tag with the given ``name``.

        :param name: name of the tag to be removed

        :param user: full username, i.e.: "Joe Doe <joe.doe@example.com>"

        :param message: message of the tag's removal commit

        :param date: date of tag's removal commit

        :raises TagDoesNotExistError: if tag with given name does not exists

        """

        if name not in self.tags:

            raise TagDoesNotExistError("Tag %s does not exist" % name)

        if message is None:

            message = "Removed tag %s" % name

        if date is None:

            date = safe_bytes(datetime.datetime.now().strftime('%a, %d %b %Y %H:%M:%S'))

        local = False

        try:

            mercurial.tags.tag(self._repo, safe_bytes(name), mercurial.commands.nullid, safe_bytes(message), local, safe_bytes(user), date)

            self.tags = self._get_tags()

        except mercurial.error.Abort as e:

            raise RepositoryError(e.args[0])

    @LazyProperty

    def bookmarks(self):

        """

        Gets bookmarks for this repository

        """

        return self._get_bookmarks()

    def _get_bookmarks(self):

        if self._empty:

            return {}

        return OrderedDict(sorted(

            ((safe_str(n), ascii_str(mercurial.node.hex(h))) for n, h in self._repo._bookmarks.items()),

            reverse=True,

            key=lambda x: x[0],  # sort by name

        ))

    def _get_all_revisions(self):

        return [ascii_str(self._repo[x].hex()) for x in self._repo.filtered(b'visible').changelog.revs()]

    def get_diff(self, rev1, rev2, path='', ignore_whitespace=False,

                  context=3):

        """

        Returns (git like) *diff*, as plain text. Shows changes introduced by

        ``rev2`` since ``rev1``.

        :param rev1: Entry point from which diff is shown. Can be

          ``self.EMPTY_CHANGESET`` - in this case, patch showing all

          the changes since empty state of the repository until ``rev2``

        :param rev2: Until which revision changes should be shown.

        :param ignore_whitespace: If set to ``True``, would not show whitespace

          changes. Defaults to ``False``.

        :param context: How many lines before/after changed lines should be

          shown. Defaults to ``3``. If negative value is passed-in, it will be

          set to ``0`` instead.

        """

        # Negative context values make no sense, and will result in

        # errors. Ensure this does not happen.

        if context < 0:

            context = 0

        if hasattr(rev1, 'raw_id'):

            rev1 = getattr(rev1, 'raw_id')

        if hasattr(rev2, 'raw_id'):

            rev2 = getattr(rev2, 'raw_id')

        # Check if given revisions are present at repository (may raise

        # ChangesetDoesNotExistError)

        if rev1 != self.EMPTY_CHANGESET:

            self.get_changeset(rev1)

        self.get_changeset(rev2)

        if path:

            file_filter = mercurial.match.exact(path)

        else:

            file_filter = None

        return b''.join(mercurial.patch.diff(self._repo, rev1, rev2, match=file_filter,

                          opts=mercurial.mdiff.diffopts(git=True,

                                        showfunc=True,

                                        ignorews=ignore_whitespace,

                                        context=context)))

    @classmethod

    def _check_url(cls, url, repoui=None):

        """

        Function will check given url and try to verify if it's a valid

        link. Sometimes it may happened that mercurial will issue basic

        auth request that can cause whole API to hang when used from python

        or other external calls.

        On failures it'll raise urllib2.HTTPError, exception is also thrown

        when the return code is non 200

        """

        # check first if it's not an local url

        url = safe_bytes(url)

        if os.path.isdir(url) or url.startswith(b'file:'):

            return True

        if url.startswith(b'ssh:'):

            # in case of invalid uri or authentication issues, sshpeer will

            # throw an exception.

            mercurial.sshpeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')

            return True

        url_prefix = None

        if b'+' in url[:url.find(b'://')]:

            url_prefix, url = url.split(b'+', 1)

        url_obj = mercurial.util.url(url)

        test_uri, handlers = get_urllib_request_handlers(url_obj)

        url_obj.passwd = b'*****'

        cleaned_uri = str(url_obj)

        o = urllib.request.build_opener(*handlers)

        o.addheaders = [('Content-Type', 'application/mercurial-0.1'),

                        ('Accept', 'application/mercurial-0.1')]

        req = urllib.request.Request(

            "%s?%s" % (

                urllib.parse.urlencode({

                    'cmd': 'between',

                    'pairs': "%s-%s" % ('0' * 40, '0' * 40),

                })

            ))

        try:

            resp = o.open(req)

            if resp.code != 200:

                raise Exception('Return Code is not 200')

        except Exception as e:

            # means it cannot be cloned

            raise urllib.error.URLError("[%s] org_exc: %s" % (cleaned_uri, e))

        if not url_prefix: # skip svn+http://... (and git+... too)

            # now check if it's a proper hg repo

            try:

                mercurial.httppeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')

            except Exception as e:

                raise urllib.error.URLError(

                    "url [%s] does not look like an hg repo org_exc: %s"

                    % (cleaned_uri, e))

        return True

    def _get_repo(self, create, src_url=None, update_after_clone=False):

        """

        Function will check for mercurial repository in given path and return

        a localrepo object. If there is no repository in that path it will

        raise an exception unless ``create`` parameter is set to True - in

        that case repository would be created and returned.

        If ``src_url`` is given, would try to clone repository from the

        location at given clone_point. Additionally it'll make update to

        working copy accordingly to ``update_after_clone`` flag

        """

        try:

            if src_url:

                url = safe_bytes(self._get_url(src_url))

                opts = {}

                if not update_after_clone:

                    opts.update({'noupdate': True})

                MercurialRepository._check_url(url, self.baseui)

                mercurial.commands.clone(self.baseui, url, safe_bytes(self.path), **opts)

                # Don't try to create if we've already cloned repo

                create = False

            return mercurial.localrepo.instance(self.baseui, safe_bytes(self.path), create=create)

        except (mercurial.error.Abort, mercurial.error.RepoError) as err:

            if create:

                msg = "Cannot create repository at %s. Original error was %s" \

                    % (self.name, err)

            else:

                msg = "Not valid repository at %s. Original error was %s" \

                    % (self.name, err)

            raise RepositoryError(msg)

    @LazyProperty

    def in_memory_changeset(self):

        return MercurialInMemoryChangeset(self)

    @LazyProperty

    def description(self):

        _desc = self._repo.ui.config(b'web', b'description', None, untrusted=True)

        return safe_str(_desc or b'unknown')

    @LazyProperty

    def contact(self):

        return safe_str(mercurial.hgweb.common.get_contact(self._repo.ui.config)

                            or b'Unknown')

    @LazyProperty

    def last_change(self):

        """

        Returns last change made on this repository as datetime object

        """

        return date_fromtimestamp(self._get_mtime(), makedate()[1])

    def _get_mtime(self):

        try:

            return time.mktime(self.get_changeset().date.timetuple())

        except RepositoryError:

            # fallback to filesystem

            cl_path = os.path.join(self.path, '.hg', "00changelog.i")

            st_path = os.path.join(self.path, '.hg', "store")

            if os.path.exists(cl_path):

                return os.stat(cl_path).st_mtime

            else:

                return os.stat(st_path).st_mtime

    def _get_revision(self, revision):

        """

        Given any revision identifier, returns a 40 char string with revision hash.

        :param revision: str or int or None

        """

        if self._empty:

            raise EmptyRepositoryError("There are no changesets yet")

        if revision in [-1, None]:

            revision = b'tip'

        elif isinstance(revision, str):

            revision = safe_bytes(revision)

        try:

            if isinstance(revision, int):

                return ascii_str(self._repo[revision].hex())

            return ascii_str(mercurial.scmutil.revsymbol(self._repo, revision).hex())

        except (IndexError, ValueError, mercurial.error.RepoLookupError, TypeError):

            msg = "Revision %r does not exist for %s" % (safe_str(revision), self.name)

            raise ChangesetDoesNotExistError(msg)

        except (LookupError, ):

            msg = "Ambiguous identifier `%s` for %s" % (safe_str(revision), self.name)

            raise ChangesetDoesNotExistError(msg)

    def get_ref_revision(self, ref_type, ref_name):

        """

        Returns revision number for the given reference.

        """

        if ref_type == 'rev' and not ref_name.strip('0'):

            return self.EMPTY_CHANGESET

        # lookup up the exact node id

        _revset_predicates = {

                'branch': 'branch',

                'book': 'bookmark',

                'tag': 'tag',

                'rev': 'id',

            }

        # avoid expensive branch(x) iteration over whole repo

        rev_spec = "%%s & %s(%%s)" % _revset_predicates[ref_type]

        try:

            revs = self._repo.revs(rev_spec, ref_name, ref_name)

        except LookupError:

            msg = "Ambiguous identifier %s:%s for %s" % (ref_type, ref_name, self.name)

            raise ChangesetDoesNotExistError(msg)

        except mercurial.error.RepoLookupError:

            msg = "Revision %s:%s does not exist for %s" % (ref_type, ref_name, self.name)

            raise ChangesetDoesNotExistError(msg)

        if revs:

            revision = revs.last()

        else:

            # TODO: just report 'not found'?

            revision = ref_name

        return self._get_revision(revision)

    def _get_archives(self, archive_name='tip'):

        allowed = self.baseui.configlist(b"web", b"allow_archive",

                                         untrusted=True)

        for name, ext in [(b'zip', '.zip'), (b'gz', '.tar.gz'), (b'bz2', '.tar.bz2')]:

            if name in allowed or self._repo.ui.configbool(b"web",

                                                           b"allow" + name,

                                                           untrusted=True):

                yield {"type": safe_str(name), "extension": ext, "node": archive_name}

    def _get_url(self, url):

        """

        Returns normalized url. If schema is not given, fall back to

        filesystem (``file:///``) schema.

        """

        if url != 'default' and '://' not in url:

            url = "file:" + urllib.request.pathname2url(url)

        return url

    def get_changeset(self, revision=None):

        """

        Returns ``MercurialChangeset`` object representing repository's

        changeset at the given ``revision``.

        """

        return MercurialChangeset(repository=self, revision=self._get_revision(revision))

    def get_changesets(self, start=None, end=None, start_date=None,

                       end_date=None, branch_name=None, reverse=False, max_revisions=None):

        """

        Returns iterator of ``MercurialChangeset`` objects from start to end

        (both are inclusive)

        :param start: None, str, int or mercurial lookup format

        :param end:  None, str, int or mercurial lookup format

        :param start_date:

        :param end_date:

        :param branch_name:

        :param reversed: return changesets in reversed order

        """

        start_raw_id = self._get_revision(start)

        start_pos = None if start is None else self.revisions.index(start_raw_id)

        end_raw_id = self._get_revision(end)

        end_pos = None if end is None else self.revisions.index(end_raw_id)

        if start_pos is not None and end_pos is not None and start_pos > end_pos:

            raise RepositoryError("Start revision '%s' cannot be "

                                  "after end revision '%s'" % (start, end))

"""

Utilities aimed to help achieve mostly basic tasks.

"""

import datetime

import os

import re

import time

import urllib.request

import mercurial.url

from kallithea.lib.vcs.exceptions import RepositoryError, VCSError

from kallithea.lib.vcs.utils.paths import abspath

ALIASES = ['hg', 'git']

def get_scm(path, search_up=False, explicit_alias=None):

    """

    Returns one of alias from ``ALIASES`` (in order of precedence same as

    shortcuts given in ``ALIASES``) and top working dir path for the given

    argument. If no scm-specific directory is found or more than one scm is

    found at that directory, ``VCSError`` is raised.

    :param search_up: if set to ``True``, this function would try to

      move up to parent directory every time no scm is recognized for the

      currently checked path. Default: ``False``.

    :param explicit_alias: can be one of available backend aliases, when given

      it will return given explicit alias in repositories under more than one

      version control, if explicit_alias is different than found it will raise

      VCSError

    """

    if not os.path.isdir(path):

        raise VCSError("Given path %s is not a directory" % path)

    while True:

        found_scms = [(scm, path) for scm in get_scms_for_path(path)]

        if found_scms or not search_up:

            break

        newpath = abspath(path, '..')

        if newpath == path:

            break

        path = newpath

    if len(found_scms) > 1:

        for scm in found_scms:

            if scm[0] == explicit_alias:

                return scm

        raise VCSError('More than one [%s] scm found at given path %s'

                       % (', '.join((x[0] for x in found_scms)), path))

    if len(found_scms) == 0:

        raise VCSError('No scm found at given path %s' % path)

    return found_scms[0]

def get_scms_for_path(path):

    """

    Returns all scm's found at the given path. If no scm is recognized

    - empty list is returned.

    :param path: path to directory which should be checked. May be callable.

    :raises VCSError: if given ``path`` is not a directory

    """

    from kallithea.lib.vcs.backends import get_backend

    if hasattr(path, '__call__'):

        path = path()

    if not os.path.isdir(path):

        raise VCSError("Given path %r is not a directory" % path)

    result = []

    for key in ALIASES:

        # find .hg / .git

        dirname = os.path.join(path, '.' + key)

        if os.path.isdir(dirname):

            result.append(key)

            continue

        # find rm__.hg / rm__.git too - left overs from old method for deleting

        dirname = os.path.join(path, 'rm__.' + key)

        if os.path.isdir(dirname):

            return result

        # We still need to check if it's not bare repository as

        # bare repos don't have working directories

        try:

            get_backend(key)(path)

            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.lexers import guess_lexer_for_filename, ClassNotFound

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

    CID_RE = r'[a-zA-Z0-9]+'

    if '..' not in text:

        m = re.match(r'^(?P<cid>%s)$' % CID_RE, text)

        if m:

            return {

                'start': None,

                'main': text,

                'end': None,

            }

    else:

        RE = r'^(?P<start>%s)?\.{2,3}(?P<end>%s)?$' % (CID_RE, CID_RE)

        m = re.match(RE, text)

        if m:

            result = m.groupdict()

            result['main'] = None

            return result

    raise ValueError("IDs not recognized")

def parse_datetime(text):

    """

    Parses given text and returns ``datetime.datetime`` instance or raises

    ``ValueError``.

    :param text: string of desired date/datetime or something more verbose,

      like *yesterday*, *2weeks 3days*, etc.

    """

    text = text.strip().lower()

    INPUT_FORMATS = (

        '%Y-%m-%d %H:%M:%S',

        '%Y-%m-%d %H:%M',

        '%Y-%m-%d',

        '%m/%d/%Y %H:%M:%S',

        '%m/%d/%Y %H:%M',

        '%m/%d/%Y',

        '%m/%d/%y %H:%M:%S',

        '%m/%d/%y %H:%M',

        '%m/%d/%y',

    )

    for format in INPUT_FORMATS:

        try:

            return datetime.datetime(*time.strptime(text, format)[:6])

        except ValueError:

            pass

    # Try descriptive texts

    if text == 'tomorrow':

        future = datetime.datetime.now() + datetime.timedelta(days=1)

        args = future.timetuple()[:3] + (23, 59, 59)

        return datetime.datetime(*args)

    elif text == 'today':

        return datetime.datetime(*datetime.datetime.today().timetuple()[:3])

    elif text == 'now':

        return datetime.datetime.now()

    elif text == 'yesterday':

        past = datetime.datetime.now() - datetime.timedelta(days=1)

        return datetime.datetime(*past.timetuple()[:3])

    else:

        days = 0

        matched = re.match(

            r'^((?P<weeks>\d+) ?w(eeks?)?)? ?((?P<days>\d+) ?d(ays?)?)?$', text)

        if matched:

            groupdict = matched.groupdict()

            if groupdict['days']:

                days += int(matched.groupdict()['days'])

            if groupdict['weeks']:

                days += int(matched.groupdict()['weeks']) * 7

            past = datetime.datetime.now() - datetime.timedelta(days=days)

            return datetime.datetime(*past.timetuple()[:3])

    raise ValueError('Wrong date: "%s"' % text)

def get_dict_for_attrs(obj, attrs):

    """

    Returns dictionary for each attribute from given ``obj``.

    """

    data = {}

    for attr in attrs:

        data[attr] = getattr(obj, attr)

    return data

def get_urllib_request_handlers(url_obj):

    handlers = []

    test_uri, authinfo = url_obj.authinfo()

    if authinfo:

        # create a password manager

        passmgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()

        handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),

                         mercurial.url.httpdigestauthhandler(passmgr)))

    return test_uri, handlers

            Session().commit()

            pytest.fail('no repo %s in filesystem' % repo_name)

        RepoModel().delete(repo_name_full)

        RepoGroupModel().delete(group_name)

        Session().commit()

    def test_create_in_group_without_needed_permissions(self):

        usr = self.log_user(base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR_PASS)

        # avoid spurious RepoGroup DetachedInstanceError ...

        session_csrf_secret_token = self.session_csrf_secret_token()

        # revoke

        user_model = UserModel()

        # disable fork and create on default user

        user_model.revoke_perm(User.DEFAULT_USER_NAME, 'hg.create.repository')

        user_model.grant_perm(User.DEFAULT_USER_NAME, 'hg.create.none')

        user_model.revoke_perm(User.DEFAULT_USER_NAME, 'hg.fork.repository')

        user_model.grant_perm(User.DEFAULT_USER_NAME, 'hg.fork.none')

        # disable on regular user

        user_model.revoke_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.create.repository')

        user_model.grant_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.create.none')

        user_model.revoke_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')

        user_model.grant_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.fork.none')

        Session().commit()

        ## create GROUP

        group_name = 'reg_sometest_%s' % self.REPO_TYPE

        gr = RepoGroupModel().create(group_name=group_name,

                                     group_description='test',

                                     owner=base.TEST_USER_ADMIN_LOGIN)

        Session().commit()

        group_name_allowed = 'reg_sometest_allowed_%s' % self.REPO_TYPE

        gr_allowed = RepoGroupModel().create(group_name=group_name_allowed,

                                     group_description='test',

                                     owner=base.TEST_USER_REGULAR_LOGIN)

        Session().commit()

        repo_name = 'ingroup'

        repo_name_full = db.URL_SEP.join([group_name, repo_name])

        description = 'description for newly created repo'

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

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                repo_group=gr.group_id,

                                                _session_csrf_secret_token=session_csrf_secret_token))

        response.mustcontain('Invalid value')

        # user is allowed to create in this group

        repo_name = 'ingroup'

        repo_name_full = db.URL_SEP.join([group_name_allowed, repo_name])

        description = 'description for newly created repo'

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

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                repo_group=gr_allowed.group_id,

                                                _session_csrf_secret_token=session_csrf_secret_token))

        ## run the check page that triggers the flash message

        response = self.app.get(base.url('repo_check_home', repo_name=repo_name_full))

        assert response.json == {'result': True}

        self.checkSessionFlash(response,

                               'Created repository <a href="/%s">%s</a>'

                               % (repo_name_full, repo_name_full))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_full).one()

        new_repo_id = new_repo.repo_id

        assert new_repo.repo_name == repo_name_full

        assert new_repo.description == description

        # test if the repository is visible in the list ?

        response = self.app.get(base.url('summary_home', repo_name=repo_name_full))

        response.mustcontain(repo_name_full)

        response.mustcontain(self.REPO_TYPE)

        inherited_perms = UserRepoToPerm.query() \

            .filter(UserRepoToPerm.repository_id == new_repo_id).all()

        assert len(inherited_perms) == 1

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full))

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

            pytest.fail('no repo %s in filesystem' % repo_name)

        RepoModel().delete(repo_name_full)

        RepoGroupModel().delete(group_name)

        RepoGroupModel().delete(group_name_allowed)

        Session().commit()

    def test_create_in_group_inherit_permissions(self):

        self.log_user()

        ## create GROUP

        group_name = 'sometest_%s' % self.REPO_TYPE

        gr = RepoGroupModel().create(group_name=group_name,

                                     group_description='test',

                                     owner=base.TEST_USER_ADMIN_LOGIN)

        perm = Permission.get_by_key('repository.write')

        RepoGroupModel().grant_user_permission(gr, base.TEST_USER_REGULAR_LOGIN, perm)

        ## add repo permissions

        Session().commit()

        repo_name = 'ingroup_inherited_%s' % self.REPO_TYPE

        repo_name_full = db.URL_SEP.join([group_name, repo_name])

        description = 'description for newly created repo'

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

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                repo_group=gr.group_id,

                                                repo_copy_permissions=True,

                                                _session_csrf_secret_token=self.session_csrf_secret_token()))

        ## run the check page that triggers the flash message

        response = self.app.get(base.url('repo_check_home', repo_name=repo_name_full))

        self.checkSessionFlash(response,

                               'Created repository <a href="/%s">%s</a>'

                               % (repo_name_full, repo_name_full))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_full).one()

        new_repo_id = new_repo.repo_id

        assert new_repo.repo_name == repo_name_full

        assert new_repo.description == description

        # test if the repository is visible in the list ?

        response = self.app.get(base.url('summary_home', repo_name=repo_name_full))

        response.mustcontain(repo_name_full)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full))

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

            pytest.fail('no repo %s in filesystem' % repo_name)

        # check if inherited permissiona are applied

        inherited_perms = UserRepoToPerm.query() \

            .filter(UserRepoToPerm.repository_id == new_repo_id).all()

        assert len(inherited_perms) == 2

        assert base.TEST_USER_REGULAR_LOGIN in [x.user.username

                                                    for x in inherited_perms]

        assert 'repository.write' in [x.permission.permission_name

                                               for x in inherited_perms]

        RepoModel().delete(repo_name_full)

        RepoGroupModel().delete(group_name)

        Session().commit()

    def test_create_remote_repo_wrong_clone_uri(self):

        self.log_user()

        repo_name = self.NEW_REPO

        description = 'description for newly created repo'

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

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                clone_uri='http://127.0.0.1/repo',

                                                _session_csrf_secret_token=self.session_csrf_secret_token()))

        response.mustcontain('Invalid repository URL')

    def test_create_remote_repo_wrong_clone_uri_hg_svn(self):

        self.log_user()

        repo_name = self.NEW_REPO

        description = 'description for newly created repo'

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

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                clone_uri='svn+http://127.0.0.1/repo',

                                                _session_csrf_secret_token=self.session_csrf_secret_token()))

        response.mustcontain('Invalid repository URL')

    def test_delete(self):

        self.log_user()

        repo_name = 'vcs_test_new_to_delete_%s' % self.REPO_TYPE

        description = 'description for newly created repo'

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

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_type=self.REPO_TYPE,

                                                repo_name=repo_name,

                                                repo_description=description,

                                                _session_csrf_secret_token=self.session_csrf_secret_token()))

        ## run the check page that triggers the flash message

        response = self.app.get(base.url('repo_check_home', repo_name=repo_name))

        self.checkSessionFlash(response,

                               'Created repository <a href="/%s">%s</a>'

                               % (repo_name, repo_name))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name).one()

        assert new_repo.repo_name == repo_name

        assert new_repo.description == description

        # test if the repository is visible in the list ?