"""

Unit tests configuration module for vcs.

"""

import os

import tempfile

import shutil

import uuid

__all__ = (

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

    'SCM_TESTS',

)

SCM_TESTS = ['hg', 'git']

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

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

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

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

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

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

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

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

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

TEST_REPO_PREFIX = 'vcs-test'

def get_new_dir(title=None):

    """

    Resulting directory name will have format:

    prefix-[title-]hexuuid

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

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

    specified.

    Args:

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

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

            to None.

    Returns:

        Path to the new directory as a string.

    """

    if title:

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

    else:

        name = TEST_REPO_PREFIX

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

    hex_uuid = uuid.uuid4().hex

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

        hex_uuid = uuid.uuid4().hex

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

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

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

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

TEST_USER_CONFIG_FILE = _dest

# overide default configurations with kallithea ones

from kallithea.tests.base import *

import os

import tarfile

import zipfile

import datetime

import tempfile

import StringIO

from kallithea.tests.vcs.base import _BackendTestMixin

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

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

        with open(path, 'wb') as f:

            self.tip.fill_archive(stream=f, kind='tgz', prefix='repo')

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

        with open(path, 'w+b') as f:

            self.tip.fill_archive(stream=f, kind='tbz2', prefix='repo')

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

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

            self.tip.fill_archive(stream=stream)

        mystream = StringIO.StringIO()

        self.tip.fill_archive(stream=mystream)

        mystream.seek(0)

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

            file_content = f.read()

            stringio_content = mystream.read()

            # the gzip header contains a MTIME header

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

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

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

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

    def test_archive_wrong_kind(self):

        with self.assertRaises(VCSError):

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

    def test_archive_empty_prefix(self):

        with self.assertRaises(VCSError):

            self.tip.fill_archive(prefix='')

    def test_archive_prefix_with_leading_slash(self):

        with self.assertRaises(VCSError):

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

# For each backend create test case class

for alias in SCM_TESTS:

    attrs = {

        'backend_alias': alias,

    }

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

    bases = (ArchivesTestCaseMixin, unittest.TestCase)

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

import 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

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

        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

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

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

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

            self.assertTrue(isinstance(

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

    def _test_single_changeset_cache(self, revision):

        chset = self.repo.get_changeset(revision)

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

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

    def test_initial_changeset(self):

        id = self.repo.revisions[0]

        init_chset = self.repo.get_changeset(id)

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

        self.assertEqual(init_chset.author,

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

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

                     'vcs/backends/BaseRepository.py',

                     'vcs/backends/__init__.py'):

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

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

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

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

        node = init_chset.get_node('vcs/')

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

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

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

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

        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)

        node = init_chset.get_node('vcs')

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

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

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

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

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

    def test_not_existing_changeset(self):

        # rawid

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

            'abcd' * 10)

        # shortid

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

            'erro' * 4)

        # numeric

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

            self.repo.count() + 1)

        # Small chance we ever get to this one

        revision = pow(2, 30)

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

    def test_changeset10(self):

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

import os

import mock

import time

import shutil

import datetime

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

from kallithea.lib.vcs.utils.paths import get_dirs_for_path

from kallithea.lib.vcs.utils.helpers import get_dict_for_attrs

from kallithea.lib.vcs.utils.helpers import get_scm

from kallithea.lib.vcs.utils.helpers import get_scms_for_path

from kallithea.lib.vcs.utils.helpers import get_total_seconds

from kallithea.lib.vcs.utils.helpers import parse_changesets

from kallithea.lib.vcs.utils.helpers import parse_datetime

from kallithea.lib.vcs.utils import author_email, author_name

from kallithea.lib.vcs.utils.paths import get_user_home

from kallithea.lib.vcs.exceptions import VCSError

class PathsTest(unittest.TestCase):

    def _test_get_dirs_for_path(self, path, expected):

        """

        Tests if get_dirs_for_path returns same as expected.

        """

        expected = sorted(expected)

        result = sorted(get_dirs_for_path(path))

        self.assertEqual(result, expected,

            msg="%s != %s which was expected result for path %s"

            % (result, expected, path))

    def test_get_dirs_for_path(self):

        path = 'foo/bar/baz/file'

        paths_and_results = (

            ('foo/bar/baz/file', ['foo', 'foo/bar', 'foo/bar/baz']),

            ('foo/bar/', ['foo', 'foo/bar']),

            ('foo/bar', ['foo']),

        )

        for path, expected in paths_and_results:

            self._test_get_dirs_for_path(path, expected)

    def test_get_scm(self):

        self.assertEqual(('hg', TEST_HG_REPO), get_scm(TEST_HG_REPO))

        self.assertEqual(('git', TEST_GIT_REPO), get_scm(TEST_GIT_REPO))

    def test_get_two_scms_for_path(self):

        if os.path.isdir(multialias_repo_path):

            shutil.rmtree(multialias_repo_path)

        os.mkdir(multialias_repo_path)

        self.assertRaises(VCSError, get_scm, multialias_repo_path)

    def test_get_scm_error_path(self):

        self.assertRaises(VCSError, get_scm, 'err')

    def test_get_scms_for_path(self):

        os.mkdir(new)

        self.assertEqual(get_scms_for_path(new), [])

        os.mkdir(os.path.join(new, '.tux'))

        self.assertEqual(get_scms_for_path(new), [])

        os.mkdir(os.path.join(new, '.git'))

        self.assertEqual(set(get_scms_for_path(new)), set(['git']))

        os.mkdir(os.path.join(new, '.hg'))

        self.assertEqual(set(get_scms_for_path(new)), set(['git', 'hg']))

class TestParseChangesets(unittest.TestCase):

    def test_main_is_returned_correctly(self):

        self.assertEqual(parse_changesets('123456'), {

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