def get_changeset(self, revision=None):

        """

        Returns ``GitChangeset`` object representing commit from git repository

        at the given revision or head (most recent commit) if None given.

        """

        if isinstance(revision, GitChangeset):

            return revision

        revision = self._get_revision(revision)

        changeset = GitChangeset(repository=self, revision=revision)

        return changeset

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

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

        """

        Returns iterator of ``GitChangeset`` objects from start to end (both

        are inclusive), in ascending date order (unless ``reverse`` is set).

        :param start: changeset ID, as str; first returned changeset

        :param end: changeset ID, as str; last returned changeset

        :param start_date: if specified, changesets with commit date less than

          ``start_date`` would be filtered out from returned set

        :param end_date: if specified, changesets with commit date greater than

          ``end_date`` would be filtered out from returned set

        :param branch_name: if specified, changesets not reachable from given

          branch would be filtered out from returned set

        :param reverse: if ``True``, returned generator would be reversed

          (meaning that returned changesets would have descending date order)

        :raise BranchDoesNotExistError: If given ``branch_name`` does not

            exist.

        :raise ChangesetDoesNotExistError: If changeset for given ``start`` or

          ``end`` could not be found.

        """

        if branch_name and branch_name not in self.branches:

            raise BranchDoesNotExistError("Branch '%s' not found" \

                                          % branch_name)

        # actually we should check now if it's not an empty repo to not spaw

        # subprocess commands

        if self._empty:

            raise EmptyRepositoryError("There are no changesets yet")

        # %H at format means (full) commit hash, initial hashes are retrieved

        # in ascending date order

        cmd = ['log', '--date-order', '--reverse', '--pretty=format:%H']

        if start_date:

            cmd += ['--since', start_date.strftime('%m/%d/%y %H:%M:%S')]

        if end_date:

            cmd += ['--until', end_date.strftime('%m/%d/%y %H:%M:%S')]

        if branch_name:

            cmd.append(branch_name)

        else:

            cmd.append(settings.GIT_REV_FILTER)

        revs = self.run_git_command(cmd)[0].splitlines()

        start_pos = 0

        end_pos = len(revs)

        if start:

            _start = self._get_revision(start)

            try:

                start_pos = revs.index(_start)

            except ValueError:

                pass

        if end is not None:

            _end = self._get_revision(end)

            try:

                end_pos = revs.index(_end)

            except ValueError:

                pass

        if None not in [start, end] and start_pos > end_pos:

            raise RepositoryError('start cannot be after end')

        if end_pos is not None:

            end_pos += 1

        revs = revs[start_pos:end_pos]

        if reverse:

            revs = reversed(revs)

        return CollectionGenerator(self, revs)

    def get_diff(self, rev1, rev2, path=None, 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

        """

        flags = ['-U%s' % context, '--full-index', '--binary', '-p', '-M', '--abbrev=40']

        if ignore_whitespace:

            flags.append('-w')

        if hasattr(rev1, 'raw_id'):

            rev1 = getattr(rev1, 'raw_id')

        if hasattr(rev2, 'raw_id'):

            rev2 = getattr(rev2, 'raw_id')

        if rev1 == self.EMPTY_CHANGESET:

            rev2 = self.get_changeset(rev2).raw_id

            cmd = ['show'] + flags + [rev2]

        else:

            rev1 = self.get_changeset(rev1).raw_id

            rev2 = self.get_changeset(rev2).raw_id

            cmd = ['diff'] + flags + [rev1, rev2]

        if path:

            cmd += ['--', path]

        stdout, stderr = self.run_git_command(cmd)

        # TODO: don't ignore stderr

        # If we used 'show' command, strip first few lines (until actual diff

        # starts)

        if rev1 == self.EMPTY_CHANGESET:

            parts = stdout.split('\ndiff ', 1)

            if len(parts) > 1:

                stdout = 'diff ' + parts[1]

        return stdout

    @LazyProperty

    def in_memory_changeset(self):

        """

        Returns ``GitInMemoryChangeset`` object for this repository.

        """

        return GitInMemoryChangeset(self)

    def clone(self, url, update_after_clone=True, bare=False):

        """

        Tries to clone changes from external location.

        :param update_after_clone: If set to ``False``, git won't checkout

          working directory

        :param bare: If set to ``True``, repository would be cloned into

          *bare* git repository (no working directory at all).

        """

        url = self._get_url(url)

        cmd = ['clone', '-q']

        if bare:

            cmd.append('--bare')

        elif not update_after_clone:

            cmd.append('--no-checkout')

        cmd += ['--', url, self.path]

        # If error occurs run_git_command raises RepositoryError already

        self.run_git_command(cmd)

    def pull(self, url):

        """

        Tries to pull changes from external location.

        """

        url = self._get_url(url)

        cmd = ['pull', '--ff-only', url]

        # If error occurs run_git_command raises RepositoryError already

        self.run_git_command(cmd)

    def fetch(self, url):

        """

        Tries to pull changes from external location.

        """

        url = self._get_url(url)

        so, se = self.run_git_command(['ls-remote', '-h', url])

        cmd = ['fetch', url, '--']

        for line in (x for x in so.splitlines()):

            sha, ref = line.split('\t')

            cmd.append('+%s:%s' % (ref, ref))

        self.run_git_command(cmd)

    def _update_server_info(self):

        """

        runs gits update-server-info command in this repo instance

        """

        from dulwich.server import update_server_info

        try:

            update_server_info(self._repo)

        except OSError as e:

            if e.errno not in [errno.ENOENT, errno.EROFS]:

                raise

            # Workaround for dulwich crashing on for example its own dulwich/tests/data/repos/simple_merge.git/info/refs.lock

            log.error('Ignoring %s running update-server-info: %s', type(e).__name__, e)

    @LazyProperty

    def workdir(self):

        """

        Returns ``Workdir`` instance for this repository.

        """

    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 = datetime.datetime.now().ctime()

        try:

            self._repo.tag(name, changeset._ctx.node(), message, local, user,

                date)

        except Abort as e:

            raise RepositoryError(e.message)

        # 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 = datetime.datetime.now().ctime()

        local = False

        try:

            self._repo.tag(name, nullid, message, local, user, date)

            self.tags = self._get_tags()

        except Abort as e:

            raise RepositoryError(e.message)

    @LazyProperty

    def bookmarks(self):

        """

        Gets bookmarks for this repository

        """

        return self._get_bookmarks()

    def _get_bookmarks(self):

        if self._empty:

            return {}

        sortkey = lambda ctx: ctx[0]  # sort by name

        _bookmarks = [(safe_unicode(n), hex(h),) for n, h in

                 self._repo._bookmarks.items()]

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

    def _get_all_revisions(self):

        return [self._repo[x].hex() for x in self._repo.filtered('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

        """

        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 = match(self.path, '', [path])

        else:

            file_filter = None

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

                          opts=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

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

            return True

        if url.startswith('ssh:'):

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

            # throw an exception.

            sshpeer(repoui or ui.ui(), url).lookup('tip')

            return True

        url_prefix = None

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

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

        handlers = []

        url_obj = hg_url(url)

        test_uri, authinfo = url_obj.authinfo()

        url_obj.passwd = '*****'

        cleaned_uri = str(url_obj)

        if authinfo:

            #create a password manager

            passmgr = urllib2.HTTPPasswordMgrWithDefaultRealm()

            passmgr.add_password(*authinfo)

            handlers.extend((httpbasicauthhandler(passmgr),

                             httpdigestauthhandler(passmgr)))

        o = urllib2.build_opener(*handlers)

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

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

        q = {"cmd": 'between'}

        q.update({'pairs': "%s-%s" % ('0' * 40, '0' * 40)})

        qs = '?%s' % urllib.urlencode(q)

        cu = "%s%s" % (test_uri, qs)

        req = urllib2.Request(cu, None, {})

        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 urllib2.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:

                httppeer(repoui or ui.ui(), url).lookup('tip')

            except Exception as e:

                raise urllib2.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.

        self.assertEqual('marcink',

          self.repo.get_changeset('8430a588b43b5d6da365400117c89400326e7992')\

          .author_name)

class GitSpecificTest(unittest.TestCase):

    def test_error_is_raised_for_added_if_diff_name_status_is_wrong(self):

        repo = mock.MagicMock()

        changeset = GitChangeset(repo, 'foobar')

        changeset._diff_name_status = 'foobar'

        with self.assertRaises(VCSError):

            changeset.added

    def test_error_is_raised_for_changed_if_diff_name_status_is_wrong(self):

        repo = mock.MagicMock()

        changeset = GitChangeset(repo, 'foobar')

        changeset._diff_name_status = 'foobar'

        with self.assertRaises(VCSError):

            changeset.added

    def test_error_is_raised_for_removed_if_diff_name_status_is_wrong(self):

        repo = mock.MagicMock()

        changeset = GitChangeset(repo, 'foobar')

        changeset._diff_name_status = 'foobar'

        with self.assertRaises(VCSError):

            changeset.added

class GitSpecificWithRepoTest(_BackendTestMixin, unittest.TestCase):

    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('foobar/static/js/admin/base.js', content='base'),

                    FileNode('foobar/static/admin', content='admin',

                        mode=0120000), # this is a link

                    FileNode('foo', content='foo'),

                ],

            },

            {

                'message': 'Second',

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

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

                'added': [

                    FileNode('foo2', content='foo2'),

                ],

            },

        ]

    def test_paths_slow_traversing(self):

        cs = self.repo.get_changeset()

        self.assertEqual(cs.get_node('foobar').get_node('static').get_node('js')

            .get_node('admin').get_node('base.js').content, 'base')

    def test_paths_fast_traversing(self):

        cs = self.repo.get_changeset()

        self.assertEqual(cs.get_node('foobar/static/js/admin/base.js').content,

            'base')

    def test_workdir_get_branch(self):

        self.repo.run_git_command(['checkout', '-b', 'production'])

        # Regression test: one of following would fail if we don't check

        # .git/HEAD file

        self.repo.run_git_command(['checkout', 'production'])

        self.assertEqual(self.repo.workdir.get_branch(), 'production')

        self.repo.run_git_command(['checkout', 'master'])

        self.assertEqual(self.repo.workdir.get_branch(), 'master')

    def test_get_diff_runs_git_command_with_hashes(self):

        self.repo.run_git_command = mock.Mock(return_value=['', ''])

        self.repo.get_diff(0, 1)

        self.repo.run_git_command.assert_called_once_with(

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

             self.repo._get_revision(0), self.repo._get_revision(1)])

    def test_get_diff_runs_git_command_with_str_hashes(self):

        self.repo.run_git_command = mock.Mock(return_value=['', ''])

        self.repo.get_diff(self.repo.EMPTY_CHANGESET, 1)

        self.repo.run_git_command.assert_called_once_with(

            ['show', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',

             self.repo._get_revision(1)])

    def test_get_diff_runs_git_command_with_path_if_its_given(self):

        self.repo.run_git_command = mock.Mock(return_value=['', ''])

        self.repo.get_diff(0, 1, 'foo')

        self.repo.run_git_command.assert_called_once_with(

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

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

class GitRegressionTest(_BackendTestMixin, unittest.TestCase):

    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]

        self.assertEqual(paths(*cs.get_nodes('bot')), ['bot/build', 'bot/templates', 'bot/__init__.py'])

        self.assertEqual(paths(*cs.get_nodes('bot/build')), ['bot/build/migrations', 'bot/build/static', 'bot/build/templates'])

        self.assertEqual(paths(*cs.get_nodes('bot/build/static')), ['bot/build/static/templates'])

        # this get_nodes below causes troubles !

        self.assertEqual(paths(*cs.get_nodes('bot/build/static/templates')), ['bot/build/static/templates/f.html', 'bot/build/static/templates/f1.html'])

        self.assertEqual(paths(*cs.get_nodes('bot/build/templates')), ['bot/build/templates/err.html', 'bot/build/templates/err2.html'])

        self.assertEqual(paths(*cs.get_nodes('bot/templates/')), ['bot/templates/404.html', 'bot/templates/500.html'])

class GitHooksTest(unittest.TestCase):

    """

    Tests related to hook functionality of Git repositories.

    """

    def setUp(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

        # them. Deduplicates code in tests a bit.

        self.hook_directory = self.repo.get_hook_location()

        self.kallithea_hooks = {h: os.path.join(self.hook_directory, h) for h in ("pre-receive", "post-receive")}

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

            self.assertTrue(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:

                self.assertNotIn("JUST_BOGUS", 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")

import os

from kallithea.lib.vcs.backends.hg import MercurialRepository, MercurialChangeset

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

from kallithea.lib.vcs.nodes import NodeKind, NodeState

from kallithea.tests.vcs.conf import TEST_HG_REPO, TEST_HG_REPO_CLONE, \

    TEST_HG_REPO_PULL

from kallithea.lib.vcs.utils.compat import unittest

class MercurialRepositoryTest(unittest.TestCase):

    def __check_for_existing_repo(self):

        if os.path.exists(TEST_HG_REPO_CLONE):

            self.fail('Cannot test mercurial clone repo as location %s already '

                      'exists. You should manually remove it first.'

                      % TEST_HG_REPO_CLONE)

    def setUp(self):

        self.repo = MercurialRepository(TEST_HG_REPO)

    def test_wrong_repo_path(self):

        wrong_repo_path = '/tmp/errorrepo'

        self.assertRaises(RepositoryError, MercurialRepository, wrong_repo_path)

    def test_unicode_path_repo(self):

        self.assertRaises(VCSError,lambda:MercurialRepository(u'iShouldFail'))

    def test_repo_clone(self):

        self.__check_for_existing_repo()

        repo = MercurialRepository(TEST_HG_REPO)

        repo_clone = MercurialRepository(TEST_HG_REPO_CLONE,

            src_url=TEST_HG_REPO, update_after_clone=True)

        self.assertEqual(len(repo.revisions), len(repo_clone.revisions))

        # Checking hashes of changesets should be enough

        for changeset in repo.get_changesets():

            raw_id = changeset.raw_id

            self.assertEqual(raw_id, repo_clone.get_changeset(raw_id).raw_id)

    def test_repo_clone_with_update(self):

        repo = MercurialRepository(TEST_HG_REPO)

        repo_clone = MercurialRepository(TEST_HG_REPO_CLONE + '_w_update',

            src_url=TEST_HG_REPO, update_after_clone=True)

        self.assertEqual(len(repo.revisions), len(repo_clone.revisions))

        #check if current workdir was updated

        self.assertEqual(os.path.isfile(os.path.join(TEST_HG_REPO_CLONE \

                                                    + '_w_update',

                                                    'MANIFEST.in')), True,)

    def test_repo_clone_without_update(self):

        repo = MercurialRepository(TEST_HG_REPO)

        repo_clone = MercurialRepository(TEST_HG_REPO_CLONE + '_wo_update',

            src_url=TEST_HG_REPO, update_after_clone=False)

        self.assertEqual(len(repo.revisions), len(repo_clone.revisions))

        self.assertEqual(os.path.isfile(os.path.join(TEST_HG_REPO_CLONE \

                                                    + '_wo_update',

                                                    'MANIFEST.in')), False,)

    def test_pull(self):

        if os.path.exists(TEST_HG_REPO_PULL):

            self.fail('Cannot test mercurial pull command as location %s '

                      'already exists. You should manually remove it first'

                      % TEST_HG_REPO_PULL)

        repo_new = MercurialRepository(TEST_HG_REPO_PULL, create=True)

        self.assertTrue(len(self.repo.revisions) > len(repo_new.revisions))

        repo_new.pull(self.repo.path)

        repo_new = MercurialRepository(TEST_HG_REPO_PULL)

        self.assertTrue(len(self.repo.revisions) == len(repo_new.revisions))

    def test_revisions(self):

        # there are 21 revisions at bitbucket now

        # so we can assume they would be available from now on

        subset = set(['b986218ba1c9b0d6a259fac9b050b1724ed8e545',

                 '3d8f361e72ab303da48d799ff1ac40d5ac37c67e',

                 '6cba7170863a2411822803fa77a0a264f1310b35',

                 '56349e29c2af3ac913b28bde9a2c6154436e615b',

                 '2dda4e345facb0ccff1a191052dd1606dba6781d',

                 '6fff84722075f1607a30f436523403845f84cd9e',

                 '7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7',

                 '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb',

                 'dc5d2c0661b61928834a785d3e64a3f80d3aad9c',

                 'be90031137367893f1c406e0a8683010fd115b79',

                 'db8e58be770518cbb2b1cdfa69146e47cd481481',

                 '84478366594b424af694a6c784cb991a16b87c21',

                 '17f8e105dddb9f339600389c6dc7175d395a535c',

                 '20a662e756499bde3095ffc9bc0643d1def2d0eb',

                 '2e319b85e70a707bba0beff866d9f9de032aa4f9',

                 '786facd2c61deb9cf91e9534735124fb8fc11842',

                 '94593d2128d38210a2fcd1aabff6dda0d6d9edf8',

                 'aa6a0de05b7612707db567078e130a6cd114a9a7',

                 'eada5a770da98ab0dd7325e29d00e0714f228d09'

                ])

        self.assertTrue(subset.issubset(set(self.repo.revisions)))

        # check if we have the proper order of revisions

        #4 1 5 10 95

        for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),

                                 (10, 20, 10), (5, 100, 95)]:

            revs = list(self.repo[sfrom:sto])

            self.assertEqual(len(revs), size)

            self.assertEqual(revs[0], self.repo.get_changeset(sfrom))

            self.assertEqual(revs[-1], self.repo.get_changeset(sto - 1))

    def test_branches(self):

        # TODO: Need more tests here

        #active branches

        self.assertTrue('default' in self.repo.branches)

        self.assertTrue('stable' in self.repo.branches)

        # closed

        self.assertTrue('git' in self.repo._get_branches(closed=True))

        self.assertTrue('web' in self.repo._get_branches(closed=True))

        for name, id in self.repo.branches.items():

            self.assertTrue(isinstance(

                self.repo.get_changeset(id), MercurialChangeset))

    def test_tip_in_tags(self):

        # tip is always a tag

        self.assertIn('tip', self.repo.tags)

    def test_tip_changeset_in_tags(self):

        tip = self.repo.get_changeset()

        self.assertEqual(self.repo.tags['tip'], tip.raw_id)

    def test_initial_changeset(self):

        init_chset = self.repo.get_changeset(0)

        self.assertEqual(init_chset.message, 'initial import')

        self.assertEqual(init_chset.author,

            'Marcin Kuzminski <marcin@python-blog.com>')

        self.assertEqual(sorted(init_chset._file_paths),

            sorted([

                'vcs/__init__.py',

                'vcs/backends/BaseRepository.py',

                'vcs/backends/__init__.py',

            ])

        )

        self.assertEqual(sorted(init_chset._dir_paths),

            sorted(['', 'vcs', 'vcs/backends']))

        self.assertRaises(NodeDoesNotExistError, init_chset.get_node, path='foobar')

        node = init_chset.get_node('vcs/')

        self.assertTrue(hasattr(node, 'kind'))

        self.assertEqual(node.kind, NodeKind.DIR)

        node = init_chset.get_node('vcs')

        self.assertTrue(hasattr(node, 'kind'))

        self.assertEqual(node.kind, NodeKind.DIR)

        node = init_chset.get_node('vcs/__init__.py')

        self.assertTrue(hasattr(node, 'kind'))

        self.assertEqual(node.kind, NodeKind.FILE)

    def test_not_existing_changeset(self):

        #rawid

        self.assertRaises(RepositoryError, self.repo.get_changeset,

            'abcd' * 10)

        #shortid

        self.assertRaises(RepositoryError, self.repo.get_changeset,

            'erro' * 4)

        #numeric

        self.assertRaises(RepositoryError, self.repo.get_changeset,

            self.repo.count() + 1)

        # Small chance we ever get to this one

        revision = pow(2, 30)

        self.assertRaises(RepositoryError, self.repo.get_changeset, revision)

    def test_changeset10(self):

        chset10 = self.repo.get_changeset(10)

        README = """===

VCS

===

Various Version Control System management abstraction layer for Python.

Introduction

------------

TODO: To be written...

"""

        node = chset10.get_node('README.rst')

        self.assertEqual(node.kind, NodeKind.FILE)

        self.assertEqual(node.content, README)