# 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