from kallithea.model.db import Notification, User, UserNotification

from kallithea.model.meta import Session

from kallithea.lib.utils2 import safe_str

log = logging.getLogger(__name__)

skipif = pytest.mark.skipif

parametrize = pytest.mark.parametrize

# Hack: These module global values MUST be set to actual values before running any tests. This is currently done by conftest.py.

url = None

testapp = None

__all__ = [

    'skipif', 'parametrize', 'url', 'TestController',

    'ldap_lib_installed', 'pam_lib_installed', 'invalidate_all_caches',

    'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',

    'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',

    'TEST_USER_ADMIN_EMAIL', '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_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',

    'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',

]

# Invoke websetup with the current config file

# SetupCommand('setup-app').run([config_file])

## SOME GLOBALS FOR TESTS

TESTS_TMP_PATH = os.environ.get('KALLITHEA_TESTS_TMP_PATH', tempfile.mkdtemp(prefix='kallithea-test-'))

TEST_USER_ADMIN_LOGIN = 'test_admin'

TEST_USER_ADMIN_PASS = 'test12'

TEST_USER_ADMIN_EMAIL = 'test_admin@example.com'

TEST_USER_REGULAR_LOGIN = 'test_regular'

TEST_USER_REGULAR_PASS = 'test12'

TEST_USER_REGULAR_EMAIL = 'test_regular@example.com'

TEST_USER_REGULAR2_LOGIN = 'test_regular2'

TEST_USER_REGULAR2_PASS = 'test12'

TEST_USER_REGULAR2_EMAIL = 'test_regular2@example.com'

HG_REPO = u'vcs_test_hg'

GIT_REPO = u'vcs_test_git'

NEW_HG_REPO = u'vcs_test_hg_new'

NEW_GIT_REPO = u'vcs_test_git_new'

HG_FORK = u'vcs_test_hg_fork'

GIT_FORK = u'vcs_test_git_fork'

HG_TEST_REVISION = u"a53d9201d4bc278910d416d94941b7ea007ecd52"

GIT_TEST_REVISION = u"7ab37bc680b4aa72c34d07b230c866c28e9fc204"

## VCS

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

GIT_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)

TEST_GIT_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)

TEST_GIT_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcs-git-clone-%s' % uniq_suffix)

TEST_GIT_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcs-git-pull-%s' % uniq_suffix)

HG_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)

TEST_HG_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)

TEST_HG_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcs-hg-clone-%s' % uniq_suffix)

TEST_HG_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcs-hg-pull-%s' % uniq_suffix)

# By default, some of the tests will utilise locally available

# repositories stored within tar.gz archives as source for

# cloning. Should you wish to use some other, remote archive, simply

# uncomment these entries and/or update the URLs to use.

#

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

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

# skip ldap tests if LDAP lib is not installed

ldap_lib_installed = False

"""

Unit tests for vcs_ library.

For each SCM we run tests for, we need some repository. We would use

repositories location provided in 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.

"""

import os

import shutil

from kallithea.tests.base import TEST_HG_REPO, HG_REMOTE_REPO, TEST_GIT_REPO, GIT_REMOTE_REPO, TESTS_TMP_PATH

from kallithea.tests.vcs.utils import SCMFetcher

# Base directory for the VCS tests.

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

# Path to user configuration file used during tests.

TEST_USER_CONFIG_FILE = os.path.join(TESTS_TMP_PATH, 'aconfig')

"""

Module providing backend independent mixin class. It requires that

InMemoryChangeset class is working properly at backend class.

"""

import os

import time

import datetime

from kallithea.lib import vcs

from kallithea.lib.vcs.nodes import FileNode

class _BackendTestMixin(object):

    """

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

    with ``type`` method.

    It is required to set following attributes at subclass:

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

    - ``repo_path``: path to the repository which would be created for set of

      tests

    - ``recreate_repo_per_test``: If set to ``False``, repo would NOT be created

      before every single test. Defaults to ``True``.

    """

    recreate_repo_per_test = True

    @classmethod

    def get_backend(cls):

        return vcs.get_backend(cls.backend_alias)

    @classmethod

    def _get_commits(cls):

        commits = [

    @classmethod

    def setup_class(cls):

        Backend = cls.get_backend()

        cls.backend_class = Backend

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

        cls.repo = Backend(cls.repo_path, create=True)

        cls.imc = cls.repo.in_memory_changeset

        cls.default_branch = cls.repo.DEFAULT_BRANCH_NAME

        for commit in cls._get_commits():

            for node in commit.get('added', []):

                cls.imc.add(FileNode(node.path, content=node.content))

            for node in commit.get('changed', []):

                cls.imc.change(FileNode(node.path, content=node.content))

            for node in commit.get('removed', []):

                cls.imc.remove(FileNode(node.path))

            cls.tip = cls.imc.commit(message=unicode(commit['message']),

                                     author=unicode(commit['author']),

                                     date=commit['date'])

    def setup_method(self, method):

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

            self.__class__.setup_class()

"""

Unit tests configuration module for vcs.

"""

import os

import uuid

# Retrieve the necessary configuration options from the test base

# module. Some of these configuration options are subsequently

# consumed by the VCS test module.

from kallithea.tests.base import (

    TEST_HG_REPO, HG_REMOTE_REPO,

    TEST_HG_REPO_CLONE, TEST_HG_REPO_PULL,

    TEST_GIT_REPO, GIT_REMOTE_REPO,

    TEST_GIT_REPO_CLONE,

)

__all__ = (

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

    'TEST_HG_REPO_CLONE', 'TEST_GIT_REPO_CLONE', 'TEST_HG_REPO_PULL',

)

def get_new_dir(title=None):

    """

    Calculates a path for a new, non-existant, unique sub-directory in TESTS_TMP_PATH.

    Resulting directory name will have format:

    vcs-test-[title-]hexuuid

    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.

    """

    test_repo_prefix = 'vcs-test'

import os

import tarfile

import zipfile

import datetime

import tempfile

import StringIO

import pytest

from kallithea.lib.vcs.exceptions import VCSError

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.base import _BackendTestMixin

class ArchivesTestCaseMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

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

        for x in xrange(5):

            yield {

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

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('%d/file_%d.txt' % (x, x),

                        content='Foobar %d' % x),

                ],

            }

    def test_archive_zip(self):

        path = tempfile.mkstemp(dir=TESTS_TMP_PATH, prefix='test_archive_zip-')[1]

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

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

        out = zipfile.ZipFile(path)

        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

            assert file_content[:4] == stringio_content[:4]

            assert file_content[8:] == stringio_content[8:]

    def test_archive_wrong_kind(self):

        with pytest.raises(VCSError):

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

    def test_archive_empty_prefix(self):

        with pytest.raises(VCSError):

            self.tip.fill_archive(prefix='')

    def test_archive_prefix_with_leading_slash(self):

        with pytest.raises(VCSError):

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

import datetime

from kallithea.lib import vcs

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.base import _BackendTestMixin

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

                ],

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

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

        assert 'issue/123' in self.repo.branches

        assert '123' in self.repo.branches

# 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

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': {

        changeset = self.repo.get_changeset()

        assert isinstance(changeset.added, AddedFileNodesGenerator)

        assert list(changeset.added) == [

            changeset.get_node('fallout'),

        ]

        assert isinstance(changeset.changed, ChangedFileNodesGenerator)

        assert list(changeset.changed) == [

            changeset.get_node('foo/bar'),

            changeset.get_node('foobar'),

        ]

        assert isinstance(changeset.removed, RemovedFileNodesGenerator)

        assert len(changeset.removed) == 1

        assert list(changeset.removed)[0].path == 'qwe'

    def test_get_filemode(self):

        changeset = self.repo.get_changeset()

        assert 33188 == changeset.get_file_mode('foo/bar')

    def test_get_filemode_non_ascii(self):

        changeset = self.repo.get_changeset()

        assert 33188 == changeset.get_file_mode('foo/bał')

        assert 33188 == changeset.get_file_mode(u'foo/bał')

# encoding: utf8

import datetime

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.test_inmemchangesets import BackendBaseTestCase

class FileNodeUnicodePathTestsMixin(object):

    fname = 'ąśðąęłąć.txt'

    ufname = (fname).decode('utf-8')

    def get_commits(self):

        self.nodes = [

            FileNode(self.fname, content='Foobar'),

        ]

        commits = [

            {

                'message': 'Initial commit',

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': datetime.datetime(2010, 1, 1, 20),

                'added': self.nodes,

            },

        ]

        return commits

    def test_filenode_path(self):

        node = self.tip.get_node(self.fname)

        unode = self.tip.get_node(self.ufname)

        assert node == unode

import datetime

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.lib.vcs.nodes import FileNode

class GetitemTestCaseMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

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

        for x in xrange(5):

            yield {

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

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),

                ],

            }

    def test__getitem__last_item_is_tip(self):

        assert self.repo[-1] == self.repo.get_changeset()

    def test__getitem__returns_correct_items(self):

        changesets = [self.repo[x] for x in xrange(len(self.repo.revisions))]

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

import datetime

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.base import _BackendTestMixin

class GetsliceTestCaseMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

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

        for x in xrange(5):

            yield {

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

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),

                ],

            }

    def test__getslice__last_item_is_tip(self):

        assert list(self.repo[-1:])[0] == self.repo.get_changeset()

    def test__getslice__respects_start_index(self):

        assert list(self.repo[2:]) == [self.repo.get_changeset(rev) for rev in self.repo.revisions[2:]]

    def test__getslice__respects_negative_start_index(self):

        assert list(self.repo[-2:]) == [self.repo.get_changeset(rev) for rev in self.repo.revisions[-2:]]

    def test__getslice__respects_end_index(self):

        assert list(self.repo[:2]) == [self.repo.get_changeset(rev) for rev in self.repo.revisions[:2]]

    def test__getslice__respects_negative_end_index(self):

        assert list(self.repo[:-2]) == [self.repo.get_changeset(rev) for rev in self.repo.revisions[:-2]]

# 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

class InMemoryChangesetTestMixin(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

    """

    def get_backend(self):

        return vcs.get_backend(self.backend_alias)

    def setup_method(self):

        Backend = self.get_backend()

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

        self.repo = Backend(self.repo_path, create=True)

        self.imc = self.repo.in_memory_changeset

        self.nodes = [

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

        repo_path = get_new_dir(key)

        return repo_path

    def setup_method(self):

        Backend = self.get_backend()

        self.backend_class = Backend

        self.repo_path = self.get_new_repo_path()

        self.repo = Backend(self.repo_path, create=True)

        self.imc = self.repo.in_memory_changeset

        for commit in self.get_commits():

            for node in commit.get('added', []):

                self.imc.add(FileNode(node.path, content=node.content))

            for node in commit.get('changed', []):

                self.imc.change(FileNode(node.path, content=node.content))

            for node in commit.get('removed', []):

                self.imc.remove(FileNode(node.path))

            self.imc.commit(message=unicode(commit['message']),

                            author=unicode(commit['author']),

                date=commit['date'])

        self.tip = self.repo.get_changeset()

import datetime

import pytest

from kallithea.lib.vcs.nodes import FileNode

from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.tests.vcs import TEST_USER_CONFIG_FILE

class RepositoryBaseTest(_BackendTestMixin):

    recreate_repo_per_test = False

    @classmethod

    def _get_commits(cls):

        return super(RepositoryBaseTest, cls)._get_commits()[:1]

    def test_get_config_value(self):

        assert self.repo.get_config_value('universal', 'foo', TEST_USER_CONFIG_FILE) == 'bar'

    def test_get_config_value_defaults_to_None(self):

        assert self.repo.get_config_value('universal', 'nonexist', TEST_USER_CONFIG_FILE) == None

    def test_get_user_name(self):

        assert self.repo.get_user_name(TEST_USER_CONFIG_FILE) == 'Foo Bar'

    def test_get_user_email(self):

        assert self.repo.get_user_email(TEST_USER_CONFIG_FILE) == 'foo.bar@example.com'

    def test_repo_equality(self):

        assert self.repo == self.repo

    def test_repo_equality_broken_object(self):

        import copy

        _repo = copy.copy(self.repo)

        delattr(_repo, 'path')

        assert self.repo != _repo

    def test_repo_equality_other_object(self):

        class dummy(object):

            path = self.repo.path

        assert self.repo != dummy()

class RepositoryGetDiffTest(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        commits = [

            {

                'message': 'Initial commit',

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': datetime.datetime(2010, 1, 1, 20),

                'added': [

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

                    FileNode('foobar2', content='foobar2'),

                ],

            },

            {

                'message': 'Changed foobar, added foobar3',

                'author': 'Jane Doe <jane.doe@example.com>',

                'date': datetime.datetime(2010, 1, 1, 21),

                'added': [

                    FileNode('foobar3', content='foobar3'),

                ],

                'changed': [

                    FileNode('foobar', 'FOOBAR'),

                ],

@@ -1,1 +0,0 @@

-FOOBAR

\ No newline at end of file

diff --git a/foobar3 b/foobar3

--- a/foobar3

+++ b/foobar3

@@ -1,1 +1,3 @@

-foobar3

\ No newline at end of file

+FOOBAR

+FOOBAR

+FOOBAR

'''

    def test_fourth_changeset_diff(self):

        revs = self.repo.revisions

        assert self.repo.get_diff(revs[2], revs[3]) == '''diff --git a/README{ b/README{

new file mode 100644

--- /dev/null

+++ b/README{

@@ -0,0 +1,1 @@

+Strangely-named README file

\ No newline at end of file

'''