#LOAD VCS test stuff

    from rhodecode.tests.vcs import setup_package

    setup_package()

                                  'an <unicode> instance got %r & %r instead'

                                  % (type(message), type(author)))

                                 getattr(self.changeset, 'short_id', ''))

                                 getattr(self.changeset, 'short_id', ''))

                                 getattr(self.changeset, 'short_id', ''))

import datetime

import hashlib

import tempfile

    'TEST_USER_ADMIN_LOGIN', 'TEST_USER_REGULAR_LOGIN',

    'TEST_USER_REGULAR_PASS', 'TEST_USER_REGULAR_EMAIL',

    'TEST_USER_REGULAR2_LOGIN', 'TEST_USER_REGULAR2_PASS',

    'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO', 'TEST_GIT_REPO',

    'HG_REMOTE_REPO', 'GIT_REMOTE_REPO', 'SCM_TESTS',

## VCS

SCM_TESTS = ['hg', 'git']

uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))

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

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

TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)

TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)

TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)

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

TEST_HG_REPO = jn(TESTS_TMP_PATH, 'vcs-hg')

TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)

TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)

TEST_DIR = tempfile.gettempdir()

TEST_REPO_PREFIX = 'vcs-test'

def get_new_dir(title):

    """

    Returns always new directory path.

    """

    from rhodecode.tests.vcs.utils import get_normalized_path

    name = TEST_REPO_PREFIX

    if title:

        name = '-'.join((name, title))

    hex = hashlib.sha1(str(time.time())).hexdigest()

    name = '-'.join((name, hex))

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

    return get_normalized_path(path)

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

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

TEST_USER_CONFIG_FILE = jn(THIS, 'aconfig')

                                 {'username': username,

                                  'password': password})

    def test_file_annotation_git(self):

        self.log_user()

        response = self.app.get(url(controller='files', action='index',

                                    repo_name=GIT_REPO,

                                    revision='master',

                                    f_path='vcs/nodes.py',

                                    annotate=True))

                                    {'q': HG_REPO})

                                {'q': 'def repo'})

        response.mustcontain('10 results')

        response.mustcontain('Permission denied')

                                {'q': 'repository:%s def test' % HG_REPO})

        response.mustcontain('4 results')

        response.mustcontain('Permission denied')

            """title="Mercurial repository" alt="Mercurial repository" """

            """src="/images/icons/hgicon.png"/>"""

        response.mustcontain("""<input style="width:80%%;margin-left:105px" type="text" id="clone_url" readonly="readonly" value="http://test_admin@localhost:80/%s"/>""" % HG_REPO)

        response.mustcontain("""<input style="display:none;width:80%%;margin-left:105px" type="text" id="clone_url_id" readonly="readonly" value="http://test_admin@localhost:80/_%s"/>""" % ID)

    def test_index_git(self):

        self.log_user()

        ID = Repository.get_by_repo_name(GIT_REPO).repo_id

        response = self.app.get(url(controller='summary',

                                    action='index',

                                    repo_name=GIT_REPO))

        #repo type

        response.mustcontain(

            """<img style="margin-bottom:2px" class="icon" """

            """title="Git repository" alt="Git repository" """

            """src="/images/icons/giticon.png"/>"""

        )

        response.mustcontain(

            """<img style="margin-bottom:2px" class="icon" """

            """title="public repository" alt="public """

            """repository" src="/images/icons/lock_open.png"/>"""

        )

        # clone url...

        response.mustcontain("""<input style="width:80%%;margin-left:105px" type="text" id="clone_url" readonly="readonly" value="http://test_admin@localhost:80/%s"/>""" % GIT_REPO)

        response.mustcontain("""<input style="display:none;width:80%%;margin-left:105px" type="text" id="clone_url_id" readonly="readonly" value="http://test_admin@localhost:80/_%s"/>""" % ID)

    def test_index_by_id_hg(self):

    def test_index_by_id_git(self):

        self.log_user()

        ID = Repository.get_by_repo_name(GIT_REPO).repo_id

        response = self.app.get(url(controller='summary',

                                    action='index',

                                    repo_name='_%s' % ID))

        #repo type

        response.mustcontain("""<img style="margin-bottom:2px" class="icon" """

                        """title="Git repository" alt="Git """

                        """repository" src="/images/icons/hgicon.png"/>""")

        response.mustcontain("""<img style="margin-bottom:2px" class="icon" """

                        """title="public repository" alt="public """

                        """repository" src="/images/icons/lock_open.png"/>""")

"""

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

"""

import os

from rhodecode.lib import vcs

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

from rhodecode.tests import *

from utils import VCSTestError, SCMFetcher

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

        },

        'git': {

            'alias': 'git',

            'test_repo_path': TEST_GIT_REPO,

            'remote_repo': GIT_REMOTE_REPO,

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

        },

    }

    try:

        for scm, fetcher_info in fetchers.items():

            fetcher = SCMFetcher(**fetcher_info)

            fetcher.setup()

    except VCSTestError, err:

        raise RuntimeError(str(err))

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

unittest.defaultTestLoader.discover(start_dir)

[user]

name = Foo Bar

email = foo.bar@example.com

[ui]

username = Foo Bar foo.bar@example.com

[universal]

foo = bar

"""

Module providing backend independent mixin class. It requires that

InMemoryChangeset class is working properly at backend class.

"""

import os

from rhodecode.lib import vcs

import time

import shutil

import datetime

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

from conf import SCM_TESTS, get_new_dir

from rhodecode.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

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

"""

Unit tests configuration module for vcs.

"""

import os

import time

import hashlib

import tempfile

import datetime

from utils import get_normalized_path

from os.path import join as jn

__all__ = (

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

    'SCM_TESTS',

)

SCM_TESTS = ['hg', 'git']

uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))

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

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

TEST_TMP_PATH = os.environ.get('VCS_TEST_ROOT', '/tmp')

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

                              jn(TEST_TMP_PATH, 'vcs-git'))

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

                            jn(TEST_TMP_PATH, 'vcsgitclone%s' % uniq_suffix))

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

                            jn(TEST_TMP_PATH, 'vcsgitpull%s' % uniq_suffix))

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

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

                              jn(TEST_TMP_PATH, 'vcs-hg'))

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

                              jn(TEST_TMP_PATH, 'vcshgclone%s' % uniq_suffix))

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

                              jn(TEST_TMP_PATH, 'vcshgpull%s' % uniq_suffix))

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

TEST_REPO_PREFIX = 'vcs-test'

def get_new_dir(title):

    """

    Returns always new directory path.

    """

    name = TEST_REPO_PREFIX

    if title:

        name = '-'.join((name, title))

    hex = hashlib.sha1(str(time.time())).hexdigest()

    name = '-'.join((name, hex))

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

    return get_normalized_path(path)

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

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

TEST_USER_CONFIG_FILE = jn(THIS, 'aconfig')

from __future__ import with_statement

import os

import tarfile

import zipfile

import datetime

import tempfile

import StringIO

from base import BackendTestMixin

from conf import SCM_TESTS

from rhodecode.lib.vcs.exceptions import VCSError

from rhodecode.lib.vcs.nodes import FileNode

from rhodecode.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()[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()[1]

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

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

        outdir = tempfile.mkdtemp()

        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()[1]

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

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

        outdir = tempfile.mkdtemp()

        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()[1]

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

            self.tip.fill_archive(stream=stream)

        mystream = StringIO.StringIO()

        self.tip.fill_archive(stream=mystream)

        mystream.seek(0)

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

            self.assertEqual(f.read(), mystream.read())

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

from __future__ import with_statement

from rhodecode.lib import vcs

import datetime

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

from base import BackendTestMixin

from conf import SCM_TESTS

from rhodecode.lib.vcs.nodes import FileNode

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

from __future__ import with_statement

from rhodecode.lib import vcs

import datetime

from base import BackendTestMixin

from conf import SCM_TESTS

from rhodecode.lib.vcs.backends.base import BaseChangeset

from rhodecode.lib.vcs.nodes import FileNode

from rhodecode.lib.vcs.exceptions import BranchDoesNotExistError

from rhodecode.lib.vcs.exceptions import ChangesetDoesNotExistError

from rhodecode.lib.vcs.exceptions import RepositoryError

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

class TestBaseChangeset(unittest.TestCase):

    def test_as_dict(self):

        changeset = BaseChangeset()

        changeset.id = 'ID'

        changeset.raw_id = 'RAW_ID'

        changeset.short_id = 'SHORT_ID'

        changeset.revision = 1009

        changeset.date = datetime.datetime(2011, 1, 30, 1, 45)

        changeset.message = 'Message of a commit'

        changeset.author = 'Joe Doe <joe.doe@example.com>'

        changeset.added = [FileNode('foo/bar/baz'), FileNode('foobar')]

        changeset.changed = []

        changeset.removed = []

        self.assertEqual(changeset.as_dict(), {

            'id': 'ID',

            'raw_id': 'RAW_ID',

            'short_id': 'SHORT_ID',

            'revision': 1009,

            'date': datetime.datetime(2011, 1, 30, 1, 45),

            'message': 'Message of a commit',

            'author': {

                'name': 'Joe Doe',

                'email': 'joe.doe@example.com',

            },

            'added': ['foo/bar/baz', 'foobar'],

            'changed': [],

            'removed': [],

        })

class ChangesetsWithCommitsTestCaseixin(BackendTestMixin):

    recreate_repo_per_test = True

    @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_new_branch(self):

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

        # 'foobar' should be the only branch that contains the new commit

        self.assertNotEqual(*self.repo.branches.values())

    def test_new_head_in_default_branch(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',