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

tests: rewrote code for running vcs_operations. Now it starts it's own

Browse source 
configurable instances for exactly those tests.
This commit is contained in:
Marcin Kuzminski 2017-12-18 21:19:29 +01:00
parent 6ba1fb962c
commit 4f426c33f5
13 changed files with 325 additions and 327 deletions
Download patch file Download diff file Expand all files Collapse all files

1
rhodecode/tests/lib/middleware/utils/test_scm_app_http.py
View file

@ -43,7 +43,6 @@ def vcsserver_http_echo_app(request, vcsserver_factory):
"""
vcsserver = vcsserver_factory(
request=request,
use_http=True,
overrides=[{'app:main': {'dev.use_echo_app': 'true'}}])
return vcsserver

161
rhodecode/tests/pylons_plugin.py
View file

@ -18,25 +18,15 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import os
import json
import time
import platform
import socket
import tempfile
import subprocess32
from urllib2 import urlopen, URLError
import configobj
import pytest
from rhodecode.lib.pyramid_utils import get_app_config
from rhodecode.tests.fixture import TestINI
from rhodecode.tests.vcs_operations.conftest import get_host_url, get_port
VCSSERVER_LOG = os.path.join(tempfile.gettempdir(), 'rc-vcsserver.log')
from rhodecode.tests.server_utils import RcVCSServer
def _parse_json(value):
@ -94,17 +84,8 @@ def pytest_addoption(parser):
"Start the VCSServer with HTTP protocol support.")
@pytest.fixture(scope="session")
def vcs_server_config_override(request):
"""
Allows injecting the overrides by specifying this inside test class
"""
return ()
@pytest.fixture(scope='session')
def vcsserver(request, vcsserver_port, vcsserver_factory, vcs_server_config_override):
def vcsserver(request, vcsserver_port, vcsserver_factory):
"""
Session scope VCSServer.
@ -125,10 +106,8 @@ def vcsserver(request, vcsserver_port, vcsserver_factory, vcs_server_config_over
if not request.config.getoption('with_vcsserver'):
return None
use_http = _use_vcs_http_server(request.config)
return vcsserver_factory(
request, use_http=use_http, vcsserver_port=vcsserver_port,
overrides=vcs_server_config_override)
request, vcsserver_port=vcsserver_port)
@pytest.fixture(scope='session')
@ -137,23 +116,21 @@ def vcsserver_factory(tmpdir_factory):
Use this if you need a running vcsserver with a special configuration.
"""
def factory(request, use_http=True, overrides=(), vcsserver_port=None):
def factory(request, overrides=(), vcsserver_port=None,
log_file=None):
if vcsserver_port is None:
vcsserver_port = get_available_port()
overrides = list(overrides)
if use_http:
overrides.append({'server:main': {'port': vcsserver_port}})
else:
overrides.append({'DEFAULT': {'port': vcsserver_port}})
overrides.append({'server:main': {'port': vcsserver_port}})
if is_cygwin():
platform_override = {'DEFAULT': {
'beaker.cache.repo_object.type': 'nocache'}}
overrides.append(platform_override)
option_name = 'vcsserver_config_http' if use_http else ''
option_name = 'vcsserver_config_http'
override_option_name = 'vcsserver_config_override'
config_file = get_config(
request.config, option_name=option_name,
@ -161,9 +138,7 @@ def vcsserver_factory(tmpdir_factory):
basetemp=tmpdir_factory.getbasetemp().strpath,
prefix='test_vcs_')
print("Using the VCSServer configuration:{}".format(config_file))
ServerClass = HttpVCSServer if use_http else None
server = ServerClass(config_file)
server = RcVCSServer(config_file, log_file)
server.start()
@request.addfinalizer
@ -180,93 +155,11 @@ def is_cygwin():
return 'cygwin' in platform.system().lower()
def _use_vcs_http_server(config):
protocol_option = 'vcsserver_protocol'
protocol = (
config.getoption(protocol_option) or
config.getini(protocol_option) or
'http')
return protocol == 'http'
def _use_log_level(config):
level = config.getoption('test_loglevel') or 'warn'
return level.upper()
class VCSServer(object):
"""
Represents a running VCSServer instance.
"""
_args = []
def start(self):
print("Starting the VCSServer: {}".format(self._args))
self.process = subprocess32.Popen(self._args)
def wait_until_ready(self, timeout=30):
raise NotImplementedError()
def shutdown(self):
self.process.kill()
class HttpVCSServer(VCSServer):
"""
Represents a running VCSServer instance.
"""
def __init__(self, config_file):
self.config_file = config_file
config_data = configobj.ConfigObj(config_file)
self._config = config_data['server:main']
args = ['gunicorn', '--paste', config_file]
self._args = args
@property
def http_url(self):
template = 'http://{host}:{port}/'
return template.format(**self._config)
def start(self):
env = os.environ.copy()
host_url = 'http://' + get_host_url(self.config_file)
rc_log = list(VCSSERVER_LOG.partition('.log'))
rc_log.insert(1, get_port(self.config_file))
rc_log = ''.join(rc_log)
server_out = open(rc_log, 'w')
command = ' '.join(self._args)
print('rhodecode-vcsserver starting at: {}'.format(host_url))
print('rhodecode-vcsserver command: {}'.format(command))
print('rhodecode-vcsserver logfile: {}'.format(rc_log))
self.process = subprocess32.Popen(
self._args, bufsize=0, env=env, stdout=server_out, stderr=server_out)
def wait_until_ready(self, timeout=30):
host = self._config['host']
port = self._config['port']
status_url = 'http://{host}:{port}/status'.format(host=host, port=port)
start = time.time()
while time.time() - start < timeout:
try:
urlopen(status_url)
break
except URLError:
time.sleep(0.2)
else:
pytest.exit(
"Starting the VCSServer failed or took more than {} "
"seconds. cmd: `{}`".format(timeout, ' '.join(self._args)))
def shutdown(self):
self.process.kill()
@pytest.fixture(scope='session')
def ini_config(request, tmpdir_factory, rcserver_port, vcsserver_port):
option_name = 'pyramid_config'
@ -280,6 +173,10 @@ def ini_config(request, tmpdir_factory, rcserver_port, vcsserver_port):
# fixtures of the test cases. For the test run it must always be
# off in the INI file.
'vcs.start_server': 'false',
'vcs.server.protocol': 'http',
'vcs.scm_app_implementation': 'http',
'vcs.hooks.protocol': 'http',
}},
{'handler_console': {
@ -289,14 +186,6 @@ def ini_config(request, tmpdir_factory, rcserver_port, vcsserver_port):
}},
]
if _use_vcs_http_server(request.config):
overrides.append({
'app:main': {
'vcs.server.protocol': 'http',
'vcs.scm_app_implementation': 'http',
'vcs.hooks.protocol': 'http',
}
})
filename = get_config(
request.config, option_name=option_name,
@ -313,6 +202,19 @@ def ini_settings(ini_config):
return get_app_config(ini_path)
def get_available_port():
family = socket.AF_INET
socktype = socket.SOCK_STREAM
host = '127.0.0.1'
mysocket = socket.socket(family, socktype)
mysocket.bind((host, 0))
port = mysocket.getsockname()[1]
mysocket.close()
del mysocket
return port
@pytest.fixture(scope='session')
def rcserver_port(request):
port = get_available_port()
@ -329,19 +231,6 @@ def vcsserver_port(request):
return port
def get_available_port():
family = socket.AF_INET
socktype = socket.SOCK_STREAM
host = '127.0.0.1'
mysocket = socket.socket(family, socktype)
mysocket.bind((host, 0))
port = mysocket.getsockname()[1]
mysocket.close()
del mysocket
return port
@pytest.fixture(scope='session')
def available_port_factory():
"""

194
rhodecode/tests/server_utils.py Normal file
View file

@ -0,0 +1,194 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2017 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# 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 time
import tempfile
import pytest
import subprocess32
import configobj
from urllib2 import urlopen, URLError
from pyramid.compat import configparser
from rhodecode.tests import TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS
from rhodecode.tests.utils import is_url_reachable
def get_port(pyramid_config):
config = configparser.ConfigParser()
config.read(pyramid_config)
return config.get('server:main', 'port')
def get_host_url(pyramid_config):
"""Construct the host url using the port in the test configuration."""
return '127.0.0.1:%s' % get_port(pyramid_config)
def assert_no_running_instance(url):
if is_url_reachable(url):
print("Hint: Usually this means another instance of server "
"is running in the background at %s." % url)
pytest.fail(
"Port is not free at %s, cannot start server at" % url)
class ServerBase(object):
_args = []
log_file_name = 'NOT_DEFINED.log'
status_url_tmpl = 'http://{host}:{port}'
def __init__(self, config_file, log_file):
self.config_file = config_file
config_data = configobj.ConfigObj(config_file)
self._config = config_data['server:main']
self._args = []
self.log_file = log_file or os.path.join(
tempfile.gettempdir(), self.log_file_name)
self.process = None
self.server_out = None
print("Using the {} configuration:{}".format(
self.__class__.__name__, config_file))
if not os.path.isfile(config_file):
raise RuntimeError('Failed to get config at {}'.format(config_file))
@property
def command(self):
return ' '.join(self._args)
@property
def http_url(self):
template = 'http://{host}:{port}/'
return template.format(**self._config)
def host_url(self):
return 'http://' + get_host_url(self.config_file)
def get_rc_log(self):
with open(self.log_file) as f:
return f.read()
def wait_until_ready(self, timeout=15):
host = self._config['host']
port = self._config['port']
status_url = self.status_url_tmpl.format(host=host, port=port)
start = time.time()
while time.time() - start < timeout:
try:
urlopen(status_url)
break
except URLError:
time.sleep(0.2)
else:
pytest.exit(
"Starting the {} failed or took more than {} "
"seconds. cmd: `{}`".format(
self.__class__.__name__, timeout, self.command))
def shutdown(self):
self.process.kill()
self.server_out.flush()
self.server_out.close()
def get_log_file_with_port(self):
log_file = list(self.log_file.partition('.log'))
log_file.insert(1, get_port(self.config_file))
log_file = ''.join(log_file)
return log_file
class RcVCSServer(ServerBase):
"""
Represents a running VCSServer instance.
"""
log_file_name = 'rc-vcsserver.log'
status_url_tmpl = 'http://{host}:{port}/status'
def __init__(self, config_file, log_file=None):
super(RcVCSServer, self).__init__(config_file, log_file)
self._args = [
'gunicorn', '--paste', self.config_file]
def start(self):
env = os.environ.copy()
self.log_file = self.get_log_file_with_port()
self.server_out = open(self.log_file, 'w')
host_url = self.host_url()
assert_no_running_instance(host_url)
print('rhodecode-vcsserver starting at: {}'.format(host_url))
print('rhodecode-vcsserver command: {}'.format(self.command))
print('rhodecode-vcsserver logfile: {}'.format(self.log_file))
self.process = subprocess32.Popen(
self._args, bufsize=0, env=env,
stdout=self.server_out, stderr=self.server_out)
class RcWebServer(ServerBase):
"""
Represents a running RCE web server used as a test fixture.
"""
log_file_name = 'rc-web.log'
status_url_tmpl = 'http://{host}:{port}/_admin/ops/ping'
def __init__(self, config_file, log_file=None):
super(RcWebServer, self).__init__(config_file, log_file)
self._args = [
'gunicorn', '--worker-class', 'gevent', '--paste', config_file]
def start(self):
env = os.environ.copy()
env['RC_NO_TMP_PATH'] = '1'
self.log_file = self.get_log_file_with_port()
self.server_out = open(self.log_file, 'w')
host_url = self.host_url()
assert_no_running_instance(host_url)
print('rhodecode-web starting at: {}'.format(host_url))
print('rhodecode-web command: {}'.format(self.command))
print('rhodecode-web logfile: {}'.format(self.log_file))
self.process = subprocess32.Popen(
self._args, bufsize=0, env=env,
stdout=self.server_out, stderr=self.server_out)
def repo_clone_url(self, repo_name, **kwargs):
params = {
'user': TEST_USER_ADMIN_LOGIN,
'passwd': TEST_USER_ADMIN_PASS,
'host': get_host_url(self.config_file),
'cloned_repo': repo_name,
}
params.update(**kwargs)
_url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s' % params
return _url

5
rhodecode/tests/vcs/conftest.py
View file

@ -31,11 +31,6 @@ from rhodecode.tests import get_new_dir
from rhodecode.tests.utils import check_skip_backends, check_xfail_backends
@pytest.fixture(scope="session")
def vcs_server_config_override():
return ({'server:main': {'workers': 1}},)
@pytest.fixture()
def vcs_repository_support(
request, backend_alias, baseapp, _vcs_repo_container):

174
rhodecode/tests/vcs_operations/conftest.py
View file

@ -27,76 +27,30 @@ py.test config for test suite for making push/pull operations.
to redirect things to stderr instead of stdout.
"""
import ConfigParser
import os
import subprocess32
import tempfile
import textwrap
import pytest
import rhodecode
from rhodecode import events
from rhodecode.model.db import Integration
from rhodecode.model.integration import IntegrationModel
from rhodecode.model.db import Repository
from rhodecode.model.meta import Session
from rhodecode.model.settings import SettingsModel
from rhodecode.tests import (
GIT_REPO, HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,)
from rhodecode.tests.fixture import Fixture
from rhodecode.tests.utils import is_url_reachable, wait_for_url
from rhodecode.integrations.types.webhook import WebhookIntegrationType
from rhodecode.tests import GIT_REPO, HG_REPO
from rhodecode.tests.fixture import Fixture
from rhodecode.tests.server_utils import RcWebServer
RC_LOG = os.path.join(tempfile.gettempdir(), 'rc.log')
REPO_GROUP = 'a_repo_group'
HG_REPO_WITH_GROUP = '%s/%s' % (REPO_GROUP, HG_REPO)
GIT_REPO_WITH_GROUP = '%s/%s' % (REPO_GROUP, GIT_REPO)
def assert_no_running_instance(url):
if is_url_reachable(url):
print("Hint: Usually this means another instance of Enterprise "
"is running in the background.")
pytest.fail(
"Port is not free at %s, cannot start web interface" % url)
def get_port(pyramid_config):
config = ConfigParser.ConfigParser()
config.read(pyramid_config)
return config.get('server:main', 'port')
def get_host_url(pyramid_config):
"""Construct the host url using the port in the test configuration."""
return '127.0.0.1:%s' % get_port(pyramid_config)
class RcWebServer(object):
"""
Represents a running RCE web server used as a test fixture.
"""
def __init__(self, pyramid_config, log_file):
self.pyramid_config = pyramid_config
self.log_file = log_file
def repo_clone_url(self, repo_name, **kwargs):
params = {
'user': TEST_USER_ADMIN_LOGIN,
'passwd': TEST_USER_ADMIN_PASS,
'host': get_host_url(self.pyramid_config),
'cloned_repo': repo_name,
}
params.update(**kwargs)
_url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s' % params
return _url
def host_url(self):
return 'http://' + get_host_url(self.pyramid_config)
def get_rc_log(self):
with open(self.log_file) as f:
return f.read()
@pytest.fixture(scope="module")
def rcextensions(request, baseapp, tmpdir_factory):
def rcextensions(request, db_connection, tmpdir_factory):
"""
Installs a testing rcextensions pack to ensure they work as expected.
"""
@ -122,7 +76,7 @@ def rcextensions(request, baseapp, tmpdir_factory):
@pytest.fixture(scope="module")
def repos(request, baseapp):
def repos(request, db_connection):
"""Create a copy of each test repo in a repo group."""
fixture = Fixture()
repo_group = fixture.create_repo_group(REPO_GROUP)
@ -141,62 +95,56 @@ def repos(request, baseapp):
fixture.destroy_repo_group(repo_group_id)
@pytest.fixture(scope="session")
def vcs_server_config_override():
return ({'server:main': {'workers': 2}},)
@pytest.fixture(scope="module")
def rc_web_server_config_modification():
return []
@pytest.fixture(scope="module")
def rc_web_server_config(testini_factory):
def rc_web_server_config_factory(testini_factory, rc_web_server_config_modification):
"""
Configuration file used for the fixture `rc_web_server`.
"""
CUSTOM_PARAMS = [
{'handler_console': {'level': 'DEBUG'}},
]
return testini_factory(CUSTOM_PARAMS)
def factory(vcsserver_port):
custom_params = [
{'handler_console': {'level': 'DEBUG'}},
{'app:main': {'vcs.server': 'localhost:%s' % vcsserver_port}}
]
custom_params.extend(rc_web_server_config_modification)
return testini_factory(custom_params)
return factory
@pytest.fixture(scope="module")
def rc_web_server(
request, baseapp, rc_web_server_config, repos, rcextensions):
request, vcsserver_factory, available_port_factory,
rc_web_server_config_factory, repos, rcextensions):
"""
Run the web server as a subprocess.
Since we have already a running vcsserver, this is not spawned again.
Run the web server as a subprocess. with it's own instance of vcsserver
"""
env = os.environ.copy()
env['RC_NO_TMP_PATH'] = '1'
rc_log = list(RC_LOG.partition('.log'))
rc_log.insert(1, get_port(rc_web_server_config))
rc_log = ''.join(rc_log)
vcsserver_port = available_port_factory()
print('Using vcsserver ops test port {}'.format(vcsserver_port))
server_out = open(rc_log, 'w')
vcs_log = os.path.join(tempfile.gettempdir(), 'rc_op_vcs.log')
vcsserver_factory(
request, vcsserver_port=vcsserver_port,
log_file=vcs_log,
overrides=({'server:main': {'workers': 2}},))
host_url = 'http://' + get_host_url(rc_web_server_config)
assert_no_running_instance(host_url)
command = ['gunicorn', '--worker-class', 'gevent', '--paste', rc_web_server_config]
print('rhodecode-web starting at: {}'.format(host_url))
print('rhodecode-web command: {}'.format(command))
print('rhodecode-web logfile: {}'.format(rc_log))
proc = subprocess32.Popen(
command, bufsize=0, env=env, stdout=server_out, stderr=server_out)
wait_for_url(host_url, timeout=30)
rc_log = os.path.join(tempfile.gettempdir(), 'rc_op_web.log')
rc_web_server_config = rc_web_server_config_factory(
vcsserver_port=vcsserver_port)
server = RcWebServer(rc_web_server_config, log_file=rc_log)
server.start()
@request.addfinalizer
def stop_web_server():
# TODO: Find out how to integrate with the reporting of py.test to
# make this information available.
print("\nServer log file written to %s" % (rc_log, ))
proc.kill()
server_out.flush()
server_out.close()
def cleanup():
server.shutdown()
return RcWebServer(rc_web_server_config, log_file=rc_log)
server.wait_until_ready()
return server
@pytest.fixture
@ -274,3 +222,41 @@ def fs_repo_only(request, rhodecode_fixtures):
request.addfinalizer(cleanup)
return fs_repo_fabric
@pytest.fixture
def enable_webhook_push_integration(request):
integration = Integration()
integration.integration_type = WebhookIntegrationType.key
Session().add(integration)
settings = dict(
url='http://httpbin.org',
secret_token='secret',
username=None,
password=None,
custom_header_key=None,
custom_header_val=None,
method_type='get',
events=[events.RepoPushEvent.name],
log_data=True
)
IntegrationModel().update_integration(
integration,
name='IntegrationWebhookTest',
enabled=True,
settings=settings,
repo=None,
repo_group=None,
child_repos_only=False,
)
Session().commit()
integration_id = integration.integration_id
@request.addfinalizer
def cleanup():
integration = Integration.get(integration_id)
Session().delete(integration)
Session().commit()

8
rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_403.py
View file

@ -33,14 +33,12 @@ from rhodecode.tests import (GIT_REPO, HG_REPO)
from rhodecode.tests.vcs_operations import Command
# override rc_web_server_config fixture with custom INI
@pytest.fixture(scope='module')
def rc_web_server_config(testini_factory):
CUSTOM_PARAMS = [
@pytest.fixture(scope="module")
def rc_web_server_config_modification():
return [
{'app:main': {'auth_ret_code': '403'}},
{'app:main': {'auth_ret_code_detection': 'true'}},
]
return testini_factory(CUSTOM_PARAMS)
@pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")

8
rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_404.py
View file

@ -33,14 +33,12 @@ from rhodecode.tests import (GIT_REPO, HG_REPO)
from rhodecode.tests.vcs_operations import Command
# override rc_web_server_config fixture with custom INI
@pytest.fixture(scope='module')
def rc_web_server_config(testini_factory):
CUSTOM_PARAMS = [
@pytest.fixture(scope="module")
def rc_web_server_config_modification():
return [
{'app:main': {'auth_ret_code': '404'}},
{'app:main': {'auth_ret_code_detection': 'false'}},
]
return testini_factory(CUSTOM_PARAMS)
@pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")

8
rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_bad_code.py
View file

@ -33,14 +33,12 @@ from rhodecode.tests import (GIT_REPO, HG_REPO)
from rhodecode.tests.vcs_operations import Command
# override rc_web_server_config fixture with custom INI
@pytest.fixture(scope='module')
def rc_web_server_config(testini_factory):
CUSTOM_PARAMS = [
@pytest.fixture(scope="module")
def rc_web_server_config_modification():
return [
{'app:main': {'auth_ret_code': '600'}},
{'app:main': {'auth_ret_code_detection': 'false'}},
]
return testini_factory(CUSTOM_PARAMS)
@pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")

12
rhodecode/tests/vcs_operations/test_vcs_calls_small_post_buffer.py
View file

@ -37,18 +37,6 @@ from rhodecode.tests.vcs_operations import Command
from .test_vcs_operations import _check_proper_clone, _check_proper_git_push
# override rc_web_server_config fixture with custom INI
@pytest.fixture(scope="module")
def rc_web_server_config(testini_factory):
"""
Configuration file used for the fixture `rc_web_server`.
"""
CUSTOM_PARAMS = [
{'handler_console': {'level': 'DEBUG'}},
]
return testini_factory(CUSTOM_PARAMS)
def test_git_clone_with_small_push_buffer(backend_git, rc_web_server, tmpdir):
clone_url = rc_web_server.repo_clone_url(GIT_REPO)
cmd = Command('/tmp')

8
rhodecode/tests/vcs_operations/test_vcs_operations_locking.py
View file

@ -41,13 +41,11 @@ from rhodecode.tests.vcs_operations import (
Command, _check_proper_clone, _check_proper_git_push, _add_files_and_push)
# override rc_web_server_config fixture with custom INI
@pytest.fixture(scope='module')
def rc_web_server_config(testini_factory):
CUSTOM_PARAMS = [
@pytest.fixture(scope="module")
def rc_web_server_config_modification():
return [
{'app:main': {'lock_ret_code': '423'}},
]
return testini_factory(CUSTOM_PARAMS)
@pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")

8
rhodecode/tests/vcs_operations/test_vcs_operations_locking_custom_code.py
View file

@ -40,13 +40,11 @@ from rhodecode.tests import (
from rhodecode.tests.vcs_operations import Command, _add_files_and_push
# override rc_web_server_config fixture with custom INI
@pytest.fixture(scope='module')
def rc_web_server_config(testini_factory):
CUSTOM_PARAMS = [
@pytest.fixture(scope="module")
def rc_web_server_config_modification():
return [
{'app:main': {'lock_ret_code': '400'}},
]
return testini_factory(CUSTOM_PARAMS)
@pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")

22
rhodecode/tests/vcs_operations/test_vcs_operations_special.py
View file

@ -61,17 +61,17 @@ class TestVCSOperationsSpecial(object):
# Doing an explicit commit in order to get latest user logs on MySQL
Session().commit()
# def test_git_fetches_from_remote_repository_with_annotated_tags(
# self, backend_git, rc_web_server):
# # Note: This is a test specific to the git backend. It checks the
# # integration of fetching from a remote repository which contains
# # annotated tags.
#
# # Dulwich shows this specific behavior only when
# # operating against a remote repository.
# source_repo = backend_git['annotated-tag']
# target_vcs_repo = backend_git.create_repo().scm_instance()
# target_vcs_repo.fetch(rc_web_server.repo_clone_url(source_repo.repo_name))
def test_git_fetches_from_remote_repository_with_annotated_tags(
self, backend_git, rc_web_server):
# Note: This is a test specific to the git backend. It checks the
# integration of fetching from a remote repository which contains
# annotated tags.
# Dulwich shows this specific behavior only when
# operating against a remote repository.
source_repo = backend_git['annotated-tag']
target_vcs_repo = backend_git.create_repo().scm_instance()
target_vcs_repo.fetch(rc_web_server.repo_clone_url(source_repo.repo_name))
def test_git_push_shows_pull_request_refs(self, backend_git, rc_web_server, tmpdir):
"""

43
rhodecode/tests/vcs_operations/test_vcs_operations_tag_push.py
View file

@ -30,14 +30,8 @@ Test suite for making push/pull operations, on specially modified INI files
import pytest
import requests
from rhodecode import events
from rhodecode.model.db import Integration
from rhodecode.model.integration import IntegrationModel
from rhodecode.model.meta import Session
from rhodecode.tests import GIT_REPO, HG_REPO
from rhodecode.tests.vcs_operations import Command, _add_files_and_push
from rhodecode.integrations.types.webhook import WebhookIntegrationType
def check_connection():
@ -54,43 +48,6 @@ connection_available = pytest.mark.skipif(
not check_connection(), reason="No outside internet connection available")
@pytest.fixture
def enable_webhook_push_integration(request):
integration = Integration()
integration.integration_type = WebhookIntegrationType.key
Session().add(integration)
settings = dict(
url='http://httpbin.org',
secret_token='secret',
username=None,
password=None,
custom_header_key=None,
custom_header_val=None,
method_type='get',
events=[events.RepoPushEvent.name],
log_data=True
)
IntegrationModel().update_integration(
integration,
name='IntegrationWebhookTest',
enabled=True,
settings=settings,
repo=None,
repo_group=None,
child_repos_only=False,
)
Session().commit()
integration_id = integration.integration_id
@request.addfinalizer
def cleanup():
integration = Integration.get(integration_id)
Session().delete(integration)
Session().commit()
@pytest.mark.usefixtures(
"disable_locking", "disable_anonymous_user",
"enable_webhook_push_integration")