# Invoke websetup with the current config file

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

#SOME GLOBALS FOR TESTS

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'

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

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

TEST_HG_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)

TEST_HG_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)

# cached repos if any !

# comment out to get some other repos from bb or github

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

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

#skip ldap tests if LDAP lib is not installed

# -*- coding: utf-8 -*-

from kallithea.model.db import Setting, Ui

from kallithea.tests.base import *

from kallithea.tests.fixture import Fixture

fixture = Fixture()

        response = self.app.get(url('admin_settings_hooks'))

    def test_create_custom_hook(self):

        self.log_user()

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

                                params=dict(new_hook_ui_key='test_hooks_1',

                                            _authentication_token=self.authentication_token()))

        response = response.follow()

        response.mustcontain('test_hooks_1')

    def test_create_custom_hook_delete(self):

        self.log_user()

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

                                params=dict(new_hook_ui_key='test_hooks_2',

                                            _authentication_token=self.authentication_token()))

        response = response.follow()

        response.mustcontain('test_hooks_2')

        hook_id = Ui.get_by_key('hooks', 'test_hooks_2').ui_id

        ## delete

        self.app.post(url('admin_settings_hooks'),

                        params=dict(hook_id=hook_id, _authentication_token=self.authentication_token()))

        response = self.app.get(url('admin_settings_hooks'))

        response.mustcontain(no=['test_hooks_2'])

    def test_index_search(self):

        self.log_user()

        response = self.app.get(url('admin_settings_search'))

    def test_index_system(self):

# -*- coding: utf-8 -*-

import formencode

import pytest

from kallithea.tests.base import *

from kallithea.model import validators as v

from kallithea.model.user_group import UserGroupModel

        assert {'user2': 'test', 'pass': 'pass'} == validator.to_python(value={'user2': 'test',

                                                    'pass': 'pass'})

    def test_ValidPath(self):

            validator = v.ValidPath()

            with pytest.raises(formencode.Invalid):

                validator.to_python('/no_such_dir')

    def test_UniqSystemEmail(self):

        validator = v.UniqSystemEmail(old_data={})

                print 'stderr:', repr(stderr)

        if not ignoreReturnCode:

            assert p.returncode == 0

        return stdout, stderr

def _add_files_and_push(webserver, vcs, DEST, ignoreReturnCode=False, files_no=3,

                        clone_url=None, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS):

    """

    Generate some files, add it to DEST repo and push back

        Repository.unlock(r)

        r.enable_locking = False

        Session().commit()

    def test_clone_hg_repo_by_admin(self, webserver):

        clone_url = webserver.repo_url(HG_REPO)

        assert 'requesting all changes' in stdout

        assert 'adding changesets' in stdout

        assert 'adding manifests' in stdout

        assert 'adding file changes' in stdout

        assert stderr == ''

    def test_clone_git_repo_by_admin(self, webserver):

        clone_url = webserver.repo_url(GIT_REPO)

        assert 'Cloning into' in stdout + stderr

        assert stderr == '' or stdout == ''

    def test_clone_wrong_credentials_hg(self, webserver):

        clone_url = webserver.repo_url(HG_REPO, password='bad!')

        assert 'abort: authorization failed' in stderr

    def test_clone_wrong_credentials_git(self, webserver):

        clone_url = webserver.repo_url(GIT_REPO, password='bad!')

        assert 'fatal: Authentication failed' in stderr

    def test_clone_git_dir_as_hg(self, webserver):

        clone_url = webserver.repo_url(GIT_REPO)

        assert 'HTTP Error 404: Not Found' in stderr

    def test_clone_hg_repo_as_git(self, webserver):

        clone_url = webserver.repo_url(HG_REPO)

        assert 'not found' in stderr

    def test_clone_non_existing_path_hg(self, webserver):

        clone_url = webserver.repo_url('trololo')

        assert 'HTTP Error 404: Not Found' in stderr

    def test_clone_non_existing_path_git(self, webserver):

        clone_url = webserver.repo_url('trololo')

        assert 'not found' in stderr

    def test_push_new_file_hg(self, webserver):

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(HG_REPO)

        fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())

        fixture.create_fork(HG_REPO, fork_name)

        clone_url = webserver.repo_url(fork_name)

        stdout, stderr = _add_files_and_push(webserver, 'hg', DEST, clone_url=clone_url)

        assert 'Repository size' in stdout

        assert 'Last revision is now' in stdout

    def test_push_new_file_git(self, webserver):

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO)

        # commit some stuff into this repo

        fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())

        fixture.create_fork(GIT_REPO, fork_name)

        clone_url = webserver.repo_url(fork_name)

        stdout, stderr = _add_files_and_push(webserver, 'git', DEST, clone_url=clone_url)

        key.cache_active = True

        Session().commit()

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(HG_REPO)

        fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())

        fixture.create_fork(HG_REPO, fork_name)

        clone_url = webserver.repo_url(fork_name)

        stdout, stderr = _add_files_and_push(webserver, 'hg', DEST, files_no=1, clone_url=clone_url)

        key.cache_active = True

        Session().commit()

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO)

        # commit some stuff into this repo

        fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())

        fixture.create_fork(GIT_REPO, fork_name)

        clone_url = webserver.repo_url(fork_name)

        stdout, stderr = _add_files_and_push(webserver, 'git', DEST, files_no=1, clone_url=clone_url)

                                               ==fork_name).all()

        assert key == []

    def test_push_wrong_credentials_hg(self, webserver):

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(HG_REPO)

        stdout, stderr = _add_files_and_push(webserver, 'hg', DEST, username='bad',

                                             password='name', ignoreReturnCode=True)

        assert 'abort: authorization failed' in stderr

    def test_push_wrong_credentials_git(self, webserver):

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO)

        stdout, stderr = _add_files_and_push(webserver, 'git', DEST, username='bad',

                                             password='name', ignoreReturnCode=True)

        assert 'fatal: Authentication failed' in stderr

    def test_push_back_to_wrong_url_hg(self, webserver):

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(HG_REPO)

        stdout, stderr = _add_files_and_push(webserver, 'hg', DEST,

                                    clone_url='http://%s:%s/tmp' % (webserver.server_address[0], webserver.server_address[1]),

                                    ignoreReturnCode = True)

        assert 'HTTP Error 404: Not Found' in stderr

    def test_push_back_to_wrong_url_git(self, webserver):

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO)

        stdout, stderr = _add_files_and_push(webserver, 'git', DEST,

                                    clone_url='http://%s:%s/tmp' % (webserver.server_address[0], webserver.server_address[1]),

                                    ignoreReturnCode = True)

        assert 'not found' in stderr

        # enable locking

        r = Repository.get_by_repo_name(HG_REPO)

        r.enable_locking = True

        Session().commit()

        # clone

        clone_url = webserver.repo_url(HG_REPO)

        #check if lock was made

        r = Repository.get_by_repo_name(HG_REPO)

        assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

    def test_clone_and_create_lock_git(self, webserver):

        # enable locking

        r = Repository.get_by_repo_name(GIT_REPO)

        r.enable_locking = True

        Session().commit()

        # clone

        clone_url = webserver.repo_url(GIT_REPO)

        #check if lock was made

        r = Repository.get_by_repo_name(GIT_REPO)

        assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

    def test_clone_after_repo_was_locked_hg(self, webserver):

        #lock repo

        r = Repository.get_by_repo_name(HG_REPO)

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #pull fails since repo is locked

        clone_url = webserver.repo_url(HG_REPO)

        msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""

                % (HG_REPO, TEST_USER_ADMIN_LOGIN))

        assert msg in stderr

    def test_clone_after_repo_was_locked_git(self, webserver):

        #lock repo

        r = Repository.get_by_repo_name(GIT_REPO)

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #pull fails since repo is locked

        clone_url = webserver.repo_url(GIT_REPO)

        msg = ("""The requested URL returned error: 423""")

        assert msg in stderr

    def test_push_on_locked_repo_by_other_user_hg(self, webserver):

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(HG_REPO)

        #lock repo

        r = Repository.get_by_repo_name(HG_REPO)

        # let this user actually push !

        RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,

                                          perm='repository.write')

        # Note: Git hooks must be executable on unix. This test will thus fail

        # for example on Linux if /tmp is mounted noexec.

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO)

        #lock repo

        r = Repository.get_by_repo_name(GIT_REPO)

        # let this user actually push !

        RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,

                                          perm='repository.write')

        r = Repository.get_by_repo_name(fork_name)

        r.enable_locking = True

        Session().commit()

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(fork_name)

        #check for lock repo after clone

        r = Repository.get_by_repo_name(fork_name)

        uid = User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

        assert r.locked[0] == uid

        r = Repository.get_by_repo_name(fork_name)

        r.enable_locking = True

        Session().commit()

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = webserver.repo_url(fork_name)

        #check for lock repo after clone

        r = Repository.get_by_repo_name(fork_name)

        assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

        #push is ok and repo is now unlocked

    def test_ip_restriction_hg(self, webserver):

        user_model = UserModel()

        try:

            user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

            Session().commit()

            clone_url = webserver.repo_url(HG_REPO)

            assert 'abort: HTTP Error 403: Forbidden' in stderr

        finally:

            #release IP restrictions

            for ip in UserIpMap.query():

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        # IP permissions are cached, need to wait for the cache in the server process to expire

        time.sleep(1.5)

        clone_url = webserver.repo_url(HG_REPO)

        assert 'requesting all changes' in stdout

        assert 'adding changesets' in stdout

        assert 'adding manifests' in stdout

        assert 'adding file changes' in stdout

    def test_ip_restriction_git(self, webserver):

        user_model = UserModel()

        try:

            user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

            Session().commit()

            clone_url = webserver.repo_url(GIT_REPO)

            # The message apparently changed in Git 1.8.3, so match it loosely.

            assert re.search(r'\b403\b', stderr)

        finally:

            #release IP restrictions

            for ip in UserIpMap.query():

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        # IP permissions are cached, need to wait for the cache in the server process to expire

        time.sleep(1.5)

        clone_url = webserver.repo_url(GIT_REPO)

        assert 'Cloning into' in stdout + stderr

        assert stderr == '' or stdout == ''

#==============================================================================

# TESTS

#==============================================================================

def test_clone_with_credentials(no_errors=False, repo=HG_REPO, method=METHOD,

    cwd = path = os.path.join(Ui.get_by_key('paths', '/').ui_value, repo)

    try:

        shutil.rmtree(path, ignore_errors=True)

        os.makedirs(path)

        #print 'made dirs %s' % os.path.join(path)

    except OSError:

        raise

    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \

                  {'user': USER,

                   'pass': PASS,

                   'host': HOST,

                   'cloned_repo': repo, }

    if method == 'pull':

        stdout, stderr = Command(cwd).execute(backend, method, '--cwd', dest, clone_url)

    else:

        stdout, stderr = Command(cwd).execute(backend, method, clone_url, dest)

        if not no_errors:

            if backend == 'hg':

                assert """adding file changes""" in stdout, 'no messages about cloning'

                assert """abort""" not in stderr , 'got error from clone'

            elif backend == 'git':

                assert """Cloning into""" in stdout, 'no messages about cloning'

if __name__ == '__main__':

    try:

        create_test_user(force=False)

        import time

        try:

            METHOD = sys.argv[3]

        except IndexError:

            pass

        except IndexError:

            backend = 'hg'

        if METHOD == 'pull':

            seq = _RandomNameSequence().next()

            test_clone_with_credentials(repo=sys.argv[1], method='clone',

        s = time.time()

        for i in range(1, int(sys.argv[2]) + 1):

            print 'take', i

            test_clone_with_credentials(repo=sys.argv[1], method=METHOD,

        print 'time taken %.3f' % (time.time() - s)

    except Exception as e:

        sys.exit('stop on %s' % e)

__all__ = (

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

    'SCM_TESTS',

)

SCM_TESTS = ['hg', 'git']

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

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

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

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

                               os.path.join(TEST_TMP_PATH, 'vcs-git'))

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

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

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

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

                              os.path.join(TEST_TMP_PATH, 'vcs-hg'))

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

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

TEST_REPO_PREFIX = 'vcs-test'

def get_new_dir(title=None):

    """

    Resulting directory name will have format:

    prefix-[title-]hexuuid

    Prefix is equal to value of variable TEST_REPO_PREFIX. The "hexuuid" is a

    if title:

        name = "%s-%s" % (TEST_REPO_PREFIX, title)

    else:

        name = TEST_REPO_PREFIX

    # Generate new hexes until we get a unique name (just in case).

    hex_uuid = uuid.uuid4().hex

    while os.path.exists("%s-%s" % (path, hex_uuid)):

        hex_uuid = uuid.uuid4().hex

import os

import tarfile

import zipfile

import datetime

import tempfile

import StringIO

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.lib.vcs.exceptions import VCSError

from kallithea.lib.vcs.nodes import FileNode

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

class ArchivesTestCaseMixin(_BackendTestMixin):

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

                        content='Foobar %d' % x),

                ],

            }

    def test_archive_zip(self):

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

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

        out = zipfile.ZipFile(path)

        for x in xrange(5):

            node_path = '%d/file_%d.txt' % (x, x)

            decompressed.write(out.read('repo/' + node_path))

            self.assertEqual(

                decompressed.getvalue(),

                self.tip.get_node(node_path).content)

    def test_archive_tgz(self):

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

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

        outfile = tarfile.open(path, 'r|gz')

        outfile.extractall(outdir)

        for x in xrange(5):

            node_path = '%d/file_%d.txt' % (x, x)

            self.assertEqual(

                open(os.path.join(outdir, 'repo/' + node_path)).read(),

                self.tip.get_node(node_path).content)

    def test_archive_tbz2(self):

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

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

        outfile = tarfile.open(path, 'r|bz2')

        outfile.extractall(outdir)

        for x in xrange(5):

            node_path = '%d/file_%d.txt' % (x, x)

            self.assertEqual(

                open(os.path.join(outdir, 'repo/' + node_path)).read(),

                self.tip.get_node(node_path).content)

    def test_archive_default_stream(self):

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

            self.tip.fill_archive(stream=stream)

        mystream = StringIO.StringIO()

        self.tip.fill_archive(stream=mystream)

        mystream.seek(0)

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

import os

import sys

import mock

import datetime

import urllib2

import pytest

from kallithea.lib.vcs.backends.git import GitRepository, GitChangeset

from kallithea.lib.vcs.exceptions import RepositoryError, VCSError, NodeDoesNotExistError

from kallithea.lib.vcs.nodes import NodeKind, FileNode, DirNode, NodeState

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

from kallithea.model.scm import ScmModel

from kallithea.tests.vcs.base import _BackendTestMixin

class GitRepositoryTest(unittest.TestCase):

    def __check_for_existing_repo(self):

        if os.path.exists(TEST_GIT_REPO_CLONE):

                      % TEST_GIT_REPO_CLONE)

    def setUp(self):

        self.repo = GitRepository(TEST_GIT_REPO)

    def test_wrong_repo_path(self):

        self.assertRaises(RepositoryError, GitRepository, wrong_repo_path)

    def test_git_cmd_injection(self):

        repo_inject_path = TEST_GIT_REPO + '; echo "Cake";'

        with self.assertRaises(urllib2.URLError):

            # Should fail because URL will contain the parts after ; too

import os

import pytest

from kallithea.lib.utils2 import safe_str

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

from kallithea.lib.vcs.exceptions import RepositoryError, VCSError, NodeDoesNotExistError

from kallithea.lib.vcs.nodes import NodeKind, NodeState

from kallithea.tests.vcs.conf import TEST_HG_REPO, TEST_HG_REPO_CLONE, \

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

# Use only clean mercurial's ui

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

mercurial.scmutil.rcpath()

                      % TEST_HG_REPO_CLONE)

    def setUp(self):

        self.repo = MercurialRepository(safe_str(TEST_HG_REPO))

    def test_wrong_repo_path(self):

        self.assertRaises(RepositoryError, MercurialRepository, wrong_repo_path)

    def test_unicode_path_repo(self):

        self.assertRaises(VCSError,lambda:MercurialRepository(u'iShouldFail'))

    def test_repo_clone(self):

# -*- coding: utf-8 -*-

import os

import mock

import time

import shutil

import datetime

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

from kallithea.lib.vcs.utils.paths import get_dirs_for_path

from kallithea.lib.vcs.utils.helpers import get_dict_for_attrs

from kallithea.lib.vcs.utils.helpers import get_scm

from kallithea.lib.vcs.utils.helpers import get_scms_for_path

from kallithea.lib.vcs.utils.helpers import get_total_seconds

        self.assertRaises(VCSError, get_scm, multialias_repo_path)

    def test_get_scm_error_path(self):

        self.assertRaises(VCSError, get_scm, 'err')

    def test_get_scms_for_path(self):

        os.mkdir(new)

        self.assertEqual(get_scms_for_path(new), [])

        os.mkdir(os.path.join(new, '.tux'))

        self.assertEqual(get_scms_for_path(new), [])