def __init__(self, *args, **kwargs):

        self.wsgiapp = pylons.test.pylonsapp

        init_stack(self.wsgiapp.config)

        unittest.TestCase.__init__(self, *args, **kwargs)

class BaseTestController(object):

    """Base test controller used by pytest and unittest tests controllers."""

    # Note: pytest base classes cannot have an __init__ method

    def init(self):

        self.app = TestApp(self.wsgiapp)

        self.maxDiff = None

        self.index_location = config['app_conf']['index_dir']

    def log_user(self, username=TEST_USER_ADMIN_LOGIN,

                 password=TEST_USER_ADMIN_PASS):

        self._logged_username = username

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': username,

                                  'password': password})

        if 'Invalid username or password' in response.body:

        self.assertEqual(response.status, '302 Found')

        self.assert_authenticated_user(response, username)

        response = response.follow()

        return response.session['authuser']

    def _get_logged_user(self):

        return User.get_by_username(self._logged_username)

    def assert_authenticated_user(self, response, expected_username):

        cookie = response.session.get('authuser')

        user = cookie and cookie.get('user_id')

        user = user and User.get(user)

        user = user and user.username

        self.assertEqual(user, expected_username)

    def authentication_token(self):

        return self.app.get(url('authentication_token')).body

    def checkSessionFlash(self, response, msg=None, skip=0, _matcher=lambda msg, m: msg in m):

        if 'flash' not in response.session:

        try:

            level, m = response.session['flash'][-1 - skip]

            if _matcher(msg, m):

                return

        except IndexError:

            pass

                           (msg, skip,

                            ', '.join('`%s`' % m for level, m in response.session['flash']))))

    def checkSessionFlashRegex(self, response, regex, skip=0):

        self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)

class TestController(BaseTestCase, BaseTestController):

    """Deprecated unittest-style test controller"""

    def __init__(self, *args, **kwargs):

        super(TestController, self).__init__(*args, **kwargs)

        self.init()

class TestControllerPytest(BaseTestController):

    """Pytest-style test controller"""

    # Note: pytest base classes cannot have an __init__ method

    @pytest.fixture(autouse=True)

    def app_fixture(self):

        self.wsgiapp = pylons.test.pylonsapp

        init_stack(self.wsgiapp.config)

        self.init()

        return self.app

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

        response = self.app.get(url('repo_check_home', repo_name=repo_name))

        self.assertEqual(response.json, {u'result': True})

        self.checkSessionFlash(response,

                               'Created repository <a href="/%s">%s</a>'

                               % (repo_name, repo_name))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name).one()

        self.assertEqual(new_repo.repo_name, repo_name)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name))

        response.mustcontain(repo_name)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(safe_str(os.path.join(TESTS_TMP_PATH, repo_name)))

        except vcs.exceptions.VCSError:

        RepoModel().delete(repo_name)

        Session().commit()

    def test_create_in_group(self):

        self.log_user()

        ## create GROUP

        group_name = u'sometest_%s' % self.REPO_TYPE

        gr = RepoGroupModel().create(group_name=group_name,

                                     group_description=u'test',

                                     owner=TEST_USER_ADMIN_LOGIN)

        Session().commit()

        repo_name = u'ingroup'

        repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])

        description = u'description for newly created repo'

        response = self.app.post(url('repos'),

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                repo_group=gr.group_id,

                                                _authentication_token=self.authentication_token()))

                               % (repo_name_full, repo_name_full))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_full).one()

        new_repo_id = new_repo.repo_id

        self.assertEqual(new_repo.repo_name, repo_name_full)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name_full))

        response.mustcontain(repo_name_full)

        response.mustcontain(self.REPO_TYPE)

        inherited_perms = UserRepoToPerm.query() \

            .filter(UserRepoToPerm.repository_id == new_repo_id).all()

        self.assertEqual(len(inherited_perms), 1)

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(safe_str(os.path.join(TESTS_TMP_PATH, repo_name_full)))

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

        RepoModel().delete(repo_name_full)

        RepoGroupModel().delete(group_name)

        Session().commit()

    def test_create_in_group_without_needed_permissions(self):

        usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)

        # avoid spurious RepoGroup DetachedInstanceError ...

        authentication_token = self.authentication_token()

        # revoke

        user_model = UserModel()

        # disable fork and create on default user

        user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')

        user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')

        user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')

        user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')

        # disable on regular user

        user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')

        user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')

        user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')

        user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')

        Session().commit()

                               % (repo_name_full, repo_name_full))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_full).one()

        new_repo_id = new_repo.repo_id

        self.assertEqual(new_repo.repo_name, repo_name_full)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name_full))

        response.mustcontain(repo_name_full)

        response.mustcontain(self.REPO_TYPE)

        inherited_perms = UserRepoToPerm.query() \

            .filter(UserRepoToPerm.repository_id == new_repo_id).all()

        self.assertEqual(len(inherited_perms), 1)

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(safe_str(os.path.join(TESTS_TMP_PATH, repo_name_full)))

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

        RepoModel().delete(repo_name_full)

        RepoGroupModel().delete(group_name)

        RepoGroupModel().delete(group_name_allowed)

        Session().commit()

    def test_create_in_group_inherit_permissions(self):

        self.log_user()

        ## create GROUP

        group_name = u'sometest_%s' % self.REPO_TYPE

        gr = RepoGroupModel().create(group_name=group_name,

                                     group_description=u'test',

                                     owner=TEST_USER_ADMIN_LOGIN)

        perm = Permission.get_by_key('repository.write')

        RepoGroupModel().grant_user_permission(gr, TEST_USER_REGULAR_LOGIN, perm)

        ## add repo permissions

        Session().commit()

        repo_name = u'ingroup_inherited_%s' % self.REPO_TYPE

        repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])

        description = u'description for newly created repo'

        response = self.app.post(url('repos'),

        ## run the check page that triggers the flash message

        response = self.app.get(url('repo_check_home', repo_name=repo_name_full))

        self.checkSessionFlash(response,

                               'Created repository <a href="/%s">%s</a>'

                               % (repo_name_full, repo_name_full))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_full).one()

        new_repo_id = new_repo.repo_id

        self.assertEqual(new_repo.repo_name, repo_name_full)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name_full))

        response.mustcontain(repo_name_full)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(safe_str(os.path.join(TESTS_TMP_PATH, repo_name_full)))

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

        #check if inherited permissiona are applied

        inherited_perms = UserRepoToPerm.query() \

            .filter(UserRepoToPerm.repository_id == new_repo_id).all()

        self.assertEqual(len(inherited_perms), 2)

        self.assertTrue(TEST_USER_REGULAR_LOGIN in [x.user.username

                                                    for x in inherited_perms])

        self.assertTrue('repository.write' in [x.permission.permission_name

                                               for x in inherited_perms])

        RepoModel().delete(repo_name_full)

        RepoGroupModel().delete(group_name)

        Session().commit()

    def test_create_remote_repo_wrong_clone_uri(self):

        self.log_user()

        repo_name = self.NEW_REPO

        description = u'description for newly created repo'

        response = self.app.post(url('repos'),

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                repo_name=repo_name,

                                                repo_description=description,

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

        response = self.app.get(url('repo_check_home', repo_name=repo_name))

        self.checkSessionFlash(response,

                               'Created repository <a href="/%s">%s</a>'

                               % (repo_name, repo_name))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name).one()

        self.assertEqual(new_repo.repo_name, repo_name)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name))

        response.mustcontain(repo_name)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(safe_str(os.path.join(TESTS_TMP_PATH, repo_name)))

        except vcs.exceptions.VCSError:

        response = self.app.post(url('delete_repo', repo_name=repo_name),

            params={'_method': 'delete', '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))

        response.follow()

        #check if repo was deleted from db

        deleted_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name).scalar()

        self.assertEqual(deleted_repo, None)

        self.assertEqual(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)),

                                  False)

    def test_delete_non_ascii(self):

        self.log_user()

        non_ascii = "ąęł"

        repo_name = "%s%s" % (safe_str(self.NEW_REPO), non_ascii)

        repo_name_unicode = safe_unicode(repo_name)

        description = 'description for newly created repo' + non_ascii

        description_unicode = safe_unicode(description)

                                                repo_description=description,

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

        response = self.app.get(url('repo_check_home', repo_name=repo_name))

        self.assertEqual(response.json, {u'result': True})

        self.checkSessionFlash(response,

                               u'Created repository <a href="/%s">%s</a>'

                               % (urllib.quote(repo_name), repo_name_unicode))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_unicode).one()

        self.assertEqual(new_repo.repo_name, repo_name_unicode)

        self.assertEqual(new_repo.description, description_unicode)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name))

        response.mustcontain(repo_name)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(safe_str(os.path.join(TESTS_TMP_PATH, repo_name)))

        except vcs.exceptions.VCSError:

        response = self.app.post(url('delete_repo', repo_name=repo_name),

            params={'_method': 'delete', '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))

        response.follow()

        #check if repo was deleted from db

        deleted_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_unicode).scalar()

        self.assertEqual(deleted_repo, None)

        self.assertEqual(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)),

                                  False)

    def test_delete_repo_with_group(self):

        #TODO:

        pass

    def test_delete_browser_fakeout(self):

        response = self.app.post(url('delete_repo', repo_name=self.REPO),

                                 params=dict(_method='delete', _authentication_token=self.authentication_token()))

    def test_show(self):

import os

import sys

import mock

import datetime

import urllib2

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

from kallithea.tests.vcs.conf import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, get_new_dir

class GitRepositoryTest(unittest.TestCase):

    def __check_for_existing_repo(self):

        if os.path.exists(TEST_GIT_REPO_CLONE):

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

        wrong_repo_path = '/tmp/errorrepo'

        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

            urlerror_fail_repo = 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

        chset10 = self.repo.get_changeset(rev10)

        self.assertEqual(chset10.branch, 'master')

        self.assertEqual(chset10.tags, [])

        rev44 = self.repo.revisions[44]

        chset44 = self.repo.get_changeset(rev44)

        self.assertEqual(chset44.branch, 'web-branch')

        tip = self.repo.get_changeset('tip')

        self.assertTrue('tip' in tip.tags)

        """

        # Those tests would fail - branches are now going

        # to be changed at main API in order to support git backend

        pass

    def _test_slices(self, limit, offset):

        count = self.repo.count()

        changesets = self.repo.get_changesets(limit=limit, offset=offset)

        idx = 0

        for changeset in changesets:

            rev = offset + idx

            idx += 1

            rev_id = self.repo.revisions[rev]

            if idx > limit:

                    "there are %s total revisions, offset=%s, limit=%s)"

                    % (rev_id, count, offset, limit))

            self.assertEqual(changeset, self.repo.get_changeset(rev_id))

        result = list(self.repo.get_changesets(limit=limit, offset=offset))

        start = offset

        end = limit and offset + limit or None

        sliced = list(self.repo[start:end])

            msg="Comparison failed for limit=%s, offset=%s"

            "(get_changeset returned: %s and sliced: %s"

            % (limit, offset, result, sliced))

    def _test_file_size(self, revision, path, size):

        node = self.repo.get_changeset(revision).get_node(path)

        self.assertTrue(node.is_file())

        self.assertEqual(node.size, size)

    def test_file_size(self):

        to_check = (

            ('c1214f7e79e02fc37156ff215cd71275450cffc3',

                'vcs/backends/BaseRepository.py', 502),

            ('d7e0d30fbcae12c90680eb095a4f5f02505ce501',

                'vcs/backends/hg.py', 854),

            ('6e125e7c890379446e98980d8ed60fba87d0f6d1',

                'setup.py', 1068),

            ('d955cd312c17b02143c04fa1099a352b04368118',

                'vcs/backends/base.py', 2921),

            ('ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',

                'vcs/backends/base.py', 3936),

            ('f50f42baeed5af6518ef4b0cb2f1423f3851a941',

                'vcs/backends/base.py', 6189),

import os

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

    TEST_HG_REPO_PULL

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

# Use only clean mercurial's ui

from kallithea.lib.vcs.utils.hgcompat import mercurial

mercurial.scmutil.rcpath()

if mercurial.scmutil._rcpath:

    mercurial.scmutil._rcpath = mercurial.scmutil._rcpath[:1]

class MercurialRepositoryTest(unittest.TestCase):

    def __check_for_existing_repo(self):

        if os.path.exists(TEST_HG_REPO_CLONE):

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

        wrong_repo_path = '/tmp/errorrepo'

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

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

        """

        Tests if node's parent path are properly computed.

        """

        node = Node(node_path, NodeKind.DIR)

        parent_path = node.get_parent_path()

        self.assertTrue(parent_path.endswith('/') or \

            node.is_root() and parent_path == '')

        self.assertEqual(parent_path, expected_parent_path,

            "Node's path is %r and parent path is %r but should be %r"

            % (node.path, parent_path, expected_parent_path))

    def test_parent_path(self):

        test_paths = (

            # (node_path, expected_parent_path)

            ('', ''),

            ('some/path/', 'some/'),

            ('some/longer/path/', 'some/longer/'),

        )

        for node_path, expected_parent_path in test_paths:

            self._test_parent_path(node_path, expected_parent_path)

    '''

    def _test_trailing_slash(self, path):

        if not path.endswith('/'):

        for kind in NodeKind.FILE, NodeKind.DIR:

            self.assertRaises(NodeError, Node, path=path, kind=kind)

    def test_trailing_slash(self):

        for path in ('/', 'foo/', 'foo/bar/', 'foo/bar/biz/'):

            self._test_trailing_slash(path)

    '''

    def test_is_file(self):

        node = Node('any', NodeKind.FILE)

        self.assertTrue(node.is_file())

        node = FileNode('any')

        self.assertTrue(node.is_file())

        self.assertRaises(AttributeError, getattr, node, 'nodes')

    def test_is_dir(self):

        node = Node('any_dir', NodeKind.DIR)

        self.assertTrue(node.is_dir())

        node = DirNode('any_dir')

        self.assertTrue(node.is_dir())

        self.assertRaises(NodeError, getattr, node, 'content')