"""

Module providing backend independent mixin class. It requires that

InMemoryChangeset class is working properly at backend class.

"""

import os

import time

import datetime

import pytest

from kallithea.lib import vcs

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.conf import get_new_dir

class _BackendTestMixin(object):

    """

    This is a backend independent test case class which should be created

    with ``type`` method.

    - ``backend_alias``: alias of used backend (see ``vcs.BACKENDS``)

    - ``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_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

    @pytest.fixture(autouse=True,

                    scope='class')

    def _configure_backend(cls, request):

        Backend = vcs.get_backend(cls.backend_alias)

        cls.backend_class = Backend

        cls.setup_repo(Backend)

    @classmethod

    def setup_empty_repo(cls, backend):

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

        repo = backend(repo_path, create=True)

        return repo

    @classmethod

    def setup_repo(cls, backend):

        cls.repo = cls.setup_empty_repo(backend)

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

    @pytest.fixture(autouse=True)

    def _possibly_recreate_repo(self):

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

            self.setup_repo(self.backend_class)

# encoding: utf8

import time

import datetime

import pytest

from kallithea.lib import vcs

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

from kallithea.lib.vcs.nodes import (

    FileNode, AddedFileNodesGenerator,

    ChangedFileNodesGenerator, RemovedFileNodesGenerator

)

from kallithea.lib.vcs.exceptions import (

    BranchDoesNotExistError, ChangesetDoesNotExistError,

    RepositoryError, EmptyRepositoryError

)

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.tests.vcs.conf import get_new_dir

class TestBaseChangeset(object):

    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(u'foobar'), FileNode(u'blåbærgrød')]

        changeset.changed = []

        changeset.removed = []

        assert 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', u'bl\xe5b\xe6rgr\xf8d'],

            'changed': [],

            'removed': [],

        }

class _ChangesetsWithCommitsTestCaseixin(_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_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',

        )

        assert 'foobar' in self.repo.branches

        assert foobar_tip.branch == 'foobar'

        assert foobar_tip.branches == ['foobar']

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

        branch_tips = self.repo.branches.values()

        assert branch_tips.count(str(foobar_tip.raw_id)) == 1

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

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

        )

        assert newest_tip.branch == self.backend_class.DEFAULT_BRANCH_NAME

        assert newest_tip.branches == [self.backend_class.DEFAULT_BRANCH_NAME]

    def test_get_changesets_respects_branch_name(self):

        tip = self.repo.get_changeset()

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

            content='Documentation\n'))

        doc_changeset = self.imc.commit(

            message=u'New branch: docs',

            author=u'joe',

            branch='docs',

        )

        self.imc.add(vcs.nodes.FileNode('newfile', content=''))

        self.imc.commit(

            message=u'Back in default branch',

            author=u'joe',

            parents=[tip],

        )

        default_branch_changesets = self.repo.get_changesets(

            branch_name=self.repo.DEFAULT_BRANCH_NAME)

        assert doc_changeset not in default_branch_changesets

    def test_get_changeset_by_branch(self):

        for branch, sha in self.repo.branches.iteritems():

            assert sha == self.repo.get_changeset(branch).raw_id

    def test_get_changeset_by_tag(self):

        for tag, sha in self.repo.tags.iteritems():

            assert sha == self.repo.get_changeset(tag).raw_id

    def test_get_changeset_parents(self):

        for test_rev in [1, 2, 3]:

            sha = self.repo.get_changeset(test_rev-1)

            assert [sha] == self.repo.get_changeset(test_rev).parents

    def test_get_changeset_children(self):

        for test_rev in [1, 2, 3]:

            sha = self.repo.get_changeset(test_rev+1)

            assert [sha] == self.repo.get_changeset(test_rev).children

class _ChangesetsTestCaseMixin(_BackendTestMixin):

    recreate_repo_per_test = False

    @classmethod

    def _get_commits(cls):

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

        for x in xrange(5):

            yield {

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

                'author': u'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_simple(self):

        tip = self.repo.get_changeset()

        assert tip.date == datetime.datetime(2010, 1, 3, 20)

    def test_get_changesets_is_ordered_by_date(self):

        changesets = list(self.repo.get_changesets())

        ordered_by_date = sorted(changesets,

            key=lambda cs: cs.date)

        assert changesets == ordered_by_date

    def test_get_changesets_respects_start(self):

        second_id = self.repo.revisions[1]

        changesets = list(self.repo.get_changesets(start=second_id))

        assert len(changesets) == 4

    def test_get_changesets_numerical_id_respects_start(self):

        second_id = 1

        changesets = list(self.repo.get_changesets(start=second_id))

        assert len(changesets) == 4

    def test_get_changesets_includes_start_changeset(self):

        second_id = self.repo.revisions[1]

        changesets = list(self.repo.get_changesets(start=second_id))

        assert changesets[0].raw_id == second_id

    def test_get_changesets_respects_end(self):

        second_id = self.repo.revisions[1]

        changesets = list(self.repo.get_changesets(end=second_id))

        assert changesets[-1].raw_id == second_id

        assert len(changesets) == 2

    def test_get_changesets_numerical_id_respects_end(self):

        second_id = 1

        changesets = list(self.repo.get_changesets(end=second_id))

        assert changesets.index(changesets[-1]) == second_id

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

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

        assert len(changesets) == 2

    def test_get_changesets_on_empty_repo_raises_EmptyRepository_error(self):

        repo = self.setup_empty_repo(self.backend_class)

        with pytest.raises(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))

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

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

            assert cs.date >= start_date

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

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

# encoding: utf8

"""

Tests so called "in memory changesets" commit API of vcs.

"""

import time

import datetime

import pytest

from kallithea.lib import vcs

from kallithea.lib.vcs.exceptions import EmptyRepositoryError

from kallithea.lib.vcs.exceptions import NodeAlreadyAddedError

from kallithea.lib.vcs.exceptions import NodeAlreadyExistsError

from kallithea.lib.vcs.exceptions import NodeAlreadyRemovedError

from kallithea.lib.vcs.exceptions import NodeAlreadyChangedError

from kallithea.lib.vcs.exceptions import NodeDoesNotExistError

from kallithea.lib.vcs.exceptions import NodeNotChangedError

from kallithea.lib.vcs.nodes import DirNode

from kallithea.lib.vcs.nodes import FileNode

from kallithea.lib.vcs.utils import safe_unicode

from kallithea.tests.vcs.base import _BackendTestMixin

class InMemoryChangesetTestMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        # Note: this is slightly different than the regular _get_commits methods

        # as we don't actually return any commits. The creation of commits is

        # handled in the tests themselves.

        cls.nodes = [

            FileNode('foobar', content='Foo & bar'),

            FileNode('foobar2', content='Foo & bar, doubled!'),

            FileNode('foo bar with spaces', content=''),

            FileNode('foo/bar/baz', content='Inside'),

            FileNode('foo/bar/file.bin', content='\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00;\x00\x03\x00\xfe\xff\t\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x1a\x00\x00\x00\x00\x00\x00\x00\x00\x10\x00\x00\x18\x00\x00\x00\x01\x00\x00\x00\xfe\xff\xff\xff\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff'),

        ]

        commits = []

        return commits

    def test_add(self):

        rev_count = len(self.repo.revisions)

        to_add = [FileNode(node.path, content=node.content)

            for node in self.nodes]

        for node in to_add:

            self.imc.add(node)

        message = u'Added: %s' % ', '.join((node.path for node in self.nodes))

        author = unicode(self.__class__)

        changeset = self.imc.commit(message=message, author=author)

        newtip = self.repo.get_changeset()

        assert changeset == newtip

        assert rev_count + 1 == len(self.repo.revisions)

        assert newtip.message == message

        assert newtip.author == author

        assert not any((

            self.imc.added,

            self.imc.changed,

            self.imc.removed

        ))

        for node in to_add:

            assert newtip.get_node(node.path).content == node.content

    def test_add_in_bulk(self):

        rev_count = len(self.repo.revisions)

        to_add = [FileNode(node.path, content=node.content)

            for node in self.nodes]

        self.imc.add(*to_add)

        message = u'Added: %s' % ', '.join((node.path for node in self.nodes))

        author = unicode(self.__class__)

        changeset = self.imc.commit(message=message, author=author)

        newtip = self.repo.get_changeset()

        assert changeset == newtip

        assert rev_count + 1 == len(self.repo.revisions)

        assert newtip.message == message

        assert newtip.author == author

        assert not any((

            self.imc.added,

            self.imc.changed,

            self.imc.removed

        ))

        for node in to_add:

            assert newtip.get_node(node.path).content == node.content

    def test_add_actually_adds_all_nodes_at_second_commit_too(self):

        self.imc.add(FileNode('foo/bar/image.png', content='\0'))

        self.imc.add(FileNode('foo/README.txt', content='readme!'))

        changeset = self.imc.commit(u'Initial', u'joe.doe@example.com')

        assert isinstance(changeset.get_node('foo'), DirNode)

        assert isinstance(changeset.get_node('foo/bar'), DirNode)

        assert changeset.get_node('foo/bar/image.png').content == '\0'

        assert changeset.get_node('foo/README.txt').content == 'readme!'

        # commit some more files again

        to_add = [

            FileNode('foo/bar/foobaz/bar', content='foo'),

            FileNode('foo/bar/another/bar', content='foo'),

            FileNode('foo/baz.txt', content='foo'),

            FileNode('foobar/foobaz/file', content='foo'),

            FileNode('foobar/barbaz', content='foo'),

        ]

        self.imc.add(*to_add)

        changeset = self.imc.commit(u'Another', u'joe.doe@example.com')

        changeset.get_node('foo/bar/foobaz/bar').content == 'foo'

        changeset.get_node('foo/bar/another/bar').content == 'foo'

        changeset.get_node('foo/baz.txt').content == 'foo'

        changeset.get_node('foobar/foobaz/file').content == 'foo'

        changeset.get_node('foobar/barbaz').content == 'foo'

    def test_add_non_ascii_files(self):

        rev_count = len(self.repo.revisions)

        to_add = [

            FileNode('żółwik/zwierzątko', content='ćććć'),

            FileNode(u'żółwik/zwierzątko_uni', content=u'ćććć'),

        ]

        for node in to_add:

            self.imc.add(node)

        message = u'Added: %s' % ', '.join((node.path for node in self.nodes))

        author = unicode(self.__class__)

        changeset = self.imc.commit(message=message, author=author)

        newtip = self.repo.get_changeset()

        assert changeset == newtip

        assert rev_count + 1 == len(self.repo.revisions)

        assert newtip.message == message

        assert newtip.author == author

        assert not any((

            self.imc.added,

            self.imc.changed,

            self.imc.removed

        ))

        for node in to_add:

            assert newtip.get_node(node.path).content == node.content

    def test_add_raise_already_added(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        with pytest.raises(NodeAlreadyAddedError):

            self.imc.add(node)

    def test_check_integrity_raise_already_exist(self):

        node = FileNode('foobar', content='baz')

        self.imc.add(node)

        self.imc.commit(message=u'Added foobar', author=unicode(self))

        self.imc.add(node)

        with pytest.raises(NodeAlreadyExistsError):

            self.imc.commit(message='new message',

                            author=str(self))

    def test_change(self):

        self.imc.add(FileNode('foo/bar/baz', content='foo'))

        self.imc.add(FileNode('foo/fbar', content='foobar'))

        tip = self.imc.commit(u'Initial', u'joe.doe@example.com')

        # Change node's content

        node = FileNode('foo/bar/baz', content='My **changed** content')

        self.imc.change(node)

        self.imc.commit(u'Changed %s' % node.path, u'joe.doe@example.com')

        newtip = self.repo.get_changeset()

        assert tip != newtip

        assert tip.id != newtip.id

        assert newtip.get_node('foo/bar/baz').content == 'My **changed** content'

    def test_change_non_ascii(self):

        to_add = [

            FileNode('żółwik/zwierzątko', content='ćććć'),

            FileNode(u'żółwik/zwierzątko_uni', content=u'ćććć'),

        ]

        for node in to_add:

            self.imc.add(node)

        tip = self.imc.commit(u'Initial', u'joe.doe@example.com')

        # Change node's content

        node = FileNode('żółwik/zwierzątko', content='My **changed** content')

        self.imc.change(node)

        self.imc.commit(u'Changed %s' % safe_unicode(node.path),

                        u'joe.doe@example.com')

        node = FileNode(u'żółwik/zwierzątko_uni', content=u'My **changed** content')

        self.imc.change(node)

        self.imc.commit(u'Changed %s' % safe_unicode(node.path),

                        u'joe.doe@example.com')

        newtip = self.repo.get_changeset()

        assert tip != newtip

        assert tip.id != newtip.id

        assert newtip.get_node('żółwik/zwierzątko').content == 'My **changed** content'

        assert newtip.get_node('żółwik/zwierzątko_uni').content == 'My **changed** content'

    def test_change_raise_empty_repository(self):

        node = FileNode('foobar')

        with pytest.raises(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)

        with pytest.raises(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)

        with pytest.raises(NodeAlreadyChangedError):

            self.imc.change(node)

    def test_check_integrity_change_raise_node_not_changed(self):