if not key:

            key = CacheInvalidation(HG_REPO, HG_REPO)

            Session().add(key)

        key.cache_active = True

        Session().commit()

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(testfork['hg'])

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)

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

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

                                               == testfork['hg']).all()

        assert key == []

    def test_push_invalidates_cache_git(self, webserver, testfork):

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

                                               == GIT_REPO).scalar()

        if not key:

            key = CacheInvalidation(GIT_REPO, GIT_REPO)

            Session().add(key)

        key.cache_active = True

        Session().commit()

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(testfork['git'])

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)

        # commit some stuff into this repo

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

        _check_proper_git_push(stdout, stderr)

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

                                               == testfork['git']).all()

        assert key == []

    def test_push_wrong_credentials_hg(self, webserver):

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(HG_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)

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

                                             password='name', ignoreReturnCode=True)

        assert 'abort: authorization failed' in stderr

    def test_push_wrong_credentials_git(self, webserver):

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)

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

                                             password='name', ignoreReturnCode=True)

        assert 'fatal: Authentication failed' in stderr

    def test_push_with_readonly_credentials_hg(self, webserver):

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(HG_REPO, username=TEST_USER_REGULAR_LOGIN, password=TEST_USER_REGULAR_PASS)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, username=TEST_USER_REGULAR_LOGIN,

                                             password=TEST_USER_REGULAR_PASS, ignoreReturnCode=True)

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

    def test_push_with_readonly_credentials_git(self, webserver):

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO, username=TEST_USER_REGULAR_LOGIN, password=TEST_USER_REGULAR_PASS)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir, username=TEST_USER_REGULAR_LOGIN,

                                             password=TEST_USER_REGULAR_PASS, ignoreReturnCode=True)

        assert 'The requested URL returned error: 403' in stderr

    def test_push_back_to_wrong_url_hg(self, webserver):

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(HG_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(

            webserver, 'hg', dest_dir, 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_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(

            webserver, 'git', dest_dir, clone_url='http://%s:%s/tmp' % (

                webserver.server_address[0], webserver.server_address[1]),

            ignoreReturnCode=True)

        assert 'not found' in stderr

    def test_clone_and_create_lock_hg(self, webserver):

        # enable locking

        r = Repository.get_by_repo_name(HG_REPO)

        r.enable_locking = True

        Session().commit()

        # clone

        clone_url = webserver.repo_url(HG_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())

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

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir())

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

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        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)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        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_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(HG_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)

        # 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(webserver, 'hg', dest_dir,

                                             username=TEST_USER_REGULAR_LOGIN,

                                             password=TEST_USER_REGULAR_PASS,

                                             ignoreReturnCode=True)

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

        # 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_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)

        # 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(webserver, 'git', dest_dir,

                                             username=TEST_USER_REGULAR_LOGIN,

                                             password=TEST_USER_REGULAR_PASS,

                                             ignoreReturnCode=True)

        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, webserver, testfork):

        # enable locking

        r = Repository.get_by_repo_name(testfork['hg'])

        r.enable_locking = True

        Session().commit()

        # clone some temp

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(testfork['hg'])

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)

        # check for lock repo after clone

        r = Repository.get_by_repo_name(testfork['hg'])

        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(webserver, 'hg', dest_dir, clone_url=clone_url)

        assert str('remote: Released lock on repo `%s`' % testfork['hg']) in stdout

        # we need to cleanup the Session Here !

        Session.remove()

        r = Repository.get_by_repo_name(testfork['hg'])

        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, webserver, testfork):

        # enable locking

        r = Repository.get_by_repo_name(testfork['git'])

        r.enable_locking = True

        Session().commit()

        # clone some temp

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(testfork['git'])

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)

        # check for lock repo after clone

        r = Repository.get_by_repo_name(testfork['git'])

        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(webserver, 'git', dest_dir, clone_url=clone_url)

        _check_proper_git_push(stdout, stderr)

        assert ('remote: Released lock on repo `%s`' % testfork['git']) in stderr

        # we need to cleanup the Session Here !

        Session.remove()

        r = Repository.get_by_repo_name(testfork['git'])

        assert r.locked == [None, None]

    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)

            stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

            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)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())

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

            stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

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

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir())

        assert 'Cloning into' in stdout + stderr

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

    @parametrize('repo_type, repo_name', [

                #('git',      GIT_REPO), # git hooks doesn't work like hg hooks

                ('hg',       HG_REPO),

    ])

    def test_custom_hooks_preoutgoing(self, testhook_cleanup, webserver, testfork, repo_type, repo_name):

        # set prechangegroup to failing hook (returns True)

        Ui.create_or_update_hook('preoutgoing.testhook', 'python:kallithea.tests.fixture.failing_test_hook')

        Session().commit()

        # clone repo

        clone_url = webserver.repo_url(testfork[repo_type], username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS)

        dest_dir = _get_tmp_dir()

        stdout, stderr = Command(TESTS_TMP_PATH) \

            .execute('%s clone' % repo_type, clone_url, dest_dir, ignoreReturnCode=True)

        if repo_type == 'hg':

            assert 'preoutgoing.testhook hook failed' in stdout

        elif repo_type == 'git':

            assert 'error: 406' in stderr

    @parametrize('repo_type, repo_name', [

                #('git',      GIT_REPO), # git hooks doesn't work like hg hooks

                ('hg',       HG_REPO),

    ])

    def test_custom_hooks_prechangegroup(self, testhook_cleanup, webserver, testfork, repo_type, repo_name):

        # set prechangegroup to failing hook (returns True)

        Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.failing_test_hook')

        Session().commit()

        # clone repo

        clone_url = webserver.repo_url(testfork[repo_type], username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS)

        dest_dir = _get_tmp_dir()

        stdout, stderr = Command(TESTS_TMP_PATH).execute('%s clone' % repo_type, clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(webserver, repo_type, dest_dir,

                                             username=TEST_USER_ADMIN_LOGIN,

                                             password=TEST_USER_ADMIN_PASS,

                                             ignoreReturnCode=True)

        assert 'failing_test_hook failed' in stdout + stderr

        assert 'Traceback' not in stdout + stderr

        assert 'prechangegroup.testhook hook failed' in stdout + stderr

        # there are still outgoing changesets

        stdout, stderr = _check_outgoing(repo_type, dest_dir, clone_url)

        assert stdout != ''

        # set prechangegroup hook to exception throwing method

        Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.exception_test_hook')

        Session().commit()

        # re-try to push

        stdout, stderr = Command(dest_dir).execute('%s push' % repo_type, clone_url, ignoreReturnCode=True)

        if repo_type == 'hg':

            # like with 'hg serve...' 'HTTP Error 500: INTERNAL SERVER ERROR' should be returned

            assert 'HTTP Error 500: INTERNAL SERVER ERROR' in stderr

        elif repo_type == 'git':

            assert 'exception_test_hook threw an exception' in stderr

        # there are still outgoing changesets

        stdout, stderr = _check_outgoing(repo_type, dest_dir, clone_url)

        assert stdout != ''

        # set prechangegroup hook to method that returns False

        Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.passing_test_hook')

        Session().commit()

        # re-try to push

        stdout, stderr = Command(dest_dir).execute('%s push' % repo_type, clone_url, ignoreReturnCode=True)

        assert 'passing_test_hook succeeded' in stdout + stderr

        assert 'Traceback' not in stdout + stderr

        assert 'prechangegroup.testhook hook failed' not in stdout + stderr

        # no more outgoing changesets

        stdout, stderr = _check_outgoing(repo_type, dest_dir, clone_url)

        assert stdout == ''

        assert stderr == ''

    def test_add_submodule(self, webserver, testfork):

        dest_dir = _get_tmp_dir()

        clone_url = webserver.repo_url(GIT_REPO)

        fork_url = webserver.repo_url(testfork['git'])

        # add submodule

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', fork_url, dest_dir)

        stdout, stderr = Command(dest_dir).execute('git submodule add', clone_url, 'testsubmodule')

        stdout, stderr = Command(dest_dir).execute('git push', fork_url, 'master')

        # check for testsubmodule link in files page

        self.log_user()

        response = self.app.get(url(controller='files', action='index',

                                    repo_name=testfork['git'],

                                    revision='tip',

                                    f_path='/'))

        response.mustcontain('<a class="submodule-dir" href="%s" target="_blank"><i class="icon-file-submodule"></i><span>testsubmodule @ ' % clone_url)

        # check that following a submodule link actually works - and redirects

        response = self.app.get(url(controller='files', action='index',

                                    repo_name=testfork['git'],

                                    revision='tip',

                                    f_path='/testsubmodule'),

                                status=302)

        assert response.location == clone_url