tests: use gunicorn for testing. This is close to production testing

and actually will reproduce any chunked encoding problems as we hit on production
servers.

rhodecode/tests/other/vcs_operations/conftest.py

@@ -1,271 +1,271 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 py.test config for test suite for making push/pull operations.
 
 .. important::
 
    You must have git >= 1.8.5 for tests to work fine. With 68b939b git started
    to redirect things to stderr instead of stdout.
 """
 
 import ConfigParser
 import os
 import subprocess32
 import tempfile
 import textwrap
 import pytest
 
 import rhodecode
 from rhodecode.model.db import Repository
 from rhodecode.model.meta import Session
 from rhodecode.model.settings import SettingsModel
 from rhodecode.tests import (
     GIT_REPO, HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,)
 from rhodecode.tests.fixture import Fixture
 from rhodecode.tests.utils import is_url_reachable, wait_for_url
 
 RC_LOG = os.path.join(tempfile.gettempdir(), 'rc.log')
 REPO_GROUP = 'a_repo_group'
 HG_REPO_WITH_GROUP = '%s/%s' % (REPO_GROUP, HG_REPO)
 GIT_REPO_WITH_GROUP = '%s/%s' % (REPO_GROUP, GIT_REPO)
 
 
 def assert_no_running_instance(url):
     if is_url_reachable(url):
         print("Hint: Usually this means another instance of Enterprise "
               "is running in the background.")
         pytest.fail(
             "Port is not free at %s, cannot start web interface" % url)
 
 
 def get_port(pyramid_config):
     config = ConfigParser.ConfigParser()
     config.read(pyramid_config)
     return config.get('server:main', 'port')
 
 
 def get_host_url(pyramid_config):
     """Construct the host url using the port in the test configuration."""
     return '127.0.0.1:%s' % get_port(pyramid_config)
 
 
 class RcWebServer(object):
     """
     Represents a running RCE web server used as a test fixture.
     """
     def __init__(self, pyramid_config, log_file):
         self.pyramid_config = pyramid_config
         self.log_file = log_file
 
     def repo_clone_url(self, repo_name, **kwargs):
         params = {
             'user': TEST_USER_ADMIN_LOGIN,
             'passwd': TEST_USER_ADMIN_PASS,
             'host': get_host_url(self.pyramid_config),
             'cloned_repo': repo_name,
         }
         params.update(**kwargs)
         _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s' % params
         return _url
 
     def host_url(self):
         return 'http://' + get_host_url(self.pyramid_config)
 
     def get_rc_log(self):
         with open(self.log_file) as f:
             return f.read()
 
 
 @pytest.fixture(scope="module")
 def rcextensions(request, baseapp, tmpdir_factory):
     """
     Installs a testing rcextensions pack to ensure they work as expected.
     """
     init_content = textwrap.dedent("""
         # Forward import the example rcextensions to make it
         # active for our tests.
         from rhodecode.tests.other.example_rcextensions import *
     """)
 
     # Note: rcextensions are looked up based on the path of the ini file
     root_path = tmpdir_factory.getbasetemp()
     rcextensions_path = root_path.join('rcextensions')
     init_path = rcextensions_path.join('__init__.py')
 
     if rcextensions_path.check():
         pytest.fail(
             "Path for rcextensions already exists, please clean up before "
             "test run this path: %s" % (rcextensions_path, ))
         return
 
     request.addfinalizer(rcextensions_path.remove)
     init_path.write_binary(init_content, ensure=True)
 
 
 @pytest.fixture(scope="module")
 def repos(request, baseapp):
     """Create a copy of each test repo in a repo group."""
     fixture = Fixture()
     repo_group = fixture.create_repo_group(REPO_GROUP)
     repo_group_id = repo_group.group_id
     fixture.create_fork(HG_REPO, HG_REPO,
                         repo_name_full=HG_REPO_WITH_GROUP,
                         repo_group=repo_group_id)
     fixture.create_fork(GIT_REPO, GIT_REPO,
                         repo_name_full=GIT_REPO_WITH_GROUP,
                         repo_group=repo_group_id)
 
     @request.addfinalizer
     def cleanup():
         fixture.destroy_repo(HG_REPO_WITH_GROUP)
         fixture.destroy_repo(GIT_REPO_WITH_GROUP)
         fixture.destroy_repo_group(repo_group_id)
 
 
 @pytest.fixture(scope="module")
 def rc_web_server_config(testini_factory):
     """
     Configuration file used for the fixture `rc_web_server`.
     """
     CUSTOM_PARAMS = [
         {'handler_console': {'level': 'DEBUG'}},
     ]
     return testini_factory(CUSTOM_PARAMS)
 
 
 @pytest.fixture(scope="module")
 def rc_web_server(
         request, baseapp, rc_web_server_config, repos, rcextensions):
     """
     Run the web server as a subprocess.
 
     Since we have already a running vcsserver, this is not spawned again.
     """
     env = os.environ.copy()
     env['RC_NO_TMP_PATH'] = '1'
 
     rc_log = list(RC_LOG.partition('.log'))
     rc_log.insert(1, get_port(rc_web_server_config))
     rc_log = ''.join(rc_log)
 
     server_out = open(rc_log, 'w')
 
     host_url = 'http://' + get_host_url(rc_web_server_config)
     assert_no_running_instance(host_url)
-    command = ['pserve', rc_web_server_config]
+    command = ['gunicorn', '--worker-class', 'gevent', '--paste', rc_web_server_config]
 
     print('Starting rhodecode server: {}'.format(host_url))
     print('Command: {}'.format(command))
     print('Logfile: {}'.format(rc_log))
 
     proc = subprocess32.Popen(
         command, bufsize=0, env=env, stdout=server_out, stderr=server_out)
 
     wait_for_url(host_url, timeout=30)
 
     @request.addfinalizer
     def stop_web_server():
         # TODO: Find out how to integrate with the reporting of py.test to
         # make this information available.
         print("\nServer log file written to %s" % (rc_log, ))
         proc.kill()
         server_out.flush()
         server_out.close()
 
     return RcWebServer(rc_web_server_config, log_file=rc_log)
 
 
 @pytest.fixture
 def disable_locking(baseapp):
     r = Repository.get_by_repo_name(GIT_REPO)
     Repository.unlock(r)
     r.enable_locking = False
     Session().add(r)
     Session().commit()
 
     r = Repository.get_by_repo_name(HG_REPO)
     Repository.unlock(r)
     r.enable_locking = False
     Session().add(r)
     Session().commit()
 
 
 @pytest.fixture
 def enable_auth_plugins(request, baseapp, csrf_token):
     """
     Return a factory object that when called, allows to control which
     authentication plugins are enabled.
     """
     def _enable_plugins(plugins_list, override=None):
         override = override or {}
         params = {
             'auth_plugins': ','.join(plugins_list),
         }
 
         # helper translate some names to others
         name_map = {
             'token': 'authtoken'
         }
 
         for module in plugins_list:
             plugin_name = module.partition('#')[-1]
             if plugin_name in name_map:
                 plugin_name = name_map[plugin_name]
             enabled_plugin = 'auth_%s_enabled' % plugin_name
             cache_ttl = 'auth_%s_cache_ttl' % plugin_name
 
             # default params that are needed for each plugin,
             # `enabled` and `cache_ttl`
             params.update({
                 enabled_plugin: True,
                 cache_ttl: 0
             })
             if override.get:
                 params.update(override.get(module, {}))
 
             validated_params = params
             for k, v in validated_params.items():
                 setting = SettingsModel().create_or_update_setting(k, v)
                 Session().add(setting)
             Session().commit()
 
     def cleanup():
         _enable_plugins(['egg:rhodecode-enterprise-ce#rhodecode'])
 
     request.addfinalizer(cleanup)
 
     return _enable_plugins
 
 
 @pytest.fixture
 def fs_repo_only(request, rhodecode_fixtures):
     def fs_repo_fabric(repo_name, repo_type):
         rhodecode_fixtures.create_repo(repo_name, repo_type=repo_type)
         rhodecode_fixtures.destroy_repo(repo_name, fs_remove=False)
 
         def cleanup():
             rhodecode_fixtures.destroy_repo(repo_name, fs_remove=True)
             rhodecode_fixtures.destroy_repo_on_filesystem(repo_name)
 
         request.addfinalizer(cleanup)
 
     return fs_repo_fabric

rhodecode/tests/pylons_plugin.py

@@ -1,380 +1,400 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import json
 import platform
 import socket
+import tempfile
+
 import subprocess32
 import time
 from urllib2 import urlopen, URLError
 
 import configobj
 import pytest
 
 import pyramid.paster
 
 from rhodecode.lib.pyramid_utils import get_app_config
 from rhodecode.tests.fixture import TestINI
 import rhodecode
+from rhodecode.tests.other.vcs_operations.conftest import get_host_url, get_port
+
+VCSSERVER_LOG = os.path.join(tempfile.gettempdir(), 'rc-vcsserver.log')
 
 
 def _parse_json(value):
     return json.loads(value) if value else None
 
 
 def pytest_addoption(parser):
     parser.addoption(
         '--test-loglevel', dest='test_loglevel',
         help="Set default Logging level for tests, warn (default), info, debug")
     group = parser.getgroup('pylons')
     group.addoption(
         '--with-pylons', dest='pyramid_config',
         help="Set up a Pylons environment with the specified config file.")
     group.addoption(
         '--ini-config-override', action='store', type=_parse_json,
         default=None, dest='pyramid_config_override', help=(
             "Overrides the .ini file settings. Should be specified in JSON"
             " format, e.g. '{\"section\": {\"parameter\": \"value\", ...}}'"
         )
     )
     parser.addini(
         'pyramid_config',
         "Set up a Pyramid environment with the specified config file.")
 
     vcsgroup = parser.getgroup('vcs')
     vcsgroup.addoption(
         '--without-vcsserver', dest='with_vcsserver', action='store_false',
         help="Do not start the VCSServer in a background process.")
     vcsgroup.addoption(
         '--with-vcsserver-http', dest='vcsserver_config_http',
         help="Start the HTTP VCSServer with the specified config file.")
     vcsgroup.addoption(
         '--vcsserver-protocol', dest='vcsserver_protocol',
         help="Start the VCSServer with HTTP protocol support.")
     vcsgroup.addoption(
         '--vcsserver-config-override', action='store', type=_parse_json,
         default=None, dest='vcsserver_config_override', help=(
             "Overrides the .ini file settings for the VCSServer. "
             "Should be specified in JSON "
             "format, e.g. '{\"section\": {\"parameter\": \"value\", ...}}'"
         )
     )
     vcsgroup.addoption(
         '--vcsserver-port', action='store', type=int,
         default=None, help=(
             "Allows to set the port of the vcsserver. Useful when testing "
             "against an already running server and random ports cause "
             "trouble."))
     parser.addini(
         'vcsserver_config_http',
         "Start the HTTP VCSServer with the specified config file.")
     parser.addini(
         'vcsserver_protocol',
         "Start the VCSServer with HTTP protocol support.")
 
 
 @pytest.fixture(scope='session')
 def vcsserver(request, vcsserver_port, vcsserver_factory):
     """
     Session scope VCSServer.
 
     Tests wich need the VCSServer have to rely on this fixture in order
     to ensure it will be running.
 
     For specific needs, the fixture vcsserver_factory can be used. It allows to
     adjust the configuration file for the test run.
 
     Command line args:
 
     --without-vcsserver: Allows to switch this fixture off. You have to
     manually start the server.
 
     --vcsserver-port: Will expect the VCSServer to listen on this port.
     """
 
     if not request.config.getoption('with_vcsserver'):
         return None
 
     use_http = _use_vcs_http_server(request.config)
     return vcsserver_factory(
         request, use_http=use_http, vcsserver_port=vcsserver_port)
 
 
 @pytest.fixture(scope='session')
 def vcsserver_factory(tmpdir_factory):
     """
     Use this if you need a running vcsserver with a special configuration.
     """
 
     def factory(request, use_http=True, overrides=(), vcsserver_port=None):
 
         if vcsserver_port is None:
             vcsserver_port = get_available_port()
 
         overrides = list(overrides)
         if use_http:
             overrides.append({'server:main': {'port': vcsserver_port}})
         else:
             overrides.append({'DEFAULT': {'port': vcsserver_port}})
 
         if is_cygwin():
             platform_override = {'DEFAULT': {
                 'beaker.cache.repo_object.type': 'nocache'}}
             overrides.append(platform_override)
 
         option_name = 'vcsserver_config_http' if use_http else ''
         override_option_name = 'vcsserver_config_override'
         config_file = get_config(
             request.config, option_name=option_name,
             override_option_name=override_option_name, overrides=overrides,
             basetemp=tmpdir_factory.getbasetemp().strpath,
             prefix='test_vcs_')
 
         print("Using the VCSServer configuration:{}".format(config_file))
         ServerClass = HttpVCSServer if use_http else None
         server = ServerClass(config_file)
         server.start()
 
         @request.addfinalizer
         def cleanup():
             server.shutdown()
 
         server.wait_until_ready()
         return server
 
     return factory
 
 
 def is_cygwin():
     return 'cygwin' in platform.system().lower()
 
 
 def _use_vcs_http_server(config):
     protocol_option = 'vcsserver_protocol'
     protocol = (
         config.getoption(protocol_option) or
         config.getini(protocol_option) or
         'http')
     return protocol == 'http'
 
 
 def _use_log_level(config):
     level = config.getoption('test_loglevel') or 'warn'
     return level.upper()
 
 
 class VCSServer(object):
     """
     Represents a running VCSServer instance.
     """
 
     _args = []
 
     def start(self):
         print("Starting the VCSServer: {}".format(self._args))
         self.process = subprocess32.Popen(self._args)
 
     def wait_until_ready(self, timeout=30):
         raise NotImplementedError()
 
     def shutdown(self):
         self.process.kill()
 
 
 class HttpVCSServer(VCSServer):
     """
     Represents a running VCSServer instance.
     """
     def __init__(self, config_file):
+        self.config_file = config_file
         config_data = configobj.ConfigObj(config_file)
         self._config = config_data['server:main']
 
-        args = ['pserve', config_file]
+        args = ['gunicorn', '--workers', '1', '--paste', config_file]
         self._args = args
 
     @property
     def http_url(self):
         template = 'http://{host}:{port}/'
         return template.format(**self._config)
 
     def start(self):
-        self.process = subprocess32.Popen(self._args)
+        env = os.environ.copy()
+        host_url = 'http://' + get_host_url(self.config_file)
+
+        rc_log = list(VCSSERVER_LOG.partition('.log'))
+        rc_log.insert(1, get_port(self.config_file))
+        rc_log = ''.join(rc_log)
+
+        server_out = open(rc_log, 'w')
+
+        command = ' '.join(self._args)
+        print('Starting rhodecode-vcsserver: {}'.format(host_url))
+        print('Command: {}'.format(command))
+        print('Logfile: {}'.format(rc_log))
+        self.process = subprocess32.Popen(
+            self._args, bufsize=0, env=env, stdout=server_out, stderr=server_out)
 
     def wait_until_ready(self, timeout=30):
         host = self._config['host']
         port = self._config['port']
         status_url = 'http://{host}:{port}/status'.format(host=host, port=port)
         start = time.time()
 
         while time.time() - start < timeout:
             try:
                 urlopen(status_url)
                 break
             except URLError:
                 time.sleep(0.2)
         else:
             pytest.exit(
                 "Starting the VCSServer failed or took more than {} "
                 "seconds. cmd: `{}`".format(timeout, ' '.join(self._args)))
 
     def shutdown(self):
         self.process.kill()
 
 
 @pytest.fixture(scope='session')
 def ini_config(request, tmpdir_factory, rcserver_port, vcsserver_port):
     option_name = 'pyramid_config'
     log_level = _use_log_level(request.config)
 
     overrides = [
         {'server:main': {'port': rcserver_port}},
         {'app:main': {
             'vcs.server': 'localhost:%s' % vcsserver_port,
             # johbo: We will always start the VCSServer on our own based on the
             # fixtures of the test cases. For the test run it must always be
             # off in the INI file.
             'vcs.start_server': 'false',
         }},
 
         {'handler_console': {
             'class ': 'StreamHandler',
             'args ': '(sys.stderr,)',
             'level': log_level,
         }},
 
     ]
     if _use_vcs_http_server(request.config):
         overrides.append({
             'app:main': {
                 'vcs.server.protocol': 'http',
                 'vcs.scm_app_implementation': 'http',
                 'vcs.hooks.protocol': 'http',
             }
         })
 
     filename = get_config(
         request.config, option_name=option_name,
         override_option_name='{}_override'.format(option_name),
         overrides=overrides,
         basetemp=tmpdir_factory.getbasetemp().strpath,
         prefix='test_rce_')
     return filename
 
 
 @pytest.fixture(scope='session')
 def ini_settings(ini_config):
     ini_path = ini_config
     return get_app_config(ini_path)
 
 
 @pytest.fixture(scope='session')
 def rcserver_port(request):
     port = get_available_port()
     print('Using rcserver port {}'.format(port))
     return port
 
 
 @pytest.fixture(scope='session')
 def vcsserver_port(request):
     port = request.config.getoption('--vcsserver-port')
     if port is None:
         port = get_available_port()
         print('Using vcsserver port {}'.format(port))
     return port
 
 
 def get_available_port():
     family = socket.AF_INET
     socktype = socket.SOCK_STREAM
     host = '127.0.0.1'
 
     mysocket = socket.socket(family, socktype)
     mysocket.bind((host, 0))
     port = mysocket.getsockname()[1]
     mysocket.close()
     del mysocket
     return port
 
 
 @pytest.fixture(scope='session')
 def available_port_factory():
     """
     Returns a callable which returns free port numbers.
     """
     return get_available_port
 
 
 @pytest.fixture
 def available_port(available_port_factory):
     """
     Gives you one free port for the current test.
 
     Uses "available_port_factory" to retrieve the port.
     """
     return available_port_factory()
 
 
 @pytest.fixture(scope='session')
 def testini_factory(tmpdir_factory, ini_config):
     """
     Factory to create an INI file based on TestINI.
 
     It will make sure to place the INI file in the correct directory.
     """
     basetemp = tmpdir_factory.getbasetemp().strpath
     return TestIniFactory(basetemp, ini_config)
 
 
 class TestIniFactory(object):
 
     def __init__(self, basetemp, template_ini):
         self._basetemp = basetemp
         self._template_ini = template_ini
 
     def __call__(self, ini_params, new_file_prefix='test'):
         ini_file = TestINI(
             self._template_ini, ini_params=ini_params,
             new_file_prefix=new_file_prefix, dir=self._basetemp)
         result = ini_file.create()
         return result
 
 
 def get_config(
         config, option_name, override_option_name, overrides=None,
         basetemp=None, prefix='test'):
     """
     Find a configuration file and apply overrides for the given `prefix`.
     """
     config_file = (
         config.getoption(option_name) or config.getini(option_name))
     if not config_file:
         pytest.exit(
             "Configuration error, could not extract {}.".format(option_name))
 
     overrides = overrides or []
     config_override = config.getoption(override_option_name)
     if config_override:
         overrides.append(config_override)
     temp_ini_file = TestINI(
         config_file, ini_params=overrides, new_file_prefix=prefix,
         dir=basetemp)
 
     return temp_ini_file.create()

rhodecode/tests/rhodecode.ini

@@ -1,768 +1,693 b''
 
 
 ################################################################################
 ##                 RHODECODE COMMUNITY EDITION CONFIGURATION                  ##
 # The %(here)s variable will be replaced with the parent directory of this file#
 ################################################################################
 
 [DEFAULT]
 debug = true
 
 ################################################################################
 ##                            EMAIL CONFIGURATION                             ##
 ## Uncomment and replace with the email address which should receive          ##
 ## any error reports after an application crash                               ##
 ## Additionally these settings will be used by the RhodeCode mailing system   ##
 ################################################################################
 
 ## prefix all emails subjects with given prefix, helps filtering out emails
 #email_prefix = [RhodeCode]
 
 ## email FROM address all mails will be sent
 #app_email_from = rhodecode-noreply@localhost
 
 ## Uncomment and replace with the address which should receive any error report
 ## note: using appenlight for error handling doesn't need this to be uncommented
 #email_to = admin@localhost
 
 ## in case of Application errors, sent an error email form
 #error_email_from = rhodecode_error@localhost
 
 ## additional error message to be send in case of server crash
 #error_message =
 
 
 #smtp_server = mail.server.com
 #smtp_username =
 #smtp_password =
 #smtp_port =
 #smtp_use_tls = false
 #smtp_use_ssl = true
 ## Specify available auth parameters here (e.g. LOGIN PLAIN CRAM-MD5, etc.)
 #smtp_auth =
 
 [server:main]
 ## COMMON ##
 host = 0.0.0.0
 port = 5000
 
-##################################
-##     WAITRESS WSGI SERVER     ##
-## Recommended for Development  ##
-##################################
-
-use = egg:waitress#main
-## number of worker threads
-threads = 5
-## MAX BODY SIZE 100GB
-max_request_body_size = 107374182400
-## Use poll instead of select, fixes file descriptors limits problems.
-## May not work on old windows systems.
-asyncore_use_poll = true
-
-
 ##########################
 ## GUNICORN WSGI SERVER ##
 ##########################
 ## run with gunicorn --log-config rhodecode.ini --paste rhodecode.ini
 
-#use = egg:gunicorn#main
+use = egg:gunicorn#main
 ## Sets the number of process workers. You must set `instance_id = *`
 ## when this option is set to more than one worker, recommended
 ## value is (2 * NUMBER_OF_CPUS + 1), eg 2CPU = 5 workers
 ## The `instance_id = *` must be set in the [app:main] section below
 #workers = 2
 ## number of threads for each of the worker, must be set to 1 for gevent
 ## generally recommened to be at 1
 #threads = 1
 ## process name
 #proc_name = rhodecode
 ## type of worker class, one of sync, gevent
 ## recommended for bigger setup is using of of other than sync one
 #worker_class = sync
 ## The maximum number of simultaneous clients. Valid only for Gevent
 #worker_connections = 10
 ## max number of requests that worker will handle before being gracefully
 ## restarted, could prevent memory leaks
 #max_requests = 1000
 #max_requests_jitter = 30
 ## amount of time a worker can spend with handling a request before it
 ## gets killed and restarted. Set to 6hrs
 #timeout = 21600
 
-## UWSGI ##
-## run with uwsgi --ini-paste-logged <inifile.ini>
-#[uwsgi]
-#socket = /tmp/uwsgi.sock
-#master = true
-#http = 127.0.0.1:5000
-
-## set as deamon and redirect all output to file
-#daemonize = ./uwsgi_rhodecode.log
-
-## master process PID
-#pidfile = ./uwsgi_rhodecode.pid
-
-## stats server with workers statistics, use uwsgitop
-## for monitoring, `uwsgitop 127.0.0.1:1717`
-#stats = 127.0.0.1:1717
-#memory-report = true
-
-## log 5XX errors
-#log-5xx = true
-
-## Set the socket listen queue size.
-#listen = 256
-
-## Gracefully Reload workers after the specified amount of managed requests
-## (avoid memory leaks).
-#max-requests = 1000
-
-## enable large buffers
-#buffer-size=65535
-
-## socket and http timeouts ##
-#http-timeout=3600
-#socket-timeout=3600
-
-## Log requests slower than the specified number of milliseconds.
-#log-slow = 10
-
-## Exit if no app can be loaded.
-#need-app = true
-
-## Set lazy mode (load apps in workers instead of master).
-#lazy = true
-
-## scaling ##
-## set cheaper algorithm to use, if not set default will be used
-#cheaper-algo = spare
-
-## minimum number of workers to keep at all times
-#cheaper = 1
-
-## number of workers to spawn at startup
-#cheaper-initial = 1
-
-## maximum number of workers that can be spawned
-#workers = 4
-
-## how many workers should be spawned at a time
-#cheaper-step = 1
-
 ## prefix middleware for RhodeCode.
 ## recommended when using proxy setup.
 ## allows to set RhodeCode under a prefix in server.
 ## eg https://server.com/custom_prefix. Enable `filter-with =` option below as well.
 ## And set your prefix like: `prefix = /custom_prefix`
 ## be sure to also set beaker.session.cookie_path = /custom_prefix if you need
 ## to make your cookies only work on prefix url
 [filter:proxy-prefix]
 use = egg:PasteDeploy#prefix
 prefix = /
 
 [app:main]
 is_test = True
 use = egg:rhodecode-enterprise-ce
 
 ## enable proxy prefix middleware, defined above
 #filter-with = proxy-prefix
 
 
 ## RHODECODE PLUGINS ##
 rhodecode.includes = rhodecode.api
 
 # api prefix url
 rhodecode.api.url = /_admin/api
 
 
 ## END RHODECODE PLUGINS ##
 
 ## encryption key used to encrypt social plugin tokens,
 ## remote_urls with credentials etc, if not set it defaults to
 ## `beaker.session.secret`
 #rhodecode.encrypted_values.secret =
 
 ## decryption strict mode (enabled by default). It controls if decryption raises
 ## `SignatureVerificationError` in case of wrong key, or damaged encryption data.
 #rhodecode.encrypted_values.strict = false
 
 ## return gzipped responses from Rhodecode (static files/application)
 gzip_responses = false
 
 ## autogenerate javascript routes file on startup
 generate_js_files = false
 
 ## Optional Languages
 ## en(default), be, de, es, fr, it, ja, pl, pt, ru, zh
 lang = en
 
 ## perform a full repository scan on each server start, this should be
 ## set to false after first startup, to allow faster server restarts.
 startup.import_repos = true
 
 ## Uncomment and set this path to use archive download cache.
 ## Once enabled, generated archives will be cached at this location
 ## and served from the cache during subsequent requests for the same archive of
 ## the repository.
 #archive_cache_dir = /tmp/tarballcache
 
 ## URL at which the application is running. This is used for bootstraping
 ## requests in context when no web request is available. Used in ishell, or
 ## SSH calls. Set this for events to receive proper url for SSH calls.
 app.base_url = http://rhodecode.local
 
 ## change this to unique ID for security
 app_instance_uuid = rc-production
 
 ## cut off limit for large diffs (size in bytes)
 cut_off_limit_diff = 1024000
 cut_off_limit_file = 256000
 
 ## use cache version of scm repo everywhere
 vcs_full_cache = false
 
 ## force https in RhodeCode, fixes https redirects, assumes it's always https
 ## Normally this is controlled by proper http flags sent from http server
 force_https = false
 
 ## use Strict-Transport-Security headers
 use_htsts = false
 
 ## number of commits stats will parse on each iteration
 commit_parse_limit = 25
 
 ## git rev filter option, --all is the default filter, if you need to
 ## hide all refs in changelog switch this to --branches --tags
 git_rev_filter = --all
 
 # Set to true if your repos are exposed using the dumb protocol
 git_update_server_info = false
 
 ## RSS/ATOM feed options
 rss_cut_off_limit = 256000
 rss_items_per_page = 10
 rss_include_diff = false
 
 ## gist URL alias, used to create nicer urls for gist. This should be an
 ## url that does rewrites to _admin/gists/{gistid}.
 ## example: http://gist.rhodecode.org/{gistid}. Empty means use the internal
 ## RhodeCode url, ie. http[s]://rhodecode.server/_admin/gists/{gistid}
 gist_alias_url =
 
 ## List of views (using glob pattern syntax) that AUTH TOKENS could be
 ## used for access.
 ## Adding ?auth_token=TOKEN_HASH to the url authenticates this request as if it
 ## came from the the logged in user who own this authentication token.
 ## Additionally @TOKEN syntaxt can be used to bound the view to specific
 ## authentication token. Such view would be only accessible when used together
 ## with this authentication token
 ##
 ## list of all views can be found under `/_admin/permissions/auth_token_access`
 ## The list should be "," separated and on a single line.
 ##
 ## Most common views to enable:
 #    RepoCommitsView:repo_commit_download
 #    RepoCommitsView:repo_commit_patch
 #    RepoCommitsView:repo_commit_raw
 #    RepoCommitsView:repo_commit_raw@TOKEN
 #    RepoFilesView:repo_files_diff
 #    RepoFilesView:repo_archivefile
 #    RepoFilesView:repo_file_raw
 #    GistView:*
 api_access_controllers_whitelist =
 
 ## default encoding used to convert from and to unicode
 ## can be also a comma separated list of encoding in case of mixed encodings
 default_encoding = UTF-8
 
 ## instance-id prefix
 ## a prefix key for this instance used for cache invalidation when running
 ## multiple instances of rhodecode, make sure it's globally unique for
 ## all running rhodecode instances. Leave empty if you don't use it
 instance_id =
 
 ## Fallback authentication plugin. Set this to a plugin ID to force the usage
 ## of an authentication plugin also if it is disabled by it's settings.
 ## This could be useful if you are unable to log in to the system due to broken
 ## authentication settings. Then you can enable e.g. the internal rhodecode auth
 ## module to log in again and fix the settings.
 ##
 ## Available builtin plugin IDs (hash is part of the ID):
 ## egg:rhodecode-enterprise-ce#rhodecode
 ## egg:rhodecode-enterprise-ce#pam
 ## egg:rhodecode-enterprise-ce#ldap
 ## egg:rhodecode-enterprise-ce#jasig_cas
 ## egg:rhodecode-enterprise-ce#headers
 ## egg:rhodecode-enterprise-ce#crowd
 #rhodecode.auth_plugin_fallback = egg:rhodecode-enterprise-ce#rhodecode
 
 ## alternative return HTTP header for failed authentication. Default HTTP
 ## response is 401 HTTPUnauthorized. Currently HG clients have troubles with
 ## handling that causing a series of failed authentication calls.
 ## Set this variable to 403 to return HTTPForbidden, or any other HTTP code
 ## This will be served instead of default 401 on bad authnetication
 auth_ret_code =
 
 ## use special detection method when serving auth_ret_code, instead of serving
 ## ret_code directly, use 401 initially (Which triggers credentials prompt)
 ## and then serve auth_ret_code to clients
 auth_ret_code_detection = false
 
 ## locking return code. When repository is locked return this HTTP code. 2XX
 ## codes don't break the transactions while 4XX codes do
 lock_ret_code = 423
 
 ## allows to change the repository location in settings page
 allow_repo_location_change = true
 
 ## allows to setup custom hooks in settings page
 allow_custom_hooks_settings = true
 
 ## generated license token, goto license page in RhodeCode settings to obtain
 ## new token
 license_token = abra-cada-bra1-rce3
 
 ## supervisor connection uri, for managing supervisor and logs.
 supervisor.uri =
 ## supervisord group name/id we only want this RC instance to handle
 supervisor.group_id = dev
 
 ## Display extended labs settings
 labs_settings_active = true
 
 ####################################
 ###        CELERY CONFIG        ####
 ####################################
 use_celery = false
 broker.host = localhost
 broker.vhost = rabbitmqhost
 broker.port = 5672
 broker.user = rabbitmq
 broker.password = qweqwe
 
 celery.imports = rhodecode.lib.celerylib.tasks
 
 celery.result.backend = amqp
 celery.result.dburi = amqp://
 celery.result.serialier = json
 
 #celery.send.task.error.emails = true
 #celery.amqp.task.result.expires = 18000
 
 celeryd.concurrency = 2
 #celeryd.log.file = celeryd.log
 celeryd.log.level = debug
 celeryd.max.tasks.per.child = 1
 
 ## tasks will never be sent to the queue, but executed locally instead.
 celery.always.eager = false
 
 ####################################
 ###         BEAKER CACHE        ####
 ####################################
 # default cache dir for templates.  Putting this into a ramdisk
 ## can boost performance, eg. %(here)s/data_ramdisk
 cache_dir = %(here)s/data
 
 ## locking and default file storage for Beaker. Putting this into a ramdisk
 ## can boost performance, eg. %(here)s/data_ramdisk/cache/beaker_data
 beaker.cache.data_dir = %(here)s/rc/data/cache/beaker_data
 beaker.cache.lock_dir = %(here)s/rc/data/cache/beaker_lock
 
 beaker.cache.regions = super_short_term, short_term, long_term, sql_cache_short, auth_plugins, repo_cache_long
 
 beaker.cache.super_short_term.type = memory
 beaker.cache.super_short_term.expire = 1
 beaker.cache.super_short_term.key_length = 256
 
 beaker.cache.short_term.type = memory
 beaker.cache.short_term.expire = 60
 beaker.cache.short_term.key_length = 256
 
 beaker.cache.long_term.type = memory
 beaker.cache.long_term.expire = 36000
 beaker.cache.long_term.key_length = 256
 
 beaker.cache.sql_cache_short.type = memory
 beaker.cache.sql_cache_short.expire = 1
 beaker.cache.sql_cache_short.key_length = 256
 
 ## default is memory cache, configure only if required
 ## using multi-node or multi-worker setup
 #beaker.cache.auth_plugins.type = memory
 #beaker.cache.auth_plugins.lock_dir = %(here)s/data/cache/auth_plugin_lock
 #beaker.cache.auth_plugins.url = postgresql://postgres:secret@localhost/rhodecode
 #beaker.cache.auth_plugins.url = mysql://root:secret@127.0.0.1/rhodecode
 #beaker.cache.auth_plugins.sa.pool_recycle = 3600
 #beaker.cache.auth_plugins.sa.pool_size = 10
 #beaker.cache.auth_plugins.sa.max_overflow = 0
 
 beaker.cache.repo_cache_long.type = memorylru_base
 beaker.cache.repo_cache_long.max_items = 4096
 beaker.cache.repo_cache_long.expire = 2592000
 
 ## default is memorylru_base cache, configure only if required
 ## using multi-node or multi-worker setup
 #beaker.cache.repo_cache_long.type = ext:memcached
 #beaker.cache.repo_cache_long.url = localhost:11211
 #beaker.cache.repo_cache_long.expire = 1209600
 #beaker.cache.repo_cache_long.key_length = 256
 
 ####################################
 ###       BEAKER SESSION        ####
 ####################################
 
 ## .session.type is type of storage options for the session, current allowed
 ## types are file, ext:memcached, ext:database, and memory (default).
 beaker.session.type = file
 beaker.session.data_dir = %(here)s/rc/data/sessions/data
 
 ## db based session, fast, and allows easy management over logged in users
 #beaker.session.type = ext:database
 #beaker.session.table_name = db_session
 #beaker.session.sa.url = postgresql://postgres:secret@localhost/rhodecode
 #beaker.session.sa.url = mysql://root:secret@127.0.0.1/rhodecode
 #beaker.session.sa.pool_recycle = 3600
 #beaker.session.sa.echo = false
 
 beaker.session.key = rhodecode
 beaker.session.secret = test-rc-uytcxaz
 beaker.session.lock_dir = %(here)s/rc/data/sessions/lock
 
 ## Secure encrypted cookie. Requires AES and AES python libraries
 ## you must disable beaker.session.secret to use this
 #beaker.session.encrypt_key = key_for_encryption
 #beaker.session.validate_key = validation_key
 
 ## sets session as invalid(also logging out user) if it haven not been
 ## accessed for given amount of time in seconds
 beaker.session.timeout = 2592000
 beaker.session.httponly = true
 ## Path to use for the cookie. Set to prefix if you use prefix middleware
 #beaker.session.cookie_path = /custom_prefix
 
 ## uncomment for https secure cookie
 beaker.session.secure = false
 
 ## auto save the session to not to use .save()
 beaker.session.auto = false
 
 ## default cookie expiration time in seconds, set to `true` to set expire
 ## at browser close
 #beaker.session.cookie_expires = 3600
 
 ###################################
 ## SEARCH INDEXING CONFIGURATION ##
 ###################################
 ## Full text search indexer is available in rhodecode-tools under
 ## `rhodecode-tools index` command
 
 ## WHOOSH Backend, doesn't require additional services to run
 ## it works good with few dozen repos
 search.module = rhodecode.lib.index.whoosh
 search.location = %(here)s/data/index
 
 ########################################
 ###    CHANNELSTREAM CONFIG         ####
 ########################################
 ## channelstream enables persistent connections and live notification
 ## in the system. It's also used by the chat system
 
 channelstream.enabled = false
 
 ## server address for channelstream server on the backend
 channelstream.server = 127.0.0.1:9800
 ## location of the channelstream server from outside world
 ## use ws:// for http or wss:// for https. This address needs to be handled
 ## by external HTTP server such as Nginx or Apache
 ## see nginx/apache configuration examples in our docs
 channelstream.ws_url = ws://rhodecode.yourserver.com/_channelstream
 channelstream.secret = secret
 channelstream.history.location = %(here)s/channelstream_history
 
 ## Internal application path that Javascript uses to connect into.
 ## If you use proxy-prefix the prefix should be added before /_channelstream
 channelstream.proxy_path = /_channelstream
 
 
 ###################################
 ##       APPENLIGHT CONFIG       ##
 ###################################
 
 ## Appenlight is tailored to work with RhodeCode, see
 ## http://appenlight.com for details how to obtain an account
 
 ## appenlight integration enabled
 appenlight = false
 
 appenlight.server_url = https://api.appenlight.com
 appenlight.api_key = YOUR_API_KEY
 #appenlight.transport_config = https://api.appenlight.com?threaded=1&timeout=5
 
 # used for JS client
 appenlight.api_public_key = YOUR_API_PUBLIC_KEY
 
 ## TWEAK AMOUNT OF INFO SENT HERE
 
 ## enables 404 error logging (default False)
 appenlight.report_404 = false
 
 ## time in seconds after request is considered being slow (default 1)
 appenlight.slow_request_time = 1
 
 ## record slow requests in application
 ## (needs to be enabled for slow datastore recording and time tracking)
 appenlight.slow_requests = true
 
 ## enable hooking to application loggers
 appenlight.logging = true
 
 ## minimum log level for log capture
 appenlight.logging.level = WARNING
 
 ## send logs only from erroneous/slow requests
 ## (saves API quota for intensive logging)
 appenlight.logging_on_error = false
 
 ## list of additonal keywords that should be grabbed from environ object
 ## can be string with comma separated list of words in lowercase
 ## (by default client will always send following info:
 ## 'REMOTE_USER', 'REMOTE_ADDR', 'SERVER_NAME', 'CONTENT_TYPE' + all keys that
 ## start with HTTP* this list be extended with additional keywords here
 appenlight.environ_keys_whitelist =
 
 ## list of keywords that should be blanked from request object
 ## can be string with comma separated list of words in lowercase
 ## (by default client will always blank keys that contain following words
 ## 'password', 'passwd', 'pwd', 'auth_tkt', 'secret', 'csrf'
 ## this list be extended with additional keywords set here
 appenlight.request_keys_blacklist =
 
 ## list of namespaces that should be ignores when gathering log entries
 ## can be string with comma separated list of namespaces
 ## (by default the client ignores own entries: appenlight_client.client)
 appenlight.log_namespace_blacklist =
 
 
 ################################################################################
 ## WARNING: *THE LINE BELOW MUST BE UNCOMMENTED ON A PRODUCTION ENVIRONMENT*  ##
 ## Debug mode will enable the interactive debugging tool, allowing ANYONE to  ##
 ## execute malicious code after an exception is raised.                       ##
 ################################################################################
 set debug = false
 
 
 ##############
 ## STYLING  ##
 ##############
 debug_style = false
 
 ###########################################
 ###   MAIN RHODECODE DATABASE CONFIG    ###
 ###########################################
 #sqlalchemy.db1.url = sqlite:///%(here)s/rhodecode_test.db?timeout=30
 #sqlalchemy.db1.url = postgresql://postgres:qweqwe@localhost/rhodecode_test
 #sqlalchemy.db1.url = mysql://root:qweqwe@localhost/rhodecode_test
 sqlalchemy.db1.url = sqlite:///%(here)s/rhodecode_test.db?timeout=30
 
 # see sqlalchemy docs for other advanced settings
 
 ## print the sql statements to output
 sqlalchemy.db1.echo = false
 ## recycle the connections after this amount of seconds
 sqlalchemy.db1.pool_recycle = 3600
 sqlalchemy.db1.convert_unicode = true
 
 ## the number of connections to keep open inside the connection pool.
 ## 0 indicates no limit
 #sqlalchemy.db1.pool_size = 5
 
 ## the number of connections to allow in connection pool "overflow", that is
 ## connections that can be opened above and beyond the pool_size setting,
 ## which defaults to five.
 #sqlalchemy.db1.max_overflow = 10
 
 
 ##################
 ### VCS CONFIG ###
 ##################
 vcs.server.enable = true
 vcs.server = localhost:9901
 
 ## Web server connectivity protocol, responsible for web based VCS operatations
 ## Available protocols are:
 ## `http` - use http-rpc backend (default)
 vcs.server.protocol = http
 
 ## Push/Pull operations protocol, available options are:
 ## `http` - use http-rpc backend (default)
 ## `vcsserver.scm_app` - internal app (EE only)
 vcs.scm_app_implementation = http
 
 ## Push/Pull operations hooks protocol, available options are:
 ## `http` - use http-rpc backend (default)
 vcs.hooks.protocol = http
 
 vcs.server.log_level = debug
 ## Start VCSServer with this instance as a subprocess, usefull for development
 vcs.start_server = false
 
 ## List of enabled VCS backends, available options are:
 ## `hg`  - mercurial
 ## `git` - git
 ## `svn` - subversion
 vcs.backends = hg, git, svn
 
 vcs.connection_timeout = 3600
 ## Compatibility version when creating SVN repositories. Defaults to newest version when commented out.
 ## Available options are: pre-1.4-compatible, pre-1.5-compatible, pre-1.6-compatible, pre-1.8-compatible, pre-1.9-compatible
 #vcs.svn.compatible_version = pre-1.8-compatible
 
 
 ############################################################
 ### Subversion proxy support (mod_dav_svn)               ###
 ### Maps RhodeCode repo groups into SVN paths for Apache ###
 ############################################################
 ## Enable or disable the config file generation.
 svn.proxy.generate_config = false
 ## Generate config file with `SVNListParentPath` set to `On`.
 svn.proxy.list_parent_path = true
 ## Set location and file name of generated config file.
 svn.proxy.config_file_path = %(here)s/mod_dav_svn.conf
 ## Used as a prefix to the `Location` block in the generated config file.
 ## In most cases it should be set to `/`.
 svn.proxy.location_root = /
 ## Command to reload the mod dav svn configuration on change.
 ## Example: `/etc/init.d/apache2 reload`
 #svn.proxy.reload_cmd = /etc/init.d/apache2 reload
 ## If the timeout expires before the reload command finishes, the command will
 ## be killed. Setting it to zero means no timeout. Defaults to 10 seconds.
 #svn.proxy.reload_timeout = 10
 
 ############################################################
 ### SSH Support Settings                                 ###
 ############################################################
 
 ## Defines if the authorized_keys file should be written on any change of
 ## user ssh keys, setting this to false also disables posibility of adding
 ## ssh keys for users from web interface.
 ssh.generate_authorized_keyfile = true
 
 ## Options for ssh, default is `no-pty,no-port-forwarding,no-X11-forwarding,no-agent-forwarding`
 # ssh.authorized_keys_ssh_opts =
 
 ## File to generate the authorized keys together with options
 ## It is possible to have multiple key files specified in `sshd_config` e.g.
 ## AuthorizedKeysFile %h/.ssh/authorized_keys %h/.ssh/authorized_keys_rhodecode
 ssh.authorized_keys_file_path = %(here)s/rc/authorized_keys_rhodecode
 
 ## Command to execute the SSH wrapper. The binary is available in the
 ## rhodecode installation directory.
 ## e.g ~/.rccontrol/community-1/profile/bin/rc-ssh-wrapper
 ssh.wrapper_cmd = ~/.rccontrol/community-1/rc-ssh-wrapper
 
 ## Allow shell when executing the ssh-wrapper command
 ssh.wrapper_cmd_allow_shell = false
 
 ## Enables logging, and detailed output send back to the client. Usefull for
 ## debugging, shouldn't be used in production.
 ssh.enable_debug_logging = false
 
 ## Paths to binary executrables, by default they are the names, but we can
 ## override them if we want to use a custom one
 ssh.executable.hg = ~/.rccontrol/vcsserver-1/profile/bin/hg
 ssh.executable.git = ~/.rccontrol/vcsserver-1/profile/bin/git
 ssh.executable.svn = ~/.rccontrol/vcsserver-1/profile/bin/svnserve
 
 
 ## Dummy marker to add new entries after.
 ## Add any custom entries below. Please don't remove.
 custom.conf = 1
 
 
 ################################
 ### LOGGING CONFIGURATION   ####
 ################################
 [loggers]
 keys = root, sqlalchemy, beaker, rhodecode, ssh_wrapper
 
 [handlers]
 keys = console, console_sql
 
 [formatters]
 keys = generic, color_formatter, color_formatter_sql
 
 #############
 ## LOGGERS ##
 #############
 [logger_root]
 level = NOTSET
 handlers = console
 
 [logger_routes]
 level = DEBUG
 handlers =
 qualname = routes.middleware
 ## "level = DEBUG" logs the route matched and routing variables.
 propagate = 1
 
 [logger_beaker]
 level = DEBUG
 handlers =
 qualname = beaker.container
 propagate = 1
 
 [logger_rhodecode]
 level = DEBUG
 handlers =
 qualname = rhodecode
 propagate = 1
 
 [logger_sqlalchemy]
 level = ERROR
 handlers = console_sql
 qualname = sqlalchemy.engine
 propagate = 0
 
 [logger_ssh_wrapper]
 level = DEBUG
 handlers =
 qualname = ssh_wrapper
 propagate = 1
 
 
 ##############
 ## HANDLERS ##
 ##############
 
 [handler_console]
 class = StreamHandler
 args = (sys.stderr,)
 level = DEBUG
 formatter = generic
 
 [handler_console_sql]
 class = StreamHandler
 args = (sys.stderr,)
 level = WARN
 formatter = generic
 
 ################
 ## FORMATTERS ##
 ################
 
 [formatter_generic]
 class = rhodecode.lib.logging_formatter.ExceptionAwareFormatter
 format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 datefmt = %Y-%m-%d %H:%M:%S
 
 [formatter_color_formatter]
 class = rhodecode.lib.logging_formatter.ColorFormatter
 format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 datefmt = %Y-%m-%d %H:%M:%S
 
 [formatter_color_formatter_sql]
 class = rhodecode.lib.logging_formatter.ColorFormatterSql
 format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 datefmt = %Y-%m-%d %H:%M:%S

rhodecode/tests/vcs/conftest.py

@@ -1,209 +1,257 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
+import time
 import shutil
-import time
 import datetime
 
 import pytest
 
 from rhodecode.lib.vcs.backends import get_backend
 from rhodecode.lib.vcs.backends.base import Config
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests import get_new_dir
 from rhodecode.tests.utils import check_skip_backends, check_xfail_backends
-from rhodecode.tests.vcs.base import BackendTestMixin
 
 
 @pytest.fixture()
 def vcs_repository_support(
         request, backend_alias, baseapp, _vcs_repo_container):
     """
     Provide a test repository for the test run.
 
     Depending on the value of `recreate_repo_per_test` a new repo for each
     test will be created.
 
     The parameter `--backends` can be used to limit this fixture to specific
     backend implementations.
     """
     cls = request.cls
 
     check_skip_backends(request.node, backend_alias)
     check_xfail_backends(request.node, backend_alias)
 
     if _should_create_repo_per_test(cls):
         _vcs_repo_container = _create_vcs_repo_container(request)
 
     repo = _vcs_repo_container.get_repo(cls, backend_alias=backend_alias)
 
     # TODO: johbo: Supporting old test class api, think about removing this
     cls.repo = repo
     cls.repo_path = repo.path
     cls.default_branch = repo.DEFAULT_BRANCH_NAME
     cls.Backend = cls.backend_class = repo.__class__
     cls.imc = repo.in_memory_commit
 
-    return (backend_alias, repo)
+    return backend_alias, repo
 
 
 @pytest.fixture(scope='class')
 def _vcs_repo_container(request):
     """
     Internal fixture intended to help support class based scoping on demand.
     """
     return _create_vcs_repo_container(request)
 
 
 def _create_vcs_repo_container(request):
     repo_container = VcsRepoContainer()
     if not request.config.getoption('--keep-tmp-path'):
         request.addfinalizer(repo_container.cleanup)
     return repo_container
 
 
 class VcsRepoContainer(object):
 
     def __init__(self):
         self._cleanup_paths = []
         self._repos = {}
 
     def get_repo(self, test_class, backend_alias):
         if backend_alias not in self._repos:
             repo = _create_empty_repository(test_class, backend_alias)
+
             self._cleanup_paths.append(repo.path)
             self._repos[backend_alias] = repo
         return self._repos[backend_alias]
 
     def cleanup(self):
         for repo_path in reversed(self._cleanup_paths):
             shutil.rmtree(repo_path)
 
 
 def _should_create_repo_per_test(cls):
     return getattr(cls, 'recreate_repo_per_test', False)
 
 
 def _create_empty_repository(cls, backend_alias=None):
     Backend = get_backend(backend_alias or cls.backend_alias)
     repo_path = get_new_dir(str(time.time()))
     repo = Backend(repo_path, create=True)
     if hasattr(cls, '_get_commits'):
-        cls.tip = _add_commits_to_repo(repo, cls._get_commits())
+        commits = cls._get_commits()
+        cls.tip = _add_commits_to_repo(repo, commits)
 
     return repo
 
 
 @pytest.fixture
 def config():
     """
     Instance of a repository config.
 
     The instance contains only one value:
 
     - Section: "section-a"
     - Key:     "a-1"
     - Value:   "value-a-1"
 
     The intended usage is for cases where a config instance is needed but no
     specific content is required.
     """
     config = Config()
     config.set('section-a', 'a-1', 'value-a-1')
     return config
 
 
 def _add_commits_to_repo(repo, commits):
     imc = repo.in_memory_commit
-    commit = None
+    tip = None
 
     for commit in commits:
         for node in commit.get('added', []):
             imc.add(FileNode(node.path, content=node.content))
         for node in commit.get('changed', []):
             imc.change(FileNode(node.path, content=node.content))
         for node in commit.get('removed', []):
             imc.remove(FileNode(node.path))
 
-        commit = imc.commit(
+        tip = imc.commit(
             message=unicode(commit['message']),
             author=unicode(commit['author']),
             date=commit['date'],
             branch=commit.get('branch'))
 
-    return commit
+    return tip
 
 
 @pytest.fixture
 def vcs_repo(request, backend_alias):
     Backend = get_backend(backend_alias)
     repo_path = get_new_dir(str(time.time()))
     repo = Backend(repo_path, create=True)
 
     @request.addfinalizer
     def cleanup():
         shutil.rmtree(repo_path)
 
     return repo
 
 
 @pytest.fixture
 def generate_repo_with_commits(vcs_repo):
     """
     Creates a fabric to generate N comits with some file nodes on a randomly
     generated repository
     """
 
     def commit_generator(num):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in xrange(num):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <[e-mail]>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('file_%d.txt' % x, content='Foobar %d' % x),
                 ],
                 'modified': [
                     FileNode('file_%d.txt' % x,
                              content='Foobar %d modified' % (x-1)),
                 ]
             }
 
     def commit_maker(num=5):
         _add_commits_to_repo(vcs_repo, commit_generator(num))
         return vcs_repo
 
     return commit_maker
 
 
 @pytest.fixture
 def hg_repo(request, vcs_repo):
     repo = vcs_repo
 
-    commits = BackendTestMixin._get_commits()
+    commits = repo._get_commits()
     _add_commits_to_repo(repo, commits)
 
     return repo
 
 
 @pytest.fixture
 def hg_commit(hg_repo):
     return hg_repo.get_commit()
+
+
+class BackendTestMixin(object):
+    """
+    This is a backend independent test case class which should be created
+    with ``type`` method.
+
+    It is required to set following attributes at subclass:
+
+    - ``backend_alias``: alias of used backend (see ``vcs.BACKENDS``)
+    - ``repo_path``: path to the repository which would be created for set of
+      tests
+    - ``recreate_repo_per_test``: If set to ``False``, repo would NOT be
+      created
+      before every single test. Defaults to ``True``.
+    """
+    recreate_repo_per_test = True
+
+    @classmethod
+    def _get_commits(cls):
+        commits = [
+            {
+                'message': u'Initial commit',
+                'author': u'Joe Doe <[e-mail]>',
+                'date': datetime.datetime(2010, 1, 1, 20),
+                'added': [
+                    FileNode('foobar', content='Foobar'),
+                    FileNode('foobar2', content='Foobar II'),
+                    FileNode('foo/bar/baz', content='baz here!'),
+                ],
+            },
+            {
+                'message': u'Changes...',
+                'author': u'Jane Doe <[e-mail]>',
+                'date': datetime.datetime(2010, 1, 1, 21),
+                'added': [
+                    FileNode('some/new.txt', content='news...'),
+                ],
+                'changed': [
+                    FileNode('foobar', 'Foobar I'),
+                ],
+                'removed': [],
+            },
+        ]
+        return commits
+
+

rhodecode/tests/vcs/test_archives.py

@@ -1,150 +1,151 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import datetime
 import os
 import shutil
 import tarfile
 import tempfile
 import zipfile
 import StringIO
 
 import mock
 import pytest
 
 from rhodecode.lib.vcs.backends import base
 from rhodecode.lib.vcs.exceptions import ImproperArchiveTypeError, VCSError
 from rhodecode.lib.vcs.nodes import FileNode
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
[e-mail]("vcs_repository_support")
 class TestArchives(BackendTestMixin):
 
     @pytest.fixture(autouse=True)
     def tempfile(self, request):
         self.temp_file = tempfile.mkstemp()[1]
 
         @request.addfinalizer
         def cleanup():
             os.remove(self.temp_file)
 
     @classmethod
     def _get_commits(cls):
         start_date = datetime.datetime(2010, 1, 1, 20)
-        for x in xrange(5):
+        for x in range(5):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <[e-mail]>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode(
                         '%d/file_%d.txt' % (x, x), content='Foobar %d' % x),
                 ],
             }
 
     @pytest.mark.parametrize('compressor', ['gz', 'bz2'])
     def test_archive_tar(self, compressor):
         self.tip.archive_repo(
             self.temp_file, kind='t' + compressor, prefix='repo')
         out_dir = tempfile.mkdtemp()
         out_file = tarfile.open(self.temp_file, 'r|' + compressor)
         out_file.extractall(out_dir)
         out_file.close()
 
-        for x in xrange(5):
+        for x in range(5):
             node_path = '%d/file_%d.txt' % (x, x)
             with open(os.path.join(out_dir, 'repo/' + node_path)) as f:
                 file_content = f.read()
             assert file_content == self.tip.get_node(node_path).content
 
         shutil.rmtree(out_dir)
 
     def test_archive_zip(self):
         self.tip.archive_repo(self.temp_file, kind='zip', prefix='repo')
         out = zipfile.ZipFile(self.temp_file)
 
-        for x in xrange(5):
+        for x in range(5):
             node_path = '%d/file_%d.txt' % (x, x)
             decompressed = StringIO.StringIO()
             decompressed.write(out.read('repo/' + node_path))
             assert decompressed.getvalue() == \
                 self.tip.get_node(node_path).content
             decompressed.close()
 
     def test_archive_zip_with_metadata(self):
         self.tip.archive_repo(self.temp_file, kind='zip',
                               prefix='repo', write_metadata=True)
 
         out = zipfile.ZipFile(self.temp_file)
         metafile = out.read('.archival.txt')
 
         raw_id = self.tip.raw_id
         assert 'rev:%s' % raw_id in metafile
 
-        for x in xrange(5):
+        for x in range(5):
             node_path = '%d/file_%d.txt' % (x, x)
             decompressed = StringIO.StringIO()
             decompressed.write(out.read('repo/' + node_path))
             assert decompressed.getvalue() == \
                 self.tip.get_node(node_path).content
             decompressed.close()
 
     def test_archive_wrong_kind(self):
         with pytest.raises(ImproperArchiveTypeError):
             self.tip.archive_repo(self.temp_file, kind='wrong kind')
 
 
 @pytest.fixture
 def base_commit():
     """
     Prepare a `base.BaseCommit` just enough for `_validate_archive_prefix`.
     """
     commit = base.BaseCommit()
     commit.repository = mock.Mock()
     commit.repository.name = u'fake_repo'
     commit.short_id = 'fake_id'
     return commit
 
 
 @pytest.mark.parametrize("prefix", [u"unicode-prefix", u"Ünïcödë"])
 def test_validate_archive_prefix_enforces_bytes_as_prefix(prefix, base_commit):
     with pytest.raises(ValueError):
         base_commit._validate_archive_prefix(prefix)
 
 
 def test_validate_archive_prefix_empty_prefix(base_commit):
     # TODO: johbo: Should raise a ValueError here.
     with pytest.raises(VCSError):
         base_commit._validate_archive_prefix('')
 
 
 def test_validate_archive_prefix_with_leading_slash(base_commit):
     # TODO: johbo: Should raise a ValueError here.
     with pytest.raises(VCSError):
         base_commit._validate_archive_prefix('/any')
 
 
 def test_validate_archive_prefix_falls_back_to_repository_name(base_commit):
     prefix = base_commit._validate_archive_prefix(None)
     expected_prefix = base_commit._ARCHIVE_PREFIX_TEMPLATE.format(
         repo_name='fake_repo',
         short_id='fake_id')
     assert isinstance(prefix, str)
     assert prefix == expected_prefix

rhodecode/tests/vcs/test_branches.py

@@ -1,146 +1,147 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import datetime
 
 import pytest
 
 from rhodecode.lib.vcs.nodes import FileNode
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
[e-mail]("vcs_repository_support")
 class TestBranches(BackendTestMixin):
 
     def test_empty_repository_has_no_branches(self, vcsbackend):
         empty_repo = vcsbackend.create_repo()
         assert empty_repo.branches == {}
 
     def test_branches_all(self, vcsbackend):
         branch_count = {
             'git': 1,
             'hg': 1,
             'svn': 0,
         }
         assert len(self.repo.branches_all) == branch_count[vcsbackend.alias]
 
     def test_closed_branches(self):
         assert len(self.repo.branches_closed) == 0
 
     def test_simple(self, local_dt_to_utc):
         tip = self.repo.get_commit()
         assert tip.message == 'Changes...'
         assert tip.date == local_dt_to_utc(datetime.datetime(2010, 1, 1, 21))
 
     @pytest.mark.backends("git", "hg")
     def test_new_branch(self):
         # This check must not be removed to ensure the 'branches' LazyProperty
         # gets hit *before* the new 'foobar' branch got created:
         assert 'foobar' not in self.repo.branches
         self.imc.add(FileNode(
             'docs/index.txt',
             content='Documentation\n'))
         foobar_tip = self.imc.commit(
             message=u'New branch: foobar',
             author=u'joe',
             branch='foobar',
         )
         assert 'foobar' in self.repo.branches
         assert foobar_tip.branch == 'foobar'
 
     @pytest.mark.backends("git", "hg")
     def test_new_head(self):
         tip = self.repo.get_commit()
         self.imc.add(FileNode(
             'docs/index.txt',
             content='Documentation\n'))
         foobar_tip = self.imc.commit(
             message=u'New branch: foobar',
             author=u'joe',
             branch='foobar',
             parents=[tip],
         )
         self.imc.change(FileNode(
             'docs/index.txt',
             content='Documentation\nand more...\n'))
         newtip = self.imc.commit(
             message=u'At default branch',
             author=u'joe',
             branch=foobar_tip.branch,
             parents=[foobar_tip],
         )
 
         newest_tip = self.imc.commit(
             message=u'Merged with %s' % foobar_tip.raw_id,
             author=u'joe',
             branch=self.backend_class.DEFAULT_BRANCH_NAME,
             parents=[newtip, foobar_tip],
         )
 
         assert newest_tip.branch == \
             self.backend_class.DEFAULT_BRANCH_NAME
 
     @pytest.mark.backends("git", "hg")
     def test_branch_with_slash_in_name(self):
         self.imc.add(FileNode('extrafile', content='Some data\n'))
         self.imc.commit(
             u'Branch with a slash!', author=u'joe',
             branch='issue/123')
         assert 'issue/123' in self.repo.branches
 
     @pytest.mark.backends("git", "hg")
     def test_branch_with_slash_in_name_and_similar_without(self):
         self.imc.add(FileNode('extrafile', content='Some data\n'))
         self.imc.commit(
             u'Branch with a slash!', author=u'joe',
             branch='issue/123')
         self.imc.add(FileNode('extrafile II', content='Some data\n'))
         self.imc.commit(
             u'Branch without a slash...', author=u'joe',
             branch='123')
         assert 'issue/123' in self.repo.branches
         assert '123' in self.repo.branches
 
 
 class TestSvnBranches(object):
 
     def test_empty_repository_has_no_tags_and_branches(self, vcsbackend_svn):
         empty_repo = vcsbackend_svn.create_repo()
         assert empty_repo.branches == {}
         assert empty_repo.tags == {}
 
     def test_missing_structure_has_no_tags_and_branches(self, vcsbackend_svn):
         repo = vcsbackend_svn.create_repo(number_of_commits=1)
         assert repo.branches == {}
         assert repo.tags == {}
 
     def test_discovers_ordered_branches(self, vcsbackend_svn):
         repo = vcsbackend_svn['svn-simple-layout']
         expected_branches = [
             'branches/add-docs',
             'branches/argparse',
             'trunk',
         ]
         assert repo.branches.keys() == expected_branches
 
     def test_discovers_ordered_tags(self, vcsbackend_svn):
         repo = vcsbackend_svn['svn-simple-layout']
         expected_tags = [
             'tags/v0.1', 'tags/v0.2', 'tags/v0.3', 'tags/v0.5']
         assert repo.tags.keys() == expected_tags

rhodecode/tests/vcs/test_commits.py

@@ -1,590 +1,593 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import datetime
 import time
 
 import pytest
 
 from rhodecode.lib.vcs.backends.base import (
     CollectionGenerator, FILEMODE_DEFAULT, EmptyCommit)
 from rhodecode.lib.vcs.exceptions import (
     BranchDoesNotExistError, CommitDoesNotExistError,
     RepositoryError, EmptyRepositoryError)
 from rhodecode.lib.vcs.nodes import (
     FileNode, AddedFileNodesGenerator,
     ChangedFileNodesGenerator, RemovedFileNodesGenerator)
 from rhodecode.tests import get_new_dir
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
 class TestBaseChangeset:
 
     def test_is_deprecated(self):
         from rhodecode.lib.vcs.backends.base import BaseChangeset
         pytest.deprecated_call(BaseChangeset)
 
 
-class TestEmptyCommit:
+class TestEmptyCommit(object):
 
     def test_branch_without_alias_returns_none(self):
         commit = EmptyCommit()
         assert commit.branch is None
 
 
[e-mail]("vcs_repository_support")
 class TestCommitsInNonEmptyRepo(BackendTestMixin):
     recreate_repo_per_test = True
 
     @classmethod
     def _get_commits(cls):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in xrange(5):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <[e-mail]>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('file_%d.txt' % x, content='Foobar %d' % x),
                 ],
             }
 
     def test_walk_returns_empty_list_in_case_of_file(self):
         result = list(self.tip.walk('file_0.txt'))
         assert result == []
 
     @pytest.mark.backends("git", "hg")
     def test_new_branch(self):
         self.imc.add(FileNode('docs/index.txt',
                      content='Documentation\n'))
         foobar_tip = self.imc.commit(
             message=u'New branch: foobar',
             author=u'joe',
             branch='foobar',
         )
         assert 'foobar' in self.repo.branches
         assert foobar_tip.branch == 'foobar'
         # 'foobar' should be the only branch that contains the new commit
         branch = self.repo.branches.values()
         assert branch[0] != branch[1]
 
     @pytest.mark.backends("git", "hg")
     def test_new_head_in_default_branch(self):
         tip = self.repo.get_commit()
         self.imc.add(FileNode('docs/index.txt',
                      content='Documentation\n'))
         foobar_tip = self.imc.commit(
             message=u'New branch: foobar',
             author=u'joe',
             branch='foobar',
             parents=[tip],
         )
         self.imc.change(FileNode('docs/index.txt',
                         content='Documentation\nand more...\n'))
         newtip = self.imc.commit(
             message=u'At default branch',
             author=u'joe',
             branch=foobar_tip.branch,
             parents=[foobar_tip],
         )
 
         newest_tip = self.imc.commit(
             message=u'Merged with %s' % foobar_tip.raw_id,
             author=u'joe',
             branch=self.backend_class.DEFAULT_BRANCH_NAME,
             parents=[newtip, foobar_tip],
         )
 
         assert newest_tip.branch == self.backend_class.DEFAULT_BRANCH_NAME
 
     @pytest.mark.backends("git", "hg")
     def test_get_commits_respects_branch_name(self):
         """
         * e1930d0 (HEAD, master) Back in default branch
         | * e1930d0 (docs) New Branch: docs2
         | * dcc14fa New branch: docs
         |/
         * e63c41a Initial commit
         ...
         * 624d3db Commit 0
 
         :return:
         """
         DEFAULT_BRANCH = self.repo.DEFAULT_BRANCH_NAME
         TEST_BRANCH = 'docs'
         org_tip = self.repo.get_commit()
 
         self.imc.add(FileNode('readme.txt', content='Document\n'))
         initial = self.imc.commit(
             message=u'Initial commit',
             author=u'joe',
             parents=[org_tip],
             branch=DEFAULT_BRANCH,)
 
         self.imc.add(FileNode('newdoc.txt', content='foobar\n'))
         docs_branch_commit1 = self.imc.commit(
             message=u'New branch: docs',
             author=u'joe',
             parents=[initial],
             branch=TEST_BRANCH,)
 
         self.imc.add(FileNode('newdoc2.txt', content='foobar2\n'))
         docs_branch_commit2 = self.imc.commit(
             message=u'New branch: docs2',
             author=u'joe',
             parents=[docs_branch_commit1],
             branch=TEST_BRANCH,)
 
         self.imc.add(FileNode('newfile', content='hello world\n'))
         self.imc.commit(
             message=u'Back in default branch',
             author=u'joe',
             parents=[initial],
             branch=DEFAULT_BRANCH,)
 
         default_branch_commits = self.repo.get_commits(
             branch_name=DEFAULT_BRANCH)
         assert docs_branch_commit1 not in list(default_branch_commits)
         assert docs_branch_commit2 not in list(default_branch_commits)
 
         docs_branch_commits = self.repo.get_commits(
             start_id=self.repo.commit_ids[0], end_id=self.repo.commit_ids[-1],
             branch_name=TEST_BRANCH)
         assert docs_branch_commit1 in list(docs_branch_commits)
         assert docs_branch_commit2 in list(docs_branch_commits)
 
     @pytest.mark.backends("svn")
     def test_get_commits_respects_branch_name_svn(self, vcsbackend_svn):
         repo = vcsbackend_svn['svn-simple-layout']
         commits = repo.get_commits(branch_name='trunk')
         commit_indexes = [c.idx for c in commits]
         assert commit_indexes == [1, 2, 3, 7, 12, 15]
 
     def test_get_commit_by_branch(self):
         for branch, commit_id in self.repo.branches.iteritems():
             assert commit_id == self.repo.get_commit(branch).raw_id
 
     def test_get_commit_by_tag(self):
         for tag, commit_id in self.repo.tags.iteritems():
             assert commit_id == self.repo.get_commit(tag).raw_id
 
     def test_get_commit_parents(self):
         repo = self.repo
         for test_idx in [1, 2, 3]:
             commit = repo.get_commit(commit_idx=test_idx - 1)
             assert [commit] == repo.get_commit(commit_idx=test_idx).parents
 
     def test_get_commit_children(self):
         repo = self.repo
         for test_idx in [1, 2, 3]:
             commit = repo.get_commit(commit_idx=test_idx + 1)
             assert [commit] == repo.get_commit(commit_idx=test_idx).children
 
 
[e-mail]("vcs_repository_support")
 class TestCommits(BackendTestMixin):
     recreate_repo_per_test = False
 
     @classmethod
     def _get_commits(cls):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in xrange(5):
             yield {
                 'message': u'Commit %d' % x,
                 'author': u'Joe Doe <[e-mail]>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('file_%d.txt' % x, content='Foobar %d' % x),
                 ],
             }
 
     def test_simple(self):
         tip = self.repo.get_commit()
         assert tip.date, datetime.datetime(2010, 1, 3 == 20)
 
     def test_simple_serialized_commit(self):
         tip = self.repo.get_commit()
         # json.dumps(tip) uses .__json__() method
         data = tip.__json__()
         assert 'branch' in data
         assert data['revision']
 
     def test_retrieve_tip(self):
         tip = self.repo.get_commit('tip')
         assert tip == self.repo.get_commit()
 
     def test_invalid(self):
         with pytest.raises(CommitDoesNotExistError):
             self.repo.get_commit(commit_idx=123456789)
 
     def test_idx(self):
         commit = self.repo[0]
         assert commit.idx == 0
 
     def test_negative_idx(self):
         commit = self.repo.get_commit(commit_idx=-1)
         assert commit.idx >= 0
 
     def test_revision_is_deprecated(self):
         def get_revision(commit):
             return commit.revision
 
         commit = self.repo[0]
         pytest.deprecated_call(get_revision, commit)
 
     def test_size(self):
         tip = self.repo.get_commit()
         size = 5 * len('Foobar N')  # Size of 5 files
         assert tip.size == size
 
     def test_size_at_commit(self):
         tip = self.repo.get_commit()
         size = 5 * len('Foobar N')  # Size of 5 files
         assert self.repo.size_at_commit(tip.raw_id) == size
 
     def test_size_at_first_commit(self):
         commit = self.repo[0]
         size = len('Foobar N')  # Size of 1 file
         assert self.repo.size_at_commit(commit.raw_id) == size
 
     def test_author(self):
         tip = self.repo.get_commit()
         assert_text_equal(tip.author, u'Joe Doe <[e-mail]>')
 
     def test_author_name(self):
         tip = self.repo.get_commit()
         assert_text_equal(tip.author_name, u'Joe Doe')
 
     def test_author_email(self):
         tip = self.repo.get_commit()
         assert_text_equal(tip.author_email, u'[e-mail]')
 
     def test_message(self):
         tip = self.repo.get_commit()
         assert_text_equal(tip.message, u'Commit 4')
 
     def test_diff(self):
         tip = self.repo.get_commit()
         diff = tip.diff()
         assert "+Foobar 4" in diff.raw
 
     def test_prev(self):
         tip = self.repo.get_commit()
         prev_commit = tip.prev()
         assert prev_commit.message == 'Commit 3'
 
     def test_prev_raises_on_first_commit(self):
         commit = self.repo.get_commit(commit_idx=0)
         with pytest.raises(CommitDoesNotExistError):
             commit.prev()
 
     def test_prev_works_on_second_commit_issue_183(self):
         commit = self.repo.get_commit(commit_idx=1)
         prev_commit = commit.prev()
         assert prev_commit.idx == 0
 
     def test_next(self):
         commit = self.repo.get_commit(commit_idx=2)
         next_commit = commit.next()
         assert next_commit.message == 'Commit 3'
 
     def test_next_raises_on_tip(self):
         commit = self.repo.get_commit()
         with pytest.raises(CommitDoesNotExistError):
             commit.next()
 
     def test_get_file_commit(self):
         commit = self.repo.get_commit()
         commit.get_file_commit('file_4.txt')
         assert commit.message == 'Commit 4'
 
     def test_get_filenodes_generator(self):
         tip = self.repo.get_commit()
         filepaths = [node.path for node in tip.get_filenodes_generator()]
         assert filepaths == ['file_%d.txt' % x for x in xrange(5)]
 
     def test_get_file_annotate(self):
         file_added_commit = self.repo.get_commit(commit_idx=3)
         annotations = list(file_added_commit.get_file_annotate('file_3.txt'))
 
         line_no, commit_id, commit_loader, line = annotations[0]
 
         assert line_no == 1
         assert commit_id == file_added_commit.raw_id
         assert commit_loader() == file_added_commit
         assert 'Foobar 3' in line
 
     def test_get_file_annotate_does_not_exist(self):
         file_added_commit = self.repo.get_commit(commit_idx=2)
         # TODO: Should use a specific exception class here?
         with pytest.raises(Exception):
             list(file_added_commit.get_file_annotate('file_3.txt'))
 
     def test_get_file_annotate_tip(self):
         tip = self.repo.get_commit()
         commit = self.repo.get_commit(commit_idx=3)
         expected_values = list(commit.get_file_annotate('file_3.txt'))
         annotations = list(tip.get_file_annotate('file_3.txt'))
 
         # Note: Skip index 2 because the loader function is not the same
         for idx in (0, 1, 3):
             assert annotations[0][idx] == expected_values[0][idx]
 
     def test_get_commits_is_ordered_by_date(self):
         commits = self.repo.get_commits()
         assert isinstance(commits, CollectionGenerator)
         assert len(commits) == 0 or len(commits) != 0
         commits = list(commits)
         ordered_by_date = sorted(commits, key=lambda commit: commit.date)
         assert commits == ordered_by_date
 
     def test_get_commits_respects_start(self):
         second_id = self.repo.commit_ids[1]
         commits = self.repo.get_commits(start_id=second_id)
         assert isinstance(commits, CollectionGenerator)
         commits = list(commits)
         assert len(commits) == 4
 
     def test_get_commits_includes_start_commit(self):
         second_id = self.repo.commit_ids[1]
         commits = self.repo.get_commits(start_id=second_id)
         assert isinstance(commits, CollectionGenerator)
         commits = list(commits)
         assert commits[0].raw_id == second_id
 
     def test_get_commits_respects_end(self):
         second_id = self.repo.commit_ids[1]
         commits = self.repo.get_commits(end_id=second_id)
         assert isinstance(commits, CollectionGenerator)
         commits = list(commits)
         assert commits[-1].raw_id == second_id
         assert len(commits) == 2
 
     def test_get_commits_respects_both_start_and_end(self):
         second_id = self.repo.commit_ids[1]
         third_id = self.repo.commit_ids[2]
         commits = self.repo.get_commits(start_id=second_id, end_id=third_id)
         assert isinstance(commits, CollectionGenerator)
         commits = list(commits)
         assert len(commits) == 2
 
     def test_get_commits_on_empty_repo_raises_EmptyRepository_error(self):
         repo_path = get_new_dir(str(time.time()))
         repo = self.Backend(repo_path, create=True)
 
         with pytest.raises(EmptyRepositoryError):
             list(repo.get_commits(start_id='foobar'))
 
     def test_get_commits_respects_hidden(self):
         commits = self.repo.get_commits(show_hidden=True)
         assert isinstance(commits, CollectionGenerator)
         assert len(commits) == 5
 
     def test_get_commits_includes_end_commit(self):
         second_id = self.repo.commit_ids[1]
         commits = self.repo.get_commits(end_id=second_id)
         assert isinstance(commits, CollectionGenerator)
         assert len(commits) == 2
         commits = list(commits)
         assert commits[-1].raw_id == second_id
 
     def test_get_commits_respects_start_date(self):
         start_date = datetime.datetime(2010, 1, 2)
         commits = self.repo.get_commits(start_date=start_date)
         assert isinstance(commits, CollectionGenerator)
         # Should be 4 commits after 2010-01-02 00:00:00
         assert len(commits) == 4
         for c in commits:
             assert c.date >= start_date
 
     def test_get_commits_respects_start_date_with_branch(self):
         start_date = datetime.datetime(2010, 1, 2)
         commits = self.repo.get_commits(
             start_date=start_date, branch_name=self.repo.DEFAULT_BRANCH_NAME)
         assert isinstance(commits, CollectionGenerator)
         # Should be 4 commits after 2010-01-02 00:00:00
         assert len(commits) == 4
         for c in commits:
             assert c.date >= start_date
 
     def test_get_commits_respects_start_date_and_end_date(self):
         start_date = datetime.datetime(2010, 1, 2)
         end_date = datetime.datetime(2010, 1, 3)
         commits = self.repo.get_commits(start_date=start_date,
                                         end_date=end_date)
         assert isinstance(commits, CollectionGenerator)
         assert len(commits) == 2
         for c in commits:
             assert c.date >= start_date
             assert c.date <= end_date
 
     def test_get_commits_respects_end_date(self):
         end_date = datetime.datetime(2010, 1, 2)
         commits = self.repo.get_commits(end_date=end_date)
         assert isinstance(commits, CollectionGenerator)
         assert len(commits) == 1
         for c in commits:
             assert c.date <= end_date
 
     def test_get_commits_respects_reverse(self):
         commits = self.repo.get_commits()  # no longer reverse support
         assert isinstance(commits, CollectionGenerator)
         assert len(commits) == 5
         commit_ids = reversed([c.raw_id for c in commits])
         assert list(commit_ids) == list(reversed(self.repo.commit_ids))
 
     def test_get_commits_slice_generator(self):
         commits = self.repo.get_commits(
             branch_name=self.repo.DEFAULT_BRANCH_NAME)
         assert isinstance(commits, CollectionGenerator)
         commit_slice = list(commits[1:3])
         assert len(commit_slice) == 2
 
     def test_get_commits_raise_commitdoesnotexist_for_wrong_start(self):
         with pytest.raises(CommitDoesNotExistError):
             list(self.repo.get_commits(start_id='foobar'))
 
     def test_get_commits_raise_commitdoesnotexist_for_wrong_end(self):
         with pytest.raises(CommitDoesNotExistError):
             list(self.repo.get_commits(end_id='foobar'))
 
     def test_get_commits_raise_branchdoesnotexist_for_wrong_branch_name(self):
         with pytest.raises(BranchDoesNotExistError):
             list(self.repo.get_commits(branch_name='foobar'))
 
     def test_get_commits_raise_repositoryerror_for_wrong_start_end(self):
         start_id = self.repo.commit_ids[-1]
         end_id = self.repo.commit_ids[0]
         with pytest.raises(RepositoryError):
             list(self.repo.get_commits(start_id=start_id, end_id=end_id))
 
     def test_get_commits_raises_for_numerical_ids(self):
         with pytest.raises(TypeError):
             self.repo.get_commits(start_id=1, end_id=2)
 
     def test_commit_equality(self):
         commit1 = self.repo.get_commit(self.repo.commit_ids[0])
         commit2 = self.repo.get_commit(self.repo.commit_ids[1])
 
         assert commit1 == commit1
         assert commit2 == commit2
         assert commit1 != commit2
         assert commit2 != commit1
         assert commit1 != None
         assert None != commit1
         assert 1 != commit1
         assert 'string' != commit1
 
 
 @pytest.mark.parametrize("filename, expected", [
     ("README.rst", False),
     ("README", True),
 ])
 def test_commit_is_link(vcsbackend, filename, expected):
     commit = vcsbackend.repo.get_commit()
     link_status = commit.is_link(filename)
     assert link_status is expected
 
 
[e-mail]("vcs_repository_support")
 class TestCommitsChanges(BackendTestMixin):
     recreate_repo_per_test = False
 
     @classmethod
     def _get_commits(cls):
         return [
             {
                 'message': u'Initial',
                 'author': u'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foo/bar', content='foo'),
                     FileNode('foo/bał', content='foo'),
                     FileNode('foobar', content='foo'),
                     FileNode('qwe', content='foo'),
                 ],
             },
             {
                 'message': u'Massive changes',
                 'author': u'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [FileNode('fallout', content='War never changes')],
                 'changed': [
                     FileNode('foo/bar', content='baz'),
                     FileNode('foobar', content='baz'),
                 ],
                 'removed': [FileNode('qwe')],
             },
         ]
 
     def test_initial_commit(self, local_dt_to_utc):
         commit = self.repo.get_commit(commit_idx=0)
         assert set(commit.added) == set([
             commit.get_node('foo/bar'),
             commit.get_node('foo/bał'),
             commit.get_node('foobar'),
             commit.get_node('qwe'),
         ])
         assert set(commit.changed) == set()
         assert set(commit.removed) == set()
         assert set(commit.affected_files) == set(
             ['foo/bar', 'foo/bał', 'foobar', 'qwe'])
         assert commit.date == local_dt_to_utc(
             datetime.datetime(2010, 1, 1, 20, 0))
 
     def test_head_added(self):
         commit = self.repo.get_commit()
         assert isinstance(commit.added, AddedFileNodesGenerator)
         assert set(commit.added) == set([commit.get_node('fallout')])
         assert isinstance(commit.changed, ChangedFileNodesGenerator)
         assert set(commit.changed) == set([
             commit.get_node('foo/bar'),
             commit.get_node('foobar'),
         ])
         assert isinstance(commit.removed, RemovedFileNodesGenerator)
         assert len(commit.removed) == 1
         assert list(commit.removed)[0].path == 'qwe'
 
     def test_get_filemode(self):
         commit = self.repo.get_commit()
         assert FILEMODE_DEFAULT == commit.get_file_mode('foo/bar')
 
     def test_get_filemode_non_ascii(self):
         commit = self.repo.get_commit()
         assert FILEMODE_DEFAULT == commit.get_file_mode('foo/bał')
         assert FILEMODE_DEFAULT == commit.get_file_mode(u'foo/bał')
 
     def test_get_file_history(self):
         commit = self.repo.get_commit()
         history = commit.get_file_history('foo/bar')
         assert len(history) == 2
 
     def test_get_file_history_with_limit(self):
         commit = self.repo.get_commit()
         history = commit.get_file_history('foo/bar', limit=1)
         assert len(history) == 1
 
     def test_get_file_history_first_commit(self):
         commit = self.repo[0]
         history = commit.get_file_history('foo/bar')
         assert len(history) == 1
 
 
 def assert_text_equal(expected, given):
     assert expected == given
     assert isinstance(expected, unicode)
     assert isinstance(given, unicode)

rhodecode/tests/vcs/test_diff.py

@@ -1,546 +1,548 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import datetime
 import pytest
 
 from rhodecode.lib.vcs.nodes import FileNode
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
 class TestGetDiffValidation:
 
     def test_raises_on_string_input(self, vcsbackend):
         repo = vcsbackend.repo
         with pytest.raises(TypeError):
             repo.get_diff("1", "2")
 
     def test_raises_if_commits_not_of_this_repository(self, vcsbackend):
         repo = vcsbackend.repo
         target_repo = vcsbackend.create_repo(number_of_commits=1)
         repo_commit = repo[0]
         wrong_commit = target_repo[0]
         with pytest.raises(ValueError):
             repo.get_diff(repo_commit, wrong_commit)
 
     def test_allows_empty_commit(self, vcsbackend):
         repo = vcsbackend.repo
         commit = repo[0]
         repo.get_diff(repo.EMPTY_COMMIT, commit)
 
     def test_raise_if_both_commits_are_empty(self, vcsbackend):
         repo = vcsbackend.repo
         empty_commit = repo.EMPTY_COMMIT
         with pytest.raises(ValueError):
             repo.get_diff(empty_commit, empty_commit)
 
     def test_supports_path1_parameter(self, vcsbackend):
         repo = vcsbackend.repo
         commit = repo[1]
         repo.get_diff(
             repo.EMPTY_COMMIT, commit,
             path='vcs/__init__.py', path1='vcs/__init__.py')
 
     @pytest.mark.backends("git", "hg")
     def test_raises_value_error_if_paths_not_supported(self, vcsbackend):
         repo = vcsbackend.repo
         commit = repo[1]
         with pytest.raises(ValueError):
             repo.get_diff(
                 repo.EMPTY_COMMIT, commit,
                 path='trunk/example.py', path1='branches/argparse/example.py')
 
 
[e-mail]("vcs_repository_support")
 class TestRepositoryGetDiff(BackendTestMixin):
 
     recreate_repo_per_test = False
 
     @classmethod
     def _get_commits(cls):
         commits = [
             {
                 'message': 'Initial commit',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar', content='foobar'),
                     FileNode('foobar2', content='foobar2'),
                 ],
             },
             {
                 'message': 'Changed foobar, added foobar3',
                 'author': 'Jane Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 21),
                 'added': [
                     FileNode('foobar3', content='foobar3'),
                 ],
                 'changed': [
                     FileNode('foobar', 'FOOBAR'),
                 ],
             },
             {
                 'message': 'Removed foobar, changed foobar3',
                 'author': 'Jane Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'changed': [
                     FileNode('foobar3', content='FOOBAR\nFOOBAR\nFOOBAR\n'),
                 ],
                 'removed': [FileNode('foobar')],
             },
             {
                 'message': 'Whitespace changes',
                 'author': 'Jane Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 23),
                 'changed': [
                     FileNode('foobar3', content='FOOBAR  \nFOOBAR\nFOOBAR\n'),
                 ],
             },
         ]
         return commits
 
     def test_initial_commit_diff(self):
         initial_commit = self.repo[0]
         diff = self.repo.get_diff(self.repo.EMPTY_COMMIT, initial_commit)
         assert diff.raw == self.first_commit_diffs[self.repo.alias]
 
     def test_second_commit_diff(self):
         diff = self.repo.get_diff(self.repo[0], self.repo[1])
         assert diff.raw == self.second_commit_diffs[self.repo.alias]
 
     def test_third_commit_diff(self):
         diff = self.repo.get_diff(self.repo[1], self.repo[2])
         assert diff.raw == self.third_commit_diffs[self.repo.alias]
 
     def test_ignore_whitespace(self):
         diff = self.repo.get_diff(
             self.repo[2], self.repo[3], ignore_whitespace=True)
         assert '@@' not in diff.raw
 
     def test_only_one_file(self):
         diff = self.repo.get_diff(
             self.repo.EMPTY_COMMIT, self.repo[0], path='foobar')
         assert 'foobar2' not in diff.raw
 
     def test_context_parameter(self):
         first_commit = self.repo.get_commit(commit_idx=0)
         diff = self.repo.get_diff(
             self.repo.EMPTY_COMMIT, first_commit, context=2)
         assert diff.raw == self.first_commit_diffs[self.repo.alias]
 
     def test_context_only_one_file(self):
         diff = self.repo.get_diff(
             self.repo.EMPTY_COMMIT, self.repo[0], path='foobar', context=2)
         assert diff.raw == self.first_commit_one_file[self.repo.alias]
 
     first_commit_diffs = {
         'git': r"""diff --git a/foobar b/foobar
 new file mode 100644
 index 0000000000000000000000000000000000000000..f6ea0495187600e7b2288c8ac19c5886383a4632
 --- /dev/null
 +++ b/foobar
 @@ -0,0 +1 @@
 +foobar
 \ No newline at end of file
 diff --git a/foobar2 b/foobar2
 new file mode 100644
 index 0000000000000000000000000000000000000000..e8c9d6b98e3dce993a464935e1a53f50b56a3783
 --- /dev/null
 +++ b/foobar2
 @@ -0,0 +1 @@
 +foobar2
 \ No newline at end of file
 """,
         'hg': r"""diff --git a/foobar b/foobar
 new file mode 100644
 --- /dev/null
 +++ b/foobar
 @@ -0,0 +1,1 @@
 +foobar
 \ No newline at end of file
 diff --git a/foobar2 b/foobar2
 new file mode 100644
 --- /dev/null
 +++ b/foobar2
 @@ -0,0 +1,1 @@
 +foobar2
 \ No newline at end of file
 """,
         'svn': """Index: foobar
 ===================================================================
 diff --git a/foobar b/foobar
 new file mode 10644
 --- /dev/null\t(revision 0)
 +++ b/foobar\t(revision 1)
 @@ -0,0 +1 @@
 +foobar
 \\ No newline at end of file
 Index: foobar2
 ===================================================================
 diff --git a/foobar2 b/foobar2
 new file mode 10644
 --- /dev/null\t(revision 0)
 +++ b/foobar2\t(revision 1)
 @@ -0,0 +1 @@
 +foobar2
 \\ No newline at end of file
 """,
     }
 
     second_commit_diffs = {
         'git': r"""diff --git a/foobar b/foobar
 index f6ea0495187600e7b2288c8ac19c5886383a4632..389865bb681b358c9b102d79abd8d5f941e96551 100644
 --- a/foobar
 +++ b/foobar
 @@ -1 +1 @@
 -foobar
 \ No newline at end of file
 +FOOBAR
 \ No newline at end of file
 diff --git a/foobar3 b/foobar3
 new file mode 100644
 index 0000000000000000000000000000000000000000..c11c37d41d33fb47741cff93fa5f9d798c1535b0
 --- /dev/null
 +++ b/foobar3
 @@ -0,0 +1 @@
 +foobar3
 \ No newline at end of file
 """,
         'hg': r"""diff --git a/foobar b/foobar
 --- a/foobar
 +++ b/foobar
 @@ -1,1 +1,1 @@
 -foobar
 \ No newline at end of file
 +FOOBAR
 \ No newline at end of file
 diff --git a/foobar3 b/foobar3
 new file mode 100644
 --- /dev/null
 +++ b/foobar3
 @@ -0,0 +1,1 @@
 +foobar3
 \ No newline at end of file
 """,
         'svn': """Index: foobar
 ===================================================================
 diff --git a/foobar b/foobar
 --- a/foobar\t(revision 1)
 +++ b/foobar\t(revision 2)
 @@ -1 +1 @@
 -foobar
 \\ No newline at end of file
 +FOOBAR
 \\ No newline at end of file
 Index: foobar3
 ===================================================================
 diff --git a/foobar3 b/foobar3
 new file mode 10644
 --- /dev/null\t(revision 0)
 +++ b/foobar3\t(revision 2)
 @@ -0,0 +1 @@
 +foobar3
 \\ No newline at end of file
 """,
     }
 
     third_commit_diffs = {
         'git': r"""diff --git a/foobar b/foobar
 deleted file mode 100644
 index 389865bb681b358c9b102d79abd8d5f941e96551..0000000000000000000000000000000000000000
 --- a/foobar
 +++ /dev/null
 @@ -1 +0,0 @@
 -FOOBAR
 \ No newline at end of file
 diff --git a/foobar3 b/foobar3
 index c11c37d41d33fb47741cff93fa5f9d798c1535b0..f9324477362684ff692aaf5b9a81e01b9e9a671c 100644
 --- a/foobar3
 +++ b/foobar3
 @@ -1 +1,3 @@
 -foobar3
 \ No newline at end of file
 +FOOBAR
 +FOOBAR
 +FOOBAR
 """,
         'hg': r"""diff --git a/foobar b/foobar
 deleted file mode 100644
 --- a/foobar
 +++ /dev/null
 @@ -1,1 +0,0 @@
 -FOOBAR
 \ No newline at end of file
 diff --git a/foobar3 b/foobar3
 --- a/foobar3
 +++ b/foobar3
 @@ -1,1 +1,3 @@
 -foobar3
 \ No newline at end of file
 +FOOBAR
 +FOOBAR
 +FOOBAR
 """,
         'svn': """Index: foobar
 ===================================================================
 diff --git a/foobar b/foobar
 deleted file mode 10644
 --- a/foobar\t(revision 2)
 +++ /dev/null\t(revision 3)
 @@ -1 +0,0 @@
 -FOOBAR
 \\ No newline at end of file
 Index: foobar3
 ===================================================================
 diff --git a/foobar3 b/foobar3
 --- a/foobar3\t(revision 2)
 +++ b/foobar3\t(revision 3)
 @@ -1 +1,3 @@
 -foobar3
 \\ No newline at end of file
 +FOOBAR
 +FOOBAR
 +FOOBAR
 """,
     }
 
     first_commit_one_file = {
         'git': r"""diff --git a/foobar b/foobar
 new file mode 100644
 index 0000000000000000000000000000000000000000..f6ea0495187600e7b2288c8ac19c5886383a4632
 --- /dev/null
 +++ b/foobar
 @@ -0,0 +1 @@
 +foobar
 \ No newline at end of file
 """,
         'hg': r"""diff --git a/foobar b/foobar
 new file mode 100644
 --- /dev/null
 +++ b/foobar
 @@ -0,0 +1,1 @@
 +foobar
 \ No newline at end of file
 """,
         'svn': """Index: foobar
 ===================================================================
 diff --git a/foobar b/foobar
 new file mode 10644
 --- /dev/null\t(revision 0)
 +++ b/foobar\t(revision 1)
 @@ -0,0 +1 @@
 +foobar
 \\ No newline at end of file
 """,
     }
 
 
 class TestSvnGetDiff:
 
     @pytest.mark.parametrize('path, path1', [
         ('trunk/example.py', 'tags/v0.2/example.py'),
         ('trunk', 'tags/v0.2')
     ], ids=['file', 'dir'])
     def test_diff_to_tagged_version(self, vcsbackend_svn, path, path1):
         repo = vcsbackend_svn['svn-simple-layout']
         commit1 = repo[-2]
         commit2 = repo[-1]
         diff = repo.get_diff(commit1, commit2, path=path, path1=path1)
         assert diff.raw == self.expected_diff_v_0_2
 
     expected_diff_v_0_2 = '''Index: example.py
 ===================================================================
 diff --git a/example.py b/example.py
 --- a/example.py\t(revision 25)
 +++ b/example.py\t(revision 26)
 @@ -7,8 +7,12 @@
  
  @click.command()
  def main():
 +    """
 +    Will print out a useful message on invocation.
 +    """
      click.echo("Hello world!")
  
  
 +# Main entry point
  if __name__ == '__main__':
      main()
 '''
 
     def test_diff_of_moved_directory(self, vcsbackend_svn):
         repo = vcsbackend_svn['svn-move-directory']
         diff = repo.get_diff(repo[0], repo[1])
         # TODO: johbo: Think about supporting svn directory nodes
         # a little bit better, source is here like a file
         expected_diff = """Index: source
 ===================================================================
 diff --git a/source b/source
 deleted file mode 10644
 --- a/source\t(revision 1)
 +++ /dev/null\t(revision 2)
 Index: target/file
 ===================================================================
 diff --git a/target/file b/target/file
 new file mode 10644
 --- /dev/null\t(revision 0)
 +++ b/target/file\t(revision 2)
 """
         assert diff.raw == expected_diff
 
 
[e-mail]("vcs_repository_support")
 class TestGetDiffBinary(BackendTestMixin):
 
     recreate_repo_per_test = False
 
     # Note: "Fake" PNG files, has the correct magic as prefix
     BINARY = """\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00"""
     BINARY2 = """\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x01\x00\x00"""
 
     @staticmethod
     def _get_commits():
         commits = [
             {
                 'message': 'Add binary file image.png',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('image.png', content=TestGetDiffBinary.BINARY),
                 ]},
             {
                 'message': 'Modify image.png',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 21),
                 'changed': [
                     FileNode('image.png', content=TestGetDiffBinary.BINARY2),
                 ]},
             {
                 'message': 'Remove image.png',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 21),
                 'removed': [
                     FileNode('image.png'),
                 ]},
         ]
         return commits
 
     def test_add_a_binary_file(self):
         diff = self.repo.get_diff(self.repo.EMPTY_COMMIT, self.repo[0])
 
         expected = {
             'git': """diff --git a/image.png b/image.png
 new file mode 100644
 index 0000000000000000000000000000000000000000..28380fd4a25c58be1b68b523ba2a314f4459ee9c
 GIT binary patch
 literal 19
 YcmeAS@N?(olHy`uVBq!ia0vp^03%2O-T(jq
 
 literal 0
 HcmV?d00001
 
 """,
             'hg': """diff --git a/image.png b/image.png
 new file mode 100644
 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..28380fd4a25c58be1b68b523ba2a314f4459ee9c
 GIT binary patch
 literal 19
 Yc%17D@N?(olHy`uVBq!ia0vp^03%2O-T(jq
 
 """,
             'svn': """===================================================================
 Cannot display: file marked as a binary type.
 svn:mime-type = application/octet-stream
 Index: image.png
 ===================================================================
 diff --git a/image.png b/image.png
 new file mode 10644
 --- /dev/null\t(revision 0)
 +++ b/image.png\t(revision 1)
 """,
         }
         assert diff.raw == expected[self.repo.alias]
 
     def test_update_a_binary_file(self):
         diff = self.repo.get_diff(self.repo[0], self.repo[1])
 
         expected = {
             'git': """diff --git a/image.png b/image.png
 index 28380fd4a25c58be1b68b523ba2a314f4459ee9c..1008a77cd372386a1c24fbd96019333f67ad0065 100644
 GIT binary patch
 literal 19
 acmeAS@N?(olHy`uVBq!ia0y~$U;qFkO9I~j
 
 literal 19
 YcmeAS@N?(olHy`uVBq!ia0vp^03%2O-T(jq
 
 """,
             'hg': """diff --git a/image.png b/image.png
 index 28380fd4a25c58be1b68b523ba2a314f4459ee9c..1008a77cd372386a1c24fbd96019333f67ad0065
 GIT binary patch
 literal 19
 ac%17D@N?(olHy`uVBq!ia0y~$U;qFkO9I~j
 
 """,
             'svn': """===================================================================
 Cannot display: file marked as a binary type.
 svn:mime-type = application/octet-stream
 Index: image.png
 ===================================================================
 diff --git a/image.png b/image.png
 --- a/image.png\t(revision 1)
 +++ b/image.png\t(revision 2)
 """,
         }
         assert diff.raw == expected[self.repo.alias]
 
     def test_remove_a_binary_file(self):
         diff = self.repo.get_diff(self.repo[1], self.repo[2])
 
         expected = {
             'git': """diff --git a/image.png b/image.png
 deleted file mode 100644
 index 1008a77cd372386a1c24fbd96019333f67ad0065..0000000000000000000000000000000000000000
 GIT binary patch
 literal 0
 HcmV?d00001
 
 literal 19
 acmeAS@N?(olHy`uVBq!ia0y~$U;qFkO9I~j
 
 """,
             'hg': """diff --git a/image.png b/image.png
 deleted file mode 100644
 index 1008a77cd372386a1c24fbd96019333f67ad0065..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
 GIT binary patch
 literal 0
 Hc$@<O00001
 
 """,
             'svn': """===================================================================
 Cannot display: file marked as a binary type.
 svn:mime-type = application/octet-stream
 Index: image.png
 ===================================================================
 diff --git a/image.png b/image.png
 deleted file mode 10644
 --- a/image.png\t(revision 2)
 +++ /dev/null\t(revision 3)
 """,
         }
         assert diff.raw == expected[self.repo.alias]

rhodecode/tests/vcs/test_filenodes_unicode_path.py

@@ -1,50 +1,53 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import datetime
+
+import pytest
 from rhodecode.lib.vcs.nodes import FileNode
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
[e-mail]("vcs_repository_support")
 class TestFileNodeUnicodePath(BackendTestMixin):
 
     fname = 'ąśðąęłąć.txt'
     ufname = (fname).decode('utf-8')
 
     @classmethod
     def _get_commits(cls):
         nodes = [
             FileNode(cls.fname, content='Foobar'),
         ]
 
         commits = [
             {
                 'message': 'Initial commit',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': nodes,
             },
         ]
         return commits
 
     def test_filenode_path(self):
         node = self.tip.get_node(self.fname)
         unode = self.tip.get_node(self.ufname)
         assert node == unode

rhodecode/tests/vcs/test_getitem.py

@@ -1,68 +1,69 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import datetime
 
 import pytest
 
 from rhodecode.lib.vcs.exceptions import CommitDoesNotExistError
 from rhodecode.lib.vcs.nodes import FileNode
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
[e-mail]("vcs_repository_support")
 class TestGetitem(BackendTestMixin):
 
     @classmethod
     def _get_commits(cls):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in xrange(5):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <[e-mail]>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('file_%d.txt' % x, content='Foobar %d' % x),
                 ],
             }
 
     def test_last_item_is_tip(self):
         assert self.repo[-1] == self.repo.get_commit()
 
     @pytest.mark.parametrize("offset, message", [
         (-1, 'Commit 4'),
         (-2, 'Commit 3'),
         (-5, 'Commit 0'),
     ])
     def test_negative_offset_fetches_correct_commit(self, offset, message):
         assert self.repo[offset].message == message
 
     def test_returns_correct_items(self):
         commits = [self.repo[x] for x in xrange(len(self.repo.commit_ids))]
         assert commits == list(self.repo.get_commits())
 
     def test_raises_for_next_commit(self):
         next_commit_idx = len(self.repo.commit_ids)
         with pytest.raises(CommitDoesNotExistError):
             self.repo[next_commit_idx]
 
     def test_raises_for_not_existing_commit_idx(self):
         not_existing_commit_idx = 1000
         with pytest.raises(CommitDoesNotExistError):
             self.repo[not_existing_commit_idx]

rhodecode/tests/vcs/test_getslice.py

@@ -1,75 +1,77 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 import datetime
 
+import pytest
 from rhodecode.lib.vcs.nodes import FileNode
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
[e-mail]("vcs_repository_support")
 class TestGetslice(BackendTestMixin):
 
     @classmethod
     def _get_commits(cls):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in xrange(5):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <[e-mail]>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('file_%d.txt' % x, content='Foobar %d' % x),
                 ],
             }
 
     def test__getslice__last_item_is_tip(self):
         assert list(self.repo[-1:])[0] == self.repo.get_commit()
 
     def test__getslice__respects_start_index(self):
         assert list(self.repo[2:]) == \
             [self.repo.get_commit(commit_id)
              for commit_id in self.repo.commit_ids[2:]]
 
     def test__getslice__respects_negative_start_index(self):
         assert list(self.repo[-2:]) == \
             [self.repo.get_commit(commit_id)
              for commit_id in self.repo.commit_ids[-2:]]
 
     def test__getslice__respects_end_index(self):
         assert list(self.repo[:2]) == \
             [self.repo.get_commit(commit_id)
              for commit_id in self.repo.commit_ids[:2]]
 
     def test__getslice__respects_negative_end_index(self):
         assert list(self.repo[:-2]) == \
             [self.repo.get_commit(commit_id)
              for commit_id in self.repo.commit_ids[:-2]]
 
     def test__getslice__start_grater_than_end(self):
         assert list(self.repo[10:0]) == []
 
     def test__getslice__negative_iteration(self):
         assert list(self.repo[::-1]) == \
             [self.repo.get_commit(commit_id)
              for commit_id in self.repo.commit_ids[::-1]]
 
     def test__getslice__iterate_even(self):
         assert list(self.repo[0:10:2]) == \
             [self.repo.get_commit(commit_id)
              for commit_id in self.repo.commit_ids[0:10:2]]

rhodecode/tests/vcs/test_git.py

@@ -1,1269 +1,1271 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import datetime
 import mock
 import os
 import sys
 import shutil
 
 import pytest
 
 from rhodecode.lib.utils import make_db_config
 from rhodecode.lib.vcs.backends.base import Reference
 from rhodecode.lib.vcs.backends.git import (
     GitRepository, GitCommit, discover_git_version)
 from rhodecode.lib.vcs.exceptions import (
     RepositoryError, VCSError, NodeDoesNotExistError)
 from rhodecode.lib.vcs.nodes import (
     NodeKind, FileNode, DirNode, NodeState, SubModuleNode)
 from rhodecode.tests import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, get_new_dir
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
 pytestmark = pytest.mark.backends("git")
 
 
 def repo_path_generator():
     """
     Return a different path to be used for cloning repos.
     """
     i = 0
     while True:
         i += 1
         yield '%s-%d' % (TEST_GIT_REPO_CLONE, i)
 
 
 REPO_PATH_GENERATOR = repo_path_generator()
 
 
 class TestGitRepository:
 
     # pylint: disable=protected-access
 
     def __check_for_existing_repo(self):
         if os.path.exists(TEST_GIT_REPO_CLONE):
             self.fail('Cannot test git clone repo as location %s already '
                       'exists. You should manually remove it first.'
                       % TEST_GIT_REPO_CLONE)
 
     @pytest.fixture(autouse=True)
     def prepare(self, request, baseapp):
         self.repo = GitRepository(TEST_GIT_REPO, bare=True)
 
     def get_clone_repo(self):
         """
         Return a non bare clone of the base repo.
         """
         clone_path = next(REPO_PATH_GENERATOR)
         repo_clone = GitRepository(
             clone_path, create=True, src_url=self.repo.path, bare=False)
 
         return repo_clone
 
     def get_empty_repo(self, bare=False):
         """
         Return a non bare empty repo.
         """
         return GitRepository(next(REPO_PATH_GENERATOR), create=True, bare=bare)
 
     def test_wrong_repo_path(self):
         wrong_repo_path = '/tmp/errorrepo_git'
         with pytest.raises(RepositoryError):
             GitRepository(wrong_repo_path)
 
     def test_repo_clone(self):
         self.__check_for_existing_repo()
         repo = GitRepository(TEST_GIT_REPO)
         repo_clone = GitRepository(
             TEST_GIT_REPO_CLONE,
             src_url=TEST_GIT_REPO, create=True, update_after_clone=True)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
         # Checking hashes of commits should be enough
         for commit in repo.get_commits():
             raw_id = commit.raw_id
             assert raw_id == repo_clone.get_commit(raw_id).raw_id
 
     def test_repo_clone_without_create(self):
         with pytest.raises(RepositoryError):
             GitRepository(
                 TEST_GIT_REPO_CLONE + '_wo_create', src_url=TEST_GIT_REPO)
 
     def test_repo_clone_with_update(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_with_update'
         repo_clone = GitRepository(
             clone_path,
             create=True, src_url=TEST_GIT_REPO, update_after_clone=True)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
 
         # check if current workdir was updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         assert os.path.isfile(fpath)
 
     def test_repo_clone_without_update(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_without_update'
         repo_clone = GitRepository(
             clone_path,
             create=True, src_url=TEST_GIT_REPO, update_after_clone=False)
         assert len(repo.commit_ids) == len(repo_clone.commit_ids)
         # check if current workdir was *NOT* updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         # Make sure it's not bare repo
         assert not repo_clone.bare
         assert not os.path.isfile(fpath)
 
     def test_repo_clone_into_bare_repo(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_bare.git'
         repo_clone = GitRepository(
             clone_path, create=True, src_url=repo.path, bare=True)
         assert repo_clone.bare
 
     def test_create_repo_is_not_bare_by_default(self):
         repo = GitRepository(get_new_dir('not-bare-by-default'), create=True)
         assert not repo.bare
 
     def test_create_bare_repo(self):
         repo = GitRepository(get_new_dir('bare-repo'), create=True, bare=True)
         assert repo.bare
 
     def test_update_server_info(self):
         self.repo._update_server_info()
 
     def test_fetch(self, vcsbackend_git):
         # Note: This is a git specific part of the API, it's only implemented
         # by the git backend.
         source_repo = vcsbackend_git.repo
         target_repo = vcsbackend_git.create_repo()
         target_repo.fetch(source_repo.path)
         # Note: Get a fresh instance, avoids caching trouble
         target_repo = vcsbackend_git.backend(target_repo.path)
         assert len(source_repo.commit_ids) == len(target_repo.commit_ids)
 
     def test_commit_ids(self):
         # there are 112 commits (by now)
         # so we can assume they would be available from now on
         subset = set([
             'c1214f7e79e02fc37156ff215cd71275450cffc3',
             '38b5fe81f109cb111f549bfe9bb6b267e10bc557',
             'fa6600f6848800641328adbf7811fd2372c02ab2',
             '102607b09cdd60e2793929c4f90478be29f85a17',
             '49d3fd156b6f7db46313fac355dca1a0b94a0017',
             '2d1028c054665b962fa3d307adfc923ddd528038',
             'd7e0d30fbcae12c90680eb095a4f5f02505ce501',
             'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
             '8430a588b43b5d6da365400117c89400326e7992',
             'd955cd312c17b02143c04fa1099a352b04368118',
             'f67b87e5c629c2ee0ba58f85197e423ff28d735b',
             'add63e382e4aabc9e1afdc4bdc24506c269b7618',
             'f298fe1189f1b69779a4423f40b48edf92a703fc',
             'bd9b619eb41994cac43d67cf4ccc8399c1125808',
             '6e125e7c890379446e98980d8ed60fba87d0f6d1',
             'd4a54db9f745dfeba6933bf5b1e79e15d0af20bd',
             '0b05e4ed56c802098dfc813cbe779b2f49e92500',
             '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
             '45223f8f114c64bf4d6f853e3c35a369a6305520',
             'ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
             'f5ea29fc42ef67a2a5a7aecff10e1566699acd68',
             '27d48942240f5b91dfda77accd2caac94708cc7d',
             '622f0eb0bafd619d2560c26f80f09e3b0b0d78af',
             'e686b958768ee96af8029fe19c6050b1a8dd3b2b'])
         assert subset.issubset(set(self.repo.commit_ids))
 
     def test_slicing(self):
         # 4 1 5 10 95
         for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),
                                  (10, 20, 10), (5, 100, 95)]:
             commit_ids = list(self.repo[sfrom:sto])
             assert len(commit_ids) == size
             assert commit_ids[0] == self.repo.get_commit(commit_idx=sfrom)
             assert commit_ids[-1] == self.repo.get_commit(commit_idx=sto - 1)
 
     def test_branches(self):
         # TODO: Need more tests here
         # Removed (those are 'remotes' branches for cloned repo)
         # assert 'master' in self.repo.branches
         # assert 'gittree' in self.repo.branches
         # assert 'web-branch' in self.repo.branches
         for __, commit_id in self.repo.branches.items():
             assert isinstance(self.repo.get_commit(commit_id), GitCommit)
 
     def test_tags(self):
         # TODO: Need more tests here
         assert 'v0.1.1' in self.repo.tags
         assert 'v0.1.2' in self.repo.tags
         for __, commit_id in self.repo.tags.items():
             assert isinstance(self.repo.get_commit(commit_id), GitCommit)
 
     def _test_single_commit_cache(self, commit_id):
         commit = self.repo.get_commit(commit_id)
         assert commit_id in self.repo.commits
         assert commit is self.repo.commits[commit_id]
 
     def test_initial_commit(self):
         commit_id = self.repo.commit_ids[0]
         init_commit = self.repo.get_commit(commit_id)
         init_author = init_commit.author
 
         assert init_commit.message == 'initial import\n'
         assert init_author == 'Marcin Kuzminski <[e-mail]>'
         assert init_author == init_commit.committer
         for path in ('vcs/__init__.py',
                      'vcs/backends/BaseRepository.py',
                      'vcs/backends/__init__.py'):
             assert isinstance(init_commit.get_node(path), FileNode)
         for path in ('', 'vcs', 'vcs/backends'):
             assert isinstance(init_commit.get_node(path), DirNode)
 
         with pytest.raises(NodeDoesNotExistError):
             init_commit.get_node(path='foobar')
 
         node = init_commit.get_node('vcs/')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
 
         node = init_commit.get_node('vcs')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
 
         node = init_commit.get_node('vcs/__init__.py')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.FILE
 
     def test_not_existing_commit(self):
         with pytest.raises(RepositoryError):
             self.repo.get_commit('f' * 40)
 
     def test_commit10(self):
 
         commit10 = self.repo.get_commit(self.repo.commit_ids[9])
         README = """===
 VCS
 ===
 
 Various Version Control System management abstraction layer for Python.
 
 Introduction
 ------------
 
 TODO: To be written...
 
 """
         node = commit10.get_node('README.rst')
         assert node.kind == NodeKind.FILE
         assert node.content == README
 
     def test_head(self):
         assert self.repo.head == self.repo.get_commit().raw_id
 
     def test_checkout_with_create(self):
         repo_clone = self.get_clone_repo()
 
         new_branch = 'new_branch'
         assert repo_clone._current_branch() == 'master'
         assert set(repo_clone.branches) == set(('master',))
         repo_clone._checkout(new_branch, create=True)
 
         # Branches is a lazy property so we need to recrete the Repo object.
         repo_clone = GitRepository(repo_clone.path)
         assert set(repo_clone.branches) == set(('master', new_branch))
         assert repo_clone._current_branch() == new_branch
 
     def test_checkout(self):
         repo_clone = self.get_clone_repo()
 
         repo_clone._checkout('new_branch', create=True)
         repo_clone._checkout('master')
 
         assert repo_clone._current_branch() == 'master'
 
     def test_checkout_same_branch(self):
         repo_clone = self.get_clone_repo()
 
         repo_clone._checkout('master')
         assert repo_clone._current_branch() == 'master'
 
     def test_checkout_branch_already_exists(self):
         repo_clone = self.get_clone_repo()
 
         with pytest.raises(RepositoryError):
             repo_clone._checkout('master', create=True)
 
     def test_checkout_bare_repo(self):
         with pytest.raises(RepositoryError):
             self.repo._checkout('master')
 
     def test_current_branch_bare_repo(self):
         with pytest.raises(RepositoryError):
             self.repo._current_branch()
 
     def test_current_branch_empty_repo(self):
         repo = self.get_empty_repo()
         assert repo._current_branch() is None
 
     def test_local_clone(self):
         clone_path = next(REPO_PATH_GENERATOR)
         self.repo._local_clone(clone_path, 'master')
         repo_clone = GitRepository(clone_path)
 
         assert self.repo.commit_ids == repo_clone.commit_ids
 
     def test_local_clone_with_specific_branch(self):
         source_repo = self.get_clone_repo()
 
         # Create a new branch in source repo
         new_branch_commit = source_repo.commit_ids[-3]
         source_repo._checkout(new_branch_commit)
         source_repo._checkout('new_branch', create=True)
 
         clone_path = next(REPO_PATH_GENERATOR)
         source_repo._local_clone(clone_path, 'new_branch')
         repo_clone = GitRepository(clone_path)
 
         assert source_repo.commit_ids[:-3 + 1] == repo_clone.commit_ids
 
         clone_path = next(REPO_PATH_GENERATOR)
         source_repo._local_clone(clone_path, 'master')
         repo_clone = GitRepository(clone_path)
 
         assert source_repo.commit_ids == repo_clone.commit_ids
 
     def test_local_clone_fails_if_target_exists(self):
         with pytest.raises(RepositoryError):
             self.repo._local_clone(self.repo.path, 'master')
 
     def test_local_fetch(self):
         target_repo = self.get_empty_repo()
         source_repo = self.get_clone_repo()
 
         # Create a new branch in source repo
         master_commit = source_repo.commit_ids[-1]
         new_branch_commit = source_repo.commit_ids[-3]
         source_repo._checkout(new_branch_commit)
         source_repo._checkout('new_branch', create=True)
 
         target_repo._local_fetch(source_repo.path, 'new_branch')
         assert target_repo._last_fetch_heads() == [new_branch_commit]
 
         target_repo._local_fetch(source_repo.path, 'master')
         assert target_repo._last_fetch_heads() == [master_commit]
 
     def test_local_fetch_from_bare_repo(self):
         target_repo = self.get_empty_repo()
         target_repo._local_fetch(self.repo.path, 'master')
 
         master_commit = self.repo.commit_ids[-1]
         assert target_repo._last_fetch_heads() == [master_commit]
 
     def test_local_fetch_from_same_repo(self):
         with pytest.raises(ValueError):
             self.repo._local_fetch(self.repo.path, 'master')
 
     def test_local_fetch_branch_does_not_exist(self):
         target_repo = self.get_empty_repo()
 
         with pytest.raises(RepositoryError):
             target_repo._local_fetch(self.repo.path, 'new_branch')
 
     def test_local_pull(self):
         target_repo = self.get_empty_repo()
         source_repo = self.get_clone_repo()
 
         # Create a new branch in source repo
         master_commit = source_repo.commit_ids[-1]
         new_branch_commit = source_repo.commit_ids[-3]
         source_repo._checkout(new_branch_commit)
         source_repo._checkout('new_branch', create=True)
 
         target_repo._local_pull(source_repo.path, 'new_branch')
         target_repo = GitRepository(target_repo.path)
         assert target_repo.head == new_branch_commit
 
         target_repo._local_pull(source_repo.path, 'master')
         target_repo = GitRepository(target_repo.path)
         assert target_repo.head == master_commit
 
     def test_local_pull_in_bare_repo(self):
         with pytest.raises(RepositoryError):
             self.repo._local_pull(self.repo.path, 'master')
 
     def test_local_merge(self):
         target_repo = self.get_empty_repo()
         source_repo = self.get_clone_repo()
 
         # Create a new branch in source repo
         master_commit = source_repo.commit_ids[-1]
         new_branch_commit = source_repo.commit_ids[-3]
         source_repo._checkout(new_branch_commit)
         source_repo._checkout('new_branch', create=True)
 
         # This is required as one cannot do a -ff-only merge in an empty repo.
         target_repo._local_pull(source_repo.path, 'new_branch')
 
         target_repo._local_fetch(source_repo.path, 'master')
         merge_message = 'Merge message\n\nDescription:...'
         user_name = 'Albert Einstein'
         user_email = '[e-mail]'
         target_repo._local_merge(merge_message, user_name, user_email,
                                  target_repo._last_fetch_heads())
 
         target_repo = GitRepository(target_repo.path)
         assert target_repo.commit_ids[-2] == master_commit
         last_commit = target_repo.get_commit(target_repo.head)
         assert last_commit.message.strip() == merge_message
         assert last_commit.author == '%s <%s>' % (user_name, user_email)
 
         assert not os.path.exists(
             os.path.join(target_repo.path, '.git', 'MERGE_HEAD'))
 
     def test_local_merge_raises_exception_on_conflict(self, vcsbackend_git):
         target_repo = vcsbackend_git.create_repo(number_of_commits=1)
         vcsbackend_git.ensure_file('README', 'I will conflict with you!!!')
 
         target_repo._local_fetch(self.repo.path, 'master')
         with pytest.raises(RepositoryError):
             target_repo._local_merge(
                 'merge_message', 'user name', '[e-mail]',
                 target_repo._last_fetch_heads())
 
         # Check we are not left in an intermediate merge state
         assert not os.path.exists(
             os.path.join(target_repo.path, '.git', 'MERGE_HEAD'))
 
     def test_local_merge_into_empty_repo(self):
         target_repo = self.get_empty_repo()
 
         # This is required as one cannot do a -ff-only merge in an empty repo.
         target_repo._local_fetch(self.repo.path, 'master')
         with pytest.raises(RepositoryError):
             target_repo._local_merge(
                 'merge_message', 'user name', '[e-mail]',
                 target_repo._last_fetch_heads())
 
     def test_local_merge_in_bare_repo(self):
         with pytest.raises(RepositoryError):
             self.repo._local_merge(
                 'merge_message', 'user name', '[e-mail]', None)
 
     def test_local_push_non_bare(self):
         target_repo = self.get_empty_repo()
 
         pushed_branch = 'pushed_branch'
         self.repo._local_push('master', target_repo.path, pushed_branch)
         # Fix the HEAD of the target repo, or otherwise GitRepository won't
         # report any branches.
         with open(os.path.join(target_repo.path, '.git', 'HEAD'), 'w') as f:
             f.write('ref: refs/heads/%s' % pushed_branch)
 
         target_repo = GitRepository(target_repo.path)
 
         assert (target_repo.branches[pushed_branch] ==
                 self.repo.branches['master'])
 
     def test_local_push_bare(self):
         target_repo = self.get_empty_repo(bare=True)
 
         pushed_branch = 'pushed_branch'
         self.repo._local_push('master', target_repo.path, pushed_branch)
         # Fix the HEAD of the target repo, or otherwise GitRepository won't
         # report any branches.
         with open(os.path.join(target_repo.path, 'HEAD'), 'w') as f:
             f.write('ref: refs/heads/%s' % pushed_branch)
 
         target_repo = GitRepository(target_repo.path)
 
         assert (target_repo.branches[pushed_branch] ==
                 self.repo.branches['master'])
 
     def test_local_push_non_bare_target_branch_is_checked_out(self):
         target_repo = self.get_clone_repo()
 
         pushed_branch = 'pushed_branch'
         # Create a new branch in source repo
         new_branch_commit = target_repo.commit_ids[-3]
         target_repo._checkout(new_branch_commit)
         target_repo._checkout(pushed_branch, create=True)
 
         self.repo._local_push('master', target_repo.path, pushed_branch)
 
         target_repo = GitRepository(target_repo.path)
 
         assert (target_repo.branches[pushed_branch] ==
                 self.repo.branches['master'])
 
     def test_local_push_raises_exception_on_conflict(self, vcsbackend_git):
         target_repo = vcsbackend_git.create_repo(number_of_commits=1)
         with pytest.raises(RepositoryError):
             self.repo._local_push('master', target_repo.path, 'master')
 
     def test_hooks_can_be_enabled_via_env_variable_for_local_push(self):
         target_repo = self.get_empty_repo(bare=True)
 
         with mock.patch.object(self.repo, 'run_git_command') as run_mock:
             self.repo._local_push(
                 'master', target_repo.path, 'master', enable_hooks=True)
         env = run_mock.call_args[1]['extra_env']
         assert 'RC_SKIP_HOOKS' not in env
 
     def _add_failing_hook(self, repo_path, hook_name, bare=False):
         path_components = (
             ['hooks', hook_name] if bare else ['.git', 'hooks', hook_name])
         hook_path = os.path.join(repo_path, *path_components)
         with open(hook_path, 'w') as f:
             script_lines = [
                 '#!%s' % sys.executable,
                 'import os',
                 'import sys',
                 'if os.environ.get("RC_SKIP_HOOKS"):',
                 '    sys.exit(0)',
                 'sys.exit(1)',
             ]
             f.write('\n'.join(script_lines))
         os.chmod(hook_path, 0755)
 
     def test_local_push_does_not_execute_hook(self):
         target_repo = self.get_empty_repo()
 
         pushed_branch = 'pushed_branch'
         self._add_failing_hook(target_repo.path, 'pre-receive')
         self.repo._local_push('master', target_repo.path, pushed_branch)
         # Fix the HEAD of the target repo, or otherwise GitRepository won't
         # report any branches.
         with open(os.path.join(target_repo.path, '.git', 'HEAD'), 'w') as f:
             f.write('ref: refs/heads/%s' % pushed_branch)
 
         target_repo = GitRepository(target_repo.path)
 
         assert (target_repo.branches[pushed_branch] ==
                 self.repo.branches['master'])
 
     def test_local_push_executes_hook(self):
         target_repo = self.get_empty_repo(bare=True)
         self._add_failing_hook(target_repo.path, 'pre-receive', bare=True)
         with pytest.raises(RepositoryError):
             self.repo._local_push(
                 'master', target_repo.path, 'master', enable_hooks=True)
 
     def test_maybe_prepare_merge_workspace(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             'pr2', Reference('branch', 'master', 'unused'))
 
         assert os.path.isdir(workspace)
         workspace_repo = GitRepository(workspace)
         assert workspace_repo.branches == self.repo.branches
 
         # Calling it a second time should also succeed
         workspace = self.repo._maybe_prepare_merge_workspace(
             'pr2', Reference('branch', 'master', 'unused'))
         assert os.path.isdir(workspace)
 
     def test_cleanup_merge_workspace(self):
         workspace = self.repo._maybe_prepare_merge_workspace(
             'pr3', Reference('branch', 'master', 'unused'))
         self.repo.cleanup_merge_workspace('pr3')
 
         assert not os.path.exists(workspace)
 
     def test_cleanup_merge_workspace_invalid_workspace_id(self):
         # No assert: because in case of an inexistent workspace this function
         # should still succeed.
         self.repo.cleanup_merge_workspace('pr4')
 
     def test_set_refs(self):
         test_ref = 'refs/test-refs/abcde'
         test_commit_id = 'ecb86e1f424f2608262b130db174a7dfd25a6623'
 
         self.repo.set_refs(test_ref, test_commit_id)
         stdout, _ = self.repo.run_git_command(['show-ref'])
         assert test_ref in stdout
         assert test_commit_id in stdout
 
     def test_remove_ref(self):
         test_ref = 'refs/test-refs/abcde'
         test_commit_id = 'ecb86e1f424f2608262b130db174a7dfd25a6623'
         self.repo.set_refs(test_ref, test_commit_id)
         stdout, _ = self.repo.run_git_command(['show-ref'])
         assert test_ref in stdout
         assert test_commit_id in stdout
 
         self.repo.remove_ref(test_ref)
         stdout, _ = self.repo.run_git_command(['show-ref'])
         assert test_ref not in stdout
         assert test_commit_id not in stdout
 
 
 class TestGitCommit(object):
 
     @pytest.fixture(autouse=True)
     def prepare(self):
         self.repo = GitRepository(TEST_GIT_REPO)
 
     def test_default_commit(self):
         tip = self.repo.get_commit()
         assert tip == self.repo.get_commit(None)
         assert tip == self.repo.get_commit('tip')
 
     def test_root_node(self):
         tip = self.repo.get_commit()
         assert tip.root is tip.get_node('')
 
     def test_lazy_fetch(self):
         """
         Test if commit's nodes expands and are cached as we walk through
         the commit. This test is somewhat hard to write as order of tests
         is a key here. Written by running command after command in a shell.
         """
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         assert commit_id in self.repo.commit_ids
         commit = self.repo.get_commit(commit_id)
         assert len(commit.nodes) == 0
         root = commit.root
         assert len(commit.nodes) == 1
         assert len(root.nodes) == 8
         # accessing root.nodes updates commit.nodes
         assert len(commit.nodes) == 9
 
         docs = root.get_node('docs')
         # we haven't yet accessed anything new as docs dir was already cached
         assert len(commit.nodes) == 9
         assert len(docs.nodes) == 8
         # accessing docs.nodes updates commit.nodes
         assert len(commit.nodes) == 17
 
         assert docs is commit.get_node('docs')
         assert docs is root.nodes[0]
         assert docs is root.dirs[0]
         assert docs is commit.get_node('docs')
 
     def test_nodes_with_commit(self):
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         commit = self.repo.get_commit(commit_id)
         root = commit.root
         docs = root.get_node('docs')
         assert docs is commit.get_node('docs')
         api = docs.get_node('api')
         assert api is commit.get_node('docs/api')
         index = api.get_node('index.rst')
         assert index is commit.get_node('docs/api/index.rst')
         assert index is commit.get_node('docs')\
             .get_node('api')\
             .get_node('index.rst')
 
     def test_branch_and_tags(self):
         """
         rev0 = self.repo.commit_ids[0]
         commit0 = self.repo.get_commit(rev0)
         assert commit0.branch == 'master'
         assert commit0.tags == []
 
         rev10 = self.repo.commit_ids[10]
         commit10 = self.repo.get_commit(rev10)
         assert commit10.branch == 'master'
         assert commit10.tags == []
 
         rev44 = self.repo.commit_ids[44]
         commit44 = self.repo.get_commit(rev44)
         assert commit44.branch == 'web-branch'
 
         tip = self.repo.get_commit('tip')
         assert 'tip' in tip.tags
         """
         # Those tests would fail - branches are now going
         # to be changed at main API in order to support git backend
         pass
 
     def test_file_size(self):
         to_check = (
             ('c1214f7e79e02fc37156ff215cd71275450cffc3',
                 'vcs/backends/BaseRepository.py', 502),
             ('d7e0d30fbcae12c90680eb095a4f5f02505ce501',
                 'vcs/backends/hg.py', 854),
             ('6e125e7c890379446e98980d8ed60fba87d0f6d1',
                 'setup.py', 1068),
 
             ('d955cd312c17b02143c04fa1099a352b04368118',
                 'vcs/backends/base.py', 2921),
             ('ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
                 'vcs/backends/base.py', 3936),
             ('f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'vcs/backends/base.py', 6189),
         )
         for commit_id, path, size in to_check:
             node = self.repo.get_commit(commit_id).get_node(path)
             assert node.is_file()
             assert node.size == size
 
     def test_file_history_from_commits(self):
         node = self.repo[10].get_node('setup.py')
         commit_ids = [commit.raw_id for commit in node.history]
         assert ['ff7ca51e58c505fec0dd2491de52c622bb7a806b'] == commit_ids
 
         node = self.repo[20].get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b'] == node_ids
 
         # special case we check history from commit that has this particular
         # file changed this means we check if it's included as well
         node = self.repo.get_commit('191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e') \
             .get_node('setup.py')
         node_ids = [commit.raw_id for commit in node.history]
         assert ['191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b'] == node_ids
 
     def test_file_history(self):
         # we can only check if those commits are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [
                 '54386793436c938cff89326944d4c2702340037d',
                 '51d254f0ecf5df2ce50c0b115741f4cf13985dab',
                 '998ed409c795fec2012b1c0ca054d99888b22090',
                 '5e0eb4c47f56564395f76333f319d26c79e2fb09',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '7cb3fd1b6d8c20ba89e2264f1c8baebc8a52d36e',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/nodes.py': [
                 '33fa3223355104431402a888fa77a4e9956feb3e',
                 'fa014c12c26d10ba682fadb78f2a11c24c8118e1',
                 'e686b958768ee96af8029fe19c6050b1a8dd3b2b',
                 'ab5721ca0a081f26bf43d9051e615af2cc99952f',
                 'c877b68d18e792a66b7f4c529ea02c8f80801542',
                 '4313566d2e417cb382948f8d9d7c765330356054',
                 '6c2303a793671e807d1cfc70134c9ca0767d98c2',
                 '54386793436c938cff89326944d4c2702340037d',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
                 '1c6b3677b37ea064cb4b51714d8f7498f93f4b2b',
                 '2d03ca750a44440fb5ea8b751176d1f36f8e8f46',
                 '2a08b128c206db48c2f0b8f70df060e6db0ae4f8',
                 '30c26513ff1eb8e5ce0e1c6b477ee5dc50e2f34b',
                 'ac71e9503c2ca95542839af0ce7b64011b72ea7c',
                 '12669288fd13adba2a9b7dd5b870cc23ffab92d2',
                 '5a0c84f3e6fe3473e4c8427199d5a6fc71a9b382',
                 '12f2f5e2b38e6ff3fbdb5d722efed9aa72ecb0d5',
                 '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                 'f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'd7e390a45f6aa96f04f5e7f583ad4f867431aa25',
                 'f15c21f97864b4f071cddfbf2750ec2e23859414',
                 'e906ef056cf539a4e4e5fc8003eaf7cf14dd8ade',
                 'ea2b108b48aa8f8c9c4a941f66c1a03315ca1c3b',
                 '84dec09632a4458f79f50ddbbd155506c460b4f9',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '3bf1c5868e570e39569d094f922d33ced2fa3b2b',
                 'b8d04012574729d2c29886e53b1a43ef16dd00a1',
                 '6970b057cffe4aab0a792aa634c89f4bebf01441',
                 'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/backends/git.py': [
                 '4cf116ad5a457530381135e2f4c453e68a1b0105',
                 '9a751d84d8e9408e736329767387f41b36935153',
                 'cb681fb539c3faaedbcdf5ca71ca413425c18f01',
                 '428f81bb652bcba8d631bce926e8834ff49bdcc6',
                 '180ab15aebf26f98f714d8c68715e0f05fa6e1c7',
                 '2b8e07312a2e89e92b90426ab97f349f4bce2a3a',
                 '50e08c506174d8645a4bb517dd122ac946a0f3bf',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
             ],
         }
         for path, commit_ids in files.items():
             node = self.repo.get_commit(commit_ids[0]).get_node(path)
             node_ids = [commit.raw_id for commit in node.history]
             assert set(commit_ids).issubset(set(node_ids)), (
                 "We assumed that %s is subset of commit_ids for which file %s "
                 "has been changed, and history of that node returned: %s"
                 % (commit_ids, path, node_ids))
 
     def test_file_annotate(self):
         files = {
             'vcs/backends/__init__.py': {
                 'c1214f7e79e02fc37156ff215cd71275450cffc3': {
                     'lines_no': 1,
                     'commits': [
                         'c1214f7e79e02fc37156ff215cd71275450cffc3',
                     ],
                 },
                 '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647': {
                     'lines_no': 21,
                     'commits': [
                         '49d3fd156b6f7db46313fac355dca1a0b94a0017',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
                 'e29b67bd158580fc90fc5e9111240b90e6e86064': {
                     'lines_no': 32,
                     'commits': [
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '54000345d2e78b03a99d561399e8e548de3f3203',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
             },
         }
 
         for fname, commit_dict in files.items():
             for commit_id, __ in commit_dict.items():
                 commit = self.repo.get_commit(commit_id)
 
                 l1_1 = [x[1] for x in commit.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in commit.get_file_annotate(fname)]
                 assert l1_1 == l1_2
                 l1 = l1_1
                 l2 = files[fname][commit_id]['commits']
                 assert l1 == l2, (
                     "The lists of commit_ids for %s@commit_id %s"
                     "from annotation list should match each other, "
                     "got \n%s \nvs \n%s " % (fname, commit_id, l1, l2))
 
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         node = self.repo\
             .get_commit('e6ea6d16e2f26250124a1f4b4fe37a912f9d86a0')\
             .get_node('vcs/utils/diffs.py')
         assert node.state, NodeState.ADDED
         assert node.added
         assert not node.changed
         assert not node.not_changed
         assert not node.removed
 
         node = self.repo\
             .get_commit('33fa3223355104431402a888fa77a4e9956feb3e')\
             .get_node('.hgignore')
         assert node.state, NodeState.CHANGED
         assert not node.added
         assert node.changed
         assert not node.not_changed
         assert not node.removed
 
         node = self.repo\
             .get_commit('e29b67bd158580fc90fc5e9111240b90e6e86064')\
             .get_node('setup.py')
         assert node.state, NodeState.NOT_CHANGED
         assert not node.added
         assert not node.changed
         assert node.not_changed
         assert not node.removed
 
         # If node has REMOVED state then trying to fetch it would raise
         # CommitError exception
         commit = self.repo.get_commit(
             'fa6600f6848800641328adbf7811fd2372c02ab2')
         path = 'vcs/backends/BaseRepository.py'
         with pytest.raises(NodeDoesNotExistError):
             commit.get_node(path)
         # but it would be one of ``removed`` (commit's attribute)
         assert path in [rf.path for rf in commit.removed]
 
         commit = self.repo.get_commit(
             '54386793436c938cff89326944d4c2702340037d')
         changed = [
             'setup.py', 'tests/test_nodes.py', 'vcs/backends/hg.py',
             'vcs/nodes.py']
         assert set(changed) == set([f.path for f in commit.changed])
 
     def test_unicode_branch_refs(self):
         unicode_branches = {
             'refs/heads/unicode': '6c0ce52b229aa978889e91b38777f800e85f330b',
             u'refs/heads/uniçö∂e': 'ürl',
         }
         with mock.patch(
             ("rhodecode.lib.vcs.backends.git.repository"
                 ".GitRepository._refs"),
                 unicode_branches):
             branches = self.repo.branches
 
         assert 'unicode' in branches
         assert u'uniçö∂e' in branches
 
     def test_unicode_tag_refs(self):
         unicode_tags = {
             'refs/tags/unicode': '6c0ce52b229aa978889e91b38777f800e85f330b',
             u'refs/tags/uniçö∂e': '6c0ce52b229aa978889e91b38777f800e85f330b',
         }
         with mock.patch(
             ("rhodecode.lib.vcs.backends.git.repository"
                 ".GitRepository._refs"),
                 unicode_tags):
             tags = self.repo.tags
 
         assert 'unicode' in tags
         assert u'uniçö∂e' in tags
 
     def test_commit_message_is_unicode(self):
         for commit in self.repo:
             assert type(commit.message) == unicode
 
     def test_commit_author_is_unicode(self):
         for commit in self.repo:
             assert type(commit.author) == unicode
 
     def test_repo_files_content_is_unicode(self):
         commit = self.repo.get_commit()
         for node in commit.get_node('/'):
             if node.is_file():
                 assert type(node.content) == unicode
 
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         tip = self.repo.get_commit()
         with pytest.raises(VCSError):
             tip.get_node(path)
 
     @pytest.mark.parametrize("author_email, commit_id", [
         ('[e-mail]', 'c1214f7e79e02fc37156ff215cd71275450cffc3'),
         ('[e-mail]',
          'ff7ca51e58c505fec0dd2491de52c622bb7a806b'),
         ('none@none', '8430a588b43b5d6da365400117c89400326e7992'),
     ])
     def test_author_email(self, author_email, commit_id):
         commit = self.repo.get_commit(commit_id)
         assert author_email == commit.author_email
 
     @pytest.mark.parametrize("author, commit_id", [
         ('Marcin Kuzminski', 'c1214f7e79e02fc37156ff215cd71275450cffc3'),
         ('Lukasz Balcerzak', 'ff7ca51e58c505fec0dd2491de52c622bb7a806b'),
         ('marcink', '8430a588b43b5d6da365400117c89400326e7992'),
     ])
     def test_author_username(self, author, commit_id):
         commit = self.repo.get_commit(commit_id)
         assert author == commit.author_name
 
 
 class TestLargeFileRepo(object):
 
     def test_large_file(self, backend_git):
         conf = make_db_config()
         repo = backend_git.create_test_repo('largefiles', conf)
 
         tip = repo.scm_instance().get_commit()
 
         # extract stored LF node into the origin cache
         lfs_store = os.path.join(repo.repo_path, repo.repo_name, 'lfs_store')
 
         oid = '7b331c02e313c7599d5a90212e17e6d3cb729bd2e1c9b873c302a63c95a2f9bf'
         oid_path = os.path.join(lfs_store, oid)
         oid_destination = os.path.join(
             conf.get('vcs_git_lfs', 'store_location'), oid)
         shutil.copy(oid_path, oid_destination)
 
         node = tip.get_node('1MB.zip')
 
         lf_node = node.get_largefile_node()
 
         assert lf_node.is_largefile() is True
         assert lf_node.size == 1024000
         assert lf_node.name == '1MB.zip'
 
 
[e-mail]("vcs_repository_support")
 class TestGitSpecificWithRepo(BackendTestMixin):
 
     @classmethod
     def _get_commits(cls):
         return [
             {
                 'message': 'Initial',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar/static/js/admin/base.js', content='base'),
                     FileNode(
                         'foobar/static/admin', content='admin',
                         mode=0120000),  # this is a link
                     FileNode('foo', content='foo'),
                 ],
             },
             {
                 'message': 'Second',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('foo2', content='foo2'),
                 ],
             },
         ]
 
     def test_paths_slow_traversing(self):
         commit = self.repo.get_commit()
         assert commit.get_node('foobar').get_node('static').get_node('js')\
             .get_node('admin').get_node('base.js').content == 'base'
 
     def test_paths_fast_traversing(self):
         commit = self.repo.get_commit()
         assert (
             commit.get_node('foobar/static/js/admin/base.js').content ==
             'base')
 
     def test_get_diff_runs_git_command_with_hashes(self):
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(self.repo[0], self.repo[1])
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M',
              '--abbrev=40', self.repo._get_commit_id(0),
              self.repo._get_commit_id(1)])
 
     def test_get_diff_runs_git_command_with_str_hashes(self):
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(self.repo.EMPTY_COMMIT, self.repo[1])
         self.repo.run_git_command.assert_called_once_with(
             ['show', '-U3', '--full-index', '--binary', '-p', '-M',
              '--abbrev=40', self.repo._get_commit_id(1)])
 
     def test_get_diff_runs_git_command_with_path_if_its_given(self):
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(self.repo[0], self.repo[1], 'foo')
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M',
              '--abbrev=40', self.repo._get_commit_id(0),
              self.repo._get_commit_id(1), '--', 'foo'])
 
 
[e-mail]("vcs_repository_support")
 class TestGitRegression(BackendTestMixin):
 
     @classmethod
     def _get_commits(cls):
         return [
             {
                 'message': 'Initial',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('bot/__init__.py', content='base'),
                     FileNode('bot/templates/404.html', content='base'),
                     FileNode('bot/templates/500.html', content='base'),
                 ],
             },
             {
                 'message': 'Second',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('bot/build/migrations/1.py', content='foo2'),
                     FileNode('bot/build/migrations/2.py', content='foo2'),
                     FileNode(
                         'bot/build/static/templates/f.html', content='foo2'),
                     FileNode(
                         'bot/build/static/templates/f1.html', content='foo2'),
                     FileNode('bot/build/templates/err.html', content='foo2'),
                     FileNode('bot/build/templates/err2.html', content='foo2'),
                 ],
             },
         ]
 
     @pytest.mark.parametrize("path, expected_paths", [
         ('bot', [
             'bot/build',
             'bot/templates',
             'bot/__init__.py']),
         ('bot/build', [
             'bot/build/migrations',
             'bot/build/static',
             'bot/build/templates']),
         ('bot/build/static', [
             'bot/build/static/templates']),
         ('bot/build/static/templates', [
             'bot/build/static/templates/f.html',
             'bot/build/static/templates/f1.html']),
         ('bot/build/templates', [
             'bot/build/templates/err.html',
             'bot/build/templates/err2.html']),
         ('bot/templates/', [
             'bot/templates/404.html',
             'bot/templates/500.html']),
     ])
     def test_similar_paths(self, path, expected_paths):
         commit = self.repo.get_commit()
         paths = [n.path for n in commit.get_nodes(path)]
         assert paths == expected_paths
 
 
 class TestDiscoverGitVersion:
 
     def test_returns_git_version(self, baseapp):
         version = discover_git_version()
         assert version
 
     def test_returns_empty_string_without_vcsserver(self):
         mock_connection = mock.Mock()
         mock_connection.discover_git_version = mock.Mock(
             side_effect=Exception)
         with mock.patch('rhodecode.lib.vcs.connection.Git', mock_connection):
             version = discover_git_version()
         assert version == ''
 
 
 class TestGetSubmoduleUrl(object):
     def test_submodules_file_found(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         node = mock.Mock()
         with mock.patch.object(
                 commit, 'get_node', return_value=node) as get_node_mock:
             node.content = (
                 '[submodule "subrepo1"]\n'
                 '\tpath = subrepo1\n'
                 '\turl = https://code.rhodecode.com/dulwich\n'
             )
             result = commit._get_submodule_url('subrepo1')
         get_node_mock.assert_called_once_with('.gitmodules')
         assert result == 'https://code.rhodecode.com/dulwich'
 
     def test_complex_submodule_path(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         node = mock.Mock()
         with mock.patch.object(
                 commit, 'get_node', return_value=node) as get_node_mock:
             node.content = (
                 '[submodule "complex/subrepo/path"]\n'
                 '\tpath = complex/subrepo/path\n'
                 '\turl = https://code.rhodecode.com/dulwich\n'
             )
             result = commit._get_submodule_url('complex/subrepo/path')
         get_node_mock.assert_called_once_with('.gitmodules')
         assert result == 'https://code.rhodecode.com/dulwich'
 
     def test_submodules_file_not_found(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         with mock.patch.object(
                 commit, 'get_node', side_effect=NodeDoesNotExistError):
             result = commit._get_submodule_url('complex/subrepo/path')
         assert result is None
 
     def test_path_not_found(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         node = mock.Mock()
         with mock.patch.object(
                 commit, 'get_node', return_value=node) as get_node_mock:
             node.content = (
                 '[submodule "subrepo1"]\n'
                 '\tpath = subrepo1\n'
                 '\turl = https://code.rhodecode.com/dulwich\n'
             )
             result = commit._get_submodule_url('subrepo2')
         get_node_mock.assert_called_once_with('.gitmodules')
         assert result is None
 
     def test_returns_cached_values(self):
         commit = GitCommit(repository=mock.Mock(), raw_id='abcdef12', idx=1)
         node = mock.Mock()
         with mock.patch.object(
                 commit, 'get_node', return_value=node) as get_node_mock:
             node.content = (
                 '[submodule "subrepo1"]\n'
                 '\tpath = subrepo1\n'
                 '\turl = https://code.rhodecode.com/dulwich\n'
             )
             for _ in range(3):
                 commit._get_submodule_url('subrepo1')
         get_node_mock.assert_called_once_with('.gitmodules')
 
     def test_get_node_returns_a_link(self):
         repository = mock.Mock()
         repository.alias = 'git'
         commit = GitCommit(repository=repository, raw_id='abcdef12', idx=1)
         submodule_url = 'https://code.rhodecode.com/dulwich'
         get_id_patch = mock.patch.object(
             commit, '_get_id_for_path', return_value=(1, 'link'))
         get_submodule_patch = mock.patch.object(
             commit, '_get_submodule_url', return_value=submodule_url)
 
         with get_id_patch, get_submodule_patch as submodule_mock:
             node = commit.get_node('/abcde')
 
         submodule_mock.assert_called_once_with('/abcde')
         assert type(node) == SubModuleNode
         assert node.url == submodule_url
 
     def test_get_nodes_returns_links(self):
         repository = mock.MagicMock()
         repository.alias = 'git'
         repository._remote.tree_items.return_value = [
             ('subrepo', 'stat', 1, 'link')
         ]
         commit = GitCommit(repository=repository, raw_id='abcdef12', idx=1)
         submodule_url = 'https://code.rhodecode.com/dulwich'
         get_id_patch = mock.patch.object(
             commit, '_get_id_for_path', return_value=(1, 'tree'))
         get_submodule_patch = mock.patch.object(
             commit, '_get_submodule_url', return_value=submodule_url)
 
         with get_id_patch, get_submodule_patch as submodule_mock:
             nodes = commit.get_nodes('/abcde')
 
         submodule_mock.assert_called_once_with('/abcde/subrepo')
         assert len(nodes) == 1
         assert type(nodes[0]) == SubModuleNode
         assert nodes[0].url == submodule_url

rhodecode/tests/vcs/test_inmemory.py

@@ -1,344 +1,345 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Tests so called "in memory commits" commit API of vcs.
 """
 import datetime
 
 import pytest
 
 from rhodecode.lib.utils2 import safe_unicode
 from rhodecode.lib.vcs.exceptions import (
     EmptyRepositoryError, NodeAlreadyAddedError, NodeAlreadyExistsError,
     NodeAlreadyRemovedError, NodeAlreadyChangedError, NodeDoesNotExistError,
     NodeNotChangedError)
 from rhodecode.lib.vcs.nodes import DirNode, FileNode
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
 @pytest.fixture
 def nodes():
     nodes = [
         FileNode('foobar', content='Foo & bar'),
         FileNode('foobar2', content='Foo & bar, doubled!'),
         FileNode('foo bar with spaces', content=''),
         FileNode('foo/bar/baz', content='Inside'),
         FileNode(
             'foo/bar/file.bin',
             content=(
                 '\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1\x00\x00\x00\x00\x00\x00'
                 '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00;\x00\x03\x00\xfe'
                 '\xff\t\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
                 '\x01\x00\x00\x00\x1a\x00\x00\x00\x00\x00\x00\x00\x00\x10\x00'
                 '\x00\x18\x00\x00\x00\x01\x00\x00\x00\xfe\xff\xff\xff\x00\x00'
                 '\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff'
                 '\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff'
             )
         ),
     ]
     return nodes
 
 
[e-mail]("vcs_repository_support")
 class TestInMemoryCommit(BackendTestMixin):
     """
     This is a backend independent test case class which should be created
     with ``type`` method.
 
     It is required to set following attributes at subclass:
 
     - ``backend_alias``: alias of used backend (see ``vcs.BACKENDS``)
     """
 
     @classmethod
     def _get_commits(cls):
         return []
 
     def test_add(self, nodes):
         for node in nodes:
             self.imc.add(node)
 
         self.commit()
         self.assert_succesful_commit(nodes)
 
     @pytest.mark.skip_backends(
         'svn', reason="Svn does not support commits on branches.")
     def test_add_on_branch(self, nodes):
         for node in nodes:
             self.imc.add(node)
         self.commit(branch=u'stable')
         self.assert_succesful_commit(nodes)
 
     def test_add_in_bulk(self, nodes):
         self.imc.add(*nodes)
 
         self.commit()
         self.assert_succesful_commit(nodes)
 
     def test_add_non_ascii_files(self):
         nodes = [
             FileNode('żółwik/zwierzątko_utf8_str', content='ćććć'),
             FileNode(u'żółwik/zwierzątko_unicode', content=u'ćććć'),
         ]
 
         for node in nodes:
             self.imc.add(node)
 
         self.commit()
         self.assert_succesful_commit(nodes)
 
     def commit(self, branch=None):
         self.old_commit_count = len(self.repo.commit_ids)
         self.commit_message = u'Test commit with unicode: żółwik'
         self.commit_author = unicode(self.__class__)
         self.commit = self.imc.commit(
             message=self.commit_message, author=self.commit_author,
             branch=branch)
 
     def test_add_actually_adds_all_nodes_at_second_commit_too(self):
         to_add = [
             FileNode('foo/bar/image.png', content='\0'),
             FileNode('foo/README.txt', content='readme!'),
         ]
         self.imc.add(*to_add)
         commit = self.imc.commit(u'Initial', u'[e-mail]')
         assert isinstance(commit.get_node('foo'), DirNode)
         assert isinstance(commit.get_node('foo/bar'), DirNode)
         self.assert_nodes_in_commit(commit, to_add)
 
         # commit some more files again
         to_add = [
             FileNode('foo/bar/foobaz/bar', content='foo'),
             FileNode('foo/bar/another/bar', content='foo'),
             FileNode('foo/baz.txt', content='foo'),
             FileNode('foobar/foobaz/file', content='foo'),
             FileNode('foobar/barbaz', content='foo'),
         ]
         self.imc.add(*to_add)
         commit = self.imc.commit(u'Another', u'[e-mail]')
         self.assert_nodes_in_commit(commit, to_add)
 
     def test_add_raise_already_added(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
         with pytest.raises(NodeAlreadyAddedError):
             self.imc.add(node)
 
     def test_check_integrity_raise_already_exist(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
         self.imc.commit(message=u'Added foobar', author=unicode(self))
         self.imc.add(node)
         with pytest.raises(NodeAlreadyExistsError):
             self.imc.commit(message='new message', author=str(self))
 
     def test_change(self):
         self.imc.add(FileNode('foo/bar/baz', content='foo'))
         self.imc.add(FileNode('foo/fbar', content='foobar'))
         tip = self.imc.commit(u'Initial', u'[e-mail]')
 
         # Change node's content
         node = FileNode('foo/bar/baz', content='My **changed** content')
         self.imc.change(node)
         self.imc.commit(u'Changed %s' % node.path, u'[e-mail]')
 
         newtip = self.repo.get_commit()
         assert tip != newtip
         assert tip.id != newtip.id
         self.assert_nodes_in_commit(newtip, (node,))
 
     def test_change_non_ascii(self):
         to_add = [
             FileNode('żółwik/zwierzątko', content='ćććć'),
             FileNode(u'żółwik/zwierzątko_uni', content=u'ćććć'),
         ]
         for node in to_add:
             self.imc.add(node)
 
         tip = self.imc.commit(u'Initial', u'[e-mail]')
 
         # Change node's content
         node = FileNode('żółwik/zwierzątko', content='My **changed** content')
         self.imc.change(node)
         self.imc.commit(u'Changed %s' % safe_unicode(node.path),
                         u'[e-mail]')
 
         node_uni = FileNode(
             u'żółwik/zwierzątko_uni', content=u'My **changed** content')
         self.imc.change(node_uni)
         self.imc.commit(u'Changed %s' % safe_unicode(node_uni.path),
                         u'[e-mail]')
 
         newtip = self.repo.get_commit()
         assert tip != newtip
         assert tip.id != newtip.id
 
         self.assert_nodes_in_commit(newtip, (node, node_uni))
 
     def test_change_raise_empty_repository(self):
         node = FileNode('foobar')
         with pytest.raises(EmptyRepositoryError):
             self.imc.change(node)
 
     def test_check_integrity_change_raise_node_does_not_exist(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
         self.imc.commit(message=u'Added foobar', author=unicode(self))
         node = FileNode('not-foobar', content='')
         self.imc.change(node)
         with pytest.raises(NodeDoesNotExistError):
             self.imc.commit(
                 message='Changed not existing node',
                 author=str(self))
 
     def test_change_raise_node_already_changed(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
         self.imc.commit(message=u'Added foobar', author=unicode(self))
         node = FileNode('foobar', content='more baz')
         self.imc.change(node)
         with pytest.raises(NodeAlreadyChangedError):
             self.imc.change(node)
 
     def test_check_integrity_change_raise_node_not_changed(self, nodes):
         self.test_add(nodes)  # Performs first commit
 
         node = FileNode(nodes[0].path, content=nodes[0].content)
         self.imc.change(node)
         with pytest.raises(NodeNotChangedError):
             self.imc.commit(
                 message=u'Trying to mark node as changed without touching it',
                 author=unicode(self))
 
     def test_change_raise_node_already_removed(self):
         node = FileNode('foobar', content='baz')
         self.imc.add(node)
         self.imc.commit(message=u'Added foobar', author=unicode(self))
         self.imc.remove(FileNode('foobar'))
         with pytest.raises(NodeAlreadyRemovedError):
             self.imc.change(node)
 
     def test_remove(self, nodes):
         self.test_add(nodes)  # Performs first commit
 
         tip = self.repo.get_commit()
         node = nodes[0]
         assert node.content == tip.get_node(node.path).content
         self.imc.remove(node)
         self.imc.commit(
             message=u'Removed %s' % node.path, author=unicode(self))
 
         newtip = self.repo.get_commit()
         assert tip != newtip
         assert tip.id != newtip.id
         with pytest.raises(NodeDoesNotExistError):
             newtip.get_node(node.path)
 
     def test_remove_last_file_from_directory(self):
         node = FileNode('omg/qwe/foo/bar', content='foobar')
         self.imc.add(node)
         self.imc.commit(u'added', u'joe doe')
 
         self.imc.remove(node)
         tip = self.imc.commit(u'removed', u'joe doe')
         with pytest.raises(NodeDoesNotExistError):
             tip.get_node('omg/qwe/foo/bar')
 
     def test_remove_raise_node_does_not_exist(self, nodes):
         self.imc.remove(nodes[0])
         with pytest.raises(NodeDoesNotExistError):
             self.imc.commit(
                 message='Trying to remove node at empty repository',
                 author=str(self))
 
     def test_check_integrity_remove_raise_node_does_not_exist(self, nodes):
         self.test_add(nodes)  # Performs first commit
 
         node = FileNode('no-such-file')
         self.imc.remove(node)
         with pytest.raises(NodeDoesNotExistError):
             self.imc.commit(
                 message=u'Trying to remove not existing node',
                 author=unicode(self))
 
     def test_remove_raise_node_already_removed(self, nodes):
         self.test_add(nodes)  # Performs first commit
 
         node = FileNode(nodes[0].path)
         self.imc.remove(node)
         with pytest.raises(NodeAlreadyRemovedError):
             self.imc.remove(node)
 
     def test_remove_raise_node_already_changed(self, nodes):
         self.test_add(nodes)  # Performs first commit
 
         node = FileNode(nodes[0].path, content='Bending time')
         self.imc.change(node)
         with pytest.raises(NodeAlreadyChangedError):
             self.imc.remove(node)
 
     def test_reset(self):
         self.imc.add(FileNode('foo', content='bar'))
         # self.imc.change(FileNode('baz', content='new'))
         # self.imc.remove(FileNode('qwe'))
         self.imc.reset()
         assert not any((self.imc.added, self.imc.changed, self.imc.removed))
 
     def test_multiple_commits(self):
         N = 3  # number of commits to perform
         last = None
         for x in xrange(N):
             fname = 'file%s' % str(x).rjust(5, '0')
             content = 'foobar\n' * x
             node = FileNode(fname, content=content)
             self.imc.add(node)
             commit = self.imc.commit(u"Commit no. %s" % (x + 1), author=u'vcs')
             assert last != commit
             last = commit
 
         # Check commit number for same repo
         assert len(self.repo.commit_ids) == N
 
         # Check commit number for recreated repo
         repo = self.Backend(self.repo_path)
         assert len(repo.commit_ids) == N
 
     def test_date_attr(self, local_dt_to_utc):
         node = FileNode('foobar.txt', content='Foobared!')
         self.imc.add(node)
         date = datetime.datetime(1985, 1, 30, 1, 45)
         commit = self.imc.commit(
             u"Committed at time when I was born ;-)",
             author=u'lb', date=date)
 
         assert commit.date == local_dt_to_utc(date)
 
     def assert_succesful_commit(self, added_nodes):
         newtip = self.repo.get_commit()
         assert self.commit == newtip
         assert self.old_commit_count + 1 == len(self.repo.commit_ids)
         assert newtip.message == self.commit_message
         assert newtip.author == self.commit_author
         assert not any((self.imc.added, self.imc.changed, self.imc.removed))
         self.assert_nodes_in_commit(newtip, added_nodes)
 
     def assert_nodes_in_commit(self, commit, nodes):
         for node in nodes:
             assert commit.get_node(node.path).content == node.content

rhodecode/tests/vcs/test_nodes.py

@@ -1,274 +1,275 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import stat
 
 import pytest
 
 from rhodecode.lib.vcs.nodes import DirNode
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.lib.vcs.nodes import Node
 from rhodecode.lib.vcs.nodes import NodeError
 from rhodecode.lib.vcs.nodes import NodeKind
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
 @pytest.fixture()
 def binary_filenode():
     def node_maker(filename):
         data = (
             "\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x10\x00\x00\x00"
             "\x10\x08\x06\x00\x00\x00\x1f??a\x00\x00\x00\x04gAMA\x00\x00\xaf?7"
             "\x05\x8a?\x00\x00\x00\x19tEXtSoftware\x00Adobe ImageReadyq?e<\x00"
             "\x00\x025IDAT8?\xa5\x93?K\x94Q\x14\x87\x9f\xf7?Q\x1bs4?\x03\x9a"
             "\xa8?B\x02\x8b$\x10[U;i\x13?6h?&h[?\"\x14j?\xa2M\x7fB\x14F\x9aQ?&"
             "\x842?\x0b\x89\"\x82??!?\x9c!\x9c2l??{N\x8bW\x9dY\xb4\t/\x1c?="
             "\x9b?}????\xa9*;9!?\x83\x91?[?\\v*?D\x04\'`EpNp\xa2X\'U?pVq\"Sw."
             "\x1e?\x08\x01D?jw????\xbc??7{|\x9b?\x89$\x01??W@\x15\x9c\x05q`Lt/"
             "\x97?\x94\xa1d?\x18~?\x18?\x18W[%\xb0?\x83??\x14\x88\x8dB?\xa6H"
             "\tL\tl\x19>/\x01`\xac\xabx?\x9cl\nx\xb0\x98\x07\x95\x88D$\"q["
             "\x19?d\x00(o\n\xa0??\x7f\xb9\xa4?\x1bF\x1f\x8e\xac\xa8?j??eUU}?.?"
             "\x9f\x8cE??x\x94??\r\xbdtoJU5\"0N\x10U?\x00??V\t\x02\x9f\x81?U?"
             "\x00\x9eM\xae2?r\x9b7\x83\x82\x8aP3????.?&\"?\xb7ZP \x0c<?O"
             "\xa5\t}\xb8?\x99\xa6?\x87?\x1di|/\xa0??0\xbe\x1fp?d&\x1a\xad"
             "\x95\x8a\x07?\t*\x10??b:?d?.\x13C\x8a?\x12\xbe\xbf\x8e?{???"
             "\x08?\x80\xa7\x13+d\x13>J?\x80\x15T\x95\x9a\x00??S\x8c\r?\xa1"
             "\x03\x07?\x96\x9b\xa7\xab=E??\xa4\xb3?\x19q??B\x91=\x8d??k?J"
             "\x0bV\"??\xf7x?\xa1\x00?\\.\x87\x87???\x02F@D\x99],??\x10#?X"
             "\xb7=\xb9\x10?Z\x1by???cI??\x1ag?\x92\xbc?T?t[\x92\x81?<_\x17~"
             "\x92\x88?H%?\x10Q\x02\x9f\n\x81qQ\x0bm?\x1bX?\xb1AK\xa6\x9e\xb9?u"
             "\xb2?1\xbe|/\x92M@\xa2!F?\xa9>\"\r<DT?>\x92\x8e?>\x9a9Qv\x127?a"
             "\xac?Y?8?:??]X???9\x80\xb7?u?\x0b#BZ\x8d=\x1d?p\x00\x00\x00\x00"
             "IEND\xaeB`\x82")
         return FileNode(filename, content=data)
     return node_maker
 
 
 class TestNodeBasics:
 
     @pytest.mark.parametrize("path", ['/foo', '/foo/bar'])
     @pytest.mark.parametrize(
         "kind", [NodeKind.FILE, NodeKind.DIR], ids=["FILE", "DIR"])
     def test_init_wrong_paths(self, path, kind):
         """
         Cannot innitialize Node objects with path with slash at the beginning.
         """
         with pytest.raises(NodeError):
             Node(path, kind)
 
     @pytest.mark.parametrize("path", ['path', 'some/path'])
     @pytest.mark.parametrize(
         "kind", [NodeKind.FILE, NodeKind.DIR], ids=["FILE", "DIR"])
     def test_name(self, path, kind):
         node = Node(path, kind)
         assert node.name == 'path'
 
     def test_name_root(self):
         node = Node('', NodeKind.DIR)
         assert node.name == ''
 
     def test_root_node_cannot_be_file(self):
         with pytest.raises(NodeError):
             Node('', NodeKind.FILE)
 
     def test_kind_setter(self):
         node = Node('', NodeKind.DIR)
         with pytest.raises(NodeError):
             node.kind = NodeKind.FILE
 
     def test_compare_equal(self):
         node1 = FileNode('test', content='')
         node2 = FileNode('test', content='')
         assert node1 == node2
         assert not node1 != node2
 
     def test_compare_unequal(self):
         node1 = FileNode('test', content='a')
         node2 = FileNode('test', content='b')
         assert node1 != node2
         assert not node1 == node2
 
     @pytest.mark.parametrize("node_path, expected_parent_path", [
         ('', ''),
         ('some/path/', 'some/'),
         ('some/longer/path/', 'some/longer/'),
     ])
     def test_parent_path_new(self, node_path, expected_parent_path):
         """
         Tests if node's parent path are properly computed.
         """
         node = Node(node_path, NodeKind.DIR)
         parent_path = node.get_parent_path()
         assert (parent_path.endswith('/') or
                 node.is_root() and parent_path == '')
         assert parent_path == expected_parent_path
 
     '''
     def _test_trailing_slash(self, path):
         if not path.endswith('/'):
             pytest.fail("Trailing slash tests needs paths to end with slash")
         for kind in NodeKind.FILE, NodeKind.DIR:
             with pytest.raises(NodeError):
                 Node(path=path, kind=kind)
 
     def test_trailing_slash(self):
         for path in ('/', 'foo/', 'foo/bar/', 'foo/bar/biz/'):
             self._test_trailing_slash(path)
     '''
 
     def test_is_file(self):
         node = Node('any', NodeKind.FILE)
         assert node.is_file()
 
         node = FileNode('any')
         assert node.is_file()
         with pytest.raises(AttributeError):
             node.nodes
 
     def test_is_dir(self):
         node = Node('any_dir', NodeKind.DIR)
         assert node.is_dir()
 
         node = DirNode('any_dir')
 
         assert node.is_dir()
         with pytest.raises(NodeError):
             node.content
 
     def test_dir_node_iter(self):
         nodes = [
             DirNode('docs'),
             DirNode('tests'),
             FileNode('bar'),
             FileNode('foo'),
             FileNode('readme.txt'),
             FileNode('setup.py'),
         ]
         dirnode = DirNode('', nodes=nodes)
         for node in dirnode:
             assert node == dirnode.get_node(node.path)
 
     def test_node_state(self):
         """
         Without link to commit nodes should raise NodeError.
         """
         node = FileNode('anything')
         with pytest.raises(NodeError):
             node.state
         node = DirNode('anything')
         with pytest.raises(NodeError):
             node.state
 
     def test_file_node_stat(self):
         node = FileNode('foobar', 'empty... almost')
         mode = node.mode  # default should be 0100644
         assert mode & stat.S_IRUSR
         assert mode & stat.S_IWUSR
         assert mode & stat.S_IRGRP
         assert mode & stat.S_IROTH
         assert not mode & stat.S_IWGRP
         assert not mode & stat.S_IWOTH
         assert not mode & stat.S_IXUSR
         assert not mode & stat.S_IXGRP
         assert not mode & stat.S_IXOTH
 
     def test_file_node_is_executable(self):
         node = FileNode('foobar', 'empty... almost', mode=0100755)
         assert node.is_executable
 
         node = FileNode('foobar', 'empty... almost', mode=0100500)
         assert node.is_executable
 
         node = FileNode('foobar', 'empty... almost', mode=0100644)
         assert not node.is_executable
 
     def test_file_node_is_not_symlink(self):
         node = FileNode('foobar', 'empty...')
         assert not node.is_link()
 
     def test_mimetype(self):
         py_node = FileNode('test.py')
         tar_node = FileNode('test.tar.gz')
 
         ext = 'CustomExtension'
 
         my_node2 = FileNode('myfile2')
         my_node2._mimetype = [ext]
 
         my_node3 = FileNode('myfile3')
         my_node3._mimetype = [ext, ext]
 
         assert py_node.mimetype == 'text/x-python'
         assert py_node.get_mimetype() == ('text/x-python', None)
 
         assert tar_node.mimetype == 'application/x-tar'
         assert tar_node.get_mimetype() == ('application/x-tar', 'gzip')
 
         with pytest.raises(NodeError):
             my_node2.get_mimetype()
 
         assert my_node3.mimetype == ext
         assert my_node3.get_mimetype() == [ext, ext]
 
     def test_lines_counts(self):
         lines = [
             'line1\n',
             'line2\n',
             'line3\n',
             '\n',
             '\n',
             'line4\n',
         ]
         py_node = FileNode('test.py', ''.join(lines))
 
         assert (len(lines), len(lines)) == py_node.lines()
         assert (len(lines), len(lines) - 2) == py_node.lines(count_empty=True)
 
     def test_lines_no_newline(self):
         py_node = FileNode('test.py', 'oneline')
 
         assert (1, 1) == py_node.lines()
         assert (1, 1) == py_node.lines(count_empty=True)
 
 
-class TestNodeContent:
+class TestNodeContent(object):
 
     def test_if_binary(self, binary_filenode):
         filenode = binary_filenode('calendar.jpg')
         assert filenode.is_binary
 
     def test_binary_line_counts(self, binary_filenode):
         tar_node = binary_filenode('archive.tar.gz')
         assert (0, 0) == tar_node.lines(count_empty=True)
 
     def test_binary_mimetype(self, binary_filenode):
         tar_node = binary_filenode('archive.tar.gz')
         assert tar_node.mimetype == 'application/x-tar'
 
 
[e-mail]("vcs_repository_support")
 class TestNodesCommits(BackendTestMixin):
 
     def test_node_last_commit(self, generate_repo_with_commits):
         repo = generate_repo_with_commits(20)
         last_commit = repo.get_commit()
 
         for x in xrange(3):
             node = last_commit.get_node('file_%s.txt' % x)
             assert node.last_commit == repo[x]

rhodecode/tests/vcs/test_repository.py

@@ -1,534 +1,537 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import datetime
 from urllib2 import URLError
 
 import mock
 import pytest
 
 from rhodecode.lib.vcs import backends
 from rhodecode.lib.vcs.backends.base import (
     Config, BaseInMemoryCommit, Reference, MergeResponse, MergeFailureReason)
 from rhodecode.lib.vcs.exceptions import VCSError, RepositoryError
 from rhodecode.lib.vcs.nodes import FileNode
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 
 
[e-mail]("vcs_repository_support")
 class TestRepositoryBase(BackendTestMixin):
     recreate_repo_per_test = False
 
     def test_init_accepts_unicode_path(self, tmpdir):
         path = unicode(tmpdir.join(u'unicode ä'))
         self.Backend(path, create=True)
 
     def test_init_accepts_str_path(self, tmpdir):
         path = str(tmpdir.join('str ä'))
         self.Backend(path, create=True)
 
     def test_init_fails_if_path_does_not_exist(self, tmpdir):
         path = unicode(tmpdir.join('i-do-not-exist'))
         with pytest.raises(VCSError):
             self.Backend(path)
 
     def test_init_fails_if_path_is_not_a_valid_repository(self, tmpdir):
         path = unicode(tmpdir.mkdir(u'unicode ä'))
         with pytest.raises(VCSError):
             self.Backend(path)
 
     def test_has_commits_attribute(self):
         self.repo.commit_ids
 
     def test_name(self):
         assert self.repo.name.startswith('vcs-test')
 
     @pytest.mark.backends("hg", "git")
     def test_has_default_branch_name(self):
         assert self.repo.DEFAULT_BRANCH_NAME is not None
 
     @pytest.mark.backends("svn")
     def test_has_no_default_branch_name(self):
         assert self.repo.DEFAULT_BRANCH_NAME is None
 
     def test_has_empty_commit(self):
         assert self.repo.EMPTY_COMMIT_ID is not None
         assert self.repo.EMPTY_COMMIT is not None
 
     def test_empty_changeset_is_deprecated(self):
         def get_empty_changeset(repo):
             return repo.EMPTY_CHANGESET
         pytest.deprecated_call(get_empty_changeset, self.repo)
 
     def test_bookmarks(self):
         assert len(self.repo.bookmarks) == 0
 
     # TODO: Cover two cases: Local repo path, remote URL
     def test_check_url(self):
         config = Config()
         assert self.Backend.check_url(self.repo.path, config)
 
     def test_check_url_invalid(self):
         config = Config()
         with pytest.raises(URLError):
             self.Backend.check_url(self.repo.path + "invalid", config)
 
     def test_get_contact(self):
         assert self.repo.contact
 
     def test_get_description(self):
         assert self.repo.description
 
     def test_get_hook_location(self):
         assert len(self.repo.get_hook_location()) != 0
 
     def test_last_change(self, local_dt_to_utc):
         assert self.repo.last_change >= local_dt_to_utc(
             datetime.datetime(2010, 1, 1, 21, 0))
 
     def test_last_change_in_empty_repository(self, vcsbackend, local_dt_to_utc):
         delta = datetime.timedelta(seconds=1)
 
         start = local_dt_to_utc(datetime.datetime.now())
         empty_repo = vcsbackend.create_repo()
         now = local_dt_to_utc(datetime.datetime.now())
         assert empty_repo.last_change >= start - delta
         assert empty_repo.last_change <= now + delta
 
     def test_repo_equality(self):
         assert self.repo == self.repo
 
     def test_repo_equality_broken_object(self):
         import copy
         _repo = copy.copy(self.repo)
         delattr(_repo, 'path')
         assert self.repo != _repo
 
     def test_repo_equality_other_object(self):
         class dummy(object):
             path = self.repo.path
         assert self.repo != dummy()
 
     def test_get_commit_is_implemented(self):
         self.repo.get_commit()
 
     def test_get_commits_is_implemented(self):
         commit_iter = iter(self.repo.get_commits())
         commit = next(commit_iter)
         assert commit.idx == 0
 
     def test_supports_iteration(self):
         repo_iter = iter(self.repo)
         commit = next(repo_iter)
         assert commit.idx == 0
 
     def test_in_memory_commit(self):
         imc = self.repo.in_memory_commit
         assert isinstance(imc, BaseInMemoryCommit)
 
     @pytest.mark.backends("hg")
     def test__get_url_unicode(self):
         url = u'/home/repos/malmö'
         assert self.repo._get_url(url)
 
 
[e-mail]("vcs_repository_support")
 class TestDeprecatedRepositoryAPI(BackendTestMixin):
     recreate_repo_per_test = False
 
     def test_revisions_is_deprecated(self):
         def get_revisions(repo):
             return repo.revisions
         pytest.deprecated_call(get_revisions, self.repo)
 
     def test_get_changeset_is_deprecated(self):
         pytest.deprecated_call(self.repo.get_changeset)
 
     def test_get_changesets_is_deprecated(self):
         pytest.deprecated_call(self.repo.get_changesets)
 
     def test_in_memory_changeset_is_deprecated(self):
         def get_imc(repo):
             return repo.in_memory_changeset
         pytest.deprecated_call(get_imc, self.repo)
 
 
 # TODO: these tests are incomplete, must check the resulting compare result for
 # correcteness
 class TestRepositoryCompare:
 
     @pytest.mark.parametrize('merge', [True, False])
     def test_compare_commits_of_same_repository(self, vcsbackend, merge):
         target_repo = vcsbackend.create_repo(number_of_commits=5)
         target_repo.compare(
             target_repo[1].raw_id, target_repo[3].raw_id, target_repo,
             merge=merge)
 
     @pytest.mark.xfail_backends('svn')
     @pytest.mark.parametrize('merge', [True, False])
     def test_compare_cloned_repositories(self, vcsbackend, merge):
         target_repo = vcsbackend.create_repo(number_of_commits=5)
         source_repo = vcsbackend.clone_repo(target_repo)
         assert target_repo != source_repo
 
         vcsbackend.add_file(source_repo, 'newfile', 'somecontent')
         source_commit = source_repo.get_commit()
 
         target_repo.compare(
             target_repo[1].raw_id, source_repo[3].raw_id, source_repo,
             merge=merge)
 
     @pytest.mark.xfail_backends('svn')
     @pytest.mark.parametrize('merge', [True, False])
     def test_compare_unrelated_repositories(self, vcsbackend, merge):
         orig = vcsbackend.create_repo(number_of_commits=5)
         unrelated = vcsbackend.create_repo(number_of_commits=5)
         assert orig != unrelated
 
         orig.compare(
             orig[1].raw_id, unrelated[3].raw_id, unrelated, merge=merge)
 
 
 class TestRepositoryGetCommonAncestor:
 
     def test_get_common_ancestor_from_same_repo_existing(self, vcsbackend):
         target_repo = vcsbackend.create_repo(number_of_commits=5)
 
         expected_ancestor = target_repo[2].raw_id
 
         assert target_repo.get_common_ancestor(
             commit_id1=target_repo[2].raw_id,
             commit_id2=target_repo[4].raw_id,
             repo2=target_repo
         ) == expected_ancestor
 
         assert target_repo.get_common_ancestor(
             commit_id1=target_repo[4].raw_id,
             commit_id2=target_repo[2].raw_id,
             repo2=target_repo
         ) == expected_ancestor
 
     @pytest.mark.xfail_backends("svn")
     def test_get_common_ancestor_from_cloned_repo_existing(self, vcsbackend):
         target_repo = vcsbackend.create_repo(number_of_commits=5)
         source_repo = vcsbackend.clone_repo(target_repo)
         assert target_repo != source_repo
 
         vcsbackend.add_file(source_repo, 'newfile', 'somecontent')
         source_commit = source_repo.get_commit()
 
         expected_ancestor = target_repo[4].raw_id
 
         assert target_repo.get_common_ancestor(
             commit_id1=target_repo[4].raw_id,
             commit_id2=source_commit.raw_id,
             repo2=source_repo
         ) == expected_ancestor
 
         assert target_repo.get_common_ancestor(
             commit_id1=source_commit.raw_id,
             commit_id2=target_repo[4].raw_id,
             repo2=target_repo
         ) == expected_ancestor
 
     @pytest.mark.xfail_backends("svn")
     def test_get_common_ancestor_from_unrelated_repo_missing(self, vcsbackend):
         original = vcsbackend.create_repo(number_of_commits=5)
         unrelated = vcsbackend.create_repo(number_of_commits=5)
         assert original != unrelated
 
         assert original.get_common_ancestor(
             commit_id1=original[0].raw_id,
             commit_id2=unrelated[0].raw_id,
             repo2=unrelated
         ) == None
 
         assert original.get_common_ancestor(
             commit_id1=original[-1].raw_id,
             commit_id2=unrelated[-1].raw_id,
             repo2=unrelated
         ) == None
 
 
 @pytest.mark.backends("git", "hg")
 class TestRepositoryMerge:
     def prepare_for_success(self, vcsbackend):
         self.target_repo = vcsbackend.create_repo(number_of_commits=1)
         self.source_repo = vcsbackend.clone_repo(self.target_repo)
         vcsbackend.add_file(self.target_repo, 'README_MERGE1', 'Version 1')
         vcsbackend.add_file(self.source_repo, 'README_MERGE2', 'Version 2')
         imc = self.source_repo.in_memory_commit
         imc.add(FileNode('file_x', content=self.source_repo.name))
         imc.commit(
             message=u'Automatic commit from repo merge test',
             author=u'Automatic')
         self.target_commit = self.target_repo.get_commit()
         self.source_commit = self.source_repo.get_commit()
         # This only works for Git and Mercurial
         default_branch = self.target_repo.DEFAULT_BRANCH_NAME
         self.target_ref = Reference(
             'branch', default_branch, self.target_commit.raw_id)
         self.source_ref = Reference(
             'branch', default_branch, self.source_commit.raw_id)
         self.workspace = 'test-merge'
 
     def prepare_for_conflict(self, vcsbackend):
         self.target_repo = vcsbackend.create_repo(number_of_commits=1)
         self.source_repo = vcsbackend.clone_repo(self.target_repo)
         vcsbackend.add_file(self.target_repo, 'README_MERGE', 'Version 1')
         vcsbackend.add_file(self.source_repo, 'README_MERGE', 'Version 2')
         self.target_commit = self.target_repo.get_commit()
         self.source_commit = self.source_repo.get_commit()
         # This only works for Git and Mercurial
         default_branch = self.target_repo.DEFAULT_BRANCH_NAME
         self.target_ref = Reference(
             'branch', default_branch, self.target_commit.raw_id)
         self.source_ref = Reference(
             'branch', default_branch, self.source_commit.raw_id)
         self.workspace = 'test-merge'
 
     def test_merge_success(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
 
         merge_response = self.target_repo.merge(
             self.target_ref, self.source_repo, self.source_ref, self.workspace,
             'test user', '[e-mail]', 'merge message 1',
             dry_run=False)
         expected_merge_response = MergeResponse(
             True, True, merge_response.merge_ref,
             MergeFailureReason.NONE)
         assert merge_response == expected_merge_response
 
         target_repo = backends.get_backend(vcsbackend.alias)(
             self.target_repo.path)
         target_commits = list(target_repo.get_commits())
         commit_ids = [c.raw_id for c in target_commits[:-1]]
         assert self.source_ref.commit_id in commit_ids
         assert self.target_ref.commit_id in commit_ids
 
         merge_commit = target_commits[-1]
         assert merge_commit.raw_id == merge_response.merge_ref.commit_id
         assert merge_commit.message.strip() == 'merge message 1'
         assert merge_commit.author == 'test user <[e-mail]>'
 
         # We call it twice so to make sure we can handle updates
         target_ref = Reference(
             self.target_ref.type, self.target_ref.name,
             merge_response.merge_ref.commit_id)
 
         merge_response = target_repo.merge(
             target_ref, self.source_repo, self.source_ref, self.workspace,
             'test user', '[e-mail]', 'merge message 2',
             dry_run=False)
         expected_merge_response = MergeResponse(
             True, True, merge_response.merge_ref,
             MergeFailureReason.NONE)
         assert merge_response == expected_merge_response
 
         target_repo = backends.get_backend(
             vcsbackend.alias)(self.target_repo.path)
         merge_commit = target_repo.get_commit(
             merge_response.merge_ref.commit_id)
         assert merge_commit.message.strip() == 'merge message 1'
         assert merge_commit.author == 'test user <[e-mail]>'
 
     def test_merge_success_dry_run(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
 
         merge_response = self.target_repo.merge(
             self.target_ref, self.source_repo, self.source_ref, self.workspace,
             dry_run=True)
 
         # We call it twice so to make sure we can handle updates
         merge_response_update = self.target_repo.merge(
             self.target_ref, self.source_repo, self.source_ref, self.workspace,
             dry_run=True)
 
         # Multiple merges may differ in their commit id. Therefore we set the
         # commit id to `None` before comparing the merge responses.
         merge_response = merge_response._replace(
             merge_ref=merge_response.merge_ref._replace(commit_id=None))
         merge_response_update = merge_response_update._replace(
             merge_ref=merge_response_update.merge_ref._replace(commit_id=None))
 
         assert merge_response == merge_response_update
         assert merge_response.possible is True
         assert merge_response.executed is False
         assert merge_response.merge_ref
         assert merge_response.failure_reason is MergeFailureReason.NONE
 
     @pytest.mark.parametrize('dry_run', [True, False])
     def test_merge_conflict(self, vcsbackend, dry_run):
         self.prepare_for_conflict(vcsbackend)
         expected_merge_response = MergeResponse(
             False, False, None, MergeFailureReason.MERGE_FAILED)
 
         merge_response = self.target_repo.merge(
             self.target_ref, self.source_repo, self.source_ref, self.workspace,
             'test_user', '[e-mail]', 'test message', dry_run=dry_run)
         assert merge_response == expected_merge_response
 
         # We call it twice so to make sure we can handle updates
         merge_response = self.target_repo.merge(
             self.target_ref, self.source_repo, self.source_ref, self.workspace,
             'test_user', '[e-mail]', 'test message', dry_run=dry_run)
         assert merge_response == expected_merge_response
 
     def test_merge_target_is_not_head(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
         expected_merge_response = MergeResponse(
             False, False, None, MergeFailureReason.TARGET_IS_NOT_HEAD)
 
         target_ref = Reference(
             self.target_ref.type, self.target_ref.name, '0' * 40)
 
         merge_response = self.target_repo.merge(
             target_ref, self.source_repo, self.source_ref, self.workspace,
             dry_run=True)
 
         assert merge_response == expected_merge_response
 
     def test_merge_missing_source_reference(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
         expected_merge_response = MergeResponse(
             False, False, None, MergeFailureReason.MISSING_SOURCE_REF)
 
         source_ref = Reference(
             self.source_ref.type, 'not_existing', self.source_ref.commit_id)
 
         merge_response = self.target_repo.merge(
             self.target_ref, self.source_repo, source_ref, self.workspace,
             dry_run=True)
 
         assert merge_response == expected_merge_response
 
     def test_merge_raises_exception(self, vcsbackend):
         self.prepare_for_success(vcsbackend)
         expected_merge_response = MergeResponse(
             False, False, None, MergeFailureReason.UNKNOWN)
 
         with mock.patch.object(self.target_repo, '_merge_repo',
                                side_effect=RepositoryError()):
             merge_response = self.target_repo.merge(
                 self.target_ref, self.source_repo, self.source_ref,
                 self.workspace, dry_run=True)
 
         assert merge_response == expected_merge_response
 
     def test_merge_invalid_user_name(self, vcsbackend):
         repo = vcsbackend.create_repo(number_of_commits=1)
         ref = Reference('branch', 'master', 'not_used')
         with pytest.raises(ValueError):
             repo.merge(ref, self, ref, 'workspace_id')
 
     def test_merge_invalid_user_email(self, vcsbackend):
         repo = vcsbackend.create_repo(number_of_commits=1)
         ref = Reference('branch', 'master', 'not_used')
         with pytest.raises(ValueError):
             repo.merge(ref, self, ref, 'workspace_id', 'user name')
 
     def test_merge_invalid_message(self, vcsbackend):
         repo = vcsbackend.create_repo(number_of_commits=1)
         ref = Reference('branch', 'master', 'not_used')
         with pytest.raises(ValueError):
             repo.merge(
                 ref, self, ref, 'workspace_id', 'user name', '[e-mail]')
 
 
[e-mail]("vcs_repository_support")
 class TestRepositoryStrip(BackendTestMixin):
     recreate_repo_per_test = True
 
     @classmethod
     def _get_commits(cls):
         commits = [
             {
                 'message': 'Initial commit',
                 'author': 'Joe Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'branch': 'master',
                 'added': [
                     FileNode('foobar', content='foobar'),
                     FileNode('foobar2', content='foobar2'),
                 ],
             },
         ]
         for x in xrange(10):
             commit_data = {
                 'message': 'Changed foobar - commit%s' % x,
                 'author': 'Jane Doe <[e-mail]>',
                 'date': datetime.datetime(2010, 1, 1, 21, x),
                 'branch': 'master',
                 'changed': [
                     FileNode('foobar', 'FOOBAR - %s' % x),
                 ],
             }
             commits.append(commit_data)
         return commits
 
     @pytest.mark.backends("git", "hg")
     def test_strip_commit(self):
         tip = self.repo.get_commit()
         assert tip.idx == 10
         self.repo.strip(tip.raw_id, self.repo.DEFAULT_BRANCH_NAME)
 
         tip = self.repo.get_commit()
         assert tip.idx == 9
 
     @pytest.mark.backends("git", "hg")
     def test_strip_multiple_commits(self):
         tip = self.repo.get_commit()
         assert tip.idx == 10
 
         old = self.repo.get_commit(commit_idx=5)
         self.repo.strip(old.raw_id, self.repo.DEFAULT_BRANCH_NAME)
 
         tip = self.repo.get_commit()
         assert tip.idx == 4
 
 
 @pytest.mark.backends('hg', 'git')
 class TestRepositoryPull:
 
     def test_pull(self, vcsbackend):
         source_repo = vcsbackend.repo
         target_repo = vcsbackend.create_repo()
         assert len(source_repo.commit_ids) > len(target_repo.commit_ids)
 
         target_repo.pull(source_repo.path)
         # Note: Get a fresh instance, avoids caching trouble
         target_repo = vcsbackend.backend(target_repo.path)
         assert len(source_repo.commit_ids) == len(target_repo.commit_ids)
 
     def test_pull_wrong_path(self, vcsbackend):
         target_repo = vcsbackend.create_repo()
         with pytest.raises(RepositoryError):
             target_repo.pull(target_repo.path + "wrong")
 
     def test_pull_specific_commits(self, vcsbackend):
         source_repo = vcsbackend.repo
         target_repo = vcsbackend.create_repo()
 
         second_commit = source_repo[1].raw_id
         if vcsbackend.alias == 'git':
             second_commit_ref = 'refs/test-refs/a'
             source_repo.set_refs(second_commit_ref, second_commit)
 
         target_repo.pull(source_repo.path, commit_ids=[second_commit])
         target_repo = vcsbackend.backend(target_repo.path)
         assert 2 == len(target_repo.commit_ids)
         assert second_commit == target_repo.get_commit().raw_id

rhodecode/tests/vcs/test_tags.py

@@ -1,69 +1,70 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import pytest
 
-from rhodecode.tests.vcs.base import BackendTestMixin
+from rhodecode.tests.vcs.conftest import BackendTestMixin
 from rhodecode.lib.vcs.exceptions import (
     TagAlreadyExistError, TagDoesNotExistError)
 
 
 pytestmark = pytest.mark.backends("git", "hg")
 
 
[e-mail]("vcs_repository_support")
 class TestTags(BackendTestMixin):
 
     def test_new_tag(self):
         tip = self.repo.get_commit()
         tagsize = len(self.repo.tags)
         tag = self.repo.tag('last-commit', 'joe', tip.raw_id)
 
         assert len(self.repo.tags) == tagsize + 1
         for top, __, __ in tip.walk():
             assert top == tag.get_node(top.path)
 
     def test_tag_already_exist(self):
         tip = self.repo.get_commit()
         self.repo.tag('last-commit', 'joe', tip.raw_id)
 
         with pytest.raises(TagAlreadyExistError):
             self.repo.tag('last-commit', 'joe', tip.raw_id)
 
         commit = self.repo.get_commit(commit_idx=0)
         with pytest.raises(TagAlreadyExistError):
             self.repo.tag('last-commit', 'jane', commit.raw_id)
 
     def test_remove_tag(self):
         tip = self.repo.get_commit()
         self.repo.tag('last-commit', 'joe', tip.raw_id)
         tagsize = len(self.repo.tags)
 
         self.repo.remove_tag('last-commit', user='evil joe')
         assert len(self.repo.tags) == tagsize - 1
 
     def test_remove_tag_which_does_not_exist(self):
         with pytest.raises(TagDoesNotExistError):
             self.repo.remove_tag('last-commit', user='evil joe')
 
     def test_name_with_slash(self):
         self.repo.tag('19/10/11', 'joe')
         assert '19/10/11' in self.repo.tags
         self.repo.tag('11', 'joe')
         assert '11' in self.repo.tags

rhodecode/tests/vcs/test_utils.py

@@ -1,195 +1,195 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
+import os
 import datetime
-import os
 import subprocess32
 
 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:
+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:
+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])
 
         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:
+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:
+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/vcsserver_http.ini

@@ -1,77 +1,77 b''
 ################################################################################
 # RhodeCode VCSServer with HTTP Backend - configuration                        #
 #                                                                              #
 ################################################################################
 
 [app:main]
 use = egg:rhodecode-vcsserver
 
 pyramid.default_locale_name = en
 pyramid.includes =
-pyramid.reload_templates = true
 
 # default locale used by VCS systems
 locale = en_US.UTF-8
 
 # cache regions, please don't change
 beaker.cache.regions = repo_object
 beaker.cache.repo_object.type = memorylru
 beaker.cache.repo_object.max_items = 100
 # cache auto-expires after N seconds
 beaker.cache.repo_object.expire = 300
 beaker.cache.repo_object.enabled = true
 
 [server:main]
-use = egg:waitress#main
 host = 127.0.0.1
 port = 9900
 
+use = egg:gunicorn#main
+
 ################################
 ### LOGGING CONFIGURATION   ####
 ################################
 [loggers]
 keys = root, vcsserver, beaker
 
 [handlers]
 keys = console
 
 [formatters]
 keys = generic
 
 #############
 ## LOGGERS ##
 #############
 [logger_root]
 level = NOTSET
 handlers = console
 
 [logger_vcsserver]
 level = DEBUG
 handlers =
 qualname = vcsserver
 propagate = 1
 
 [logger_beaker]
 level = DEBUG
 handlers =
 qualname = beaker
 propagate = 1
 
 
 ##############
 ## HANDLERS ##
 ##############
 
 [handler_console]
 class = StreamHandler
 args = (sys.stderr,)
 level = INFO
 formatter = generic
 
 ################
 ## FORMATTERS ##
 ################
 
 [formatter_generic]
 format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 datefmt = %Y-%m-%d %H:%M:%S