Mercurial > kallithea

changeset 7187:522cfb2be9e1

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests: add custom hook tests Git testing is omitted for now - errors are not reported as they are for Hg.
author domruf <dominikruf@gmail.com>
date Sun, 26 Nov 2017 13:12:46 +0100
parents 79fd9b50bd3e
children 02aef1484695
files kallithea/tests/fixture.py kallithea/tests/other/test_vcs_operations.py
diffstat 2 files changed, 113 insertions(+), 1 deletions(-) [+]
line wrap: on
line diff
--- a/kallithea/tests/fixture.py	Thu Jan 25 23:04:37 2018 +0100
+++ b/kallithea/tests/fixture.py	Sun Nov 26 13:12:46 2017 +0100
@@ -416,3 +416,17 @@
                          repo_location=repo_location) \
         .run(full_index=full_index)
     l.release()
+
+
+def failing_test_hook(ui, repo, **kwargs):
+    ui.write("failing_test_hook failed\n")
+    return 1
+
+
+def exception_test_hook(ui, repo, **kwargs):
+    raise Exception("exception_test_hook threw an exception")
+
+
+def passing_test_hook(ui, repo, **kwargs):
+    ui.write("passing_test_hook succeeded\n")
+    return 0
--- a/kallithea/tests/other/test_vcs_operations.py	Thu Jan 25 23:04:37 2018 +0100
+++ b/kallithea/tests/other/test_vcs_operations.py	Sun Nov 26 13:12:46 2017 +0100
@@ -36,7 +36,7 @@
 
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
-from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation
+from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation, Ui
 from kallithea.model.meta import Session
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
@@ -136,6 +136,15 @@
     return stdout, stderr
 
 
+def _check_outgoing(vcs, cwd, clone_url=''):
+    if vcs == 'hg':
+        # hg removes the password from default URLs, so we have to provide it here via the clone_url
+        return Command(cwd).execute('hg -q outgoing', clone_url, ignoreReturnCode=True)
+    elif vcs == 'git':
+        Command(cwd).execute('git remote update')
+        return Command(cwd).execute('git log origin/master..master')
+
+
 def set_anonymous_access(enable=True):
     user = User.get_default_user()
     user.active = enable
@@ -177,6 +186,25 @@
         r.enable_locking = False
         Session().commit()
 
+    @pytest.fixture()
+    def testhook_cleanup(self):
+        yield
+        # remove hook
+        for hook in ['prechangegroup', 'pretxnchangegroup', 'preoutgoing', 'changegroup', 'outgoing', 'incoming']:
+            entry = Ui.get_by_key('hooks', '%s.testhook' % hook)
+            if entry:
+                Session().delete(entry)
+        Session().commit()
+
+    @pytest.fixture(scope="module")
+    def testfork(self):
+        # create fork so the repo stays untouched
+        git_fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
+        fixture.create_fork(GIT_REPO, git_fork_name)
+        hg_fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
+        fixture.create_fork(HG_REPO, hg_fork_name)
+        return {'git': git_fork_name, 'hg': hg_fork_name}
+
     def test_clone_hg_repo_by_admin(self, webserver):
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())
@@ -569,3 +597,73 @@
 
         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 == ''