# HG changeset patch # User Marcin Kuzminski # Date 2017-12-18 20:19:29 # Node ID 240dad60f8f3c6c28dce29f8f0c5c5588ea1aef2 # Parent db31248984a80061450370fffac28b65102e7878 tests: rewrote code for running vcs_operations. Now it starts it's own configurable instances for exactly those tests. diff --git a/rhodecode/tests/lib/middleware/utils/test_scm_app_http.py b/rhodecode/tests/lib/middleware/utils/test_scm_app_http.py --- a/rhodecode/tests/lib/middleware/utils/test_scm_app_http.py +++ b/rhodecode/tests/lib/middleware/utils/test_scm_app_http.py @@ -43,7 +43,6 @@ def vcsserver_http_echo_app(request, vcs """ vcsserver = vcsserver_factory( request=request, - use_http=True, overrides=[{'app:main': {'dev.use_echo_app': 'true'}}]) return vcsserver diff --git a/rhodecode/tests/pylons_plugin.py b/rhodecode/tests/pylons_plugin.py --- a/rhodecode/tests/pylons_plugin.py +++ b/rhodecode/tests/pylons_plugin.py @@ -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, v 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, # 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, }}, ] - 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(): """ diff --git a/rhodecode/tests/server_utils.py b/rhodecode/tests/server_utils.py new file mode 100644 --- /dev/null +++ b/rhodecode/tests/server_utils.py @@ -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 . +# +# 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 diff --git a/rhodecode/tests/vcs/conftest.py b/rhodecode/tests/vcs/conftest.py --- a/rhodecode/tests/vcs/conftest.py +++ b/rhodecode/tests/vcs/conftest.py @@ -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): diff --git a/rhodecode/tests/vcs_operations/conftest.py b/rhodecode/tests/vcs_operations/conftest.py --- a/rhodecode/tests/vcs_operations/conftest.py +++ b/rhodecode/tests/vcs_operations/conftest.py @@ -27,76 +27,30 @@ py.test config for test suite for making 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.integrations.types.webhook import WebhookIntegrationType + +from rhodecode.tests import GIT_REPO, HG_REPO from rhodecode.tests.fixture import Fixture -from rhodecode.tests.utils import is_url_reachable, wait_for_url +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, tmpdi @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): - """ - Run the web server as a subprocess. - - Since we have already a running vcsserver, this is not spawned again. + request, vcsserver_factory, available_port_factory, + rc_web_server_config_factory, repos, rcextensions): """ - env = os.environ.copy() - env['RC_NO_TMP_PATH'] = '1' + Run the web server as a subprocess. with it's own instance of vcsserver + """ - 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') - - 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] + 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}},)) - 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_fixt 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() + diff --git a/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_403.py b/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_403.py --- a/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_403.py +++ b/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_403.py @@ -33,14 +33,12 @@ from rhodecode.tests import (GIT_REPO, H 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") diff --git a/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_404.py b/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_404.py --- a/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_404.py +++ b/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_404.py @@ -33,14 +33,12 @@ from rhodecode.tests import (GIT_REPO, H 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") diff --git a/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_bad_code.py b/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_bad_code.py --- a/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_bad_code.py +++ b/rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_bad_code.py @@ -33,14 +33,12 @@ from rhodecode.tests import (GIT_REPO, H 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") diff --git a/rhodecode/tests/vcs_operations/test_vcs_calls_small_post_buffer.py b/rhodecode/tests/vcs_operations/test_vcs_calls_small_post_buffer.py --- a/rhodecode/tests/vcs_operations/test_vcs_calls_small_post_buffer.py +++ b/rhodecode/tests/vcs_operations/test_vcs_calls_small_post_buffer.py @@ -37,18 +37,6 @@ from rhodecode.tests.vcs_operations impo 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') diff --git a/rhodecode/tests/vcs_operations/test_vcs_operations_locking.py b/rhodecode/tests/vcs_operations/test_vcs_operations_locking.py --- a/rhodecode/tests/vcs_operations/test_vcs_operations_locking.py +++ b/rhodecode/tests/vcs_operations/test_vcs_operations_locking.py @@ -41,13 +41,11 @@ from rhodecode.tests.vcs_operations impo 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") diff --git a/rhodecode/tests/vcs_operations/test_vcs_operations_locking_custom_code.py b/rhodecode/tests/vcs_operations/test_vcs_operations_locking_custom_code.py --- a/rhodecode/tests/vcs_operations/test_vcs_operations_locking_custom_code.py +++ b/rhodecode/tests/vcs_operations/test_vcs_operations_locking_custom_code.py @@ -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") diff --git a/rhodecode/tests/vcs_operations/test_vcs_operations_special.py b/rhodecode/tests/vcs_operations/test_vcs_operations_special.py --- a/rhodecode/tests/vcs_operations/test_vcs_operations_special.py +++ b/rhodecode/tests/vcs_operations/test_vcs_operations_special.py @@ -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): """ diff --git a/rhodecode/tests/vcs_operations/test_vcs_operations_tag_push.py b/rhodecode/tests/vcs_operations/test_vcs_operations_tag_push.py --- a/rhodecode/tests/vcs_operations/test_vcs_operations_tag_push.py +++ b/rhodecode/tests/vcs_operations/test_vcs_operations_tag_push.py @@ -30,14 +30,8 @@ Test suite for making push/pull operatio 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.skipi 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")