Changeset - fa5788feed0f
Parent rev.
Child rev.
[Not reviewed]
default
0 1 0
domruf - 9 years ago 2017-04-27 20:46:16
dominikruf@gmail.com
tests: check returncode from executing external commands

If a command fails, it could lead to undefined behavior,
which lets tests fail without clear indication of the cause of the failures.
Therefore, if not stated otherwise, check the return code.
1 file changed with 26 insertions and 19 deletions:
kallithea/tests/other/manual_test_vcs_operations.py
26
19
0 comments (0 inline, 0 general)
kallithea/tests/other/manual_test_vcs_operations.py
➞
Show inline comments
 
@@ -62,12 +62,13 @@ class Command(object):
 
    def execute(self, cmd, *args, **environ):
 
        """
 
        Runs command on the system with given ``args``.
 
        """
 

			




	
	
	
		
 
        command = cmd + ' ' + ' '.join(args)

			




	
	
	
		
 
        ignoreReturnCode = environ.pop('ignoreReturnCode', False)

			




	
	
	
		
 
        if DEBUG:

			




	
	
	
		
 
            print '*** CMD %s ***' % command

			




	
	
	
		
 
        testenv = dict(os.environ)

			




	
	
	
		
 
        testenv['LANG'] = 'en_US.UTF-8'

			




	
	
	
		
 
        testenv['LANGUAGE'] = 'en_US:en'

			




	
	
	
		
 
        testenv.update(environ)

			




	
	
		
 
@@ -75,12 +76,14 @@ class Command(object):

			




	
	
	
		
 
        stdout, stderr = p.communicate()

			




	
	
	
		
 
        if DEBUG:

			




	
	
	
		
 
            if stdout:

			




	
	
	
		
 
                print 'stdout:', repr(stdout)

			




	
	
	
		
 
            if stderr:

			




	
	
	
		
 
                print 'stderr:', repr(stderr)

			




	
	
	
		
 
        if not ignoreReturnCode:

			




	
	
	
		
 
            assert p.returncode == 0

			




	
	
	
		
 
        return stdout, stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _get_tmp_dir():

			




	
	
	
		
 
    return tempfile.mkdtemp(prefix='rc_integration_test')

			




	
	
	
		
 

			




	
	
		
 
@@ -99,13 +102,13 @@ def _construct_url(repo, **kwargs):

			




	
	
	
		
 
        _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s' % params

			




	
	
	
		
 
    else:

			




	
	
	
		
 
        _url = 'http://(host)s/%(cloned_repo)s' % params

			




	
	
	
		
 
    return _url

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _add_files_and_push(vcs, DEST, **kwargs):

			




	
	
	
		
 
def _add_files_and_push(vcs, DEST, ignoreReturnCode=False, **kwargs):

			




	
	
	
		
 
    """

			




	
	
	
		
 
    Generate some files, add it to DEST repo and push back

			




	
	
	
		
 
    vcs is git or hg and defines what VCS we want to make those files for

			




	
	
	
		
 

			




	
	
	
		
 
    :param vcs:

			




	
	
	
		
 
    :param DEST:

			




	
	
		
 
@@ -145,15 +148,15 @@ def _add_files_and_push(vcs, DEST, **kwa

			




	
	
	
		
 

			




	
	
	
		
 
    clone_url = _construct_url(_REPO, **kwargs)

			




	
	
	
		
 
    if 'clone_url' in kwargs:

			




	
	
	
		
 
        clone_url = kwargs['clone_url']

			




	
	
	
		
 
    stdout = stderr = None

			




	
	
	
		
 
    if vcs == 'hg':

			




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

			




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

			




	
	
	
		
 
    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", ignoreReturnCode=ignoreReturnCode)

			




	
	
	
		
 

			




	
	
	
		
 
    return stdout, stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def set_anonymous_access(enable=True):

			




	
	
	
		
 
    user = User.get_default_user()

			




	
	
		
 
@@ -213,38 +216,38 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
        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, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        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, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        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, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        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, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        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, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        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, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        assert 'not found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_new_file_hg(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

			




	
	
		
 
@@ -322,43 +325,45 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_push_wrong_credentials_hg(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




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

			




	
	
	
		
 
                                             passwd='name', ignoreReturnCode=True)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'abort: authorization failed' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_wrong_credentials_git(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




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

			




	
	
	
		
 
                                             passwd='name', ignoreReturnCode=True)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'fatal: Authentication failed' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_back_to_wrong_url_hg(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        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)

			




	
	
	
		
 
                                    clone_url='http://%s/tmp' % HOST,

			




	
	
	
		
 
                                    ignoreReturnCode = True)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'HTTP Error 404: Not Found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_back_to_wrong_url_git(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        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)

			




	
	
	
		
 
                                    clone_url='http://%s/tmp' % HOST,

			




	
	
	
		
 
                                    ignoreReturnCode = True)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'not found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_and_create_lock_hg(self):

			




	
	
	
		
 
        # enable locking

			




	
	
	
		
 
        r = Repository.get_by_repo_name(HG_REPO)

			




	
	
		
 
@@ -388,24 +393,24 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_clone_after_repo_was_locked_hg(self):

			




	
	
	
		
 
        #lock repo

			




	
	
	
		
 
        r = Repository.get_by_repo_name(HG_REPO)

			




	
	
	
		
 
        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, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""

			




	
	
	
		
 
                % (HG_REPO, TEST_USER_ADMIN_LOGIN))

			




	
	
	
		
 
        assert msg in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_after_repo_was_locked_git(self):

			




	
	
	
		
 
        #lock repo

			




	
	
	
		
 
        r = Repository.get_by_repo_name(GIT_REPO)

			




	
	
	
		
 
        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, _get_tmp_dir())

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




	
	
	
		
 
        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()

			




	
	
		
 
@@ -420,13 +425,14 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

			




	
	
	
		
 

			




	
	
	
		
 
        #push fails repo is locked by other user !

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('hg', DEST,

			




	
	
	
		
 
                                             user=TEST_USER_REGULAR_LOGIN,

			




	
	
	
		
 
                                             passwd=TEST_USER_REGULAR_PASS)

			




	
	
	
		
 
                                             passwd=TEST_USER_REGULAR_PASS,

			




	
	
	
		
 
                                             ignoreReturnCode=True)

			




	
	
	
		
 
        msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""

			




	
	
	
		
 
                % (HG_REPO, TEST_USER_ADMIN_LOGIN))

			




	
	
	
		
 
        assert msg in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_on_locked_repo_by_other_user_git(self):

			




	
	
	
		
 
        #clone some temp

			




	
	
		
 
@@ -442,13 +448,14 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

			




	
	
	
		
 

			




	
	
	
		
 
        #push fails repo is locked by other user !

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('git', DEST,

			




	
	
	
		
 
                                             user=TEST_USER_REGULAR_LOGIN,

			




	
	
	
		
 
                                             passwd=TEST_USER_REGULAR_PASS)

			




	
	
	
		
 
                                             passwd=TEST_USER_REGULAR_PASS,

			




	
	
	
		
 
                                             ignoreReturnCode=True)

			




	
	
	
		
 
        err = 'Repository `%s` locked by user `%s`' % (GIT_REPO, TEST_USER_ADMIN_LOGIN)

			




	
	
	
		
 
        assert err in stderr

			




	
	
	
		
 

			




	
	
	
		
 
        #TODO: fix this somehow later on Git, Git is stupid and even if we throw

			




	
	
	
		
 
        #back 423 to it, it makes ANOTHER request and we fail there with 405 :/

			




	
	
	
		
 

			




	
	
		
 
@@ -512,13 +519,13 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_ip_restriction_hg(self):

			




	
	
	
		
 
        user_model = UserModel()

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            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, _get_tmp_dir())

			




	
	
	
		
 
            stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

			




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

			




	
	
	
		
 
        finally:

			




	
	
	
		
 
            #release IP restrictions

			




	
	
	
		
 
            for ip in UserIpMap.query():

			




	
	
	
		
 
                UserIpMap.delete(ip.ip_id)

			




	
	
	
		
 
            Session().commit()

			




	
	
		
 
@@ -539,13 +546,13 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
    def test_ip_restriction_git(self):

			




	
	
	
		
 
        user_model = UserModel()

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            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, _get_tmp_dir())

			




	
	
	
		
 
            stdout, stderr = Command(tempfile.gettempdir()).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)

			




        

    



    


    



    



      

      





    
    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.