"""

Unit tests for vcs_ library.

In order to run tests we need to prepare our environment first. Tests would be

run for each engine listed at ``conf.SCM_TESTS`` - keys are aliases from

``vcs.backends.BACKENDS``.

For each SCM we run tests for, we need some repository. We would use

repositories location from system environment variables or test suite defaults

- see ``conf`` module for more detail. We simply try to check if repository at

certain location exists, if not we would try to fetch them. At ``test_vcs`` or

``test_common`` we run unit tests common for each repository type and for

example specific mercurial tests are located at ``test_hg`` module.

Oh, and tests are run with ``unittest.collector`` wrapped by ``collector``

function at ``tests/__init__.py``.

.. _vcs: http://bitbucket.org/marcinkuzminski/vcs

.. _unittest: http://pypi.python.org/pypi/unittest

"""

from kallithea.tests.vcs.conf import *

from kallithea.tests.vcs.utils import SCMFetcher

from kallithea.tests.base import *

def setup_package():

    """

    Prepares whole package for tests which mainly means it would try to fetch

    test repositories or use already existing ones.

    """

    fetchers = {

        'hg': {

            'alias': 'hg',

            'test_repo_path': TEST_HG_REPO,

            'remote_repo': HG_REMOTE_REPO,

            'clone_cmd': 'hg clone --insecure',

        },

        'git': {

            'alias': 'git',

            'test_repo_path': TEST_GIT_REPO,

            'remote_repo': GIT_REMOTE_REPO,

            'clone_cmd': 'git clone --bare',

        },

    }

    for scm, fetcher_info in fetchers.items():

        fetcher = SCMFetcher(**fetcher_info)

        fetcher.setup()

        ]

        return commits

    @classmethod

    def setUpClass(cls):

        Backend = cls.get_backend()

        cls.backend_class = Backend

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

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

        cls.imc = cls.repo.in_memory_changeset

        cls.default_branch = cls.repo.DEFAULT_BRANCH_NAME

        for commit in cls._get_commits():

            for node in commit.get('added', []):

                cls.imc.add(FileNode(node.path, content=node.content))

            for node in commit.get('changed', []):

                cls.imc.change(FileNode(node.path, content=node.content))

            for node in commit.get('removed', []):

                cls.imc.remove(FileNode(node.path))

            cls.tip = cls.imc.commit(message=unicode(commit['message']),

                                     author=unicode(commit['author']),

                                     date=commit['date'])

    @classmethod

    def tearDownClass(cls):

        if not getattr(cls, 'recreate_repo_per_test', False) and \

            'VCS_REMOVE_TEST_DIRS' in os.environ:

            shutil.rmtree(cls.repo_path)

    def setUp(self):

        if getattr(self, 'recreate_repo_per_test', False):

            self.__class__.setUpClass()

    def tearDown(self):

        if getattr(self, 'recreate_repo_per_test', False) and \

            'VCS_REMOVE_TEST_DIRS' in os.environ:

            shutil.rmtree(self.repo_path)

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    cls_name = ''.join(('%s base backend test' % alias).title().split())

    bases = (_BackendTestMixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

        outdir = tempfile.mkdtemp(dir=TEST_TMP_PATH, prefix='test_archive_tbz2-', suffix='-outdir')

        outfile = tarfile.open(path, 'r|bz2')

        outfile.extractall(outdir)

        for x in xrange(5):

            node_path = '%d/file_%d.txt' % (x, x)

            self.assertEqual(

                open(os.path.join(outdir, 'repo/' + node_path)).read(),

                self.tip.get_node(node_path).content)

    def test_archive_default_stream(self):

        tmppath = tempfile.mkstemp(dir=TEST_TMP_PATH, prefix='test_archive_default_stream-')[1]

        with open(tmppath, 'wb') as stream:

            self.tip.fill_archive(stream=stream)

        mystream = StringIO.StringIO()

        self.tip.fill_archive(stream=mystream)

        mystream.seek(0)

        with open(tmppath, 'rb') as f:

            file_content = f.read()

            stringio_content = mystream.read()

            # the gzip header contains a MTIME header

            # because is takes a little bit of time from one fill_archive call to the next

            # this part may differ so don't include that part in the comparison

            self.assertEqual(file_content[:4], stringio_content[:4])

            self.assertEqual(file_content[8:], stringio_content[8:])

    def test_archive_wrong_kind(self):

        with self.assertRaises(VCSError):

            self.tip.fill_archive(kind='wrong kind')

    def test_archive_empty_prefix(self):

        with self.assertRaises(VCSError):

            self.tip.fill_archive(prefix='')

    def test_archive_prefix_with_leading_slash(self):

        with self.assertRaises(VCSError):

            self.tip.fill_archive(prefix='/any')

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    cls_name = ''.join(('%s archive test' % alias).title().split())

    bases = (ArchivesTestCaseMixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

import datetime

from kallithea.lib import vcs

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

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.tests.vcs.conf import SCM_TESTS

class BranchesTestCaseMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        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!'),

                ],

            },

            {

                'message': 'Changes...',

                'author': 'Jane Doe <jane.doe@example.com>',

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

                'added': [

                    FileNode('some/new.txt', content='news...'),

                ],

                'changed': [

                    FileNode('foobar', 'Foobar I'),

                ],

                'removed': [],

            },

        ]

        return commits

    def test_simple(self):

        tip = self.repo.get_changeset()

        self.assertEqual(tip.date, datetime.datetime(2010, 1, 1, 21))

    def test_new_branch(self):

        # This check must not be removed to ensure the 'branches' LazyProperty

        # gets hit *before* the new 'foobar' branch got created:

        self.assertFalse('foobar' in self.repo.branches)

        self.imc.add(vcs.nodes.FileNode('docs/index.txt',

            author=u'joe',

            branch='foobar',

            parents=[tip],

        )

        self.imc.change(vcs.nodes.FileNode('docs/index.txt',

            content='Documentation\nand more...\n'))

        newtip = self.imc.commit(

            message=u'At default branch',

            author=u'joe',

            branch=foobar_tip.branch,

            parents=[foobar_tip],

        )

        newest_tip = self.imc.commit(

            message=u'Merged with %s' % foobar_tip.raw_id,

            author=u'joe',

            branch=self.backend_class.DEFAULT_BRANCH_NAME,

            parents=[newtip, foobar_tip],

        )

        self.assertEqual(newest_tip.branch,

            self.backend_class.DEFAULT_BRANCH_NAME)

    def test_branch_with_slash_in_name(self):

        self.imc.add(vcs.nodes.FileNode('extrafile', content='Some data\n'))

        self.imc.commit(u'Branch with a slash!', author=u'joe',

            branch='issue/123')

        self.assertTrue('issue/123' in self.repo.branches)

    def test_branch_with_slash_in_name_and_similar_without(self):

        self.imc.add(vcs.nodes.FileNode('extrafile', content='Some data\n'))

        self.imc.commit(u'Branch with a slash!', author=u'joe',

            branch='issue/123')

        self.imc.add(vcs.nodes.FileNode('extrafile II', content='Some data\n'))

        self.imc.commit(u'Branch without a slash...', author=u'joe',

            branch='123')

        self.assertIn('issue/123', self.repo.branches)

        self.assertIn('123', self.repo.branches)

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    cls_name = ''.join(('%s branches test' % alias).title().split())

    bases = (BranchesTestCaseMixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

        self.assertItemsEqual(changeset.removed, [])

        assert u'foo/ba\u0142' in changeset.as_dict()['added']

        assert u'foo/ba\u0142' in changeset.__json__(with_file_list=True)['added']

    def test_head_added(self):

        changeset = self.repo.get_changeset()

        self.assertTrue(isinstance(changeset.added, AddedFileNodesGenerator))

        self.assertItemsEqual(changeset.added, [

            changeset.get_node('fallout'),

        ])

        self.assertTrue(isinstance(changeset.changed, ChangedFileNodesGenerator))

        self.assertItemsEqual(changeset.changed, [

            changeset.get_node('foo/bar'),

            changeset.get_node('foobar'),

        ])

        self.assertTrue(isinstance(changeset.removed, RemovedFileNodesGenerator))

        self.assertEqual(len(changeset.removed), 1)

        self.assertEqual(list(changeset.removed)[0].path, 'qwe')

    def test_get_filemode(self):

        changeset = self.repo.get_changeset()

        self.assertEqual(33188, changeset.get_file_mode('foo/bar'))

    def test_get_filemode_non_ascii(self):

        changeset = self.repo.get_changeset()

        self.assertEqual(33188, changeset.get_file_mode('foo/bał'))

        self.assertEqual(33188, changeset.get_file_mode(u'foo/bał'))

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    # tests with additional commits

    cls_name = alias.title() + 'ChangesetsWithCommitsTest'

    bases = (_ChangesetsWithCommitsTestCaseixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

    # tests without additional commits

    cls_name = alias.title() + 'ChangesetsTest'

    bases = (_ChangesetsTestCaseMixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

    # tests changes

    cls_name = alias.title() + 'ChangesetsChangesTest'

    bases = (_ChangesetsChangesTestCaseMixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

# encoding: utf8

import datetime

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.test_inmemchangesets import BackendBaseTestCase

from kallithea.tests.vcs.conf import SCM_TESTS

class FileNodeUnicodePathTestsMixin(object):

    fname = 'ąśðąęłąć.txt'

    ufname = (fname).decode('utf-8')

    def get_commits(self):

        self.nodes = [

            FileNode(self.fname, content='Foobar'),

        ]

        commits = [

            {

                'message': 'Initial commit',

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

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

                'added': self.nodes,

            },

        ]

        return commits

    def test_filenode_path(self):

        node = self.tip.get_node(self.fname)

        unode = self.tip.get_node(self.ufname)

        self.assertEqual(node, unode)

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    cls_name = ''.join(('%s file node unicode path test' % alias).title()

        .split())

    bases = (FileNodeUnicodePathTestsMixin, BackendBaseTestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

import datetime

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.tests.vcs.conf import SCM_TESTS

from kallithea.lib.vcs.nodes import FileNode

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

class GetitemTestCaseMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        start_date = datetime.datetime(2010, 1, 1, 20)

        for x in xrange(5):

            yield {

                'message': 'Commit %d' % x,

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

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),

                ],

            }

    def test__getitem__last_item_is_tip(self):

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

    def test__getitem__returns_correct_items(self):

        changesets = [self.repo[x] for x in xrange(len(self.repo.revisions))]

        self.assertEqual(changesets, list(self.repo.get_changesets()))

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    cls_name = ''.join(('%s getitem test' % alias).title().split())

    bases = (GetitemTestCaseMixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

import datetime

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.tests.vcs.conf import SCM_TESTS

from kallithea.lib.vcs.nodes import FileNode

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

class GetsliceTestCaseMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        start_date = datetime.datetime(2010, 1, 1, 20)

        for x in xrange(5):

            yield {

                'message': 'Commit %d' % x,

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

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),

                ],

            }

    def test__getslice__last_item_is_tip(self):

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

    def test__getslice__respects_start_index(self):

        self.assertEqual(list(self.repo[2:]),

            [self.repo.get_changeset(rev) for rev in self.repo.revisions[2:]])

    def test__getslice__respects_negative_start_index(self):

        self.assertEqual(list(self.repo[-2:]),

            [self.repo.get_changeset(rev) for rev in self.repo.revisions[-2:]])

    def test__getslice__respects_end_index(self):

        self.assertEqual(list(self.repo[:2]),

            [self.repo.get_changeset(rev) for rev in self.repo.revisions[:2]])

    def test__getslice__respects_negative_end_index(self):

        self.assertEqual(list(self.repo[:-2]),

            [self.repo.get_changeset(rev) for rev in self.repo.revisions[:-2]])

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    cls_name = ''.join(('%s getslice test' % alias).title().split())

    bases = (GetsliceTestCaseMixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

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

            GitRepository(get_new_dir('injection-repo'), src_url=repo_inject_path, update_after_clone=True, create=True)

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

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

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

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

            with open(hook_path) as f:

                self.assertIn("CUSTOM_HOOK", f.read())

    def test_custom_hooks_forced_update(self):

        """

        Tests if hooks are forcefully updated even though they are custom hooks.

        """

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

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

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

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

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

            with open(hook_path) as f:

                self.assertIn("KALLITHEA_HOOK_VER", f.read())

import os

import pytest

from kallithea.lib.utils2 import safe_str

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, TEST_TMP_PATH

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

            pytest.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(safe_str(TEST_HG_REPO))

    def test_wrong_repo_path(self):

        wrong_repo_path = os.path.join(TEST_TMP_PATH, '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(safe_str(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(safe_str(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))

    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)

        return repo_path

    def setUp(self):

        Backend = self.get_backend()

        self.backend_class = Backend

        self.repo_path = self.get_new_repo_path()

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

        self.imc = self.repo.in_memory_changeset

        for commit in self.get_commits():

            for node in commit.get('added', []):

                self.imc.add(FileNode(node.path, content=node.content))

            for node in commit.get('changed', []):

                self.imc.change(FileNode(node.path, content=node.content))

            for node in commit.get('removed', []):

                self.imc.remove(FileNode(node.path))

            self.imc.commit(message=unicode(commit['message']),

                            author=unicode(commit['author']),

                date=commit['date'])

        self.tip = self.repo.get_changeset()

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    cls_name = ''.join(('%s in memory changeset test' % alias).title().split())

    bases = (InMemoryChangesetTestMixin, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)

import stat

import mimetypes

from kallithea.lib.vcs.nodes import DirNode

from kallithea.lib.vcs.nodes import FileNode

from kallithea.lib.vcs.nodes import Node

from kallithea.lib.vcs.nodes import NodeError

from kallithea.lib.vcs.nodes import NodeKind

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

class NodeBasicTest(unittest.TestCase):

    def test_init(self):

        """

        Cannot initialize Node objects with path with slash at the beginning.

        """

        wrong_paths = (

            '/foo',

            '/foo/bar'

        )

        for path in wrong_paths:

            self.assertRaises(NodeError, Node, path, NodeKind.FILE)

        wrong_paths = (

            '/foo/',

            '/foo/bar/'

        )

        for path in wrong_paths:

            self.assertRaises(NodeError, Node, path, NodeKind.DIR)

    def test_name(self):

        node = Node('', NodeKind.DIR)

        self.assertEqual(node.name, '')

        node = Node('path', NodeKind.FILE)

        self.assertEqual(node.name, 'path')

        node = Node('path/', NodeKind.DIR)

        self.assertEqual(node.name, 'path')

        node = Node('some/path', NodeKind.FILE)

        self.assertEqual(node.name, 'path')

        node = Node('some/path/', NodeKind.DIR)

        self.assertEqual(node.name, 'path')

    def test_root_node(self):

        self.assertRaises(NodeError, Node, '', NodeKind.FILE)

        self.assertTrue(mode & stat.S_IWUSR)

        self.assertTrue(mode & stat.S_IRGRP)

        self.assertTrue(mode & stat.S_IROTH)

        self.assertFalse(mode & stat.S_IWGRP)

        self.assertFalse(mode & stat.S_IWOTH)

        self.assertFalse(mode & stat.S_IXUSR)

        self.assertFalse(mode & stat.S_IXGRP)

        self.assertFalse(mode & stat.S_IXOTH)

    def test_file_node_is_executable(self):

        node = FileNode('foobar', 'empty... almost', mode=0100755)

        self.assertTrue(node.is_executable)

        node = FileNode('foobar', 'empty... almost', mode=0100500)

        self.assertTrue(node.is_executable)

        node = FileNode('foobar', 'empty... almost', mode=0100644)

        self.assertFalse(node.is_executable)

    def test_mimetype(self):

        py_node = FileNode('test.py')

        tar_node = FileNode('test.tar.gz')

        my_node2 = FileNode('myfile2')

        my_node2._content = 'foobar'

        my_node3 = FileNode('myfile3')

        my_node3._content = '\0foobar'

        self.assertEqual(py_node.mimetype, mimetypes.guess_type(py_node.name)[0])

        self.assertEqual(py_node.get_mimetype(), mimetypes.guess_type(py_node.name))

        self.assertEqual(tar_node.mimetype, mimetypes.guess_type(tar_node.name)[0])

        self.assertEqual(tar_node.get_mimetype(), mimetypes.guess_type(tar_node.name))

        self.assertEqual(my_node2.mimetype, 'text/plain')

        self.assertEqual(my_node2.get_mimetype(), ('text/plain', None))

        self.assertEqual(my_node3.mimetype, 'application/octet-stream')

        self.assertEqual(my_node3.get_mimetype(), ('application/octet-stream', None))

class NodeContentTest(unittest.TestCase):

    def test_if_binary(self):

        data = """\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x10\x00\x00\x00\x10\x08\x06\x00\x00\x00\x1f??a\x00\x00\x00\x04gAMA\x00\x00\xaf?7\x05\x8a?\x00\x00\x00\x19tEXtSoftware\x00Adobe ImageReadyq?e<\x00\x00\x025IDAT8?\xa5\x93?K\x94Q\x14\x87\x9f\xf7?Q\x1bs4?\x03\x9a\xa8?B\x02\x8b$\x10[U;i\x13?6h?&h[?"\x14j?\xa2M\x7fB\x14F\x9aQ?&\x842?\x0b\x89"\x82??!?\x9c!\x9c2l??{N\x8bW\x9dY\xb4\t/\x1c?=\x9b?}????\xa9*;9!?\x83\x91?[?\\v*?D\x04\'`EpNp\xa2X\'U?pVq"Sw.\x1e?\x08\x01D?jw????\xbc??7{|\x9b?\x89$\x01??W@\x15\x9c\x05q`Lt/\x97?\x94\xa1d?\x18~?\x18?\x18W[%\xb0?\x83??\x14\x88\x8dB?\xa6H\tL\tl\x19>/\x01`\xac\xabx?\x9cl\nx\xb0\x98\x07\x95\x88D$"q[\x19?d\x00(o\n\xa0??\x7f\xb9\xa4?\x1bF\x1f\x8e\xac\xa8?j??eUU}?.?\x9f\x8cE??x\x94??\r\xbdtoJU5"0N\x10U?\x00??V\t\x02\x9f\x81?U?\x00\x9eM\xae2?r\x9b7\x83\x82\x8aP3????.?&"?\xb7ZP \x0c<?O\xa5\t}\xb8?\x99\xa6?\x87?\x1di|/\xa0??0\xbe\x1fp?d&\x1a\xad\x95\x8a\x07?\t*\x10??b:?d?.\x13C\x8a?\x12\xbe\xbf\x8e?{???\x08?\x80\xa7\x13+d\x13>J?\x80\x15T\x95\x9a\x00??S\x8c\r?\xa1\x03\x07?\x96\x9b\xa7\xab=E??\xa4\xb3?\x19q??B\x91=\x8d??k?J\x0bV"??\xf7x?\xa1\x00?\\.\x87\x87???\x02F@D\x99],??\x10#?X\xb7=\xb9\x10?Z\x1by???cI??\x1ag?\x92\xbc?T?t[\x92\x81?<_\x17~\x92\x88?H%?\x10Q\x02\x9f\n\x81qQ\x0bm?\x1bX?\xb1AK\xa6\x9e\xb9?u\xb2?1\xbe|/\x92M@\xa2!F?\xa9>"\r<DT?>\x92\x8e?>\x9a9Qv\x127?a\xac?Y?8?:??]X???9\x80\xb7?u?\x0b#BZ\x8d=\x1d?p\x00\x00\x00\x00IEND\xaeB`\x82"""

        filenode = FileNode('calendar.png', content=data)

        self.assertTrue(filenode.is_binary)

    def test_second_changeset_diff(self):

        revs = self.repo.revisions

        self.assertEqual(self.repo.get_diff(revs[0], revs[1]), '''diff --git a/foobar b/foobar

--- a/foobar

+++ b/foobar

@@ -1,1 +1,1 @@

-foobar

\ No newline at end of file

+FOOBAR

\ No newline at end of file

diff --git a/foobar3 b/foobar3

new file mode 100644

--- /dev/null

+++ b/foobar3

@@ -0,0 +1,1 @@

+foobar3

\ No newline at end of file

''')

    def test_third_changeset_diff(self):

        revs = self.repo.revisions

        self.assertEqual(self.repo.get_diff(revs[1], revs[2]), '''diff --git a/foobar b/foobar

deleted file mode 100644

--- a/foobar

+++ /dev/null

@@ -1,1 +0,0 @@

-FOOBAR

\ No newline at end of file

diff --git a/foobar3 b/foobar3

--- a/foobar3

+++ b/foobar3

@@ -1,1 +1,3 @@

-foobar3

\ No newline at end of file

+FOOBAR

+FOOBAR

+FOOBAR

''')

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

    cls_name = alias.capitalize() + RepositoryBaseTest.__name__

    bases = (RepositoryBaseTest, unittest.TestCase)

    globals()[cls_name] = type(cls_name, bases, attrs)