tests: move vcs_operations into its own module.

rhodecode/tests/vcs_operations/test_vcs_operations_special.py

@@ -0,0 +1,244 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 pytest
+
+from rhodecode.lib.vcs.backends.git.repository import GitRepository
+from rhodecode.lib.vcs.backends.hg.repository import MercurialRepository
+from rhodecode.lib.vcs.nodes import FileNode
+from rhodecode.model.meta import Session
+
+from rhodecode.tests.vcs_operations import (
+    Command, _check_proper_clone, _check_proper_git_push, _check_proper_hg_push)
+
+
[e-mail]("disable_locking")
+class TestVCSOperationsSpecial(object):
+
+    def test_git_sets_default_branch_if_not_master(
+            self, backend_git, tmpdir, rc_web_server):
+        empty_repo = backend_git.create_repo()
+        clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
+
+        cmd = Command(tmpdir.strpath)
+        cmd.execute('git clone', clone_url)
+
+        repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo.in_memory_commit.add(FileNode('file', content=''))
+        repo.in_memory_commit.commit(
+            message='Commit on branch test',
+            author='Automatic test',
+            branch='test')
+
+        repo_cmd = Command(repo.path)
+        stdout, stderr = repo_cmd.execute('git push --verbose origin test')
+        _check_proper_git_push(
+            stdout, stderr, branch='test', should_set_default_branch=True)
+
+        stdout, stderr = cmd.execute(
+            'git clone', clone_url, empty_repo.repo_name + '-clone')
+        _check_proper_clone(stdout, stderr, 'git')
+
+        # 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_push_shows_pull_request_refs(self, backend_git, rc_web_server, tmpdir):
+        """
+        test if remote info about refs is visible
+        """
+        empty_repo = backend_git.create_repo()
+
+        clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
+
+        cmd = Command(tmpdir.strpath)
+        cmd.execute('git clone', clone_url)
+
+        repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo.in_memory_commit.add(FileNode('readme.md', content='## Hello'))
+        repo.in_memory_commit.commit(
+            message='Commit on branch Master',
+            author='Automatic test',
+            branch='master')
+
+        repo_cmd = Command(repo.path)
+        stdout, stderr = repo_cmd.execute('git push --verbose origin master')
+        _check_proper_git_push(stdout, stderr, branch='master')
+
+        ref = '{}/{}/pull-request/new?branch=master'.format(
+            rc_web_server.host_url(), empty_repo.repo_name)
+        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stderr
+        assert 'remote: RhodeCode: push completed' in stderr
+
+        # push on the same branch
+        repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo.in_memory_commit.add(FileNode('setup.py', content='print\n'))
+        repo.in_memory_commit.commit(
+            message='Commit2 on branch Master',
+            author='Automatic test2',
+            branch='master')
+
+        repo_cmd = Command(repo.path)
+        stdout, stderr = repo_cmd.execute('git push --verbose origin master')
+        _check_proper_git_push(stdout, stderr, branch='master')
+
+        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stderr
+        assert 'remote: RhodeCode: push completed' in stderr
+
+        # new Branch
+        repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo.in_memory_commit.add(FileNode('feature1.py', content='## Hello world'))
+        repo.in_memory_commit.commit(
+            message='Commit on branch feature',
+            author='Automatic test',
+            branch='feature')
+
+        repo_cmd = Command(repo.path)
+        stdout, stderr = repo_cmd.execute('git push --verbose origin feature')
+        _check_proper_git_push(stdout, stderr, branch='feature')
+
+        ref = '{}/{}/pull-request/new?branch=feature'.format(
+            rc_web_server.host_url(), empty_repo.repo_name)
+        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stderr
+        assert 'remote: RhodeCode: push completed' in stderr
+
+    def test_hg_push_shows_pull_request_refs(self, backend_hg, rc_web_server, tmpdir):
+        empty_repo = backend_hg.create_repo()
+
+        clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
+
+        cmd = Command(tmpdir.strpath)
+        cmd.execute('hg clone', clone_url)
+
+        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo.in_memory_commit.add(FileNode(u'readme.md', content=u'## Hello'))
+        repo.in_memory_commit.commit(
+            message=u'Commit on branch default',
+            author=u'Automatic test',
+            branch='default')
+
+        repo_cmd = Command(repo.path)
+        repo_cmd.execute('hg checkout default')
+
+        stdout, stderr = repo_cmd.execute('hg push --verbose', clone_url)
+        _check_proper_hg_push(stdout, stderr, branch='default')
+
+        ref = '{}/{}/pull-request/new?branch=default'.format(
+            rc_web_server.host_url(), empty_repo.repo_name)
+        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
+        assert 'remote: RhodeCode: push completed' in stdout
+
+        # push on the same branch
+        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo.in_memory_commit.add(FileNode(u'setup.py', content=u'print\n'))
+        repo.in_memory_commit.commit(
+            message=u'Commit2 on branch default',
+            author=u'Automatic test2',
+            branch=u'default')
+
+        repo_cmd = Command(repo.path)
+        repo_cmd.execute('hg checkout default')
+
+        stdout, stderr = repo_cmd.execute('hg push --verbose', clone_url)
+        _check_proper_hg_push(stdout, stderr, branch='default')
+
+        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
+        assert 'remote: RhodeCode: push completed' in stdout
+
+        # new Branch
+        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo.in_memory_commit.add(FileNode(u'feature1.py', content=u'## Hello world'))
+        repo.in_memory_commit.commit(
+            message=u'Commit on branch feature',
+            author=u'Automatic test',
+            branch=u'feature')
+
+        repo_cmd = Command(repo.path)
+        repo_cmd.execute('hg checkout feature')
+
+        stdout, stderr = repo_cmd.execute('hg push --new-branch --verbose', clone_url)
+        _check_proper_hg_push(stdout, stderr, branch='feature')
+
+        ref = '{}/{}/pull-request/new?branch=feature'.format(
+            rc_web_server.host_url(), empty_repo.repo_name)
+        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
+        assert 'remote: RhodeCode: push completed' in stdout
+
+    def test_hg_push_shows_pull_request_refs_book(self, backend_hg, rc_web_server, tmpdir):
+        empty_repo = backend_hg.create_repo()
+
+        clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
+
+        cmd = Command(tmpdir.strpath)
+        cmd.execute('hg clone', clone_url)
+
+        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo.in_memory_commit.add(FileNode(u'readme.md', content=u'## Hello'))
+        repo.in_memory_commit.commit(
+            message=u'Commit on branch default',
+            author=u'Automatic test',
+            branch='default')
+
+        repo_cmd = Command(repo.path)
+        repo_cmd.execute('hg checkout default')
+
+        stdout, stderr = repo_cmd.execute('hg push --verbose', clone_url)
+        _check_proper_hg_push(stdout, stderr, branch='default')
+
+        ref = '{}/{}/pull-request/new?branch=default'.format(
+            rc_web_server.host_url(), empty_repo.repo_name)
+        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
+        assert 'remote: RhodeCode: push completed' in stdout
+
+        # add bookmark
+        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
+        repo.in_memory_commit.add(FileNode(u'setup.py', content=u'print\n'))
+        repo.in_memory_commit.commit(
+            message=u'Commit2 on branch default',
+            author=u'Automatic test2',
+            branch=u'default')
+
+        repo_cmd = Command(repo.path)
+        repo_cmd.execute('hg checkout default')
+        repo_cmd.execute('hg bookmark feature2')
+        stdout, stderr = repo_cmd.execute('hg push -B feature2 --verbose', clone_url)
+        _check_proper_hg_push(stdout, stderr, branch='default')
+
+        ref = '{}/{}/pull-request/new?branch=default'.format(
+            rc_web_server.host_url(), empty_repo.repo_name)
+        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
+        ref = '{}/{}/pull-request/new?bookmark=feature2'.format(
+            rc_web_server.host_url(), empty_repo.repo_name)
+        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
+        assert 'remote: RhodeCode: push completed' in stdout
+        assert 'exporting bookmark feature2' in stdout

rhodecode/tests/pylons_plugin.py

@@ -1,410 +1,408 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 time
 import platform
 import socket
 import tempfile
+import subprocess32
 
-import subprocess32
-import time
 from urllib2 import urlopen, URLError
 
 import configobj
 import pytest
 
-import pyramid.paster
 
 from rhodecode.lib.pyramid_utils import get_app_config
 from rhodecode.tests.fixture import TestINI
-import rhodecode
-from rhodecode.tests.other.vcs_operations.conftest import get_host_url, get_port
+from rhodecode.tests.vcs_operations.conftest import get_host_url, get_port
 
 VCSSERVER_LOG = os.path.join(tempfile.gettempdir(), 'rc-vcsserver.log')
 
 
 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='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='pyramid_config_override', help=(
             "Overrides the .ini file settings. Should be specified in JSON"
             " format, e.g. '{\"section\": {\"parameter\": \"value\", ...}}'"
         )
     )
     parser.addini(
         '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 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):
     """
     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,
         overrides=vcs_server_config_override)
 
 
 @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):
         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'
     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 ini_settings(ini_config):
     ini_path = ini_config
     return get_app_config(ini_path)
 
 
 @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 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()

rhodecode/tests/vcs/conftest.py

@@ -1,257 +1,262 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 time
 import shutil
 import datetime
 
 import pytest
 
 from rhodecode.lib.vcs.backends import get_backend
 from rhodecode.lib.vcs.backends.base import Config
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests import get_new_dir
 from rhodecode.tests.utils import check_skip_backends, check_xfail_backends
 
 
[e-mail](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):
     """
     Provide a test repository for the test run.
 
     Depending on the value of `recreate_repo_per_test` a new repo for each
     test will be created.
 
     The parameter `--backends` can be used to limit this fixture to specific
     backend implementations.
     """
     cls = request.cls
 
     check_skip_backends(request.node, backend_alias)
     check_xfail_backends(request.node, backend_alias)
 
     if _should_create_repo_per_test(cls):
         _vcs_repo_container = _create_vcs_repo_container(request)
 
     repo = _vcs_repo_container.get_repo(cls, backend_alias=backend_alias)
 
     # TODO: johbo: Supporting old test class api, think about removing this
     cls.repo = repo
     cls.repo_path = repo.path
     cls.default_branch = repo.DEFAULT_BRANCH_NAME
     cls.Backend = cls.backend_class = repo.__class__
     cls.imc = repo.in_memory_commit
 
     return backend_alias, repo
 
 
 @pytest.fixture(scope='class')
 def _vcs_repo_container(request):
     """
     Internal fixture intended to help support class based scoping on demand.
     """
     return _create_vcs_repo_container(request)
 
 
 def _create_vcs_repo_container(request):
     repo_container = VcsRepoContainer()
     if not request.config.getoption('--keep-tmp-path'):
         request.addfinalizer(repo_container.cleanup)
     return repo_container
 
 
 class VcsRepoContainer(object):
 
     def __init__(self):
         self._cleanup_paths = []
         self._repos = {}
 
     def get_repo(self, test_class, backend_alias):
         if backend_alias not in self._repos:
             repo = _create_empty_repository(test_class, backend_alias)
 
             self._cleanup_paths.append(repo.path)
             self._repos[backend_alias] = repo
         return self._repos[backend_alias]
 
     def cleanup(self):
         for repo_path in reversed(self._cleanup_paths):
             shutil.rmtree(repo_path)
 
 
 def _should_create_repo_per_test(cls):
     return getattr(cls, 'recreate_repo_per_test', False)
 
 
 def _create_empty_repository(cls, backend_alias=None):
     Backend = get_backend(backend_alias or cls.backend_alias)
     repo_path = get_new_dir(str(time.time()))
     repo = Backend(repo_path, create=True)
     if hasattr(cls, '_get_commits'):
         commits = cls._get_commits()
         cls.tip = _add_commits_to_repo(repo, commits)
 
     return repo
 
 
 @pytest.fixture
 def config():
     """
     Instance of a repository config.
 
     The instance contains only one value:
 
     - Section: "section-a"
     - Key:     "a-1"
     - Value:   "value-a-1"
 
     The intended usage is for cases where a config instance is needed but no
     specific content is required.
     """
     config = Config()
     config.set('section-a', 'a-1', 'value-a-1')
     return config
 
 
 def _add_commits_to_repo(repo, commits):
     imc = repo.in_memory_commit
     tip = None
 
     for commit in commits:
         for node in commit.get('added', []):
             imc.add(FileNode(node.path, content=node.content))
         for node in commit.get('changed', []):
             imc.change(FileNode(node.path, content=node.content))
         for node in commit.get('removed', []):
             imc.remove(FileNode(node.path))
 
         tip = imc.commit(
             message=unicode(commit['message']),
             author=unicode(commit['author']),
             date=commit['date'],
             branch=commit.get('branch'))
 
     return tip
 
 
 @pytest.fixture
 def vcs_repo(request, backend_alias):
     Backend = get_backend(backend_alias)
     repo_path = get_new_dir(str(time.time()))
     repo = Backend(repo_path, create=True)
 
     @request.addfinalizer
     def cleanup():
         shutil.rmtree(repo_path)
 
     return repo
 
 
 @pytest.fixture
 def generate_repo_with_commits(vcs_repo):
     """
     Creates a fabric to generate N comits with some file nodes on a randomly
     generated repository
     """
 
     def commit_generator(num):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in xrange(num):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <[e-mail]>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('file_%d.txt' % x, content='Foobar %d' % x),
                 ],
                 'modified': [
                     FileNode('file_%d.txt' % x,
                              content='Foobar %d modified' % (x-1)),
                 ]
             }
 
     def commit_maker(num=5):
         _add_commits_to_repo(vcs_repo, commit_generator(num))
         return vcs_repo
 
     return commit_maker
 
 
 @pytest.fixture
 def hg_repo(request, vcs_repo):
     repo = vcs_repo
 
     commits = repo._get_commits()
     _add_commits_to_repo(repo, commits)
 
     return repo
 
 
 @pytest.fixture
 def hg_commit(hg_repo):
     return hg_repo.get_commit()
 
 
 class BackendTestMixin(object):
     """
     This is a backend independent test case class which should be created
     with ``type`` method.
 
     It is required to set following attributes at subclass:
 
     - ``backend_alias``: alias of used backend (see ``vcs.BACKENDS``)
     - ``repo_path``: path to the repository which would be created for set of
       tests
     - ``recreate_repo_per_test``: If set to ``False``, repo would NOT be
       created
       before every single test. Defaults to ``True``.
     """
     recreate_repo_per_test = True
 
     @classmethod
     def _get_commits(cls):
         commits = [
             {
                 'message': u'Initial commit',
                 'author': u'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar', content='Foobar'),
                     FileNode('foobar2', content='Foobar II'),
                     FileNode('foo/bar/baz', content='baz here!'),
                 ],
             },
             {
                 'message': u'Changes...',
                 'author': u'Jane Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 21),
                 'added': [
                     FileNode('some/new.txt', content='news...'),
                 ],
                 'changed': [
                     FileNode('foobar', 'Foobar I'),
                 ],
                 'removed': [],
             },
         ]
         return commits
 
 

rhodecode/tests/vcs_operations/conftest.py

@@ -1,271 +1,276 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_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):
     """
     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)
 
 
[e-mail](scope="session")
+def vcs_server_config_override():
+    return ({'server:main': {'workers': 2}},)
+
+
 @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 = list(RC_LOG.partition('.log'))
     rc_log.insert(1, get_port(rc_web_server_config))
     rc_log = ''.join(rc_log)
 
     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]
 
     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)
 
     @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, log_file=rc_log)
 
 
 @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/vcs_operations/test_vcs_calls_custom_auth_code_403.py

@@ -1,59 +1,59 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/
 
 """
 Test suite for making push/pull operations, on specially modified INI files
 
 .. 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 pytest
 
 from rhodecode.tests import (GIT_REPO, HG_REPO)
-from rhodecode.tests.other.vcs_operations import Command
+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 = [
         {'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")
 class TestVCSOperationsOnCustomIniConfig(object):
 
     def test_clone_wrong_credentials_hg_ret_code(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO, passwd='bad!')
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'abort: HTTP Error 403: Forbidden' in stderr
 
     def test_clone_wrong_credentials_git_ret_code(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO, passwd='bad!')
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
         assert 'The requested URL returned error: 403' in stderr

rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_404.py

@@ -1,59 +1,59 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/
 
 """
 Test suite for making push/pull operations, on specially modified INI files
 
 .. 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 pytest
 
 from rhodecode.tests import (GIT_REPO, HG_REPO)
-from rhodecode.tests.other.vcs_operations import Command
+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 = [
         {'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")
 class TestVCSOperationsOnCustomIniConfig(object):
 
     def test_clone_wrong_credentials_hg_ret_code(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO, passwd='bad!')
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'abort: HTTP Error 404: Not Found' in stderr
 
     def test_clone_wrong_credentials_git_ret_code(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO, passwd='bad!')
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
         assert 'not found' in stderr

rhodecode/tests/vcs_operations/test_vcs_calls_custom_auth_code_bad_code.py

@@ -1,59 +1,59 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/
 
 """
 Test suite for making push/pull operations, on specially modified INI files
 
 .. 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 pytest
 
 from rhodecode.tests import (GIT_REPO, HG_REPO)
-from rhodecode.tests.other.vcs_operations import Command
+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 = [
         {'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")
 class TestVCSOperationsOnCustomIniConfig(object):
 
     def test_clone_wrong_credentials_hg_ret_code(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO, passwd='bad!')
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'abort: HTTP Error 403: Forbidden' in stderr
 
     def test_clone_wrong_credentials_git_ret_code(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO, passwd='bad!')
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
         assert 'The requested URL returned error: 403' in stderr

rhodecode/tests/vcs_operations/test_vcs_calls_small_post_buffer.py

@@ -1,79 +1,79 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/
 
 """
 Test suite for making push/pull operations, on specially modified INI files
 
 .. 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 os
 import pytest
 
 from rhodecode.lib.vcs.backends.git.repository import GitRepository
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests import GIT_REPO
-from rhodecode.tests.other.vcs_operations import Command
+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')
     stdout, stderr = cmd.execute(
         'git -c http.postBuffer=1024 clone', clone_url, tmpdir.strpath)
     _check_proper_clone(stdout, stderr, 'git')
     cmd.assert_returncode_success()
 
 
 def test_git_push_with_small_push_buffer(backend_git, rc_web_server, tmpdir):
     empty_repo = backend_git.create_repo()
 
     clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
 
     cmd = Command(tmpdir.strpath)
     cmd.execute('git clone', clone_url)
 
     repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
     repo.in_memory_commit.add(FileNode('readme.md', content='## Hello'))
     repo.in_memory_commit.commit(
         message='Commit on branch Master',
         author='Automatic test',
         branch='master')
 
     repo_cmd = Command(repo.path)
     stdout, stderr = repo_cmd.execute(
         'git -c http.postBuffer=1024 push --verbose origin master')
     _check_proper_git_push(stdout, stderr, branch='master')

rhodecode/tests/vcs_operations/test_vcs_operations.py

@@ -1,682 +1,460 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/
 
 """
 Test suite for making push/pull operations, on specially modified INI files
 
 .. 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 os
 import time
 
 import pytest
 
-from rhodecode.lib.vcs.backends.git.repository import GitRepository
-from rhodecode.lib.vcs.backends.hg.repository import MercurialRepository
-from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model.db import Repository, UserIpMap, CacheKey
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import (GIT_REPO, HG_REPO, TEST_USER_ADMIN_LOGIN)
 
-from rhodecode.tests.other.vcs_operations import (
+from rhodecode.tests.vcs_operations import (
     Command, _check_proper_clone, _check_proper_git_push,
-    _check_proper_hg_push, _add_files_and_push,
-    HG_REPO_WITH_GROUP, GIT_REPO_WITH_GROUP)
-
-
[e-mail](scope="session")
-def vcs_server_config_override():
-    return ({'server:main': {'workers': 2}},)
+    _add_files_and_push, HG_REPO_WITH_GROUP, GIT_REPO_WITH_GROUP)
 
 
 @pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")
 class TestVCSOperations(object):
 
     def test_clone_hg_repo_by_admin(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'hg')
 
     def test_clone_git_repo_by_admin(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         cmd = Command('/tmp')
         stdout, stderr = cmd.execute('git clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'git')
         cmd.assert_returncode_success()
 
     def test_clone_git_repo_by_admin_with_git_suffix(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         cmd = Command('/tmp')
         stdout, stderr = cmd.execute('git clone', clone_url+".git", tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'git')
         cmd.assert_returncode_success()
 
     def test_clone_hg_repo_by_id_by_admin(self, rc_web_server, tmpdir):
         repo_id = Repository.get_by_repo_name(HG_REPO).repo_id
         clone_url = rc_web_server.repo_clone_url('_%s' % repo_id)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'hg')
 
     def test_clone_git_repo_by_id_by_admin(self, rc_web_server, tmpdir):
         repo_id = Repository.get_by_repo_name(GIT_REPO).repo_id
         clone_url = rc_web_server.repo_clone_url('_%s' % repo_id)
         cmd = Command('/tmp')
         stdout, stderr = cmd.execute('git clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'git')
         cmd.assert_returncode_success()
 
     def test_clone_hg_repo_with_group_by_admin(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO_WITH_GROUP)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'hg')
 
     def test_clone_git_repo_with_group_by_admin(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO_WITH_GROUP)
         cmd = Command('/tmp')
         stdout, stderr = cmd.execute('git clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'git')
         cmd.assert_returncode_success()
 
     def test_clone_git_repo_shallow_by_admin(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         cmd = Command('/tmp')
         stdout, stderr = cmd.execute(
             'git clone --depth=1', clone_url, tmpdir.strpath)
 
         assert '' == stdout
         assert 'Cloning into' in stderr
         cmd.assert_returncode_success()
 
     def test_clone_wrong_credentials_hg(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO, passwd='bad!')
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'abort: authorization failed' in stderr
 
     def test_clone_wrong_credentials_git(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO, passwd='bad!')
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
         assert 'fatal: Authentication failed' in stderr
 
     def test_clone_git_dir_as_hg(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'HTTP Error 404: Not Found' in stderr
 
     def test_clone_hg_repo_as_git(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
         assert 'not found' in stderr
 
     def test_clone_non_existing_path_hg(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url('trololo')
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'HTTP Error 404: Not Found' in stderr
 
     def test_clone_non_existing_path_git(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url('trololo')
         stdout, stderr = Command('/tmp').execute('git clone', clone_url)
         assert 'not found' in stderr
 
     def test_clone_existing_path_hg_not_in_database(
             self, rc_web_server, tmpdir, fs_repo_only):
 
         db_name = fs_repo_only('not-in-db-hg', repo_type='hg')
         clone_url = rc_web_server.repo_clone_url(db_name)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'HTTP Error 404: Not Found' in stderr
 
     def test_clone_existing_path_git_not_in_database(
             self, rc_web_server, tmpdir, fs_repo_only):
         db_name = fs_repo_only('not-in-db-git', repo_type='git')
         clone_url = rc_web_server.repo_clone_url(db_name)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
         assert 'not found' in stderr
 
     def test_clone_existing_path_hg_not_in_database_different_scm(
             self, rc_web_server, tmpdir, fs_repo_only):
         db_name = fs_repo_only('not-in-db-git', repo_type='git')
         clone_url = rc_web_server.repo_clone_url(db_name)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'HTTP Error 404: Not Found' in stderr
 
     def test_clone_existing_path_git_not_in_database_different_scm(
             self, rc_web_server, tmpdir, fs_repo_only):
         db_name = fs_repo_only('not-in-db-hg', repo_type='hg')
         clone_url = rc_web_server.repo_clone_url(db_name)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
         assert 'not found' in stderr
 
     def test_clone_non_existing_store_path_hg(self, rc_web_server, tmpdir, user_util):
         repo = user_util.create_repo()
         clone_url = rc_web_server.repo_clone_url(repo.repo_name)
 
         # Damage repo by removing it's folder
         RepoModel()._delete_filesystem_repo(repo)
 
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'HTTP Error 404: Not Found' in stderr
 
     def test_clone_non_existing_store_path_git(self, rc_web_server, tmpdir, user_util):
         repo = user_util.create_repo(repo_type='git')
         clone_url = rc_web_server.repo_clone_url(repo.repo_name)
 
         # Damage repo by removing it's folder
         RepoModel()._delete_filesystem_repo(repo)
 
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
         assert 'not found' in stderr
 
     def test_push_new_file_hg(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
 
         stdout, stderr = _add_files_and_push(
             'hg', tmpdir.strpath, clone_url=clone_url)
 
         assert 'pushing to' in stdout
         assert 'size summary' in stdout
 
     def test_push_new_file_git(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         # commit some stuff into this repo
         stdout, stderr = _add_files_and_push(
             'git', tmpdir.strpath, clone_url=clone_url)
 
         _check_proper_git_push(stdout, stderr)
 
     def test_push_invalidates_cache_hg(self, rc_web_server, tmpdir):
         key = CacheKey.query().filter(CacheKey.cache_key == HG_REPO).scalar()
         if not key:
             key = CacheKey(HG_REPO, HG_REPO)
 
         key.cache_active = True
         Session().add(key)
         Session().commit()
 
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
 
         stdout, stderr = _add_files_and_push(
             'hg', tmpdir.strpath, clone_url=clone_url, files_no=1)
 
         key = CacheKey.query().filter(CacheKey.cache_key == HG_REPO).one()
         assert key.cache_active is False
 
     def test_push_invalidates_cache_git(self, rc_web_server, tmpdir):
         key = CacheKey.query().filter(CacheKey.cache_key == GIT_REPO).scalar()
         if not key:
             key = CacheKey(GIT_REPO, GIT_REPO)
 
         key.cache_active = True
         Session().add(key)
         Session().commit()
 
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         # commit some stuff into this repo
         stdout, stderr = _add_files_and_push(
             'git', tmpdir.strpath, clone_url=clone_url, files_no=1)
         _check_proper_git_push(stdout, stderr)
 
         key = CacheKey.query().filter(CacheKey.cache_key == GIT_REPO).one()
 
         assert key.cache_active is False
 
     def test_push_wrong_credentials_hg(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
 
         push_url = rc_web_server.repo_clone_url(
             HG_REPO, user='bad', passwd='name')
         stdout, stderr = _add_files_and_push(
             'hg', tmpdir.strpath, clone_url=push_url)
 
         assert 'abort: authorization failed' in stderr
 
     def test_push_wrong_credentials_git(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         push_url = rc_web_server.repo_clone_url(
             GIT_REPO, user='bad', passwd='name')
         stdout, stderr = _add_files_and_push(
             'git', tmpdir.strpath, clone_url=push_url)
 
         assert 'fatal: Authentication failed' in stderr
 
     def test_push_back_to_wrong_url_hg(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
 
         stdout, stderr = _add_files_and_push(
             'hg', tmpdir.strpath,
             clone_url=rc_web_server.repo_clone_url('not-existing'))
 
         assert 'HTTP Error 404: Not Found' in stderr
 
     def test_push_back_to_wrong_url_git(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         stdout, stderr = _add_files_and_push(
             'git', tmpdir.strpath,
             clone_url=rc_web_server.repo_clone_url('not-existing'))
 
         assert 'not found' in stderr
 
     def test_ip_restriction_hg(self, rc_web_server, tmpdir):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             time.sleep(2)
             clone_url = rc_web_server.repo_clone_url(HG_REPO)
             stdout, stderr = Command('/tmp').execute(
                 'hg clone', clone_url, tmpdir.strpath)
             assert 'abort: HTTP Error 403: Forbidden' in stderr
         finally:
             # release IP restrictions
             for ip in UserIpMap.getAll():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
 
         time.sleep(2)
 
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'hg')
 
     def test_ip_restriction_git(self, rc_web_server, tmpdir):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             time.sleep(2)
             clone_url = rc_web_server.repo_clone_url(GIT_REPO)
             stdout, stderr = Command('/tmp').execute(
                 'git clone', clone_url, tmpdir.strpath)
             msg = "The requested URL returned error: 403"
             assert msg in stderr
         finally:
             # release IP restrictions
             for ip in UserIpMap.getAll():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
 
         time.sleep(2)
 
         cmd = Command('/tmp')
         stdout, stderr = cmd.execute('git clone', clone_url, tmpdir.strpath)
         cmd.assert_returncode_success()
         _check_proper_clone(stdout, stderr, 'git')
 
     def test_clone_by_auth_token(
             self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
         enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
                              'egg:rhodecode-enterprise-ce#rhodecode'])
 
         user = user_util.create_user()
         token = user.auth_tokens[1]
 
         clone_url = rc_web_server.repo_clone_url(
             HG_REPO, user=user.username, passwd=token)
 
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'hg')
 
     def test_clone_by_auth_token_expired(
             self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
         enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
                              'egg:rhodecode-enterprise-ce#rhodecode'])
 
         user = user_util.create_user()
         auth_token = AuthTokenModel().create(
             user.user_id, 'test-token', -10, AuthTokenModel.cls.ROLE_VCS)
         token = auth_token.api_key
 
         clone_url = rc_web_server.repo_clone_url(
             HG_REPO, user=user.username, passwd=token)
 
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'abort: authorization failed' in stderr
 
     def test_clone_by_auth_token_bad_role(
             self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
         enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
                              'egg:rhodecode-enterprise-ce#rhodecode'])
 
         user = user_util.create_user()
         auth_token = AuthTokenModel().create(
             user.user_id, 'test-token', -1, AuthTokenModel.cls.ROLE_API)
         token = auth_token.api_key
 
         clone_url = rc_web_server.repo_clone_url(
             HG_REPO, user=user.username, passwd=token)
 
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'abort: authorization failed' in stderr
 
     def test_clone_by_auth_token_user_disabled(
             self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
         enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
                              'egg:rhodecode-enterprise-ce#rhodecode'])
         user = user_util.create_user()
         user.active = False
         Session().add(user)
         Session().commit()
         token = user.auth_tokens[1]
 
         clone_url = rc_web_server.repo_clone_url(
             HG_REPO, user=user.username, passwd=token)
 
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'abort: authorization failed' in stderr
 
     def test_clone_by_auth_token_with_scope(
             self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
         enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
                              'egg:rhodecode-enterprise-ce#rhodecode'])
         user = user_util.create_user()
         auth_token = AuthTokenModel().create(
             user.user_id, 'test-token', -1, AuthTokenModel.cls.ROLE_VCS)
         token = auth_token.api_key
 
         # manually set scope
         auth_token.repo = Repository.get_by_repo_name(HG_REPO)
         Session().add(auth_token)
         Session().commit()
 
         clone_url = rc_web_server.repo_clone_url(
             HG_REPO, user=user.username, passwd=token)
 
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'hg')
 
     def test_clone_by_auth_token_with_wrong_scope(
             self, rc_web_server, tmpdir, user_util, enable_auth_plugins):
         enable_auth_plugins(['egg:rhodecode-enterprise-ce#token',
                              'egg:rhodecode-enterprise-ce#rhodecode'])
         user = user_util.create_user()
         auth_token = AuthTokenModel().create(
             user.user_id, 'test-token', -1, AuthTokenModel.cls.ROLE_VCS)
         token = auth_token.api_key
 
         # manually set scope
         auth_token.repo = Repository.get_by_repo_name(GIT_REPO)
         Session().add(auth_token)
         Session().commit()
 
         clone_url = rc_web_server.repo_clone_url(
             HG_REPO, user=user.username, passwd=token)
 
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         assert 'abort: authorization failed' in stderr
 
-
[e-mail]("disable_locking")
-class TestVCSOperationsSpecial(object):
-
-    def test_git_sets_default_branch_if_not_master(
-            self, backend_git, tmpdir, rc_web_server):
-        empty_repo = backend_git.create_repo()
-        clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
-
-        cmd = Command(tmpdir.strpath)
-        cmd.execute('git clone', clone_url)
-
-        repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
-        repo.in_memory_commit.add(FileNode('file', content=''))
-        repo.in_memory_commit.commit(
-            message='Commit on branch test',
-            author='Automatic test',
-            branch='test')
-
-        repo_cmd = Command(repo.path)
-        stdout, stderr = repo_cmd.execute('git push --verbose origin test')
-        _check_proper_git_push(
-            stdout, stderr, branch='test', should_set_default_branch=True)
-
-        stdout, stderr = cmd.execute(
-            'git clone', clone_url, empty_repo.repo_name + '-clone')
-        _check_proper_clone(stdout, stderr, 'git')
-
-        # 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_push_shows_pull_request_refs(self, backend_git, rc_web_server, tmpdir):
-        """
-        test if remote info about refs is visible
-        """
-        empty_repo = backend_git.create_repo()
-
-        clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
-
-        cmd = Command(tmpdir.strpath)
-        cmd.execute('git clone', clone_url)
-
-        repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
-        repo.in_memory_commit.add(FileNode('readme.md', content='## Hello'))
-        repo.in_memory_commit.commit(
-            message='Commit on branch Master',
-            author='Automatic test',
-            branch='master')
-
-        repo_cmd = Command(repo.path)
-        stdout, stderr = repo_cmd.execute('git push --verbose origin master')
-        _check_proper_git_push(stdout, stderr, branch='master')
-
-        ref = '{}/{}/pull-request/new?branch=master'.format(
-            rc_web_server.host_url(), empty_repo.repo_name)
-        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stderr
-        assert 'remote: RhodeCode: push completed' in stderr
-
-        # push on the same branch
-        repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
-        repo.in_memory_commit.add(FileNode('setup.py', content='print\n'))
-        repo.in_memory_commit.commit(
-            message='Commit2 on branch Master',
-            author='Automatic test2',
-            branch='master')
-
-        repo_cmd = Command(repo.path)
-        stdout, stderr = repo_cmd.execute('git push --verbose origin master')
-        _check_proper_git_push(stdout, stderr, branch='master')
-
-        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stderr
-        assert 'remote: RhodeCode: push completed' in stderr
-
-        # new Branch
-        repo = GitRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
-        repo.in_memory_commit.add(FileNode('feature1.py', content='## Hello world'))
-        repo.in_memory_commit.commit(
-            message='Commit on branch feature',
-            author='Automatic test',
-            branch='feature')
-
-        repo_cmd = Command(repo.path)
-        stdout, stderr = repo_cmd.execute('git push --verbose origin feature')
-        _check_proper_git_push(stdout, stderr, branch='feature')
-
-        ref = '{}/{}/pull-request/new?branch=feature'.format(
-            rc_web_server.host_url(), empty_repo.repo_name)
-        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stderr
-        assert 'remote: RhodeCode: push completed' in stderr
-
-    def test_hg_push_shows_pull_request_refs(self, backend_hg, rc_web_server, tmpdir):
-        empty_repo = backend_hg.create_repo()
-
-        clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
-
-        cmd = Command(tmpdir.strpath)
-        cmd.execute('hg clone', clone_url)
-
-        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
-        repo.in_memory_commit.add(FileNode(u'readme.md', content=u'## Hello'))
-        repo.in_memory_commit.commit(
-            message=u'Commit on branch default',
-            author=u'Automatic test',
-            branch='default')
-
-        repo_cmd = Command(repo.path)
-        repo_cmd.execute('hg checkout default')
-
-        stdout, stderr = repo_cmd.execute('hg push --verbose', clone_url)
-        _check_proper_hg_push(stdout, stderr, branch='default')
-
-        ref = '{}/{}/pull-request/new?branch=default'.format(
-            rc_web_server.host_url(), empty_repo.repo_name)
-        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
-        assert 'remote: RhodeCode: push completed' in stdout
-
-        # push on the same branch
-        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
-        repo.in_memory_commit.add(FileNode(u'setup.py', content=u'print\n'))
-        repo.in_memory_commit.commit(
-            message=u'Commit2 on branch default',
-            author=u'Automatic test2',
-            branch=u'default')
-
-        repo_cmd = Command(repo.path)
-        repo_cmd.execute('hg checkout default')
-
-        stdout, stderr = repo_cmd.execute('hg push --verbose', clone_url)
-        _check_proper_hg_push(stdout, stderr, branch='default')
-
-        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
-        assert 'remote: RhodeCode: push completed' in stdout
-
-        # new Branch
-        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
-        repo.in_memory_commit.add(FileNode(u'feature1.py', content=u'## Hello world'))
-        repo.in_memory_commit.commit(
-            message=u'Commit on branch feature',
-            author=u'Automatic test',
-            branch=u'feature')
-
-        repo_cmd = Command(repo.path)
-        repo_cmd.execute('hg checkout feature')
-
-        stdout, stderr = repo_cmd.execute('hg push --new-branch --verbose', clone_url)
-        _check_proper_hg_push(stdout, stderr, branch='feature')
-
-        ref = '{}/{}/pull-request/new?branch=feature'.format(
-            rc_web_server.host_url(), empty_repo.repo_name)
-        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
-        assert 'remote: RhodeCode: push completed' in stdout
-
-    def test_hg_push_shows_pull_request_refs_book(self, backend_hg, rc_web_server, tmpdir):
-        empty_repo = backend_hg.create_repo()
-
-        clone_url = rc_web_server.repo_clone_url(empty_repo.repo_name)
-
-        cmd = Command(tmpdir.strpath)
-        cmd.execute('hg clone', clone_url)
-
-        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
-        repo.in_memory_commit.add(FileNode(u'readme.md', content=u'## Hello'))
-        repo.in_memory_commit.commit(
-            message=u'Commit on branch default',
-            author=u'Automatic test',
-            branch='default')
-
-        repo_cmd = Command(repo.path)
-        repo_cmd.execute('hg checkout default')
-
-        stdout, stderr = repo_cmd.execute('hg push --verbose', clone_url)
-        _check_proper_hg_push(stdout, stderr, branch='default')
-
-        ref = '{}/{}/pull-request/new?branch=default'.format(
-            rc_web_server.host_url(), empty_repo.repo_name)
-        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
-        assert 'remote: RhodeCode: push completed' in stdout
-
-        # add bookmark
-        repo = MercurialRepository(os.path.join(tmpdir.strpath, empty_repo.repo_name))
-        repo.in_memory_commit.add(FileNode(u'setup.py', content=u'print\n'))
-        repo.in_memory_commit.commit(
-            message=u'Commit2 on branch default',
-            author=u'Automatic test2',
-            branch=u'default')
-
-        repo_cmd = Command(repo.path)
-        repo_cmd.execute('hg checkout default')
-        repo_cmd.execute('hg bookmark feature2')
-        stdout, stderr = repo_cmd.execute('hg push -B feature2 --verbose', clone_url)
-        _check_proper_hg_push(stdout, stderr, branch='default')
-
-        ref = '{}/{}/pull-request/new?branch=default'.format(
-            rc_web_server.host_url(), empty_repo.repo_name)
-        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
-        ref = '{}/{}/pull-request/new?bookmark=feature2'.format(
-            rc_web_server.host_url(), empty_repo.repo_name)
-        assert 'remote: RhodeCode: open pull request link: {}'.format(ref) in stdout
-        assert 'remote: RhodeCode: push completed' in stdout
-        assert 'exporting bookmark feature2' in stdout

rhodecode/tests/vcs_operations/test_vcs_operations_locking.py

@@ -1,222 +1,222 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/
 
 """
 Test suite for making push/pull operations, on specially modified INI files
 
 .. 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 pytest
 
 from rhodecode.model.db import User, Repository
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 
 from rhodecode.tests import (
     GIT_REPO, HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN,
     TEST_USER_REGULAR_PASS)
-from rhodecode.tests.other.vcs_operations import (
+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 = [
         {'app:main': {'lock_ret_code': '423'}},
     ]
     return testini_factory(CUSTOM_PARAMS)
 
 
 @pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")
 class TestVCSOperationsOnCustomIniConfig(object):
 
     def test_clone_and_create_lock_hg(self, rc_web_server, tmpdir):
         # enable locking
         r = Repository.get_by_repo_name(HG_REPO)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         # clone
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
 
         # check if lock was made
         r = Repository.get_by_repo_name(HG_REPO)
         assert r.locked[0] == User.get_by_username(
             TEST_USER_ADMIN_LOGIN).user_id
 
     def test_clone_and_create_lock_git(self, rc_web_server, tmpdir):
         # enable locking
         r = Repository.get_by_repo_name(GIT_REPO)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         # clone
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         # check if lock was made
         r = Repository.get_by_repo_name(GIT_REPO)
         assert r.locked[0] == User.get_by_username(
             TEST_USER_ADMIN_LOGIN).user_id
 
     def test_clone_after_repo_was_locked_hg(self, rc_web_server, tmpdir):
         # lock repo
         r = Repository.get_by_repo_name(HG_REPO)
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         # pull fails since repo is locked
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
 
     def test_clone_after_repo_was_locked_git(self, rc_web_server, tmpdir):
         # lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         # pull fails since repo is locked
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         lock_msg = (
             'remote: ERROR: Repository `vcs_test_git` locked by user ' +
             '`test_admin`. Reason:`lock_auto`')
         assert lock_msg in stderr
         assert 'remote: Pre pull hook failed: aborting' in stderr
         assert 'fatal: remote did not send all necessary objects' in stderr
 
     def test_push_on_locked_repo_by_other_user_hg(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
 
         # lock repo
         r = Repository.get_by_repo_name(HG_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
 
         # push fails repo is locked by other user !
         push_url = rc_web_server.repo_clone_url(
             HG_REPO,
             user=TEST_USER_REGULAR_LOGIN, passwd=TEST_USER_REGULAR_PASS)
         stdout, stderr = _add_files_and_push(
             'hg', tmpdir.strpath, clone_url=push_url)
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
 
     def test_push_on_locked_repo_by_other_user_git(
             self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         # lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
 
         # push fails repo is locked by other user!
         push_url = rc_web_server.repo_clone_url(
             GIT_REPO,
             user=TEST_USER_REGULAR_LOGIN, passwd=TEST_USER_REGULAR_PASS)
         stdout, stderr = _add_files_and_push(
             'git', tmpdir.strpath, clone_url=push_url)
 
         err = 'Repository `%s` locked by user `%s`' % (
             GIT_REPO, TEST_USER_ADMIN_LOGIN)
         # err = 'RPC failed; result=22, HTTP code = 423'
         assert err in stderr
 
     def test_push_unlocks_repository_hg(self, rc_web_server, tmpdir):
         # enable locking
         r = Repository.get_by_repo_name(HG_REPO)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
 
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'hg')
 
         # check for lock repo after clone
         r = Repository.get_by_repo_name(HG_REPO)
         uid = User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         assert r.locked[0] == uid
 
         # push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push(
             'hg', tmpdir.strpath, clone_url=clone_url)
         assert ('remote: Released lock on repo `%s`' % HG_REPO) in stdout
         # we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(HG_REPO)
         assert r.locked == [None, None, None]
 
     def test_push_unlocks_repository_git(self, rc_web_server, tmpdir):
 
         # Note: Did a first debugging session. Seems that
         # Repository.get_locking_state is called twice. The second call
         # has the action "pull" and does not reset the lock.
 
         # enable locking
         r = Repository.get_by_repo_name(GIT_REPO)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
 
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
         _check_proper_clone(stdout, stderr, 'git')
 
         # check for lock repo after clone
         r = Repository.get_by_repo_name(GIT_REPO)
         assert r.locked[0] == User.get_by_username(
             TEST_USER_ADMIN_LOGIN).user_id
 
         # push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push(
             'git', tmpdir.strpath, clone_url=clone_url)
         _check_proper_git_push(stdout, stderr)
 
         # assert ('remote: Released lock on repo `%s`' % GIT_REPO) in stdout
         # we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(GIT_REPO)
         assert r.locked == [None, None, None]

rhodecode/tests/vcs_operations/test_vcs_operations_locking_custom_code.py

@@ -1,130 +1,130 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/
 
 """
 Test suite for making push/pull operations, on specially modified INI files
 
 .. 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 pytest
 
 from rhodecode.model.db import User, Repository
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 
 from rhodecode.tests import (
     GIT_REPO, HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN,
     TEST_USER_REGULAR_PASS)
-from rhodecode.tests.other.vcs_operations import Command, _add_files_and_push
+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 = [
         {'app:main': {'lock_ret_code': '400'}},
     ]
     return testini_factory(CUSTOM_PARAMS)
 
 
 @pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")
 class TestVCSOperationsOnCustomIniConfig(object):
 
     def test_clone_after_repo_was_locked_hg(self, rc_web_server, tmpdir):
         # lock repo
         r = Repository.get_by_repo_name(HG_REPO)
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         # pull fails since repo is locked
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
         msg = ("""abort: HTTP Error 400: Repository `%s` locked by user `%s`"""
                % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
 
     def test_clone_after_repo_was_locked_git(self, rc_web_server, tmpdir):
         # lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         # pull fails since repo is locked
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         lock_msg = (
             'remote: ERROR: Repository `vcs_test_git` locked by user ' +
             '`test_admin`. Reason:`lock_auto`')
         assert lock_msg in stderr
         assert 'remote: Pre pull hook failed: aborting' in stderr
         assert 'fatal: remote did not send all necessary objects' in stderr
 
     def test_push_on_locked_repo_by_other_user_hg(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
 
         # lock repo
         r = Repository.get_by_repo_name(HG_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
 
         # push fails repo is locked by other user !
         push_url = rc_web_server.repo_clone_url(
             HG_REPO,
             user=TEST_USER_REGULAR_LOGIN, passwd=TEST_USER_REGULAR_PASS)
         stdout, stderr = _add_files_and_push(
             'hg', tmpdir.strpath, clone_url=push_url)
         msg = ("""abort: HTTP Error 400: Repository `%s` locked by user `%s`"""
                % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
 
     def test_push_on_locked_repo_by_other_user_git(
             self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         # lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
 
         # push fails repo is locked by other user!
         push_url = rc_web_server.repo_clone_url(
             GIT_REPO,
             user=TEST_USER_REGULAR_LOGIN, passwd=TEST_USER_REGULAR_PASS)
         stdout, stderr = _add_files_and_push(
             'git', tmpdir.strpath, clone_url=push_url)
 
         err = 'Repository `%s` locked by user `%s`' % (
             GIT_REPO, TEST_USER_ADMIN_LOGIN)
 
         assert err in stderr

rhodecode/tests/vcs_operations/test_vcs_operations_tag_push.py

@@ -1,146 +1,141 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/
 
 """
 Test suite for making push/pull operations, on specially modified INI files
 
 .. 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 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.other.vcs_operations import Command, _add_files_and_push
+from rhodecode.tests.vcs_operations import Command, _add_files_and_push
 from rhodecode.integrations.types.webhook import WebhookIntegrationType
 
 
 def check_connection():
     try:
         response = requests.get('http://httpbin.org')
         return response.status_code == 200
     except Exception as e:
         print(e)
 
     return False
 
 
 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()
 
 
[e-mail](scope="session")
-def vcs_server_config_override():
-    return ({'server:main': {'workers': 2}},)
-
-
 @pytest.mark.usefixtures(
     "disable_locking", "disable_anonymous_user",
     "enable_webhook_push_integration")
 class TestVCSOperationsOnCustomIniConfig(object):
 
     def test_push_tag_with_commit_hg(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(HG_REPO)
         stdout, stderr = Command('/tmp').execute(
             'hg clone', clone_url, tmpdir.strpath)
 
         push_url = rc_web_server.repo_clone_url(HG_REPO)
         _add_files_and_push(
             'hg', tmpdir.strpath, clone_url=push_url,
             tags=[{'name': 'v1.0.0', 'commit': 'added tag v1.0.0'}])
 
         rc_log = rc_web_server.get_rc_log()
         assert 'ERROR' not in rc_log
         assert "'name': u'v1.0.0'" in rc_log
 
     def test_push_tag_with_commit_git(
             self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         push_url = rc_web_server.repo_clone_url(GIT_REPO)
         _add_files_and_push(
             'git', tmpdir.strpath, clone_url=push_url,
             tags=[{'name': 'v1.0.0', 'commit': 'added tag v1.0.0'}])
 
         rc_log = rc_web_server.get_rc_log()
         assert 'ERROR' not in rc_log
         assert "'name': u'v1.0.0'" in rc_log
 
     def test_push_tag_with_no_commit_git(
             self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(GIT_REPO)
         stdout, stderr = Command('/tmp').execute(
             'git clone', clone_url, tmpdir.strpath)
 
         push_url = rc_web_server.repo_clone_url(GIT_REPO)
         _add_files_and_push(
             'git', tmpdir.strpath, clone_url=push_url,
             tags=[{'name': 'v1.0.0', 'commit': 'added tag v1.0.0'}])
 
         rc_log = rc_web_server.get_rc_log()
         assert 'ERROR' not in rc_log
         assert "'name': u'v1.0.0'" in rc_log