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:

"""

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

    """

    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

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

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

        response.mustcontain(repo_name)

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

        except vcs.exceptions.VCSError:

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

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

            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})

        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))

        response.follow()

        # check if repo was deleted from db

        deleted_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name).scalar()

        assert deleted_repo is None

        assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)) == False