kallithea Changeset - 1e0cd9f75fbc

Changeset - 1e0cd9f75fbc

Parent rev.

Child rev.

[Not reviewed]

default

0 1 0

Mads Kiilerich (mads) - 6 years ago 2020-06-24 23:58:53
mads@kiilerich.com

Grafted from: bb99b8bfa16c

tests: fix test_vcs_operations for MariaDB/MySQL

This is apparently only needed for MariaDB/MySQL where it will fix tests that
were failing due to missing action log entries.

1 file changed with 5 insertions and 2 deletions:

kallithea/tests/other/test_vcs_operations.py

0 comments (0 inline, 0 general)

kallithea/tests/other/test_vcs_operations.py

➞

Show inline comments

@@ @@ -349,24 +349,25 @@ class TestVCSOperations(base.TestControl @@
         stdout, stderr = _add_files_and_push(webserver, vt, local_clone_dir, clone_url=clone_url)
         if vt.repo_type == 'git':
             _check_proper_git_push(stdout, stderr)
         elif vt.repo_type == 'hg':
             assert 'pushing to ' in stdout
             assert 'remote: added ' in stdout
         # Verify that we got the right events in UserLog. Expect something like:
         # <UserLog('id:new_git_XXX:started_following_repo')>
         # <UserLog('id:new_git_XXX:user_created_repo')>
         # <UserLog('id:new_git_XXX:pull')>
         # <UserLog('id:new_git_XXX:push:aed9d4c1732a1927da3be42c47eb9afdc200d427,d38b083a07af10a9f44193486959a96a23db78da,4841ff9a2b385bec995f4679ef649adb3f437622')>
         Session.close()  # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...)
         action_parts = [ul.action.split(':', 1) for ul in UserLog.query().order_by(UserLog.user_log_id)]
         assert [(t[0], (t[1].count(',') + 1) if len(t) == 2 else 0) for t in action_parts] == ([
             ('started_following_repo', 0),
             ('user_created_repo', 0),
             ('pull', 0),
             ('push', 3)]
             if vt.repo_type == 'git' else [
             ('started_following_repo', 0),
             ('user_created_repo', 0),
             # (u'pull', 0), # Mercurial outgoing hook is not called for empty clones
             ('push', 3)])
@@ @@ -380,38 +381,40 @@ class TestVCSOperations(base.TestControl @@
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type])
         stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, clone_url=clone_url)
         if vt.repo_type == 'git':
             _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
         Session.close()  # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...)
         action_parts = [ul.action.split(':', 1) for ul in UserLog.query().order_by(UserLog.user_log_id)]
         assert [(t[0], (t[1].count(',') + 1) if len(t) == 2 else 0) for t in action_parts] == \
             [('pull', 0), ('push', 3)]
     @parametrize_vcs_test
     def test_pull(self, webserver, testfork, vt):
         UserLog.query().delete()
         Session().commit()
         dest_dir = _get_tmp_dir()
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'init', dest_dir)
         clone_url = vt.repo_url_param(webserver, vt.repo_name)
         stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url)
         Session.close()  # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...)
         if vt.repo_type == 'git':
             assert 'FETCH_HEAD' in stderr
         elif vt.repo_type == 'hg':
             assert 'new changesets' in stdout
         action_parts = [ul.action for ul in UserLog.query().order_by(UserLog.user_log_id)]
         assert action_parts == ['pull']
         # Test handling of URLs with extra '/' around repo_name
         stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/./%s/' % vt.repo_name), ignoreReturnCode=True)
         if issubclass(vt, HttpVcsTest):
@@ @@ -439,29 +442,28 @@ class TestVCSOperations(base.TestControl @@
         assert "404" not in stdout
     @parametrize_vcs_test
     def test_push_invalidates_cache(self, webserver, testfork, vt):
         pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in Repository.query().filter(Repository.repo_name == testfork[vt.repo_type])]
         dest_dir = _get_tmp_dir()
         clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type])
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, files_no=1, clone_url=clone_url)
         Session().commit()  # expire test session to make sure SA fetch new Repository instances after last_changeset has been updated server side hook in other process
         if vt.repo_type == 'git':
             _check_proper_git_push(stdout, stderr)
         Session.close()  # expire session to make sure SA fetches new Repository instances after last_changeset has been updated by server side hook in another process
         post_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in Repository.query().filter(Repository.repo_name == testfork[vt.repo_type])]
         assert pre_cached_tip != post_cached_tip
     @parametrize_vcs_test_http
     def test_push_wrong_credentials(self, webserver, vt):
         dest_dir = _get_tmp_dir()
         clone_url = vt.repo_url_param(webserver, vt.repo_name)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         clone_url = webserver.repo_url(vt.repo_name, username='bad', password='name')
         stdout, stderr = _add_files_and_push(webserver, vt, dest_dir,
                                              clone_url=clone_url, ignoreReturnCode=True)
@@ @@ -478,24 +480,25 @@ class TestVCSOperations(base.TestControl @@
         dest_dir = _get_tmp_dir()
         clone_url = vt.repo_url_param(webserver, vt.repo_name, username=base.TEST_USER_REGULAR_LOGIN, password=base.TEST_USER_REGULAR_PASS)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, ignoreReturnCode=True, clone_url=clone_url)
         if vt.repo_type == 'git':
             assert 'The requested URL returned error: 403' in stderr or 'abort: Push access to %r denied' % str(vt.repo_name) in stderr
         elif vt.repo_type == 'hg':
             assert 'abort: HTTP Error 403: Forbidden' in stderr or 'abort: push failed on remote' in stderr and 'remote: Push access to %r denied' % str(vt.repo_name) in stdout
         Session.close()  # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...)
         action_parts = [ul.action.split(':', 1) for ul in UserLog.query().order_by(UserLog.user_log_id)]
         assert [(t[0], (t[1].count(',') + 1) if len(t) == 2 else 0) for t in action_parts] == \
             [('pull', 0)]
     @parametrize_vcs_test
     def test_push_back_to_wrong_url(self, webserver, vt):
         dest_dir = _get_tmp_dir()
         clone_url = vt.repo_url_param(webserver, vt.repo_name)
         stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(
             webserver, vt, dest_dir, clone_url='http://%s:%s/tmp' % (

0 comments (0 inline, 0 general)