tests: multiple tests cases fixes for python3
This commit is contained in:
parent
9469fdd4ed
commit
dd0b0f490c
16 changed files with 166 additions and 147 deletions
|
|
@ -51,7 +51,7 @@ class TestMyAccountSimpleViews(TestController):
|
|||
Repository.user == User.get_by_username(
|
||||
TEST_USER_ADMIN_LOGIN)).all()
|
||||
for repo in repos:
|
||||
response.mustcontain('"name_raw": "%s"' % repo.repo_name)
|
||||
response.mustcontain(f'"name_raw":"{repo.repo_name}"')
|
||||
|
||||
def test_my_account_my_watched(self, autologin_user):
|
||||
response = self.app.get(route_path('my_account_watched'))
|
||||
|
|
@ -60,8 +60,7 @@ class TestMyAccountSimpleViews(TestController):
|
|||
UserFollowing.user == User.get_by_username(
|
||||
TEST_USER_ADMIN_LOGIN)).all()
|
||||
for repo in repos:
|
||||
response.mustcontain(
|
||||
'"name_raw": "%s"' % repo.follows_repository.repo_name)
|
||||
response.mustcontain(f'"name_raw":"{repo.follows_repository.repo_name}"')
|
||||
|
||||
def test_my_account_perms(self, autologin_user):
|
||||
response = self.app.get(route_path('my_account_perms'))
|
||||
|
|
|
|||
|
|
@ -107,7 +107,7 @@ class TestChangelogController(TestController):
|
|||
assert expected_url in response.location
|
||||
response = response.follow()
|
||||
expected_warning = 'Branch {} is not found.'.format(branch)
|
||||
assert expected_warning in response.body
|
||||
assert expected_warning in response.text
|
||||
|
||||
@pytest.mark.xfail_backends("svn", reason="Depends on branch support")
|
||||
def test_changelog_filtered_by_branch_with_merges(
|
||||
|
|
|
|||
|
|
@ -59,20 +59,20 @@ class TestRepoCommitView(object):
|
|||
response = self.app.get(route_path(
|
||||
'repo_commit_raw',
|
||||
repo_name=backend.repo_name, commit_id=commit_id))
|
||||
assert response.body == self.diffs[backend.alias]
|
||||
assert response.text == 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.body == self.patches[backend.alias]
|
||||
assert response.text == self.patches[backend.alias]
|
||||
|
||||
def test_commit_download(self, backend):
|
||||
response = self.app.get(route_path(
|
||||
'repo_commit_download',
|
||||
repo_name=backend.repo_name,
|
||||
commit_id=self.commit_id[backend.alias]))
|
||||
assert response.body == self.diffs[backend.alias]
|
||||
assert response.text == self.diffs[backend.alias]
|
||||
|
||||
def test_single_commit_page_different_ops(self, backend):
|
||||
commit_id = {
|
||||
|
|
|
|||
|
|
@ -600,7 +600,7 @@ class TestCompareControllerSvn(object):
|
|||
status=200)
|
||||
|
||||
# It should show commits
|
||||
assert 'No commits in this compare' not in response.body
|
||||
assert 'No commits in this compare' not in response.text
|
||||
|
||||
# Should find only one file changed when comparing those two tags
|
||||
response.mustcontain('example.py')
|
||||
|
|
|
|||
|
|
@ -398,7 +398,7 @@ class TestFilesViews(object):
|
|||
repo_name=backend.repo_name,
|
||||
commit_id=commit.raw_id, f_path='README.rst'),
|
||||
extra_environ=xhr_header)
|
||||
assert response.body == ''
|
||||
assert response.text == ''
|
||||
|
||||
def test_nodetree_wrong_path(self, backend, xhr_header):
|
||||
commit = backend.repo.get_commit(commit_idx=173)
|
||||
|
|
@ -410,7 +410,7 @@ class TestFilesViews(object):
|
|||
|
||||
err = 'error: There is no file nor ' \
|
||||
'directory at the given path'
|
||||
assert err in response.body
|
||||
assert err in response.text
|
||||
|
||||
def test_nodetree_missing_xhr(self, backend):
|
||||
self.app.get(
|
||||
|
|
|
|||
|
|
@ -83,7 +83,7 @@ class TestRepoIssueTracker(object):
|
|||
repo_name=backend.repo.repo_name),
|
||||
extra_environ=xhr_header, params=data)
|
||||
|
||||
assert response.body == \
|
||||
assert response.text == \
|
||||
'example of <a class="tooltip issue-tracker-link" href="http://url" title="description">prefix</a> replacement'
|
||||
|
||||
@request.addfinalizer
|
||||
|
|
|
|||
|
|
@ -1644,7 +1644,7 @@ class TestPullrequestsControllerDelete(object):
|
|||
params={'csrf_token': csrf_token},
|
||||
status=200
|
||||
)
|
||||
assert response.body == 'true'
|
||||
assert response.text == 'true'
|
||||
|
||||
@pytest.mark.parametrize('url_type', [
|
||||
'pullrequest_new',
|
||||
|
|
|
|||
|
|
@ -22,11 +22,9 @@ import os
|
|||
import time
|
||||
import logging
|
||||
import datetime
|
||||
import hashlib
|
||||
import tempfile
|
||||
from os.path import join as jn
|
||||
|
||||
from tempfile import _RandomNameSequence
|
||||
import urllib.parse
|
||||
|
||||
import pytest
|
||||
|
||||
|
|
@ -34,8 +32,8 @@ from rhodecode.model.db import User
|
|||
from rhodecode.lib import auth
|
||||
from rhodecode.lib import helpers as h
|
||||
from rhodecode.lib.helpers import flash
|
||||
from rhodecode.lib.utils2 import safe_str
|
||||
|
||||
from rhodecode.lib.str_utils import safe_str
|
||||
from rhodecode.lib.hash_utils import sha1_safe
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -57,7 +55,7 @@ __all__ = [
|
|||
# SOME GLOBALS FOR TESTS
|
||||
TEST_DIR = tempfile.gettempdir()
|
||||
|
||||
TESTS_TMP_PATH = jn(TEST_DIR, 'rc_test_{}'.format(next(_RandomNameSequence())))
|
||||
TESTS_TMP_PATH = jn(TEST_DIR, 'rc_test_{}'.format(next(tempfile._RandomNameSequence())))
|
||||
TEST_USER_ADMIN_LOGIN = 'test_admin'
|
||||
TEST_USER_ADMIN_PASS = 'test12'
|
||||
TEST_USER_ADMIN_EMAIL = 'test_admin@mail.com'
|
||||
|
|
@ -85,12 +83,12 @@ SCM_TESTS = ['hg', 'git']
|
|||
uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
|
||||
|
||||
TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
|
||||
TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
|
||||
TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
|
||||
TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, f'vcsgitclone{uniq_suffix}')
|
||||
TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, f'vcsgitpull{uniq_suffix}')
|
||||
|
||||
TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)
|
||||
TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
|
||||
TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)
|
||||
TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, f'vcshgclone{uniq_suffix}')
|
||||
TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, f'vcshgpull{uniq_suffix}')
|
||||
|
||||
TEST_REPO_PREFIX = 'vcs-test'
|
||||
|
||||
|
|
@ -111,7 +109,7 @@ def get_new_dir(title):
|
|||
name_parts = [TEST_REPO_PREFIX]
|
||||
if title:
|
||||
name_parts.append(title)
|
||||
hex_str = hashlib.sha1('%s %s' % (os.getpid(), time.time())).hexdigest()
|
||||
hex_str = sha1_safe(f'{os.getpid()} {time.time()}')
|
||||
name_parts.append(hex_str)
|
||||
name = '-'.join(name_parts)
|
||||
path = os.path.join(TEST_DIR, name)
|
||||
|
|
@ -154,8 +152,8 @@ def login_user_session(
|
|||
response = app.post(
|
||||
h.route_path('login'),
|
||||
{'username': username, 'password': password})
|
||||
if 'invalid user name' in response.body:
|
||||
pytest.fail('could not login using %s %s' % (username, password))
|
||||
if 'invalid user name' in response.text:
|
||||
pytest.fail(f'could not login using {username} {password}')
|
||||
|
||||
assert response.status == '302 Found'
|
||||
response = response.follow()
|
||||
|
|
@ -203,9 +201,9 @@ def assert_session_flash(response, msg=None, category=None, no_=None):
|
|||
msg = _eval_if_lazy(msg)
|
||||
|
||||
if no_:
|
||||
error_msg = 'unable to detect no_ message `%s` in empty flash list' % no_
|
||||
error_msg = f'unable to detect no_ message `{no_}` in empty flash list'
|
||||
else:
|
||||
error_msg = 'unable to find message `%s` in empty flash list' % msg
|
||||
error_msg = f'unable to find message `{msg}` in empty flash list'
|
||||
assert messages, error_msg
|
||||
message = messages[0]
|
||||
|
||||
|
|
@ -213,13 +211,12 @@ def assert_session_flash(response, msg=None, category=None, no_=None):
|
|||
|
||||
if no_:
|
||||
if no_ in message_text:
|
||||
msg = u'msg `%s` found in session flash.' % (no_,)
|
||||
msg = f'msg `{no_}` found in session flash.'
|
||||
pytest.fail(safe_str(msg))
|
||||
else:
|
||||
if msg not in message_text:
|
||||
fail_msg = u'msg `%s` not found in session ' \
|
||||
u'flash: got `%s` (type:%s) instead' % (
|
||||
msg, message_text, type(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:
|
||||
|
|
@ -235,7 +232,7 @@ def no_newline_id_generator(test_name):
|
|||
Generates a test name without spaces or newlines characters. Used for
|
||||
nicer output of progress of test
|
||||
"""
|
||||
org_name = test_name
|
||||
|
||||
test_name = safe_str(test_name)\
|
||||
.replace('\n', '_N') \
|
||||
.replace('\r', '_N') \
|
||||
|
|
@ -246,7 +243,6 @@ def no_newline_id_generator(test_name):
|
|||
|
||||
|
||||
def route_path_generator(url_defs, name, params=None, **kwargs):
|
||||
import urllib.request, urllib.parse, urllib.error
|
||||
|
||||
base_url = url_defs[name].format(**kwargs)
|
||||
|
||||
|
|
|
|||
|
|
@ -143,7 +143,7 @@ def get_available_port(min_port=40000, max_port=55555):
|
|||
@pytest.fixture(scope='session')
|
||||
def rcserver_port(request):
|
||||
port = get_available_port()
|
||||
print('Using rhodecode port {}'.format(port))
|
||||
print(f'Using rhodecode port {port}')
|
||||
return port
|
||||
|
||||
|
||||
|
|
@ -152,7 +152,7 @@ def vcsserver_port(request):
|
|||
port = request.config.getoption('--vcsserver-port')
|
||||
if port is None:
|
||||
port = get_available_port()
|
||||
print('Using vcsserver port {}'.format(port))
|
||||
print(f'Using vcsserver port {port}')
|
||||
return port
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -20,7 +20,6 @@
|
|||
|
||||
import collections
|
||||
import datetime
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
import pprint
|
||||
|
|
@ -31,6 +30,7 @@ import time
|
|||
import uuid
|
||||
import dateutil.tz
|
||||
import logging
|
||||
import functools
|
||||
|
||||
import mock
|
||||
import pyramid.testing
|
||||
|
|
@ -40,7 +40,7 @@ import requests
|
|||
import pyramid.paster
|
||||
|
||||
import rhodecode
|
||||
from rhodecode.lib.utils2 import AttributeDict
|
||||
import rhodecode.lib
|
||||
from rhodecode.model.changeset_status import ChangesetStatusModel
|
||||
from rhodecode.model.comment import CommentsModel
|
||||
from rhodecode.model.db import (
|
||||
|
|
@ -57,6 +57,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.hash_utils import sha1_safe
|
||||
from rhodecode.lib.vcs.backends import get_backend
|
||||
from rhodecode.lib.vcs.nodes import FileNode
|
||||
from rhodecode.tests import (
|
||||
|
|
@ -74,6 +75,7 @@ def cmp(a, b):
|
|||
# backport cmp from python2 so we can still use it in the custom code in this module
|
||||
return (a > b) - (a < b)
|
||||
|
||||
|
||||
@pytest.fixture(scope='session', autouse=True)
|
||||
def activate_example_rcextensions(request):
|
||||
"""
|
||||
|
|
@ -1265,7 +1267,7 @@ class UserUtility(object):
|
|||
return cmp(second_group_parts, first_group_parts)
|
||||
|
||||
sorted_repo_group_ids = sorted(
|
||||
self.repo_group_ids, cmp=_repo_group_compare)
|
||||
self.repo_group_ids, key=functools.cmp_to_key(_repo_group_compare))
|
||||
for repo_group_id in sorted_repo_group_ids:
|
||||
self.fixture.destroy_repo_group(repo_group_id)
|
||||
|
||||
|
|
@ -1290,7 +1292,7 @@ class UserUtility(object):
|
|||
return cmp(second_group_parts, first_group_parts)
|
||||
|
||||
sorted_user_group_ids = sorted(
|
||||
self.user_group_ids, cmp=_user_group_compare)
|
||||
self.user_group_ids, key=functools.cmp_to_key(_user_group_compare))
|
||||
for user_group_id in sorted_user_group_ids:
|
||||
self.fixture.destroy_user_group(user_group_id)
|
||||
|
||||
|
|
@ -1429,8 +1431,7 @@ class SettingsUtility(object):
|
|||
|
||||
def create_repo_rhodecode_ui(
|
||||
self, repo, section, value, key=None, active=True, cleanup=True):
|
||||
key = key or hashlib.sha1(
|
||||
'{}{}{}'.format(section, value, repo.repo_id)).hexdigest()
|
||||
key = key or sha1_safe(f'{section}{value}{repo.repo_id}')
|
||||
|
||||
setting = RepoRhodeCodeUi()
|
||||
setting.repository_id = repo.repo_id
|
||||
|
|
@ -1447,7 +1448,7 @@ class SettingsUtility(object):
|
|||
|
||||
def create_rhodecode_ui(
|
||||
self, section, value, key=None, active=True, cleanup=True):
|
||||
key = key or hashlib.sha1('{}{}'.format(section, value)).hexdigest()
|
||||
key = key or sha1_safe(f'{section}{value}')
|
||||
|
||||
setting = RhodeCodeUi()
|
||||
setting.ui_section = section
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@ def test_vcs_available_returns_summary_page(app, backend):
|
|||
url = '/{repo_name}'.format(repo_name=backend.repo.repo_name)
|
||||
response = app.get(url)
|
||||
assert response.status_code == 200
|
||||
assert 'Summary' in response.body
|
||||
assert 'Summary' in response.text
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('autologin_user', 'app')
|
||||
|
|
@ -48,4 +48,4 @@ def test_vcs_unavailable_returns_vcs_error_page(app, backend):
|
|||
response = app.get(url, expect_errors=True)
|
||||
|
||||
assert response.status_code == 502
|
||||
assert 'Could not connect to VCS Server' in response.body
|
||||
assert 'Could not connect to VCS Server' in response.text
|
||||
|
|
|
|||
|
|
@ -96,4 +96,4 @@ def test_remote_app_caller():
|
|||
('Content-Type', 'text/plain'),
|
||||
('Content-Length', '7'),
|
||||
])
|
||||
assert response.body == 'content'
|
||||
assert response.text == 'content'
|
||||
|
|
|
|||
|
|
@ -21,113 +21,116 @@
|
|||
import pytest
|
||||
|
||||
from rhodecode.lib.encrypt import (
|
||||
AESCipher, SignatureVerificationError, InvalidDecryptedValue)
|
||||
from rhodecode.lib.encrypt2 import (Encryptor, InvalidToken)
|
||||
AESCipher, InvalidDecryptedValue)
|
||||
from rhodecode.lib import enc_utils
|
||||
from rhodecode.lib.str_utils import safe_str
|
||||
from rhodecode.lib.exceptions import SignatureVerificationError
|
||||
|
||||
|
||||
class TestEncryptModule(object):
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"key, text",
|
||||
[
|
||||
('a', 'short'),
|
||||
('a'*64, 'too long(trimmed to 32)'),
|
||||
('a'*32, 'just enough'),
|
||||
('ąćęćę', 'non asci'),
|
||||
('$asa$asa', 'special $ used'),
|
||||
]
|
||||
)
|
||||
def test_encryption(self, key, text):
|
||||
enc = AESCipher(key).encrypt(text)
|
||||
assert AESCipher(key).decrypt(enc) == text
|
||||
|
||||
def test_encryption_with_hmac(self):
|
||||
key = 'secret'
|
||||
text = 'ihatemysql'
|
||||
enc = AESCipher(key, hmac=True).encrypt(text)
|
||||
assert AESCipher(key, hmac=True).decrypt(enc) == text
|
||||
|
||||
def test_encryption_with_hmac_with_bad_key(self):
|
||||
key = 'secretstring'
|
||||
text = 'ihatemysql'
|
||||
enc = AESCipher(key, hmac=True).encrypt(text)
|
||||
|
||||
with pytest.raises(SignatureVerificationError) as e:
|
||||
assert AESCipher('differentsecret', hmac=True).decrypt(enc) == ''
|
||||
|
||||
assert 'Encryption signature verification failed' in str(e)
|
||||
|
||||
def test_encryption_with_hmac_with_bad_data(self):
|
||||
key = 'secret'
|
||||
text = 'ihatemysql'
|
||||
enc = AESCipher(key, hmac=True).encrypt(text)
|
||||
enc = 'xyz' + enc[3:]
|
||||
with pytest.raises(SignatureVerificationError) as e:
|
||||
assert AESCipher(key, hmac=True).decrypt(enc) == text
|
||||
|
||||
assert 'Encryption signature verification failed' in str(e)
|
||||
|
||||
def test_encryption_with_hmac_with_bad_key_not_strict(self):
|
||||
key = 'secretstring'
|
||||
text = 'ihatemysql'
|
||||
enc = AESCipher(key, hmac=True).encrypt(text)
|
||||
|
||||
assert isinstance(AESCipher(
|
||||
'differentsecret', hmac=True, strict_verification=False
|
||||
).decrypt(enc), InvalidDecryptedValue)
|
||||
@pytest.mark.parametrize(
|
||||
"algo", ['fernet', 'aes'],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"key, text",
|
||||
[
|
||||
(b'a', 'short'),
|
||||
(b'a' * 64, 'too long(trimmed to 32)'),
|
||||
(b'a' * 32, 'just enough'),
|
||||
('ąćęćę', 'non asci'),
|
||||
('$asa$asa', 'special $ used'),
|
||||
]
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"strict_mode", [True, False],
|
||||
)
|
||||
def test_common_encryption_module(algo, key, text, strict_mode):
|
||||
encrypted = enc_utils.encrypt_value(text, algo=algo, enc_key=key)
|
||||
decrypted = enc_utils.decrypt_value(encrypted, algo=algo, enc_key=key, strict_mode=strict_mode)
|
||||
assert text == safe_str(decrypted)
|
||||
|
||||
|
||||
class TestEncryptModule2(object):
|
||||
@pytest.mark.parametrize(
|
||||
"algo", ['fernet', 'aes'],
|
||||
)
|
||||
def test_encryption_with_bad_key(algo):
|
||||
key = b'secretstring'
|
||||
text = b'ihatemysql'
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"key, text",
|
||||
[
|
||||
('a', 'short'),
|
||||
('a'*64, 'too long(trimmed to 32)'),
|
||||
('a'*32, 'just enough'),
|
||||
('ąćęćę', 'non asci'),
|
||||
('$asa$asa', 'special $ used'),
|
||||
]
|
||||
)
|
||||
def test_encryption(self, key, text):
|
||||
enc = Encryptor(key).encrypt(text)
|
||||
assert Encryptor(key).decrypt(enc) == text
|
||||
encrypted = enc_utils.encrypt_value(text, algo=algo, enc_key=key)
|
||||
decrypted = enc_utils.decrypt_value(encrypted, algo=algo, enc_key=b'different-key', strict_mode=False)
|
||||
|
||||
def test_encryption_with_bad_key(self):
|
||||
key = 'secretstring'
|
||||
text = 'ihatemysql'
|
||||
enc = Encryptor(key).encrypt(text)
|
||||
assert decrypted[:22] == '<InvalidDecryptedValue'
|
||||
|
||||
assert Encryptor('differentsecret').decrypt(enc) == ''
|
||||
|
||||
def test_encryption_with_bad_key_raises(self):
|
||||
key = 'secretstring'
|
||||
text = 'ihatemysql'
|
||||
enc = Encryptor(key).encrypt(text)
|
||||
@pytest.mark.parametrize(
|
||||
"algo", ['fernet', 'aes'],
|
||||
)
|
||||
def test_encryption_with_bad_key_raises(algo):
|
||||
key = b'secretstring'
|
||||
text = b'ihatemysql'
|
||||
encrypted = enc_utils.encrypt_value(text, algo=algo, enc_key=key)
|
||||
|
||||
with pytest.raises(InvalidToken) as e:
|
||||
Encryptor('differentsecret').decrypt(enc, safe=False)
|
||||
with pytest.raises(SignatureVerificationError) as e:
|
||||
enc_utils.decrypt_value(encrypted, algo=algo, enc_key=b'different-key', strict_mode=True)
|
||||
|
||||
assert 'InvalidToken' in str(e)
|
||||
assert 'InvalidDecryptedValue' in str(e)
|
||||
|
||||
def test_encryption_with_bad_format_data(self):
|
||||
key = 'secret'
|
||||
text = 'ihatemysql'
|
||||
enc = Encryptor(key).encrypt(text)
|
||||
enc = '$xyz' + enc[3:]
|
||||
|
||||
with pytest.raises(ValueError) as e:
|
||||
Encryptor(key).decrypt(enc, safe=False)
|
||||
@pytest.mark.parametrize(
|
||||
"algo", ['fernet', 'aes'],
|
||||
)
|
||||
def test_encryption_with_bad_format_data(algo):
|
||||
key = b'secret'
|
||||
text = b'ihatemysql'
|
||||
encrypted = enc_utils.encrypt_value(text, algo=algo, enc_key=key)
|
||||
encrypted = b'$xyz' + encrypted[3:]
|
||||
|
||||
assert 'Encrypted Data has invalid format' in str(e)
|
||||
with pytest.raises(ValueError) as e:
|
||||
enc_utils.decrypt_value(encrypted, algo=algo, enc_key=key, strict_mode=True)
|
||||
|
||||
def test_encryption_with_bad_data(self):
|
||||
key = 'secret'
|
||||
text = 'ihatemysql'
|
||||
enc = Encryptor(key).encrypt(text)
|
||||
enc = enc[:-5]
|
||||
assert 'Encrypted Data has invalid format' in str(e)
|
||||
|
||||
with pytest.raises(InvalidToken) as e:
|
||||
Encryptor(key).decrypt(enc, safe=False)
|
||||
|
||||
assert 'InvalidToken' in str(e)
|
||||
@pytest.mark.parametrize(
|
||||
"algo", ['fernet', 'aes'],
|
||||
)
|
||||
def test_encryption_with_bad_data(algo):
|
||||
key = b'secret'
|
||||
text = b'ihatemysql'
|
||||
encrypted = enc_utils.encrypt_value(text, algo=algo, enc_key=key)
|
||||
encrypted = encrypted[:-5]
|
||||
|
||||
with pytest.raises(SignatureVerificationError) as e:
|
||||
enc_utils.decrypt_value(encrypted, algo=algo, enc_key=key, strict_mode=True)
|
||||
|
||||
assert 'SignatureVerificationError' in str(e)
|
||||
|
||||
|
||||
def test_encryption_with_hmac():
|
||||
key = b'secret'
|
||||
text = b'ihatemysql'
|
||||
enc = AESCipher(key, hmac=True).encrypt(text)
|
||||
assert AESCipher(key, hmac=True).decrypt(enc) == text
|
||||
|
||||
|
||||
def test_encryption_with_hmac_with_bad_data():
|
||||
key = b'secret'
|
||||
text = b'ihatemysql'
|
||||
enc = AESCipher(key, hmac=True).encrypt(text)
|
||||
enc = b'xyz' + enc[3:]
|
||||
with pytest.raises(SignatureVerificationError) as e:
|
||||
assert AESCipher(key, hmac=True).decrypt(enc, safe=False) == text
|
||||
|
||||
assert 'SignatureVerificationError' in str(e)
|
||||
|
||||
|
||||
def test_encryption_with_hmac_with_bad_key_not_strict():
|
||||
key = b'secretstring'
|
||||
text = b'ihatemysql'
|
||||
enc = AESCipher(key, hmac=True).encrypt(text)
|
||||
|
||||
decrypted = AESCipher(
|
||||
b'differentsecret', hmac=True, strict_verification=False
|
||||
).decrypt(enc)
|
||||
|
||||
assert isinstance(decrypted, InvalidDecryptedValue)
|
||||
|
|
|
|||
|
|
@ -439,7 +439,7 @@ def main(argv):
|
|||
return 1
|
||||
|
||||
sizes = options.sizes.split(',')
|
||||
sizes = map(int, sizes)
|
||||
sizes = list(map(int, sizes))
|
||||
|
||||
base_dir = options.dir
|
||||
api_key = options.api_key
|
||||
|
|
|
|||
|
|
@ -24,7 +24,9 @@ import logging
|
|||
import os.path
|
||||
import subprocess
|
||||
import tempfile
|
||||
import urllib.request, urllib.error, urllib.parse
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
from lxml.html import fromstring, tostring
|
||||
from lxml.cssselect import CSSSelector
|
||||
from urllib.parse import unquote_plus
|
||||
|
|
@ -54,7 +56,7 @@ log = logging.getLogger(__name__)
|
|||
class CustomTestResponse(TestResponse):
|
||||
|
||||
def _save_output(self, out):
|
||||
f = tempfile.NamedTemporaryFile(delete=False, prefix='rc-test-', suffix='.html')
|
||||
f = tempfile.NamedTemporaryFile(mode='w', delete=False, prefix='rc-test-', suffix='.html')
|
||||
f.write(out)
|
||||
return f.name
|
||||
|
||||
|
|
@ -82,7 +84,7 @@ class CustomTestResponse(TestResponse):
|
|||
f = self._save_output(str(self))
|
||||
|
||||
for s in strings:
|
||||
if not s in self:
|
||||
if s not in self:
|
||||
print_stderr("Actual response (no %r):" % s)
|
||||
print_stderr("body output saved as `%s`" % f)
|
||||
if print_body:
|
||||
|
|
@ -113,13 +115,24 @@ class CustomTestResponse(TestResponse):
|
|||
|
||||
class TestRequest(webob.BaseRequest):
|
||||
|
||||
# for py.test
|
||||
# for py.test, so it doesn't try to run this tas by name starting with test...
|
||||
disabled = True
|
||||
ResponseClass = CustomTestResponse
|
||||
|
||||
def add_response_callback(self, callback):
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def blank(cls, path, environ=None, base_url=None,
|
||||
headers=None, POST=None, **kw):
|
||||
|
||||
if not path.isascii():
|
||||
# our custom quote path if it contains non-ascii chars
|
||||
path = urllib.parse.quote(path)
|
||||
|
||||
return super(TestRequest, cls).blank(
|
||||
path, environ=environ, base_url=base_url, headers=headers, POST=POST, **kw)
|
||||
|
||||
|
||||
class CustomTestApp(TestApp):
|
||||
"""
|
||||
|
|
@ -260,6 +273,7 @@ class AssertResponse(object):
|
|||
def element_equals_to(self, css_selector, expected_content):
|
||||
element = self.get_element(css_selector)
|
||||
element_text = self._element_to_string(element)
|
||||
|
||||
assert expected_content in element_text
|
||||
|
||||
def element_contains(self, css_selector, expected_content):
|
||||
|
|
@ -306,7 +320,7 @@ class AssertResponse(object):
|
|||
|
||||
def _element_to_string(self, element):
|
||||
fromstring, tostring, CSSSelector = self.get_imports()
|
||||
return tostring(element)
|
||||
return tostring(element, encoding='unicode')
|
||||
|
||||
|
||||
class _Url(object):
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ import logging
|
|||
import os
|
||||
import tempfile
|
||||
|
||||
from rhodecode.lib.str_utils import safe_str
|
||||
from rhodecode.tests import GIT_REPO, HG_REPO
|
||||
|
||||
DEBUG = True
|
||||
|
|
@ -68,6 +69,10 @@ class Command(object):
|
|||
self.process = Popen(command, shell=True, stdout=PIPE, stderr=PIPE,
|
||||
cwd=self.cwd, env=env)
|
||||
stdout, stderr = self.process.communicate()
|
||||
|
||||
stdout = safe_str(stdout)
|
||||
stderr = safe_str(stderr)
|
||||
|
||||
if DEBUG:
|
||||
log.debug('STDOUT:%s', stdout)
|
||||
log.debug('STDERR:%s', stderr)
|
||||
|
|
@ -78,8 +83,9 @@ class Command(object):
|
|||
|
||||
|
||||
def _add_files(vcs, dest, clone_url=None, tags=None, target_branch=None, new_branch=False, **kwargs):
|
||||
git_ident = "git config user.name {} && git config user.email {}".format(
|
||||
'Marcin Kuźminski', 'me@email.com')
|
||||
full_name = 'Marcin Kuźminski'
|
||||
email = 'me@email.com'
|
||||
git_ident = f"git config user.name {full_name} && git config user.email {email}"
|
||||
cwd = path = jn(dest)
|
||||
|
||||
tags = tags or []
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue