engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests: fixed all tests for python3 BIG changes

Browse source
This commit is contained in:
RhodeCode Admin 2023-07-18 11:46:31 +02:00
parent 1627525cb1
commit bf561e5914
208 changed files with 2545 additions and 2063 deletions
Download patch file Download diff file Expand all files Collapse all files

163
conftest.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -18,8 +17,14 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import pytest
from rhodecode.lib import ext_json
import pytest # noqa
# keep the imports to have a toplevel conftest.py but still importable from EE edition
from rhodecode.tests.conftest_common import ( # noqa
pytest_generate_tests,
pytest_runtest_makereport,
pytest_addoption
)
pytest_plugins = [
@ -29,121 +34,7 @@ pytest_plugins = [
def pytest_configure(config):
from rhodecode.config import patches
def pytest_addoption(parser):
def _parse_json(value):
return ext_json.str_json(value) if value else None
def _split_comma(value):
return value.split(',')
parser.addoption(
'--keep-tmp-path', action='store_true',
help="Keep the test temporary directories")
parser.addoption(
'--backends', action='store', type=_split_comma,
default=['git', 'hg', 'svn'],
help="Select which backends to test for backend specific tests.")
parser.addoption(
'--dbs', action='store', type=_split_comma,
default=['sqlite'],
help="Select which database to test for database specific tests. "
"Possible options are sqlite,postgres,mysql")
parser.addoption(
'--appenlight', '--ae', action='store_true',
help="Track statistics in appenlight.")
parser.addoption(
'--appenlight-api-key', '--ae-key',
help="API key for Appenlight.")
parser.addoption(
'--appenlight-url', '--ae-url',
default="https://ae.rhodecode.com",
help="Appenlight service URL, defaults to https://ae.rhodecode.com")
parser.addoption(
'--sqlite-connection-string', action='store',
default='', help="Connection string for the dbs tests with SQLite")
parser.addoption(
'--postgres-connection-string', action='store',
default='', help="Connection string for the dbs tests with Postgres")
parser.addoption(
'--mysql-connection-string', action='store',
default='', help="Connection string for the dbs tests with MySQL")
parser.addoption(
'--repeat', type=int, default=100,
help="Number of repetitions in performance tests.")
parser.addoption(
'--test-loglevel', dest='test_loglevel',
help="Set default Logging level for tests, critical(default), error, warn , 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.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
"""
Adding the remote traceback if the exception has this information.
VCSServer attaches this information as the attribute `_vcs_server_traceback`
to the exception instance.
"""
outcome = yield
report = outcome.get_result()
if call.excinfo:
exc = call.excinfo.value
vcsserver_traceback = getattr(exc, '_vcs_server_traceback', None)
if vcsserver_traceback and report.outcome == 'failed':
section = f'VCSServer remote traceback {report.when}'
report.sections.append((section, vcsserver_traceback))
from rhodecode.config import patches # noqa
def pytest_collection_modifyitems(session, config, items):
@ -152,7 +43,7 @@ def pytest_collection_modifyitems(session, config, items):
i for i in items if getattr(i.obj, '__test__', True)]
items[:] = remaining
# NOTE(marcink): custom test ordering, db tests and vcstests are slowes and should
# NOTE(marcink): custom test ordering, db tests and vcstests are slowest and should
# be executed at the end for faster test feedback
def sorter(item):
pos = 0
@ -165,37 +56,3 @@ def pytest_collection_modifyitems(session, config, items):
return pos
items.sort(key=sorter)
def get_backends_from_metafunc(metafunc):
requested_backends = set(metafunc.config.getoption('--backends'))
backend_mark = metafunc.definition.get_closest_marker('backends')
if backend_mark:
# Supported backends by this test function, created from
# pytest.mark.backends
backends = backend_mark.args
elif hasattr(metafunc.cls, 'backend_alias'):
# Support class attribute "backend_alias", this is mainly
# for legacy reasons for tests not yet using pytest.mark.backends
backends = [metafunc.cls.backend_alias]
else:
backends = metafunc.config.getoption('--backends')
return requested_backends.intersection(backends)
def pytest_generate_tests(metafunc):
# Support test generation based on --backend parameter
if 'backend_alias' in metafunc.fixturenames:
backends = get_backends_from_metafunc(metafunc)
scope = None
if not backends:
pytest.skip("Not enabled for any of selected backends")
metafunc.parametrize('backend_alias', backends, scope=scope)
backend_mark = metafunc.definition.get_closest_marker('backends')
if backend_mark:
backends = get_backends_from_metafunc(metafunc)
if not backends:
pytest.skip("Not enabled for any of selected backends")

2
pytest.ini
View file

@ -13,6 +13,8 @@ addopts =
--capture=no
--show-capture=all
# --test-loglevel=INFO, show log-level during execution
markers =
vcs_operations: Mark tests depending on a running RhodeCode instance.
xfail_backends: Mark tests as xfail for given backends.

8
rhodecode/apps/admin/tests/test_admin_audit_logs.py
View file

@ -23,15 +23,17 @@ import datetime
import pytest
from rhodecode.lib.str_utils import safe_str
from rhodecode.tests import *
from rhodecode.tests.fixture import FIXTURES
from rhodecode.model.db import UserLog
from rhodecode.model.meta import Session
from rhodecode.lib.utils2 import safe_unicode
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {
@ -69,7 +71,7 @@ class TestAdminController(object):
for row in csv.DictReader(f):
ul = UserLog()
for k, v in row.items():
v = safe_unicode(v)
v = safe_str(v)
if k == 'action_date':
v = strptime(v)
if k in ['user_id', 'repository_id']:

4
rhodecode/apps/admin/tests/test_admin_defaults.py
View file

@ -24,7 +24,9 @@ from rhodecode.model.settings import SettingsModel
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

9
rhodecode/apps/admin/tests/test_admin_main_views.py
View file

@ -26,7 +26,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {
@ -50,7 +52,10 @@ class TestAdminMainView(TestController):
response = self.app.get(route_path('admin_home'), status=200)
response.mustcontain("Administration area")
def test_redirect_pull_request_view(self, view):
@pytest.mark.parametrize('view', [
'pull_requests_global',
])
def test_redirect_pull_request_view_global(self, view):
self.log_user()
self.app.get(
route_path(view, pull_request_id='xxxx'),

4
rhodecode/apps/admin/tests/test_admin_permissions.py
View file

@ -28,7 +28,9 @@ from rhodecode.tests import (
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

44
rhodecode/apps/admin/tests/test_admin_repos.py
View file

@ -17,7 +17,9 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
import mock
import pytest
@ -42,7 +44,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repos': ADMIN_PREFIX + '/repos',
@ -92,12 +96,14 @@ class TestAdminRepos(object):
assert ['hg', 'git', 'svn'] == [x.get('value') for x in assert_response.get_elements('[name=repo_type]')]
@pytest.mark.parametrize(
"suffix", [u'', u'xxa'], ids=['', 'non-ascii'])
"suffix", ['', 'xxa'], ids=['', 'non-ascii'])
def test_create(self, autologin_user, backend, suffix, csrf_token):
repo_name_unicode = backend.new_repo_name(suffix=suffix)
repo_name = repo_name_unicode.encode('utf8')
description_unicode = u'description for newly created repo' + suffix
description = description_unicode.encode('utf8')
repo_name = repo_name_unicode
description_unicode = 'description for newly created repo' + suffix
description = description_unicode
response = self.app.post(
route_path('repo_create'),
fixture._get_repo_create_params(
@ -127,20 +133,20 @@ class TestAdminRepos(object):
self.assert_repository_is_created_correctly(
repo_name, description, backend)
@pytest.mark.parametrize("suffix", [u'', u'ąćę'], ids=['', 'non-ascii'])
@pytest.mark.parametrize("suffix", ['', '_ąćę'], ids=['', 'non-ascii'])
def test_create_in_group(
self, autologin_user, backend, suffix, csrf_token):
# create GROUP
group_name = 'sometest_%s' % backend.alias
group_name = f'sometest_{backend.alias}'
gr = RepoGroupModel().create(group_name=group_name,
group_description='test',
owner=TEST_USER_ADMIN_LOGIN)
Session().commit()
repo_name = u'ingroup' + suffix
repo_name_full = RepoGroup.url_sep().join(
[group_name, repo_name])
description = u'description for newly created repo'
repo_name = f'ingroup{suffix}'
repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
description = 'description for newly created repo'
self.app.post(
route_path('repo_create'),
fixture._get_repo_create_params(
@ -483,17 +489,15 @@ class TestAdminRepos(object):
# repo must not be in filesystem !
assert not repo_on_filesystem(repo_name)
def assert_repository_is_created_correctly(
self, repo_name, description, backend):
repo_name_utf8 = safe_str(repo_name)
def assert_repository_is_created_correctly(self, repo_name, description, backend):
url_quoted_repo_name = urllib.parse.quote(repo_name)
# run the check page that triggers the flash message
response = self.app.get(
route_path('repo_creating_check', repo_name=safe_str(repo_name)))
assert response.json == {u'result': True}
route_path('repo_creating_check', repo_name=repo_name))
assert response.json == {'result': True}
flash_msg = u'Created repository <a href="/{}">{}</a>'.format(
urllib.parse.quote(repo_name_utf8), repo_name)
flash_msg = 'Created repository <a href="/{}">{}</a>'.format(url_quoted_repo_name, repo_name)
assert_session_flash(response, flash_msg)
# test if the repo was created in the database
@ -504,7 +508,7 @@ class TestAdminRepos(object):
# test if the repository is visible in the list ?
response = self.app.get(
h.route_path('repo_summary', repo_name=safe_str(repo_name)))
h.route_path('repo_summary', repo_name=repo_name))
response.mustcontain(repo_name)
response.mustcontain(backend.alias)

20
rhodecode/apps/admin/tests/test_admin_repository_groups.py
View file

@ -33,7 +33,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_groups': ADMIN_PREFIX + '/repo_groups',
@ -106,7 +108,7 @@ class TestAdminRepositoryGroups(object):
'hg_repo_ąć',
])
def test_create(self, autologin_user, repo_group_name, csrf_token):
repo_group_name_unicode = repo_group_name.decode('utf8')
repo_group_name_non_ascii = repo_group_name
description = 'description for newly created repo group'
response = self.app.post(
@ -123,14 +125,14 @@ class TestAdminRepositoryGroups(object):
assert_session_flash(
response,
'Created repository group <a href="%s">%s</a>' % (
repo_gr_url, repo_group_name_unicode))
repo_gr_url, repo_group_name_non_ascii))
# # test if the repo group was created in the database
new_repo_group = RepoGroupModel()._get_repo_group(
repo_group_name_unicode)
repo_group_name_non_ascii)
assert new_repo_group is not None
assert new_repo_group.group_name == repo_group_name_unicode
assert new_repo_group.group_name == repo_group_name_non_ascii
assert new_repo_group.group_description == description
# test if the repository is visible in the list ?
@ -143,7 +145,7 @@ class TestAdminRepositoryGroups(object):
if not is_on_filesystem:
self.fail('no repo group %s in filesystem' % repo_group_name)
RepoGroupModel().delete(repo_group_name_unicode)
RepoGroupModel().delete(repo_group_name_non_ascii)
Session().commit()
@pytest.mark.parametrize('repo_group_name', [
@ -159,7 +161,7 @@ class TestAdminRepositoryGroups(object):
expected_group_name = '{}/{}'.format(
parent_group_name, repo_group_name)
expected_group_name_unicode = expected_group_name.decode('utf8')
expected_group_name_non_ascii = expected_group_name
try:
response = self.app.post(
@ -175,9 +177,9 @@ class TestAdminRepositoryGroups(object):
u'Created repository group <a href="%s">%s</a>' % (
h.route_path('repo_group_home',
repo_group_name=expected_group_name),
expected_group_name_unicode))
expected_group_name_non_ascii))
finally:
RepoGroupModel().delete(expected_group_name_unicode)
RepoGroupModel().delete(expected_group_name_non_ascii)
Session().commit()
def test_user_with_creation_permissions_cannot_create_subgroups(

28
rhodecode/apps/admin/tests/test_admin_settings.py
View file

@ -22,19 +22,20 @@ import pytest
import rhodecode
from rhodecode.apps._base import ADMIN_PREFIX
from rhodecode.lib.utils2 import md5
from rhodecode.lib.hash_utils import md5_safe
from rhodecode.model.db import RhodeCodeUi
from rhodecode.model.meta import Session
from rhodecode.model.settings import SettingsModel, IssueTrackerSettingsModel
from rhodecode.tests import assert_session_flash
from rhodecode.tests.utils import AssertResponse
UPDATE_DATA_QUALNAME = 'rhodecode.model.update.UpdateModel.get_update_data'
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {
@ -233,6 +234,7 @@ class TestAdminSettingsGlobal(object):
route_path('admin_settings_global_update'), params=params)
assert_session_flash(response, 'Updated application settings')
app_settings = SettingsModel().get_all_settings()
del settings['csrf_token']
for key, value in settings.items():
@ -413,8 +415,9 @@ class TestAdminSettingsVcs(object):
@pytest.fixture()
def disable_sql_cache(self, request):
# patch _do_orm_execute so it returns None similar like if we don't use a cached query
patcher = mock.patch(
'rhodecode.lib.caching_query.FromCache.process_query')
'rhodecode.lib.caching_query.ORMCache._do_orm_execute', return_value=None)
request.addfinalizer(patcher.stop)
patcher.start()
@ -428,8 +431,7 @@ class TestAdminSettingsVcs(object):
@pytest.fixture(scope='class', autouse=True)
def cleanup_settings(self, request, baseapp):
ui_id = RhodeCodeUi.ui_id
original_ids = list(
r.ui_id for r in RhodeCodeUi.query().values(ui_id))
original_ids = [r.ui_id for r in RhodeCodeUi.query().with_entities(ui_id)]
@request.addfinalizer
def cleanup():
@ -644,9 +646,9 @@ class TestAdminSettingsIssueTracker(object):
}
self.app.post(post_url, post_data, status=302)
settings = SettingsModel().get_all_settings()
self.uid = md5(pattern)
self.uid = md5_safe(pattern)
assert settings[self.PATTERN_KEY+self.uid] == pattern
self.another_uid = md5(another_pattern)
self.another_uid = md5_safe(another_pattern)
assert settings[self.PATTERN_KEY+self.another_uid] == another_pattern
@request.addfinalizer
@ -654,7 +656,7 @@ class TestAdminSettingsIssueTracker(object):
defaults = SettingsModel().get_all_settings()
entries = [name for name in defaults if (
(self.uid in name) or (self.another_uid) in name)]
(self.uid in name) or (self.another_uid in name))]
start = len(self.RC_PREFIX)
for del_key in entries:
# TODO: anderson: get_by_name needs name without prefix
@ -667,7 +669,7 @@ class TestAdminSettingsIssueTracker(object):
self, autologin_user, backend, csrf_token, request):
old_pattern = 'issuetracker_pat1'
old_uid = md5(old_pattern)
old_uid = md5_safe(old_pattern)
post_url = route_path('admin_settings_issuetracker_update')
post_data = {
@ -681,7 +683,7 @@ class TestAdminSettingsIssueTracker(object):
self.app.post(post_url, post_data, status=302)
new_pattern = 'issuetracker_pat1_edited'
self.new_uid = md5(new_pattern)
self.new_uid = md5_safe(new_pattern)
post_url = route_path('admin_settings_issuetracker_update')
post_data = {
@ -708,7 +710,7 @@ class TestAdminSettingsIssueTracker(object):
self, autologin_user, csrf_token, request, settings_util):
prefix = 'issuetracker'
pattern = 'issuetracker_pat'
self.uid = md5(pattern)
self.uid = md5_safe(pattern)
pattern_key = '_'.join([prefix, 'pat', self.uid])
rc_pattern_key = '_'.join(['rhodecode', pattern_key])
desc_key = '_'.join([prefix, 'desc', self.uid])
@ -742,7 +744,7 @@ class TestAdminSettingsIssueTracker(object):
self, autologin_user, backend, csrf_token, settings_util, xhr_header):
old_pattern = 'issuetracker_pat_deleted'
old_uid = md5(old_pattern)
old_uid = md5_safe(old_pattern)
post_url = route_path('admin_settings_issuetracker_update')
post_data = {

4
rhodecode/apps/admin/tests/test_admin_user_groups.py
View file

@ -30,7 +30,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

4
rhodecode/apps/admin/tests/test_admin_users.py
View file

@ -34,7 +34,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

4
rhodecode/apps/admin/tests/test_admin_users_ssh_keys.py
View file

@ -28,7 +28,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

10
rhodecode/apps/file_store/tests/test_upload_file.py
View file

@ -27,7 +27,9 @@ from rhodecode.apps.file_store import utils, config_keys
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'upload_file': '/_file_store/upload',
@ -59,7 +61,7 @@ class TestFileStoreViews(TestController):
status = 200
store = utils.get_file_storage({config_keys.store_path: store_path})
filesystem_file = os.path.join(str(tmpdir), fid)
with open(filesystem_file, 'wb') as f:
with open(filesystem_file, 'wt') as f:
f.write(content)
with open(filesystem_file, 'rb') as f:
@ -120,7 +122,7 @@ class TestFileStoreViews(TestController):
self.log_user()
response = self.app.post(
route_path('upload_file'),
upload_files=[('store_file', 'myfile.txt', 'SOME CONTENT')],
upload_files=[('store_file', b'myfile.txt', b'SOME CONTENT')],
params={'csrf_token': self.csrf_token},
status=200)
@ -134,7 +136,7 @@ class TestFileStoreViews(TestController):
fid = 'example.txt'
filesystem_file = os.path.join(str(tmpdir), fid)
with open(filesystem_file, 'wb') as f:
with open(filesystem_file, 'wt') as f:
f.write(content)
with open(filesystem_file, 'rb') as f:

64
rhodecode/apps/gist/tests/test_admin_gists.py
View file

@ -30,7 +30,8 @@ from rhodecode.tests import (
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {
@ -59,7 +60,7 @@ class GistUtility(object):
self._gist_ids = []
def __call__(
self, f_name, content='some gist', lifetime=-1,
self, f_name: bytes, content: bytes = b'some gist', lifetime=-1,
description='gist-desc', gist_type='public',
acl_level=Gist.GIST_PUBLIC, owner=TEST_USER_ADMIN_LOGIN):
gist_mapping = {
@ -94,14 +95,14 @@ class TestGistsController(TestController):
def test_index_empty(self, create_gist):
self.log_user()
response = self.app.get(route_path('gists_show'))
response.mustcontain('data: [],')
response.mustcontain('var gist_data = [];')
def test_index(self, create_gist):
self.log_user()
g1 = create_gist('gist1')
g2 = create_gist('gist2', lifetime=1400)
g3 = create_gist('gist3', description='gist3-desc')
g4 = create_gist('gist4', gist_type='private').gist_access_id
g1 = create_gist(b'gist1')
g2 = create_gist(b'gist2', lifetime=1400)
g3 = create_gist(b'gist3', description='gist3-desc')
g4 = create_gist(b'gist4', gist_type='private').gist_access_id
response = self.app.get(route_path('gists_show'))
response.mustcontain(g1.gist_access_id)
@ -111,13 +112,12 @@ class TestGistsController(TestController):
response.mustcontain(no=[g4])
# Expiration information should be visible
expires_tag = '%s' % h.age_component(
h.time_to_utcdatetime(g2.gist_expires))
expires_tag = str(h.age_component(h.time_to_utcdatetime(g2.gist_expires)))
response.mustcontain(expires_tag.replace('"', '\\"'))
def test_index_private_gists(self, create_gist):
self.log_user()
gist = create_gist('gist5', gist_type='private')
gist = create_gist(b'gist5', gist_type='private')
response = self.app.get(route_path('gists_show', params=dict(private=1)))
# and privates
@ -125,10 +125,10 @@ class TestGistsController(TestController):
def test_index_show_all(self, create_gist):
self.log_user()
create_gist('gist1')
create_gist('gist2', lifetime=1400)
create_gist('gist3', description='gist3-desc')
create_gist('gist4', gist_type='private')
create_gist(b'gist1')
create_gist(b'gist2', lifetime=1400)
create_gist(b'gist3', description='gist3-desc')
create_gist(b'gist4', gist_type='private')
response = self.app.get(route_path('gists_show', params=dict(all=1)))
@ -139,9 +139,9 @@ class TestGistsController(TestController):
def test_index_show_all_hidden_from_regular(self, create_gist):
self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
create_gist('gist2', gist_type='private')
create_gist('gist3', gist_type='private')
create_gist('gist4', gist_type='private')
create_gist(b'gist2', gist_type='private')
create_gist(b'gist3', gist_type='private')
create_gist(b'gist4', gist_type='private')
response = self.app.get(route_path('gists_show', params=dict(all=1)))
@ -181,7 +181,7 @@ class TestGistsController(TestController):
def test_access_expired_gist(self, create_gist):
self.log_user()
gist = create_gist('never-see-me')
gist = create_gist(b'never-see-me')
gist.gist_expires = 0 # 1970
Session().add(gist)
Session().commit()
@ -269,7 +269,7 @@ class TestGistsController(TestController):
def test_delete(self, create_gist):
self.log_user()
gist = create_gist('delete-me')
gist = create_gist(b'delete-me')
response = self.app.post(
route_path('gist_delete', gist_id=gist.gist_id),
params={'csrf_token': self.csrf_token})
@ -277,7 +277,7 @@ class TestGistsController(TestController):
def test_delete_normal_user_his_gist(self, create_gist):
self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
gist = create_gist('delete-me', owner=TEST_USER_REGULAR_LOGIN)
gist = create_gist(b'delete-me', owner=TEST_USER_REGULAR_LOGIN)
response = self.app.post(
route_path('gist_delete', gist_id=gist.gist_id),
@ -286,14 +286,14 @@ class TestGistsController(TestController):
def test_delete_normal_user_not_his_own_gist(self, create_gist):
self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
gist = create_gist('delete-me-2')
gist = create_gist(b'delete-me-2')
self.app.post(
route_path('gist_delete', gist_id=gist.gist_id),
params={'csrf_token': self.csrf_token}, status=404)
def test_show(self, create_gist):
gist = create_gist('gist-show-me')
gist = create_gist(b'gist-show-me')
response = self.app.get(route_path('gist_show', gist_id=gist.gist_access_id))
response.mustcontain('added file: gist-show-me<')
@ -308,12 +308,12 @@ class TestGistsController(TestController):
def test_show_without_hg(self, create_gist):
with mock.patch(
'rhodecode.lib.vcs.settings.ALIASES', ['git']):
gist = create_gist('gist-show-me-again')
gist = create_gist(b'gist-show-me-again')
self.app.get(
route_path('gist_show', gist_id=gist.gist_access_id), status=200)
def test_show_acl_private(self, create_gist):
gist = create_gist('gist-show-me-only-when-im-logged-in',
gist = create_gist(b'gist-show-me-only-when-im-logged-in',
acl_level=Gist.ACL_LEVEL_PRIVATE)
self.app.get(
route_path('gist_show', gist_id=gist.gist_access_id), status=404)
@ -331,7 +331,7 @@ class TestGistsController(TestController):
response.mustcontain('gist-desc')
def test_show_as_raw(self, create_gist):
gist = create_gist('gist-show-me', content='GIST CONTENT')
gist = create_gist(b'gist-show-me', content=b'GIST CONTENT')
response = self.app.get(
route_path('gist_show_formatted',
gist_id=gist.gist_access_id, revision='tip',
@ -339,7 +339,7 @@ class TestGistsController(TestController):
assert response.text == 'GIST CONTENT'
def test_show_as_raw_individual_file(self, create_gist):
gist = create_gist('gist-show-me-raw', content='GIST BODY')
gist = create_gist(b'gist-show-me-raw', content=b'GIST BODY')
response = self.app.get(
route_path('gist_show_formatted_path',
gist_id=gist.gist_access_id, format='raw',
@ -348,24 +348,24 @@ class TestGistsController(TestController):
def test_edit_page(self, create_gist):
self.log_user()
gist = create_gist('gist-for-edit', content='GIST EDIT BODY')
gist = create_gist(b'gist-for-edit', content=b'GIST EDIT BODY')
response = self.app.get(route_path('gist_edit', gist_id=gist.gist_access_id))
response.mustcontain('GIST EDIT BODY')
def test_edit_page_non_logged_user(self, create_gist):
gist = create_gist('gist-for-edit', content='GIST EDIT BODY')
gist = create_gist(b'gist-for-edit', content=b'GIST EDIT BODY')
self.app.get(route_path('gist_edit', gist_id=gist.gist_access_id),
status=302)
def test_edit_normal_user_his_gist(self, create_gist):
self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
gist = create_gist('gist-for-edit', owner=TEST_USER_REGULAR_LOGIN)
gist = create_gist(b'gist-for-edit', owner=TEST_USER_REGULAR_LOGIN)
self.app.get(route_path('gist_edit', gist_id=gist.gist_access_id,
status=200))
def test_edit_normal_user_not_his_own_gist(self, create_gist):
self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
gist = create_gist('delete-me')
gist = create_gist(b'delete-me')
self.app.get(route_path('gist_edit', gist_id=gist.gist_access_id),
status=404)
@ -375,7 +375,7 @@ class TestGistsController(TestController):
password = 'test'
user = user_util.create_user(
firstname=xss_atack_string, password=password)
create_gist('gist', gist_type='public', owner=user.username)
create_gist(b'gist', gist_type='public', owner=user.username)
response = self.app.get(route_path('gists_show'))
response.mustcontain(xss_escaped_string)
@ -385,6 +385,6 @@ class TestGistsController(TestController):
password = 'test'
user = user_util.create_user(
lastname=xss_atack_string, password=password)
create_gist('gist', gist_type='public', owner=user.username)
create_gist(b'gist', gist_type='public', owner=user.username)
response = self.app.get(route_path('gists_show'))
response.mustcontain(xss_escaped_string)

8
rhodecode/apps/home/tests/test_get_repo_list_data.py
View file

@ -18,18 +18,20 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import json
from . import assert_and_get_repo_list_content
from rhodecode.tests import TestController
from rhodecode.tests.fixture import Fixture
from rhodecode.model.db import Repository
from rhodecode.lib.ext_json import json
fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_list_data': '/_repos',

8
rhodecode/apps/home/tests/test_get_user_data.py
View file

@ -17,19 +17,19 @@
# 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 json
import pytest
from rhodecode.tests import TestController
from rhodecode.tests.fixture import Fixture
from rhodecode.lib.ext_json import json
fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'user_autocomplete_data': '/_users',

42
rhodecode/apps/home/tests/test_get_user_group_data.py
View file

@ -1,22 +1,4 @@
# Copyright (C) 2016-2020 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2020 RhodeCode GmbH
#
@ -36,19 +18,39 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import json
# Copyright (C) 2016-2020 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import pytest
from rhodecode.tests import TestController
from rhodecode.tests.fixture import Fixture
from rhodecode.lib.ext_json import json
fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'user_autocomplete_data': '/_users',

2
rhodecode/apps/home/tests/test_home.py
View file

@ -163,7 +163,7 @@ class TestHomeController(TestController):
'show_version', state, 'bool')
Session().add(sett)
Session().commit()
SettingsModel().invalidate_settings_cache()
SettingsModel().invalidate_settings_cache(hard=True)
response = self.app.get(route_path('home'))
if state is True:

4
rhodecode/apps/journal/tests/test_journal.py
View file

@ -27,7 +27,9 @@ from rhodecode.model.db import UserFollowing, Repository
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'journal': ADMIN_PREFIX + '/journal',

39
rhodecode/apps/login/tests/test_login.py
View file

@ -38,7 +38,9 @@ whitelist_view = ['RepoCommitsView:repo_commit_raw']
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {
@ -160,18 +162,28 @@ class TestLoginController(object):
'file:///etc/passwd',
'ftp://some.ftp.server',
'http://other.domain',
'/\r\nX-Forwarded-Host: http://example.org',
], ids=no_newline_id_generator)
def test_login_bad_came_froms(self, url_came_from):
_url = '{}?came_from={}'.format(route_path('login'), url_came_from)
response = self.app.post(
_url,
{'username': 'test_admin', 'password': 'test12'})
_url, {'username': 'test_admin', 'password': 'test12'}, status=302)
assert response.status == '302 Found'
response = response.follow()
assert response.status == '200 OK'
assert response.request.path == '/'
@pytest.mark.xfail(reason="newline params changed behaviour in python3")
@pytest.mark.parametrize("url_came_from", [
'/\r\nX-Forwarded-Host: \rhttp://example.org',
], ids=no_newline_id_generator)
def test_login_bad_came_froms_404(self, url_came_from):
_url = '{}?came_from={}'.format(route_path('login'), url_came_from)
response = self.app.post(
_url, {'username': 'test_admin', 'password': 'test12'}, status=302)
response = response.follow()
assert response.status == '404 Not Found'
def test_login_short_password(self):
response = self.app.post(route_path('login'),
{'username': 'test_admin',
@ -184,7 +196,7 @@ class TestLoginController(object):
response = self.app.post(
route_path('login'),
{'username': user_regular.username,
'password': u'invalid-non-asci\xe4'.encode('utf8')})
'password': 'invalid-non-asci\xe4'.encode('utf8')})
response.mustcontain('invalid user name')
response.mustcontain('invalid password')
@ -486,6 +498,10 @@ class TestLoginController(object):
auth_token = user_admin.api_key
with fixture.anon_access(False):
# webtest uses linter to check if response is bytes,
# and we use memoryview here as a wrapper, quick turn-off
self.app.lint = False
self.app.get(
route_path('repo_commit_raw',
repo_name=HG_REPO, commit_id='tip',
@ -511,6 +527,9 @@ class TestLoginController(object):
assert auth_token
with fixture.anon_access(False):
# webtest uses linter to check if response is bytes,
# and we use memoryview here as a wrapper, quick turn-off
self.app.lint = False
self.app.get(
route_path('repo_commit_raw',
repo_name=HG_REPO, commit_id='tip',
@ -536,6 +555,10 @@ class TestLoginController(object):
with mock.patch.dict('rhodecode.CONFIG', whitelist):
with fixture.anon_access(False):
# webtest uses linter to check if response is bytes,
# and we use memoryview here as a wrapper, quick turn-off
self.app.lint = False
self.app.get(
route_path('repo_commit_raw',
repo_name=HG_REPO, commit_id='tip',
@ -552,6 +575,9 @@ class TestLoginController(object):
TEST_USER_ADMIN_LOGIN, 'test')
Session().commit()
with fixture.anon_access(False):
# webtest uses linter to check if response is bytes,
# and we use memoryview here as a wrapper, quick turn-off
self.app.lint = False
self.app.get(
route_path('repo_commit_raw',
repo_name=HG_REPO, commit_id='tip',
@ -572,6 +598,9 @@ class TestLoginController(object):
Session().add(new_auth_token)
Session().commit()
with fixture.anon_access(False):
# webtest uses linter to check if response is bytes,
# and we use memoryview here as a wrapper, quick turn-off
self.app.lint = False
self.app.get(
route_path('repo_commit_raw',
repo_name=HG_REPO, commit_id='tip',

4
rhodecode/apps/login/tests/test_password_reset.py
View file

@ -30,7 +30,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

7
rhodecode/apps/my_account/tests/test_my_account_notifications.py
View file

@ -34,7 +34,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {
@ -133,7 +135,8 @@ class TestNotificationsController(TestController):
u2 = User.get(u2.user_id)
# check DB
get_notif = lambda un: [x.notification for x in un]
def get_notif(un):
return [x.notification for x in un]
assert get_notif(cur_user.notifications) == [notification]
assert get_notif(u1.notifications) == [notification]
assert get_notif(u2.notifications) == [notification]

4
rhodecode/apps/my_account/tests/test_my_account_ssh_keys.py
View file

@ -28,7 +28,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

4
rhodecode/apps/repo_group/tests/test_repo_groups_advanced.py
View file

@ -23,7 +23,9 @@ from rhodecode.tests import assert_session_flash
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'edit_repo_group_advanced':

4
rhodecode/apps/repo_group/tests/test_repo_groups_permissions.py
View file

@ -23,7 +23,9 @@ from rhodecode.tests.utils import permission_update_data_generator
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'edit_repo_group_perms':

4
rhodecode/apps/repo_group/tests/test_repo_groups_settings.py
View file

@ -23,7 +23,9 @@ from rhodecode.tests import assert_session_flash
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'edit_repo_group': '/{repo_group_name}/_edit',

4
rhodecode/apps/repository/tests/test_pull_requests_list.py
View file

@ -22,7 +22,9 @@ from rhodecode.model.db import Repository
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'pullrequest_show_all': '/{repo_name}/pull-request',

4
rhodecode/apps/repository/tests/test_repo_bookmarks.py
View file

@ -22,7 +22,9 @@ from rhodecode.model.db import Repository
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'bookmarks_home': '/{repo_name}/bookmarks',

4
rhodecode/apps/repository/tests/test_repo_branches.py
View file

@ -22,7 +22,9 @@ from rhodecode.model.db import Repository
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'branches_home': '/{repo_name}/branches',

9
rhodecode/apps/repository/tests/test_repo_changelog.py
View file

@ -28,7 +28,9 @@ MATCH_HASH = re.compile(r'<span class="commit_hash">r(\d+):[\da-f]+</span>')
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_changelog': '/{repo_name}/changelog',
@ -43,7 +45,7 @@ def route_path(name, params=None, **kwargs):
def assert_commits_on_page(response, indexes):
found_indexes = [int(idx) for idx in MATCH_HASH.findall(response.body)]
found_indexes = [int(idx) for idx in MATCH_HASH.findall(response.text)]
assert found_indexes == indexes
@ -109,8 +111,7 @@ class TestChangelogController(TestController):
assert expected_warning in response.text
@pytest.mark.xfail_backends("svn", reason="Depends on branch support")
def test_changelog_filtered_by_branch_with_merges(
self, autologin_user, backend):
def test_changelog_filtered_by_branch_with_merges(self, autologin_user, backend):
# Note: The changelog of branch "b" does not contain the commit "a1"
# although this is a parent of commit "b1". And branch "b" has commits

32
rhodecode/apps/repository/tests/test_repo_commit_comments.py
View file

@ -27,7 +27,9 @@ from rhodecode.lib import helpers as h
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_commit': '/{repo_name}/changeset/{commit_id}',
@ -71,7 +73,7 @@ class TestRepoCommitCommentsView(TestController):
self.log_user()
commit = backend.repo.get_commit('300')
commit_id = commit.raw_id
text = u'CommentOnCommit'
text = 'CommentOnCommit'
params = {'text': text, 'csrf_token': self.csrf_token,
'comment_type': comment_type}
@ -101,7 +103,7 @@ class TestRepoCommitCommentsView(TestController):
author, comment_type, h.show_id(commit), backend.repo_name)
assert sbj == notification.subject
lnk = (u'/{0}/changeset/{1}#comment-{2}'.format(
lnk = ('/{0}/changeset/{1}#comment-{2}'.format(
backend.repo_name, commit_id, comment_id))
assert lnk in notification.body
@ -110,7 +112,7 @@ class TestRepoCommitCommentsView(TestController):
self.log_user()
commit = backend.repo.get_commit('300')
commit_id = commit.raw_id
text = u'CommentOnCommit'
text = 'CommentOnCommit'
f_path = 'vcs/web/simplevcs/views/repository.py'
line = 'n1'
@ -163,7 +165,7 @@ class TestRepoCommitCommentsView(TestController):
assert sbj == notification.subject
lnk = (u'/{0}/changeset/{1}#comment-{2}'.format(
lnk = ('/{0}/changeset/{1}#comment-{2}'.format(
backend.repo_name, commit_id, comment.comment_id))
assert lnk in notification.body
assert 'on line n1' in notification.body
@ -172,7 +174,7 @@ class TestRepoCommitCommentsView(TestController):
self.log_user()
commit_id = backend.repo.get_commit('300').raw_id
text = u'@test_regular check CommentOnCommit'
text = '@test_regular check CommentOnCommit'
params = {'text': text, 'csrf_token': self.csrf_token}
self.app.post(
@ -193,13 +195,13 @@ class TestRepoCommitCommentsView(TestController):
users = [x.username for x in notification.recipients]
# test_regular gets notification by @mention
assert sorted(users) == [u'test_admin', u'test_regular']
assert sorted(users) == ['test_admin', 'test_regular']
def test_create_with_status_change(self, backend):
self.log_user()
commit = backend.repo.get_commit('300')
commit_id = commit.raw_id
text = u'CommentOnCommit'
text = 'CommentOnCommit'
f_path = 'vcs/web/simplevcs/views/repository.py'
line = 'n1'
@ -233,14 +235,14 @@ class TestRepoCommitCommentsView(TestController):
author, h.show_id(commit), backend.repo_name)
assert sbj == notification.subject
lnk = (u'/{0}/changeset/{1}#comment-{2}'.format(
lnk = ('/{0}/changeset/{1}#comment-{2}'.format(
backend.repo_name, commit_id, comment_id))
assert lnk in notification.body
def test_delete(self, backend):
self.log_user()
commit_id = backend.repo.get_commit('300').raw_id
text = u'CommentOnCommit'
text = 'CommentOnCommit'
params = {'text': text, 'csrf_token': self.csrf_token}
self.app.post(
@ -271,7 +273,7 @@ class TestRepoCommitCommentsView(TestController):
def test_edit(self, backend):
self.log_user()
commit_id = backend.repo.get_commit('300').raw_id
text = u'CommentOnCommit'
text = 'CommentOnCommit'
params = {'text': text, 'csrf_token': self.csrf_token}
self.app.post(
@ -304,7 +306,7 @@ class TestRepoCommitCommentsView(TestController):
def test_edit_without_change(self, backend):
self.log_user()
commit_id = backend.repo.get_commit('300').raw_id
text = u'CommentOnCommit'
text = 'CommentOnCommit'
params = {'text': text, 'csrf_token': self.csrf_token}
self.app.post(
@ -336,7 +338,7 @@ class TestRepoCommitCommentsView(TestController):
def test_edit_try_edit_already_edited(self, backend):
self.log_user()
commit_id = backend.repo.get_commit('300').raw_id
text = u'CommentOnCommit'
text = 'CommentOnCommit'
params = {'text': text, 'csrf_token': self.csrf_token}
self.app.post(
@ -390,7 +392,7 @@ class TestRepoCommitCommentsView(TestController):
def test_edit_forbidden_for_immutable_comments(self, backend):
self.log_user()
commit_id = backend.repo.get_commit('300').raw_id
text = u'CommentOnCommit'
text = 'CommentOnCommit'
params = {'text': text, 'csrf_token': self.csrf_token, 'version': '0'}
self.app.post(
@ -429,7 +431,7 @@ class TestRepoCommitCommentsView(TestController):
def test_delete_forbidden_for_immutable_comments(self, backend):
self.log_user()
commit_id = backend.repo.get_commit('300').raw_id
text = u'CommentOnCommit'
text = 'CommentOnCommit'
params = {'text': text, 'csrf_token': self.csrf_token}
self.app.post(

30
rhodecode/apps/repository/tests/test_repo_commits.py
View file

@ -24,7 +24,9 @@ from rhodecode.lib.helpers import _shorten_commit_id
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_commit': '/{repo_name}/changeset/{commit_id}',
@ -55,23 +57,31 @@ class TestRepoCommitView(object):
def test_show_raw(self, backend):
commit_id = self.commit_id[backend.alias]
# webtest uses linter to check if response is bytes,
# and we use memoryview here as a wrapper, quick turn-off
self.app.lint = False
response = self.app.get(route_path(
'repo_commit_raw',
repo_name=backend.repo_name, commit_id=commit_id))
assert response.text == self.diffs[backend.alias]
assert response.body == self.diffs[backend.alias]
def test_show_raw_patch(self, backend):
response = self.app.get(route_path(
'repo_commit_patch', repo_name=backend.repo_name,
commit_id=self.commit_id[backend.alias]))
assert response.text == self.patches[backend.alias]
assert response.body == self.patches[backend.alias]
def test_commit_download(self, backend):
# webtest uses linter to check if response is bytes,
# and we use memoryview here as a wrapper, quick turn-off
self.app.lint = False
response = self.app.get(route_path(
'repo_commit_download',
repo_name=backend.repo_name,
commit_id=self.commit_id[backend.alias]))
assert response.text == self.diffs[backend.alias]
assert response.body == self.diffs[backend.alias]
def test_single_commit_page_different_ops(self, backend):
commit_id = {
@ -257,7 +267,7 @@ class TestRepoCommitView(object):
}
diffs = {
'hg': r"""diff --git a/README b/README
'hg': br"""diff --git a/README b/README
new file mode 120000
--- /dev/null
+++ b/README
@ -265,7 +275,7 @@ new file mode 120000
+README.rst
\ No newline at end of file
""",
'git': r"""diff --git a/README b/README
'git': br"""diff --git a/README b/README
new file mode 120000
index 0000000..92cacd2
--- /dev/null
@ -274,7 +284,7 @@ index 0000000..92cacd2
+README.rst
\ No newline at end of file
""",
'svn': """Index: README
'svn': b"""Index: README
===================================================================
diff --git a/README b/README
new file mode 10644
@ -287,7 +297,7 @@ new file mode 10644
}
patches = {
'hg': r"""# HG changeset patch
'hg': br"""# HG changeset patch
# User Marcin Kuzminski <marcin@python-works.com>
# Date 2014-01-07 12:21:40
# Node ID 2062ec7beeeaf9f44a1c25c41479565040b930b2
@ -296,7 +306,7 @@ new file mode 10644
Added a symlink
""" + diffs['hg'],
'git': r"""From fd627b9e0dd80b47be81af07c4a98518244ed2f7 2014-01-07 12:22:20
'git': br"""From fd627b9e0dd80b47be81af07c4a98518244ed2f7 2014-01-07 12:22:20
From: Marcin Kuzminski <marcin@python-works.com>
Date: 2014-01-07 12:22:20
Subject: [PATCH] Added a symlink
@ -304,7 +314,7 @@ Subject: [PATCH] Added a symlink
---
""" + diffs['git'],
'svn': r"""# SVN changeset patch
'svn': br"""# SVN changeset patch
# User marcin
# Date 2014-09-02 12:25:22.071142
# Revision 393

99
rhodecode/apps/repository/tests/test_repo_compare.py
View file

@ -27,7 +27,9 @@ from rhodecode.tests.utils import AssertResponse, commit_change
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_compare_select': '/{repo_name}/compare',
@ -65,30 +67,30 @@ class TestCompareView(object):
#
fork = backend.create_repo()
origin = backend.create_repo()
# prepare fork
commit0 = commit_change(
fork.repo_name, filename='file1', content='A',
message='A', vcs_type=backend.alias, parent=None, newfile=True)
fork.repo_name, filename=b'file1', content=b'A',
message='A - Initial Commit', vcs_type=backend.alias, parent=None, newfile=True)
commit1 = commit_change(
fork.repo_name, filename='file1', content='B',
fork.repo_name, filename=b'file1', content=b'B',
message='B, child of A', vcs_type=backend.alias, parent=commit0)
commit_change( # commit 2
fork.repo_name, filename='file1', content='C',
fork.repo_name, filename=b'file1', content=b'C',
message='C, child of B', vcs_type=backend.alias, parent=commit1)
commit3 = commit_change(
fork.repo_name, filename='file1', content='D',
fork.repo_name, filename=b'file1', content=b'D',
message='D, child of A', vcs_type=backend.alias, parent=commit0)
commit4 = commit_change(
fork.repo_name, filename='file1', content='E',
fork.repo_name, filename=b'file1', content=b'E',
message='E, child of D', vcs_type=backend.alias, parent=commit3)
# prepare origin repository, taking just the history up to D
origin = backend.create_repo()
origin_repo = origin.scm_instance(cache=False)
origin_repo.config.clear_section('hooks')
@ -98,7 +100,7 @@ class TestCompareView(object):
# Verify test fixture setup
# This does not work for git
if backend.alias != 'git':
assert 5 == len(fork.scm_instance().commit_ids)
assert 5 == len(fork.scm_instance(cache=False).commit_ids)
assert 2 == len(origin_repo.commit_ids)
# Comparing the revisions
@ -108,7 +110,8 @@ class TestCompareView(object):
source_ref_type="rev", source_ref=commit3.raw_id,
target_ref_type="rev", target_ref=commit4.raw_id,
params=dict(merge='1', target_repo=fork.repo_name)
))
),
status=200)
compare_page = ComparePage(response)
compare_page.contains_commits([commit4])
@ -119,7 +122,7 @@ class TestCompareView(object):
# commit something !
commit0 = commit_change(
repo1.repo_name, filename='file1', content='line1\n',
repo1.repo_name, filename=b'file1', content=b'line1\n',
message='commit1', vcs_type=backend.alias, parent=None,
newfile=True)
@ -128,11 +131,11 @@ class TestCompareView(object):
# add two extra commit into fork
commit1 = commit_change(
repo2.repo_name, filename='file1', content='line1\nline2\n',
repo2.repo_name, filename=b'file1', content=b'line1\nline2\n',
message='commit2', vcs_type=backend.alias, parent=commit0)
commit2 = commit_change(
repo2.repo_name, filename='file1', content='line1\nline2\nline3\n',
repo2.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
message='commit3', vcs_type=backend.alias, parent=commit1)
commit_id1 = repo1.scm_instance().DEFAULT_BRANCH_NAME
@ -167,7 +170,7 @@ class TestCompareView(object):
# commit something !
commit0 = commit_change(
repo1.repo_name, filename='file1', content='line1\n',
repo1.repo_name, filename=b'file1', content=b'line1\n',
message='commit1', vcs_type=backend.alias, parent=None,
newfile=True)
@ -176,17 +179,17 @@ class TestCompareView(object):
# now commit something to origin repo
commit_change(
repo1.repo_name, filename='file2', content='line1file2\n',
repo1.repo_name, filename=b'file2', content=b'line1file2\n',
message='commit2', vcs_type=backend.alias, parent=commit0,
newfile=True)
# add two extra commit into fork
commit1 = commit_change(
repo2.repo_name, filename='file1', content='line1\nline2\n',
repo2.repo_name, filename=b'file1', content=b'line1\nline2\n',
message='commit2', vcs_type=backend.alias, parent=commit0)
commit2 = commit_change(
repo2.repo_name, filename='file1', content='line1\nline2\nline3\n',
repo2.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
message='commit3', vcs_type=backend.alias, parent=commit1)
commit_id1 = repo1.scm_instance().DEFAULT_BRANCH_NAME
@ -250,11 +253,11 @@ class TestCompareView(object):
# commit something !
commit0 = commit_change(
repo1.repo_name, filename='file1', content='line1\n',
repo1.repo_name, filename=b'file1', content=b'line1\n',
message='commit1', vcs_type=backend.alias, parent=None,
newfile=True)
commit1 = commit_change(
repo1.repo_name, filename='file1', content='line1\nline2\n',
repo1.repo_name, filename=b'file1', content=b'line1\nline2\n',
message='commit2', vcs_type=backend.alias, parent=commit0)
# fork this repo
@ -262,19 +265,16 @@ class TestCompareView(object):
# now make commit3-6
commit2 = commit_change(
repo1.repo_name, filename='file1', content='line1\nline2\nline3\n',
repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
message='commit3', vcs_type=backend.alias, parent=commit1)
commit3 = commit_change(
repo1.repo_name, filename='file1',
content='line1\nline2\nline3\nline4\n', message='commit4',
vcs_type=backend.alias, parent=commit2)
repo1.repo_name, filename=b'file1',content=b'line1\nline2\nline3\nline4\n',
message='commit4', vcs_type=backend.alias, parent=commit2)
commit4 = commit_change(
repo1.repo_name, filename='file1',
content='line1\nline2\nline3\nline4\nline5\n', message='commit5',
vcs_type=backend.alias, parent=commit3)
repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\nline4\nline5\n',
message='commit5', vcs_type=backend.alias, parent=commit3)
commit_change( # commit 5
repo1.repo_name, filename='file1',
content='line1\nline2\nline3\nline4\nline5\nline6\n',
repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\nline4\nline5\nline6\n',
message='commit6', vcs_type=backend.alias, parent=commit4)
response = self.app.get(
@ -313,11 +313,11 @@ class TestCompareView(object):
# commit something !
commit0 = commit_change(
repo1.repo_name, filename='file1', content='line1\n',
repo1.repo_name, filename=b'file1', content=b'line1\n',
message='commit1', vcs_type=backend.alias, parent=None,
newfile=True)
commit1 = commit_change(
repo1.repo_name, filename='file1', content='line1\nline2\n',
repo1.repo_name, filename=b'file1', content=b'line1\nline2\n',
message='commit2', vcs_type=backend.alias, parent=commit0)
# fork this repo
@ -325,19 +325,19 @@ class TestCompareView(object):
# now make commit3-6
commit2 = commit_change(
repo1.repo_name, filename='file1', content='line1\nline2\nline3\n',
repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
message='commit3', vcs_type=backend.alias, parent=commit1)
commit3 = commit_change(
repo1.repo_name, filename='file1',
content='line1\nline2\nline3\nline4\n', message='commit4',
repo1.repo_name, filename=b'file1',
content=b'line1\nline2\nline3\nline4\n', message='commit4',
vcs_type=backend.alias, parent=commit2)
commit4 = commit_change(
repo1.repo_name, filename='file1',
content='line1\nline2\nline3\nline4\nline5\n', message='commit5',
repo1.repo_name, filename=b'file1',
content=b'line1\nline2\nline3\nline4\nline5\n', message='commit5',
vcs_type=backend.alias, parent=commit3)
commit5 = commit_change(
repo1.repo_name, filename='file1',
content='line1\nline2\nline3\nline4\nline5\nline6\n',
repo1.repo_name, filename=b'file1',
content=b'line1\nline2\nline3\nline4\nline5\nline6\n',
message='commit6', vcs_type=backend.alias, parent=commit4)
response = self.app.get(
@ -399,8 +399,8 @@ class TestCompareView(object):
r1_name = repo1.repo_name
commit0 = commit_change(
repo=r1_name, filename='file1',
content='line1', message='commit1', vcs_type=backend.alias,
repo=r1_name, filename=b'file1',
content=b'line1', message='commit1', vcs_type=backend.alias,
newfile=True)
assert repo1.scm_instance().commit_ids == [commit0.raw_id]
@ -412,20 +412,20 @@ class TestCompareView(object):
r2_name = repo2.repo_name
commit1 = commit_change(
repo=r2_name, filename='file1-fork',
content='file1-line1-from-fork', message='commit1-fork',
repo=r2_name, filename=b'file1-fork',
content=b'file1-line1-from-fork', message='commit1-fork',
vcs_type=backend.alias, parent=repo2.scm_instance()[-1],
newfile=True)
commit2 = commit_change(
repo=r2_name, filename='file2-fork',
content='file2-line1-from-fork', message='commit2-fork',
repo=r2_name, filename=b'file2-fork',
content=b'file2-line1-from-fork', message='commit2-fork',
vcs_type=backend.alias, parent=commit1,
newfile=True)
commit_change( # commit 3
repo=r2_name, filename='file3-fork',
content='file3-line1-from-fork', message='commit3-fork',
repo=r2_name, filename=b'file3-fork',
content=b'file3-line1-from-fork', message='commit3-fork',
vcs_type=backend.alias, parent=commit2, newfile=True)
# compare !
@ -446,8 +446,8 @@ class TestCompareView(object):
response.mustcontain('No commits in this compare')
commit0 = commit_change(
repo=r1_name, filename='file2',
content='line1-added-after-fork', message='commit2-parent',
repo=r1_name, filename=b'file2',
content=b'line1-added-after-fork', message='commit2-parent',
vcs_type=backend.alias, parent=None, newfile=True)
# compare !
@ -487,11 +487,10 @@ class TestCompareView(object):
def test_errors_when_comparing_unknown_source_repo(self, backend):
repo = backend.repo
badrepo = 'badrepo'
response = self.app.get(
self.app.get(
route_path('repo_compare',
repo_name=badrepo,
repo_name='badrepo',
source_ref_type="rev", source_ref='tip',
target_ref_type="rev", target_ref='tip',
params=dict(merge='1', target_repo=repo.repo_name)

8
rhodecode/apps/repository/tests/test_repo_compare_local.py
View file

@ -23,7 +23,9 @@ from .test_repo_compare import ComparePage
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_compare_select': '/{repo_name}/compare',
@ -55,8 +57,8 @@ class TestCompareView(object):
# outgoing commits between tags
commit_indexes = {
'git': [113] + range(115, 121),
'hg': [112] + range(115, 121),
'git': [113] + list(range(115, 121)),
'hg': [112] + list(range(115, 121)),
}
repo = backend.repo
commits = (repo.get_commit(commit_idx=idx)

18
rhodecode/apps/repository/tests/test_repo_compare_on_single_file.py
View file

@ -29,7 +29,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_compare_select': '/{repo_name}/compare',
@ -151,9 +153,9 @@ class TestSideBySideDiff(object):
@pytest.mark.xfail(reason='GIT does not handle empty commit compare correct (missing 1 commit)')
def test_diff_side_by_side_from_0_commit(self, app, backend, backend_stub):
f_path = 'test_sidebyside_file.py'
commit1_content = 'content-25d7e49c18b159446c\n'
commit2_content = 'content-603d6c72c46d953420\n'
f_path = b'test_sidebyside_file.py'
commit1_content = b'content-25d7e49c18b159446c\n'
commit2_content = b'content-603d6c72c46d953420\n'
repo = backend.create_repo()
commit1 = commit_change(
@ -185,9 +187,9 @@ class TestSideBySideDiff(object):
@pytest.mark.xfail(reason='GIT does not handle empty commit compare correct (missing 1 commit)')
def test_diff_side_by_side_from_0_commit_with_file_filter(self, app, backend, backend_stub):
f_path = 'test_sidebyside_file.py'
commit1_content = 'content-25d7e49c18b159446c\n'
commit2_content = 'content-603d6c72c46d953420\n'
f_path = b'test_sidebyside_file.py'
commit1_content = b'content-25d7e49c18b159446c\n'
commit2_content = b'content-603d6c72c46d953420\n'
repo = backend.create_repo()
commit1 = commit_change(
@ -222,7 +224,7 @@ class TestSideBySideDiff(object):
{'message': 'First commit'},
{'message': 'Second commit'},
{'message': 'Commit with binary',
'added': [nodes.FileNode('file.empty', content='')]},
'added': [nodes.FileNode(b'file.empty', content=b'')]},
]
f_path = 'file.empty'
repo = backend.create_repo(commits=commits)

5
rhodecode/apps/repository/tests/test_repo_feed.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -24,7 +23,9 @@ from rhodecode.tests import TestController
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'rss_feed_home': '/{repo_name}/feed-rss',

55
rhodecode/apps/repository/tests/test_repo_files.py
View file

@ -23,10 +23,11 @@ import mock
import pytest
from rhodecode.apps.repository.tests.test_repo_compare import ComparePage
from rhodecode.apps.repository.views.repo_files import RepoFilesView
from rhodecode.apps.repository.views.repo_files import RepoFilesView, get_archive_name, get_path_sha
from rhodecode.lib import helpers as h
from collections import OrderedDict
from rhodecode.lib.ext_json import json
from rhodecode.lib.str_utils import safe_str
from rhodecode.lib.vcs import nodes
from rhodecode.lib.vcs.conf import settings
@ -46,7 +47,9 @@ def get_node_history(backend_type):
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_summary': '/{repo_name}',
@ -506,7 +509,7 @@ class TestRawFileHandling(object):
def test_raw_svg_should_not_be_rendered(self, backend):
backend.create_repo()
backend.ensure_file("xss.svg")
backend.ensure_file(b"xss.svg")
response = self.app.get(
route_path('repo_file_raw',
repo_name=backend.repo_name,
@ -523,10 +526,10 @@ class TestRepositoryArchival(object):
backend.enable_downloads()
commit = backend.repo.get_commit(commit_idx=173)
for a_type, content_type, extension in settings.ARCHIVE_SPECS:
path_sha = get_path_sha('/')
filename = get_archive_name(backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha)
short = commit.short_id + extension
fname = commit.raw_id + extension
filename = '%s-%s' % (backend.repo_name, short)
response = self.app.get(
route_path('repo_archivefile',
repo_name=backend.repo_name,
@ -545,10 +548,10 @@ class TestRepositoryArchival(object):
backend.enable_downloads()
commit = backend.repo.get_commit(commit_idx=173)
for a_type, content_type, extension in settings.ARCHIVE_SPECS:
path_sha = get_path_sha('/')
filename = get_archive_name(backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha, with_hash=False)
short = 'plain' + extension
fname = commit.raw_id + extension
filename = '%s-%s' % (backend.repo_name, short)
response = self.app.get(
route_path('repo_archivefile',
repo_name=backend.repo_name,
@ -622,7 +625,7 @@ class TestFilesDiff(object):
commits = [
{'message': 'First commit'},
{'message': 'Commit with binary',
'added': [nodes.FileNode('file.bin', content='\0BINARY\0')]},
'added': [nodes.FileNode(b'file.bin', content='\0BINARY\0')]},
]
repo = backend.create_repo(commits=commits)
@ -899,7 +902,7 @@ class TestModifyFilesWithWebInterface(object):
def test_edit_file_view_not_on_branch(self, backend):
repo = backend.create_repo()
backend.ensure_file("vcs/nodes.py")
backend.ensure_file(b"vcs/nodes.py")
response = self.app.get(
route_path('repo_files_edit_file',
@ -912,7 +915,7 @@ class TestModifyFilesWithWebInterface(object):
def test_edit_file_view_commit_changes(self, backend, csrf_token):
repo = backend.create_repo()
backend.ensure_file("vcs/nodes.py", content="print 'hello'")
backend.ensure_file(b"vcs/nodes.py", content=b"print 'hello'")
response = self.app.post(
route_path('repo_files_update_file',
@ -934,7 +937,7 @@ class TestModifyFilesWithWebInterface(object):
def test_edit_file_view_commit_changes_default_message(self, backend,
csrf_token):
repo = backend.create_repo()
backend.ensure_file("vcs/nodes.py", content="print 'hello'")
backend.ensure_file(b"vcs/nodes.py", content=b"print 'hello'")
commit_id = (
backend.default_branch_name or
@ -967,7 +970,7 @@ class TestModifyFilesWithWebInterface(object):
def test_delete_file_view_not_on_branch(self, backend):
repo = backend.create_repo()
backend.ensure_file('vcs/nodes.py')
backend.ensure_file(b'vcs/nodes.py')
response = self.app.get(
route_path('repo_files_remove_file',
@ -980,7 +983,7 @@ class TestModifyFilesWithWebInterface(object):
def test_delete_file_view_commit_changes(self, backend, csrf_token):
repo = backend.create_repo()
backend.ensure_file("vcs/nodes.py")
backend.ensure_file(b"vcs/nodes.py")
response = self.app.post(
route_path('repo_files_delete_file',
@ -1015,12 +1018,18 @@ class TestFilesViewOtherCases(object):
'repo_files_add_file',
repo_name=repo.repo_name,
commit_id=0, f_path='')
add_new = f'<a class="alert-link" href="{repo_file_add_url}">add a new file</a>'
repo_file_upload_url = route_path(
'repo_files_upload_file',
repo_name=repo.repo_name,
commit_id=0, f_path='')
upload_new = f'<a class="alert-link" href="{repo_file_upload_url}">upload a new file</a>'
assert_session_flash(
response,
'There are no files yet. <a class="alert-link" '
'href="{}">Click here to add a new file.</a>'
.format(repo_file_add_url))
'There are no files yet. Click here to %s or %s.' % (add_new, upload_new)
)
def test_access_empty_repo_redirect_to_summary_with_alert_no_write_perms(
self, backend_stub, autologin_regular_user):
@ -1041,12 +1050,12 @@ class TestFilesViewOtherCases(object):
assert_session_flash(response, no_=repo_file_add_url)
@pytest.mark.parametrize('file_node', [
'archive/file.zip',
'diff/my-file.txt',
'render.py',
'render',
'remove_file',
'remove_file/to-delete.txt',
b'archive/file.zip',
b'diff/my-file.txt',
b'render.py',
b'render',
b'remove_file',
b'remove_file/to-delete.txt',
])
def test_file_names_equal_to_routes_parts(self, backend, file_node):
backend.create_repo()
@ -1055,7 +1064,7 @@ class TestFilesViewOtherCases(object):
self.app.get(
route_path('repo_files',
repo_name=backend.repo_name,
commit_id='tip', f_path=file_node),
commit_id='tip', f_path=safe_str(file_node)),
status=200)

4
rhodecode/apps/repository/tests/test_repo_forks.py
View file

@ -33,7 +33,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_summary': '/{repo_name}',

16
rhodecode/apps/repository/tests/test_repo_issue_tracker.py
View file

@ -19,14 +19,16 @@
import pytest
from rhodecode.lib.utils2 import md5
from rhodecode.lib.hash_utils import md5_safe
from rhodecode.model.db import Repository
from rhodecode.model.meta import Session
from rhodecode.model.settings import SettingsModel, IssueTrackerSettingsModel
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_summary': '/{repo_name}',
@ -69,9 +71,9 @@ class TestRepoIssueTracker(object):
self.app.post(post_url, post_data, status=302)
self.settings_model = IssueTrackerSettingsModel(repo=backend.repo)
settings = self.settings_model.get_repo_settings()
self.uid = md5(pattern)
self.uid = md5_safe(pattern)
assert settings[self.uid]['pat'] == pattern
self.another_uid = md5(another_pattern)
self.another_uid = md5_safe(another_pattern)
assert settings[self.another_uid]['pat'] == another_pattern
# test pattern
@ -95,7 +97,7 @@ class TestRepoIssueTracker(object):
entry_key = 'issuetracker_pat_'
pattern = 'issuetracker_pat2'
old_pattern = 'issuetracker_pat'
old_uid = md5(old_pattern)
old_uid = md5_safe(old_pattern)
sett = SettingsModel(repo=backend.repo).create_or_update_setting(
entry_key+old_uid, old_pattern, 'unicode')
@ -114,7 +116,7 @@ class TestRepoIssueTracker(object):
self.app.post(post_url, post_data, status=302)
self.settings_model = IssueTrackerSettingsModel(repo=backend.repo)
settings = self.settings_model.get_repo_settings()
self.uid = md5(pattern)
self.uid = md5_safe(pattern)
assert settings[self.uid]['pat'] == pattern
with pytest.raises(KeyError):
key = settings[old_uid]
@ -129,7 +131,7 @@ class TestRepoIssueTracker(object):
repo_name = repo.repo_name
entry_key = 'issuetracker_pat_'
pattern = 'issuetracker_pat3'
uid = md5(pattern)
uid = md5_safe(pattern)
settings_util.create_repo_rhodecode_setting(
repo=backend.repo, name=entry_key+uid,
value=entry_key, type_='unicode', cleanup=False)

4
rhodecode/apps/repository/tests/test_repo_maintainance.py
View file

@ -32,7 +32,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'edit_repo_maintenance': '/{repo_name}/settings/maintenance',

4
rhodecode/apps/repository/tests/test_repo_permissions.py
View file

@ -23,7 +23,9 @@ from rhodecode.tests.utils import permission_update_data_generator
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'edit_repo_perms': '/{repo_name}/settings/permissions'

65
rhodecode/apps/repository/tests/test_repo_pullrequests.py
View file

@ -36,7 +36,9 @@ from rhodecode.tests import (
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_changelog': '/{repo_name}/changelog',
@ -119,21 +121,21 @@ class TestPullrequestsView(object):
def test_show_versions_of_pr(self, backend, csrf_token):
commits = [
{'message': 'initial-commit',
'added': [FileNode('test-file.txt', 'LINE1\n')]},
'added': [FileNode(b'test-file.txt', b'LINE1\n')]},
{'message': 'commit-1',
'changed': [FileNode('test-file.txt', 'LINE1\nLINE2\n')]},
'changed': [FileNode(b'test-file.txt', b'LINE1\nLINE2\n')]},
# Above is the initial version of PR that changes a single line
# from now on we'll add 3x commit adding a nother line on each step
{'message': 'commit-2',
'changed': [FileNode('test-file.txt', 'LINE1\nLINE2\nLINE3\n')]},
'changed': [FileNode(b'test-file.txt', b'LINE1\nLINE2\nLINE3\n')]},
{'message': 'commit-3',
'changed': [FileNode('test-file.txt', 'LINE1\nLINE2\nLINE3\nLINE4\n')]},
'changed': [FileNode(b'test-file.txt', b'LINE1\nLINE2\nLINE3\nLINE4\n')]},
{'message': 'commit-4',
'changed': [FileNode('test-file.txt', 'LINE1\nLINE2\nLINE3\nLINE4\nLINE5\n')]},
'changed': [FileNode(b'test-file.txt', b'LINE1\nLINE2\nLINE3\nLINE4\nLINE5\n')]},
]
commit_ids = backend.create_master_repo(commits)
@ -413,14 +415,14 @@ class TestPullrequestsView(object):
'csrf_token': csrf_token})
assert_session_flash(
response, u'Pull request title & description updated.',
response, 'Pull request title & description updated.',
category='success')
pull_request = PullRequest.get(pull_request_id)
assert pull_request.title == 'New title'
assert pull_request.description == 'New description'
def test_edit_title_description(self, pr_util, csrf_token):
def test_edit_title_description_special(self, pr_util, csrf_token):
pull_request = pr_util.create_pull_request()
pull_request_id = pull_request.pull_request_id
@ -435,7 +437,7 @@ class TestPullrequestsView(object):
'csrf_token': csrf_token})
assert_session_flash(
response, u'Pull request title & description updated.',
response, 'Pull request title & description updated.',
category='success')
pull_request = PullRequest.get(pull_request_id)
@ -456,7 +458,7 @@ class TestPullrequestsView(object):
'description': 'New description',
'csrf_token': csrf_token}, status=200)
assert_session_flash(
response, u'Cannot update closed pull requests.',
response, 'Cannot update closed pull requests.',
category='error')
def test_update_invalid_source_reference(self, pr_util, csrf_token):
@ -483,7 +485,7 @@ class TestPullrequestsView(object):
from rhodecode.lib.vcs.backends.base import MergeFailureReason
pull_request = pr_util.create_pull_request(
approved=True, mergeable=True)
unicode_reference = u'branch:invalid-branch:invalid-commit-id'
unicode_reference = 'branch:invalid-branch:invalid-commit-id'
pull_request.target_ref = unicode_reference
Session().add(pull_request)
Session().commit()
@ -687,7 +689,7 @@ class TestPullrequestsView(object):
ChangesetComment.comment_id == comment_id).first().text
assert test_text == text_form_db
def test_comment_and_comment_edit(self, pr_util, csrf_token, xhr_header):
def test_comment_and_comment_edit_special(self, pr_util, csrf_token, xhr_header):
pull_request = pr_util.create_pull_request()
target_scm = pull_request.target_repo.scm_instance()
target_scm_name = target_scm.name
@ -867,13 +869,12 @@ class TestPullrequestsView(object):
# notifications properly with the new PR
commits = [
{'message': 'ancestor',
'added': [FileNode('file_A', content='content_of_ancestor')]},
'added': [FileNode(b'file_A', content=b'content_of_ancestor')]},
{'message': 'change',
'added': [FileNode('file_a', content='content_of_change')]},
'added': [FileNode(b'file_a', content=b'content_of_change')]},
{'message': 'change-child'},
{'message': 'ancestor-child', 'parents': ['ancestor'],
'added': [
FileNode('file_B', content='content_of_ancestor_child')]},
'added': [ FileNode(b'file_B', content=b'content_of_ancestor_child')]},
{'message': 'ancestor-child-2'},
]
commit_ids = backend.create_master_repo(commits)
@ -935,13 +936,13 @@ class TestPullrequestsView(object):
def test_create_pull_request_stores_ancestor_commit_id(self, backend, csrf_token):
commits = [
{'message': 'ancestor',
'added': [FileNode('file_A', content='content_of_ancestor')]},
'added': [FileNode(b'file_A', content=b'content_of_ancestor')]},
{'message': 'change',
'added': [FileNode('file_a', content='content_of_change')]},
'added': [FileNode(b'file_a', content=b'content_of_change')]},
{'message': 'change-child'},
{'message': 'ancestor-child', 'parents': ['ancestor'],
'added': [
FileNode('file_B', content='content_of_ancestor_child')]},
FileNode(b'file_B', content=b'content_of_ancestor_child')]},
{'message': 'ancestor-child-2'},
]
commit_ids = backend.create_master_repo(commits)
@ -1021,9 +1022,9 @@ class TestPullrequestsView(object):
actions = [log.action for log in user_logs]
pr_commit_ids = PullRequestModel()._get_commit_ids(pull_request)
expected_actions = [
u'repo.pull_request.close',
u'repo.pull_request.merge',
u'repo.pull_request.comment.create'
'repo.pull_request.close',
'repo.pull_request.merge',
'repo.pull_request.comment.create'
]
assert actions == expected_actions
@ -1121,8 +1122,8 @@ class TestPullrequestsView(object):
branch=backend.default_branch_name, commit_id=commit_ids['ancestor'])
pull_request.revisions = [commit_ids['change']]
pull_request.title = u"Test"
pull_request.description = u"Description"
pull_request.title = "Test"
pull_request.description = "Description"
pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
pull_request.pull_request_state = PullRequest.STATE_CREATED
Session().add(pull_request)
@ -1175,8 +1176,8 @@ class TestPullrequestsView(object):
branch=backend.default_branch_name, commit_id=commit_ids['ancestor'])
pull_request.revisions = [commit_ids['change']]
pull_request.title = u"Test"
pull_request.description = u"Description"
pull_request.title = "Test"
pull_request.description = "Description"
pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
pull_request.pull_request_state = PullRequest.STATE_CREATED
@ -1242,8 +1243,8 @@ class TestPullrequestsView(object):
commit_ids['feat-commit-1'],
commit_ids['feat-commit-2']
]
pull_request.title = u"Test"
pull_request.description = u"Description"
pull_request.title = "Test"
pull_request.description = "Description"
pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
pull_request.pull_request_state = PullRequest.STATE_CREATED
Session().add(pull_request)
@ -1292,8 +1293,8 @@ class TestPullrequestsView(object):
pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
branch=backend.default_branch_name, commit_id=commit_ids['ancestor'])
pull_request.revisions = [commit_ids['change']]
pull_request.title = u"Test"
pull_request.description = u"Description"
pull_request.title = "Test"
pull_request.description = "Description"
pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
pull_request.pull_request_state = PullRequest.STATE_CREATED
Session().add(pull_request)
@ -1340,8 +1341,8 @@ class TestPullrequestsView(object):
pull_request.target_ref = 'branch:{branch}:{commit_id}'.format(
branch=backend_git.default_branch_name, commit_id=commit_ids['old-feature'])
pull_request.revisions = [commit_ids['new-feature']]
pull_request.title = u"Test"
pull_request.description = u"Description"
pull_request.title = "Test"
pull_request.description = "Description"
pull_request.author = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
pull_request.pull_request_state = PullRequest.STATE_CREATED
Session().add(pull_request)

4
rhodecode/apps/repository/tests/test_repo_settings.py
View file

@ -32,7 +32,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'edit_repo': '/{repo_name}/settings',

6
rhodecode/apps/repository/tests/test_repo_settings_advanced.py
View file

@ -19,7 +19,7 @@
import pytest
from rhodecode.lib.utils2 import safe_unicode, safe_str
from rhodecode.lib.str_utils import safe_str
from rhodecode.model.db import Repository
from rhodecode.model.repo import RepoModel
from rhodecode.tests import (
@ -31,7 +31,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_summary_explicit': '/{repo_name}/summary',

8
rhodecode/apps/repository/tests/test_repo_summary.py
View file

@ -40,7 +40,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_summary': '/{repo_name}',
@ -276,8 +278,8 @@ class TestRepoLocation(object):
response = self.app.get(
route_path('repo_summary', repo_name=safe_str(repo_name)), status=302)
msg = 'The repository `%s` cannot be loaded in filesystem. ' \
'Please check if it exist, or is not damaged.' % repo_name
msg = f'The repository `{repo_name}` cannot be loaded in filesystem. ' \
f'Please check if it exist, or is not damaged.'
assert_session_flash(response, msg)
@pytest.mark.parametrize("suffix", [u'', u'ąęł'], ids=['', 'non-ascii'])

5
rhodecode/apps/repository/tests/test_repo_tags.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -23,7 +22,9 @@ from rhodecode.model.db import Repository
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'tags_home': '/{repo_name}/tags',

4
rhodecode/apps/repository/tests/test_repo_vcs_settings.py
View file

@ -37,7 +37,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'repo_summary': '/{repo_name}',

4
rhodecode/apps/repository/tests/test_vcs_settings.py
View file

@ -27,7 +27,9 @@ from rhodecode.tests.utils import AssertResponse
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
base_url = {
'edit_repo': '/{repo_name}/settings',

1
rhodecode/apps/search/tests/test_search.py
View file

@ -179,6 +179,7 @@ class TestSearchController(TestController):
def test_filters_are_not_applied_for_admin_user(self):
self.log_user()
with mock.patch('whoosh.searching.Searcher.search') as search_mock:
self.app.get(route_path('search'),
{'q': 'test query', 'type': 'commit'})
assert search_mock.call_count == 1

2
rhodecode/apps/ssh_support/tests/conftest.py
View file

@ -35,7 +35,7 @@ def dummy_conf_file(tmpdir):
conf.set('app:main', 'ssh.executable.svn', '/usr/bin/svnserve')
f_path = os.path.join(str(tmpdir), 'ssh_wrapper_test.ini')
with open(f_path, 'wb') as f:
with open(f_path, 'wt') as f:
conf.write(f)
return os.path.join(f_path)

3
rhodecode/apps/ssh_support/tests/test_server_git.py
View file

@ -18,7 +18,6 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import json
import os
import mock
@ -26,7 +25,7 @@ import pytest
from rhodecode.apps.ssh_support.lib.backends.git import GitServer
from rhodecode.apps.ssh_support.tests.conftest import plain_dummy_env, plain_dummy_user
from rhodecode.lib.ext_json import json
class GitServerCreator(object):
root = '/tmp/repo/path/'

12
rhodecode/apps/svn_support/tests/test_mod_dav_svn_config.py
View file

@ -39,7 +39,7 @@ class TestModDavSvnConfig(object):
def get_repo_group_mocks(cls, count=1):
repo_groups = []
for num in range(0, count):
full_path = u'/path/to/RepöGröúp-°µ {}'.format(num)
full_path = f'/path/to/RepöGröúp-°µ {num}'
repo_group_mock = mock.MagicMock()
repo_group_mock.full_path = full_path
repo_group_mock.full_path_splitted = full_path.split('/')
@ -78,7 +78,7 @@ class TestModDavSvnConfig(object):
def test_render_mod_dav_svn_config_with_alternative_template(self, tmpdir):
repo_groups = self.get_repo_group_mocks(count=10)
test_file_path = os.path.join(str(tmpdir), 'example.mako')
with open(test_file_path, 'wb') as f:
with open(test_file_path, 'wt') as f:
f.write('TEST_EXAMPLE\n')
generated_config = utils._render_mod_dav_svn_config(
@ -107,11 +107,11 @@ class TestModDavSvnConfig(object):
# Assert that correct configuration directive is present.
if list_parent_path:
assert not re.search('SVNListParentPath\s+Off', generated_config)
assert re.search('SVNListParentPath\s+On', generated_config)
assert not re.search(r'SVNListParentPath\s+Off', generated_config)
assert re.search(r'SVNListParentPath\s+On', generated_config)
else:
assert re.search('SVNListParentPath\s+Off', generated_config)
assert not re.search('SVNListParentPath\s+On', generated_config)
assert re.search(r'SVNListParentPath\s+Off', generated_config)
assert not re.search(r'SVNListParentPath\s+On', generated_config)
if use_ssl:
assert 'RequestHeader edit Destination ^https: http: early' \

4
rhodecode/apps/user_group/tests/test_user_groups.py
View file

@ -29,7 +29,9 @@ fixture = Fixture()
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

4
rhodecode/apps/user_group/tests/test_user_groups_permissions.py
View file

@ -23,7 +23,9 @@ from rhodecode.tests.utils import permission_update_data_generator
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

3
rhodecode/tests/__init__.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -214,11 +213,13 @@ def assert_session_flash(response, msg=None, category=None, no_=None):
msg = f'msg `{no_}` found in session flash.'
pytest.fail(safe_str(msg))
else:
if msg not in message_text:
fail_msg = f'msg `{msg}` not found in ' \
f'session flash: got `{message_text}` (type:{type(message_text)}) instead'
pytest.fail(safe_str(fail_msg))
if category:
assert category == message.category

1
rhodecode/tests/auth_external_test.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/config/__init__.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/config/test_environment.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/config/test_routing_links.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

4
rhodecode/tests/config/test_sanitize_settings.py
View file

@ -1,4 +1,4 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2020 RhodeCode GmbH
#
@ -118,7 +118,7 @@ class TestSanitizeVcsSettings(object):
_string_funcs = [
('vcs.svn.compatible_version', ''),
('vcs.hooks.protocol', 'http'),
('vcs.hooks.host', '127.0.0.1'),
('vcs.hooks.host', '*'),
('vcs.scm_app_implementation', 'http'),
('vcs.server', ''),
('vcs.server.protocol', 'http'),

2
rhodecode/tests/config/test_utils.py
View file

@ -1,4 +1,4 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2020 RhodeCode GmbH
#

132
rhodecode/tests/conftest.py Normal file
View file

@ -0,0 +1,132 @@
# Copyright (C) 2010-2020 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
"""
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 pytest
import logging
from rhodecode.authentication import AuthenticationPluginRegistry
from rhodecode.model.db import Permission, User
from rhodecode.model.meta import Session
from rhodecode.model.settings import SettingsModel
from rhodecode.model.user import UserModel
log = logging.getLogger(__name__)
@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.
"""
class AuthPluginManager(object):
def cleanup(self):
self._enable_plugins(['egg:rhodecode-enterprise-ce#rhodecode'])
def enable(self, plugins_list, override=None):
return self._enable_plugins(plugins_list, override)
def _enable_plugins(self, plugins_list, override=None):
override = override or {}
params = {
'auth_plugins': ','.join(plugins_list),
}
# helper translate some names to others, to fix settings code
name_map = {
'token': 'authtoken'
}
log.debug('enable_auth_plugins: enabling following auth-plugins: %s', plugins_list)
for module in plugins_list:
plugin_name = module.partition('#')[-1]
if plugin_name in name_map:
plugin_name = name_map[plugin_name]
enabled_plugin = f'auth_{plugin_name}_enabled'
cache_ttl = f'auth_{plugin_name}_cache_ttl'
# 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()
AuthenticationPluginRegistry.invalidate_auth_plugins_cache(hard=True)
enabled_plugins = SettingsModel().get_auth_plugins()
assert plugins_list == enabled_plugins
enabler = AuthPluginManager()
request.addfinalizer(enabler.cleanup)
return enabler
@pytest.fixture()
def test_user_factory(request, baseapp):
def user_factory(username='test_user', password='qweqwe', first_name='John', last_name='Testing', **kwargs):
usr = UserModel().create_or_update(
username=username,
password=password,
email=f'{username}@rhodecode.org',
firstname=first_name, lastname=last_name)
Session().commit()
for k, v in kwargs.items():
setattr(usr, k, v)
Session().add(usr)
assert User.get_by_username(username) == usr
@request.addfinalizer
def cleanup():
if UserModel().get_user(usr.user_id) is None:
return
perm = Permission.query().all()
for p in perm:
UserModel().revoke_perm(usr, p)
UserModel().delete(usr.user_id)
Session().commit()
return usr
return user_factory

167
rhodecode/tests/conftest_common.py Normal file
View file

@ -0,0 +1,167 @@
# Copyright (C) 2010-2020 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import pytest
from rhodecode.lib import ext_json
def get_backends_from_metafunc(metafunc):
requested_backends = set(metafunc.config.getoption('--backends'))
backend_mark = metafunc.definition.get_closest_marker('backends')
if backend_mark:
# Supported backends by this test function, created from
# pytest.mark.backends
backends = backend_mark.args
elif hasattr(metafunc.cls, 'backend_alias'):
# Support class attribute "backend_alias", this is mainly
# for legacy reasons for tests not yet using pytest.mark.backends
backends = [metafunc.cls.backend_alias]
else:
backends = metafunc.config.getoption('--backends')
return requested_backends.intersection(backends)
def pytest_addoption(parser):
def _parse_json(value):
return ext_json.str_json(value) if value else None
def _split_comma(value):
return value.split(',')
parser.addoption(
'--keep-tmp-path', action='store_true',
help="Keep the test temporary directories")
parser.addoption(
'--backends', action='store', type=_split_comma,
default=['git', 'hg', 'svn'],
help="Select which backends to test for backend specific tests.")
parser.addoption(
'--dbs', action='store', type=_split_comma,
default=['sqlite'],
help="Select which database to test for database specific tests. "
"Possible options are sqlite,postgres,mysql")
parser.addoption(
'--appenlight', '--ae', action='store_true',
help="Track statistics in appenlight.")
parser.addoption(
'--appenlight-api-key', '--ae-key',
help="API key for Appenlight.")
parser.addoption(
'--appenlight-url', '--ae-url',
default="https://ae.rhodecode.com",
help="Appenlight service URL, defaults to https://ae.rhodecode.com")
parser.addoption(
'--sqlite-connection-string', action='store',
default='', help="Connection string for the dbs tests with SQLite")
parser.addoption(
'--postgres-connection-string', action='store',
default='', help="Connection string for the dbs tests with Postgres")
parser.addoption(
'--mysql-connection-string', action='store',
default='', help="Connection string for the dbs tests with MySQL")
parser.addoption(
'--repeat', type=int, default=100,
help="Number of repetitions in performance tests.")
parser.addoption(
'--test-loglevel', dest='test_loglevel',
help="Set default Logging level for tests, critical(default), error, warn , 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.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
"""
Adding the remote traceback if the exception has this information.
VCSServer attaches this information as the attribute `_vcs_server_traceback`
to the exception instance.
"""
outcome = yield
report = outcome.get_result()
if call.excinfo:
exc = call.excinfo.value
vcsserver_traceback = getattr(exc, '_vcs_server_traceback', None)
if vcsserver_traceback and report.outcome == 'failed':
section = f'VCSServer remote traceback {report.when}'
report.sections.append((section, vcsserver_traceback))
def pytest_generate_tests(metafunc):
# Support test generation based on --backend parameter
if 'backend_alias' in metafunc.fixturenames:
backends = get_backends_from_metafunc(metafunc)
scope = None
if not backends:
pytest.skip("Not enabled for any of selected backends")
metafunc.parametrize('backend_alias', backends, scope=scope)
backend_mark = metafunc.definition.get_closest_marker('backends')
if backend_mark:
backends = get_backends_from_metafunc(metafunc)
if not backends:
pytest.skip("Not enabled for any of selected backends")

41
rhodecode/tests/database/conftest.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -20,13 +19,13 @@
from subprocess import Popen, PIPE
import os
import shutil
import sys
import tempfile
import pytest
from sqlalchemy.engine import url
from rhodecode.lib.str_utils import safe_str, safe_bytes
from rhodecode.tests.fixture import TestINI
@ -145,17 +144,19 @@ class DBBackend(object):
_env.update(env)
self.p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, env=_env)
self.stdout, self.stderr = self.p.communicate()
sys.stdout.write('COMMAND:'+command+'\n')
sys.stdout.write(self.stdout)
stdout_str = safe_str(self.stdout)
sys.stdout.write(f'COMMAND:{command}\n')
sys.stdout.write(stdout_str)
return self.stdout, self.stderr
def assert_returncode_success(self):
from rich import print as pprint
if not self.p.returncode == 0:
print(self.stderr)
raise AssertionError('non 0 retcode:{}'.format(self.p.returncode))
pprint(safe_str(self.stderr))
raise AssertionError(f'non 0 retcode:{self.p.returncode}')
def assert_correct_output(self, stdout, version):
assert 'UPGRADE FOR STEP {} COMPLETED'.format(version) in stdout
assert b'UPGRADE FOR STEP %b COMPLETED' % safe_bytes(version) in stdout
def setup_rhodecode_db(self, ini_params=None, env=None):
if not ini_params:
@ -233,11 +234,11 @@ class SQLiteDBBackend(DBBackend):
def import_dump(self, dumpname):
dump = os.path.join(self.fixture_store, dumpname)
target = os.path.join(self._basetemp, '{0.db_name}.sqlite'.format(self))
return self.execute('cp -v {} {}'.format(dump, target))
return self.execute(f'cp -v {dump} {target}')
def teardown_db(self):
return self.execute("rm -rf {}.sqlite".format(
os.path.join(self._basetemp, self.db_name)))
target_db = os.path.join(self._basetemp, self.db_name)
return self.execute(f"rm -rf {target_db}.sqlite")
class MySQLDBBackend(DBBackend):
@ -273,21 +274,15 @@ class PostgresDBBackend(DBBackend):
def setup_db(self):
# dump schema for tests
# pg_dump -U postgres -h localhost $TEST_DB_NAME
self._db_url = [{'app:main': {
'sqlalchemy.db1.url':
self.connection_string}}]
return self.execute("PGPASSWORD={} psql -U {} -h localhost "
"-c 'create database '{}';'".format(
self.password, self.user, self.db_name))
self._db_url = [{'app:main': {'sqlalchemy.db1.url': self.connection_string}}]
cmd = f"PGPASSWORD={self.password} psql -U {self.user} -h localhost -c 'create database '{self.db_name}';'"
return self.execute(cmd)
def teardown_db(self):
return self.execute("PGPASSWORD={} psql -U {} -h localhost "
"-c 'drop database if exists '{}';'".format(
self.password, self.user, self.db_name))
cmd = f"PGPASSWORD={self.password} psql -U {self.user} -h localhost -c 'drop database if exists '{self.db_name}';'"
return self.execute(cmd)
def import_dump(self, dumpname):
dump = os.path.join(self.fixture_store, dumpname)
return self.execute(
"PGPASSWORD={} psql -U {} -h localhost -d {} -1 "
"-f {}".format(
self.password, self.user, self.db_name, dump))
cmd = f"PGPASSWORD={self.password} psql -U {self.user} -h localhost -d {self.db_name} -1 -f {dump}"
return self.execute(cmd)

1
rhodecode/tests/database/test_creation.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/database/test_migration.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/events/__init__.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/events/test_pullrequest.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/events/test_repo.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

13
rhodecode/tests/fixture.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -148,7 +147,7 @@ class Fixture(object):
plugin = self._get_plugin()
plugin.create_or_update_setting('auth_restriction', auth_restriction)
Session().commit()
SettingsModel().invalidate_settings_cache()
SettingsModel().invalidate_settings_cache(hard=True)
def __exit__(self, exc_type, exc_val, exc_tb):
@ -156,7 +155,7 @@ class Fixture(object):
plugin.create_or_update_setting(
'auth_restriction', RhodeCodeAuthPlugin.AUTH_RESTRICTION_NONE)
Session().commit()
SettingsModel().invalidate_settings_cache()
SettingsModel().invalidate_settings_cache(hard=True)
return context()
@ -181,14 +180,14 @@ class Fixture(object):
plugin = self._get_plugin()
plugin.create_or_update_setting('scope_restriction', scope_restriction)
Session().commit()
SettingsModel().invalidate_settings_cache()
SettingsModel().invalidate_settings_cache(hard=True)
def __exit__(self, exc_type, exc_val, exc_tb):
plugin = self._get_plugin()
plugin.create_or_update_setting(
'scope_restriction', RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_ALL)
Session().commit()
SettingsModel().invalidate_settings_cache()
SettingsModel().invalidate_settings_cache(hard=True)
return context()
@ -399,7 +398,7 @@ class Fixture(object):
'gist_type': GistModel.cls.GIST_PUBLIC,
'lifetime': -1,
'acl_level': Gist.ACL_LEVEL_PUBLIC,
'gist_mapping': {'filename1.txt': {'content': 'hello world'},}
'gist_mapping': {b'filename1.txt': {'content': b'hello world'},}
}
form_data.update(kwargs)
gist = GistModel().create(
@ -420,7 +419,7 @@ class Fixture(object):
Session().commit()
def load_resource(self, resource_name, strip=False):
with open(os.path.join(FIXTURES, resource_name)) as f:
with open(os.path.join(FIXTURES, resource_name), 'rb') as f:
source = f.read()
if strip:
source = source.strip()

12
rhodecode/tests/fixture_mods/fixture_pyramid.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -30,7 +29,7 @@ def vcsserver(request, vcsserver_port, vcsserver_factory):
"""
Session scope VCSServer.
Tests wich need the VCSServer have to rely on this fixture in order
Tests which 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
@ -58,7 +57,7 @@ def vcsserver_factory(tmpdir_factory):
"""
def factory(request, overrides=(), vcsserver_port=None,
log_file=None):
log_file=None, workers='2'):
if vcsserver_port is None:
vcsserver_port = get_available_port()
@ -74,7 +73,7 @@ def vcsserver_factory(tmpdir_factory):
basetemp=tmpdir_factory.getbasetemp().strpath,
prefix='test_vcs_')
server = RcVCSServer(config_file, log_file)
server = RcVCSServer(config_file, log_file, workers)
server.start()
@request.addfinalizer
@ -100,7 +99,8 @@ def ini_config(request, tmpdir_factory, rcserver_port, vcsserver_port):
overrides = [
{'server:main': {'port': rcserver_port}},
{'app:main': {
'vcs.server': 'localhost:%s' % vcsserver_port,
'cache_dir': '%(here)s/rc_data',
'vcs.server': f'localhost:{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.
@ -109,7 +109,7 @@ def ini_config(request, tmpdir_factory, rcserver_port, vcsserver_port):
'vcs.server.protocol': 'http',
'vcs.scm_app_implementation': 'http',
'vcs.hooks.protocol': 'http',
'vcs.hooks.host': '127.0.0.1',
'vcs.hooks.host': '*',
}},
{'handler_console': {

187
rhodecode/tests/fixture_mods/fixture_utils.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -57,6 +56,7 @@ from rhodecode.model.integration import IntegrationModel
from rhodecode.integrations import integration_type_registry
from rhodecode.integrations.types.base import IntegrationTypeBase
from rhodecode.lib.utils import repo2db_mapper
from rhodecode.lib.str_utils import safe_bytes
from rhodecode.lib.hash_utils import sha1_safe
from rhodecode.lib.vcs.backends import get_backend
from rhodecode.lib.vcs.nodes import FileNode
@ -540,7 +540,7 @@ class Backend(object):
def create_repo(
self, commits=None, number_of_commits=0, heads=None,
name_suffix=u'', bare=False, **kwargs):
name_suffix='', bare=False, **kwargs):
"""
Create a repository and record it for later cleanup.
@ -585,14 +585,14 @@ class Backend(object):
self._cleanup_repos.append(self.repo_name)
return repo
def new_repo_name(self, suffix=u''):
def new_repo_name(self, suffix=''):
self.repo_name = self._next_repo_name() + suffix
self._cleanup_repos.append(self.repo_name)
return self.repo_name
def _next_repo_name(self):
return u"%s_%s" % (
self.invalid_repo_name.sub(u'_', self._test_name), len(self._cleanup_repos))
self.invalid_repo_name.sub('_', self._test_name), len(self._cleanup_repos))
def ensure_file(self, filename, content='Test content\n'):
assert self._cleanup_repos, "Avoid writing into vcs_test repos"
@ -634,14 +634,98 @@ class Backend(object):
repo.set_refs(ref_name, refs[ref_name])
def vcsbackend_base(request, backend_alias, tests_tmp_path, baseapp, test_repo):
class VcsBackend(object):
"""
Represents the test configuration for one supported vcs backend.
"""
invalid_repo_name = re.compile(r'[^0-9a-zA-Z]+')
def __init__(self, alias, repo_path, test_name, test_repo_container):
self.alias = alias
self._repo_path = repo_path
self._cleanup_repos = []
self._test_name = test_name
self._test_repo_container = test_repo_container
def __getitem__(self, key):
return self._test_repo_container(key, self.alias).scm_instance()
def __repr__(self):
return f'{self.__class__.__name__}(alias={self.alias}, repo={self._repo_path})'
@property
def repo(self):
"""
Returns the "current" repository. This is the vcs_test repo of the last
repo which has been created.
"""
Repository = get_backend(self.alias)
return Repository(self._repo_path)
@property
def backend(self):
"""
Returns the backend implementation class.
"""
return get_backend(self.alias)
def create_repo(self, commits=None, number_of_commits=0, _clone_repo=None,
bare=False):
repo_name = self._next_repo_name()
self._repo_path = get_new_dir(repo_name)
repo_class = get_backend(self.alias)
src_url = None
if _clone_repo:
src_url = _clone_repo.path
repo = repo_class(self._repo_path, create=True, src_url=src_url, bare=bare)
self._cleanup_repos.append(repo)
commits = commits or [
{'message': 'Commit %s of %s' % (x, repo_name)}
for x in range(number_of_commits)]
_add_commits_to_repo(repo, commits)
return repo
def clone_repo(self, repo):
return self.create_repo(_clone_repo=repo)
def cleanup(self):
for repo in self._cleanup_repos:
shutil.rmtree(repo.path)
def new_repo_path(self):
repo_name = self._next_repo_name()
self._repo_path = get_new_dir(repo_name)
return self._repo_path
def _next_repo_name(self):
return "{}_{}".format(
self.invalid_repo_name.sub('_', self._test_name),
len(self._cleanup_repos)
)
def add_file(self, repo, filename, content='Test content\n'):
imc = repo.in_memory_commit
imc.add(FileNode(safe_bytes(filename), content=safe_bytes(content)))
imc.commit(
message='Automatic commit from vcsbackend fixture',
author='Automatic <automatic@rhodecode.com>')
def ensure_file(self, filename, content='Test content\n'):
assert self._cleanup_repos, "Avoid writing into vcs_test repos"
self.add_file(self.repo, filename, content)
def vcsbackend_base(request, backend_alias, tests_tmp_path, baseapp, test_repo) -> VcsBackend:
if backend_alias not in request.config.getoption('--backends'):
pytest.skip("Backend %s not selected." % (backend_alias, ))
utils.check_xfail_backends(request.node, backend_alias)
utils.check_skip_backends(request.node, backend_alias)
repo_name = 'vcs_test_%s' % (backend_alias, )
repo_name = f'vcs_test_{backend_alias}'
repo_path = os.path.join(tests_tmp_path, repo_name)
backend = VcsBackend(
alias=backend_alias,
@ -691,85 +775,6 @@ def vcsbackend_stub(vcsbackend_git):
return vcsbackend_git
class VcsBackend(object):
"""
Represents the test configuration for one supported vcs backend.
"""
invalid_repo_name = re.compile(r'[^0-9a-zA-Z]+')
def __init__(self, alias, repo_path, test_name, test_repo_container):
self.alias = alias
self._repo_path = repo_path
self._cleanup_repos = []
self._test_name = test_name
self._test_repo_container = test_repo_container
def __getitem__(self, key):
return self._test_repo_container(key, self.alias).scm_instance()
@property
def repo(self):
"""
Returns the "current" repository. This is the vcs_test repo of the last
repo which has been created.
"""
Repository = get_backend(self.alias)
return Repository(self._repo_path)
@property
def backend(self):
"""
Returns the backend implementation class.
"""
return get_backend(self.alias)
def create_repo(self, commits=None, number_of_commits=0, _clone_repo=None,
bare=False):
repo_name = self._next_repo_name()
self._repo_path = get_new_dir(repo_name)
repo_class = get_backend(self.alias)
src_url = None
if _clone_repo:
src_url = _clone_repo.path
repo = repo_class(self._repo_path, create=True, src_url=src_url, bare=bare)
self._cleanup_repos.append(repo)
commits = commits or [
{'message': 'Commit %s of %s' % (x, repo_name)}
for x in range(number_of_commits)]
_add_commits_to_repo(repo, commits)
return repo
def clone_repo(self, repo):
return self.create_repo(_clone_repo=repo)
def cleanup(self):
for repo in self._cleanup_repos:
shutil.rmtree(repo.path)
def new_repo_path(self):
repo_name = self._next_repo_name()
self._repo_path = get_new_dir(repo_name)
return self._repo_path
def _next_repo_name(self):
return "%s_%s" % (
self.invalid_repo_name.sub('_', self._test_name),
len(self._cleanup_repos))
def add_file(self, repo, filename, content='Test content\n'):
imc = repo.in_memory_commit
imc.add(FileNode(filename, content=content))
imc.commit(
message=u'Automatic commit from vcsbackend fixture',
author=u'Automatic <automatic@rhodecode.com>')
def ensure_file(self, filename, content='Test content\n'):
assert self._cleanup_repos, "Avoid writing into vcs_test repos"
self.add_file(self.repo, filename, content)
def _add_commits_to_repo(vcs_repo, commits):
commit_ids = {}
if not commits:
@ -782,11 +787,11 @@ def _add_commits_to_repo(vcs_repo, commits):
message = str(commit.get('message', 'Commit %s' % idx))
for node in commit.get('added', []):
imc.add(FileNode(node.path, content=node.content))
imc.add(FileNode(safe_bytes(node.path), content=node.content))
for node in commit.get('changed', []):
imc.change(FileNode(node.path, content=node.content))
imc.change(FileNode(safe_bytes(node.path), content=node.content))
for node in commit.get('removed', []):
imc.remove(FileNode(node.path))
imc.remove(FileNode(safe_bytes(node.path)))
parents = [
vcs_repo.get_commit(commit_id=commit_ids[p])
@ -794,7 +799,7 @@ def _add_commits_to_repo(vcs_repo, commits):
operations = ('added', 'changed', 'removed')
if not any((commit.get(o) for o in operations)):
imc.add(FileNode('file_%s' % idx, content=message))
imc.add(FileNode(b'file_%b' % safe_bytes(str(idx)), content=safe_bytes(message)))
commit = imc.commit(
message=message,
@ -877,7 +882,7 @@ class PRTestUtility(object):
def create_pull_request(
self, commits=None, target_head=None, source_head=None,
revisions=None, approved=False, author=None, mergeable=False,
enable_notifications=True, name_suffix=u'', reviewers=None, observers=None,
enable_notifications=True, name_suffix='', reviewers=None, observers=None,
title=u"Test", description=u"Description"):
self.set_mergeable(mergeable)
if not enable_notifications:
@ -1002,7 +1007,7 @@ class PRTestUtility(object):
return comment
def create_inline_comment(
self, linked_to=None, line_no=u'n1', file_path='file_1'):
self, linked_to=None, line_no='n1', file_path='file_1'):
comment = CommentsModel().create(
text=u"Test comment",
repo=self.target_repository.repo_name,

12
rhodecode/tests/fixtures/diff_with_diff_data.diff vendored
View file

@ -245,8 +245,8 @@ index e34033e29fa9b3d3366b723beab129cee73869b9..b6e3f419778d6009229e9108824acaf8
+ 'author': 'Joe Doe <joe.doe@example.com>',
+ 'date': datetime.datetime(2010, 1, 1, 20),
+ 'added': [
+ FileNode('foobar', content='foobar'),
+ FileNode('foobar2', content='foobar2'),
+ FileNode(b'foobar', content='foobar'),
+ FileNode(b'foobar2', content='foobar2'),
+ ],
+ },
+ {
@ -254,10 +254,10 @@ index e34033e29fa9b3d3366b723beab129cee73869b9..b6e3f419778d6009229e9108824acaf8
+ 'author': 'Jane Doe <jane.doe@example.com>',
+ 'date': datetime.datetime(2010, 1, 1, 21),
+ 'added': [
+ FileNode('foobar3', content='foobar3'),
+ FileNode(b'foobar3', content='foobar3'),
+ ],
+ 'changed': [
+ FileNode('foobar', 'FOOBAR'),
+ FileNode(b'foobar', 'FOOBAR'),
+ ],
+ },
+ {
@ -265,9 +265,9 @@ index e34033e29fa9b3d3366b723beab129cee73869b9..b6e3f419778d6009229e9108824acaf8
+ 'author': 'Jane Doe <jane.doe@example.com>',
+ 'date': datetime.datetime(2010, 1, 1, 22),
+ 'changed': [
+ FileNode('foobar3', content='FOOBAR\nFOOBAR\nFOOBAR\n'),
+ FileNode(b'foobar3', content='FOOBAR\nFOOBAR\nFOOBAR\n'),
+ ],
+ 'removed': [FileNode('foobar')],
+ 'removed': [FileNode(b'foobar')],
+ },
+ ]
+ return commits

1
rhodecode/tests/functional/__init__.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

2
rhodecode/tests/functional/test_bad_request_data.py
View file

@ -1,4 +1,4 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2020 RhodeCode GmbH
#

6
rhodecode/tests/functional/test_delegated_admin.py
View file

@ -1,4 +1,4 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2020 RhodeCode GmbH
#
@ -25,7 +25,9 @@ from rhodecode.tests.fixture import Fixture
def route_path(name, params=None, **kwargs):
import urllib.request, urllib.parse, urllib.error
import urllib.request
import urllib.parse
import urllib.error
from rhodecode.apps._base import ADMIN_PREFIX
base_url = {

1
rhodecode/tests/functional/test_sessions.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/integrations/conftest.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/integrations/test_integration.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/integrations/test_slack.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/integrations/test_webhook.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/__init__.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/auth_modules/test_auth_modules.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/middleware/__init__.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/middleware/mock_scm_app.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/middleware/test_appenlight.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/middleware/test_simplegit.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/middleware/test_simplehg.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/middleware/test_simplesvn.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

67
rhodecode/tests/lib/middleware/test_simplevcs.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -18,11 +17,10 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import base64
import mock
import pytest
from rhodecode.lib.str_utils import base64_to_str
from rhodecode.lib.utils2 import AttributeDict
from rhodecode.tests.utils import CustomTestApp
@ -32,7 +30,7 @@ from rhodecode.lib.middleware import simplevcs
from rhodecode.lib.middleware.https_fixup import HttpsFixup
from rhodecode.lib.middleware.utils import scm_app_http
from rhodecode.model.db import User, _hash_key
from rhodecode.model.meta import Session
from rhodecode.model.meta import Session, cache as db_cache
from rhodecode.tests import (
HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
from rhodecode.tests.lib.middleware import mock_scm_app
@ -75,10 +73,13 @@ class StubVCSController(simplevcs.SimpleVCS):
@pytest.fixture()
def vcscontroller(baseapp, config_stub, request_stub):
from rhodecode.config.middleware import ce_auth_resources
config_stub.testing_securitypolicy()
config_stub.include('rhodecode.authentication')
config_stub.include('rhodecode.authentication.plugins.auth_rhodecode')
config_stub.include('rhodecode.authentication.plugins.auth_token')
for resource in ce_auth_resources:
config_stub.include(resource)
controller = StubVCSController(
baseapp.config.get_settings(), request_stub.registry)
@ -98,27 +99,45 @@ def _remove_default_user_from_query_cache():
user = User.get_default_user(cache=True)
query = Session().query(User).filter(User.username == user.username)
query = query.options(
FromCache("sql_cache_short", "get_user_%s" % _hash_key(user.username)))
query.invalidate()
FromCache("sql_cache_short", f"get_user_{_hash_key(user.username)}"))
db_cache.invalidate(
query, {},
FromCache("sql_cache_short", f"get_user_{_hash_key(user.username)}"))
Session().expire(user)
def test_handles_exceptions_during_permissions_checks(
vcscontroller, disable_anonymous_user):
user_and_pass = '%s:%s' % (TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
auth_password = base64.encodestring(user_and_pass).strip()
vcscontroller, disable_anonymous_user, enable_auth_plugins, test_user_factory):
test_password = 'qweqwe'
test_user = test_user_factory(password=test_password, extern_type='headers', extern_name='headers')
test_username = test_user.username
enable_auth_plugins.enable([
'egg:rhodecode-enterprise-ce#headers',
'egg:rhodecode-enterprise-ce#token',
'egg:rhodecode-enterprise-ce#rhodecode'],
override={
'egg:rhodecode-enterprise-ce#headers': {'auth_headers_header': 'REMOTE_USER'}
})
user_and_pass = f'{test_username}:{test_password}'
auth_password = base64_to_str(user_and_pass)
extra_environ = {
'AUTH_TYPE': 'Basic',
'HTTP_AUTHORIZATION': 'Basic %s' % auth_password,
'REMOTE_USER': TEST_USER_ADMIN_LOGIN,
'HTTP_AUTHORIZATION': f'Basic {auth_password}',
'REMOTE_USER': test_username,
}
# Verify that things are hooked up correctly
# Verify that things are hooked up correctly, we pass user with headers bound auth, and headers filled in
vcscontroller.get('/', status=200, extra_environ=extra_environ)
# Simulate trouble during permission checks
with mock.patch('rhodecode.model.db.User.get_by_username',
side_effect=Exception) as get_user:
side_effect=Exception('permission_error_test')) as get_user:
# Verify that a correct 500 is returned and check that the expected
# code path was hit.
vcscontroller.get('/', status=500, extra_environ=extra_environ)
@ -230,7 +249,7 @@ class TestShadowRepoExposure(object):
controller.is_shadow_repo = True
controller._action = 'pull'
controller._is_shadow_repo_dir = True
controller.stub_response_body = 'dummy body value'
controller.stub_response_body = (b'dummy body value',)
controller._get_default_cache_ttl = mock.Mock(
return_value=(False, 0))
@ -242,10 +261,10 @@ class TestShadowRepoExposure(object):
}
response = controller(environ_stub, mock.Mock())
response_body = ''.join(response)
response_body = b''.join(response)
# Assert that we got the response from the wsgi app.
assert response_body == controller.stub_response_body
assert response_body == b''.join(controller.stub_response_body)
def test_pull_on_shadow_repo_that_is_missing(self, baseapp, request_stub):
"""
@ -258,7 +277,7 @@ class TestShadowRepoExposure(object):
controller.is_shadow_repo = True
controller._action = 'pull'
controller._is_shadow_repo_dir = False
controller.stub_response_body = 'dummy body value'
controller.stub_response_body = (b'dummy body value',)
environ_stub = {
'HTTP_HOST': 'test.example.com',
'HTTP_ACCEPT': 'application/mercurial',
@ -267,10 +286,10 @@ class TestShadowRepoExposure(object):
}
response = controller(environ_stub, mock.Mock())
response_body = ''.join(response)
response_body = b''.join(response)
# Assert that we got the response from the wsgi app.
assert '404 Not Found' in response_body
assert b'404 Not Found' in response_body
def test_push_on_shadow_repo_raises(self, baseapp, request_stub):
"""
@ -281,7 +300,7 @@ class TestShadowRepoExposure(object):
controller._check_ssl = mock.Mock()
controller.is_shadow_repo = True
controller._action = 'push'
controller.stub_response_body = 'dummy body value'
controller.stub_response_body = (b'dummy body value',)
environ_stub = {
'HTTP_HOST': 'test.example.com',
'HTTP_ACCEPT': 'application/mercurial',
@ -290,11 +309,11 @@ class TestShadowRepoExposure(object):
}
response = controller(environ_stub, mock.Mock())
response_body = ''.join(response)
response_body = b''.join(response)
assert response_body != controller.stub_response_body
# Assert that a 406 error is returned.
assert '406 Not Acceptable' in response_body
assert b'406 Not Acceptable' in response_body
def test_set_repo_names_no_shadow(self, baseapp, request_stub):
"""

1
rhodecode/tests/lib/middleware/test_vcs.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/middleware/test_vcs_unavailable.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/middleware/utils/__init__.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

10
rhodecode/tests/lib/middleware/utils/test_scm_app_http.py
View file

@ -1,4 +1,4 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2020 RhodeCode GmbH
#
@ -55,20 +55,20 @@ def data():
def test_reuse_app_no_data(repeat, vcsserver_http_echo_app):
app = vcs_http_app(vcsserver_http_echo_app)
for x in range(repeat / 10):
for x in range(repeat // 10):
response = app.post('/')
assert response.status_code == 200
def test_reuse_app_with_data(data, repeat, vcsserver_http_echo_app):
app = vcs_http_app(vcsserver_http_echo_app)
for x in range(repeat / 10):
for x in range(repeat // 10):
response = app.post('/', params=data)
assert response.status_code == 200
def test_create_app_per_request_no_data(repeat, vcsserver_http_echo_app):
for x in range(repeat / 10):
for x in range(repeat // 10):
app = vcs_http_app(vcsserver_http_echo_app)
response = app.post('/')
assert response.status_code == 200
@ -76,7 +76,7 @@ def test_create_app_per_request_no_data(repeat, vcsserver_http_echo_app):
def test_create_app_per_request_with_data(
data, repeat, vcsserver_http_echo_app):
for x in range(repeat / 10):
for x in range(repeat // 10):
app = vcs_http_app(vcsserver_http_echo_app)
response = app.post('/', params=data)
assert response.status_code == 200

2
rhodecode/tests/lib/middleware/utils/test_scm_app_http_chunking.py
View file

@ -1,4 +1,4 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2020 RhodeCode GmbH
#

3
rhodecode/tests/lib/middleware/utils/test_wsgi_app_caller_client.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -82,7 +81,7 @@ def test_remote_app_caller():
('a1', 'a2', 'a3', 'a4', None))
# Note: RemoteAppCaller is expected to return a tuple like the
# following one
return (['content'], '200 OK', [('Content-Type', 'text/plain')])
return ([b'content'], '200 OK', [('Content-Type', 'text/plain')])
wrapper_app = wsgi_app_caller_client.RemoteAppCaller(
RemoteAppCallerMock(), 'a1', 'a2', arg3='a3', arg4='a4')

1
rhodecode/tests/lib/test_action_parser.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

1
rhodecode/tests/lib/test_audit_logger.py
View file

@ -1,5 +1,4 @@
import collections
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

16
rhodecode/tests/lib/test_auth.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#
@ -19,13 +18,13 @@
# and proprietary license terms, please see https://rhodecode.com/licenses/
import os
from hashlib import sha1
import pytest
from mock import patch
from rhodecode.lib import auth
from rhodecode.lib.utils2 import md5
from rhodecode.lib.str_utils import safe_bytes
from rhodecode.lib.hash_utils import md5_safe, sha1
from rhodecode.model.auth_token import AuthTokenModel
from rhodecode.model.db import Session, User
from rhodecode.model.repo import RepoModel
@ -638,7 +637,7 @@ def test_auth_user_get_cookie_store_for_normal_user(user_util):
expected_data = {
'username': user.username,
'user_id': user.user_id,
'password': md5(user.password),
'password': md5_safe(user.password),
'is_authenticated': False
}
assert auth_user.get_cookie_store() == expected_data
@ -650,7 +649,7 @@ def test_auth_user_get_cookie_store_for_default_user():
expected_data = {
'username': User.DEFAULT_USER,
'user_id': default_user.user_id,
'password': md5(default_user.password),
'password': md5_safe(default_user.password),
'is_authenticated': True
}
assert auth_user.get_cookie_store() == expected_data
@ -678,10 +677,10 @@ def get_permissions(user, **kwargs):
class TestGenerateAuthToken(object):
def test_salt_is_used_when_specified(self):
salt = 'abcde'
salt = b'abcde'
user_name = 'test_user'
result = auth.generate_auth_token(user_name, salt)
expected_result = sha1(user_name + salt).hexdigest()
expected_result = sha1(safe_bytes(user_name) + salt)
assert result == expected_result
def test_salt_is_geneated_when_not_specified(self):
@ -690,7 +689,8 @@ class TestGenerateAuthToken(object):
with patch.object(auth, 'os') as os_mock:
os_mock.urandom.return_value = random_salt
result = auth.generate_auth_token(user_name)
expected_result = sha1(user_name + random_salt).hexdigest()
expected_result = sha1(safe_bytes(user_name) + random_salt)
assert result == expected_result

1
rhodecode/tests/lib/test_auth_crypto_backend.py
View file

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2020 RhodeCode GmbH
#

Some files were not shown because too many files have changed in this diff Show more