Changeset - fc33889cb798
kallithea/tests/vcs/__init__.py
Show inline comments
 
"""
 
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.lib.vcs.utils.compat import 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()
 

	
 

	
 
def collector():
 
    setup_package()
 
    start_dir = os.path.abspath(os.path.dirname(__file__))
 
    return unittest.defaultTestLoader.discover(start_dir)
 

	
 

	
 
def main():
 
    collector()
 
    unittest.main()
 

	
 

	
 
#if __name__ == '__main__':
 
#    main()
kallithea/tests/vcs/base.py
Show inline comments
 
"""
 
Module providing backend independent mixin class. It requires that
 
InMemoryChangeset class is working properly at backend class.
 
"""
 
import os
 
import time
 
import shutil
 
import datetime
 
from kallithea.tests.vcs.conf import SCM_TESTS, get_new_dir
 

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

	
 

	
 
class _BackendTestMixin(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
 
    - ``recreate_repo_per_test``: If set to ``False``, repo would NOT be created
 
      before every single test. Defaults to ``True``.
 
    """
 
    recreate_repo_per_test = True
 

	
 
    @classmethod
 
    def get_backend(cls):
 
        return vcs.get_backend(cls.backend_alias)
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        commits = [
 
            {
 
                'message': u'Initial commit',
 
                'author': u'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': u'Changes...',
 
                'author': u'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
 

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

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_archives.py
Show inline comments
 
import os
 
import tarfile
 
import zipfile
 
import datetime
 
import tempfile
 
import StringIO
 

	
 
from kallithea.tests.vcs.base import _BackendTestMixin
 
from kallithea.tests.vcs.conf import SCM_TESTS, TEST_TMP_PATH
 
from kallithea.lib.vcs.exceptions import VCSError
 
from kallithea.lib.vcs.nodes import FileNode
 
from kallithea.lib.vcs.utils.compat import unittest
 

	
 

	
 
class ArchivesTestCaseMixin(_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('%d/file_%d.txt' % (x, x),
 
                        content='Foobar %d' % x),
 
                ],
 
            }
 

	
 
    def test_archive_zip(self):
 
        path = tempfile.mkstemp(dir=TEST_TMP_PATH, prefix='test_archive_zip-')[1]
 
        with open(path, 'wb') as f:
 
            self.tip.fill_archive(stream=f, kind='zip', prefix='repo')
 
        out = zipfile.ZipFile(path)
 

	
 
        for x in xrange(5):
 
            node_path = '%d/file_%d.txt' % (x, x)
 
            decompressed = StringIO.StringIO()
 
            decompressed.write(out.read('repo/' + node_path))
 
            self.assertEqual(
 
                decompressed.getvalue(),
 
                self.tip.get_node(node_path).content)
 

	
 
    def test_archive_tgz(self):
 
        path = tempfile.mkstemp(dir=TEST_TMP_PATH, prefix='test_archive_tgz-')[1]
 
        with open(path, 'wb') as f:
 
            self.tip.fill_archive(stream=f, kind='tgz', prefix='repo')
 
        outdir = tempfile.mkdtemp(dir=TEST_TMP_PATH, prefix='test_archive_tgz-', suffix='-outdir')
 

	
 
        outfile = tarfile.open(path, 'r|gz')
 
        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_tbz2(self):
 
        path = tempfile.mkstemp(dir=TEST_TMP_PATH, prefix='test_archive_tbz2-')[1]
 
        with open(path, 'w+b') as f:
 
            self.tip.fill_archive(stream=f, kind='tbz2', prefix='repo')
 
        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)
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_branches.py
Show inline comments
 

	
 
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',
 
            content='Documentation\n'))
 
        foobar_tip = self.imc.commit(
 
            message=u'New branch: foobar',
 
            author=u'joe',
 
            branch='foobar',
 
        )
 
        self.assertTrue('foobar' in self.repo.branches)
 
        self.assertEqual(foobar_tip.branch, 'foobar')
 

	
 
    def test_new_head(self):
 
        tip = self.repo.get_changeset()
 
        self.imc.add(vcs.nodes.FileNode('docs/index.txt',
 
            content='Documentation\n'))
 
        foobar_tip = self.imc.commit(
 
            message=u'New branch: foobar',
 
            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)
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_changesets.py
Show inline comments
 
@@ -199,196 +199,192 @@ class _ChangesetsTestCaseMixin(_BackendT
 
        self.assertEqual(changesets.index(changesets[-1]), second_id)
 
        self.assertEqual(len(changesets), 2)
 

	
 
    def test_get_changesets_respects_both_start_and_end(self):
 
        second_id = self.repo.revisions[1]
 
        third_id = self.repo.revisions[2]
 
        changesets = list(self.repo.get_changesets(start=second_id,
 
            end=third_id))
 
        self.assertEqual(len(changesets), 2)
 

	
 
    def test_get_changesets_numerical_id_respects_both_start_and_end(self):
 
        changesets = list(self.repo.get_changesets(start=2, end=3))
 
        self.assertEqual(len(changesets), 2)
 

	
 
    def test_get_changesets_on_empty_repo_raises_EmptyRepository_error(self):
 
        Backend = self.get_backend()
 
        repo_path = get_new_dir(str(time.time()))
 
        repo = Backend(repo_path, create=True)
 

	
 
        with self.assertRaises(EmptyRepositoryError):
 
            list(repo.get_changesets(start='foobar'))
 

	
 
    def test_get_changesets_includes_end_changeset(self):
 
        second_id = self.repo.revisions[1]
 
        changesets = list(self.repo.get_changesets(end=second_id))
 
        self.assertEqual(changesets[-1].raw_id, second_id)
 

	
 
    def test_get_changesets_respects_start_date(self):
 
        start_date = datetime.datetime(2010, 2, 1)
 
        for cs in self.repo.get_changesets(start_date=start_date):
 
            self.assertGreaterEqual(cs.date, start_date)
 

	
 
    def test_get_changesets_respects_end_date(self):
 
        start_date = datetime.datetime(2010, 1, 1)
 
        end_date = datetime.datetime(2010, 2, 1)
 
        for cs in self.repo.get_changesets(start_date=start_date,
 
                                           end_date=end_date):
 
            self.assertGreaterEqual(cs.date, start_date)
 
            self.assertLessEqual(cs.date, end_date)
 

	
 
    def test_get_changesets_respects_start_date_and_end_date(self):
 
        end_date = datetime.datetime(2010, 2, 1)
 
        for cs in self.repo.get_changesets(end_date=end_date):
 
            self.assertLessEqual(cs.date, end_date)
 

	
 
    def test_get_changesets_respects_reverse(self):
 
        changesets_id_list = [cs.raw_id for cs in
 
            self.repo.get_changesets(reverse=True)]
 
        self.assertItemsEqual(changesets_id_list, reversed(self.repo.revisions))
 

	
 
    def test_get_filenodes_generator(self):
 
        tip = self.repo.get_changeset()
 
        filepaths = [node.path for node in tip.get_filenodes_generator()]
 
        self.assertItemsEqual(filepaths, ['file_%d.txt' % x for x in xrange(5)])
 

	
 
    def test_size(self):
 
        tip = self.repo.get_changeset()
 
        size = 5 * len('Foobar N') # Size of 5 files
 
        self.assertEqual(tip.size, size)
 

	
 
    def test_author(self):
 
        tip = self.repo.get_changeset()
 
        self.assertEqual(tip.author, u'Joe Doe <joe.doe@example.com>')
 

	
 
    def test_author_name(self):
 
        tip = self.repo.get_changeset()
 
        self.assertEqual(tip.author_name, u'Joe Doe')
 

	
 
    def test_author_email(self):
 
        tip = self.repo.get_changeset()
 
        self.assertEqual(tip.author_email, u'joe.doe@example.com')
 

	
 
    def test_get_changesets_raise_changesetdoesnotexist_for_wrong_start(self):
 
        with self.assertRaises(ChangesetDoesNotExistError):
 
            list(self.repo.get_changesets(start='foobar'))
 

	
 
    def test_get_changesets_raise_changesetdoesnotexist_for_wrong_end(self):
 
        with self.assertRaises(ChangesetDoesNotExistError):
 
            list(self.repo.get_changesets(end='foobar'))
 

	
 
    def test_get_changesets_raise_branchdoesnotexist_for_wrong_branch_name(self):
 
        with self.assertRaises(BranchDoesNotExistError):
 
            list(self.repo.get_changesets(branch_name='foobar'))
 

	
 
    def test_get_changesets_raise_repositoryerror_for_wrong_start_end(self):
 
        start = self.repo.revisions[-1]
 
        end = self.repo.revisions[0]
 
        with self.assertRaises(RepositoryError):
 
            list(self.repo.get_changesets(start=start, end=end))
 

	
 
    def test_get_changesets_numerical_id_reversed(self):
 
        with self.assertRaises(RepositoryError):
 
            [x for x in self.repo.get_changesets(start=3, end=2)]
 

	
 
    def test_get_changesets_numerical_id_respects_both_start_and_end_last(self):
 
        with self.assertRaises(RepositoryError):
 
            last = len(self.repo.revisions)
 
            list(self.repo.get_changesets(start=last-1, end=last-2))
 

	
 
    def test_get_changesets_numerical_id_last_zero_error(self):
 
        with self.assertRaises(RepositoryError):
 
            last = len(self.repo.revisions)
 
            list(self.repo.get_changesets(start=last-1, end=0))
 

	
 

	
 
class _ChangesetsChangesTestCaseMixin(_BackendTestMixin):
 
    recreate_repo_per_test = False
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        return [
 
            {
 
                'message': u'Initial',
 
                'author': u'Joe Doe <joe.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 20),
 
                'added': [
 
                    FileNode('foo/bar', content='foo'),
 
                    FileNode('foo/bał', content='foo'),
 
                    FileNode('foobar', content='foo'),
 
                    FileNode('qwe', content='foo'),
 
                ],
 
            },
 
            {
 
                'message': u'Massive changes',
 
                'author': u'Joe Doe <joe.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 22),
 
                'added': [FileNode('fallout', content='War never changes')],
 
                'changed': [
 
                    FileNode('foo/bar', content='baz'),
 
                    FileNode('foobar', content='baz'),
 
                ],
 
                'removed': [FileNode('qwe')],
 
            },
 
        ]
 

	
 
    def test_initial_commit(self):
 
        changeset = self.repo.get_changeset(0)
 
        self.assertItemsEqual(changeset.added, [
 
            changeset.get_node('foo/bar'),
 
            changeset.get_node('foo/bał'),
 
            changeset.get_node('foobar'),
 
            changeset.get_node('qwe'),
 
        ])
 
        self.assertItemsEqual(changeset.changed, [])
 
        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)
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_filenodes_unicode_path.py
Show inline comments
 
# encoding: utf8
 

	
 

	
 
import datetime
 
from kallithea.lib.vcs.nodes import FileNode
 
from kallithea.lib.vcs.utils.compat import unittest
 
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)
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_getitem.py
Show inline comments
 

	
 
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)
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_getslice.py
Show inline comments
 

	
 
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)
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_git.py
Show inline comments
 

	
 
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-$")
 
        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
 
@@ -656,196 +655,192 @@ class GitSpecificTest(unittest.TestCase)
 
    def test_error_is_raised_for_removed_if_diff_name_status_is_wrong(self):
 
        repo = mock.MagicMock()
 
        changeset = GitChangeset(repo, 'foobar')
 
        changeset._diff_name_status = 'foobar'
 
        with self.assertRaises(VCSError):
 
            changeset.added
 

	
 

	
 
class GitSpecificWithRepoTest(_BackendTestMixin, unittest.TestCase):
 
    backend_alias = 'git'
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        return [
 
            {
 
                'message': 'Initial',
 
                'author': 'Joe Doe <joe.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 20),
 
                'added': [
 
                    FileNode('foobar/static/js/admin/base.js', content='base'),
 
                    FileNode('foobar/static/admin', content='admin',
 
                        mode=0120000), # this is a link
 
                    FileNode('foo', content='foo'),
 
                ],
 
            },
 
            {
 
                'message': 'Second',
 
                'author': 'Joe Doe <joe.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 22),
 
                'added': [
 
                    FileNode('foo2', content='foo2'),
 
                ],
 
            },
 
        ]
 

	
 
    def test_paths_slow_traversing(self):
 
        cs = self.repo.get_changeset()
 
        self.assertEqual(cs.get_node('foobar').get_node('static').get_node('js')
 
            .get_node('admin').get_node('base.js').content, 'base')
 

	
 
    def test_paths_fast_traversing(self):
 
        cs = self.repo.get_changeset()
 
        self.assertEqual(cs.get_node('foobar/static/js/admin/base.js').content,
 
            'base')
 

	
 
    def test_workdir_get_branch(self):
 
        self.repo.run_git_command(['checkout', '-b', 'production'])
 
        # Regression test: one of following would fail if we don't check
 
        # .git/HEAD file
 
        self.repo.run_git_command(['checkout', 'production'])
 
        self.assertEqual(self.repo.workdir.get_branch(), 'production')
 
        self.repo.run_git_command(['checkout', 'master'])
 
        self.assertEqual(self.repo.workdir.get_branch(), 'master')
 

	
 
    def test_get_diff_runs_git_command_with_hashes(self):
 
        self.repo.run_git_command = mock.Mock(return_value=['', ''])
 
        self.repo.get_diff(0, 1)
 
        self.repo.run_git_command.assert_called_once_with(
 
            ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
 
             self.repo._get_revision(0), self.repo._get_revision(1)])
 

	
 
    def test_get_diff_runs_git_command_with_str_hashes(self):
 
        self.repo.run_git_command = mock.Mock(return_value=['', ''])
 
        self.repo.get_diff(self.repo.EMPTY_CHANGESET, 1)
 
        self.repo.run_git_command.assert_called_once_with(
 
            ['show', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
 
             self.repo._get_revision(1)])
 

	
 
    def test_get_diff_runs_git_command_with_path_if_its_given(self):
 
        self.repo.run_git_command = mock.Mock(return_value=['', ''])
 
        self.repo.get_diff(0, 1, 'foo')
 
        self.repo.run_git_command.assert_called_once_with(
 
            ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
 
             self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'])
 

	
 

	
 
class GitRegressionTest(_BackendTestMixin, unittest.TestCase):
 
    backend_alias = 'git'
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        return [
 
            {
 
                'message': 'Initial',
 
                'author': 'Joe Doe <joe.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 20),
 
                'added': [
 
                    FileNode('bot/__init__.py', content='base'),
 
                    FileNode('bot/templates/404.html', content='base'),
 
                    FileNode('bot/templates/500.html', content='base'),
 
                ],
 
            },
 
            {
 
                'message': 'Second',
 
                'author': 'Joe Doe <joe.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 22),
 
                'added': [
 
                    FileNode('bot/build/migrations/1.py', content='foo2'),
 
                    FileNode('bot/build/migrations/2.py', content='foo2'),
 
                    FileNode('bot/build/static/templates/f.html', content='foo2'),
 
                    FileNode('bot/build/static/templates/f1.html', content='foo2'),
 
                    FileNode('bot/build/templates/err.html', content='foo2'),
 
                    FileNode('bot/build/templates/err2.html', content='foo2'),
 
                ],
 
            },
 
        ]
 

	
 
    def test_similar_paths(self):
 
        cs = self.repo.get_changeset()
 
        paths = lambda *n: [x.path for x in n]
 
        self.assertEqual(paths(*cs.get_nodes('bot')), ['bot/build', 'bot/templates', 'bot/__init__.py'])
 
        self.assertEqual(paths(*cs.get_nodes('bot/build')), ['bot/build/migrations', 'bot/build/static', 'bot/build/templates'])
 
        self.assertEqual(paths(*cs.get_nodes('bot/build/static')), ['bot/build/static/templates'])
 
        # this get_nodes below causes troubles !
 
        self.assertEqual(paths(*cs.get_nodes('bot/build/static/templates')), ['bot/build/static/templates/f.html', 'bot/build/static/templates/f1.html'])
 
        self.assertEqual(paths(*cs.get_nodes('bot/build/templates')), ['bot/build/templates/err.html', 'bot/build/templates/err2.html'])
 
        self.assertEqual(paths(*cs.get_nodes('bot/templates/')), ['bot/templates/404.html', 'bot/templates/500.html'])
 

	
 

	
 
class GitHooksTest(unittest.TestCase):
 
    """
 
    Tests related to hook functionality of Git repositories.
 
    """
 

	
 
    def setUp(self):
 
        # For each run we want a fresh repo.
 
        self.repo_directory = get_new_dir("githookrepo")
 
        self.repo = GitRepository(self.repo_directory, create=True)
 

	
 
        # Create a dictionary where keys are hook names, and values are paths to
 
        # them. Deduplicates code in tests a bit.
 
        self.hook_directory = self.repo.get_hook_location()
 
        self.kallithea_hooks = dict((h, os.path.join(self.hook_directory, h)) for h in ("pre-receive", "post-receive"))
 

	
 
    def test_hooks_created_if_missing(self):
 
        """
 
        Tests if hooks are installed in repository if they are missing.
 
        """
 

	
 
        for hook, hook_path in self.kallithea_hooks.iteritems():
 
            if os.path.exists(hook_path):
 
                os.remove(hook_path)
 

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

	
 
        for hook, hook_path in self.kallithea_hooks.iteritems():
 
            self.assertTrue(os.path.exists(hook_path))
 

	
 
    def test_kallithea_hooks_updated(self):
 
        """
 
        Tests if hooks are updated if they are Kallithea hooks already.
 
        """
 

	
 
        for hook, hook_path in self.kallithea_hooks.iteritems():
 
            with open(hook_path, "w") as f:
 
                f.write("KALLITHEA_HOOK_VER=0.0.0\nJUST_BOGUS")
 

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

	
 
        for hook, hook_path in self.kallithea_hooks.iteritems():
 
            with open(hook_path) as f:
 
                self.assertNotIn("JUST_BOGUS", f.read())
 

	
 
    def test_custom_hooks_untouched(self):
 
        """
 
        Tests if hooks are left untouched if they are not Kallithea hooks.
 
        """
 

	
 
        for hook, hook_path in self.kallithea_hooks.iteritems():
 
            with open(hook_path, "w") as f:
 
                f.write("#!/bin/bash\n#CUSTOM_HOOK")
 

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

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_hg.py
Show inline comments
 

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

	
 
        # check if current workdir was updated
 
        self.assertEqual(os.path.isfile(os.path.join(TEST_HG_REPO_CLONE \
 
                                                    + '_w_update',
 
                                                    'MANIFEST.in')), True,)
 

	
 
    def test_repo_clone_without_update(self):
 
        repo = MercurialRepository(safe_str(TEST_HG_REPO))
 
        repo_clone = MercurialRepository(TEST_HG_REPO_CLONE + '_wo_update',
 
            src_url=TEST_HG_REPO, update_after_clone=False)
 
        self.assertEqual(len(repo.revisions), len(repo_clone.revisions))
 
        self.assertEqual(os.path.isfile(os.path.join(TEST_HG_REPO_CLONE \
 
                                                    + '_wo_update',
 
                                                    'MANIFEST.in')), False,)
 

	
 
    def test_pull(self):
 
        if os.path.exists(TEST_HG_REPO_PULL):
 
            pytest.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)
 

	
kallithea/tests/vcs/test_inmemchangesets.py
Show inline comments
 
@@ -199,196 +199,192 @@ class InMemoryChangesetTestMixin(object)
 
        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)
 
        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)
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_nodes.py
Show inline comments
 

	
 
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)
 

	
 
    def test_kind_setter(self):
 
        node = Node('', NodeKind.DIR)
 
        self.assertRaises(NodeError, setattr, node, 'kind', NodeKind.FILE)
 

	
 
    def _test_parent_path(self, node_path, expected_parent_path):
 
        """
 
        Tests if node's parent path are properly computed.
 
        """
 
        node = Node(node_path, NodeKind.DIR)
 
        parent_path = node.get_parent_path()
 
        self.assertTrue(parent_path.endswith('/') or \
 
            node.is_root() and parent_path == '')
 
        self.assertEqual(parent_path, expected_parent_path,
 
            "Node's path is %r and parent path is %r but should be %r"
 
            % (node.path, parent_path, expected_parent_path))
 

	
 
    def test_parent_path(self):
 
        test_paths = (
 
            # (node_path, expected_parent_path)
 
            ('', ''),
 
            ('some/path/', 'some/'),
 
            ('some/longer/path/', 'some/longer/'),
 
        )
 
        for node_path, expected_parent_path in test_paths:
 
            self._test_parent_path(node_path, expected_parent_path)
 

	
 
    '''
 
    def _test_trailing_slash(self, path):
 
        if not path.endswith('/'):
 
            pytest.fail("Trailing slash tests needs paths to end with slash")
 
        for kind in NodeKind.FILE, NodeKind.DIR:
 
            self.assertRaises(NodeError, Node, path=path, kind=kind)
 

	
 
    def test_trailing_slash(self):
 
        for path in ('/', 'foo/', 'foo/bar/', 'foo/bar/biz/'):
 
            self._test_trailing_slash(path)
 
    '''
 

	
 
    def test_is_file(self):
 
        node = Node('any', NodeKind.FILE)
 
        self.assertTrue(node.is_file())
 

	
 
        node = FileNode('any')
 
        self.assertTrue(node.is_file())
 
        self.assertRaises(AttributeError, getattr, node, 'nodes')
 

	
 
    def test_is_dir(self):
 
        node = Node('any_dir', NodeKind.DIR)
 
        self.assertTrue(node.is_dir())
 

	
 
        node = DirNode('any_dir')
 

	
 
        self.assertTrue(node.is_dir())
 
        self.assertRaises(NodeError, getattr, node, 'content')
 

	
 
    def test_dir_node_iter(self):
 
        nodes = [
 
            DirNode('docs'),
 
            DirNode('tests'),
 
            FileNode('bar'),
 
            FileNode('foo'),
 
            FileNode('readme.txt'),
 
            FileNode('setup.py'),
 
        ]
 
        dirnode = DirNode('', nodes=nodes)
 
        for node in dirnode:
 
            node == dirnode.get_node(node.path)
 

	
 
    def test_node_state(self):
 
        """
 
        Without link to changeset nodes should raise NodeError.
 
        """
 
        node = FileNode('anything')
 
        self.assertRaises(NodeError, getattr, node, 'state')
 
        node = DirNode('anything')
 
        self.assertRaises(NodeError, getattr, node, 'state')
 

	
 
    def test_file_node_stat(self):
 
        node = FileNode('foobar', 'empty... almost')
 
        mode = node.mode  # default should be 0100644
 
        self.assertTrue(mode & stat.S_IRUSR)
 
        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)
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_repository.py
Show inline comments
 
@@ -34,195 +34,192 @@ class RepositoryBaseTest(_BackendTestMix
 
        self.assertTrue(self.repo == self.repo)
 

	
 
    def test_repo_equality_broken_object(self):
 
        import copy
 
        _repo = copy.copy(self.repo)
 
        delattr(_repo, 'path')
 
        self.assertTrue(self.repo != _repo)
 

	
 
    def test_repo_equality_other_object(self):
 
        class dummy(object):
 
            path = self.repo.path
 
        self.assertTrue(self.repo != dummy())
 

	
 

	
 
class RepositoryGetDiffTest(_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='foobar2'),
 
                ],
 
            },
 
            {
 
                'message': 'Changed foobar, added foobar3',
 
                'author': 'Jane Doe <jane.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 21),
 
                'added': [
 
                    FileNode('foobar3', content='foobar3'),
 
                ],
 
                'changed': [
 
                    FileNode('foobar', 'FOOBAR'),
 
                ],
 
            },
 
            {
 
                'message': 'Removed foobar, changed foobar3',
 
                'author': 'Jane Doe <jane.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 22),
 
                'changed': [
 
                    FileNode('foobar3', content='FOOBAR\nFOOBAR\nFOOBAR\n'),
 
                ],
 
                'removed': [FileNode('foobar')],
 
            },
 
        ]
 
        return commits
 

	
 
    def test_raise_for_wrong(self):
 
        with self.assertRaises(ChangesetDoesNotExistError):
 
            self.repo.get_diff('a' * 40, 'b' * 40)
 

	
 

	
 
class GitRepositoryGetDiffTest(RepositoryGetDiffTest, unittest.TestCase):
 
    backend_alias = 'git'
 

	
 
    def test_initial_commit_diff(self):
 
        initial_rev = self.repo.revisions[0]
 
        self.assertEqual(self.repo.get_diff(self.repo.EMPTY_CHANGESET, initial_rev), '''diff --git a/foobar b/foobar
 
new file mode 100644
 
index 0000000000000000000000000000000000000000..f6ea0495187600e7b2288c8ac19c5886383a4632
 
--- /dev/null
 
+++ b/foobar
 
@@ -0,0 +1 @@
 
+foobar
 
\ No newline at end of file
 
diff --git a/foobar2 b/foobar2
 
new file mode 100644
 
index 0000000000000000000000000000000000000000..e8c9d6b98e3dce993a464935e1a53f50b56a3783
 
--- /dev/null
 
+++ b/foobar2
 
@@ -0,0 +1 @@
 
+foobar2
 
\ No newline at end of file
 
''')
 

	
 
    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
 
index f6ea0495187600e7b2288c8ac19c5886383a4632..389865bb681b358c9b102d79abd8d5f941e96551 100644
 
--- a/foobar
 
+++ b/foobar
 
@@ -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
 
index 0000000000000000000000000000000000000000..c11c37d41d33fb47741cff93fa5f9d798c1535b0
 
--- /dev/null
 
+++ b/foobar3
 
@@ -0,0 +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
 
index 389865bb681b358c9b102d79abd8d5f941e96551..0000000000000000000000000000000000000000
 
--- a/foobar
 
+++ /dev/null
 
@@ -1 +0,0 @@
 
-FOOBAR
 
\ No newline at end of file
 
diff --git a/foobar3 b/foobar3
 
index c11c37d41d33fb47741cff93fa5f9d798c1535b0..f9324477362684ff692aaf5b9a81e01b9e9a671c 100644
 
--- a/foobar3
 
+++ b/foobar3
 
@@ -1 +1,3 @@
 
-foobar3
 
\ No newline at end of file
 
+FOOBAR
 
+FOOBAR
 
+FOOBAR
 
''')
 

	
 

	
 
class HgRepositoryGetDiffTest(RepositoryGetDiffTest, unittest.TestCase):
 
    backend_alias = 'hg'
 

	
 
    def test_initial_commit_diff(self):
 
        initial_rev = self.repo.revisions[0]
 
        self.assertEqual(self.repo.get_diff(self.repo.EMPTY_CHANGESET, initial_rev), '''diff --git a/foobar b/foobar
 
new file mode 100644
 
--- /dev/null
 
+++ b/foobar
 
@@ -0,0 +1,1 @@
 
+foobar
 
\ No newline at end of file
 
diff --git a/foobar2 b/foobar2
 
new file mode 100644
 
--- /dev/null
 
+++ b/foobar2
 
@@ -0,0 +1,1 @@
 
+foobar2
 
\ No newline at end of file
 
''')
 

	
 
    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)
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_tags.py
Show inline comments
 

	
 
from kallithea.tests.vcs.base import _BackendTestMixin
 
from kallithea.tests.vcs.conf import SCM_TESTS
 
from kallithea.lib.vcs.exceptions import TagAlreadyExistError
 
from kallithea.lib.vcs.exceptions import TagDoesNotExistError
 
from kallithea.lib.vcs.utils.compat import unittest
 

	
 

	
 
class TagsTestCaseMixin(_BackendTestMixin):
 

	
 
    def test_new_tag(self):
 
        tip = self.repo.get_changeset()
 
        tagsize = len(self.repo.tags)
 
        tag = self.repo.tag('last-commit', 'joe', tip.raw_id)
 

	
 
        self.assertEqual(len(self.repo.tags), tagsize + 1)
 
        for top, dirs, files in tip.walk():
 
            self.assertEqual(top, tag.get_node(top.path))
 

	
 
    def test_tag_already_exist(self):
 
        tip = self.repo.get_changeset()
 
        self.repo.tag('last-commit', 'joe', tip.raw_id)
 

	
 
        self.assertRaises(TagAlreadyExistError,
 
            self.repo.tag, 'last-commit', 'joe', tip.raw_id)
 

	
 
        chset = self.repo.get_changeset(0)
 
        self.assertRaises(TagAlreadyExistError,
 
            self.repo.tag, 'last-commit', 'jane', chset.raw_id)
 

	
 
    def test_remove_tag(self):
 
        tip = self.repo.get_changeset()
 
        self.repo.tag('last-commit', 'joe', tip.raw_id)
 
        tagsize = len(self.repo.tags)
 

	
 
        self.repo.remove_tag('last-commit', user='evil joe')
 
        self.assertEqual(len(self.repo.tags), tagsize - 1)
 

	
 
    def test_remove_tag_which_does_not_exist(self):
 
        self.assertRaises(TagDoesNotExistError,
 
            self.repo.remove_tag, 'last-commit', user='evil joe')
 

	
 
    def test_name_with_slash(self):
 
        self.repo.tag('19/10/11', 'joe')
 
        self.assertTrue('19/10/11' in self.repo.tags)
 

	
 

	
 
# For each backend create test case class
 
for alias in SCM_TESTS:
 
    attrs = {
 
        'backend_alias': alias,
 
    }
 
    cls_name = ''.join(('%s tags test' % alias).title().split())
 
    bases = (TagsTestCaseMixin, unittest.TestCase)
 
    globals()[cls_name] = type(cls_name, bases, attrs)
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_utils.py
Show inline comments
 
@@ -81,196 +81,192 @@ class TestParseChangesets(unittest.TestC
 
            'start': None,
 
            'main': '123456',
 
            'end': None,
 
        })
 

	
 
    def test_start_is_returned_correctly(self):
 
        self.assertEqual(parse_changesets('aaabbb..'), {
 
            'start': 'aaabbb',
 
            'main': None,
 
            'end': None,
 
        })
 

	
 
    def test_end_is_returned_correctly(self):
 
        self.assertEqual(parse_changesets('..cccddd'), {
 
            'start': None,
 
            'main': None,
 
            'end': 'cccddd',
 
        })
 

	
 
    def test_that_two_or_three_dots_are_allowed(self):
 
        text1 = 'a..b'
 
        text2 = 'a...b'
 
        self.assertEqual(parse_changesets(text1), parse_changesets(text2))
 

	
 
    def test_that_input_is_stripped_first(self):
 
        text1 = 'a..bb'
 
        text2 = '  a..bb\t\n\t '
 
        self.assertEqual(parse_changesets(text1), parse_changesets(text2))
 

	
 
    def test_that_exception_is_raised(self):
 
        text = '123456.789012' # single dot is not recognized
 
        with self.assertRaises(ValueError):
 
            parse_changesets(text)
 

	
 
    def test_non_alphanumeric_raises_exception(self):
 
        with self.assertRaises(ValueError):
 
            parse_changesets('aaa@bbb')
 

	
 

	
 
class TestParseDatetime(unittest.TestCase):
 

	
 
    def test_datetime_text(self):
 
        self.assertEqual(parse_datetime('2010-04-07 21:29:41'),
 
            datetime.datetime(2010, 4, 7, 21, 29, 41))
 

	
 
    def test_no_seconds(self):
 
        self.assertEqual(parse_datetime('2010-04-07 21:29'),
 
            datetime.datetime(2010, 4, 7, 21, 29))
 

	
 
    def test_date_only(self):
 
        self.assertEqual(parse_datetime('2010-04-07'),
 
            datetime.datetime(2010, 4, 7))
 

	
 
    def test_another_format(self):
 
        self.assertEqual(parse_datetime('04/07/10 21:29:41'),
 
            datetime.datetime(2010, 4, 7, 21, 29, 41))
 

	
 
    def test_now(self):
 
        self.assertTrue(parse_datetime('now') - datetime.datetime.now() <
 
            datetime.timedelta(seconds=1))
 

	
 
    def test_today(self):
 
        today = datetime.date.today()
 
        self.assertEqual(parse_datetime('today'),
 
            datetime.datetime(*today.timetuple()[:3]))
 

	
 
    def test_yesterday(self):
 
        yesterday = datetime.date.today() - datetime.timedelta(days=1)
 
        self.assertEqual(parse_datetime('yesterday'),
 
            datetime.datetime(*yesterday.timetuple()[:3]))
 

	
 
    def test_tomorrow(self):
 
        tomorrow = datetime.date.today() + datetime.timedelta(days=1)
 
        args = tomorrow.timetuple()[:3] + (23, 59, 59)
 
        self.assertEqual(parse_datetime('tomorrow'), datetime.datetime(*args))
 

	
 
    def test_days(self):
 
        timestamp = datetime.datetime.today() - datetime.timedelta(days=3)
 
        args = timestamp.timetuple()[:3] + (0, 0, 0, 0)
 
        expected = datetime.datetime(*args)
 
        self.assertEqual(parse_datetime('3d'), expected)
 
        self.assertEqual(parse_datetime('3 d'), expected)
 
        self.assertEqual(parse_datetime('3 day'), expected)
 
        self.assertEqual(parse_datetime('3 days'), expected)
 

	
 
    def test_weeks(self):
 
        timestamp = datetime.datetime.today() - datetime.timedelta(days=3 * 7)
 
        args = timestamp.timetuple()[:3] + (0, 0, 0, 0)
 
        expected = datetime.datetime(*args)
 
        self.assertEqual(parse_datetime('3w'), expected)
 
        self.assertEqual(parse_datetime('3 w'), expected)
 
        self.assertEqual(parse_datetime('3 week'), expected)
 
        self.assertEqual(parse_datetime('3 weeks'), expected)
 

	
 
    def test_mixed(self):
 
        timestamp = datetime.datetime.today() - datetime.timedelta(days=2 * 7 + 3)
 
        args = timestamp.timetuple()[:3] + (0, 0, 0, 0)
 
        expected = datetime.datetime(*args)
 
        self.assertEqual(parse_datetime('2w3d'), expected)
 
        self.assertEqual(parse_datetime('2w 3d'), expected)
 
        self.assertEqual(parse_datetime('2w 3 days'), expected)
 
        self.assertEqual(parse_datetime('2 weeks 3 days'), expected)
 

	
 

	
 
class TestAuthorExtractors(unittest.TestCase):
 
    TEST_AUTHORS = [("Username Last'o'Name <username@example.com>",
 
                    ("Username Last'o'Name", "username@example.com")),
 
                  ("Username Last'o'Name Spaces < username@example.com >",
 
                    ("Username Last'o'Name Spaces", "username@example.com")),
 
                  ("Username Last'o'Name <username.lastname@example.com>",
 
                    ("Username Last'o'Name", "username.lastname@example.com")),
 
                  ('mrf RFC_SPEC <username+lastname@example.com>',
 
                    ('mrf RFC_SPEC', 'username+lastname@example.com')),
 
                  ('username <user@example.com>',
 
                    ('username', 'user@example.com')),
 
                  ('username <user@example.com',
 
                   ('username', 'user@example.com')),
 
                  ('broken missing@example.com',
 
                   ('broken', 'missing@example.com')),
 
                  ('<justemail@example.com>',
 
                   ('', 'justemail@example.com')),
 
                  ('justname',
 
                   ('justname', '')),
 
                  ('Mr Double Name withemail@example.com ',
 
                   ('Mr Double Name', 'withemail@example.com')),
 
                  (u'John Doe <джондо à éẋàṁṗłê.ç°ḿ>',
 
                   (u'John Doe <\u0434\u0436\u043e\u043d\u0434\u043e \xe0 \xe9\u1e8b\xe0\u1e41\u1e57\u0142\xea.\xe7\xb0\u1e3f>', '')),
 
                  ]
 

	
 
    def test_author_email(self):
 
        for test_str, result in self.TEST_AUTHORS:
 
            self.assertEqual(result[1], author_email(test_str))
 

	
 
    def test_author_name(self):
 
        for test_str, result in self.TEST_AUTHORS:
 
            self.assertEqual(result[0], author_name(test_str))
 

	
 

	
 
class TestGetDictForAttrs(unittest.TestCase):
 

	
 
    def test_returned_dict_has_expected_attrs(self):
 
        obj = mock.Mock()
 
        obj.NOT_INCLUDED = 'this key/value should not be included'
 
        obj.CONST = True
 
        obj.foo = 'aaa'
 
        obj.attrs = {'foo': 'bar'}
 
        obj.date = datetime.datetime(2010, 12, 31)
 
        obj.count = 1001
 

	
 
        self.assertEqual(get_dict_for_attrs(obj, ['CONST', 'foo', 'attrs',
 
            'date', 'count']), {
 
            'CONST': True,
 
            'foo': 'aaa',
 
            'attrs': {'foo': 'bar'},
 
            'date': datetime.datetime(2010, 12, 31),
 
            'count': 1001,
 
        })
 

	
 

	
 
class TestGetTotalSeconds(unittest.TestCase):
 

	
 
    def assertTotalSecondsEqual(self, timedelta, expected_seconds):
 
        result = get_total_seconds(timedelta)
 
        self.assertEqual(result, expected_seconds,
 
            "We computed %s seconds for %s but expected %s"
 
            % (result, timedelta, expected_seconds))
 

	
 
    def test_get_total_seconds_returns_proper_value(self):
 
        self.assertTotalSecondsEqual(datetime.timedelta(seconds=1001), 1001)
 

	
 
    def test_get_total_seconds_returns_proper_value_for_partial_seconds(self):
 
        self.assertTotalSecondsEqual(datetime.timedelta(seconds=50.65), 50.65)
 

	
 

	
 
class TestGetUserHome(unittest.TestCase):
 

	
 
    @mock.patch.object(os, 'environ', {})
 
    def test_defaults_to_none(self):
 
        self.assertEqual(get_user_home(), '')
 

	
 
    @mock.patch.object(os, 'environ', {'HOME': '/home/foobar'})
 
    def test_unix_like(self):
 
        self.assertEqual(get_user_home(), '/home/foobar')
 

	
 
    @mock.patch.object(os, 'environ', {'USERPROFILE': '/Users/foobar'})
 
    def test_windows_like(self):
 
        self.assertEqual(get_user_home(), '/Users/foobar')
 

	
 
    @mock.patch.object(os, 'environ', {'HOME': '/home/foobar',
 
        'USERPROFILE': '/Users/foobar'})
 
    def test_prefers_home_over_userprofile(self):
 
        self.assertEqual(get_user_home(), '/home/foobar')
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_utils_filesize.py
Show inline comments
 

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

	
 

	
 
class TestFilesizeformat(unittest.TestCase):
 

	
 
    def test_bytes(self):
 
        self.assertEqual(filesizeformat(10), '10 B')
 

	
 
    def test_kilobytes(self):
 
        self.assertEqual(filesizeformat(1024 * 2), '2 KB')
 

	
 
    def test_megabytes(self):
 
        self.assertEqual(filesizeformat(1024 * 1024 * 2.3), '2.3 MB')
 

	
 
    def test_gigabytes(self):
 
        self.assertEqual(filesizeformat(1024 * 1024 * 1024 * 12.92), '12.92 GB')
 

	
 
    def test_that_function_respects_sep_parameter(self):
 
        self.assertEqual(filesizeformat(1, ''), '1B')
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
kallithea/tests/vcs/test_workdirs.py
Show inline comments
 

	
 
import datetime
 
from kallithea.lib.vcs.nodes import FileNode
 
from kallithea.lib.vcs.utils.compat import unittest
 
from kallithea.tests.vcs.base import _BackendTestMixin
 
from kallithea.tests.vcs.conf import SCM_TESTS
 

	
 

	
 
class WorkdirTestCaseMixin(_BackendTestMixin):
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        commits = [
 
            {
 
                'message': u'Initial commit',
 
                'author': u'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': u'Changes...',
 
                'author': u'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_get_branch_for_default_branch(self):
 
        self.assertEqual(self.repo.workdir.get_branch(),
 
            self.repo.DEFAULT_BRANCH_NAME)
 

	
 
    def test_get_branch_after_adding_one(self):
 
        self.imc.add(FileNode('docs/index.txt',
 
            content='Documentation\n'))
 
        self.imc.commit(
 
            message=u'New branch: foobar',
 
            author=u'joe',
 
            branch='foobar',
 
        )
 
        self.assertEqual(self.repo.workdir.get_branch(), self.default_branch)
 

	
 
    def test_get_changeset(self):
 
        old_head = self.repo.get_changeset()
 
        self.imc.add(FileNode('docs/index.txt',
 
            content='Documentation\n'))
 
        head = self.imc.commit(
 
            message=u'New branch: foobar',
 
            author=u'joe',
 
            branch='foobar',
 
        )
 
        self.assertEqual(self.repo.workdir.get_branch(), self.default_branch)
 
        self.repo.workdir.checkout_branch('foobar')
 
        self.assertEqual(self.repo.workdir.get_changeset(), head)
 

	
 
        # Make sure that old head is still there after update to default branch
 
        self.repo.workdir.checkout_branch(self.default_branch)
 
        self.assertEqual(self.repo.workdir.get_changeset(), old_head)
 

	
 
    def test_checkout_branch(self):
 
        from kallithea.lib.vcs.exceptions import BranchDoesNotExistError
 
        # first, 'foobranch' does not exist.
 
        self.assertRaises(BranchDoesNotExistError, self.repo.workdir.checkout_branch,
 
                          branch='foobranch')
 
        # create new branch 'foobranch'.
 
        self.imc.add(FileNode('file1', content='blah'))
 
        self.imc.commit(message=u'asd', author=u'john', branch='foobranch')
 
        # go back to the default branch
 
        self.repo.workdir.checkout_branch()
 
        self.assertEqual(self.repo.workdir.get_branch(), self.backend_class.DEFAULT_BRANCH_NAME)
 
        # checkout 'foobranch'
 
        self.repo.workdir.checkout_branch('foobranch')
 
        self.assertEqual(self.repo.workdir.get_branch(), 'foobranch')
 

	
 

	
 
# For each backend create test case class
 
for alias in SCM_TESTS:
 
    attrs = {
 
        'backend_alias': alias,
 
    }
 
    cls_name = ''.join(('%s branch test' % alias).title().split())
 
    bases = (WorkdirTestCaseMixin, unittest.TestCase)
 
    globals()[cls_name] = type(cls_name, bases, attrs)
 

	
 

	
 
if __name__ == '__main__':
 
    unittest.main()
0 comments (0 inline, 0 general)