# -*- coding: utf-8 -*-

import os

import mock

import urllib

from kallithea.lib import vcs

from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User, \

    Permission

from kallithea.model.user import UserModel

from kallithea.tests import *

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.repo import RepoModel

from kallithea.model.meta import Session

from kallithea.tests.fixture import Fixture, error_function

fixture = Fixture()

def _get_permission_for_user(user, repo):

    perm = UserRepoToPerm.query() \

                .filter(UserRepoToPerm.repository ==

                        Repository.get_by_repo_name(repo)) \

                .filter(UserRepoToPerm.user == User.get_by_username(user)) \

                .all()

    return perm

class _BaseTest(object):

    """

    Write all tests here

    """

    REPO = None

    REPO_TYPE = None

    NEW_REPO = None

    OTHER_TYPE_REPO = None

    OTHER_TYPE = None

    @classmethod

    def setup_class(cls):

        pass

    @classmethod

    def teardown_class(cls):

        pass

    def test_index(self):

        self.log_user()

        response = self.app.get(url('repos'))

    def test_create(self):

        self.log_user()

        repo_name = self.NEW_REPO

        description = 'description for newly created repo'

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

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

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

        self.assertEqual(response.json, {u'result': True})

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

        self.assertEqual(new_repo.repo_name, repo_name)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(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:

        except vcs.exceptions.VCSError:

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

        RepoModel().delete(repo_name)

        Session().commit()

    def test_create_non_ascii(self):

        self.log_user()

        non_ascii = "ąęł"

        repo_name = "%s%s" % (self.NEW_REPO, non_ascii)

        repo_name_unicode = repo_name.decode('utf8')

        description = 'description for newly created repo' + non_ascii

        description_unicode = description.decode('utf8')

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

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

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

        self.assertEqual(response.json, {u'result': True})

        self.checkSessionFlash(response,

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

                               % (urllib.quote(repo_name), repo_name_unicode))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

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

        self.assertEqual(new_repo.repo_name, repo_name_unicode)

        self.assertEqual(new_repo.description, description_unicode)

        # test if the repository is visible in the list ?

        response = self.app.get(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:

        except vcs.exceptions.VCSError:

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

    def test_create_in_group(self):

        self.log_user()

        ## create GROUP

        group_name = 'sometest_%s' % self.REPO_TYPE

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

                                     group_description='test',

                                     owner=TEST_USER_ADMIN_LOGIN)

        Session().commit()

        repo_name = 'ingroup'

        repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])

        description = 'description for newly created repo'

        response = self.app.post(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,

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

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

        self.assertEqual(response.json, {u'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

        self.assertEqual(new_repo.repo_name, repo_name_full)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(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()

        self.assertEqual(len(inherited_perms), 1)

        # test if the repository was created on filesystem

        try:

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

            self.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(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)

        # avoid spurious RepoGroup DetachedInstanceError ...

        authentication_token = self.authentication_token()

        # revoke

        user_model = UserModel()

        # disable fork and create on default user

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

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

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

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

        # disable on regular user

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

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

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

        user_model.grant_perm(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=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=TEST_USER_REGULAR_LOGIN)

        Session().commit()

        repo_name = 'ingroup'

        repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])

        description = 'description for newly created repo'

        response = self.app.post(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,

                                                _authentication_token=authentication_token))

        response.mustcontain('Invalid value')

        # user is allowed to create in this group

        repo_name = 'ingroup'

        repo_name_full = RepoGroup.url_sep().join([group_name_allowed, repo_name])

        description = 'description for newly created repo'

        response = self.app.post(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,

                                                _authentication_token=authentication_token))

        ## run the check page that triggers the flash message

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

        self.assertEqual(response.json, {u'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

        self.assertEqual(new_repo.repo_name, repo_name_full)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(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()

        self.assertEqual(len(inherited_perms), 1)

        # test if the repository was created on filesystem

        try:

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

            self.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=TEST_USER_ADMIN_LOGIN)

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

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

        ## add repo permissions

        Session().commit()

        repo_name = 'ingroup_inherited_%s' % self.REPO_TYPE

        repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])

        description = 'description for newly created repo'

        response = self.app.post(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,

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

        response = self.app.get(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

        self.assertEqual(new_repo.repo_name, repo_name_full)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(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:

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

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

        self.assertEqual(len(inherited_perms), 2)

        self.assertTrue(TEST_USER_REGULAR_LOGIN in [x.user.username

                                                    for x in inherited_perms])

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

                                                _authentication_token=self.authentication_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(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',

                                                _authentication_token=self.authentication_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(url('repos'),

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_type=self.REPO_TYPE,

                                                repo_name=repo_name,

                                                repo_description=description,

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

        response = self.app.get(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()

        self.assertEqual(new_repo.repo_name, repo_name)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(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:

        except vcs.exceptions.VCSError:

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

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

            params={'_method': 'delete', '_authentication_token': self.authentication_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()

        self.assertEqual(deleted_repo, None)

        self.assertEqual(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)),

                                  False)

    def test_delete_non_ascii(self):

        self.log_user()

        non_ascii = "ąęł"

        repo_name = "%s%s" % (self.NEW_REPO, non_ascii)

        repo_name_unicode = repo_name.decode('utf8')

        description = 'description for newly created repo' + non_ascii

        description_unicode = description.decode('utf8')

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

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

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

        self.assertEqual(response.json, {u'result': True})

        self.checkSessionFlash(response,

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

                               % (urllib.quote(repo_name), repo_name_unicode))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

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

        self.assertEqual(new_repo.repo_name, repo_name_unicode)

        self.assertEqual(new_repo.description, description_unicode)

        # test if the repository is visible in the list ?

        response = self.app.get(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:

        except vcs.exceptions.VCSError:

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

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

            params={'_method': 'delete', '_authentication_token': self.authentication_token()})

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

        response.follow()

        #check if repo was deleted from db

        deleted_repo = Session().query(Repository) \

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

        self.assertEqual(deleted_repo, None)

        self.assertEqual(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)),

                                  False)

    def test_delete_repo_with_group(self):

        #TODO:

        pass

    def test_delete_browser_fakeout(self):

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

                                 params=dict(_method='delete', _authentication_token=self.authentication_token()))

    def test_show(self):

        self.log_user()

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

    def test_edit(self):

        response = self.app.get(url('edit_repo', repo_name=self.REPO))

    def test_set_private_flag_sets_default_to_none(self):

        self.log_user()

        #initially repository perm should be read

        perm = _get_permission_for_user(user='default', repo=self.REPO)

        self.assertTrue(len(perm), 1)

        self.assertEqual(perm[0].permission.permission_name, 'repository.read')

        self.assertEqual(Repository.get_by_repo_name(self.REPO).private, False)

        response = self.app.put(url('put_repo', repo_name=self.REPO),

                        fixture._get_repo_create_params(repo_private=1,

                                                repo_name=self.REPO,

                                                repo_type=self.REPO_TYPE,

                                                user=TEST_USER_ADMIN_LOGIN,

                                                _authentication_token=self.authentication_token()))

        self.checkSessionFlash(response,

                               msg='Repository %s updated successfully' % (self.REPO))

        self.assertEqual(Repository.get_by_repo_name(self.REPO).private, True)

        #now the repo default permission should be None

        perm = _get_permission_for_user(user='default', repo=self.REPO)

        self.assertTrue(len(perm), 1)

        self.assertEqual(perm[0].permission.permission_name, 'repository.none')

        response = self.app.put(url('put_repo', repo_name=self.REPO),

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=self.REPO,

                                                repo_type=self.REPO_TYPE,

                                                user=TEST_USER_ADMIN_LOGIN,

                                                _authentication_token=self.authentication_token()))

        self.checkSessionFlash(response,

                               msg='Repository %s updated successfully' % (self.REPO))

        self.assertEqual(Repository.get_by_repo_name(self.REPO).private, False)

        #we turn off private now the repo default permission should stay None

        perm = _get_permission_for_user(user='default', repo=self.REPO)

        self.assertTrue(len(perm), 1)

        self.assertEqual(perm[0].permission.permission_name, 'repository.none')

        #update this permission back

        perm[0].permission = Permission.get_by_key('repository.read')

        Session().add(perm[0])

        Session().commit()

    def test_set_repo_fork_has_no_self_id(self):

        self.log_user()

        repo = Repository.get_by_repo_name(self.REPO)

        response = self.app.get(url('edit_repo_advanced', repo_name=self.REPO))

        opt = """<option value="%s">%s</option>""" % (repo.repo_id, self.REPO)

        response.mustcontain(no=[opt])

    def test_set_fork_of_other_repo(self):

        self.log_user()

        other_repo = 'other_%s' % self.REPO_TYPE

        fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)

        repo = Repository.get_by_repo_name(self.REPO)

        repo2 = Repository.get_by_repo_name(other_repo)

        response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),

                                params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))

        repo = Repository.get_by_repo_name(self.REPO)

        repo2 = Repository.get_by_repo_name(other_repo)

        self.checkSessionFlash(response,

            'Marked repository %s as fork of %s' % (repo.repo_name, repo2.repo_name))

        assert repo.fork == repo2

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

# Use only clean mercurial's ui

from kallithea.lib.vcs.utils.hgcompat import mercurial

mercurial.scmutil.rcpath()

if mercurial.scmutil._rcpath:

    mercurial.scmutil._rcpath = mercurial.scmutil._rcpath[:1]

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

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

        org = ['b986218ba1c9b0d6a259fac9b050b1724ed8e545',

                '3d8f361e72ab303da48d799ff1ac40d5ac37c67e',

                '6cba7170863a2411822803fa77a0a264f1310b35',

                '56349e29c2af3ac913b28bde9a2c6154436e615b',

                '2dda4e345facb0ccff1a191052dd1606dba6781d',

                '6fff84722075f1607a30f436523403845f84cd9e',

                '7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7',

                '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb',

                'dc5d2c0661b61928834a785d3e64a3f80d3aad9c',

                'be90031137367893f1c406e0a8683010fd115b79',

                'db8e58be770518cbb2b1cdfa69146e47cd481481',

                '84478366594b424af694a6c784cb991a16b87c21',

                '17f8e105dddb9f339600389c6dc7175d395a535c',

                '20a662e756499bde3095ffc9bc0643d1def2d0eb',

                '2e319b85e70a707bba0beff866d9f9de032aa4f9',

                '786facd2c61deb9cf91e9534735124fb8fc11842',

                '94593d2128d38210a2fcd1aabff6dda0d6d9edf8',

                'aa6a0de05b7612707db567078e130a6cd114a9a7',

                'eada5a770da98ab0dd7325e29d00e0714f228d09',

                '2c1885c735575ca478bf9e17b0029dca68824458',

                'd9bcd465040bf869799b09ad732c04e0eea99fe9',

                '469e9c847fe1f6f7a697b8b25b4bc5b48780c1a7',

                '4fb8326d78e5120da2c7468dcf7098997be385da',

                '62b4a097164940bd66030c4db51687f3ec035eed',

                '536c1a19428381cfea92ac44985304f6a8049569',

                '965e8ab3c44b070cdaa5bf727ddef0ada980ecc4',

                '9bb326a04ae5d98d437dece54be04f830cf1edd9',

                'f8940bcb890a98c4702319fbe36db75ea309b475',

                'ff5ab059786ebc7411e559a2cc309dfae3625a3b',

                '6b6ad5f82ad5bb6190037671bd254bd4e1f4bf08',

                'ee87846a61c12153b51543bf860e1026c6d3dcba', ]

        self.assertEqual(org, self.repo.revisions[:31])

    def test_iter_slice(self):

        sliced = list(self.repo[:10])

        itered = list(self.repo)[:10]

        self.assertEqual(sliced, itered)

    def test_slicing(self):

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

class MercurialChangesetTest(unittest.TestCase):

    def setUp(self):

    def _test_equality(self, changeset):

        revision = changeset.revision

        self.assertEqual(changeset, self.repo.get_changeset(revision))

    def test_equality(self):

        self.setUp()

        revs = [0, 10, 20]

        changesets = [self.repo.get_changeset(rev) for rev in revs]

        for changeset in changesets:

            self._test_equality(changeset)

    def test_default_changeset(self):

        tip = self.repo.get_changeset('tip')

        self.assertEqual(tip, self.repo.get_changeset())

        self.assertEqual(tip, self.repo.get_changeset(revision=None))

        self.assertEqual(tip, list(self.repo[-1:])[0])

    def test_root_node(self):

        tip = self.repo.get_changeset('tip')

        self.assertTrue(tip.root is tip.get_node(''))

    def test_lazy_fetch(self):

        """

        Test if changeset's nodes expands and are cached as we walk through

        the revision. This test is somewhat hard to write as order of tests

        is a key here. Written by running command after command in a shell.

        """

        self.setUp()

        chset = self.repo.get_changeset(45)

        self.assertTrue(len(chset.nodes) == 0)

        root = chset.root

        self.assertTrue(len(chset.nodes) == 1)

        self.assertTrue(len(root.nodes) == 8)

        # accessing root.nodes updates chset.nodes

        self.assertTrue(len(chset.nodes) == 9)

        docs = root.get_node('docs')

        # we haven't yet accessed anything new as docs dir was already cached

        self.assertTrue(len(chset.nodes) == 9)

        self.assertTrue(len(docs.nodes) == 8)

        # accessing docs.nodes updates chset.nodes

        self.assertTrue(len(chset.nodes) == 17)

        self.assertTrue(docs is chset.get_node('docs'))

        self.assertTrue(docs is root.nodes[0])

        self.assertTrue(docs is root.dirs[0])

        self.assertTrue(docs is chset.get_node('docs'))

    def test_nodes_with_changeset(self):

        self.setUp()

        chset = self.repo.get_changeset(45)

        root = chset.root

        docs = root.get_node('docs')

        self.assertTrue(docs is chset.get_node('docs'))

        api = docs.get_node('api')

        self.assertTrue(api is chset.get_node('docs/api'))

        index = api.get_node('index.rst')

        self.assertTrue(index is chset.get_node('docs/api/index.rst'))

        self.assertTrue(index is chset.get_node('docs') \

            .get_node('api') \

            .get_node('index.rst'))

    def test_branch_and_tags(self):

        chset0 = self.repo.get_changeset(0)