"""

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

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)

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 \

# -*- 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..'), {

import os

import shutil

from kallithea.lib.utils2 import safe_str

from kallithea.lib.vcs import VCSError, get_repo, get_backend

from kallithea.lib.vcs.backends.hg import MercurialRepository

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

class VCSTest(unittest.TestCase):

    """

    Tests for main module's methods.

    """

    def test_get_backend(self):

        hg = get_backend('hg')

        self.assertEqual(hg, MercurialRepository)

    def test_alias_detect_hg(self):

        alias = 'hg'

        path = TEST_HG_REPO

        backend = get_backend(alias)

        repo = backend(safe_str(path))

        self.assertEqual('hg', repo.alias)

    def test_alias_detect_git(self):

        alias = 'git'

        path = TEST_GIT_REPO

        backend = get_backend(alias)

        repo = backend(safe_str(path))

        self.assertEqual('git', repo.alias)

        repo = backend(safe_str(path))

        self.assertEqual(repo.__class__, get_repo(safe_str(path), alias).__class__)

        self.assertEqual(repo.path, get_repo(safe_str(path), alias).path)

    def test_get_repo_autoalias_hg(self):

        alias = 'hg'

        path = TEST_HG_REPO

        backend = get_backend(alias)

        repo = backend(safe_str(path))

        self.assertEqual(repo.__class__, get_repo(safe_str(path)).__class__)

        self.assertEqual(repo.path, get_repo(safe_str(path)).path)

    def test_get_repo_autoalias_git(self):

        alias = 'git'

        path = TEST_GIT_REPO

        backend = get_backend(alias)

        repo = backend(safe_str(path))

        self.assertEqual(repo.__class__, get_repo(safe_str(path)).__class__)

        self.assertEqual(repo.path, get_repo(safe_str(path)).path)

    def test_get_repo_err(self):

        if os.path.isdir(blank_repo_path):

            shutil.rmtree(blank_repo_path)

        os.mkdir(blank_repo_path)

        self.assertRaises(VCSError, get_repo, blank_repo_path)

        self.assertRaises(VCSError, get_repo, blank_repo_path + 'non_existing')

    def test_get_repo_multialias(self):

        if os.path.isdir(multialias_repo_path):

            shutil.rmtree(multialias_repo_path)

        os.mkdir(multialias_repo_path)

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

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

        self.assertRaises(VCSError, get_repo, multialias_repo_path)