else:

                    return Repo.init(self.path)

            else:

                return Repo(self.path)

        except (NotGitRepository, OSError) as err:

            raise RepositoryError(err)

    def _get_all_revisions(self):

        # we must check if this repo is not empty, since later command

        # fails if it is. And it's cheaper to ask than throw the subprocess

        # errors

        try:

            self._repo.head()

        except KeyError:

            return []

        rev_filter = settings.GIT_REV_FILTER

        cmd = ['rev-list', rev_filter, '--reverse', '--date-order']

        try:

            so = self.run_git_command(cmd)

        except RepositoryError:

            # Can be raised for empty repositories

            return []

        return so.splitlines()

    def _get_all_revisions2(self):

        # alternate implementation using dulwich

        includes = [ascii_str(sha) for key, (sha, type_) in self._parsed_refs.iteritems()

                    if type_ != b'T']

        return [c.commit.id for c in self._repo.get_walker(include=includes)]

    def _get_revision(self, revision):

        """

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

        """

        if self._empty:

            raise EmptyRepositoryError("There are no changesets yet")

        if revision in (None, '', 'tip', 'HEAD', 'head', -1):

            revision = -1

        if isinstance(revision, int):

            try:

                return self.revisions[revision]

            except IndexError:

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

                raise ChangesetDoesNotExistError(msg)

        if isinstance(revision, (str, unicode)):

            if revision.isdigit() and (len(revision) < 12 or len(revision) == revision.count('0')):

                try:

                    return self.revisions[int(revision)]

                except IndexError:

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

                    raise ChangesetDoesNotExistError(msg)

            # get by branch/tag name

            _ref_revision = self._parsed_refs.get(safe_bytes(revision))

            if _ref_revision:  # and _ref_revision[1] in [b'H', b'RH', b'T']:

                return ascii_str(_ref_revision[0])

            if revision in self.revisions:

                return revision

            # maybe it's a tag ? we don't have them in self.revisions

            if revision in self.tags.values():

                return revision

            if SHA_PATTERN.match(revision):

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

                raise ChangesetDoesNotExistError(msg)

        raise ChangesetDoesNotExistError("Given revision %r not recognized" % revision)

    def get_ref_revision(self, ref_type, ref_name):

        """

        Returns ``GitChangeset`` object representing repository's

        changeset at the given ``revision``.

        """

        return self._get_revision(ref_name)

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

        for i in [('zip', '.zip'), ('gz', '.tar.gz'), ('bz2', '.tar.bz2')]:

            yield {"type": i[0], "extension": i[1], "node": archive_name}

    def _get_url(self, url):

        """

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

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

        """

        url = safe_str(url)

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

            url = ':///'.join(('file', url))

        return url

    @LazyProperty

    def name(self):

        return os.path.basename(self.path)

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

            idx_loc = '' if self.bare else '.git'

            # fallback to filesystem

            in_path = os.path.join(self.path, idx_loc, "index")

            he_path = os.path.join(self.path, idx_loc, "HEAD")

            if os.path.exists(in_path):

                return os.stat(in_path).st_mtime

            else:

                return os.stat(he_path).st_mtime

    @LazyProperty

    def description(self):

        return safe_unicode(self._repo.get_description() or b'unknown')

    @LazyProperty

    def contact(self):

        undefined_contact = u'Unknown'

        return undefined_contact

    @property

    def branches(self):

        if not self.revisions:

            return {}

        sortkey = lambda ctx: ctx[0]

        _branches = [(key, ascii_str(sha))

                     for key, (sha, type_) in self._parsed_refs.iteritems() if type_ == b'H']

        return OrderedDict(sorted(_branches, key=sortkey, reverse=False))

    @LazyProperty

    def closed_branches(self):

        return {}

    @LazyProperty

    def tags(self):

        return self._get_tags()

    def _get_tags(self):

        if not self.revisions:

            return {}

        sortkey = lambda ctx: ctx[0]

        _tags = [(key, ascii_str(sha))

                 for key, (sha, type_) in self._parsed_refs.iteritems() if type_ == b'T']

        return OrderedDict(sorted(_tags, key=sortkey, reverse=True))

    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)

        message = message or "Added tag %s for commit %s" % (name,

            changeset.raw_id)

        self._repo.refs[b"refs/tags/%s" % safe_bytes(name)] = changeset._commit.id

        self._parsed_refs = self._get_parsed_refs()

        self.tags = self._get_tags()

        return changeset

    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)

    @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, unicode):

            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_unicode(revision), self.name)

            raise ChangesetDoesNotExistError(msg)

        except (LookupError, ):

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

            raise ChangesetDoesNotExistError(msg)

    def get_ref_revision(self, ref_type, ref_name):

        """

        Returns revision number for the given reference.

        """

        ref_name = safe_str(ref_name)

        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": name, "extension": ext, "node": archive_name}

    def _get_url(self, url):

        """

        Returns normalized url. If schema is not given, would fall

        to filesystem

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

        """

        url = safe_str(url)

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

            url = "file:" + urllib.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))

        if branch_name and branch_name not in self.allbranches:

            msg = "Branch %r not found in %s" % (branch_name, self.name)

            raise BranchDoesNotExistError(msg)

        if end_pos is not None:

            end_pos += 1

        # filter branches

        filter_ = []

        if branch_name:

            filter_.append(b'branch("%s")' % safe_bytes(branch_name))

        if start_date:

            filter_.append(b'date(">%s")' % safe_bytes(str(start_date)))

        if end_date:

            filter_.append(b'date("<%s")' % safe_bytes(str(end_date)))

        if filter_ or max_revisions:

            if filter_:

                revspec = b' and '.join(filter_)

            else:

                revspec = b'all()'

            if max_revisions:

                revspec = b'limit(%s, %d)' % (revspec, max_revisions)

            revisions = mercurial.scmutil.revrange(self._repo, [revspec])

        else:

            revisions = self.revisions

        # this is very much a hack to turn this into a list; a better solution

        # would be to get rid of this function entirely and use revsets

        revs = list(revisions)[start_pos:end_pos]

        if reverse:

            revs.reverse()

        return CollectionGenerator(self, revs)

    def pull(self, url):

        """

        Tries to pull changes from external location.

        """

        other = mercurial.hg.peer(self._repo, {}, safe_bytes(self._get_url(url)))

        try:

            mercurial.exchange.pull(self._repo, other, heads=None, force=None)

        except mercurial.error.Abort as err:

            # Propagate error but with vcs's type

            raise RepositoryError(str(err))

    @LazyProperty

    def workdir(self):

        """

        Returns ``Workdir`` instance for this repository.

        """

        return MercurialWorkdir(self)

    def get_config_value(self, section, name=None, config_file=None):

        """

        Returns configuration value for a given [``section``] and ``name``.

        :param section: Section we want to retrieve value from

        :param name: Name of configuration we want to retrieve

        :param config_file: A path to file which should be used to retrieve

          configuration from (might also be a list of file paths)

        """

        if config_file is None:

            config_file = []

        elif isinstance(config_file, basestring):

            config_file = [config_file]

        config = self._repo.ui

        if config_file:

            config = mercurial.ui.ui()

    def test_get_diff_does_not_sanitize_valid_context(self):

        almost_overflowed_long_int = 2**31-1

        self.repo._run_git_command = mock.Mock(return_value=(b'', b''))

        self.repo.get_diff(0, 1, 'foo', context=almost_overflowed_long_int)

        self.repo._run_git_command.assert_called_once_with(

            ['diff', '-U' + str(almost_overflowed_long_int), '--full-index', '--binary', '-p', '-M', '--abbrev=40',

             self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)

    def test_get_diff_sanitizes_overflowing_context(self):

        overflowed_long_int = 2**31

        sanitized_overflowed_long_int = overflowed_long_int-1

        self.repo._run_git_command = mock.Mock(return_value=(b'', b''))

        self.repo.get_diff(0, 1, 'foo', context=overflowed_long_int)

        self.repo._run_git_command.assert_called_once_with(

            ['diff', '-U' + str(sanitized_overflowed_long_int), '--full-index', '--binary', '-p', '-M', '--abbrev=40',

             self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)

    def test_get_diff_does_not_sanitize_zero_context(self):

        zero_context = 0

        self.repo._run_git_command = mock.Mock(return_value=(b'', b''))

        self.repo.get_diff(0, 1, 'foo', context=zero_context)

        self.repo._run_git_command.assert_called_once_with(

            ['diff', '-U' + str(zero_context), '--full-index', '--binary', '-p', '-M', '--abbrev=40',

             self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)

    def test_get_diff_sanitizes_negative_context(self):

        negative_context = -10

        self.repo._run_git_command = mock.Mock(return_value=(b'', b''))

        self.repo.get_diff(0, 1, 'foo', context=negative_context)

        self.repo._run_git_command.assert_called_once_with(

            ['diff', '-U0', '--full-index', '--binary', '-p', '-M', '--abbrev=40',

             self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)

class TestGitRegression(_BackendTestMixin):

    backend_alias = 'git'

    @classmethod

    def _get_commits(cls):

        return [

            {

                'message': 'Initial',

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': datetime.datetime(2010, 1, 1, 20),

                'added': [

                    FileNode('bot/__init__.py', content='base'),

                    FileNode('bot/templates/404.html', content='base'),

                    FileNode('bot/templates/500.html', content='base'),

                ],

            },

            {

                'message': 'Second',

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': datetime.datetime(2010, 1, 1, 22),

                'added': [

                    FileNode('bot/build/migrations/1.py', content='foo2'),

                    FileNode('bot/build/migrations/2.py', content='foo2'),

                    FileNode('bot/build/static/templates/f.html', content='foo2'),

                    FileNode('bot/build/static/templates/f1.html', content='foo2'),

                    FileNode('bot/build/templates/err.html', content='foo2'),

                    FileNode('bot/build/templates/err2.html', content='foo2'),

                ],

            },

        ]

    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

    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.iteritems():

            if os.path.exists(hook_path):

                os.remove(hook_path)

        ScmModel().install_git_hooks(repo=self.repo)

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

            assert os.path.exists(hook_path)

    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.iteritems():

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

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

        ScmModel().install_git_hooks(repo=self.repo)

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

            with open(hook_path) as f:

                assert "JUST_BOGUS" not in f.read()

    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.iteritems():

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

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

        ScmModel().install_git_hooks(repo=self.repo)

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

            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.iteritems():

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

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

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

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

            with open(hook_path) as f:

                assert "KALLITHEA_HOOK_VER" in f.read()