Mercurial > kallithea
changeset 7676:902cbe668001
tests: run the same vcs tests both for hg and git using http ... and soon also ssh
Parametrize vcs tests and remove duplicate code.
Refactored by Mads Kiilerich.
author | domruf <dominikruf@gmail.com> |
---|---|
date | Tue, 13 Mar 2018 23:52:10 +0100 |
parents | c8239333853d |
children | cffb5e5bf7d6 |
files | kallithea/tests/other/test_vcs_operations.py |
diffstat | 1 files changed, 159 insertions(+), 196 deletions(-) [+] |
line wrap: on
line diff
--- a/kallithea/tests/other/test_vcs_operations.py Wed Jan 16 12:55:10 2019 +0100 +++ b/kallithea/tests/other/test_vcs_operations.py Tue Mar 13 23:52:10 2018 +0100 @@ -49,17 +49,51 @@ fixture = Fixture() +# Parameterize different kinds of VCS testing - both the kind of VCS and the +# access method (HTTP/SSH) + +class HttpVcsTest(object): + @staticmethod + def repo_url_param(webserver, repo_name, **kwargs): + return webserver.repo_url(repo_name, **kwargs) + +class HgVcsTest(object): + repo_type = 'hg' + repo_name = HG_REPO + +class GitVcsTest(object): + repo_type = 'git' + repo_name = GIT_REPO + +class HgHttpVcsTest(HgVcsTest, HttpVcsTest): + pass + +class GitHttpVcsTest(GitVcsTest, HttpVcsTest): + pass + +parametrize_vcs_test = parametrize('vt', [ + HgHttpVcsTest, + GitHttpVcsTest, +]) +parametrize_vcs_test_hg = parametrize('vt', [ + HgHttpVcsTest, +]) +parametrize_vcs_test_http = parametrize('vt', [ + HgHttpVcsTest, + GitHttpVcsTest, +]) + class Command(object): def __init__(self, cwd): self.cwd = cwd - def execute(self, cmd, *args, **environ): + def execute(self, *args, **environ): """ - Runs command on the system with given ``args``. + Runs command on the system with given ``args`` using simple space + join without safe quoting. """ - - command = cmd + ' ' + ' '.join(args) + command = ' '.join(args) ignoreReturnCode = environ.pop('ignoreReturnCode', False) if DEBUG: print '*** CMD %s ***' % command @@ -95,7 +129,7 @@ """ added_file = '%ssetup.py' % _RandomNameSequence().next() open(os.path.join(dest_dir, added_file), 'a').close() - Command(dest_dir).execute('%s add %s' % (vcs, added_file)) + Command(dest_dir).execute(vcs, 'add', added_file) email = 'me@example.com' if os.name == 'nt': @@ -195,33 +229,29 @@ 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()) - - assert 'requesting all changes' in stdout - assert 'adding changesets' in stdout - assert 'adding manifests' in stdout - assert 'adding file changes' in stdout - - assert stderr == '' + @parametrize_vcs_test + def test_clone_repo_by_admin(self, webserver, vt): + clone_url = webserver.repo_url(vt.repo_name) + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir()) - def test_clone_git_repo_by_admin(self, webserver): - 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 == '' + if vt.repo_type == 'git': + assert 'Cloning into' in stdout + stderr + assert stderr == '' or stdout == '' + elif vt.repo_type == 'hg': + 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_wrong_credentials_hg(self, webserver): - clone_url = webserver.repo_url(HG_REPO, password='bad!') - stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) - assert 'abort: authorization failed' in stderr - - def test_clone_wrong_credentials_git(self, webserver): - clone_url = webserver.repo_url(GIT_REPO, password='bad!') - stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) - assert 'fatal: Authentication failed' in stderr + @parametrize_vcs_test_http + def test_clone_wrong_credentials(self, webserver, vt): + clone_url = webserver.repo_url(vt.repo_name, password='bad!') + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) + if vt.repo_type == 'git': + assert 'fatal: Authentication failed' in stderr + elif vt.repo_type == 'hg': + assert 'abort: authorization failed' in stderr def test_clone_git_dir_as_hg(self, webserver): clone_url = webserver.repo_url(GIT_REPO) @@ -233,16 +263,16 @@ stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) assert 'not found' in stderr - def test_clone_non_existing_path_hg(self, webserver): + @parametrize_vcs_test + def test_clone_non_existing_path(self, webserver, vt): clone_url = webserver.repo_url('trololo') - stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) - assert 'HTTP Error 404: Not Found' in stderr + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) + if vt.repo_type == 'git': + assert 'not found' in stderr + elif vt.repo_type == 'hg': + assert 'HTTP Error 404: Not Found' in stderr - def test_clone_non_existing_path_git(self, webserver): - clone_url = webserver.repo_url('trololo') - stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) - assert 'not found' in stderr - + # TODO: use @parametrize_vcs_test and run on hg def test_push_new_repo_git(self, webserver): # Clear the log so we know what is added UserLog.query().delete() @@ -294,145 +324,106 @@ assert uls[3].action.startswith(u'push:') assert uls[3].action.count(',') == 2 # expect 3 commits - def test_push_new_file_hg(self, webserver, testfork): + @parametrize_vcs_test + def test_push_new_file(self, webserver, testfork, vt): 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) - - clone_url = webserver.repo_url(testfork['hg']) - stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, clone_url=clone_url) + clone_url = webserver.repo_url(vt.repo_name) + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) - assert 'pushing to' in stdout - assert 'Repository size' in stdout - assert 'Last revision is now' in stdout - - def test_push_new_file_git(self, webserver, testfork): - 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) - - # commit some stuff into this repo - clone_url = webserver.repo_url(testfork['git']) - stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir, clone_url=clone_url) - print [(x.repo_full_path,x.repo_path) for x in Repository.query()] # TODO: what is this for - _check_proper_git_push(stdout, stderr) + clone_url = webserver.repo_url(testfork[vt.repo_type]) + stdout, stderr = _add_files_and_push(webserver, vt.repo_type, dest_dir, clone_url=clone_url) - def test_push_invalidates_cache_hg(self, webserver, testfork): - key = CacheInvalidation.query().filter(CacheInvalidation.cache_key - == HG_REPO).scalar() - if not key: - key = CacheInvalidation(HG_REPO, HG_REPO) - Session().add(key) - - key.cache_active = True - Session().commit() + if vt.repo_type == 'git': + print [(x.repo_full_path, x.repo_path) for x in Repository.query()] # TODO: what is this for + _check_proper_git_push(stdout, stderr) + elif vt.repo_type == 'hg': + assert 'pushing to' in stdout + assert 'Repository size' in stdout + assert 'Last revision is now' in stdout - 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) - + @parametrize_vcs_test + def test_push_invalidates_cache(self, webserver, testfork, vt): 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() + == vt.repo_name).scalar() if not key: - key = CacheInvalidation(GIT_REPO, GIT_REPO) + key = CacheInvalidation(vt.repo_name, vt.repo_name) 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) + clone_url = webserver.repo_url(testfork[vt.repo_type]) + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) + + stdout, stderr = _add_files_and_push(webserver, vt.repo_type, dest_dir, files_no=1, clone_url=clone_url) - # 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) - + if vt.repo_type == 'git': + _check_proper_git_push(stdout, stderr) key = CacheInvalidation.query().filter(CacheInvalidation.cache_key - == testfork['git']).all() + == testfork[vt.repo_type]).all() assert key == [] - def test_push_wrong_credentials_hg(self, webserver): + @parametrize_vcs_test_http + def test_push_wrong_credentials(self, webserver, vt): 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) + clone_url = webserver.repo_url(vt.repo_name) + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) - 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', + stdout, stderr = _add_files_and_push(webserver, vt.repo_type, dest_dir, username='bad', password='name', ignoreReturnCode=True) - assert 'fatal: Authentication failed' in stderr + if vt.repo_type == 'git': + assert 'fatal: Authentication failed' in stderr + elif vt.repo_type == 'hg': + assert 'abort: authorization failed' in stderr - def test_push_with_readonly_credentials_hg(self, webserver): + @parametrize_vcs_test + def test_push_with_readonly_credentials(self, webserver, vt): 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) + clone_url = webserver.repo_url(vt.repo_name, username=TEST_USER_REGULAR_LOGIN, password=TEST_USER_REGULAR_PASS) + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) - stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, username=TEST_USER_REGULAR_LOGIN, + stdout, stderr = _add_files_and_push(webserver, vt.repo_type, 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) + if vt.repo_type == 'git': + assert 'The requested URL returned error: 403' in stderr + elif vt.repo_type == 'hg': + assert 'abort: HTTP Error 403: Forbidden' in stderr - 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): + @parametrize_vcs_test + def test_push_back_to_wrong_url(self, webserver, vt): 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) + clone_url = webserver.repo_url(vt.repo_name) + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) stdout, stderr = _add_files_and_push( - webserver, 'hg', dest_dir, clone_url='http://%s:%s/tmp' % ( + webserver, vt.repo_type, 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) + if vt.repo_type == 'git': + assert 'not found' in stderr + elif vt.repo_type == 'hg': + assert 'HTTP Error 404: Not Found' in stderr - 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_ip_restriction_hg(self, webserver): + @parametrize_vcs_test + def test_ip_restriction(self, webserver, vt): user_model = UserModel() try: + # Add IP constraint that excludes the test context: user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32') 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(), ignoreReturnCode=True) - assert 'abort: HTTP Error 403: Forbidden' in stderr + clone_url = webserver.repo_url(vt.repo_name) + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) + if vt.repo_type == 'git': + # The message apparently changed in Git 1.8.3, so match it loosely. + assert re.search(r'\b403\b', stderr) + elif vt.repo_type == 'hg': + assert 'abort: HTTP Error 403: Forbidden' in stderr finally: # release IP restrictions for ip in UserIpMap.query(): @@ -441,74 +432,46 @@ # 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 == '' + clone_url = webserver.repo_url(vt.repo_name) + stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir()) - 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() - # 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(), 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) + if vt.repo_type == 'git': + assert 'Cloning into' in stdout + stderr + assert stderr == '' or stdout == '' + elif vt.repo_type == 'hg': + assert 'requesting all changes' in stdout + assert 'adding changesets' in stdout + assert 'adding manifests' in stdout + assert 'adding file changes' in stdout - 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 == '' + assert stderr == '' - @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): + @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks + def test_custom_hooks_preoutgoing(self, testhook_cleanup, webserver, testfork, vt): # 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) + clone_url = webserver.repo_url(testfork[vt.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': + .execute(vt.repo_type, 'clone', clone_url, dest_dir, ignoreReturnCode=True) + if vt.repo_type == 'hg': assert 'preoutgoing.testhook hook failed' in stdout - elif repo_type == 'git': + elif vt.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): - + @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks + def test_custom_hooks_prechangegroup(self, testhook_cleanup, webserver, testfork, vt): # 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) + clone_url = webserver.repo_url(testfork[vt.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 = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) - stdout, stderr = _add_files_and_push(webserver, repo_type, dest_dir, + stdout, stderr = _add_files_and_push(webserver, vt.repo_type, dest_dir, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS, ignoreReturnCode=True) @@ -516,37 +479,37 @@ 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) + stdout, stderr = _check_outgoing(vt.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': + stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True) + if vt.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': + elif vt.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) + stdout, stderr = _check_outgoing(vt.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) + stdout, stderr = Command(dest_dir).execute('%s push' % vt.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) + stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url) assert stdout == '' assert stderr == '' - def test_add_submodule(self, webserver, testfork): + def test_add_submodule_git(self, webserver, testfork): dest_dir = _get_tmp_dir() clone_url = webserver.repo_url(GIT_REPO)