Changeset - 731a8466fb11
Parent rev.
Child rev.
[Not reviewed]
default
0 1 0
Mads Kiilerich - 7 years ago 2019-01-16 01:56:37
mads@kiilerich.com
Grafted from: bd75bd05e06e
tests: refactor vcs _add_files_and_push - pass url_fixture
1 file changed with 11 insertions and 11 deletions:
kallithea/tests/other/test_vcs_operations.py
11
11
0 comments (0 inline, 0 general)
kallithea/tests/other/test_vcs_operations.py
➞
Show inline comments
 
@@ -147,19 +147,19 @@ def _add_files(vcs, dest_dir, files_no=3
 
            cmd = """git commit -m "committed new %s" --author "%s" "%s" """ % (
 
                i, author_str, added_file
 
            )
 
        # git commit needs EMAIL on some machines
 
        Command(dest_dir).execute(cmd, EMAIL=email)
 

			




	
	
	
		
 
def _add_files_and_push(webserver, vcs, dest_dir, clone_url, ignoreReturnCode=False, files_no=3):

			




	
	
	
		
 
    _add_files(vcs, dest_dir, files_no=files_no)

			




	
	
	
		
 
def _add_files_and_push(webserver, vt, dest_dir, clone_url, ignoreReturnCode=False, files_no=3):

			




	
	
	
		
 
    _add_files(vt.repo_type, dest_dir, files_no=files_no)

			




	
	
	
		
 
    # PUSH it back

			




	
	
	
		
 
    stdout = stderr = None

			




	
	
	
		
 
    if vcs == 'hg':

			




	
	
	
		
 
    if vt.repo_type == 'hg':

			




	
	
	
		
 
        stdout, stderr = Command(dest_dir).execute('hg push --verbose', clone_url, ignoreReturnCode=ignoreReturnCode)

			




	
	
	
		
 
    elif vcs == 'git':

			




	
	
	
		
 
    elif vt.repo_type == 'git':

			




	
	
	
		
 
        stdout, stderr = Command(dest_dir).execute('git push --verbose', clone_url, "master", ignoreReturnCode=ignoreReturnCode)

			




	
	
	
		
 

			




	
	
	
		
 
    return stdout, stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _check_outgoing(vcs, cwd, clone_url):

			




	
	
		
 
@@ -294,13 +294,13 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, local_clone_dir, ignoreReturnCode=True)

			




	
	
	
		
 

			




	
	
	
		
 
        # Make 3 commits and push to the empty server repo.

			




	
	
	
		
 
        # The server repo doesn't have any other heads than the

			




	
	
	
		
 
        # refs/heads/master we are pushing, but the `git log` in the push hook

			




	
	
	
		
 
        # should still list the 3 commits.

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, 'git', local_clone_dir, clone_url=clone_url)

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, GitHttpVcsTest, local_clone_dir, clone_url=clone_url)

			




	
	
	
		
 
        _check_proper_git_push(stdout, stderr)

			




	
	
	
		
 

			




	
	
	
		
 
        # 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')>

			




	
	
		
 
@@ -317,13 +317,13 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_push_new_file(self, webserver, testfork, vt):

			




	
	
	
		
 
        dest_dir = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(vt.repo_name)

			




	
	
	
		
 
        stdout, stderr = Command(TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

			




	
	
	
		
 

			




	
	
	
		
 
        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)

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, clone_url=clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        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

			




	
	
		
 
@@ -342,13 +342,13 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        dest_dir = _get_tmp_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)

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, files_no=1, clone_url=clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        if vt.repo_type == 'git':

			




	
	
	
		
 
            _check_proper_git_push(stdout, stderr)

			




	
	
	
		
 
        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

			




	
	
	
		
 
                                               == testfork[vt.repo_type]).all()

			




	
	
	
		
 
        assert key == []

			




	
	
		
 
@@ -357,13 +357,13 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_push_wrong_credentials(self, webserver, vt):

			




	
	
	
		
 
        dest_dir = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = webserver.repo_url(vt.repo_name)

			




	
	
	
		
 
        stdout, stderr = Command(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.repo_type, dest_dir,

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir,

			




	
	
	
		
 
                                             clone_url=clone_url, ignoreReturnCode=True)

			




	
	
	
		
 

			




	
	
	
		
 
        if vt.repo_type == 'git':

			




	
	
	
		
 
            assert 'fatal: Authentication failed' in stderr

			




	
	
	
		
 
        elif vt.repo_type == 'hg':

			




	
	
	
		
 
            assert 'abort: authorization failed' in stderr

			




	
	
		
 
@@ -371,13 +371,13 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    @parametrize_vcs_test

			




	
	
	
		
 
    def test_push_with_readonly_credentials(self, webserver, vt):

			




	
	
	
		
 
        dest_dir = _get_tmp_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, vt.repo_type, dest_dir, clone_url=clone_url, ignoreReturnCode=True)

			




	
	
	
		
 
        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

			




	
	
	
		
 
        elif vt.repo_type == 'hg':

			




	
	
	
		
 
            assert 'abort: HTTP Error 403: Forbidden' in stderr

			




	
	
	
		
 

			




	
	
		
 
@@ -385,13 +385,13 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_push_back_to_wrong_url(self, webserver, vt):

			




	
	
	
		
 
        dest_dir = _get_tmp_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, vt.repo_type, dest_dir, clone_url='http://%s:%s/tmp' % (

			




	
	
	
		
 
            webserver, vt, dest_dir, clone_url='http://%s:%s/tmp' % (

			




	
	
	
		
 
                webserver.server_address[0], webserver.server_address[1]),

			




	
	
	
		
 
            ignoreReturnCode=True)

			




	
	
	
		
 

			




	
	
	
		
 
        if vt.repo_type == 'git':

			




	
	
	
		
 
            assert 'not found' in stderr

			




	
	
	
		
 
        elif vt.repo_type == 'hg':

			




	
	
		
 
@@ -457,13 +457,13 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        # clone repo

			




	
	
	
		
 
        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(vt.repo_type, 'clone', clone_url, dest_dir)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, vt.repo_type, dest_dir, clone_url,

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, clone_url,

			




	
	
	
		
 
                                             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(vt.repo_type, dest_dir, clone_url)

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.