python3: remove usage of subprocess32

rhodecode/subscribers.py

@@ -1,398 +1,398 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 io
 import shlex
 
 import math
 import re
 import os
 import datetime
 import logging
 import Queue
-import subprocess32
+import subprocess
 
 
 from dateutil.parser import parse
 from pyramid.threadlocal import get_current_request
 from pyramid.interfaces import IRoutesMapper
 from pyramid.settings import asbool
 from pyramid.path import AssetResolver
 from threading import Thread
 
 from rhodecode.config.jsroutes import generate_jsroutes_content
 from rhodecode.lib.base import get_auth_user
 
 import rhodecode
 
 
 log = logging.getLogger(__name__)
 
 
 def add_renderer_globals(event):
     from rhodecode.lib import helpers
 
     # TODO: When executed in pyramid view context the request is not available
     # in the event. Find a better solution to get the request.
     request = event['request'] or get_current_request()
 
     # Add Pyramid translation as '_' to context
     event['_'] = request.translate
     event['_ungettext'] = request.plularize
     event['h'] = helpers
 
 
 def set_user_lang(event):
     request = event.request
     cur_user = getattr(request, 'user', None)
 
     if cur_user:
         user_lang = cur_user.get_instance().user_data.get('language')
         if user_lang:
             log.debug('lang: setting current user:%s language to: %s', cur_user, user_lang)
             event.request._LOCALE_ = user_lang
 
 
 def update_celery_conf(event):
     from rhodecode.lib.celerylib.loader import set_celery_conf
     log.debug('Setting celery config from new request')
     set_celery_conf(request=event.request, registry=event.request.registry)
 
 
 def add_request_user_context(event):
     """
     Adds auth user into request context
     """
     request = event.request
     # access req_id as soon as possible
     req_id = request.req_id
 
     if hasattr(request, 'vcs_call'):
         # skip vcs calls
         return
 
     if hasattr(request, 'rpc_method'):
         # skip api calls
         return
 
     auth_user, auth_token = get_auth_user(request)
     request.user = auth_user
     request.user_auth_token = auth_token
     request.environ['rc_auth_user'] = auth_user
     request.environ['rc_auth_user_id'] = auth_user.user_id
     request.environ['rc_req_id'] = req_id
 
 
 def reset_log_bucket(event):
     """
     reset the log bucket on new request
     """
     request = event.request
     request.req_id_records_init()
 
 
 def scan_repositories_if_enabled(event):
     """
     This is subscribed to the `pyramid.events.ApplicationCreated` event. It
     does a repository scan if enabled in the settings.
     """
     settings = event.app.registry.settings
     vcs_server_enabled = settings['vcs.server.enable']
     import_on_startup = settings['startup.import_repos']
     if vcs_server_enabled and import_on_startup:
         from rhodecode.model.scm import ScmModel
         from rhodecode.lib.utils import repo2db_mapper, get_rhodecode_base_path
         repositories = ScmModel().repo_scan(get_rhodecode_base_path())
         repo2db_mapper(repositories, remove_obsolete=False)
 
 
 def write_metadata_if_needed(event):
     """
     Writes upgrade metadata
     """
     import rhodecode
     from rhodecode.lib import system_info
     from rhodecode.lib import ext_json
 
     fname = '.rcmetadata.json'
     ini_loc = os.path.dirname(rhodecode.CONFIG.get('__file__'))
     metadata_destination = os.path.join(ini_loc, fname)
 
     def get_update_age():
         now = datetime.datetime.utcnow()
 
         with open(metadata_destination, 'rb') as f:
             data = ext_json.json.loads(f.read())
             if 'created_on' in data:
                 update_date = parse(data['created_on'])
                 diff = now - update_date
                 return diff.total_seconds() / 60.0
 
         return 0
 
     def write():
         configuration = system_info.SysInfo(
             system_info.rhodecode_config)()['value']
         license_token = configuration['config']['license_token']
 
         setup = dict(
             workers=configuration['config']['server:main'].get(
                 'workers', '?'),
             worker_type=configuration['config']['server:main'].get(
                 'worker_class', 'sync'),
         )
         dbinfo = system_info.SysInfo(system_info.database_info)()['value']
         del dbinfo['url']
 
         metadata = dict(
             desc='upgrade metadata info',
             license_token=license_token,
             created_on=datetime.datetime.utcnow().isoformat(),
             usage=system_info.SysInfo(system_info.usage_info)()['value'],
             platform=system_info.SysInfo(system_info.platform_type)()['value'],
             database=dbinfo,
             cpu=system_info.SysInfo(system_info.cpu)()['value'],
             memory=system_info.SysInfo(system_info.memory)()['value'],
             setup=setup
         )
 
         with open(metadata_destination, 'wb') as f:
             f.write(ext_json.json.dumps(metadata))
 
     settings = event.app.registry.settings
     if settings.get('metadata.skip'):
         return
 
     # only write this every 24h, workers restart caused unwanted delays
     try:
         age_in_min = get_update_age()
     except Exception:
         age_in_min = 0
 
     if age_in_min > 60 * 60 * 24:
         return
 
     try:
         write()
     except Exception:
         pass
 
 
 def write_usage_data(event):
     import rhodecode
     from rhodecode.lib import system_info
     from rhodecode.lib import ext_json
 
     settings = event.app.registry.settings
     instance_tag = settings.get('metadata.write_usage_tag')
     if not settings.get('metadata.write_usage'):
         return
 
     def get_update_age(dest_file):
         now = datetime.datetime.utcnow()
 
         with open(dest_file, 'rb') as f:
             data = ext_json.json.loads(f.read())
             if 'created_on' in data:
                 update_date = parse(data['created_on'])
                 diff = now - update_date
                 return math.ceil(diff.total_seconds() / 60.0)
 
         return 0
 
     utc_date = datetime.datetime.utcnow()
     hour_quarter = int(math.ceil((utc_date.hour + utc_date.minute/60.0) / 6.))
     fname = '.rc_usage_{date.year}{date.month:02d}{date.day:02d}_{hour}.json'.format(
         date=utc_date, hour=hour_quarter)
     ini_loc = os.path.dirname(rhodecode.CONFIG.get('__file__'))
 
     usage_dir = os.path.join(ini_loc, '.rcusage')
     if not os.path.isdir(usage_dir):
         os.makedirs(usage_dir)
     usage_metadata_destination = os.path.join(usage_dir, fname)
 
     try:
         age_in_min = get_update_age(usage_metadata_destination)
     except Exception:
         age_in_min = 0
 
     # write every 6th hour
     if age_in_min and age_in_min < 60 * 6:
         log.debug('Usage file created %s minutes ago, skipping (threshold: %s minutes)...',
                   age_in_min, 60 * 6)
         return
 
     def write(dest_file):
         configuration = system_info.SysInfo(system_info.rhodecode_config)()['value']
         license_token = configuration['config']['license_token']
 
         metadata = dict(
             desc='Usage data',
             instance_tag=instance_tag,
             license_token=license_token,
             created_on=datetime.datetime.utcnow().isoformat(),
             usage=system_info.SysInfo(system_info.usage_info)()['value'],
         )
 
         with open(dest_file, 'wb') as f:
             f.write(ext_json.json.dumps(metadata, indent=2, sort_keys=True))
 
     try:
         log.debug('Writing usage file at: %s', usage_metadata_destination)
         write(usage_metadata_destination)
     except Exception:
         pass
 
 
 def write_js_routes_if_enabled(event):
     registry = event.app.registry
 
     mapper = registry.queryUtility(IRoutesMapper)
     _argument_prog = re.compile('\{(.*?)\}|:\((.*)\)')
 
     def _extract_route_information(route):
         """
         Convert a route into tuple(name, path, args), eg:
             ('show_user', '/profile/%(username)s', ['username'])
         """
 
         routepath = route.pattern
         pattern = route.pattern
 
         def replace(matchobj):
             if matchobj.group(1):
                 return "%%(%s)s" % matchobj.group(1).split(':')[0]
             else:
                 return "%%(%s)s" % matchobj.group(2)
 
         routepath = _argument_prog.sub(replace, routepath)
 
         if not routepath.startswith('/'):
             routepath = '/'+routepath
 
         return (
             route.name,
             routepath,
             [(arg[0].split(':')[0] if arg[0] != '' else arg[1])
               for arg in _argument_prog.findall(pattern)]
         )
 
     def get_routes():
         # pyramid routes
         for route in mapper.get_routes():
             if not route.name.startswith('__'):
                 yield _extract_route_information(route)
 
     if asbool(registry.settings.get('generate_js_files', 'false')):
         static_path = AssetResolver().resolve('rhodecode:public').abspath()
         jsroutes = get_routes()
         jsroutes_file_content = generate_jsroutes_content(jsroutes)
         jsroutes_file_path = os.path.join(
             static_path, 'js', 'rhodecode', 'routes.js')
 
         try:
             with io.open(jsroutes_file_path, 'w', encoding='utf-8') as f:
                 f.write(jsroutes_file_content)
         except Exception:
             log.exception('Failed to write routes.js into %s', jsroutes_file_path)
 
 
 class Subscriber(object):
     """
     Base class for subscribers to the pyramid event system.
     """
     def __call__(self, event):
         self.run(event)
 
     def run(self, event):
         raise NotImplementedError('Subclass has to implement this.')
 
 
 class AsyncSubscriber(Subscriber):
     """
     Subscriber that handles the execution of events in a separate task to not
     block the execution of the code which triggers the event. It puts the
     received events into a queue from which the worker process takes them in
     order.
     """
     def __init__(self):
         self._stop = False
         self._eventq = Queue.Queue()
         self._worker = self.create_worker()
         self._worker.start()
 
     def __call__(self, event):
         self._eventq.put(event)
 
     def create_worker(self):
         worker = Thread(target=self.do_work)
         worker.daemon = True
         return worker
 
     def stop_worker(self):
         self._stop = False
         self._eventq.put(None)
         self._worker.join()
 
     def do_work(self):
         while not self._stop:
             event = self._eventq.get()
             if event is not None:
                 self.run(event)
 
 
 class AsyncSubprocessSubscriber(AsyncSubscriber):
     """
-    Subscriber that uses the subprocess32 module to execute a command if an
+    Subscriber that uses the subprocess module to execute a command if an
     event is received. Events are handled asynchronously::
 
         subscriber = AsyncSubprocessSubscriber('ls -la', timeout=10)
         subscriber(dummyEvent) # running __call__(event)
 
     """
 
     def __init__(self, cmd, timeout=None):
         if not isinstance(cmd, (list, tuple)):
             cmd = shlex.split(cmd)
         super(AsyncSubprocessSubscriber, self).__init__()
         self._cmd = cmd
         self._timeout = timeout
 
     def run(self, event):
         cmd = self._cmd
         timeout = self._timeout
         log.debug('Executing command %s.', cmd)
 
         try:
-            output = subprocess32.check_output(
-                cmd, timeout=timeout, stderr=subprocess32.STDOUT)
+            output = subprocess.check_output(
+                cmd, timeout=timeout, stderr=subprocess.STDOUT)
             log.debug('Command finished %s', cmd)
             if output:
                 log.debug('Command output: %s', output)
-        except subprocess32.TimeoutExpired as e:
+        except subprocess.TimeoutExpired as e:
             log.exception('Timeout while executing command.')
             if e.output:
                 log.error('Command output: %s', e.output)
-        except subprocess32.CalledProcessError as e:
+        except subprocess.CalledProcessError as e:
             log.exception('Error while executing command.')
             if e.output:
                 log.error('Command output: %s', e.output)
         except Exception:
             log.exception(
                 'Exception while executing command %s.', cmd)

rhodecode/tests/database/conftest.py

@@ -1,293 +1,293 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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/
 
-from subprocess32 import Popen, PIPE
+from subprocess import Popen, PIPE
 import os
 import shutil
 import sys
 import tempfile
 
 import pytest
 from sqlalchemy.engine import url
 
 from rhodecode.tests.fixture import TestINI
 
 
 def _get_dbs_from_metafunc(metafunc):
     dbs_mark = metafunc.definition.get_closest_marker('dbs')
 
     if dbs_mark:
         # Supported backends by this test function, created from pytest.mark.dbs
         backends = dbs_mark.args
     else:
         backends = metafunc.config.getoption('--dbs')
     return backends
 
 
 def pytest_generate_tests(metafunc):
     # Support test generation based on --dbs parameter
     if 'db_backend' in metafunc.fixturenames:
         requested_backends = set(metafunc.config.getoption('--dbs'))
         backends = _get_dbs_from_metafunc(metafunc)
         backends = requested_backends.intersection(backends)
         # TODO: johbo: Disabling a backend did not work out with
         # parametrization, find better way to achieve this.
         if not backends:
             metafunc.function._skip = True
         metafunc.parametrize('db_backend_name', backends)
 
 
 def pytest_collection_modifyitems(session, config, items):
     remaining = [
         i for i in items if not getattr(i.obj, '_skip', False)]
     items[:] = remaining
 
 
 @pytest.fixture()
 def db_backend(
         request, db_backend_name, ini_config, tmpdir_factory):
     basetemp = tmpdir_factory.getbasetemp().strpath
     klass = _get_backend(db_backend_name)
 
     option_name = '--{}-connection-string'.format(db_backend_name)
     connection_string = request.config.getoption(option_name) or None
 
     return klass(
         config_file=ini_config, basetemp=basetemp,
         connection_string=connection_string)
 
 
 def _get_backend(backend_type):
     return {
         'sqlite': SQLiteDBBackend,
         'postgres': PostgresDBBackend,
         'mysql': MySQLDBBackend,
         '': EmptyDBBackend
     }[backend_type]
 
 
 class DBBackend(object):
     _store = os.path.dirname(os.path.abspath(__file__))
     _type = None
     _base_ini_config = [{'app:main': {'vcs.start_server': 'false',
                                       'startup.import_repos': 'false',
                                       'is_test': 'False'}}]
     _db_url = [{'app:main': {'sqlalchemy.db1.url': ''}}]
     _base_db_name = 'rhodecode_test_db_backend'
 
     def __init__(
             self, config_file, db_name=None, basetemp=None,
             connection_string=None):
 
         from rhodecode.lib.vcs.backends.hg import largefiles_store
         from rhodecode.lib.vcs.backends.git import lfs_store
 
         self.fixture_store = os.path.join(self._store, self._type)
         self.db_name = db_name or self._base_db_name
         self._base_ini_file = config_file
         self.stderr = ''
         self.stdout = ''
         self._basetemp = basetemp or tempfile.gettempdir()
         self._repos_location = os.path.join(self._basetemp, 'rc_test_repos')
         self._repos_hg_largefiles_store = largefiles_store(self._basetemp)
         self._repos_git_lfs_store = lfs_store(self._basetemp)
         self.connection_string = connection_string
 
     @property
     def connection_string(self):
         return self._connection_string
 
     @connection_string.setter
     def connection_string(self, new_connection_string):
         if not new_connection_string:
             new_connection_string = self.get_default_connection_string()
         else:
             new_connection_string = new_connection_string.format(
                 db_name=self.db_name)
         url_parts = url.make_url(new_connection_string)
         self._connection_string = new_connection_string
         self.user = url_parts.username
         self.password = url_parts.password
         self.host = url_parts.host
 
     def get_default_connection_string(self):
         raise NotImplementedError('default connection_string is required.')
 
     def execute(self, cmd, env=None, *args):
         """
         Runs command on the system with given ``args``.
         """
 
         command = cmd + ' ' + ' '.join(args)
         sys.stdout.write(command)
 
         # Tell Python to use UTF-8 encoding out stdout
         _env = os.environ.copy()
         _env['PYTHONIOENCODING'] = 'UTF-8'
         if env:
             _env.update(env)
         self.p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, env=_env)
         self.stdout, self.stderr = self.p.communicate()
         sys.stdout.write('COMMAND:'+command+'\n')
         sys.stdout.write(self.stdout)
         return self.stdout, self.stderr
 
     def assert_returncode_success(self):
         if not self.p.returncode == 0:
             print(self.stderr)
             raise AssertionError('non 0 retcode:{}'.format(self.p.returncode))
 
     def assert_correct_output(self, stdout, version):
         assert 'UPGRADE FOR STEP {} COMPLETED'.format(version) in stdout
 
     def setup_rhodecode_db(self, ini_params=None, env=None):
         if not ini_params:
             ini_params = self._base_ini_config
 
         ini_params.extend(self._db_url)
         with TestINI(self._base_ini_file, ini_params,
                      self._type, destroy=True) as _ini_file:
 
             if not os.path.isdir(self._repos_location):
                 os.makedirs(self._repos_location)
             if not os.path.isdir(self._repos_hg_largefiles_store):
                 os.makedirs(self._repos_hg_largefiles_store)
             if not os.path.isdir(self._repos_git_lfs_store):
                 os.makedirs(self._repos_git_lfs_store)
 
             return self.execute(
                 "rc-setup-app {0} --user=marcink "
                 "--email=[e-mail] --password={1} "
                 "--repos={2} --force-yes".format(
                     _ini_file, 'qweqwe', self._repos_location), env=env)
 
     def upgrade_database(self, ini_params=None):
         if not ini_params:
             ini_params = self._base_ini_config
         ini_params.extend(self._db_url)
 
         test_ini = TestINI(
             self._base_ini_file, ini_params, self._type, destroy=True)
         with test_ini as ini_file:
             if not os.path.isdir(self._repos_location):
                 os.makedirs(self._repos_location)
 
             return self.execute(
                 "rc-upgrade-db {0} --force-yes".format(ini_file))
 
     def setup_db(self):
         raise NotImplementedError
 
     def teardown_db(self):
         raise NotImplementedError
 
     def import_dump(self, dumpname):
         raise NotImplementedError
 
 
 class EmptyDBBackend(DBBackend):
     _type = ''
 
     def setup_db(self):
         pass
 
     def teardown_db(self):
         pass
 
     def import_dump(self, dumpname):
         pass
 
     def assert_returncode_success(self):
         assert True
 
 
 class SQLiteDBBackend(DBBackend):
     _type = 'sqlite'
 
     def get_default_connection_string(self):
         return 'sqlite:///{}/{}.sqlite'.format(self._basetemp, self.db_name)
 
     def setup_db(self):
         # dump schema for tests
         # cp -v $TEST_DB_NAME
         self._db_url = [{'app:main': {
             'sqlalchemy.db1.url': self.connection_string}}]
 
     def import_dump(self, dumpname):
         dump = os.path.join(self.fixture_store, dumpname)
         target = os.path.join(self._basetemp, '{0.db_name}.sqlite'.format(self))
         return self.execute('cp -v {} {}'.format(dump, target))
 
     def teardown_db(self):
         return self.execute("rm -rf {}.sqlite".format(
             os.path.join(self._basetemp, self.db_name)))
 
 
 class MySQLDBBackend(DBBackend):
     _type = 'mysql'
 
     def get_default_connection_string(self):
         return 'mysql://root:qweqwe@127.0.0.1/{}'.format(self.db_name)
 
     def setup_db(self):
         # dump schema for tests
         # mysqldump -uroot -pqweqwe $TEST_DB_NAME
         self._db_url = [{'app:main': {
             'sqlalchemy.db1.url': self.connection_string}}]
         return self.execute("mysql -v -u{} -p{} -e 'create database '{}';'".format(
             self.user, self.password, self.db_name))
 
     def import_dump(self, dumpname):
         dump = os.path.join(self.fixture_store, dumpname)
         return self.execute("mysql -u{} -p{} {} < {}".format(
             self.user, self.password, self.db_name, dump))
 
     def teardown_db(self):
         return self.execute("mysql -v -u{} -p{} -e 'drop database '{}';'".format(
             self.user, self.password, self.db_name))
 
 
 class PostgresDBBackend(DBBackend):
     _type = 'postgres'
 
     def get_default_connection_string(self):
         return 'postgresql://postgres:qweqwe@localhost/{}'.format(self.db_name)
 
     def setup_db(self):
         # dump schema for tests
         # pg_dump -U postgres -h localhost $TEST_DB_NAME
         self._db_url = [{'app:main': {
             'sqlalchemy.db1.url':
                 self.connection_string}}]
         return self.execute("PGPASSWORD={} psql -U {} -h localhost "
                      "-c 'create database '{}';'".format(
                          self.password, self.user, self.db_name))
 
     def teardown_db(self):
         return self.execute("PGPASSWORD={} psql -U {} -h localhost "
                      "-c 'drop database if exists '{}';'".format(
                          self.password, self.user, self.db_name))
 
     def import_dump(self, dumpname):
         dump = os.path.join(self.fixture_store, dumpname)
         return self.execute(
             "PGPASSWORD={} psql -U {} -h localhost -d {} -1 "
             "-f {}".format(
                 self.password, self.user, self.db_name, dump))

rhodecode/tests/lib/middleware/utils/test_scm_app_http_chunking.py

@@ -1,136 +1,136 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2020 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/
 
 """
 Checking the chunked data transfer via HTTP
 """
 
 import os
 import time
-import subprocess32
+import subprocess
 
 import pytest
 import requests
 
 from rhodecode.lib.middleware.utils import scm_app_http
 from rhodecode.tests.utils import wait_for_url
 
 
 def test_does_chunked_end_to_end_transfer(scm_app):
     response = requests.post(scm_app, data='', stream=True)
     assert response.headers['Transfer-Encoding'] == 'chunked'
     times = [time.time() for chunk in response.raw.read_chunked()]
     assert times[1] - times[0] > 0.1, "Chunks arrived at the same time"
 
 
 @pytest.fixture()
 def echo_app_chunking(request, available_port_factory):
     """
     Run the EchoApp via Waitress in a subprocess.
 
     Return the URL endpoint to reach the app.
     """
     port = available_port_factory()
     command = (
         'waitress-serve --send-bytes 1 --port {port} --call '
         'rhodecode.tests.lib.middleware.utils.test_scm_app_http_chunking'
         ':create_echo_app')
     command = command.format(port=port)
-    proc = subprocess32.Popen(command.split(' '), bufsize=0)
+    proc = subprocess.Popen(command.split(' '), bufsize=0)
     echo_app_url = 'http://localhost:' + str(port)
 
     @request.addfinalizer
     def stop_echo_app():
         proc.kill()
 
     return echo_app_url
 
 
 @pytest.fixture()
 def scm_app(request, available_port_factory, echo_app_chunking):
     """
     Run the scm_app in Waitress.
 
     Returns the URL endpoint where this app can be reached.
     """
     port = available_port_factory()
     command = (
         'waitress-serve --send-bytes 1 --port {port} --call '
         'rhodecode.tests.lib.middleware.utils.test_scm_app_http_chunking'
         ':create_scm_app')
     command = command.format(port=port)
     env = os.environ.copy()
     env["RC_ECHO_URL"] = echo_app_chunking
-    proc = subprocess32.Popen(command.split(' '), bufsize=0, env=env)
+    proc = subprocess.Popen(command.split(' '), bufsize=0, env=env)
     scm_app_url = 'http://localhost:' + str(port)
     wait_for_url(scm_app_url)
 
     @request.addfinalizer
     def stop_echo_app():
         proc.kill()
 
     return scm_app_url
 
 
 class EchoApp(object):
     """
     Stub WSGI application which returns a chunked response to every request.
     """
 
     def __init__(self, repo_path, repo_name, config):
         self._repo_path = repo_path
 
     def __call__(self, environ, start_response):
         environ['wsgi.input'].read()
         status = '200 OK'
         headers = []
         start_response(status, headers)
         return result_generator()
 
 
 def result_generator():
     """
     Simulate chunked results.
 
     The intended usage is to simulate a chunked response as we would get it
     out of a vcs operation during a call to "hg clone".
     """
     yield 'waiting 2 seconds'
     # Wait long enough so that the first chunk can go out
     time.sleep(2)
     yield 'final chunk'
     # Another small wait, otherwise they go together
     time.sleep(0.1)
 
 
 def create_echo_app():
     """
     Create EchoApp filled with stub data.
     """
     return EchoApp('stub_path', 'repo_name', {})
 
 
 def create_scm_app():
     """
     Create a scm_app hooked up to speak to EchoApp.
     """
     echo_app_url = os.environ["RC_ECHO_URL"]
     return scm_app_http.VcsHttpProxy(
         echo_app_url, 'stub_path', 'stub_name', None)

rhodecode/tests/load/http_performance.py

@@ -1,463 +1,463 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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/
 
 """
 Module to test the performance of pull, push and clone operations.
 
 It works by replaying a group of commits to the repo.
 """
 
 import argparse
 import collections
 import ConfigParser
 import functools
 import itertools
 import os
 import pprint
 import shutil
-import subprocess32
+import subprocess
 import sys
 import time
 
 import api
 
 
 def mean(container):
     """Return the mean of the container."""
     if not container:
         return -1.0
     return sum(container) / len(container)
 
 
 def keep_cwd(f):
     """Decorator that keeps track of the starting working directory."""
     @functools.wraps(f)
     def wrapped_f(*args, **kwargs):
         cur_dir = os.getcwd()
         try:
             return f(*args, **kwargs)
         finally:
             os.chdir(cur_dir)
 
     return wrapped_f
 
 
 def timed(f):
     """Decorator that returns the time it took to execute the function."""
     @functools.wraps(f)
     def wrapped_f(*args, **kwargs):
         start_time = time.time()
         try:
             f(*args, **kwargs)
         finally:
             return time.time() - start_time
 
     return wrapped_f
 
 
 def execute(*popenargs, **kwargs):
     """Extension of subprocess.check_output to support writing to stdin."""
     input = kwargs.pop('stdin', None)
     stdin = None
     if input:
-        stdin = subprocess32.PIPE
+        stdin = subprocess.PIPE
     #if 'stderr' not in kwargs:
-    #    kwargs['stderr'] = subprocess32.PIPE
+    #    kwargs['stderr'] = subprocess.PIPE
     if 'stdout' in kwargs:
         raise ValueError('stdout argument not allowed, it will be overridden.')
-    process = subprocess32.Popen(stdin=stdin, stdout=subprocess32.PIPE,
+    process = subprocess.Popen(stdin=stdin, stdout=subprocess.PIPE,
                                *popenargs, **kwargs)
     output, error = process.communicate(input=input)
     retcode = process.poll()
     if retcode:
         cmd = kwargs.get("args")
         if cmd is None:
             cmd = popenargs[0]
         print('{} {} {} '.format(cmd, output, error))
-        raise subprocess32.CalledProcessError(retcode, cmd, output=output)
+        raise subprocess.CalledProcessError(retcode, cmd, output=output)
     return output
 
 
 def get_repo_name(repo_url):
     """Extract the repo name from its url."""
     repo_url = repo_url.rstrip('/')
     return repo_url.split('/')[-1].split('.')[0]
 
 
 class TestPerformanceBase(object):
     def __init__(self, base_dir, repo_url, n_commits, max_commits,
                  skip_commits):
         self.repo_url = repo_url
         self.repo_name = get_repo_name(self.repo_url)
         self.upstream_repo_name = '%s_upstream' % self.repo_name
         self.base_dir = os.path.abspath(base_dir)
         self.n_commits = n_commits
         self.max_commits = max_commits
         self.skip_commits = skip_commits
         self.push_times = []
         self.pull_times = []
         self.empty_pull_times = []
         self.clone_time = -1.0
         self.last_commit = None
 
         self.cloned_repo = ''
         self.pull_repo = ''
         self.orig_repo = ''
 
     def run(self):
         try:
             self.test()
         except Exception as error:
             print(error)
         finally:
             self.cleanup()
 
             print('Clone time     :{}'.format(self.clone_time))
             print('Push time      :{}'.format(mean(self.push_times)))
             print('Pull time      :{}'.format(mean(self.pull_times)))
             print('Empty pull time:{}'.format(mean(self.empty_pull_times)))
 
             return {
                 'clone': self.clone_time,
                 'push': mean(self.push_times),
                 'pull': mean(self.pull_times),
                 'empty_pull': mean(self.empty_pull_times),
             }
 
     @keep_cwd
     def test(self):
         os.chdir(self.base_dir)
 
         self.orig_repo = os.path.join(self.base_dir, self.repo_name)
         if not os.path.exists(self.orig_repo):
             self.clone_repo(self.repo_url, default_only=True)
 
         upstream_url = self.create_repo(self.upstream_repo_name, self.repo_type)
 
         self.add_remote(self.orig_repo, upstream_url)
 
         self.pull_repo = os.path.join(self.base_dir, '%s_pull' % self.repo_name)
         self.clone_repo(upstream_url, self.pull_repo)
 
         commits = self.get_commits(self.orig_repo)
         self.last_commit = commits[-1]
         if self.skip_commits:
             self.push(
                 self.orig_repo, commits[self.skip_commits - 1], 'upstream')
         commits = commits[self.skip_commits:self.max_commits]
 
         print('Working with %d commits' % len(commits))
         for i in range(self.n_commits - 1, len(commits), self.n_commits):
             commit = commits[i]
             print('Processing commit %s (%d)' % (commit, i + 1))
             self.push_times.append(
                 self.push(self.orig_repo, commit, 'upstream'))
             self.check_remote_last_commit_is(commit, upstream_url)
 
             self.pull_times.append(self.pull(self.pull_repo))
             self.check_local_last_commit_is(commit, self.pull_repo)
 
             self.empty_pull_times.append(self.pull(self.pull_repo))
 
         self.cloned_repo = os.path.join(self.base_dir,
                                         '%s_clone' % self.repo_name)
         self.clone_time = self.clone_repo(upstream_url, self.cloned_repo)
 
     def cleanup(self):
         try:
             self.delete_repo(self.upstream_repo_name)
         except api.ApiError:
             # Continue in case we could not delete the repo. Maybe we did not
             # create it in the first place.
             pass
 
         shutil.rmtree(self.pull_repo, ignore_errors=True)
         shutil.rmtree(self.cloned_repo, ignore_errors=True)
 
         if os.path.exists(self.orig_repo):
             self.remove_remote(self.orig_repo)
 
 
 class RhodeCodeMixin(object):
     """Mixin providing the methods to create and delete repos in RhodeCode."""
     def __init__(self, api_key):
         self.api = api.RCApi(api_key=api_key)
 
     def create_repo(self, repo_name, repo_type):
         return self.api.create_repo(repo_name, repo_type,
                                     'Repo for perfomance testing')
 
     def delete_repo(self, repo_name):
         return self.api.delete_repo(repo_name)
 
 
 class GitMixin(object):
     """Mixin providing the git operations."""
     @timed
     def clone_repo(self, repo_url, destination=None, default_only=False):
         args = ['git', 'clone']
         if default_only:
             args.extend(['--branch', 'master', '--single-branch'])
         args.append(repo_url)
         if destination:
             args.append(destination)
         execute(args)
 
     @keep_cwd
     def add_remote(self, repo, remote_url, remote_name='upstream'):
         self.remove_remote(repo, remote_name)
         os.chdir(repo)
         execute(['git', 'remote', 'add', remote_name, remote_url])
 
     @keep_cwd
     def remove_remote(self, repo, remote_name='upstream'):
         os.chdir(repo)
         remotes = execute(['git', 'remote']).split('\n')
         if remote_name in remotes:
             execute(['git', 'remote', 'remove', remote_name])
 
     @keep_cwd
     def get_commits(self, repo, branch='master'):
         os.chdir(repo)
         commits_list = execute(
             ['git', 'log', '--first-parent', branch, '--pretty=%H'])
         return commits_list.strip().split('\n')[::-1]
 
     @timed
     def push(self, repo, commit, remote_name=None):
         os.chdir(repo)
         try:
             execute(['git', 'reset', '--soft', commit])
             args = ['git', 'push']
             if remote_name:
                 args.append(remote_name)
             execute(args)
         finally:
             execute(['git', 'reset', '--soft', 'HEAD@{1}'])
 
     @timed
     def pull(self, repo):
         os.chdir(repo)
         execute(['git', 'pull'])
 
     def _remote_last_commit(self, repo_url):
         output = execute(['git', 'ls-remote', repo_url, 'HEAD'])
         return output.split()[0]
 
     def check_remote_last_commit_is(self, commit, repo_url):
         last_remote_commit = self._remote_last_commit(repo_url)
         if last_remote_commit != commit:
             raise Exception('Push did not work, expected commit %s but got %s' %
                             (commit, last_remote_commit))
 
     @keep_cwd
     def _local_last_commit(self, repo):
         os.chdir(repo)
         return execute(['git', 'rev-parse', 'HEAD']).strip()
 
     def check_local_last_commit_is(self, commit, repo):
         last_local_commit = self._local_last_commit(repo)
         if last_local_commit != commit:
             raise Exception('Pull did not work, expected commit %s but got %s' %
                             (commit, last_local_commit))
 
 
 class HgMixin(object):
     """Mixin providing the mercurial operations."""
     @timed
     def clone_repo(self, repo_url, destination=None, default_only=False):
         args = ['hg', 'clone']
         if default_only:
             args.extend(['--branch', 'default'])
         args.append(repo_url)
         if destination:
             args.append(destination)
         execute(args)
 
     @keep_cwd
     def add_remote(self, repo, remote_url, remote_name='upstream'):
         self.remove_remote(repo, remote_name)
         os.chdir(repo)
         hgrc = ConfigParser.RawConfigParser()
         hgrc.read('.hg/hgrc')
         hgrc.set('paths', remote_name, remote_url)
         with open('.hg/hgrc', 'w') as f:
             hgrc.write(f)
 
     @keep_cwd
     def remove_remote(self, repo, remote_name='upstream'):
         os.chdir(repo)
         hgrc = ConfigParser.RawConfigParser()
         hgrc.read('.hg/hgrc')
         hgrc.remove_option('paths', remote_name)
         with open('.hg/hgrc', 'w') as f:
             hgrc.write(f)
 
     @keep_cwd
     def get_commits(self, repo, branch='default'):
         os.chdir(repo)
         # See http://stackoverflow.com/questions/15376649/is-there-a-mercurial-equivalent-to-git-log-first-parent
         commits_list = execute(['hg', 'log', '--branch', branch, '--template',
                                 '{node}\n', '--follow-first'])
         return commits_list.strip().split('\n')[::-1]
 
     @timed
     def push(self, repo, commit, remote_name=None):
         os.chdir(repo)
         args = ['hg', 'push', '--rev', commit, '--new-branch']
         if remote_name:
             args.append(remote_name)
         execute(args)
 
     @timed
     def pull(self, repo):
         os.chdir(repo)
         execute(['hg', '--config', 'alias.pull=pull', 'pull', '-u'])
 
     def _remote_last_commit(self, repo_url):
         return execute(['hg', 'identify', repo_url])[:12]
 
     def check_remote_last_commit_is(self, commit, repo_url):
         last_remote_commit = self._remote_last_commit(repo_url)
         if not commit.startswith(last_remote_commit):
             raise Exception('Push did not work, expected commit %s but got %s' %
                             (commit, last_remote_commit))
 
     @keep_cwd
     def _local_last_commit(self, repo):
         os.chdir(repo)
         return execute(['hg', 'identify'])[:12]
 
     def check_local_last_commit_is(self, commit, repo):
         last_local_commit = self._local_last_commit(repo)
         if not commit.startswith(last_local_commit):
             raise Exception('Pull did not work, expected commit %s but got %s' %
                             (commit, last_local_commit))
 
 
 class GitTestPerformance(GitMixin, RhodeCodeMixin, TestPerformanceBase):
     def __init__(self, base_dir, repo_url, n_commits, max_commits, skip_commits,
                  api_key):
         TestPerformanceBase.__init__(self, base_dir, repo_url, n_commits,
                                      max_commits, skip_commits)
         RhodeCodeMixin.__init__(self, api_key)
         self.repo_type = 'git'
 
 
 class HgTestPerformance(HgMixin, RhodeCodeMixin, TestPerformanceBase):
     def __init__(self, base_dir, repo_url, n_commits, max_commits, skip_commits,
                  api_key):
         TestPerformanceBase.__init__(self, base_dir, repo_url, n_commits,
                                      max_commits, skip_commits)
         RhodeCodeMixin.__init__(self, api_key)
         self.repo_type = 'hg'
 
 
 def get_test(base_dir, repo_url, repo_type, step, max_commits, skip_commits,
              api_key):
     max_commits = min(10 * step,
                       int((max_commits - skip_commits) / step) * step)
     max_commits += skip_commits
     if repo_type == 'git':
         return GitTestPerformance(
             base_dir, repo_url, step, max_commits, skip_commits, api_key)
     elif repo_type == 'hg':
         return HgTestPerformance(
             base_dir, repo_url, step, max_commits, skip_commits, api_key)
 
 
 def main(argv):
     parser = argparse.ArgumentParser(
         description='Performance tests for push/pull/clone for git and ' +
                     'mercurial repos.')
     parser.add_argument(
         '--tests', dest='tests', action='store', required=False, default='all',
         help='The tests to run. Default: all. But could be any comma ' +
              'separated list with python, hg, kernel or git')
     parser.add_argument(
         '--sizes', dest='sizes', action='store', required=False,
         default='1,10,100,1000,2500',
         help='The sizes to use. Default: 1,10,100,1000,2500')
     parser.add_argument(
         '--dir', dest='dir', action='store', required=True,
         help='The dir where to store the repos')
     parser.add_argument(
         '--api-key', dest='api_key', action='store', required=True,
         help='The api key of RhodeCode')
     options = parser.parse_args(argv[1:])
     print(options)
 
     test_config = {
         'python': {
             'url': 'https://hg.python.org/cpython/',
             'limit': 23322,
             'type': 'hg',
             # Do not time the first commit, as it is HUGE!
             'skip': 1,
         },
         'hg': {
             'url': 'http://selenic.com/hg',
             'limit': 14396,
             'type': 'hg',
         },
         'kernel': {
             'url': 'https://github.com/torvalds/linux.git',
             'limit': 46271,
             'type': 'git',
         },
         'git': {
             'url': 'https://github.com/git/git.git',
             'limit': 13525,
             'type': 'git',
         }
 
     }
 
     test_names = options.tests.split(',')
     if test_names == ['all']:
         test_names = test_config.keys()
     if not set(test_names) <= set(test_config.keys()):
         print('Invalid tests: only %s are valid but specified %s' %
               (test_config.keys(), test_names))
         return 1
 
     sizes = options.sizes.split(',')
     sizes = map(int, sizes)
 
     base_dir = options.dir
     api_key = options.api_key
     results = collections.defaultdict(dict)
     for test_name, size in itertools.product(test_names, sizes):
         test = get_test(base_dir,
                         test_config[test_name]['url'],
                         test_config[test_name]['type'],
                         size,
                         test_config[test_name]['limit'],
                         test_config[test_name].get('skip', 0),
                         api_key)
         print('*' * 80)
         print('Running performance test: %s with size %d' % (test_name, size))
         print('*' * 80)
         results[test_name][size] = test.run()
     pprint.pprint(dict(results))
 
 
 if __name__ == '__main__':
     sys.exit(main(sys.argv))

rhodecode/tests/load/profile.py

@@ -1,155 +1,155 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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/
 
 """
 This is a standalone script which will start VCS and RC.
 
 Performance numbers will be written on each interval to:
     vcs_profileX.csv
     rc_profileX.csv
 
 To stop the script by press Ctrl-C
 """
 
 import datetime
 import os
 import psutil
-import subprocess32
+import subprocess
 import sys
 import time
 import traceback
 import urllib.request, urllib.parse, urllib.error
 
 PROFILING_INTERVAL = 5
 RC_WEBSITE = "http://localhost:5001/"
 
 
 def get_file(prefix):
     out_file = None
     for i in range(100):
         file_path = "%s_profile%.3d.csv" % (prefix, i)
         if os.path.exists(file_path):
             continue
         out_file = open(file_path, "w")
         out_file.write("Time; CPU %; Memory (MB); Total FDs; Dulwich FDs; Threads\n")
         break
     return out_file
 
 
 def dump_system():
     print("System Overview...")
     print("\nCPU Count: %d (%d real)" %
           (psutil.cpu_count(), psutil.cpu_count(logical=False)))
     print("\nDisk:")
     print(psutil.disk_usage(os.sep))
     print("\nMemory:")
     print(psutil.virtual_memory())
     print("\nMemory (swap):")
     print(psutil.swap_memory())
 
 
 def count_dulwich_fds(proc):
-    p = subprocess32.Popen(["lsof", "-p", proc.pid], stdout=subprocess32.PIPE)
+    p = subprocess.Popen(["lsof", "-p", proc.pid], stdout=subprocess.PIPE)
     out, err = p.communicate()
 
     count = 0
     for line in out.splitlines():
         content = line.split()
         # http://git-scm.com/book/en/Git-Internals-Packfiles
         if content[-1].endswith(".idx"):
             count += 1
 
     return count
 
 def dump_process(pid, out_file):
     now = datetime.datetime.now()
     cpu = pid.cpu_percent()
     mem = pid.memory_info()
     fds = pid.num_fds()
     dulwich_fds = count_dulwich_fds(pid)
     threads = pid.num_threads()
 
     content = [now.strftime('%m/%d/%y %H:%M:%S'),
                cpu,
                "%.2f" % (mem[0]/1024.0/1024.0),
                fds, dulwich_fds, threads]
     out_file.write("; ".join([str(item) for item in content]))
     out_file.write("\n")
 
 
 # Open output files
 vcs_out = get_file("vcs")
 if vcs_out is None:
     print("Unable to enumerate output file for VCS")
     sys.exit(1)
 rc_out = get_file("rc")
 if rc_out is None:
     print("Unable to enumerate output file for RC")
     sys.exit(1)
 
 # Show system information
 dump_system()
 
 print("\nStarting VCS...")
 vcs = psutil.Popen(["vcsserver"])
 time.sleep(1)
 if not vcs.is_running():
     print("VCS - Failed to start")
     sys.exit(1)
 print("VCS - Ok")
 
 print("\nStarting RhodeCode...")
 rc = psutil.Popen("RC_VCSSERVER_TEST_DISABLE=1 paster serve test.ini",
-                  shell=True, stdin=subprocess32.PIPE)
+                  shell=True, stdin=subprocess.PIPE)
 time.sleep(1)
 if not rc.is_running():
     print("RC - Failed to start")
     vcs.terminate()
     sys.exit(1)
 
 # Send command to create the databases
 rc.stdin.write("y\n")
 
 # Verify that the website is up
 time.sleep(4)
 try:
     urllib.request.urlopen(RC_WEBSITE)
 except IOError:
     print("RC - Website not started")
     vcs.terminate()
     sys.exit(1)
 print("RC - Ok")
 
 print("\nProfiling...\n%s\n" % ("-"*80))
 while True:
     try:
         dump_process(vcs, vcs_out)
         dump_process(rc, rc_out)
         time.sleep(PROFILING_INTERVAL)
     except Exception:
         print(traceback.format_exc())
         break
 
 # Finalize the profiling
 vcs_out.close()
 rc_out.close()
 
 vcs.terminate()
 rc.terminate()

rhodecode/tests/load/vcs_performance.py

@@ -1,306 +1,306 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2020 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/
 
 """
 VCS Performance measurement tool
 
 Usage:
 
 - Check that required vcs keys can be found in ~/.hgrc and ~/.netrc
 
 - Start a local instance of RhodeCode Enterprise
 
 - Launch the script:
 
     TMPDIR=/tmp python vcs_performance.py \
         --host=http://vm:5000 \
         --api-key=55c4a33688577da24183dcac5fde4dddfdbf18dc \
         --commits=10 --repositories=100 --log-level=info
 """
 
 import argparse
 import functools
 import logging
 import os
 import shutil
-import subprocess32
+import subprocess
 import tempfile
 import time
 from itertools import chain
 
 from api import RCApi, ApiError
 
 
 log = logging.getLogger(__name__)
 
 
 def timed(f):
     """Decorator that returns the time it took to execute the function."""
     @functools.wraps(f)
     def wrapped_f(*args, **kwargs):
         start_time = time.time()
         try:
             f(*args, **kwargs)
         finally:
             return time.time() - start_time
 
     return wrapped_f
 
 
 def mean(container):
     """Return the mean of the container."""
     if not container:
         return -1.0
     return sum(container) / len(container)
 
 
 class Config(object):
     args = None
 
     def __init__(self):
         parser = argparse.ArgumentParser(description='Runs VCS load tests')
         parser.add_argument(
             '--host', dest='host', action='store', required=True,
             help='RhodeCode Enterprise host')
         parser.add_argument(
             '--api-key', dest='api_key', action='store', required=True,
             help='API Key')
         parser.add_argument(
             '--file-size', dest='file_size', action='store', required=False,
             default=1, type=int, help='File size in MB')
         parser.add_argument(
             '--repositories', dest='repositories', action='store',
             required=False, default=1, type=int,
             help='Number of repositories')
         parser.add_argument(
             '--commits', dest='commits', action='store', required=False,
             default=1, type=int, help='Number of commits')
         parser.add_argument(
             '--log-level', dest='log_level', action='store', required=False,
             default='error', help='Logging level')
         self.args = parser.parse_args()
 
     def __getattr__(self, attr):
         return getattr(self.args, attr)
 
 
 class Repository(object):
     FILE_NAME_TEMPLATE = "test_{:09d}.bin"
 
     def __init__(self, name, base_path, api):
         self.name = name
         self.path = os.path.join(base_path, name)
         self.api = api
         self.url = None
 
     def create(self):
         self._create_filesystem_repo(self.path)
         try:
             self.url = self.api.create_repo(self.name, self.TYPE, 'Performance tests')
         except ApiError as e:
             log.error('api: {}'.format(e))
 
     def delete(self):
         self._delete_filesystem_repo()
         try:
             self.api.delete_repo(self.name)
         except ApiError as e:
             log.error('api: {}'.format(e))
 
     def create_commits(self, number, file_size):
         for i in range(number):
             file_name = self.FILE_NAME_TEMPLATE.format(i)
             log.debug("Create commit[{}] {}".format(self.name, file_name))
             self._create_file(file_name, file_size)
             self._create_commit(file_name)
 
     @timed
     def push(self):
         raise NotImplementedError()
 
     @timed
     def clone(self, destination_path):
         raise NotImplementedError()
 
     @timed
     def pull(self):
         raise NotImplementedError()
 
     def _run(self, *args):
         command = [self.BASE_COMMAND] + list(args)
-        process = subprocess32.Popen(
-            command, stdout=subprocess32.PIPE, stderr=subprocess32.PIPE)
+        process = subprocess.Popen(
+            command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         return process.communicate()
 
     def _create_file(self, name, size):
         file_name = os.path.join(self.path, name)
         with open(file_name, 'wb') as f:
             f.write(os.urandom(1024))
 
     def _delete_filesystem_repo(self):
         shutil.rmtree(self.path)
 
     def _create_filesystem_repo(self, path):
         raise NotImplementedError()
 
     def _create_commit(self, file_name):
         raise NotImplementedError()
 
 
 class GitRepository(Repository):
     TYPE = 'git'
     BASE_COMMAND = 'git'
 
     @timed
     def push(self):
         os.chdir(self.path)
         self._run('push', '--set-upstream', self.url, 'master')
 
     @timed
     def clone(self, destination_path):
         self._run('clone', self.url, os.path.join(destination_path, self.name))
 
     @timed
     def pull(self, destination_path):
         path = os.path.join(destination_path, self.name)
         self._create_filesystem_repo(path)
         os.chdir(path)
         self._run('remote', 'add', 'origin', self.url)
         self._run('pull', 'origin', 'master')
 
     def _create_filesystem_repo(self, path):
         self._run('init', path)
 
     def _create_commit(self, file_name):
         os.chdir(self.path)
         self._run('add', file_name)
         self._run('commit', file_name, '-m', '"Add {}"'.format(file_name))
 
 
 class HgRepository(Repository):
     TYPE = 'hg'
     BASE_COMMAND = 'hg'
 
     @timed
     def push(self):
         os.chdir(self.path)
         self._run('push', self.url)
 
     @timed
     def clone(self, destination_path):
         self._run('clone', self.url, os.path.join(destination_path, self.name))
 
     @timed
     def pull(self, destination_path):
         path = os.path.join(destination_path, self.name)
         self._create_filesystem_repo(path)
         os.chdir(path)
         self._run('pull', '-r', 'tip', self.url)
 
     def _create_filesystem_repo(self, path):
         self._run('init', path)
 
     def _create_commit(self, file_name):
         os.chdir(self.path)
         self._run('add', file_name)
         self._run('commit', file_name, '-m', '"Add {}"'.format(file_name))
 
 
 class Benchmark(object):
     REPO_CLASSES = {
         'git': GitRepository,
         'hg': HgRepository
     }
     REPO_NAME = '{}_performance_{:03d}'
 
     def __init__(self, config):
         self.api = RCApi(api_key=config.api_key, rc_endpoint=config.host)
         self.source_path = tempfile.mkdtemp(suffix='vcsperformance')
 
         self.config = config
         self.git_repos = []
         self.hg_repos = []
 
         self._set_log_level()
 
     def start(self):
         self._create_repos()
         repos = {
             'git': self.git_repos,
             'hg': self.hg_repos
         }
 
         clone_destination_path = tempfile.mkdtemp(suffix='clone')
         pull_destination_path = tempfile.mkdtemp(suffix='pull')
         operations = [
             ('push', ),
             ('clone', clone_destination_path),
             ('pull', pull_destination_path)
         ]
 
         for operation in operations:
             for type_ in repos:
                 times = self._measure(repos[type_], *operation)
                 print("Mean[of {}] {:5s} {:5s} time: {:.3f} sec.".format(
                     len(times), type_, operation[0], mean(times)))
 
     def cleanup(self):
         log.info("Cleaning up...")
         for repo in chain(self.git_repos, self.hg_repos):
             repo.delete()
 
     def _measure(self, repos, operation, *args):
         times = []
         for repo in repos:
             method = getattr(repo, operation)
             times.append(method(*args))
         return times
 
     def _create_repos(self):
         log.info("Creating repositories...")
         for i in range(self.config.repositories):
             self.git_repos.append(self._create_repo('git', i))
             self.hg_repos.append(self._create_repo('hg', i))
 
     def _create_repo(self, type_, id_):
         RepoClass = self.REPO_CLASSES[type_]
         repo = RepoClass(
             self.REPO_NAME.format(type_, id_), self.source_path, self.api)
         repo.create()
         repo.create_commits(self.config.commits, self.config.file_size)
         return repo
 
     def _set_log_level(self):
         try:
             log_level = getattr(logging, config.log_level.upper())
         except:
             log_level = logging.ERROR
         handler = logging.StreamHandler()
         log.addHandler(handler)
         log.setLevel(log_level)
 
 
 if __name__ == '__main__':
     config = Config()
     benchmark = Benchmark(config)
     try:
         benchmark.start()
     finally:
         benchmark.cleanup()

rhodecode/tests/plugin.py

@@ -1,1842 +1,1842 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 collections
 import datetime
 import hashlib
 import os
 import re
 import pprint
 import shutil
 import socket
-import subprocess32
+import subprocess
 import time
 import uuid
 import dateutil.tz
 import logging
 
 import mock
 import pyramid.testing
 import pytest
 import colander
 import requests
 import pyramid.paster
 
 import rhodecode
 from rhodecode.lib.utils2 import AttributeDict
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import (
     PullRequest, PullRequestReviewers, Repository, RhodeCodeSetting, ChangesetStatus,
     RepoGroup, UserGroup, RepoRhodeCodeUi, RepoRhodeCodeSetting, RhodeCodeUi)
 from rhodecode.model.meta import Session
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.settings import VcsSettingsModel
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.model.integration import IntegrationModel
 from rhodecode.integrations import integration_type_registry
 from rhodecode.integrations.types.base import IntegrationTypeBase
 from rhodecode.lib.utils import repo2db_mapper
 from rhodecode.lib.vcs.backends import get_backend
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests import (
     login_user_session, get_new_dir, utils, TESTS_TMP_PATH,
     TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR2_LOGIN,
     TEST_USER_REGULAR_PASS)
 from rhodecode.tests.utils import CustomTestApp, set_anonymous_access
 from rhodecode.tests.fixture import Fixture
 from rhodecode.config import utils as config_utils
 
 log = logging.getLogger(__name__)
 
 def _split_comma(value):
     return value.split(',')
 
 
 def pytest_addoption(parser):
     parser.addoption(
         '--keep-tmp-path', action='store_true',
         help="Keep the test temporary directories")
     parser.addoption(
         '--backends', action='store', type=_split_comma,
         default=['git', 'hg', 'svn'],
         help="Select which backends to test for backend specific tests.")
     parser.addoption(
         '--dbs', action='store', type=_split_comma,
         default=['sqlite'],
         help="Select which database to test for database specific tests. "
              "Possible options are sqlite,postgres,mysql")
     parser.addoption(
         '--appenlight', '--ae', action='store_true',
         help="Track statistics in appenlight.")
     parser.addoption(
         '--appenlight-api-key', '--ae-key',
         help="API key for Appenlight.")
     parser.addoption(
         '--appenlight-url', '--ae-url',
         default="https://ae.rhodecode.com",
         help="Appenlight service URL, defaults to https://ae.rhodecode.com")
     parser.addoption(
         '--sqlite-connection-string', action='store',
         default='', help="Connection string for the dbs tests with SQLite")
     parser.addoption(
         '--postgres-connection-string', action='store',
         default='', help="Connection string for the dbs tests with Postgres")
     parser.addoption(
         '--mysql-connection-string', action='store',
         default='', help="Connection string for the dbs tests with MySQL")
     parser.addoption(
         '--repeat', type=int, default=100,
         help="Number of repetitions in performance tests.")
 
 
 def pytest_configure(config):
     from rhodecode.config import patches
 
 
 def pytest_collection_modifyitems(session, config, items):
     # nottest marked, compare nose, used for transition from nose to pytest
     remaining = [
         i for i in items if getattr(i.obj, '__test__', True)]
     items[:] = remaining
 
     # NOTE(marcink): custom test ordering, db tests and vcstests are slowes and should
     # be executed at the end for faster test feedback
     def sorter(item):
         pos = 0
         key = item._nodeid
         if key.startswith('rhodecode/tests/database'):
             pos = 1
         elif key.startswith('rhodecode/tests/vcs_operations'):
             pos = 2
 
         return pos
 
     items.sort(key=sorter)
 
 
 def pytest_generate_tests(metafunc):
 
     # Support test generation based on --backend parameter
     if 'backend_alias' in metafunc.fixturenames:
         backends = get_backends_from_metafunc(metafunc)
         scope = None
         if not backends:
             pytest.skip("Not enabled for any of selected backends")
 
         metafunc.parametrize('backend_alias', backends, scope=scope)
 
     backend_mark = metafunc.definition.get_closest_marker('backends')
     if backend_mark:
         backends = get_backends_from_metafunc(metafunc)
         if not backends:
             pytest.skip("Not enabled for any of selected backends")
 
 
 def get_backends_from_metafunc(metafunc):
     requested_backends = set(metafunc.config.getoption('--backends'))
     backend_mark = metafunc.definition.get_closest_marker('backends')
     if backend_mark:
         # Supported backends by this test function, created from
         # pytest.mark.backends
         backends = backend_mark.args
     elif hasattr(metafunc.cls, 'backend_alias'):
         # Support class attribute "backend_alias", this is mainly
         # for legacy reasons for tests not yet using pytest.mark.backends
         backends = [metafunc.cls.backend_alias]
     else:
         backends = metafunc.config.getoption('--backends')
     return requested_backends.intersection(backends)
 
 
 @pytest.fixture(scope='session', autouse=True)
 def activate_example_rcextensions(request):
     """
     Patch in an example rcextensions module which verifies passed in kwargs.
     """
     from rhodecode.config import rcextensions
 
     old_extensions = rhodecode.EXTENSIONS
     rhodecode.EXTENSIONS = rcextensions
     rhodecode.EXTENSIONS.calls = collections.defaultdict(list)
 
     @request.addfinalizer
     def cleanup():
         rhodecode.EXTENSIONS = old_extensions
 
 
 @pytest.fixture()
 def capture_rcextensions():
     """
     Returns the recorded calls to entry points in rcextensions.
     """
     calls = rhodecode.EXTENSIONS.calls
     calls.clear()
     # Note: At this moment, it is still the empty dict, but that will
     # be filled during the test run and since it is a reference this
     # is enough to make it work.
     return calls
 
 
 @pytest.fixture(scope='session')
 def http_environ_session():
     """
     Allow to use "http_environ" in session scope.
     """
     return plain_http_environ()
 
 
 def plain_http_host_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return 'example.com:80'
 
 
 @pytest.fixture()
 def http_host_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return plain_http_host_stub()
 
 
 def plain_http_host_only_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return plain_http_host_stub().split(':')[0]
 
 
 @pytest.fixture()
 def http_host_only_stub():
     """
     Value of HTTP_HOST in the test run.
     """
     return plain_http_host_only_stub()
 
 
 def plain_http_environ():
     """
     HTTP extra environ keys.
 
     User by the test application and as well for setting up the pylons
     environment. In the case of the fixture "app" it should be possible
     to override this for a specific test case.
     """
     return {
         'SERVER_NAME': plain_http_host_only_stub(),
         'SERVER_PORT': plain_http_host_stub().split(':')[1],
         'HTTP_HOST': plain_http_host_stub(),
         'HTTP_USER_AGENT': 'rc-test-agent',
         'REQUEST_METHOD': 'GET'
     }
 
 
 @pytest.fixture()
 def http_environ():
     """
     HTTP extra environ keys.
 
     User by the test application and as well for setting up the pylons
     environment. In the case of the fixture "app" it should be possible
     to override this for a specific test case.
     """
     return plain_http_environ()
 
 
 @pytest.fixture(scope='session')
 def baseapp(ini_config, vcsserver, http_environ_session):
     from rhodecode.lib.pyramid_utils import get_app_config
     from rhodecode.config.middleware import make_pyramid_app
 
     log.info("Using the RhodeCode configuration:{}".format(ini_config))
     pyramid.paster.setup_logging(ini_config)
 
     settings = get_app_config(ini_config)
     app = make_pyramid_app({'__file__': ini_config}, **settings)
 
     return app
 
 
 @pytest.fixture(scope='function')
 def app(request, config_stub, baseapp, http_environ):
     app = CustomTestApp(
         baseapp,
         extra_environ=http_environ)
     if request.cls:
         request.cls.app = app
     return app
 
 
 @pytest.fixture(scope='session')
 def app_settings(baseapp, ini_config):
     """
     Settings dictionary used to create the app.
 
     Parses the ini file and passes the result through the sanitize and apply
     defaults mechanism in `rhodecode.config.middleware`.
     """
     return baseapp.config.get_settings()
 
 
 @pytest.fixture(scope='session')
 def db_connection(ini_settings):
     # Initialize the database connection.
     config_utils.initialize_database(ini_settings)
 
 
 LoginData = collections.namedtuple('LoginData', ('csrf_token', 'user'))
 
 
 def _autologin_user(app, *args):
     session = login_user_session(app, *args)
     csrf_token = rhodecode.lib.auth.get_csrf_token(session)
     return LoginData(csrf_token, session['rhodecode_user'])
 
 
 @pytest.fixture()
 def autologin_user(app):
     """
     Utility fixture which makes sure that the admin user is logged in
     """
     return _autologin_user(app)
 
 
 @pytest.fixture()
 def autologin_regular_user(app):
     """
     Utility fixture which makes sure that the regular user is logged in
     """
     return _autologin_user(
         app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 
 
 @pytest.fixture(scope='function')
 def csrf_token(request, autologin_user):
     return autologin_user.csrf_token
 
 
 @pytest.fixture(scope='function')
 def xhr_header(request):
     return {'HTTP_X_REQUESTED_WITH': 'XMLHttpRequest'}
 
 
 @pytest.fixture()
 def real_crypto_backend(monkeypatch):
     """
     Switch the production crypto backend on for this test.
 
     During the test run the crypto backend is replaced with a faster
     implementation based on the MD5 algorithm.
     """
     monkeypatch.setattr(rhodecode, 'is_test', False)
 
 
 @pytest.fixture(scope='class')
 def index_location(request, baseapp):
     index_location = baseapp.config.get_settings()['search.location']
     if request.cls:
         request.cls.index_location = index_location
     return index_location
 
 
 @pytest.fixture(scope='session', autouse=True)
 def tests_tmp_path(request):
     """
     Create temporary directory to be used during the test session.
     """
     if not os.path.exists(TESTS_TMP_PATH):
         os.makedirs(TESTS_TMP_PATH)
 
     if not request.config.getoption('--keep-tmp-path'):
         @request.addfinalizer
         def remove_tmp_path():
             shutil.rmtree(TESTS_TMP_PATH)
 
     return TESTS_TMP_PATH
 
 
 @pytest.fixture()
 def test_repo_group(request):
     """
     Create a temporary repository group, and destroy it after
     usage automatically
     """
     fixture = Fixture()
     repogroupid = 'test_repo_group_%s' % str(time.time()).replace('.', '')
     repo_group = fixture.create_repo_group(repogroupid)
 
     def _cleanup():
         fixture.destroy_repo_group(repogroupid)
 
     request.addfinalizer(_cleanup)
     return repo_group
 
 
 @pytest.fixture()
 def test_user_group(request):
     """
     Create a temporary user group, and destroy it after
     usage automatically
     """
     fixture = Fixture()
     usergroupid = 'test_user_group_%s' % str(time.time()).replace('.', '')
     user_group = fixture.create_user_group(usergroupid)
 
     def _cleanup():
         fixture.destroy_user_group(user_group)
 
     request.addfinalizer(_cleanup)
     return user_group
 
 
 @pytest.fixture(scope='session')
 def test_repo(request):
     container = TestRepoContainer()
     request.addfinalizer(container._cleanup)
     return container
 
 
 class TestRepoContainer(object):
     """
     Container for test repositories which are used read only.
 
     Repositories will be created on demand and re-used during the lifetime
     of this object.
 
     Usage to get the svn test repository "minimal"::
 
        test_repo = TestContainer()
        repo = test_repo('minimal', 'svn')
 
     """
 
     dump_extractors = {
         'git': utils.extract_git_repo_from_dump,
         'hg': utils.extract_hg_repo_from_dump,
         'svn': utils.extract_svn_repo_from_dump,
     }
 
     def __init__(self):
         self._cleanup_repos = []
         self._fixture = Fixture()
         self._repos = {}
 
     def __call__(self, dump_name, backend_alias, config=None):
         key = (dump_name, backend_alias)
         if key not in self._repos:
             repo = self._create_repo(dump_name, backend_alias, config)
             self._repos[key] = repo.repo_id
         return Repository.get(self._repos[key])
 
     def _create_repo(self, dump_name, backend_alias, config):
         repo_name = '%s-%s' % (backend_alias, dump_name)
         backend = get_backend(backend_alias)
         dump_extractor = self.dump_extractors[backend_alias]
         repo_path = dump_extractor(dump_name, repo_name)
 
         vcs_repo = backend(repo_path, config=config)
         repo2db_mapper({repo_name: vcs_repo})
 
         repo = RepoModel().get_by_repo_name(repo_name)
         self._cleanup_repos.append(repo_name)
         return repo
 
     def _cleanup(self):
         for repo_name in reversed(self._cleanup_repos):
             self._fixture.destroy_repo(repo_name)
 
 
 def backend_base(request, backend_alias, baseapp, test_repo):
     if backend_alias not in request.config.getoption('--backends'):
         pytest.skip("Backend %s not selected." % (backend_alias, ))
 
     utils.check_xfail_backends(request.node, backend_alias)
     utils.check_skip_backends(request.node, backend_alias)
 
     repo_name = 'vcs_test_%s' % (backend_alias, )
     backend = Backend(
         alias=backend_alias,
         repo_name=repo_name,
         test_name=request.node.name,
         test_repo_container=test_repo)
     request.addfinalizer(backend.cleanup)
     return backend
 
 
 @pytest.fixture()
 def backend(request, backend_alias, baseapp, test_repo):
     """
     Parametrized fixture which represents a single backend implementation.
 
     It respects the option `--backends` to focus the test run on specific
     backend implementations.
 
     It also supports `pytest.mark.xfail_backends` to mark tests as failing
     for specific backends. This is intended as a utility for incremental
     development of a new backend implementation.
     """
     return backend_base(request, backend_alias, baseapp, test_repo)
 
 
 @pytest.fixture()
 def backend_git(request, baseapp, test_repo):
     return backend_base(request, 'git', baseapp, test_repo)
 
 
 @pytest.fixture()
 def backend_hg(request, baseapp, test_repo):
     return backend_base(request, 'hg', baseapp, test_repo)
 
 
 @pytest.fixture()
 def backend_svn(request, baseapp, test_repo):
     return backend_base(request, 'svn', baseapp, test_repo)
 
 
 @pytest.fixture()
 def backend_random(backend_git):
     """
     Use this to express that your tests need "a backend.
 
     A few of our tests need a backend, so that we can run the code. This
     fixture is intended to be used for such cases. It will pick one of the
     backends and run the tests.
 
     The fixture `backend` would run the test multiple times for each
     available backend which is a pure waste of time if the test is
     independent of the backend type.
     """
     # TODO: johbo: Change this to pick a random backend
     return backend_git
 
 
 @pytest.fixture()
 def backend_stub(backend_git):
     """
     Use this to express that your tests need a backend stub
 
     TODO: mikhail: Implement a real stub logic instead of returning
     a git backend
     """
     return backend_git
 
 
 @pytest.fixture()
 def repo_stub(backend_stub):
     """
     Use this to express that your tests need a repository stub
     """
     return backend_stub.create_repo()
 
 
 class Backend(object):
     """
     Represents the test configuration for one supported backend
 
     Provides easy access to different test repositories based on
     `__getitem__`. Such repositories will only be created once per test
     session.
     """
 
     invalid_repo_name = re.compile(r'[^0-9a-zA-Z]+')
     _master_repo = None
     _master_repo_path = ''
     _commit_ids = {}
 
     def __init__(self, alias, repo_name, test_name, test_repo_container):
         self.alias = alias
         self.repo_name = repo_name
         self._cleanup_repos = []
         self._test_name = test_name
         self._test_repo_container = test_repo_container
         # TODO: johbo: Used as a delegate interim. Not yet sure if Backend or
         # Fixture will survive in the end.
         self._fixture = Fixture()
 
     def __getitem__(self, key):
         return self._test_repo_container(key, self.alias)
 
     def create_test_repo(self, key, config=None):
         return self._test_repo_container(key, self.alias, config)
 
     @property
     def repo(self):
         """
         Returns the "current" repository. This is the vcs_test repo or the
         last repo which has been created with `create_repo`.
         """
         from rhodecode.model.db import Repository
         return Repository.get_by_repo_name(self.repo_name)
 
     @property
     def default_branch_name(self):
         VcsRepository = get_backend(self.alias)
         return VcsRepository.DEFAULT_BRANCH_NAME
 
     @property
     def default_head_id(self):
         """
         Returns the default head id of the underlying backend.
 
         This will be the default branch name in case the backend does have a
         default branch. In the other cases it will point to a valid head
         which can serve as the base to create a new commit on top of it.
         """
         vcsrepo = self.repo.scm_instance()
         head_id = (
             vcsrepo.DEFAULT_BRANCH_NAME or
             vcsrepo.commit_ids[-1])
         return head_id
 
     @property
     def commit_ids(self):
         """
         Returns the list of commits for the last created repository
         """
         return self._commit_ids
 
     def create_master_repo(self, commits):
         """
         Create a repository and remember it as a template.
 
         This allows to easily create derived repositories to construct
         more complex scenarios for diff, compare and pull requests.
 
         Returns a commit map which maps from commit message to raw_id.
         """
         self._master_repo = self.create_repo(commits=commits)
         self._master_repo_path = self._master_repo.repo_full_path
 
         return self._commit_ids
 
     def create_repo(
             self, commits=None, number_of_commits=0, heads=None,
             name_suffix=u'', bare=False, **kwargs):
         """
         Create a repository and record it for later cleanup.
 
         :param commits: Optional. A sequence of dict instances.
            Will add a commit per entry to the new repository.
         :param number_of_commits: Optional. If set to a number, this number of
            commits will be added to the new repository.
         :param heads: Optional. Can be set to a sequence of of commit
            names which shall be pulled in from the master repository.
         :param name_suffix: adds special suffix to generated repo name
         :param bare: set a repo as bare (no checkout)
         """
         self.repo_name = self._next_repo_name() + name_suffix
         repo = self._fixture.create_repo(
             self.repo_name, repo_type=self.alias, bare=bare, **kwargs)
         self._cleanup_repos.append(repo.repo_name)
 
         commits = commits or [
             {'message': 'Commit %s of %s' % (x, self.repo_name)}
             for x in range(number_of_commits)]
         vcs_repo = repo.scm_instance()
         vcs_repo.count()
         self._add_commits_to_repo(vcs_repo, commits)
         if heads:
             self.pull_heads(repo, heads)
 
         return repo
 
     def pull_heads(self, repo, heads):
         """
         Make sure that repo contains all commits mentioned in `heads`
         """
         vcsrepo = repo.scm_instance()
         vcsrepo.config.clear_section('hooks')
         commit_ids = [self._commit_ids[h] for h in heads]
         vcsrepo.pull(self._master_repo_path, commit_ids=commit_ids)
 
     def create_fork(self):
         repo_to_fork = self.repo_name
         self.repo_name = self._next_repo_name()
         repo = self._fixture.create_fork(repo_to_fork, self.repo_name)
         self._cleanup_repos.append(self.repo_name)
         return repo
 
     def new_repo_name(self, suffix=u''):
         self.repo_name = self._next_repo_name() + suffix
         self._cleanup_repos.append(self.repo_name)
         return self.repo_name
 
     def _next_repo_name(self):
         return u"%s_%s" % (
             self.invalid_repo_name.sub(u'_', self._test_name), len(self._cleanup_repos))
 
     def ensure_file(self, filename, content='Test content\n'):
         assert self._cleanup_repos, "Avoid writing into vcs_test repos"
         commits = [
             {'added': [
                 FileNode(filename, content=content),
             ]},
         ]
         self._add_commits_to_repo(self.repo.scm_instance(), commits)
 
     def enable_downloads(self):
         repo = self.repo
         repo.enable_downloads = True
         Session().add(repo)
         Session().commit()
 
     def cleanup(self):
         for repo_name in reversed(self._cleanup_repos):
             self._fixture.destroy_repo(repo_name)
 
     def _add_commits_to_repo(self, repo, commits):
         commit_ids = _add_commits_to_repo(repo, commits)
         if not commit_ids:
             return
         self._commit_ids = commit_ids
 
         # Creating refs for Git to allow fetching them from remote repository
         if self.alias == 'git':
             refs = {}
             for message in self._commit_ids:
                 # TODO: mikhail: do more special chars replacements
                 ref_name = 'refs/test-refs/{}'.format(
                     message.replace(' ', ''))
                 refs[ref_name] = self._commit_ids[message]
             self._create_refs(repo, refs)
 
     def _create_refs(self, repo, refs):
         for ref_name in refs:
             repo.set_refs(ref_name, refs[ref_name])
 
 
 def vcsbackend_base(request, backend_alias, tests_tmp_path, baseapp, test_repo):
     if backend_alias not in request.config.getoption('--backends'):
         pytest.skip("Backend %s not selected." % (backend_alias, ))
 
     utils.check_xfail_backends(request.node, backend_alias)
     utils.check_skip_backends(request.node, backend_alias)
 
     repo_name = 'vcs_test_%s' % (backend_alias, )
     repo_path = os.path.join(tests_tmp_path, repo_name)
     backend = VcsBackend(
         alias=backend_alias,
         repo_path=repo_path,
         test_name=request.node.name,
         test_repo_container=test_repo)
     request.addfinalizer(backend.cleanup)
     return backend
 
 
 @pytest.fixture()
 def vcsbackend(request, backend_alias, tests_tmp_path, baseapp, test_repo):
     """
     Parametrized fixture which represents a single vcs backend implementation.
 
     See the fixture `backend` for more details. This one implements the same
     concept, but on vcs level. So it does not provide model instances etc.
 
     Parameters are generated dynamically, see :func:`pytest_generate_tests`
     for how this works.
     """
     return vcsbackend_base(request, backend_alias, tests_tmp_path, baseapp, test_repo)
 
 
 @pytest.fixture()
 def vcsbackend_git(request, tests_tmp_path, baseapp, test_repo):
     return vcsbackend_base(request, 'git', tests_tmp_path, baseapp, test_repo)
 
 
 @pytest.fixture()
 def vcsbackend_hg(request, tests_tmp_path, baseapp, test_repo):
     return vcsbackend_base(request, 'hg', tests_tmp_path, baseapp, test_repo)
 
 
 @pytest.fixture()
 def vcsbackend_svn(request, tests_tmp_path, baseapp, test_repo):
     return vcsbackend_base(request, 'svn', tests_tmp_path, baseapp, test_repo)
 
 
 @pytest.fixture()
 def vcsbackend_stub(vcsbackend_git):
     """
     Use this to express that your test just needs a stub of a vcsbackend.
 
     Plan is to eventually implement an in-memory stub to speed tests up.
     """
     return vcsbackend_git
 
 
 class VcsBackend(object):
     """
     Represents the test configuration for one supported vcs backend.
     """
 
     invalid_repo_name = re.compile(r'[^0-9a-zA-Z]+')
 
     def __init__(self, alias, repo_path, test_name, test_repo_container):
         self.alias = alias
         self._repo_path = repo_path
         self._cleanup_repos = []
         self._test_name = test_name
         self._test_repo_container = test_repo_container
 
     def __getitem__(self, key):
         return self._test_repo_container(key, self.alias).scm_instance()
 
     @property
     def repo(self):
         """
         Returns the "current" repository. This is the vcs_test repo of the last
         repo which has been created.
         """
         Repository = get_backend(self.alias)
         return Repository(self._repo_path)
 
     @property
     def backend(self):
         """
         Returns the backend implementation class.
         """
         return get_backend(self.alias)
 
     def create_repo(self, commits=None, number_of_commits=0, _clone_repo=None,
                     bare=False):
         repo_name = self._next_repo_name()
         self._repo_path = get_new_dir(repo_name)
         repo_class = get_backend(self.alias)
         src_url = None
         if _clone_repo:
             src_url = _clone_repo.path
         repo = repo_class(self._repo_path, create=True, src_url=src_url, bare=bare)
         self._cleanup_repos.append(repo)
 
         commits = commits or [
             {'message': 'Commit %s of %s' % (x, repo_name)}
             for x in range(number_of_commits)]
         _add_commits_to_repo(repo, commits)
         return repo
 
     def clone_repo(self, repo):
         return self.create_repo(_clone_repo=repo)
 
     def cleanup(self):
         for repo in self._cleanup_repos:
             shutil.rmtree(repo.path)
 
     def new_repo_path(self):
         repo_name = self._next_repo_name()
         self._repo_path = get_new_dir(repo_name)
         return self._repo_path
 
     def _next_repo_name(self):
         return "%s_%s" % (
             self.invalid_repo_name.sub('_', self._test_name),
             len(self._cleanup_repos))
 
     def add_file(self, repo, filename, content='Test content\n'):
         imc = repo.in_memory_commit
         imc.add(FileNode(filename, content=content))
         imc.commit(
             message=u'Automatic commit from vcsbackend fixture',
             author=u'Automatic <[e-mail]>')
 
     def ensure_file(self, filename, content='Test content\n'):
         assert self._cleanup_repos, "Avoid writing into vcs_test repos"
         self.add_file(self.repo, filename, content)
 
 
 def _add_commits_to_repo(vcs_repo, commits):
     commit_ids = {}
     if not commits:
         return commit_ids
 
     imc = vcs_repo.in_memory_commit
     commit = None
 
     for idx, commit in enumerate(commits):
         message = unicode(commit.get('message', 'Commit %s' % idx))
 
         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))
 
         parents = [
             vcs_repo.get_commit(commit_id=commit_ids[p])
             for p in commit.get('parents', [])]
 
         operations = ('added', 'changed', 'removed')
         if not any((commit.get(o) for o in operations)):
             imc.add(FileNode('file_%s' % idx, content=message))
 
         commit = imc.commit(
             message=message,
             author=unicode(commit.get('author', 'Automatic <[e-mail]>')),
             date=commit.get('date'),
             branch=commit.get('branch'),
             parents=parents)
 
         commit_ids[commit.message] = commit.raw_id
 
     return commit_ids
 
 
 @pytest.fixture()
 def reposerver(request):
     """
     Allows to serve a backend repository
     """
 
     repo_server = RepoServer()
     request.addfinalizer(repo_server.cleanup)
     return repo_server
 
 
 class RepoServer(object):
     """
     Utility to serve a local repository for the duration of a test case.
 
     Supports only Subversion so far.
     """
 
     url = None
 
     def __init__(self):
         self._cleanup_servers = []
 
     def serve(self, vcsrepo):
         if vcsrepo.alias != 'svn':
             raise TypeError("Backend %s not supported" % vcsrepo.alias)
 
-        proc = subprocess32.Popen(
+        proc = subprocess.Popen(
             ['svnserve', '-d', '--foreground', '--listen-host', 'localhost',
              '--root', vcsrepo.path])
         self._cleanup_servers.append(proc)
         self.url = 'svn://localhost'
 
     def cleanup(self):
         for proc in self._cleanup_servers:
             proc.terminate()
 
 
 @pytest.fixture()
 def pr_util(backend, request, config_stub):
     """
     Utility for tests of models and for functional tests around pull requests.
 
     It gives an instance of :class:`PRTestUtility` which provides various
     utility methods around one pull request.
 
     This fixture uses `backend` and inherits its parameterization.
     """
 
     util = PRTestUtility(backend)
     request.addfinalizer(util.cleanup)
 
     return util
 
 
 class PRTestUtility(object):
 
     pull_request = None
     pull_request_id = None
     mergeable_patcher = None
     mergeable_mock = None
     notification_patcher = None
 
     def __init__(self, backend):
         self.backend = backend
 
     def create_pull_request(
             self, commits=None, target_head=None, source_head=None,
             revisions=None, approved=False, author=None, mergeable=False,
             enable_notifications=True, name_suffix=u'', reviewers=None, observers=None,
             title=u"Test", description=u"Description"):
         self.set_mergeable(mergeable)
         if not enable_notifications:
             # mock notification side effect
             self.notification_patcher = mock.patch(
                 'rhodecode.model.notification.NotificationModel.create')
             self.notification_patcher.start()
 
         if not self.pull_request:
             if not commits:
                 commits = [
                     {'message': 'c1'},
                     {'message': 'c2'},
                     {'message': 'c3'},
                 ]
                 target_head = 'c1'
                 source_head = 'c2'
                 revisions = ['c2']
 
             self.commit_ids = self.backend.create_master_repo(commits)
             self.target_repository = self.backend.create_repo(
                 heads=[target_head], name_suffix=name_suffix)
             self.source_repository = self.backend.create_repo(
                 heads=[source_head], name_suffix=name_suffix)
             self.author = author or UserModel().get_by_username(
                 TEST_USER_ADMIN_LOGIN)
 
             model = PullRequestModel()
             self.create_parameters = {
                 'created_by': self.author,
                 'source_repo': self.source_repository.repo_name,
                 'source_ref': self._default_branch_reference(source_head),
                 'target_repo': self.target_repository.repo_name,
                 'target_ref': self._default_branch_reference(target_head),
                 'revisions': [self.commit_ids[r] for r in revisions],
                 'reviewers': reviewers or self._get_reviewers(),
                 'observers': observers or self._get_observers(),
                 'title': title,
                 'description': description,
             }
             self.pull_request = model.create(**self.create_parameters)
             assert model.get_versions(self.pull_request) == []
 
             self.pull_request_id = self.pull_request.pull_request_id
 
             if approved:
                 self.approve()
 
             Session().add(self.pull_request)
             Session().commit()
 
         return self.pull_request
 
     def approve(self):
         self.create_status_votes(
             ChangesetStatus.STATUS_APPROVED,
             *self.pull_request.reviewers)
 
     def close(self):
         PullRequestModel().close_pull_request(self.pull_request, self.author)
 
     def _default_branch_reference(self, commit_message):
         reference = '%s:%s:%s' % (
             'branch',
             self.backend.default_branch_name,
             self.commit_ids[commit_message])
         return reference
 
     def _get_reviewers(self):
         role = PullRequestReviewers.ROLE_REVIEWER
         return [
             (TEST_USER_REGULAR_LOGIN, ['default1'], False, role, []),
             (TEST_USER_REGULAR2_LOGIN, ['default2'], False, role, []),
         ]
 
     def _get_observers(self):
         return [
 
         ]
 
     def update_source_repository(self, head=None):
         heads = [head or 'c3']
         self.backend.pull_heads(self.source_repository, heads=heads)
 
     def add_one_commit(self, head=None):
         self.update_source_repository(head=head)
         old_commit_ids = set(self.pull_request.revisions)
         PullRequestModel().update_commits(self.pull_request, self.pull_request.author)
         commit_ids = set(self.pull_request.revisions)
         new_commit_ids = commit_ids - old_commit_ids
         assert len(new_commit_ids) == 1
         return new_commit_ids.pop()
 
     def remove_one_commit(self):
         assert len(self.pull_request.revisions) == 2
         source_vcs = self.source_repository.scm_instance()
         removed_commit_id = source_vcs.commit_ids[-1]
 
         # TODO: johbo: Git and Mercurial have an inconsistent vcs api here,
         # remove the if once that's sorted out.
         if self.backend.alias == "git":
             kwargs = {'branch_name': self.backend.default_branch_name}
         else:
             kwargs = {}
         source_vcs.strip(removed_commit_id, **kwargs)
 
         PullRequestModel().update_commits(self.pull_request, self.pull_request.author)
         assert len(self.pull_request.revisions) == 1
         return removed_commit_id
 
     def create_comment(self, linked_to=None):
         comment = CommentsModel().create(
             text=u"Test comment",
             repo=self.target_repository.repo_name,
             user=self.author,
             pull_request=self.pull_request)
         assert comment.pull_request_version_id is None
 
         if linked_to:
             PullRequestModel()._link_comments_to_version(linked_to)
 
         return comment
 
     def create_inline_comment(
             self, linked_to=None, line_no=u'n1', file_path='file_1'):
         comment = CommentsModel().create(
             text=u"Test comment",
             repo=self.target_repository.repo_name,
             user=self.author,
             line_no=line_no,
             f_path=file_path,
             pull_request=self.pull_request)
         assert comment.pull_request_version_id is None
 
         if linked_to:
             PullRequestModel()._link_comments_to_version(linked_to)
 
         return comment
 
     def create_version_of_pull_request(self):
         pull_request = self.create_pull_request()
         version = PullRequestModel()._create_version_from_snapshot(
             pull_request)
         return version
 
     def create_status_votes(self, status, *reviewers):
         for reviewer in reviewers:
             ChangesetStatusModel().set_status(
                 repo=self.pull_request.target_repo,
                 status=status,
                 user=reviewer.user_id,
                 pull_request=self.pull_request)
 
     def set_mergeable(self, value):
         if not self.mergeable_patcher:
             self.mergeable_patcher = mock.patch.object(
                 VcsSettingsModel, 'get_general_settings')
             self.mergeable_mock = self.mergeable_patcher.start()
         self.mergeable_mock.return_value = {
             'rhodecode_pr_merge_enabled': value}
 
     def cleanup(self):
         # In case the source repository is already cleaned up, the pull
         # request will already be deleted.
         pull_request = PullRequest().get(self.pull_request_id)
         if pull_request:
             PullRequestModel().delete(pull_request, pull_request.author)
             Session().commit()
 
         if self.notification_patcher:
             self.notification_patcher.stop()
 
         if self.mergeable_patcher:
             self.mergeable_patcher.stop()
 
 
 @pytest.fixture()
 def user_admin(baseapp):
     """
     Provides the default admin test user as an instance of `db.User`.
     """
     user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
     return user
 
 
 @pytest.fixture()
 def user_regular(baseapp):
     """
     Provides the default regular test user as an instance of `db.User`.
     """
     user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
     return user
 
 
 @pytest.fixture()
 def user_util(request, db_connection):
     """
     Provides a wired instance of `UserUtility` with integrated cleanup.
     """
     utility = UserUtility(test_name=request.node.name)
     request.addfinalizer(utility.cleanup)
     return utility
 
 
 # TODO: johbo: Split this up into utilities per domain or something similar
 class UserUtility(object):
 
     def __init__(self, test_name="test"):
         self._test_name = self._sanitize_name(test_name)
         self.fixture = Fixture()
         self.repo_group_ids = []
         self.repos_ids = []
         self.user_ids = []
         self.user_group_ids = []
         self.user_repo_permission_ids = []
         self.user_group_repo_permission_ids = []
         self.user_repo_group_permission_ids = []
         self.user_group_repo_group_permission_ids = []
         self.user_user_group_permission_ids = []
         self.user_group_user_group_permission_ids = []
         self.user_permissions = []
 
     def _sanitize_name(self, name):
         for char in ['[', ']']:
             name = name.replace(char, '_')
         return name
 
     def create_repo_group(
             self, owner=TEST_USER_ADMIN_LOGIN, auto_cleanup=True):
         group_name = "{prefix}_repogroup_{count}".format(
             prefix=self._test_name,
             count=len(self.repo_group_ids))
         repo_group = self.fixture.create_repo_group(
             group_name, cur_user=owner)
         if auto_cleanup:
             self.repo_group_ids.append(repo_group.group_id)
         return repo_group
 
     def create_repo(self, owner=TEST_USER_ADMIN_LOGIN, parent=None,
                     auto_cleanup=True, repo_type='hg', bare=False):
         repo_name = "{prefix}_repository_{count}".format(
             prefix=self._test_name,
             count=len(self.repos_ids))
 
         repository = self.fixture.create_repo(
             repo_name, cur_user=owner, repo_group=parent, repo_type=repo_type, bare=bare)
         if auto_cleanup:
             self.repos_ids.append(repository.repo_id)
         return repository
 
     def create_user(self, auto_cleanup=True, **kwargs):
         user_name = "{prefix}_user_{count}".format(
             prefix=self._test_name,
             count=len(self.user_ids))
         user = self.fixture.create_user(user_name, **kwargs)
         if auto_cleanup:
             self.user_ids.append(user.user_id)
         return user
 
     def create_additional_user_email(self, user, email):
         uem = self.fixture.create_additional_user_email(user=user, email=email)
         return uem
 
     def create_user_with_group(self):
         user = self.create_user()
         user_group = self.create_user_group(members=[user])
         return user, user_group
 
     def create_user_group(self, owner=TEST_USER_ADMIN_LOGIN, members=None,
                           auto_cleanup=True, **kwargs):
         group_name = "{prefix}_usergroup_{count}".format(
             prefix=self._test_name,
             count=len(self.user_group_ids))
         user_group = self.fixture.create_user_group(
             group_name, cur_user=owner, **kwargs)
 
         if auto_cleanup:
             self.user_group_ids.append(user_group.users_group_id)
         if members:
             for user in members:
                 UserGroupModel().add_user_to_group(user_group, user)
         return user_group
 
     def grant_user_permission(self, user_name, permission_name):
         self.inherit_default_user_permissions(user_name, False)
         self.user_permissions.append((user_name, permission_name))
 
     def grant_user_permission_to_repo_group(
             self, repo_group, user, permission_name):
         permission = RepoGroupModel().grant_user_permission(
             repo_group, user, permission_name)
         self.user_repo_group_permission_ids.append(
             (repo_group.group_id, user.user_id))
         return permission
 
     def grant_user_group_permission_to_repo_group(
             self, repo_group, user_group, permission_name):
         permission = RepoGroupModel().grant_user_group_permission(
             repo_group, user_group, permission_name)
         self.user_group_repo_group_permission_ids.append(
             (repo_group.group_id, user_group.users_group_id))
         return permission
 
     def grant_user_permission_to_repo(
             self, repo, user, permission_name):
         permission = RepoModel().grant_user_permission(
             repo, user, permission_name)
         self.user_repo_permission_ids.append(
             (repo.repo_id, user.user_id))
         return permission
 
     def grant_user_group_permission_to_repo(
             self, repo, user_group, permission_name):
         permission = RepoModel().grant_user_group_permission(
             repo, user_group, permission_name)
         self.user_group_repo_permission_ids.append(
             (repo.repo_id, user_group.users_group_id))
         return permission
 
     def grant_user_permission_to_user_group(
             self, target_user_group, user, permission_name):
         permission = UserGroupModel().grant_user_permission(
             target_user_group, user, permission_name)
         self.user_user_group_permission_ids.append(
             (target_user_group.users_group_id, user.user_id))
         return permission
 
     def grant_user_group_permission_to_user_group(
             self, target_user_group, user_group, permission_name):
         permission = UserGroupModel().grant_user_group_permission(
             target_user_group, user_group, permission_name)
         self.user_group_user_group_permission_ids.append(
             (target_user_group.users_group_id, user_group.users_group_id))
         return permission
 
     def revoke_user_permission(self, user_name, permission_name):
         self.inherit_default_user_permissions(user_name, True)
         UserModel().revoke_perm(user_name, permission_name)
 
     def inherit_default_user_permissions(self, user_name, value):
         user = UserModel().get_by_username(user_name)
         user.inherit_default_permissions = value
         Session().add(user)
         Session().commit()
 
     def cleanup(self):
         self._cleanup_permissions()
         self._cleanup_repos()
         self._cleanup_repo_groups()
         self._cleanup_user_groups()
         self._cleanup_users()
 
     def _cleanup_permissions(self):
         if self.user_permissions:
             for user_name, permission_name in self.user_permissions:
                 self.revoke_user_permission(user_name, permission_name)
 
         for permission in self.user_repo_permission_ids:
             RepoModel().revoke_user_permission(*permission)
 
         for permission in self.user_group_repo_permission_ids:
             RepoModel().revoke_user_group_permission(*permission)
 
         for permission in self.user_repo_group_permission_ids:
             RepoGroupModel().revoke_user_permission(*permission)
 
         for permission in self.user_group_repo_group_permission_ids:
             RepoGroupModel().revoke_user_group_permission(*permission)
 
         for permission in self.user_user_group_permission_ids:
             UserGroupModel().revoke_user_permission(*permission)
 
         for permission in self.user_group_user_group_permission_ids:
             UserGroupModel().revoke_user_group_permission(*permission)
 
     def _cleanup_repo_groups(self):
         def _repo_group_compare(first_group_id, second_group_id):
             """
             Gives higher priority to the groups with the most complex paths
             """
             first_group = RepoGroup.get(first_group_id)
             second_group = RepoGroup.get(second_group_id)
             first_group_parts = (
                 len(first_group.group_name.split('/')) if first_group else 0)
             second_group_parts = (
                 len(second_group.group_name.split('/')) if second_group else 0)
             return cmp(second_group_parts, first_group_parts)
 
         sorted_repo_group_ids = sorted(
             self.repo_group_ids, cmp=_repo_group_compare)
         for repo_group_id in sorted_repo_group_ids:
             self.fixture.destroy_repo_group(repo_group_id)
 
     def _cleanup_repos(self):
         sorted_repos_ids = sorted(self.repos_ids)
         for repo_id in sorted_repos_ids:
             self.fixture.destroy_repo(repo_id)
 
     def _cleanup_user_groups(self):
         def _user_group_compare(first_group_id, second_group_id):
             """
             Gives higher priority to the groups with the most complex paths
             """
             first_group = UserGroup.get(first_group_id)
             second_group = UserGroup.get(second_group_id)
             first_group_parts = (
                 len(first_group.users_group_name.split('/'))
                 if first_group else 0)
             second_group_parts = (
                 len(second_group.users_group_name.split('/'))
                 if second_group else 0)
             return cmp(second_group_parts, first_group_parts)
 
         sorted_user_group_ids = sorted(
             self.user_group_ids, cmp=_user_group_compare)
         for user_group_id in sorted_user_group_ids:
             self.fixture.destroy_user_group(user_group_id)
 
     def _cleanup_users(self):
         for user_id in self.user_ids:
             self.fixture.destroy_user(user_id)
 
 
 # TODO: Think about moving this into a pytest-pyro package and make it a
 # pytest plugin
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_runtest_makereport(item, call):
     """
     Adding the remote traceback if the exception has this information.
 
     VCSServer attaches this information as the attribute `_vcs_server_traceback`
     to the exception instance.
     """
     outcome = yield
     report = outcome.get_result()
     if call.excinfo:
         _add_vcsserver_remote_traceback(report, call.excinfo.value)
 
 
 def _add_vcsserver_remote_traceback(report, exc):
     vcsserver_traceback = getattr(exc, '_vcs_server_traceback', None)
 
     if vcsserver_traceback:
         section = 'VCSServer remote traceback ' + report.when
         report.sections.append((section, vcsserver_traceback))
 
 
 @pytest.fixture(scope='session')
 def testrun():
     return {
         'uuid': uuid.uuid4(),
         'start': datetime.datetime.utcnow().isoformat(),
         'timestamp': int(time.time()),
     }
 
 
 class AppenlightClient(object):
 
     url_template = '{url}?protocol_version=0.5'
 
     def __init__(
             self, url, api_key, add_server=True, add_timestamp=True,
             namespace=None, request=None, testrun=None):
         self.url = self.url_template.format(url=url)
         self.api_key = api_key
         self.add_server = add_server
         self.add_timestamp = add_timestamp
         self.namespace = namespace
         self.request = request
         self.server = socket.getfqdn(socket.gethostname())
         self.tags_before = {}
         self.tags_after = {}
         self.stats = []
         self.testrun = testrun or {}
 
     def tag_before(self, tag, value):
         self.tags_before[tag] = value
 
     def tag_after(self, tag, value):
         self.tags_after[tag] = value
 
     def collect(self, data):
         if self.add_server:
             data.setdefault('server', self.server)
         if self.add_timestamp:
             data.setdefault('date', datetime.datetime.utcnow().isoformat())
         if self.namespace:
             data.setdefault('namespace', self.namespace)
         if self.request:
             data.setdefault('request', self.request)
         self.stats.append(data)
 
     def send_stats(self):
         tags = [
             ('testrun', self.request),
             ('testrun.start', self.testrun['start']),
             ('testrun.timestamp', self.testrun['timestamp']),
             ('test', self.namespace),
         ]
         for key, value in self.tags_before.items():
             tags.append((key + '.before', value))
             try:
                 delta = self.tags_after[key] - value
                 tags.append((key + '.delta', delta))
             except Exception:
                 pass
         for key, value in self.tags_after.items():
             tags.append((key + '.after', value))
         self.collect({
             'message': "Collected tags",
             'tags': tags,
         })
 
         response = requests.post(
             self.url,
             headers={
                 'X-appenlight-api-key': self.api_key},
             json=self.stats,
         )
 
         if not response.status_code == 200:
             pprint.pprint(self.stats)
             print(response.headers)
             print(response.text)
             raise Exception('Sending to appenlight failed')
 
 
 @pytest.fixture()
 def gist_util(request, db_connection):
     """
     Provides a wired instance of `GistUtility` with integrated cleanup.
     """
     utility = GistUtility()
     request.addfinalizer(utility.cleanup)
     return utility
 
 
 class GistUtility(object):
     def __init__(self):
         self.fixture = Fixture()
         self.gist_ids = []
 
     def create_gist(self, **kwargs):
         gist = self.fixture.create_gist(**kwargs)
         self.gist_ids.append(gist.gist_id)
         return gist
 
     def cleanup(self):
         for id_ in self.gist_ids:
             self.fixture.destroy_gists(str(id_))
 
 
 @pytest.fixture()
 def enabled_backends(request):
     backends = request.config.option.backends
     return backends[:]
 
 
 @pytest.fixture()
 def settings_util(request, db_connection):
     """
     Provides a wired instance of `SettingsUtility` with integrated cleanup.
     """
     utility = SettingsUtility()
     request.addfinalizer(utility.cleanup)
     return utility
 
 
 class SettingsUtility(object):
     def __init__(self):
         self.rhodecode_ui_ids = []
         self.rhodecode_setting_ids = []
         self.repo_rhodecode_ui_ids = []
         self.repo_rhodecode_setting_ids = []
 
     def create_repo_rhodecode_ui(
             self, repo, section, value, key=None, active=True, cleanup=True):
         key = key or hashlib.sha1(
             '{}{}{}'.format(section, value, repo.repo_id)).hexdigest()
 
         setting = RepoRhodeCodeUi()
         setting.repository_id = repo.repo_id
         setting.ui_section = section
         setting.ui_value = value
         setting.ui_key = key
         setting.ui_active = active
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.repo_rhodecode_ui_ids.append(setting.ui_id)
         return setting
 
     def create_rhodecode_ui(
             self, section, value, key=None, active=True, cleanup=True):
         key = key or hashlib.sha1('{}{}'.format(section, value)).hexdigest()
 
         setting = RhodeCodeUi()
         setting.ui_section = section
         setting.ui_value = value
         setting.ui_key = key
         setting.ui_active = active
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.rhodecode_ui_ids.append(setting.ui_id)
         return setting
 
     def create_repo_rhodecode_setting(
             self, repo, name, value, type_, cleanup=True):
         setting = RepoRhodeCodeSetting(
             repo.repo_id, key=name, val=value, type=type_)
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.repo_rhodecode_setting_ids.append(setting.app_settings_id)
         return setting
 
     def create_rhodecode_setting(self, name, value, type_, cleanup=True):
         setting = RhodeCodeSetting(key=name, val=value, type=type_)
         Session().add(setting)
         Session().commit()
 
         if cleanup:
             self.rhodecode_setting_ids.append(setting.app_settings_id)
 
         return setting
 
     def cleanup(self):
         for id_ in self.rhodecode_ui_ids:
             setting = RhodeCodeUi.get(id_)
             Session().delete(setting)
 
         for id_ in self.rhodecode_setting_ids:
             setting = RhodeCodeSetting.get(id_)
             Session().delete(setting)
 
         for id_ in self.repo_rhodecode_ui_ids:
             setting = RepoRhodeCodeUi.get(id_)
             Session().delete(setting)
 
         for id_ in self.repo_rhodecode_setting_ids:
             setting = RepoRhodeCodeSetting.get(id_)
             Session().delete(setting)
 
         Session().commit()
 
 
 @pytest.fixture()
 def no_notifications(request):
     notification_patcher = mock.patch(
         'rhodecode.model.notification.NotificationModel.create')
     notification_patcher.start()
     request.addfinalizer(notification_patcher.stop)
 
 
 @pytest.fixture(scope='session')
 def repeat(request):
     """
     The number of repetitions is based on this fixture.
 
     Slower calls may divide it by 10 or 100. It is chosen in a way so that the
     tests are not too slow in our default test suite.
     """
     return request.config.getoption('--repeat')
 
 
 @pytest.fixture()
 def rhodecode_fixtures():
     return Fixture()
 
 
 @pytest.fixture()
 def context_stub():
     """
     Stub context object.
     """
     context = pyramid.testing.DummyResource()
     return context
 
 
 @pytest.fixture()
 def request_stub():
     """
     Stub request object.
     """
     from rhodecode.lib.base import bootstrap_request
     request = bootstrap_request(scheme='https')
     return request
 
 
 @pytest.fixture()
 def config_stub(request, request_stub):
     """
     Set up pyramid.testing and return the Configurator.
     """
     from rhodecode.lib.base import bootstrap_config
     config = bootstrap_config(request=request_stub)
 
     @request.addfinalizer
     def cleanup():
         pyramid.testing.tearDown()
 
     return config
 
 
 @pytest.fixture()
 def StubIntegrationType():
     class _StubIntegrationType(IntegrationTypeBase):
         """ Test integration type class """
 
         key = 'test'
         display_name = 'Test integration type'
         description = 'A test integration type for testing'
 
         @classmethod
         def icon(cls):
             return 'test_icon_html_image'
 
         def __init__(self, settings):
             super(_StubIntegrationType, self).__init__(settings)
             self.sent_events = []  # for testing
 
         def send_event(self, event):
             self.sent_events.append(event)
 
         def settings_schema(self):
             class SettingsSchema(colander.Schema):
                 test_string_field = colander.SchemaNode(
                     colander.String(),
                     missing=colander.required,
                     title='test string field',
                 )
                 test_int_field = colander.SchemaNode(
                     colander.Int(),
                     title='some integer setting',
                 )
             return SettingsSchema()
 
 
     integration_type_registry.register_integration_type(_StubIntegrationType)
     return _StubIntegrationType
 
 @pytest.fixture()
 def stub_integration_settings():
     return {
         'test_string_field': 'some data',
         'test_int_field': 100,
     }
 
 
 @pytest.fixture()
 def repo_integration_stub(request, repo_stub, StubIntegrationType,
         stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test repo integration',
         repo=repo_stub, repo_group=None, child_repos_only=None)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture()
 def repogroup_integration_stub(request, test_repo_group, StubIntegrationType,
     stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test repogroup integration',
         repo=None, repo_group=test_repo_group, child_repos_only=True)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture()
 def repogroup_recursive_integration_stub(request, test_repo_group,
     StubIntegrationType, stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test recursive repogroup integration',
         repo=None, repo_group=test_repo_group, child_repos_only=False)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture()
 def global_integration_stub(request, StubIntegrationType,
     stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test global integration',
         repo=None, repo_group=None, child_repos_only=None)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture()
 def root_repos_integration_stub(request, StubIntegrationType,
     stub_integration_settings):
     integration = IntegrationModel().create(
         StubIntegrationType, settings=stub_integration_settings, enabled=True,
         name='test global integration',
         repo=None, repo_group=None, child_repos_only=True)
 
     @request.addfinalizer
     def cleanup():
         IntegrationModel().delete(integration)
 
     return integration
 
 
 @pytest.fixture()
 def local_dt_to_utc():
     def _factory(dt):
         return dt.replace(tzinfo=dateutil.tz.tzlocal()).astimezone(
             dateutil.tz.tzutc()).replace(tzinfo=None)
     return _factory
 
 
 @pytest.fixture()
 def disable_anonymous_user(request, baseapp):
     set_anonymous_access(False)
 
     @request.addfinalizer
     def cleanup():
         set_anonymous_access(True)
 
 
 @pytest.fixture(scope='module')
 def rc_fixture(request):
     return Fixture()
 
 
 @pytest.fixture()
 def repo_groups(request):
     fixture = Fixture()
 
     session = Session()
     zombie_group = fixture.create_repo_group('zombie')
     parent_group = fixture.create_repo_group('parent')
     child_group = fixture.create_repo_group('parent/child')
     groups_in_db = session.query(RepoGroup).all()
     assert len(groups_in_db) == 3
     assert child_group.group_parent_id == parent_group.group_id
 
     @request.addfinalizer
     def cleanup():
         fixture.destroy_repo_group(zombie_group)
         fixture.destroy_repo_group(child_group)
         fixture.destroy_repo_group(parent_group)
 
     return zombie_group, parent_group, child_group

rhodecode/tests/scripts/test_concurency.py

@@ -1,203 +1,203 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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
 """
 
 import os
 import sys
 import shutil
 import logging
 from os.path import join as jn
 from os.path import dirname as dn
 
 from tempfile import _RandomNameSequence
-from subprocess32 import Popen, PIPE
+from subprocess import Popen, PIPE
 
 from rhodecode.lib.utils2 import engine_from_config
 from rhodecode.lib.auth import get_crypt_password
 from rhodecode.model import init_model
 from rhodecode.model import meta
 from rhodecode.model.db import User, Repository
 
 from rhodecode.tests import TESTS_TMP_PATH, HG_REPO
 
 rel_path = dn(dn(dn(dn(os.path.abspath(__file__)))))
 
 
 USER = 'test_admin'
 PASS = 'test12'
 HOST = 'rc.local'
 METHOD = 'pull'
 DEBUG = True
 log = logging.getLogger(__name__)
 
 
 class Command(object):
 
     def __init__(self, cwd):
         self.cwd = cwd
 
     def execute(self, cmd, *args):
         """Runs command on the system with given ``args``.
         """
 
         command = cmd + ' ' + ' '.join(args)
         log.debug('Executing %s', command)
         if DEBUG:
             print(command)
         p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)
         stdout, stderr = p.communicate()
         if DEBUG:
             print('{} {}'.format(stdout, stderr))
         return stdout, stderr
 
 
 def get_session():
     conf = {}
     engine = engine_from_config(conf, 'sqlalchemy.db1.')
     init_model(engine)
     sa = meta.Session
     return sa
 
 
 def create_test_user(force=True):
     print('creating test user')
     sa = get_session()
 
     user = sa.query(User).filter(User.username == USER).scalar()
 
     if force and user is not None:
         print('removing current user')
         for repo in sa.query(Repository).filter(Repository.user == user).all():
             sa.delete(repo)
         sa.delete(user)
         sa.commit()
 
     if user is None or force:
         print('creating new one')
         new_usr = User()
         new_usr.username = USER
         new_usr.password = get_crypt_password(PASS)
         new_usr.email = '[e-mail]'
         new_usr.name = 'test'
         new_usr.lastname = 'lasttestname'
         new_usr.active = True
         new_usr.admin = True
         sa.add(new_usr)
         sa.commit()
 
     print('done')
 
 
 def create_test_repo(force=True):
     print('creating test repo')
     from rhodecode.model.repo import RepoModel
     sa = get_session()
 
     user = sa.query(User).filter(User.username == USER).scalar()
     if user is None:
         raise Exception('user not found')
 
     repo = sa.query(Repository).filter(Repository.repo_name == HG_REPO).scalar()
 
     if repo is None:
         print('repo not found creating')
 
         form_data = {'repo_name': HG_REPO,
                      'repo_type': 'hg',
                      'private':False,
                      'clone_uri': '' }
         rm = RepoModel(sa)
         rm.base_path = '/home/hg'
         rm.create(form_data, user)
 
     print('done')
 
 
 def get_anonymous_access():
     sa = get_session()
     return sa.query(User).filter(User.username == 'default').one().active
 
 
 #==============================================================================
 # TESTS
 #==============================================================================
 def test_clone_with_credentials(repo=HG_REPO, method=METHOD,
                                 seq=None, backend='hg', check_output=True):
     cwd = path = jn(TESTS_TMP_PATH, repo)
 
     if seq is None:
         seq = _RandomNameSequence().next()
 
     try:
         shutil.rmtree(path, ignore_errors=True)
         os.makedirs(path)
     except OSError:
         raise
 
     clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \
                   {'user': USER,
                    'pass': PASS,
                    'host': HOST,
                    'cloned_repo': repo, }
 
     dest = path + seq
     if method == 'pull':
         stdout, stderr = Command(cwd).execute(backend, method, '--cwd', dest, clone_url)
     else:
         stdout, stderr = Command(cwd).execute(backend, method, clone_url, dest)
         if check_output:
             if backend == 'hg':
                 assert """adding file changes""" in stdout, 'no messages about cloning'
                 assert """abort""" not in stderr, 'got error from clone'
             elif backend == 'git':
                 assert """Cloning into""" in stdout, 'no messages about cloning'
 
 
 if __name__ == '__main__':
     try:
         create_test_user(force=False)
         seq = None
         import time
 
         try:
             METHOD = sys.argv[3]
         except Exception:
             pass
 
         try:
             backend = sys.argv[4]
         except Exception:
             backend = 'hg'
 
         if METHOD == 'pull':
             seq = _RandomNameSequence().next()
             test_clone_with_credentials(repo=sys.argv[1], method='clone',
                                         seq=seq, backend=backend)
         s = time.time()
         for i in range(1, int(sys.argv[2]) + 1):
             print('take {}'.format(i))
             test_clone_with_credentials(repo=sys.argv[1], method=METHOD,
                                         seq=seq, backend=backend)
         print('time taken %.4f' % (time.time() - s))
     except Exception as e:
         sys.exit('stop on %s' % e)

rhodecode/tests/server_utils.py

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

rhodecode/tests/utils.py

@@ -1,468 +1,468 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 threading
 import time
 import logging
 import os.path
-import subprocess32
+import subprocess
 import tempfile
 import urllib.request, urllib.error, urllib.parse
 from lxml.html import fromstring, tostring
 from lxml.cssselect import CSSSelector
 import urllib.parse.urlparse
 from urllib.parse import unquote_plus
 import webob
 
 from webtest.app import TestResponse, TestApp
 from webtest.compat import print_stderr
 
 import pytest
 import rc_testdata
 
 from rhodecode.model.db import User, Repository
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.lib.vcs.backends.svn.repository import SubversionRepository
 from rhodecode.lib.vcs.backends.base import EmptyCommit
 from rhodecode.tests import login_user_session
 
 log = logging.getLogger(__name__)
 
 
 class CustomTestResponse(TestResponse):
 
     def _save_output(self, out):
         f = tempfile.NamedTemporaryFile(delete=False, prefix='rc-test-', suffix='.html')
         f.write(out)
         return f.name
 
     def mustcontain(self, *strings, **kw):
         """
         Assert that the response contains all of the strings passed
         in as arguments.
 
         Equivalent to::
 
             assert string in res
         """
         print_body = kw.pop('print_body', False)
         if 'no' in kw:
             no = kw['no']
             del kw['no']
             if isinstance(no, str):
                 no = [no]
         else:
             no = []
         if kw:
             raise TypeError(
                 "The only keyword argument allowed is 'no' got %s" % kw)
 
         f = self._save_output(str(self))
 
         for s in strings:
             if not s in self:
                 print_stderr("Actual response (no %r):" % s)
                 print_stderr("body output saved as `%s`" % f)
                 if print_body:
                     print_stderr(str(self))
                 raise IndexError(
                     "Body does not contain string %r, body output saved as %s" % (s, f))
 
         for no_s in no:
             if no_s in self:
                 print_stderr("Actual response (has %r)" % no_s)
                 print_stderr("body output saved as `%s`" % f)
                 if print_body:
                     print_stderr(str(self))
                 raise IndexError(
                     "Body contains bad string %r, body output saved as %s" % (no_s, f))
 
     def assert_response(self):
         return AssertResponse(self)
 
     def get_session_from_response(self):
         """
         This returns the session from a response object.
         """
         from rhodecode.lib.rc_beaker import session_factory_from_settings
         session = session_factory_from_settings(self.test_app._pyramid_settings)
         return session(self.request)
 
 
 class TestRequest(webob.BaseRequest):
 
     # for py.test
     disabled = True
     ResponseClass = CustomTestResponse
 
     def add_response_callback(self, callback):
         pass
 
 
 class CustomTestApp(TestApp):
     """
     Custom app to make mustcontain more Useful, and extract special methods
     """
     RequestClass = TestRequest
     rc_login_data = {}
     rc_current_session = None
 
     def login(self, username=None, password=None):
         from rhodecode.lib import auth
 
         if username and password:
             session = login_user_session(self, username, password)
         else:
             session = login_user_session(self)
 
         self.rc_login_data['csrf_token'] = auth.get_csrf_token(session)
         self.rc_current_session = session
         return session['rhodecode_user']
 
     @property
     def csrf_token(self):
         return self.rc_login_data['csrf_token']
 
     @property
     def _pyramid_registry(self):
         return self.app.config.registry
 
     @property
     def _pyramid_settings(self):
         return self._pyramid_registry.settings
 
 
 def set_anonymous_access(enabled):
     """(Dis)allows anonymous access depending on parameter `enabled`"""
     user = User.get_default_user()
     user.active = enabled
     Session().add(user)
     Session().commit()
     time.sleep(1.5)  # must sleep for cache (1s to expire)
     log.info('anonymous access is now: %s', enabled)
     assert enabled == User.get_default_user().active, (
         'Cannot set anonymous access')
 
 
 def check_xfail_backends(node, backend_alias):
     # Using "xfail_backends" here intentionally, since this marks work
     # which is "to be done" soon.
     skip_marker = node.get_closest_marker('xfail_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Support for backend %s to be developed." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.xfail(msg)
 
 
 def check_skip_backends(node, backend_alias):
     # Using "skip_backends" here intentionally, since this marks work which is
     # not supported.
     skip_marker = node.get_closest_marker('skip_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Feature not supported for backend %s." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.skip(msg)
 
 
 def extract_git_repo_from_dump(dump_name, repo_name):
     """Create git repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_git_dump(dump_name, target_path)
     return target_path
 
 
 def extract_hg_repo_from_dump(dump_name, repo_name):
     """Create hg repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_hg_dump(dump_name, target_path)
     return target_path
 
 
 def extract_svn_repo_from_dump(dump_name, repo_name):
     """Create a svn repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     SubversionRepository(target_path, create=True)
     _load_svn_dump_into_repo(dump_name, target_path)
     return target_path
 
 
 def assert_message_in_log(log_records, message, levelno, module):
     messages = [
         r.message for r in log_records
         if r.module == module and r.levelno == levelno
     ]
     assert message in messages
 
 
 def _load_svn_dump_into_repo(dump_name, repo_path):
     """
     Utility to populate a svn repository with a named dump
 
     Currently the dumps are in rc_testdata. They might later on be
     integrated with the main repository once they stabilize more.
     """
     dump = rc_testdata.load_svn_dump(dump_name)
-    load_dump = subprocess32.Popen(
+    load_dump = subprocess.Popen(
         ['svnadmin', 'load', repo_path],
-        stdin=subprocess32.PIPE, stdout=subprocess32.PIPE,
-        stderr=subprocess32.PIPE)
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE)
     out, err = load_dump.communicate(dump)
     if load_dump.returncode != 0:
         log.error("Output of load_dump command: %s", out)
         log.error("Error output of load_dump command: %s", err)
         raise Exception(
             'Failed to load dump "%s" into repository at path "%s".'
             % (dump_name, repo_path))
 
 
 class AssertResponse(object):
     """
     Utility that helps to assert things about a given HTML response.
     """
 
     def __init__(self, response):
         self.response = response
 
     def get_imports(self):
         return fromstring, tostring, CSSSelector
 
     def one_element_exists(self, css_selector):
         self.get_element(css_selector)
 
     def no_element_exists(self, css_selector):
         assert not self._get_elements(css_selector)
 
     def element_equals_to(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         element_text = self._element_to_string(element)
         assert expected_content in element_text
 
     def element_contains(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         assert expected_content in element.text_content()
 
     def element_value_contains(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         assert expected_content in element.value
 
     def contains_one_link(self, link_text, href):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector('a[href]')
         elements = [
             e for e in sel(doc) if e.text_content().strip() == link_text]
         assert len(elements) == 1, "Did not find link or found multiple links"
         self._ensure_url_equal(elements[0].attrib.get('href'), href)
 
     def contains_one_anchor(self, anchor_id):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector('#' + anchor_id)
         elements = sel(doc)
         assert len(elements) == 1, 'cannot find 1 element {}'.format(anchor_id)
 
     def _ensure_url_equal(self, found, expected):
         assert _Url(found) == _Url(expected)
 
     def get_element(self, css_selector):
         elements = self._get_elements(css_selector)
         assert len(elements) == 1, 'cannot find 1 element {}'.format(css_selector)
         return elements[0]
 
     def get_elements(self, css_selector):
         return self._get_elements(css_selector)
 
     def _get_elements(self, css_selector):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector(css_selector)
         elements = sel(doc)
         return elements
 
     def _element_to_string(self, element):
         fromstring, tostring, CSSSelector = self.get_imports()
         return tostring(element)
 
 
 class _Url(object):
     """
     A url object that can be compared with other url orbjects
     without regard to the vagaries of encoding, escaping, and ordering
     of parameters in query strings.
 
     Inspired by
     http://stackoverflow.com/questions/5371992/comparing-two-urls-in-python
     """
 
     def __init__(self, url):
         parts = urllib.parse.urlparse(url)
         _query = frozenset(urllib.parse.parse_qsl(parts.query))
         _path = unquote_plus(parts.path)
         parts = parts._replace(query=_query, path=_path)
         self.parts = parts
 
     def __eq__(self, other):
         return self.parts == other.parts
 
     def __hash__(self):
         return hash(self.parts)
 
 
 def run_test_concurrently(times, raise_catched_exc=True):
     """
     Add this decorator to small pieces of code that you want to test
     concurrently
 
     ex:
 
     @test_concurrently(25)
     def my_test_function():
         ...
     """
     def test_concurrently_decorator(test_func):
         def wrapper(*args, **kwargs):
             exceptions = []
 
             def call_test_func():
                 try:
                     test_func(*args, **kwargs)
                 except Exception as e:
                     exceptions.append(e)
                     if raise_catched_exc:
                         raise
             threads = []
             for i in range(times):
                 threads.append(threading.Thread(target=call_test_func))
             for t in threads:
                 t.start()
             for t in threads:
                 t.join()
             if exceptions:
                 raise Exception(
                     'test_concurrently intercepted %s exceptions: %s' % (
                         len(exceptions), exceptions))
         return wrapper
     return test_concurrently_decorator
 
 
 def wait_for_url(url, timeout=10):
     """
     Wait until URL becomes reachable.
 
     It polls the URL until the timeout is reached or it became reachable.
     If will call to `py.test.fail` in case the URL is not reachable.
     """
     timeout = time.time() + timeout
     last = 0
     wait = 0.1
 
     while timeout > last:
         last = time.time()
         if is_url_reachable(url):
             break
         elif (last + wait) > time.time():
             # Go to sleep because not enough time has passed since last check.
             time.sleep(wait)
     else:
         pytest.fail("Timeout while waiting for URL {}".format(url))
 
 
 def is_url_reachable(url):
     try:
         urllib.request.urlopen(url)
     except urllib.error.URLError:
         log.exception('URL `{}` reach error'.format(url))
         return False
     return True
 
 
 def repo_on_filesystem(repo_name):
     from rhodecode.lib import vcs
     from rhodecode.tests import TESTS_TMP_PATH
     repo = vcs.get_vcs_instance(
         os.path.join(TESTS_TMP_PATH, repo_name), create=False)
     return repo is not None
 
 
 def commit_change(
         repo, filename, content, message, vcs_type, parent=None, newfile=False):
     from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 
     repo = Repository.get_by_repo_name(repo)
     _commit = parent
     if not parent:
         _commit = EmptyCommit(alias=vcs_type)
 
     if newfile:
         nodes = {
             filename: {
                 'content': content
             }
         }
         commit = ScmModel().create_nodes(
             user=TEST_USER_ADMIN_LOGIN, repo=repo,
             message=message,
             nodes=nodes,
             parent_commit=_commit,
             author='{} <[e-mail]>'.format(TEST_USER_ADMIN_LOGIN),
         )
     else:
         commit = ScmModel().commit_change(
             repo=repo.scm_instance(), repo_name=repo.repo_name,
             commit=parent, user=TEST_USER_ADMIN_LOGIN,
             author='{} <[e-mail]>'.format(TEST_USER_ADMIN_LOGIN),
             message=message,
             content=content,
             f_path=filename
         )
     return commit
 
 
 def permission_update_data_generator(csrf_token, default=None, grant=None, revoke=None):
     if not default:
         raise ValueError('Permission for default user must be given')
     form_data = [(
         'csrf_token', csrf_token
     )]
     # add default
     form_data.extend([
         ('u_perm_1', default)
     ])
 
     if grant:
         for cnt, (obj_id, perm, obj_name, obj_type) in enumerate(grant, 1):
             form_data.extend([
                 ('perm_new_member_perm_new{}'.format(cnt), perm),
                 ('perm_new_member_id_new{}'.format(cnt), obj_id),
                 ('perm_new_member_name_new{}'.format(cnt), obj_name),
                 ('perm_new_member_type_new{}'.format(cnt), obj_type),
 
             ])
     if revoke:
         for obj_id, obj_type in revoke:
             form_data.extend([
                 ('perm_del_member_id_{}'.format(obj_id), obj_id),
                 ('perm_del_member_type_{}'.format(obj_id), obj_type),
             ])
     return form_data

rhodecode/tests/vcs/test_utils.py

@@ -1,195 +1,195 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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 datetime
-import subprocess32
+import subprocess
 
 import pytest
 
 from rhodecode.lib.vcs.exceptions import VCSError
 from rhodecode.lib.vcs.utils import author_email, author_name
 from rhodecode.lib.vcs.utils.helpers import get_scm
 from rhodecode.lib.vcs.utils.helpers import get_scms_for_path
 from rhodecode.lib.vcs.utils.helpers import parse_datetime
 from rhodecode.lib.vcs.utils.paths import get_dirs_for_path
 
 
 @pytest.mark.usefixtures("baseapp")
 class TestPaths(object):
 
     def _test_get_dirs_for_path(self, path, expected):
         """
         Tests if get_dirs_for_path returns same as expected.
         """
         expected = sorted(expected)
         result = sorted(get_dirs_for_path(path))
         assert result == expected, (
             "%s != %s which was expected result for path %s"
             % (result, expected, path))
 
     def test_get_dirs_for_path(self):
         path = 'foo/bar/baz/file'
         paths_and_results = (
             ('foo/bar/baz/file', ['foo', 'foo/bar', 'foo/bar/baz']),
             ('foo/bar/', ['foo', 'foo/bar']),
             ('foo/bar', ['foo']),
         )
         for path, expected in paths_and_results:
             self._test_get_dirs_for_path(path, expected)
 
     def test_get_scms_for_path(self, tmpdir):
         new = tmpdir.strpath
         assert get_scms_for_path(new) == []
 
         os.mkdir(os.path.join(new, '.tux'))
         assert get_scms_for_path(new) == []
 
         os.mkdir(os.path.join(new, '.git'))
         assert set(get_scms_for_path(new)) == set(['git'])
 
         os.mkdir(os.path.join(new, '.hg'))
         assert set(get_scms_for_path(new)) == set(['git', 'hg'])
 
 
 class TestGetScm(object):
 
     def test_existing_repository(self, vcs_repository_support):
         alias, repo = vcs_repository_support
         assert (alias, repo.path) == get_scm(repo.path)
 
     def test_raises_if_path_is_empty(self, tmpdir):
         with pytest.raises(VCSError):
             get_scm(str(tmpdir))
 
     def test_get_scm_error_path(self):
         with pytest.raises(VCSError):
             get_scm('err')
 
     def test_get_two_scms_for_path(self, tmpdir):
         multialias_repo_path = str(tmpdir)
 
-        subprocess32.check_call(['hg', 'init', multialias_repo_path])
-        subprocess32.check_call(['git', 'init', multialias_repo_path])
+        subprocess.check_call(['hg', 'init', multialias_repo_path])
+        subprocess.check_call(['git', 'init', multialias_repo_path])
 
         with pytest.raises(VCSError):
             get_scm(multialias_repo_path)
 
     def test_ignores_svn_working_copy(self, tmpdir):
         tmpdir.mkdir('.svn')
         with pytest.raises(VCSError):
             get_scm(tmpdir.strpath)
 
 
 class TestParseDatetime(object):
 
     def test_datetime_text(self):
         assert parse_datetime('2010-04-07 21:29:41') == \
             datetime.datetime(2010, 4, 7, 21, 29, 41)
 
     def test_no_seconds(self):
         assert parse_datetime('2010-04-07 21:29') == \
             datetime.datetime(2010, 4, 7, 21, 29)
 
     def test_date_only(self):
         assert parse_datetime('2010-04-07') == \
             datetime.datetime(2010, 4, 7)
 
     def test_another_format(self):
         assert parse_datetime('04/07/10 21:29:41') == \
             datetime.datetime(2010, 4, 7, 21, 29, 41)
 
     def test_now(self):
         assert parse_datetime('now') - datetime.datetime.now() < \
             datetime.timedelta(seconds=1)
 
     def test_today(self):
         today = datetime.date.today()
         assert parse_datetime('today') == \
             datetime.datetime(*today.timetuple()[:3])
 
     def test_yesterday(self):
         yesterday = datetime.date.today() - datetime.timedelta(days=1)
         assert parse_datetime('yesterday') == \
             datetime.datetime(*yesterday.timetuple()[:3])
 
     def test_tomorrow(self):
         tomorrow = datetime.date.today() + datetime.timedelta(days=1)
         args = tomorrow.timetuple()[:3] + (23, 59, 59)
         assert parse_datetime('tomorrow') == datetime.datetime(*args)
 
     def test_days(self):
         timestamp = datetime.datetime.today() - datetime.timedelta(days=3)
         args = timestamp.timetuple()[:3] + (0, 0, 0, 0)
         expected = datetime.datetime(*args)
         assert parse_datetime('3d') == expected
         assert parse_datetime('3 d') == expected
         assert parse_datetime('3 day') == expected
         assert parse_datetime('3 days') == expected
 
     def test_weeks(self):
         timestamp = datetime.datetime.today() - datetime.timedelta(days=3 * 7)
         args = timestamp.timetuple()[:3] + (0, 0, 0, 0)
         expected = datetime.datetime(*args)
         assert parse_datetime('3w') == expected
         assert parse_datetime('3 w') == expected
         assert parse_datetime('3 week') == expected
         assert parse_datetime('3 weeks') == expected
 
     def test_mixed(self):
         timestamp = (
             datetime.datetime.today() - datetime.timedelta(days=2 * 7 + 3))
         args = timestamp.timetuple()[:3] + (0, 0, 0, 0)
         expected = datetime.datetime(*args)
         assert parse_datetime('2w3d') == expected
         assert parse_datetime('2w 3d') == expected
         assert parse_datetime('2w 3 days') == expected
         assert parse_datetime('2 weeks 3 days') == expected
 
 
 @pytest.mark.parametrize("test_str, name, email", [
     ('Marcin Kuzminski <[e-mail]>',
      'Marcin Kuzminski', '[e-mail]'),
     ('Marcin Kuzminski Spaces < [e-mail] >',
      'Marcin Kuzminski Spaces', '[e-mail]'),
     ('Marcin Kuzminski <[e-mail]>',
      'Marcin Kuzminski', '[e-mail]'),
     ('mrf RFC_SPEC <[e-mail]>',
      'mrf RFC_SPEC', '[e-mail]'),
     ('username <[e-mail]>',
      'username', '[e-mail]'),
     ('username <[e-mail]',
      'username', '[e-mail]'),
     ('broken [e-mail]',
      'broken', '[e-mail]'),
     ('<[e-mail]>',
      '', '[e-mail]'),
     ('justname',
      'justname', ''),
     ('Mr Double Name [e-mail] ',
      'Mr Double Name', '[e-mail]'),
 ])
 class TestAuthorExtractors(object):
 
     def test_author_email(self, test_str, name, email):
         assert email == author_email(test_str)
 
     def test_author_name(self, test_str, name, email):
         assert name == author_name(test_str)

rhodecode/tests/vcs/utils.py

@@ -1,118 +1,118 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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/
 
 """
 Utilities for tests only. These are not or should not be used normally -
 functions here are crafted as we don't want to use ``vcs`` to verify tests.
 """
 
 import os
 import re
 import sys
 
-from subprocess32 import Popen
+from subprocess import Popen
 
 
 class VCSTestError(Exception):
     pass
 
 
 def run_command(cmd, args):
     """
     Runs command on the system with given ``args``.
     """
     command = ' '.join((cmd, args))
     p = Popen(command, shell=True)
     status = os.waitpid(p.pid, 0)[1]
     return status
 
 
 def eprint(msg):
     """
     Prints given ``msg`` into sys.stderr as nose test runner hides all output
     from sys.stdout by default and if we want to pipe stream somewhere we don't
     need those verbose messages anyway.
     Appends line break.
     """
     sys.stderr.write(msg)
     sys.stderr.write('\n')
 
 
 # TODO: Revisit once we have CI running, if this is not helping us, remove it
 class SCMFetcher(object):
 
     def __init__(self, alias, test_repo_path):
         """
         :param clone_cmd: command which would clone remote repository; pass
           only first bits - remote path and destination would be appended
           using ``remote_repo`` and ``test_repo_path``
         """
         self.alias = alias
         self.test_repo_path = test_repo_path
 
     def setup(self):
         if not os.path.isdir(self.test_repo_path):
             self.fetch_repo()
 
     def fetch_repo(self):
         """
         Tries to fetch repository from remote path.
         """
         remote = self.remote_repo
         eprint(
             "Fetching repository %s into %s" % (remote, self.test_repo_path))
         run_command(self.clone_cmd,  '%s %s' % (remote, self.test_repo_path))
 
 
 def get_normalized_path(path):
     """
     If given path exists, new path would be generated and returned. Otherwise
     same whats given is returned. Assumes that there would be no more than
     10000 same named files.
     """
     if os.path.exists(path):
         dir, basename = os.path.split(path)
         splitted_name = basename.split('.')
         if len(splitted_name) > 1:
             ext = splitted_name[-1]
         else:
             ext = None
         name = '.'.join(splitted_name[:-1])
         matcher = re.compile(r'^.*-(\d{5})$')
         start = 0
         m = matcher.match(name)
         if not m:
             # Haven't append number yet so return first
             newname = '%s-00000' % name
             newpath = os.path.join(dir, newname)
             if ext:
                 newpath = '.'.join((newpath, ext))
             return get_normalized_path(newpath)
         else:
             start = int(m.group(1)[-5:]) + 1
             for x in range(start, 10000):
                 newname = name[:-5] + str(x).rjust(5, '0')
                 newpath = os.path.join(dir, newname)
                 if ext:
                     newpath = '.'.join((newpath, ext))
                 if not os.path.exists(newpath):
                     return newpath
         raise VCSTestError("Couldn't compute new path for %s" % path)
     return path

rhodecode/tests/vcs_operations/__init__.py

@@ -1,193 +1,193 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 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/
 
 """
 Base 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.
 """
 
 from os.path import join as jn
-from subprocess32 import Popen, PIPE
+from subprocess import Popen, PIPE
 import logging
 import os
 import tempfile
 
 from rhodecode.tests import GIT_REPO, HG_REPO
 
 DEBUG = True
 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)
 
 log = logging.getLogger(__name__)
 
 
 class Command(object):
 
     def __init__(self, cwd):
         self.cwd = cwd
         self.process = None
 
     def execute(self, cmd, *args):
         """
         Runs command on the system with given ``args``.
         """
 
         command = cmd + ' ' + ' '.join(args)
         if DEBUG:
             log.debug('*** CMD %s ***', command)
 
         env = dict(os.environ)
         # Delete coverage variables, as they make the test fail for Mercurial
         for key in env.keys():
             if key.startswith('COV_CORE_'):
                 del env[key]
 
         self.process = Popen(command, shell=True, stdout=PIPE, stderr=PIPE,
                              cwd=self.cwd, env=env)
         stdout, stderr = self.process.communicate()
         if DEBUG:
             log.debug('STDOUT:%s', stdout)
             log.debug('STDERR:%s', stderr)
         return stdout, stderr
 
     def assert_returncode_success(self):
         assert self.process.returncode == 0
 
 
 def _add_files(vcs, dest, clone_url=None, tags=None, target_branch=None, new_branch=False, **kwargs):
     git_ident = "git config user.name {} && git config user.email {}".format(
             'Marcin Kuźminski', '[e-mail]')
     cwd = path = jn(dest)
 
     tags = tags or []
     added_file = jn(path, '%s_setup.py' % tempfile._RandomNameSequence().next())
     Command(cwd).execute('touch %s' % added_file)
     Command(cwd).execute('%s add %s' % (vcs, added_file))
     author_str = 'Marcin Kuźminski <[e-mail]>'
 
     for i in range(kwargs.get('files_no', 3)):
         cmd = """echo 'added_line%s' >> %s""" % (i, added_file)
         Command(cwd).execute(cmd)
 
         if vcs == 'hg':
             cmd = """hg commit -m 'committed new %s' -u '%s' %s """ % (
                 i, author_str, added_file
             )
         elif vcs == 'git':
             cmd = """%s && git commit -m 'committed new %s' %s""" % (
                 git_ident, i, added_file)
         Command(cwd).execute(cmd)
 
     for tag in tags:
         if vcs == 'hg':
             Command(cwd).execute(
                 'hg tag -m "{}" -u "{}" '.format(tag['commit'], author_str), tag['name'])
         elif vcs == 'git':
             if tag['commit']:
                 # annotated tag
                 _stdout, _stderr = Command(cwd).execute(
                     """%s && git tag -a %s -m "%s" """ % (
                         git_ident, tag['name'], tag['commit']))
             else:
                 # lightweight tag
                 _stdout, _stderr = Command(cwd).execute(
                     """%s && git tag %s""" % (
                         git_ident, tag['name']))
 
 
 def _add_files_and_push(vcs, dest, clone_url=None, tags=None, target_branch=None,
                         new_branch=False, **kwargs):
     """
     Generate some files, add it to DEST repo and push back
     vcs is git or hg and defines what VCS we want to make those files for
     """
     git_ident = "git config user.name {} && git config user.email {}".format(
             'Marcin Kuźminski', '[e-mail]')
     cwd = path = jn(dest)
 
     # commit some stuff into this repo
     _add_files(vcs, dest, clone_url, tags, target_branch, new_branch, **kwargs)
 
     default_target_branch = {
         'git': 'master',
         'hg': 'default'
     }.get(vcs)
 
     target_branch = target_branch or default_target_branch
 
     # PUSH it back
     stdout = stderr = None
     if vcs == 'hg':
         maybe_new_branch = ''
         if new_branch:
             maybe_new_branch = '--new-branch'
         stdout, stderr = Command(cwd).execute(
             'hg push --verbose {} -r {} {}'.format(maybe_new_branch, target_branch, clone_url)
         )
     elif vcs == 'git':
         stdout, stderr = Command(cwd).execute(
             """{} && 
             git push --verbose --tags {} {}""".format(git_ident, clone_url, target_branch)
         )
 
     return stdout, stderr
 
 
 def _check_proper_git_push(
         stdout, stderr, branch='master', should_set_default_branch=False):
     # Note: Git is writing most information to stderr intentionally
     assert 'fatal' not in stderr
     assert 'rejected' not in stderr
     assert 'Pushing to' in stderr
     assert '%s -> %s' % (branch, branch) in stderr
 
     if should_set_default_branch:
         assert "Setting default branch to %s" % branch in stderr
     else:
         assert "Setting default branch" not in stderr
 
 
 def _check_proper_hg_push(stdout, stderr, branch='default'):
     assert 'pushing to' in stdout
     assert 'searching for changes' in stdout
 
     assert 'abort:' not in stderr
 
 
 def _check_proper_clone(stdout, stderr, vcs):
     if vcs == 'hg':
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
 
         assert stderr == ''
 
     if vcs == 'git':
         assert '' == stdout
         assert 'Cloning into' in stderr
         assert 'abort:' not in stderr
         assert 'fatal:' not in stderr