Mercurial > kallithea

changeset 6609:29b69b044494

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests: refactor _construct_url in the manual vcs test to actually return a url - dest is handled separately Based on work by Dominik Ruf.
author Mads Kiilerich <mads@kiilerich.com>
date Wed, 03 May 2017 04:17:27 +0200
parents 52e689b37700
children 58931c2b6b0c
files kallithea/tests/other/manual_test_vcs_operations.py
diffstat 1 files changed, 46 insertions(+), 49 deletions(-) [+]
line wrap: on
line diff
--- a/kallithea/tests/other/manual_test_vcs_operations.py	Thu Apr 27 20:29:28 2017 +0200
+++ b/kallithea/tests/other/manual_test_vcs_operations.py	Wed May 03 04:17:27 2017 +0200
@@ -85,22 +85,20 @@
     return tempfile.mkdtemp(prefix='rc_integration_test')
 
 
-def _construct_url(repo, dest=None, **kwargs):
-    if dest is None:
-        #make temp clone
-        dest = _get_tmp_dir()
+def _construct_url(repo, **kwargs):
+    """Return a clone url for the provided repo path.
+    Optional named parameters: user, passwd and host."""
     params = {
         'user': TEST_USER_ADMIN_LOGIN,
         'passwd': TEST_USER_ADMIN_PASS,
         'host': HOST,
         'cloned_repo': repo,
-        'dest': dest
     }
     params.update(**kwargs)
     if params['user'] and params['passwd']:
-        _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s %(dest)s' % params
+        _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s' % params
     else:
-        _url = 'http://(host)s/%(cloned_repo)s %(dest)s' % params
+        _url = 'http://(host)s/%(cloned_repo)s' % params
     return _url
 
 
@@ -145,7 +143,6 @@
     elif vcs == 'git':
         _REPO = GIT_REPO
 
-    kwargs['dest'] = ''
     clone_url = _construct_url(_REPO, **kwargs)
     if 'clone_url' in kwargs:
         clone_url = kwargs['clone_url']
@@ -153,7 +150,7 @@
     if vcs == 'hg':
         stdout, stderr = Command(cwd).execute('hg push --verbose', clone_url)
     elif vcs == 'git':
-        stdout, stderr = Command(cwd).execute('git push --verbose', clone_url + " master")
+        stdout, stderr = Command(cwd).execute('git push --verbose', clone_url, "master")
 
     return stdout, stderr
 
@@ -201,7 +198,7 @@
 
     def test_clone_hg_repo_by_admin(self):
         clone_url = _construct_url(HG_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
 
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
@@ -212,45 +209,45 @@
 
     def test_clone_git_repo_by_admin(self):
         clone_url = _construct_url(GIT_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
 
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''
 
     def test_clone_wrong_credentials_hg(self):
         clone_url = _construct_url(HG_REPO, passwd='bad!')
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
         assert 'abort: authorization failed' in stderr
 
     def test_clone_wrong_credentials_git(self):
         clone_url = _construct_url(GIT_REPO, passwd='bad!')
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
         assert 'fatal: Authentication failed' in stderr
 
     def test_clone_git_dir_as_hg(self):
         clone_url = _construct_url(GIT_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
         assert 'HTTP Error 404: Not Found' in stderr
 
     def test_clone_hg_repo_as_git(self):
         clone_url = _construct_url(HG_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
         assert 'not found' in stderr
 
     def test_clone_non_existing_path_hg(self):
         clone_url = _construct_url('trololo')
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
         assert 'HTTP Error 404: Not Found' in stderr
 
     def test_clone_non_existing_path_git(self):
         clone_url = _construct_url('trololo')
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
         assert 'not found' in stderr
 
     def test_push_new_file_hg(self):
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        clone_url = _construct_url(HG_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
 
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
@@ -263,8 +260,8 @@
 
     def test_push_new_file_git(self):
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        clone_url = _construct_url(GIT_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
 
         # commit some stuff into this repo
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
@@ -285,8 +282,8 @@
         Session().commit()
 
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        clone_url = _construct_url(HG_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
 
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
@@ -308,8 +305,8 @@
         Session().commit()
 
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        clone_url = _construct_url(GIT_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
 
         # commit some stuff into this repo
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
@@ -324,8 +321,8 @@
 
     def test_push_wrong_credentials_hg(self):
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        clone_url = _construct_url(HG_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
 
         stdout, stderr = _add_files_and_push('hg', DEST, user='bad',
                                              passwd='name')
@@ -334,8 +331,8 @@
 
     def test_push_wrong_credentials_git(self):
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        clone_url = _construct_url(GIT_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
 
         stdout, stderr = _add_files_and_push('git', DEST, user='bad',
                                              passwd='name')
@@ -344,8 +341,8 @@
 
     def test_push_back_to_wrong_url_hg(self):
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        clone_url = _construct_url(HG_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
 
         stdout, stderr = _add_files_and_push('hg', DEST,
                                     clone_url='http://%s/tmp' % HOST)
@@ -354,8 +351,8 @@
 
     def test_push_back_to_wrong_url_git(self):
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        clone_url = _construct_url(GIT_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
 
         stdout, stderr = _add_files_and_push('git', DEST,
                                     clone_url='http://%s/tmp' % HOST)
@@ -369,7 +366,7 @@
         Session().commit()
         # clone
         clone_url = _construct_url(HG_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
 
         #check if lock was made
         r = Repository.get_by_repo_name(HG_REPO)
@@ -382,7 +379,7 @@
         Session().commit()
         # clone
         clone_url = _construct_url(GIT_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
 
         #check if lock was made
         r = Repository.get_by_repo_name(GIT_REPO)
@@ -394,7 +391,7 @@
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         #pull fails since repo is locked
         clone_url = _construct_url(HG_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
@@ -405,15 +402,15 @@
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         #pull fails since repo is locked
         clone_url = _construct_url(GIT_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
         msg = ("""The requested URL returned error: 423""")
         assert msg in stderr
 
     def test_push_on_locked_repo_by_other_user_hg(self):
         #clone some temp
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(HG_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        clone_url = _construct_url(HG_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
 
         #lock repo
         r = Repository.get_by_repo_name(HG_REPO)
@@ -434,8 +431,8 @@
     def test_push_on_locked_repo_by_other_user_git(self):
         #clone some temp
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(GIT_REPO, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        clone_url = _construct_url(GIT_REPO)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
 
         #lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
@@ -469,8 +466,8 @@
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(fork_name, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        clone_url = _construct_url(fork_name)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)
 
         #check for lock repo after clone
         r = Repository.get_by_repo_name(fork_name)
@@ -495,8 +492,8 @@
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
-        clone_url = _construct_url(fork_name, dest=DEST)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        clone_url = _construct_url(fork_name)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)
 
         #check for lock repo after clone
         r = Repository.get_by_repo_name(fork_name)
@@ -518,7 +515,7 @@
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             clone_url = _construct_url(HG_REPO)
-            stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+            stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
             assert 'abort: HTTP Error 403: Forbidden' in stderr
         finally:
             #release IP restrictions
@@ -530,7 +527,7 @@
         time.sleep(1.5)
 
         clone_url = _construct_url(HG_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())
 
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
@@ -545,7 +542,7 @@
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             clone_url = _construct_url(GIT_REPO)
-            stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+            stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
             # The message apparently changed in Git 1.8.3, so match it loosely.
             assert re.search(r'\b403\b', stderr)
         finally:
@@ -558,7 +555,7 @@
         time.sleep(1.5)
 
         clone_url = _construct_url(GIT_REPO)
-        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
+        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())
 
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''