Changeset - 8f3cf5d00d7f
Parent rev.
Child rev.
[Not reviewed]
default
0 1 0
Mads Kiilerich - 6 years ago 2019-07-31 03:56:57
mads@kiilerich.com
tests: run vcs tests on ssh too

Error reporting when using SSH is different from when using HTTP. SSH doesn't
use HTTP error codes but can write anything to stdout/stderr. Some tests thus
need to have their expectations tweaked.

Inspired by work by Dominik Ruf.
1 file changed with 55 insertions and 8 deletions:
kallithea/tests/other/test_vcs_operations.py
55
8
0 comments (0 inline, 0 general)
kallithea/tests/other/test_vcs_operations.py
➞
Show inline comments
 
@@ -39,28 +39,48 @@ import pytest
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation, Ui, UserLog
 
from kallithea.model.meta import Session
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.user import UserModel
 
from kallithea.model.ssh_key import SshKeyModel
 
from kallithea import CONFIG
 

			




	
	
	
		
 
DEBUG = True

			




	
	
	
		
 
HOST = '127.0.0.1:4999'  # test host

			




	
	
	
		
 

			




	
	
	
		
 
fixture = Fixture()

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
# Parameterize different kinds of VCS testing - both the kind of VCS and the

			




	
	
	
		
 
# access method (HTTP/SSH)

			




	
	
	
		
 

			




	
	
	
		
 
# Mixin for using HTTP URLs

			




	
	
	
		
 
# Mixin for using HTTP and SSH URLs

			




	
	
	
		
 
class HttpVcsTest(object):

			




	
	
	
		
 
    @staticmethod

			




	
	
	
		
 
    def repo_url_param(webserver, repo_name, **kwargs):

			




	
	
	
		
 
        return webserver.repo_url(repo_name, **kwargs)

			




	
	
	
		
 

			




	
	
	
		
 
class SshVcsTest(object):

			




	
	
	
		
 
    public_keys = {

			




	
	
	
		
 
        TEST_USER_REGULAR_LOGIN: u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== kallithea@localhost',

			




	
	
	
		
 
        TEST_USER_ADMIN_LOGIN: u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUq== kallithea@localhost',

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def repo_url_param(cls, webserver, repo_name, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS, client_ip=IP_ADDR):

			




	
	
	
		
 
        user = User.get_by_username(username)

			




	
	
	
		
 
        if user.ssh_keys:

			




	
	
	
		
 
            ssh_key = user.ssh_keys[0]

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            sshkeymodel = SshKeyModel()

			




	
	
	
		
 
            ssh_key = sshkeymodel.create(user, u'test key', cls.public_keys[user.username])

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        return cls._ssh_param(repo_name, user, ssh_key, client_ip)

			




	
	
	
		
 

			




	
	
	
		
 
# Mixins for using Mercurial and Git

			




	
	
	
		
 
class HgVcsTest(object):

			




	
	
	
		
 
    repo_type = 'hg'

			




	
	
	
		
 
    repo_name = HG_REPO

			




	
	
	
		
 

			




	
	
	
		
 
class GitVcsTest(object):

			




	
	
		
 
@@ -71,18 +91,43 @@ class GitVcsTest(object):

			




	
	
	
		
 
class HgHttpVcsTest(HgVcsTest, HttpVcsTest):

			




	
	
	
		
 
    pass

			




	
	
	
		
 

			




	
	
	
		
 
class GitHttpVcsTest(GitVcsTest, HttpVcsTest):

			




	
	
	
		
 
    pass

			




	
	
	
		
 

			




	
	
	
		
 
class HgSshVcsTest(HgVcsTest, SshVcsTest):

			




	
	
	
		
 
    @staticmethod

			




	
	
	
		
 
    def _ssh_param(repo_name, user, ssh_key, client_ip):

			




	
	
	
		
 
        # Specify a custom ssh command on the command line

			




	
	
	
		
 
        return r"""--config ui.ssh="bash -c 'SSH_ORIGINAL_COMMAND=\"\$2\" SSH_CONNECTION=\"%s 1024 127.0.0.1 22\" kallithea-cli ssh-serve -c %s %s %s' --" ssh://someuser@somehost/%s""" % (

			




	
	
	
		
 
            client_ip,

			




	
	
	
		
 
            CONFIG['__file__'],

			




	
	
	
		
 
            user.user_id,

			




	
	
	
		
 
            ssh_key.user_ssh_key_id,

			




	
	
	
		
 
            repo_name)

			




	
	
	
		
 

			




	
	
	
		
 
class GitSshVcsTest(GitVcsTest, SshVcsTest):

			




	
	
	
		
 
    @staticmethod

			




	
	
	
		
 
    def _ssh_param(repo_name, user, ssh_key, client_ip):

			




	
	
	
		
 
        # Set a custom ssh command in the global environment

			




	
	
	
		
 
        os.environ['GIT_SSH_COMMAND'] = r"""bash -c 'SSH_ORIGINAL_COMMAND="$2" SSH_CONNECTION="%s 1024 127.0.0.1 22" kallithea-cli ssh-serve -c %s %s %s' --""" % (

			




	
	
	
		
 
            client_ip,

			




	
	
	
		
 
            CONFIG['__file__'],

			




	
	
	
		
 
            user.user_id,

			




	
	
	
		
 
            ssh_key.user_ssh_key_id)

			




	
	
	
		
 
        return "ssh://someuser@somehost/%s""" % repo_name

			




	
	
	
		
 

			




	
	
	
		
 
parametrize_vcs_test = parametrize('vt', [

			




	
	
	
		
 
    HgHttpVcsTest,

			




	
	
	
		
 
    GitHttpVcsTest,

			




	
	
	
		
 
    HgSshVcsTest,

			




	
	
	
		
 
    GitSshVcsTest,

			




	
	
	
		
 
])

			




	
	
	
		
 
parametrize_vcs_test_hg = parametrize('vt', [

			




	
	
	
		
 
    HgHttpVcsTest,

			




	
	
	
		
 
    HgSshVcsTest,

			




	
	
	
		
 
])

			




	
	
	
		
 
parametrize_vcs_test_http = parametrize('vt', [

			




	
	
	
		
 
    HgHttpVcsTest,

			




	
	
	
		
 
    GitHttpVcsTest,

			




	
	
	
		
 
])

			




	
	
	
		
 

			




	
	
		
 
@@ -245,27 +290,27 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        elif vt.repo_type == 'hg':

			




	
	
	
		
 
            assert 'abort: authorization failed' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_git_dir_as_hg(self, webserver):

			




	
	
	
		
 
        clone_url = HgHttpVcsTest.repo_url_param(webserver, GIT_REPO)

			




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

			




	
	
	
		
 
        assert 'HTTP Error 404: Not Found' in stderr

			




	
	
	
		
 
        assert 'HTTP Error 404: Not Found' in stderr or "not a valid repository" in stdout and 'abort:' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_hg_repo_as_git(self, webserver):

			




	
	
	
		
 
        clone_url = GitHttpVcsTest.repo_url_param(webserver, HG_REPO)

			




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

			




	
	
	
		
 
        assert 'not found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    @parametrize_vcs_test

			




	
	
	
		
 
    def test_clone_non_existing_path(self, webserver, vt):

			




	
	
	
		
 
        clone_url = vt.repo_url_param(webserver, 'trololo')

			




	
	
	
		
 
        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

			




	
	
	
		
 
            assert 'not found' in stderr or 'abort: Access to %r denied' % 'trololo' in stderr

			




	
	
	
		
 
        elif vt.repo_type == 'hg':

			




	
	
	
		
 
            assert 'HTTP Error 404: Not Found' in stderr

			




	
	
	
		
 
            assert 'HTTP Error 404: Not Found' in stderr or 'abort: no suitable response from remote hg' in stderr and 'remote: abort: Access to %r denied' % 'trololo' in stdout

			




	
	
	
		
 

			




	
	
	
		
 
    @parametrize_vcs_test

			




	
	
	
		
 
    def test_push_new_repo(self, webserver, vt):

			




	
	
	
		
 
        # Clear the log so we know what is added

			




	
	
	
		
 
        UserLog.query().delete()

			




	
	
	
		
 
        Session().commit()

			




	
	
		
 
@@ -420,15 +465,15 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        clone_url = vt.repo_url_param(webserver, 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, dest_dir, ignoreReturnCode=True, clone_url=clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        if vt.repo_type == 'git':

			




	
	
	
		
 
            assert 'The requested URL returned error: 403' in stderr

			




	
	
	
		
 
            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

			




	
	
	
		
 
            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

			




	
	
	
		
 

			




	
	
	
		
 
        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] == \

			




	
	
	
		
 
            [(u'pull', 0)]

			




	
	
	
		
 

			




	
	
	
		
 
    @parametrize_vcs_test

			




	
	
		
 
@@ -457,15 +502,15 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
            # IP permissions are cached, need to wait for the cache in the server process to expire

			




	
	
	
		
 
            time.sleep(1.5)

			




	
	
	
		
 
            clone_url = vt.repo_url_param(webserver, 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)

			




	
	
	
		
 
                assert re.search(r'\b403\b', stderr) or 'abort: User test_admin from 127.0.0.127 cannot be authorized' in stderr

			




	
	
	
		
 
            elif vt.repo_type == 'hg':

			




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

			




	
	
	
		
 
                assert 'abort: HTTP Error 403: Forbidden' in stderr or 'remote: abort: User test_admin from 127.0.0.127 cannot be authorized' in stdout

			




	
	
	
		
 
        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

			




	
	
		
 
@@ -524,12 +569,14 @@ class TestVCSOperations(TestController):

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        # re-try to push

			




	
	
	
		
 
        stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True)

			




	
	
	
		
 
        if vt is HgHttpVcsTest:

			




	
	
	
		
 
            # like with 'hg serve...' 'HTTP Error 500: INTERNAL SERVER ERROR' should be returned

			




	
	
	
		
 
            assert 'HTTP Error 500: INTERNAL SERVER ERROR' in stderr

			




	
	
	
		
 
        elif vt is HgSshVcsTest:

			




	
	
	
		
 
            assert 'remote: Exception: exception_test_hook threw an exception' in stdout

			




	
	
	
		
 
        else: assert False

			




	
	
	
		
 
        # there are still outgoing changesets

			




	
	
	
		
 
        stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)

			




	
	
	
		
 
        assert stdout != ''

			




	
	
	
		
 

			




	
	
	
		
 
        # set prechangegroup hook to method that returns False

			




        

    



    


    



    



      

      





    
    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.