# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.bin.kallithea_backup

~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Repositories backup manager, it allows to backups all

repositories and send it to backup server using RSA key via ssh.

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Feb 28, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import os

import sys

import logging

import tarfile

import datetime

import subprocess

logging.basicConfig(level=logging.DEBUG,

                    format="%(asctime)s %(levelname)-5.5s %(message)s")

class BackupManager(object):

    def __init__(self, repos_location, rsa_key, backup_server):

        today = datetime.datetime.now().weekday() + 1

        self.backup_file_name = "repos.%s.tar.gz" % today

        self.id_rsa_path = self.get_id_rsa(rsa_key)

        self.repos_path = self.get_repos_path(repos_location)

        self.backup_server = backup_server

        logging.info('starting backup for %s', self.repos_path)

        logging.info('backup target %s', self.backup_file_path)

    def get_id_rsa(self, rsa_key):

        if not os.path.isfile(rsa_key):

            logging.error('Could not load id_rsa key file in %s', rsa_key)

            sys.exit()

        return rsa_key

    def get_repos_path(self, path):

        if not os.path.isdir(path):

            logging.error('Wrong location for repositories in %s', path)

            sys.exit()

        return path

    def backup_repos(self):

        bckp_file = os.path.join(self.backup_file_path, self.backup_file_name)

        tar = tarfile.open(bckp_file, "w:gz")

        for dir_name in os.listdir(self.repos_path):

            logging.info('backing up %s', dir_name)

            tar.add(os.path.join(self.repos_path, dir_name), dir_name)

        tar.close()

skipif = pytest.mark.skipif

parametrize = pytest.mark.parametrize

__all__ = [

    'skipif', 'parametrize', 'parameterized', 'environ', 'url', 'TestController', 'TestControllerPytest',

    'ldap_lib_installed', 'pam_lib_installed', 'BaseTestCase', 'init_stack',

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

    'GIT_REMOTE_REPO', 'SCM_TESTS', 'remove_all_notifications',

]

# Invoke websetup with the current config file

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

environ = {}

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

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'

## VCS

SCM_TESTS = ['hg', 'git']

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

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

from kallithea.model.db import Setting, Ui

from kallithea.tests import *

from kallithea.tests.fixture import Fixture

fixture = Fixture()

class TestAdminSettingsController(TestControllerPytest):

    def test_index_main(self):

        self.log_user()

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

    def test_index_mapping(self):

        self.log_user()

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

    def test_index_global(self):

        self.log_user()

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

    def test_index_visual(self):

        self.log_user()

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

    def test_index_email(self):

        self.log_user()

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

    def test_index_hooks(self):

        self.log_user()

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

        self.log_user()

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

    def test_ga_code_active(self):

        self.log_user()

        old_title = 'Kallithea'

        old_realm = 'Kallithea authentication'

        new_ga_code = 'ga-test-123456789'

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

                        params=dict(title=old_title,

                                 realm=old_realm,

                                 ga_code=new_ga_code,

                                 captcha_private_key='',

                                 captcha_public_key='',

                                 _authentication_token=self.authentication_token(),

                                 ))

        self.checkSessionFlash(response, 'Updated application settings')

class TestVCSOperations(BaseTestCase):

    @classmethod

    def setup_class(cls):

        #DISABLE ANONYMOUS ACCESS

        set_anonymous_access(False)

    def setUp(self):

        r = Repository.get_by_repo_name(GIT_REPO)

        Repository.unlock(r)

        r.enable_locking = False

        Session().add(r)

        Session().commit()

        r = Repository.get_by_repo_name(HG_REPO)

        Repository.unlock(r)

        r.enable_locking = False

        Session().add(r)

        Session().commit()

    def test_clone_hg_repo_by_admin(self):

        clone_url = _construct_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):

        clone_url = _construct_url(GIT_REPO)

        assert 'Cloning into' in stdout + stderr

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

    def test_clone_wrong_credentials_hg(self):

        clone_url = _construct_url(HG_REPO, passwd='bad!')

        assert 'abort: authorization failed' in stderr

    def test_clone_wrong_credentials_git(self):

        clone_url = _construct_url(GIT_REPO, passwd='bad!')

        assert 'fatal: Authentication failed' in stderr

    def test_clone_git_dir_as_hg(self):

        clone_url = _construct_url(GIT_REPO)

        assert 'HTTP Error 404: Not Found' in stderr

    def test_clone_hg_repo_as_git(self):

        clone_url = _construct_url(HG_REPO)

        assert 'not found' in stderr

    def test_clone_non_existing_path_hg(self):

        clone_url = _construct_url('trololo')

        assert 'HTTP Error 404: Not Found' in stderr

    def test_clone_non_existing_path_git(self):

        clone_url = _construct_url('trololo')

        assert 'not found' in stderr

    def test_push_new_file_hg(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

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

        assert 'pushing to' in stdout

        assert 'Repository size' in stdout

        assert 'Last revision is now' in stdout

    def test_push_new_file_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        # commit some stuff into this repo

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

        print [(x.repo_full_path,x.repo_path) for x in Repository.get_all()]

        _check_proper_git_push(stdout, stderr)

    def test_push_invalidates_cache_hg(self):

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               ==HG_REPO).scalar()

        if not key:

            key = CacheInvalidation(HG_REPO, HG_REPO)

        key.cache_active = True

        Session().add(key)

        Session().commit()

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

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

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               ==HG_REPO).all()

        self.assertEqual(key, [])

    def test_push_invalidates_cache_git(self):

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               ==GIT_REPO).scalar()

        if not key:

            key = CacheInvalidation(GIT_REPO, GIT_REPO)

        key.cache_active = True

        Session().add(key)

        Session().commit()

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        # commit some stuff into this repo

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

        _check_proper_git_push(stdout, stderr)

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               ==GIT_REPO).all()

        self.assertEqual(key, [])

    def test_push_wrong_credentials_hg(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

        stdout, stderr = _add_files_and_push('hg', DEST, user='bad',

                                             passwd='name')

        assert 'abort: authorization failed' in stderr

    def test_push_wrong_credentials_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        stdout, stderr = _add_files_and_push('git', DEST, user='bad',

                                             passwd='name')

        assert 'fatal: Authentication failed' in stderr

    def test_push_back_to_wrong_url_hg(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

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

                                    clone_url='http://%s/tmp' % HOST)

        assert 'HTTP Error 404: Not Found' in stderr

    def test_push_back_to_wrong_url_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

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

                                    clone_url='http://%s/tmp' % HOST)

        assert 'not found' in stderr

    def test_clone_and_create_lock_hg(self):

        # enable locking

        r = Repository.get_by_repo_name(HG_REPO)

        r.enable_locking = True

        Session().add(r)

        Session().commit()

        # clone

        clone_url = _construct_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):

        # enable locking

        r = Repository.get_by_repo_name(GIT_REPO)

        r.enable_locking = True

        Session().add(r)

        Session().commit()

        # clone

        clone_url = _construct_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):

        #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 = _construct_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):

        #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 = _construct_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):

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

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

        Session().commit()

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

        #push fails repo is locked by other user !

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

                                             user=TEST_USER_REGULAR_LOGIN,

                                             passwd=TEST_USER_REGULAR_PASS)

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

                % (HG_REPO, TEST_USER_ADMIN_LOGIN))

        assert msg in stderr

    def test_push_on_locked_repo_by_other_user_git(self):

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

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

        Session().commit()

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

        #push fails repo is locked by other user !

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

                                             user=TEST_USER_REGULAR_LOGIN,

                                             passwd=TEST_USER_REGULAR_PASS)

        err = 'Repository `%s` locked by user `%s`' % (GIT_REPO, TEST_USER_ADMIN_LOGIN)

        assert err in stderr

        #TODO: fix this somehow later on Git, Git is stupid and even if we throw

        #back 423 to it, it makes ANOTHER request and we fail there with 405 :/

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

                % (GIT_REPO, TEST_USER_ADMIN_LOGIN))

        #msg = "405 Method Not Allowed"

        #assert msg in stderr

    def test_push_unlocks_repository_hg(self):

        # enable locking

        r = Repository.get_by_repo_name(HG_REPO)

        r.enable_locking = True

        Session().add(r)

        Session().commit()

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

        #check for lock repo after clone

        r = Repository.get_by_repo_name(HG_REPO)

        uid = User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

        assert r.locked[0] == uid

        #push is ok and repo is now unlocked

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

        assert ('remote: Released lock on repo `%s`' % HG_REPO) in stdout

        #we need to cleanup the Session Here !

        Session.remove()

        r = Repository.get_by_repo_name(HG_REPO)

        assert r.locked == [None, None]

    #TODO: fix me ! somehow during tests hooks don't get called on Git

    def test_push_unlocks_repository_git(self):

        # enable locking

        r = Repository.get_by_repo_name(GIT_REPO)

        r.enable_locking = True

        Session().add(r)

        Session().commit()

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        #check for lock repo after clone

        r = Repository.get_by_repo_name(GIT_REPO)

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

        #push is ok and repo is now unlocked

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

        _check_proper_git_push(stdout, stderr)

        #assert ('remote: Released lock on repo `%s`' % GIT_REPO) in stdout

        #we need to cleanup the Session Here !

        Session.remove()

        r = Repository.get_by_repo_name(GIT_REPO)

        assert r.locked == [None, None]

    def test_ip_restriction_hg(self):

        user_model = UserModel()

        try:

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

            Session().commit()

            clone_url = _construct_url(HG_REPO)

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

        finally:

            #release IP restrictions

            for ip in UserIpMap.getAll():

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        time.sleep(2)

        clone_url = _construct_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_ip_restriction_git(self):

        user_model = UserModel()

        try:

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

            Session().commit()

            clone_url = _construct_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.getAll():

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        time.sleep(2)

        clone_url = _construct_url(GIT_REPO)

        assert 'Cloning into' in stdout + stderr

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

kallithea.tests.scripts.manual_test_crawler

~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Test for crawling a project for memory usage

This should be runned just as regular script together

with a watch script that will show memory usage.

watch -n1 ./kallithea/tests/mem_watch

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Apr 21, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import cookielib

import urllib

import urllib2

import time

import os

import sys

from os.path import join as jn

from os.path import dirname as dn

__here__ = os.path.abspath(__file__)

__root__ = dn(dn(dn(__here__)))

sys.path.append(__root__)

from kallithea.lib import vcs

from kallithea.lib.compat import OrderedSet

from kallithea.lib.vcs.exceptions import RepositoryError

PASES = 3

HOST = 'http://127.0.0.1'

PORT = 5000

BASE_URI = '%s:%s/' % (HOST, PORT)

if len(sys.argv) == 2:

    BASE_URI = sys.argv[1]

if not BASE_URI.endswith('/'):

    BASE_URI += '/'

print 'Crawling @ %s' % BASE_URI

BASE_URI += '%s'

PROJECT_PATH = jn('/', 'home', 'username', 'repos')

PROJECTS = [

    #'linux-magx-pbranch',

    'CPython',

    'kallithea',

]

o = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))

o.addheaders = [

    ('User-agent', 'kallithea-crawler'),

    ('Accept-Language', 'en - us, en;q = 0.5')

]

urllib2.install_opener(o)

def _get_repo(proj):

    if isinstance(proj, basestring):

        repo = vcs.get_repo(jn(PROJECT_PATH, proj))

        proj = proj

    else:

        repo = proj

        proj = repo.name

    return repo, proj

def test_changelog_walk(proj, pages=100):

    repo, proj = _get_repo(proj)

    total_time = 0

"""

Unit tests configuration module for vcs.

"""

import os

import time

import hashlib

import tempfile

import datetime

import shutil

import uuid

from os.path import join as jn

__all__ = (

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

    'SCM_TESTS',

)

SCM_TESTS = ['hg', 'git']

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

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

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

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

                              jn(TEST_TMP_PATH, 'vcs-git'))

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

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

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

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

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

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

                              jn(TEST_TMP_PATH, 'vcs-hg'))

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

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

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

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

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

TEST_REPO_PREFIX = 'vcs-test'

def get_new_dir(title=None):

    """

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

    Resulting directory name will have format:

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

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

            pytest.fail('Cannot test git clone repo as location %s already '

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

        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

            tricky_path = get_new_dir("tricky-path-repo-$'`")

        else:

            tricky_path = get_new_dir("tricky-path-repo-$'\"`")

        successfully_cloned = GitRepository(tricky_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)

        # Repo should have been created

        self.assertFalse(successfully_cloned._repo.bare)

        if sys.platform == 'win32':

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

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

            pytest.fail('Cannot test mercurial clone repo as location %s already '

                      'exists. You should manually remove it first.'

                      % TEST_HG_REPO_CLONE)

    def setUp(self):