engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests: fixed test suite for celery adoption

Browse source
This commit is contained in:
RhodeCode Admin 2024-11-08 16:17:57 +01:00
parent 840634c16f
commit 1d5d2c0155
209 changed files with 5248 additions and 5275 deletions
Download patch file Download diff file Expand all files Collapse all files

6
rhodecode/apps/ssh_support/lib/backends/base.py
View file

@ -20,7 +20,7 @@ import os
import sys
import logging
from rhodecode.lib.hook_daemon.base import prepare_callback_daemon
from rhodecode.lib.hook_daemon.utils import prepare_callback_daemon
from rhodecode.lib.ext_json import sjson as json
from rhodecode.lib.vcs.conf import settings as vcs_settings
from rhodecode.lib.api_utils import call_service_api
@ -162,9 +162,7 @@ class SshVcsServer(object):
extras = {}
extras.update(tunnel_extras)
callback_daemon, extras = prepare_callback_daemon(
extras, protocol=self.hooks_protocol,
host=vcs_settings.HOOKS_HOST)
callback_daemon, extras = prepare_callback_daemon(extras, protocol=self.hooks_protocol)
with callback_daemon:
try:

43
rhodecode/apps/ssh_support/tests/test_server_git.py
View file

@ -33,19 +33,24 @@ class GitServerCreator(object):
'app:main': {
'ssh.executable.git': git_path,
'vcs.hooks.protocol.v2': 'celery',
'app.service_api.host': 'http://localhost',
'app.service_api.token': 'secret4',
'rhodecode.api.url': '/_admin/api',
}
}
repo_name = 'test_git'
repo_mode = 'receive-pack'
user = plain_dummy_user()
def __init__(self):
pass
def __init__(self, service_api_url, ini_file):
self.service_api_url = service_api_url
self.ini_file = ini_file
def create(self, **kwargs):
self.config_data['app:main']['app.service_api.host'] = self.service_api_url
parameters = {
'store': self.root,
'ini_path': '',
'ini_path': self.ini_file,
'user': self.user,
'repo_name': self.repo_name,
'repo_mode': self.repo_mode,
@ -60,12 +65,30 @@ class GitServerCreator(object):
return server
@pytest.fixture()
def git_server(app):
return GitServerCreator()
@pytest.fixture(scope='module')
def git_server(request, module_app, rhodecode_factory, available_port_factory):
ini_file = module_app._pyramid_settings['__file__']
vcsserver_host = module_app._pyramid_settings['vcs.server']
store_dir = os.path.dirname(ini_file)
# start rhodecode for service API
rc = rhodecode_factory(
request,
store_dir=store_dir,
port=available_port_factory(),
overrides=(
{'handler_console': {'level': 'DEBUG'}},
{'app:main': {'vcs.server': vcsserver_host}},
{'app:main': {'repo_store.path': store_dir}}
))
service_api_url = f'http://{rc.bind_addr}'
return GitServerCreator(service_api_url, ini_file)
class TestGitServer(object):
class TestGitServer:
def test_command(self, git_server):
server = git_server.create()
@ -102,14 +125,14 @@ class TestGitServer(object):
assert result is value
def test_run_returns_executes_command(self, git_server):
server = git_server.create()
from rhodecode.apps.ssh_support.lib.backends.git import GitTunnelWrapper
server = git_server.create()
os.environ['SSH_CLIENT'] = '127.0.0.1'
with mock.patch.object(GitTunnelWrapper, 'create_hooks_env') as _patch:
_patch.return_value = 0
with mock.patch.object(GitTunnelWrapper, 'command', return_value='date'):
exit_code = server.run()
exit_code = server.run(tunnel_extras={'config': server.ini_path})
assert exit_code == (0, False)
@ -135,7 +158,7 @@ class TestGitServer(object):
'action': action,
'ip': '10.10.10.10',
'locked_by': [None, None],
'config': '',
'config': git_server.ini_file,
'repo_store': store,
'server_url': None,
'hooks': ['push', 'pull'],

42
rhodecode/apps/ssh_support/tests/test_server_hg.py
View file

@ -17,6 +17,7 @@
# and proprietary license terms, please see https://rhodecode.com/licenses/
import os
import mock
import pytest
@ -32,22 +33,27 @@ class MercurialServerCreator(object):
'app:main': {
'ssh.executable.hg': hg_path,
'vcs.hooks.protocol.v2': 'celery',
'app.service_api.host': 'http://localhost',
'app.service_api.token': 'secret4',
'rhodecode.api.url': '/_admin/api',
}
}
repo_name = 'test_hg'
user = plain_dummy_user()
def __init__(self):
pass
def __init__(self, service_api_url, ini_file):
self.service_api_url = service_api_url
self.ini_file = ini_file
def create(self, **kwargs):
self.config_data['app:main']['app.service_api.host'] = self.service_api_url
parameters = {
'store': self.root,
'ini_path': '',
'ini_path': self.ini_file,
'user': self.user,
'repo_name': self.repo_name,
'user_permissions': {
'test_hg': 'repository.admin'
self.repo_name: 'repository.admin'
},
'settings': self.config_data['app:main'],
'env': plain_dummy_env()
@ -57,12 +63,30 @@ class MercurialServerCreator(object):
return server
@pytest.fixture()
def hg_server(app):
return MercurialServerCreator()
@pytest.fixture(scope='module')
def hg_server(request, module_app, rhodecode_factory, available_port_factory):
ini_file = module_app._pyramid_settings['__file__']
vcsserver_host = module_app._pyramid_settings['vcs.server']
store_dir = os.path.dirname(ini_file)
# start rhodecode for service API
rc = rhodecode_factory(
request,
store_dir=store_dir,
port=available_port_factory(),
overrides=(
{'handler_console': {'level': 'DEBUG'}},
{'app:main': {'vcs.server': vcsserver_host}},
{'app:main': {'repo_store.path': store_dir}}
))
service_api_url = f'http://{rc.bind_addr}'
return MercurialServerCreator(service_api_url, ini_file)
class TestMercurialServer(object):
class TestMercurialServer:
def test_command(self, hg_server, tmpdir):
server = hg_server.create()
@ -107,7 +131,7 @@ class TestMercurialServer(object):
with mock.patch.object(MercurialTunnelWrapper, 'create_hooks_env') as _patch:
_patch.return_value = 0
with mock.patch.object(MercurialTunnelWrapper, 'command', return_value='date'):
exit_code = server.run()
exit_code = server.run(tunnel_extras={'config': server.ini_path})
assert exit_code == (0, False)

58
rhodecode/apps/ssh_support/tests/test_server_svn.py
View file

@ -15,7 +15,9 @@
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import os
import mock
import pytest
@ -26,39 +28,62 @@ from rhodecode.apps.ssh_support.tests.conftest import plain_dummy_env, plain_dum
class SubversionServerCreator(object):
root = '/tmp/repo/path/'
svn_path = '/usr/local/bin/svnserve'
config_data = {
'app:main': {
'ssh.executable.svn': svn_path,
'vcs.hooks.protocol.v2': 'celery',
'app.service_api.host': 'http://localhost',
'app.service_api.token': 'secret4',
'rhodecode.api.url': '/_admin/api',
}
}
repo_name = 'test-svn'
user = plain_dummy_user()
def __init__(self):
pass
def __init__(self, service_api_url, ini_file):
self.service_api_url = service_api_url
self.ini_file = ini_file
def create(self, **kwargs):
self.config_data['app:main']['app.service_api.host'] = self.service_api_url
parameters = {
'store': self.root,
'repo_name': self.repo_name,
'ini_path': '',
'ini_path': self.ini_file,
'user': self.user,
'repo_name': self.repo_name,
'user_permissions': {
self.repo_name: 'repository.admin'
},
'settings': self.config_data['app:main'],
'env': plain_dummy_env()
}
parameters.update(kwargs)
server = SubversionServer(**parameters)
return server
@pytest.fixture()
def svn_server(app):
return SubversionServerCreator()
@pytest.fixture(scope='module')
def svn_server(request, module_app, rhodecode_factory, available_port_factory):
ini_file = module_app._pyramid_settings['__file__']
vcsserver_host = module_app._pyramid_settings['vcs.server']
store_dir = os.path.dirname(ini_file)
# start rhodecode for service API
rc = rhodecode_factory(
request,
store_dir=store_dir,
port=available_port_factory(),
overrides=(
{'handler_console': {'level': 'DEBUG'}},
{'app:main': {'vcs.server': vcsserver_host}},
{'app:main': {'repo_store.path': store_dir}}
))
service_api_url = f'http://{rc.bind_addr}'
return SubversionServerCreator(service_api_url, ini_file)
class TestSubversionServer(object):
@ -168,8 +193,9 @@ class TestSubversionServer(object):
assert repo_name == expected_match
def test_run_returns_executes_command(self, svn_server):
server = svn_server.create()
from rhodecode.apps.ssh_support.lib.backends.svn import SubversionTunnelWrapper
server = svn_server.create()
os.environ['SSH_CLIENT'] = '127.0.0.1'
with mock.patch.object(
SubversionTunnelWrapper, 'get_first_client_response',
@ -184,20 +210,18 @@ class TestSubversionServer(object):
SubversionTunnelWrapper, 'command',
return_value=['date']):
exit_code = server.run()
exit_code = server.run(tunnel_extras={'config': server.ini_path})
# SVN has this differently configured, and we get in our mock env
# None as return code
assert exit_code == (None, False)
def test_run_returns_executes_command_that_cannot_extract_repo_name(self, svn_server):
server = svn_server.create()
from rhodecode.apps.ssh_support.lib.backends.svn import SubversionTunnelWrapper
with mock.patch.object(
SubversionTunnelWrapper, 'command',
return_value=['date']):
with mock.patch.object(
SubversionTunnelWrapper, 'get_first_client_response',
server = svn_server.create()
with mock.patch.object(SubversionTunnelWrapper, 'command', return_value=['date']):
with mock.patch.object(SubversionTunnelWrapper, 'get_first_client_response',
return_value=None):
exit_code = server.run()
exit_code = server.run(tunnel_extras={'config': server.ini_path})
assert exit_code == (1, False)