pytest: renamed pylons_config into pyramid_config.

pytest.ini

@@ -1,15 +1,15 b''
 [pytest]
 testpaths = rhodecode
 norecursedirs = rhodecode/public rhodecode/templates tests/scripts
 
-pylons_config = rhodecode/tests/rhodecode.ini
+pyramid_config = rhodecode/tests/rhodecode.ini
 vcsserver_protocol = http
 vcsserver_config_http = rhodecode/tests/vcsserver_http.ini
 
 addopts =
   --pdbcls=IPython.terminal.debugger:TerminalPdb
 
 markers =
   vcs_operations: Mark tests depending on a running RhodeCode instance.
   xfail_backends: Mark tests as xfail for given backends.
   skip_backends: Mark tests as skipped for given backends.

rhodecode/tests/other/vcs_operations/conftest.py

@@ -1,264 +1,264 b''
 # -*- 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/
 
 """
 py.test config for test suite for making push/pull operations.
 
 .. important::
 
    You must have git >= 1.8.5 for tests to work fine. With 68b939b git started
    to redirect things to stderr instead of stdout.
 """
 
 import ConfigParser
 import os
 import subprocess32
 import tempfile
 import textwrap
 import pytest
 
 import rhodecode
 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
 
 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_host_url(pylons_config):
+def get_host_url(pyramid_config):
     """Construct the host url using the port in the test configuration."""
     config = ConfigParser.ConfigParser()
-    config.read(pylons_config)
+    config.read(pyramid_config)
 
     return '127.0.0.1:%s' % config.get('server:main', 'port')
 
 
 class RcWebServer(object):
     """
     Represents a running RCE web server used as a test fixture.
     """
-    def __init__(self, pylons_config):
-        self.pylons_config = pylons_config
+    def __init__(self, pyramid_config):
+        self.pyramid_config = pyramid_config
 
     def repo_clone_url(self, repo_name, **kwargs):
         params = {
             'user': TEST_USER_ADMIN_LOGIN,
             'passwd': TEST_USER_ADMIN_PASS,
-            'host': get_host_url(self.pylons_config),
+            '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.pylons_config)
+        return 'http://' + get_host_url(self.pyramid_config)
 
 
 @pytest.fixture(scope="module")
 def rcextensions(request, baseapp, tmpdir_factory):
     """
     Installs a testing rcextensions pack to ensure they work as expected.
     """
     init_content = textwrap.dedent("""
         # Forward import the example rcextensions to make it
         # active for our tests.
         from rhodecode.tests.other.example_rcextensions import *
     """)
 
     # Note: rcextensions are looked up based on the path of the ini file
     root_path = tmpdir_factory.getbasetemp()
     rcextensions_path = root_path.join('rcextensions')
     init_path = rcextensions_path.join('__init__.py')
 
     if rcextensions_path.check():
         pytest.fail(
             "Path for rcextensions already exists, please clean up before "
             "test run this path: %s" % (rcextensions_path, ))
         return
 
     request.addfinalizer(rcextensions_path.remove)
     init_path.write_binary(init_content, ensure=True)
 
 
 @pytest.fixture(scope="module")
 def repos(request, baseapp):
     """Create a copy of each test repo in a repo group."""
     fixture = Fixture()
     repo_group = fixture.create_repo_group(REPO_GROUP)
     repo_group_id = repo_group.group_id
     fixture.create_fork(HG_REPO, HG_REPO,
                         repo_name_full=HG_REPO_WITH_GROUP,
                         repo_group=repo_group_id)
     fixture.create_fork(GIT_REPO, GIT_REPO,
                         repo_name_full=GIT_REPO_WITH_GROUP,
                         repo_group=repo_group_id)
 
     @request.addfinalizer
     def cleanup():
         fixture.destroy_repo(HG_REPO_WITH_GROUP)
         fixture.destroy_repo(GIT_REPO_WITH_GROUP)
         fixture.destroy_repo_group(repo_group_id)
 
 
 @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)
 
 
 @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.
     """
     env = os.environ.copy()
     env['RC_NO_TMP_PATH'] = '1'
 
     rc_log = RC_LOG
     server_out = open(rc_log, 'w')
 
     # TODO: Would be great to capture the output and err of the subprocess
     # and make it available in a section of the py.test report in case of an
     # error.
 
     host_url = 'http://' + get_host_url(rc_web_server_config)
     assert_no_running_instance(host_url)
     command = ['pserve', rc_web_server_config]
 
     print('Starting rhodecode server: {}'.format(host_url))
     print('Command: {}'.format(command))
     print('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)
 
     @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()
 
     return RcWebServer(rc_web_server_config)
 
 
 @pytest.fixture
 def disable_locking(baseapp):
     r = Repository.get_by_repo_name(GIT_REPO)
     Repository.unlock(r)
     r.enable_locking = False
     Session().add(r)
     Session().commit()
 
     r = Repository.get_by_repo_name(HG_REPO)
     Repository.unlock(r)
     r.enable_locking = False
     Session().add(r)
     Session().commit()
 
 
 @pytest.fixture
 def enable_auth_plugins(request, baseapp, csrf_token):
     """
     Return a factory object that when called, allows to control which
     authentication plugins are enabled.
     """
     def _enable_plugins(plugins_list, override=None):
         override = override or {}
         params = {
             'auth_plugins': ','.join(plugins_list),
         }
 
         # helper translate some names to others
         name_map = {
             'token': 'authtoken'
         }
 
         for module in plugins_list:
             plugin_name = module.partition('#')[-1]
             if plugin_name in name_map:
                 plugin_name = name_map[plugin_name]
             enabled_plugin = 'auth_%s_enabled' % plugin_name
             cache_ttl = 'auth_%s_cache_ttl' % plugin_name
 
             # default params that are needed for each plugin,
             # `enabled` and `cache_ttl`
             params.update({
                 enabled_plugin: True,
                 cache_ttl: 0
             })
             if override.get:
                 params.update(override.get(module, {}))
 
             validated_params = params
             for k, v in validated_params.items():
                 setting = SettingsModel().create_or_update_setting(k, v)
                 Session().add(setting)
             Session().commit()
 
     def cleanup():
         _enable_plugins(['egg:rhodecode-enterprise-ce#rhodecode'])
 
     request.addfinalizer(cleanup)
 
     return _enable_plugins
 
 
 @pytest.fixture
 def fs_repo_only(request, rhodecode_fixtures):
     def fs_repo_fabric(repo_name, repo_type):
         rhodecode_fixtures.create_repo(repo_name, repo_type=repo_type)
         rhodecode_fixtures.destroy_repo(repo_name, fs_remove=False)
 
         def cleanup():
             rhodecode_fixtures.destroy_repo(repo_name, fs_remove=True)
             rhodecode_fixtures.destroy_repo_on_filesystem(repo_name)
 
         request.addfinalizer(cleanup)
 
     return fs_repo_fabric

rhodecode/tests/pylons_plugin.py

@@ -1,386 +1,386 b''
 # -*- 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 json
 import platform
 import socket
 import subprocess32
 import time
 from urllib2 import urlopen, URLError
 
 import configobj
 import pytest
 
 import pyramid.paster
 from rhodecode.tests.fixture import TestINI
 import rhodecode
 
 
 def _parse_json(value):
     return json.loads(value) if value else None
 
 
 def pytest_addoption(parser):
     parser.addoption(
         '--test-loglevel', dest='test_loglevel',
         help="Set default Logging level for tests, warn (default), info, debug")
     group = parser.getgroup('pylons')
     group.addoption(
-        '--with-pylons', dest='pylons_config',
+        '--with-pylons', dest='pyramid_config',
         help="Set up a Pylons environment with the specified config file.")
     group.addoption(
         '--ini-config-override', action='store', type=_parse_json,
-        default=None, dest='pylons_config_override', help=(
+        default=None, dest='pyramid_config_override', help=(
             "Overrides the .ini file settings. Should be specified in JSON"
             " format, e.g. '{\"section\": {\"parameter\": \"value\", ...}}'"
         )
     )
     parser.addini(
-        'pylons_config',
-        "Set up a Pylons environment with the specified config file.")
+        'pyramid_config',
+        "Set up a Pyramid environment with the specified config file.")
 
     vcsgroup = parser.getgroup('vcs')
     vcsgroup.addoption(
         '--without-vcsserver', dest='with_vcsserver', action='store_false',
         help="Do not start the VCSServer in a background process.")
     vcsgroup.addoption(
         '--with-vcsserver-http', dest='vcsserver_config_http',
         help="Start the HTTP VCSServer with the specified config file.")
     vcsgroup.addoption(
         '--vcsserver-protocol', dest='vcsserver_protocol',
         help="Start the VCSServer with HTTP protocol support.")
     vcsgroup.addoption(
         '--vcsserver-config-override', action='store', type=_parse_json,
         default=None, dest='vcsserver_config_override', help=(
             "Overrides the .ini file settings for the VCSServer. "
             "Should be specified in JSON "
             "format, e.g. '{\"section\": {\"parameter\": \"value\", ...}}'"
         )
     )
     vcsgroup.addoption(
         '--vcsserver-port', action='store', type=int,
         default=None, help=(
             "Allows to set the port of the vcsserver. Useful when testing "
             "against an already running server and random ports cause "
             "trouble."))
     parser.addini(
         'vcsserver_config_http',
         "Start the HTTP VCSServer with the specified config file.")
     parser.addini(
         'vcsserver_protocol',
         "Start the VCSServer with HTTP protocol support.")
 
 
 @pytest.fixture(scope='session')
 def vcsserver(request, vcsserver_port, vcsserver_factory):
     """
     Session scope VCSServer.
 
     Tests wich need the VCSServer have to rely on this fixture in order
     to ensure it will be running.
 
     For specific needs, the fixture vcsserver_factory can be used. It allows to
     adjust the configuration file for the test run.
 
     Command line args:
 
     --without-vcsserver: Allows to switch this fixture off. You have to
     manually start the server.
 
     --vcsserver-port: Will expect the VCSServer to listen on this port.
     """
 
     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)
 
 
 @pytest.fixture(scope='session')
 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):
 
         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}})
 
         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 ''
         override_option_name = 'vcsserver_config_override'
         config_file = get_config(
             request.config, option_name=option_name,
             override_option_name=override_option_name, overrides=overrides,
             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.start()
 
         @request.addfinalizer
         def cleanup():
             server.shutdown()
 
         server.wait_until_ready()
         return server
 
     return factory
 
 
 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):
         config_data = configobj.ConfigObj(config_file)
         self._config = config_data['server:main']
 
         args = ['pserve', config_file]
         self._args = args
 
     @property
     def http_url(self):
         template = 'http://{host}:{port}/'
         return template.format(**self._config)
 
     def start(self):
         self.process = subprocess32.Popen(self._args)
 
     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 = 'pylons_config'
+    option_name = 'pyramid_config'
     log_level = _use_log_level(request.config)
 
     overrides = [
         {'server:main': {'port': rcserver_port}},
         {'app:main': {
             'vcs.server': 'localhost:%s' % vcsserver_port,
             # johbo: We will always start the VCSServer on our own based on the
             # fixtures of the test cases. For the test run it must always be
             # off in the INI file.
             'vcs.start_server': 'false',
         }},
 
         {'handler_console': {
             'class ': 'StreamHandler',
             'args ': '(sys.stderr,)',
             'level': log_level,
         }},
 
     ]
     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,
         override_option_name='{}_override'.format(option_name),
         overrides=overrides,
         basetemp=tmpdir_factory.getbasetemp().strpath,
         prefix='test_rce_')
     return filename
 
 
 @pytest.fixture(scope='session')
 def rcserver_port(request):
     port = get_available_port()
     print('Using rcserver port {}'.format(port))
     return port
 
 
 @pytest.fixture(scope='session')
 def vcsserver_port(request):
     port = request.config.getoption('--vcsserver-port')
     if port is None:
         port = get_available_port()
         print('Using vcsserver port {}'.format(port))
     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():
     """
     Returns a callable which returns free port numbers.
     """
     return get_available_port
 
 
 @pytest.fixture
 def available_port(available_port_factory):
     """
     Gives you one free port for the current test.
 
     Uses "available_port_factory" to retrieve the port.
     """
     return available_port_factory()
 
 
 @pytest.fixture(scope='session')
 def baseapp(ini_config, vcsserver, http_environ_session):
     from rhodecode.lib.pyramid_utils import get_app_config
     from rhodecode.config.middleware import make_pyramid_app
 
     print("Using the RhodeCode configuration:{}".format(ini_config))
     pyramid.paster.setup_logging(ini_config)
 
     settings = get_app_config(ini_config)
     app = make_pyramid_app({'__file__': ini_config}, **settings)
 
     return app
 
 
 @pytest.fixture(scope='session')
 def testini_factory(tmpdir_factory, ini_config):
     """
     Factory to create an INI file based on TestINI.
 
     It will make sure to place the INI file in the correct directory.
     """
     basetemp = tmpdir_factory.getbasetemp().strpath
     return TestIniFactory(basetemp, ini_config)
 
 
 class TestIniFactory(object):
 
     def __init__(self, basetemp, template_ini):
         self._basetemp = basetemp
         self._template_ini = template_ini
 
     def __call__(self, ini_params, new_file_prefix='test'):
         ini_file = TestINI(
             self._template_ini, ini_params=ini_params,
             new_file_prefix=new_file_prefix, dir=self._basetemp)
         result = ini_file.create()
         return result
 
 
 def get_config(
         config, option_name, override_option_name, overrides=None,
         basetemp=None, prefix='test'):
     """
     Find a configuration file and apply overrides for the given `prefix`.
     """
     config_file = (
         config.getoption(option_name) or config.getini(option_name))
     if not config_file:
         pytest.exit(
             "Configuration error, could not extract {}.".format(option_name))
 
     overrides = overrides or []
     config_override = config.getoption(override_option_name)
     if config_override:
         overrides.append(config_override)
     temp_ini_file = TestINI(
         config_file, ini_params=overrides, new_file_prefix=prefix,
         dir=basetemp)
 
     return temp_ini_file.create()