"""

Unit tests configuration module for vcs.

"""

import os

import tempfile

import shutil

import uuid

__all__ = (

    'TEST_HG_REPO', 'TEST_GIT_REPO', 'HG_REMOTE_REPO', 'GIT_REMOTE_REPO',

    'SCM_TESTS',

)

SCM_TESTS = ['hg', 'git']

THIS = os.path.abspath(os.path.dirname(__file__))

GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'

# Note: TEST_TMP_PATH, not TESTS_TMP_PATH

TEST_TMP_PATH = os.environ.get('VCS_TEST_ROOT', tempfile.gettempdir())

TEST_GIT_REPO = os.environ.get('VCS_TEST_GIT_REPO',

                               os.path.join(TEST_TMP_PATH, 'vcs-git'))

TEST_GIT_REPO_CLONE = os.environ.get('VCS_TEST_GIT_REPO_CLONE',

                                     os.path.join(TEST_TMP_PATH, 'vcs-git-clone'))

TEST_GIT_REPO_PULL = os.environ.get('VCS_TEST_GIT_REPO_PULL',

                                    os.path.join(TEST_TMP_PATH, 'vcs-git-pull'))

HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'

TEST_HG_REPO = os.environ.get('VCS_TEST_HG_REPO',

                              os.path.join(TEST_TMP_PATH, 'vcs-hg'))

TEST_HG_REPO_CLONE = os.environ.get('VCS_TEST_HG_REPO_CLONE',

                                    os.path.join(TEST_TMP_PATH, 'vcs-hg-clone'))

TEST_HG_REPO_PULL = os.environ.get('VCS_TEST_HG_REPO_PULL',

                                   os.path.join(TEST_TMP_PATH, 'vcs-hg-pull'))

TEST_REPO_PREFIX = 'vcs-test'

def get_new_dir(title=None):

    """

    Calculates a path for a new, non-existant, unique sub-directory in TEST_TMP_PATH.

    Resulting directory name will have format:

    prefix-[title-]hexuuid

    Prefix is equal to value of variable TEST_REPO_PREFIX. The "hexuuid" is a

    hexadecimal value of a randomly generated UUID. Title will be added if

    specified.

    Args:

        title: Custom title to include as part of the resulting sub-directory

            name. Can be useful for debugging to identify destination. Defaults

            to None.

    Returns:

        Path to the new directory as a string.

    """

    if title:

        name = "%s-%s" % (TEST_REPO_PREFIX, title)

    else:

        name = TEST_REPO_PREFIX

    path = os.path.join(TEST_TMP_PATH, name)

    # Generate new hexes until we get a unique name (just in case).

    hex_uuid = uuid.uuid4().hex

    while os.path.exists("%s-%s" % (path, hex_uuid)):

        hex_uuid = uuid.uuid4().hex

    return "%s-%s" % (path, hex_uuid)

PACKAGE_DIR = os.path.abspath(os.path.join(

    os.path.dirname(__file__), '..'))

_dest = os.path.join(TEST_TMP_PATH, 'aconfig')

shutil.copy(os.path.join(THIS, 'aconfig'), _dest)

TEST_USER_CONFIG_FILE = _dest

# overide default configurations with kallithea ones

from kallithea.tests.base import *

import os

import sys

import mock

import datetime

import urllib2

import pytest

from kallithea.lib.vcs.backends.git import GitRepository, GitChangeset

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

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

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

from kallithea.model.scm import ScmModel

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.tests.vcs.conf import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, TEST_TMP_PATH, get_new_dir

class GitRepositoryTest(unittest.TestCase):

    def __check_for_existing_repo(self):

        if os.path.exists(TEST_GIT_REPO_CLONE):

            pytest.fail('Cannot test git clone repo as location %s already '

                      'exists. You should manually remove it first.'

                      % TEST_GIT_REPO_CLONE)

    def setUp(self):

        self.repo = GitRepository(TEST_GIT_REPO)

    def test_wrong_repo_path(self):

        wrong_repo_path = os.path.join(TEST_TMP_PATH, 'errorrepo')

        self.assertRaises(RepositoryError, GitRepository, wrong_repo_path)

    def test_git_cmd_injection(self):

        repo_inject_path = TEST_GIT_REPO + '; echo "Cake";'

        with self.assertRaises(urllib2.URLError):

            # Should fail because URL will contain the parts after ; too

        with self.assertRaises(RepositoryError):

            # Should fail on direct clone call, which as of this writing does not happen outside of class

            clone_fail_repo = GitRepository(get_new_dir('injection-repo'), create=True)

            clone_fail_repo.clone(repo_inject_path, update_after_clone=True,)

        # Verify correct quoting of evil characters that should work on posix file systems

        if sys.platform == 'win32':

            # windows does not allow '"' in dir names

            # and some versions of the git client don't like ` and '

            tricky_path = get_new_dir("tricky-path-repo-$")

        else:

            tricky_path = get_new_dir("tricky-path-repo-$'\"`")

        successfully_cloned = GitRepository(tricky_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)

        # Repo should have been created

        self.assertFalse(successfully_cloned._repo.bare)

        if sys.platform == 'win32':

            # windows does not allow '"' in dir names

            # and some versions of the git client don't like ` and '

            tricky_path_2 = get_new_dir("tricky-path-2-repo-$")

        else:

            tricky_path_2 = get_new_dir("tricky-path-2-repo-$'\"`")

        successfully_cloned2 = GitRepository(tricky_path_2, src_url=tricky_path, bare=True, create=True)

        # Repo should have been created and thus used correct quoting for clone

        self.assertTrue(successfully_cloned2._repo.bare)

        # Should pass because URL has been properly quoted

        successfully_cloned.pull(tricky_path_2)

        successfully_cloned2.fetch(tricky_path)

    def test_repo_create_with_spaces_in_path(self):

        repo_path = get_new_dir("path with spaces")

        repo = GitRepository(repo_path, src_url=None, bare=True, create=True)

        # Repo should have been created

        self.assertTrue(repo._repo.bare)

    def test_repo_clone(self):

        self.__check_for_existing_repo()

        repo = GitRepository(TEST_GIT_REPO)

        repo_clone = GitRepository(TEST_GIT_REPO_CLONE,

            src_url=TEST_GIT_REPO, create=True, 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_spaces_in_path(self):

        repo_path = get_new_dir("path with spaces")

        successfully_cloned = GitRepository(repo_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)

        # Repo should have been created

        self.assertFalse(successfully_cloned._repo.bare)

        successfully_cloned.pull(TEST_GIT_REPO)

        self.repo.fetch(repo_path)

    def test_repo_clone_without_create(self):

        self.assertRaises(RepositoryError, GitRepository,

            TEST_GIT_REPO_CLONE + '_wo_create', src_url=TEST_GIT_REPO)

    def test_repo_clone_with_update(self):

        repo = GitRepository(TEST_GIT_REPO)

        clone_path = TEST_GIT_REPO_CLONE + '_with_update'

        repo_clone = GitRepository(clone_path,

            create=True, src_url=TEST_GIT_REPO, update_after_clone=True)

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

        # check if current workdir was updated

        fpath = os.path.join(clone_path, 'MANIFEST.in')

        self.assertEqual(True, os.path.isfile(fpath),

            'Repo was cloned and updated but file %s could not be found'

            % fpath)

    def test_repo_clone_without_update(self):

        repo = GitRepository(TEST_GIT_REPO)

        clone_path = TEST_GIT_REPO_CLONE + '_without_update'

        repo_clone = GitRepository(clone_path,

            create=True, src_url=TEST_GIT_REPO, update_after_clone=False)

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

        # check if current workdir was *NOT* updated

        fpath = os.path.join(clone_path, 'MANIFEST.in')

        # Make sure it's not bare repo

        self.assertFalse(repo_clone._repo.bare)

        self.assertEqual(False, os.path.isfile(fpath),

            'Repo was cloned and updated but file %s was found'

            % fpath)

    def test_repo_clone_into_bare_repo(self):

        repo = GitRepository(TEST_GIT_REPO)

        clone_path = TEST_GIT_REPO_CLONE + '_bare.git'

        repo_clone = GitRepository(clone_path, create=True,

            src_url=repo.path, bare=True)

        self.assertTrue(repo_clone._repo.bare)

    def test_create_repo_is_not_bare_by_default(self):

        repo = GitRepository(get_new_dir('not-bare-by-default'), create=True)

        self.assertFalse(repo._repo.bare)

    def test_create_bare_repo(self):

        repo = GitRepository(get_new_dir('bare-repo'), create=True, bare=True)

        self.assertTrue(repo._repo.bare)

    def test_revisions(self):

        # there are 112 revisions (by now)

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

        subset = set([

            'c1214f7e79e02fc37156ff215cd71275450cffc3',

            '38b5fe81f109cb111f549bfe9bb6b267e10bc557',

            'fa6600f6848800641328adbf7811fd2372c02ab2',

            '102607b09cdd60e2793929c4f90478be29f85a17',

            '49d3fd156b6f7db46313fac355dca1a0b94a0017',

            '2d1028c054665b962fa3d307adfc923ddd528038',

            'd7e0d30fbcae12c90680eb095a4f5f02505ce501',

            'ff7ca51e58c505fec0dd2491de52c622bb7a806b',

            'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',

            '8430a588b43b5d6da365400117c89400326e7992',

            'd955cd312c17b02143c04fa1099a352b04368118',

            'f67b87e5c629c2ee0ba58f85197e423ff28d735b',

            'add63e382e4aabc9e1afdc4bdc24506c269b7618',

            'f298fe1189f1b69779a4423f40b48edf92a703fc',

            'bd9b619eb41994cac43d67cf4ccc8399c1125808',

            '6e125e7c890379446e98980d8ed60fba87d0f6d1',

            'd4a54db9f745dfeba6933bf5b1e79e15d0af20bd',

            '0b05e4ed56c802098dfc813cbe779b2f49e92500',

            '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',

            '45223f8f114c64bf4d6f853e3c35a369a6305520',

            'ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',

            'f5ea29fc42ef67a2a5a7aecff10e1566699acd68',

            '27d48942240f5b91dfda77accd2caac94708cc7d',

            '622f0eb0bafd619d2560c26f80f09e3b0b0d78af',

            'e686b958768ee96af8029fe19c6050b1a8dd3b2b'])

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

    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

        # Removed (those are 'remotes' branches for cloned repo)

        #self.assertTrue('master' in self.repo.branches)

        #self.assertTrue('gittree' in self.repo.branches)

        #self.assertTrue('web-branch' in self.repo.branches)

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

            self.assertTrue(isinstance(

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

    def test_tags(self):

        # TODO: Need more tests here

        self.assertTrue('v0.1.1' in self.repo.tags)

        self.assertTrue('v0.1.2' in self.repo.tags)

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

            self.assertTrue(isinstance(

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

    def _test_single_changeset_cache(self, revision):

        chset = self.repo.get_changeset(revision)

        self.assertTrue(revision in self.repo.changesets)

        self.assertTrue(chset is self.repo.changesets[revision])

    def test_initial_changeset(self):

        id = self.repo.revisions[0]

        init_chset = self.repo.get_changeset(id)

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

        self.assertEqual(init_chset.author,

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

        for path in ('vcs/__init__.py',

                     'vcs/backends/BaseRepository.py',

                     'vcs/backends/__init__.py'):

            self.assertTrue(isinstance(init_chset.get_node(path), FileNode))

        for path in ('', 'vcs', 'vcs/backends'):

            self.assertTrue(isinstance(init_chset.get_node(path), DirNode))

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

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

            'f' * 40)

    def test_changeset10(self):

        chset10 = self.repo.get_changeset(self.repo.revisions[9])

        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 GitChangesetTest(unittest.TestCase):

    def setUp(self):

        self.repo = GitRepository(TEST_GIT_REPO)

    def test_default_changeset(self):

        tip = self.repo.get_changeset()

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

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

    def test_root_node(self):

        tip = self.repo.get_changeset()

        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.

        """

        commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'

        self.assertTrue(commit_id in self.repo.revisions)

        chset = self.repo.get_changeset(commit_id)

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

        commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'

        chset = self.repo.get_changeset(commit_id)

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

        """

        rev0 = self.repo.revisions[0]

        chset0 = self.repo.get_changeset(rev0)

        self.assertEqual(chset0.branch, 'master')

        self.assertEqual(chset0.tags, [])

        rev10 = self.repo.revisions[10]

        chset10 = self.repo.get_changeset(rev10)

        self.assertEqual(chset10.branch, 'master')

        self.assertEqual(chset10.tags, [])

        rev44 = self.repo.revisions[44]

        chset44 = self.repo.get_changeset(rev44)

        self.assertEqual(chset44.branch, 'web-branch')

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

        self.assertTrue('tip' in tip.tags)

        """

        # Those tests would fail - branches are now going

        # to be changed at main API in order to support git backend

        pass

    def _test_slices(self, limit, offset):

        count = self.repo.count()

        changesets = self.repo.get_changesets(limit=limit, offset=offset)

        idx = 0

        for changeset in changesets:

            rev = offset + idx

            idx += 1

            rev_id = self.repo.revisions[rev]

            if idx > limit:

                pytest.fail("Exceeded limit already (getting revision %s, "

                    "there are %s total revisions, offset=%s, limit=%s)"

                    % (rev_id, count, offset, limit))

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

        result = list(self.repo.get_changesets(limit=limit, offset=offset))

        start = offset

        end = limit and offset + limit or None

        sliced = list(self.repo[start:end])

        pytest.failUnlessEqual(result, sliced,

            msg="Comparison failed for limit=%s, offset=%s"

            "(get_changeset returned: %s and sliced: %s"

            % (limit, offset, result, sliced))

    def _test_file_size(self, revision, path, size):

        node = self.repo.get_changeset(revision).get_node(path)

        self.assertTrue(node.is_file())

        self.assertEqual(node.size, size)

    def test_file_size(self):

        to_check = (

            ('c1214f7e79e02fc37156ff215cd71275450cffc3',

                'vcs/backends/BaseRepository.py', 502),

            ('d7e0d30fbcae12c90680eb095a4f5f02505ce501',

                'vcs/backends/hg.py', 854),

            ('6e125e7c890379446e98980d8ed60fba87d0f6d1',

                'setup.py', 1068),

            ('d955cd312c17b02143c04fa1099a352b04368118',

                'vcs/backends/base.py', 2921),

            ('ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',

                'vcs/backends/base.py', 3936),

            ('f50f42baeed5af6518ef4b0cb2f1423f3851a941',

                'vcs/backends/base.py', 6189),

        )

        for revision, path, size in to_check:

            self._test_file_size(revision, path, size)

    def _test_dir_size(self, revision, path, size):

        node = self.repo.get_changeset(revision).get_node(path)

        self.assertEqual(node.size, size)

    def test_dir_size(self):

        to_check = (

            ('5f2c6ee195929b0be80749243c18121c9864a3b3', '/', 674076),

            ('7ab37bc680b4aa72c34d07b230c866c28e9fc204', '/', 674049),

            ('6892503fb8f2a552cef5f4d4cc2cdbd13ae1cd2f', '/', 671830),

        )

        for revision, path, size in to_check:

            self._test_dir_size(revision, path, size)

    def test_repo_size(self):

        self.assertEqual(self.repo.size, 674076)

    def test_file_history(self):

        # we can only check if those revisions are present in the history

        # as we cannot update this test every time file is changed

        files = {

            'setup.py': [

                '54386793436c938cff89326944d4c2702340037d',

                '51d254f0ecf5df2ce50c0b115741f4cf13985dab',

                '998ed409c795fec2012b1c0ca054d99888b22090',

                '5e0eb4c47f56564395f76333f319d26c79e2fb09',

                '0115510b70c7229dbc5dc49036b32e7d91d23acd',

                '7cb3fd1b6d8c20ba89e2264f1c8baebc8a52d36e',

                '2a13f185e4525f9d4b59882791a2d397b90d5ddc',

                '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',

                'ff7ca51e58c505fec0dd2491de52c622bb7a806b',

            ],

            'vcs/nodes.py': [

                '33fa3223355104431402a888fa77a4e9956feb3e',

                'fa014c12c26d10ba682fadb78f2a11c24c8118e1',

                'e686b958768ee96af8029fe19c6050b1a8dd3b2b',

                'ab5721ca0a081f26bf43d9051e615af2cc99952f',

                'c877b68d18e792a66b7f4c529ea02c8f80801542',

                '4313566d2e417cb382948f8d9d7c765330356054',

                '6c2303a793671e807d1cfc70134c9ca0767d98c2',

                '54386793436c938cff89326944d4c2702340037d',

                '54000345d2e78b03a99d561399e8e548de3f3203',

                '1c6b3677b37ea064cb4b51714d8f7498f93f4b2b',

                '2d03ca750a44440fb5ea8b751176d1f36f8e8f46',

# encoding: utf8

"""

Tests so called "in memory changesets" commit API of vcs.

"""

import time

import datetime

from kallithea.lib import vcs

from kallithea.tests.vcs.conf import SCM_TESTS, get_new_dir

from kallithea.lib.vcs.exceptions import EmptyRepositoryError

from kallithea.lib.vcs.exceptions import NodeAlreadyAddedError

from kallithea.lib.vcs.exceptions import NodeAlreadyExistsError

from kallithea.lib.vcs.exceptions import NodeAlreadyRemovedError

from kallithea.lib.vcs.exceptions import NodeAlreadyChangedError

from kallithea.lib.vcs.exceptions import NodeDoesNotExistError

from kallithea.lib.vcs.exceptions import NodeNotChangedError

from kallithea.lib.vcs.nodes import DirNode

from kallithea.lib.vcs.nodes import FileNode

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

from kallithea.lib.vcs.utils import safe_unicode

class InMemoryChangesetTestMixin(object):

    """

    This is a backend independent test case class which should be created

    with ``type`` method.

    It is required to set following attributes at subclass:

    - ``backend_alias``: alias of used backend (see ``vcs.BACKENDS``)

    - ``repo_path``: path to the repository which would be created for set of

      tests

    """

    def get_backend(self):

        return vcs.get_backend(self.backend_alias)

    def setUp(self):

        Backend = self.get_backend()

        self.repo_path = get_new_dir(str(time.time()))

        self.repo = Backend(self.repo_path, create=True)

        self.imc = self.repo.in_memory_changeset

        self.nodes = [

            FileNode('foobar', content='Foo & bar'),

            FileNode('foobar2', content='Foo & bar, doubled!'),

            FileNode('foo bar with spaces', content=''),

            FileNode('foo/bar/baz', content='Inside'),

            FileNode('foo/bar/file.bin', content='\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00;\x00\x03\x00\xfe\xff\t\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x1a\x00\x00\x00\x00\x00\x00\x00\x00\x10\x00\x00\x18\x00\x00\x00\x01\x00\x00\x00\xfe\xff\xff\xff\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff'),

        ]

    def test_add(self):

        rev_count = len(self.repo.revisions)

        to_add = [FileNode(node.path, content=node.content)

            for node in self.nodes]

        for node in to_add:

            self.imc.add(node)

        message = u'Added: %s' % ', '.join((node.path for node in self.nodes))

        author = unicode(self.__class__)

        changeset = self.imc.commit(message=message, author=author)

        newtip = self.repo.get_changeset()

        self.assertEqual(changeset, newtip)

        self.assertEqual(rev_count + 1, len(self.repo.revisions))

        self.assertEqual(newtip.message, message)

        self.assertEqual(newtip.author, author)

        self.assertTrue(not any((self.imc.added, self.imc.changed,

            self.imc.removed)))

        for node in to_add:

            self.assertEqual(newtip.get_node(node.path).content, node.content)

    def test_add_in_bulk(self):

        rev_count = len(self.repo.revisions)

        to_add = [FileNode(node.path, content=node.content)

            for node in self.nodes]

        self.imc.add(*to_add)

        message = u'Added: %s' % ', '.join((node.path for node in self.nodes))

        author = unicode(self.__class__)

        changeset = self.imc.commit(message=message, author=author)

        newtip = self.repo.get_changeset()

        self.assertEqual(changeset, newtip)

        self.assertEqual(rev_count + 1, len(self.repo.revisions))

        self.assertEqual(newtip.message, message)

        self.assertEqual(newtip.author, author)

        self.assertTrue(not any((self.imc.added, self.imc.changed,

            self.imc.removed)))

        for node in to_add:

            self.assertEqual(newtip.get_node(node.path).content, node.content)

    def test_add_actually_adds_all_nodes_at_second_commit_too(self):

        self.imc.add(FileNode('foo/bar/image.png', content='\0'))

        self.imc.add(FileNode('foo/README.txt', content='readme!'))

        changeset = self.imc.commit(u'Initial', u'joe.doe@example.com')

        self.assertTrue(isinstance(changeset.get_node('foo'), DirNode))

        self.assertTrue(isinstance(changeset.get_node('foo/bar'), DirNode))

        self.assertEqual(changeset.get_node('foo/bar/image.png').content, '\0')

        self.assertEqual(changeset.get_node('foo/README.txt').content, 'readme!')

        # commit some more files again

        to_add = [

            FileNode('foo/bar/foobaz/bar', content='foo'),

            FileNode('foo/bar/another/bar', content='foo'),

            FileNode('foo/baz.txt', content='foo'),

            FileNode('foobar/foobaz/file', content='foo'),

            FileNode('foobar/barbaz', content='foo'),

        ]

        self.imc.add(*to_add)

        changeset = self.imc.commit(u'Another', u'joe.doe@example.com')

        self.assertEqual(changeset.get_node('foo/bar/foobaz/bar').content, 'foo')

        self.assertEqual(changeset.get_node('foo/bar/another/bar').content, 'foo')

        self.assertEqual(changeset.get_node('foo/baz.txt').content, 'foo')

        self.assertEqual(changeset.get_node('foobar/foobaz/file').content, 'foo')

        self.assertEqual(changeset.get_node('foobar/barbaz').content, 'foo')

    def test_add_non_ascii_files(self):

        rev_count = len(self.repo.revisions)

        to_add = [

            FileNode('żółwik/zwierzątko', content='ćććć'),

            FileNode(u'żółwik/zwierzątko_uni', content=u'ćććć'),

        ]

        for node in to_add:

            self.imc.add(node)

        message = u'Added: %s' % ', '.join((node.path for node in self.nodes))

        author = unicode(self.__class__)

        changeset = self.imc.commit(message=message, author=author)

        newtip = self.repo.get_changeset()

        self.assertEqual(changeset, newtip)

        self.assertEqual(rev_count + 1, len(self.repo.revisions))

        self.assertEqual(newtip.message, message)

        self.assertEqual(newtip.author, author)

        self.assertTrue(not any((self.imc.added, self.imc.changed,

            self.imc.removed)))

        for node in to_add:

            self.assertEqual(newtip.get_node(node.path).content, node.content)

    def test_add_raise_already_added(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        self.assertRaises(NodeAlreadyAddedError, self.imc.add, node)

    def test_check_integrity_raise_already_exist(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        self.imc.commit(message=u'Added foobar', author=unicode(self))

        self.imc.add(node)

        self.assertRaises(NodeAlreadyExistsError, self.imc.commit,

            message='new message',

            author=str(self))

    def test_change(self):

        self.imc.add(FileNode('foo/bar/baz', content='foo'))

        self.imc.add(FileNode('foo/fbar', content='foobar'))

        tip = self.imc.commit(u'Initial', u'joe.doe@example.com')

        # Change node's content

        node = FileNode('foo/bar/baz', content='My **changed** content')

        self.imc.change(node)

        self.imc.commit(u'Changed %s' % node.path, u'joe.doe@example.com')

        newtip = self.repo.get_changeset()

        self.assertNotEqual(tip, newtip)

        self.assertNotEqual(tip.id, newtip.id)

        self.assertEqual(newtip.get_node('foo/bar/baz').content,

                        'My **changed** content')

    def test_change_non_ascii(self):

        to_add = [

            FileNode('żółwik/zwierzątko', content='ćććć'),

            FileNode(u'żółwik/zwierzątko_uni', content=u'ćććć'),

        ]

        for node in to_add:

            self.imc.add(node)

        tip = self.imc.commit(u'Initial', u'joe.doe@example.com')

        # Change node's content

        node = FileNode('żółwik/zwierzątko', content='My **changed** content')

        self.imc.change(node)

        self.imc.commit(u'Changed %s' % safe_unicode(node.path),

                        u'joe.doe@example.com')

        node = FileNode(u'żółwik/zwierzątko_uni', content=u'My **changed** content')

        self.imc.change(node)

        self.imc.commit(u'Changed %s' % safe_unicode(node.path),

                        u'joe.doe@example.com')

        newtip = self.repo.get_changeset()

        self.assertNotEqual(tip, newtip)

        self.assertNotEqual(tip.id, newtip.id)

        self.assertEqual(newtip.get_node('żółwik/zwierzątko').content,

                         'My **changed** content')

        self.assertEqual(newtip.get_node('żółwik/zwierzątko_uni').content,

                         'My **changed** content')

    def test_change_raise_empty_repository(self):

        node = FileNode('foobar')

        self.assertRaises(EmptyRepositoryError, self.imc.change, node)

    def test_check_integrity_change_raise_node_does_not_exist(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        self.imc.commit(message=u'Added foobar', author=unicode(self))

        node = FileNode('not-foobar', content='')

        self.imc.change(node)

        self.assertRaises(NodeDoesNotExistError, self.imc.commit,

            message='Changed not existing node',

            author=str(self))

    def test_change_raise_node_already_changed(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        self.imc.commit(message=u'Added foobar', author=unicode(self))

        node = FileNode('foobar', content='more baz')

        self.imc.change(node)

        self.assertRaises(NodeAlreadyChangedError, self.imc.change, node)

    def test_check_integrity_change_raise_node_not_changed(self):

        self.test_add()  # Performs first commit

        node = FileNode(self.nodes[0].path, content=self.nodes[0].content)

        self.imc.change(node)

        self.assertRaises(NodeNotChangedError, self.imc.commit,

            message=u'Trying to mark node as changed without touching it',

            author=unicode(self))

    def test_change_raise_node_already_removed(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        self.imc.commit(message=u'Added foobar', author=unicode(self))

        self.imc.remove(FileNode('foobar'))

        self.assertRaises(NodeAlreadyRemovedError, self.imc.change, node)

    def test_remove(self):

        self.test_add()  # Performs first commit

        tip = self.repo.get_changeset()

        node = self.nodes[0]

        self.assertEqual(node.content, tip.get_node(node.path).content)

        self.imc.remove(node)

        self.imc.commit(message=u'Removed %s' % node.path, author=unicode(self))

        newtip = self.repo.get_changeset()

        self.assertNotEqual(tip, newtip)

        self.assertNotEqual(tip.id, newtip.id)

        self.assertRaises(NodeDoesNotExistError, newtip.get_node, node.path)

    def test_remove_last_file_from_directory(self):

        node = FileNode('omg/qwe/foo/bar', content='foobar')

        self.imc.add(node)

        self.imc.commit(u'added', u'joe doe')

        self.imc.remove(node)

        tip = self.imc.commit(u'removed', u'joe doe')

        self.assertRaises(NodeDoesNotExistError, tip.get_node, 'omg/qwe/foo/bar')

    def test_remove_raise_node_does_not_exist(self):

        self.imc.remove(self.nodes[0])

        self.assertRaises(NodeDoesNotExistError, self.imc.commit,

            message='Trying to remove node at empty repository',

            author=str(self))

    def test_check_integrity_remove_raise_node_does_not_exist(self):

        self.test_add()  # Performs first commit

        node = FileNode('no-such-file')

        self.imc.remove(node)

        self.assertRaises(NodeDoesNotExistError, self.imc.commit,

            message=u'Trying to remove not existing node',

            author=unicode(self))

    def test_remove_raise_node_already_removed(self):

        self.test_add() # Performs first commit

        node = FileNode(self.nodes[0].path)

        self.imc.remove(node)

        self.assertRaises(NodeAlreadyRemovedError, self.imc.remove, node)

    def test_remove_raise_node_already_changed(self):

        self.test_add()  # Performs first commit

        node = FileNode(self.nodes[0].path, content='Bending time')

        self.imc.change(node)

        self.assertRaises(NodeAlreadyChangedError, self.imc.remove, node)

    def test_reset(self):

        self.imc.add(FileNode('foo', content='bar'))

        #self.imc.change(FileNode('baz', content='new'))

        #self.imc.remove(FileNode('qwe'))

        self.imc.reset()

        self.assertTrue(not any((self.imc.added, self.imc.changed,

            self.imc.removed)))

    def test_multiple_commits(self):

        N = 3  # number of commits to perform

        last = None

        for x in xrange(N):

            fname = 'file%s' % str(x).rjust(5, '0')

            content = 'foobar\n' * x

            node = FileNode(fname, content=content)

            self.imc.add(node)

            commit = self.imc.commit(u"Commit no. %s" % (x + 1), author=u'vcs')

            self.assertTrue(last != commit)

            last = commit

        # Check commit number for same repo

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

        # Check commit number for recreated repo

        backend = self.get_backend()

        repo = backend(self.repo_path)

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

    def test_date_attr(self):

        node = FileNode('foobar.txt', content='Foobared!')

        self.imc.add(node)

        date = datetime.datetime(1985, 1, 30, 1, 45)

        commit = self.imc.commit(u"Committed at time when I was born ;-)",

            author=u'lb <lb@example.com>', date=date)

        self.assertEqual(commit.date, date)

class BackendBaseTestCase(unittest.TestCase):

    """

    Base test class for tests which requires repository.

    """

    backend_alias = 'hg'

    commits = [

        {

            'message': 'Initial commit',

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

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

            'added': [

                FileNode('foobar', content='Foobar'),

                FileNode('foobar2', content='Foobar II'),

                FileNode('foo/bar/baz', content='baz here!'),

            ],

        },

    ]

    def get_backend(self):

        return vcs.get_backend(self.backend_alias)

    def get_commits(self):

        """

        Returns list of commits which builds repository for each tests.

        """

        if hasattr(self, 'commits'):

            return self.commits

    def get_new_repo_path(self):

        """

        Returns newly created repository's directory.

        """

        key = '%s-%s' % (self.backend_alias, str(time.time()))

        repo_path = get_new_dir(key)