Skip to content

slapos
slapos
Commit f5929506 authored by Jérome Perrin's avatar Jérome Perrin
Browse files
Options

software/erp5/test: drop python2 support 

we'll only run the test on python3
parent d0bb606d
Hide whitespace changes
Inline Side-by-side
Showing with 83 additions and 103 deletions
software/erp5/test/setup.py
View file @ f5929506
......@@ -24,7 +24,6 @@
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from __future__ import absolute_import
from setuptools import setup, find_packages
version = '0.0.1.dev0'
......@@ -49,7 +48,6 @@ setup(name=name,
'psutil',
'requests',
'mysqlclient',
'backports.lzma',
'cryptography',
'pexpect',
'pyOpenSSL',
......
software/erp5/test/test/__init__.py
View file @ f5929506
......@@ -25,7 +25,6 @@
#
##############################################################################
from __future__ import absolute_import
import json
import os
......
software/erp5/test/test/test_balancer.py
View file @ f5929506
from __future__ import absolute_import
import glob
import hashlib
import json
......@@ -9,10 +8,10 @@ import shutil
import subprocess
import tempfile
import time
import six.moves.urllib.request, six.moves.urllib.parse, six.moves.urllib.error
from six.moves.BaseHTTPServer import BaseHTTPRequestHandler
import urllib.parse
from http.server import BaseHTTPRequestHandler
import mock
from unittest import mock
import OpenSSL.SSL
import pexpect
import psutil
......@@ -27,7 +26,6 @@ from slapos.testing.utils import (CrontabMixin, ManagedHTTPServer,
findFreeTCPPort)
from . import ERP5InstanceTestCase, setUpModule
from six.moves import range
setUpModule # pyflakes
......@@ -100,7 +98,7 @@ class CaucaseService(ManagedResource):
os.mkdir(os.path.join(caucased_dir, 'user'))
os.mkdir(os.path.join(caucased_dir, 'service'))
backend_caucased_netloc = '%s:%s' % (self._cls._ipv4_address, findFreeTCPPort(self._cls._ipv4_address))
backend_caucased_netloc = f'{self._cls._ipv4_address}:{findFreeTCPPort(self._cls._ipv4_address)}'
self.url = 'http://' + backend_caucased_netloc
self._caucased_process = subprocess.Popen(
[
......@@ -213,7 +211,7 @@ class TestTimeout(BalancerTestCase, CrontabMixin):
@classmethod
def _getInstanceParameterDict(cls):
# type: () -> dict
parameter_dict = super(TestTimeout, cls)._getInstanceParameterDict()
parameter_dict = super()._getInstanceParameterDict()
# use a slow server instead
parameter_dict['dummy_http_server'] = [[cls.getManagedResource("slow_web_server", SlowHTTPServer).netloc, 1, False]]
# and set timeout of 1 second
......@@ -224,12 +222,12 @@ class TestTimeout(BalancerTestCase, CrontabMixin):
# type: () -> None
self.assertEqual(
requests.get(
six.moves.urllib.parse.urljoin(self.default_balancer_url, '/1'),
urllib.parse.urljoin(self.default_balancer_url, '/1'),
verify=False).status_code,
requests.codes.ok)
self.assertEqual(
requests.get(
six.moves.urllib.parse.urljoin(self.default_balancer_url, '/5'),
urllib.parse.urljoin(self.default_balancer_url, '/5'),
verify=False).status_code,
requests.codes.gateway_timeout)
......@@ -241,7 +239,7 @@ class TestLog(BalancerTestCase, CrontabMixin):
@classmethod
def _getInstanceParameterDict(cls):
# type: () -> dict
parameter_dict = super(TestLog, cls)._getInstanceParameterDict()
parameter_dict = super()._getInstanceParameterDict()
# use a slow server instead
parameter_dict['dummy_http_server'] = [[cls.getManagedResource("slow_web_server", SlowHTTPServer).netloc, 1, False]]
return parameter_dict
......@@ -249,7 +247,7 @@ class TestLog(BalancerTestCase, CrontabMixin):
def test_access_log_format(self):
# type: () -> None
requests.get(
six.moves.urllib.parse.urljoin(self.default_balancer_url, '/url_path'),
urllib.parse.urljoin(self.default_balancer_url, '/url_path'),
verify=False,
)
time.sleep(.5) # wait a bit more until access is logged
......@@ -288,7 +286,7 @@ class TestLog(BalancerTestCase, CrontabMixin):
'apachedex',
'ApacheDex-*.html',
))
with open(apachedex_report, 'r') as f:
with open(apachedex_report) as f:
report_text = f.read()
self.assertIn('APacheDEX', report_text)
# having this table means that apachedex could parse some lines.
......@@ -336,7 +334,7 @@ class TestLog(BalancerTestCase, CrontabMixin):
error_line = error_log_file.read().splitlines()[-1]
self.assertIn('proxy family_default has no server available!', error_line)
# this log also include a timestamp
self.assertRegexpMatches(error_line, r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}')
self.assertRegex(error_line, r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}')
class BalancerCookieHTTPServer(ManagedHTTPServer):
......@@ -377,7 +375,7 @@ class TestBalancer(BalancerTestCase):
@classmethod
def _getInstanceParameterDict(cls):
# type: () -> dict
parameter_dict = super(TestBalancer, cls)._getInstanceParameterDict()
parameter_dict = super()._getInstanceParameterDict()
# use two backend servers
parameter_dict['dummy_http_server'] = [
......@@ -409,7 +407,7 @@ class TestBalancer(BalancerTestCase):
# if backend provides a "SERVERID" cookie, balancer will overwrite it with the
# backend selected by balancing algorithm
self.assertIn(
requests.get(six.moves.urllib.parse.urljoin(self.default_balancer_url, '/set_cookie'), verify=False).cookies['SERVERID'],
requests.get(urllib.parse.urljoin(self.default_balancer_url, '/set_cookie'), verify=False).cookies['SERVERID'],
('default-0', 'default-1'),
)
......@@ -457,10 +455,7 @@ class TestTestRunnerEntryPoints(BalancerTestCase):
@classmethod
def _getInstanceParameterDict(cls):
# type: () -> dict
parameter_dict = super(
TestTestRunnerEntryPoints,
cls,
)._getInstanceParameterDict()
parameter_dict = super()._getInstanceParameterDict()
parameter_dict['dummy_http_server-test-runner-address-list'] = [
[
......@@ -485,18 +480,18 @@ class TestTestRunnerEntryPoints(BalancerTestCase):
)['default-test-runner-url-list']
url_0, url_1, url_2 = test_runner_url_list
self.assertEqual(
six.moves.urllib.parse.urlparse(url_0).netloc,
six.moves.urllib.parse.urlparse(url_1).netloc)
urllib.parse.urlparse(url_0).netloc,
urllib.parse.urlparse(url_1).netloc)
self.assertEqual(
six.moves.urllib.parse.urlparse(url_0).netloc,
six.moves.urllib.parse.urlparse(url_2).netloc)
urllib.parse.urlparse(url_0).netloc,
urllib.parse.urlparse(url_2).netloc)
path_0 = '/VirtualHostBase/https/{netloc}/VirtualHostRoot/_vh_unit_test_0/something'.format(
netloc=six.moves.urllib.parse.urlparse(url_0).netloc)
netloc=urllib.parse.urlparse(url_0).netloc)
path_1 = '/VirtualHostBase/https/{netloc}/VirtualHostRoot/_vh_unit_test_1/something'.format(
netloc=six.moves.urllib.parse.urlparse(url_0).netloc)
netloc=urllib.parse.urlparse(url_0).netloc)
path_2 = '/VirtualHostBase/https/{netloc}/VirtualHostRoot/_vh_unit_test_2/something'.format(
netloc=six.moves.urllib.parse.urlparse(url_0).netloc)
netloc=urllib.parse.urlparse(url_0).netloc)
self.assertEqual(
{
......@@ -534,7 +529,7 @@ class TestHTTP(BalancerTestCase):
@classmethod
def _getInstanceParameterDict(cls):
# type: () -> dict
parameter_dict = super(TestHTTP, cls)._getInstanceParameterDict()
parameter_dict = super()._getInstanceParameterDict()
# use a HTTP/1.1 server instead
parameter_dict['dummy_http_server'] = [[cls.getManagedResource("HTTP/1.1 Server", EchoHTTP11Server).netloc, 1, False]]
return parameter_dict
......@@ -574,7 +569,7 @@ class TestHTTP(BalancerTestCase):
session.get(self.default_balancer_url).raise_for_status()
new_conn.assert_not_called()
parsed_url = six.moves.urllib.parse.urlparse(self.default_balancer_url)
parsed_url = urllib.parse.urlparse(self.default_balancer_url)
# check that we have an open file for the ip connection
self.assertTrue([
......@@ -617,7 +612,7 @@ class TestContentEncoding(BalancerTestCase):
@classmethod
def _getInstanceParameterDict(cls):
# type: () -> dict
parameter_dict = super(TestContentEncoding, cls)._getInstanceParameterDict()
parameter_dict = super()._getInstanceParameterDict()
parameter_dict['dummy_http_server'] = [
[cls.getManagedResource("content_type_server", ContentTypeHTTPServer).netloc, 1, False],
]
......@@ -644,17 +639,17 @@ class TestContentEncoding(BalancerTestCase):
'application/font-woff2',
'application/x-font-opentype',
'application/wasm',):
resp = requests.get(six.moves.urllib.parse.urljoin(self.default_balancer_url, content_type), verify=False)
resp = requests.get(urllib.parse.urljoin(self.default_balancer_url, content_type), verify=False)
self.assertEqual(resp.headers['Content-Type'], content_type)
self.assertEqual(
resp.headers.get('Content-Encoding'),
'gzip',
'%s uses wrong encoding: %s' % (content_type, resp.headers.get('Content-Encoding')))
'{} uses wrong encoding: {}'.format(content_type, resp.headers.get('Content-Encoding')))
self.assertEqual(resp.text, 'OK')
def test_no_gzip_encoding(self):
# type: () -> None
resp = requests.get(six.moves.urllib.parse.urljoin(self.default_balancer_url, '/image/png'), verify=False)
resp = requests.get(urllib.parse.urljoin(self.default_balancer_url, '/image/png'), verify=False)
self.assertNotIn('Content-Encoding', resp.headers)
self.assertEqual(resp.text, 'OK')
......@@ -777,9 +772,9 @@ class TestFrontendXForwardedFor(BalancerTestCase):
# type: () -> dict
frontend_caucase = cls.getManagedResource('frontend_caucase', CaucaseService)
certificate = cls.getManagedResource('client_certificate', CaucaseCertificate)
certificate.request(u'shared frontend', frontend_caucase)
certificate.request('shared frontend', frontend_caucase)
parameter_dict = super(TestFrontendXForwardedFor, cls)._getInstanceParameterDict()
parameter_dict = super()._getInstanceParameterDict()
# add another "-auth" backend, that will have ssl-authentication enabled
parameter_dict['zope-family-dict']['default-auth'] = ['dummy_http_server']
parameter_dict['backend-path-dict']['default-auth'] = '/'
......@@ -815,7 +810,7 @@ class TestFrontendXForwardedFor(BalancerTestCase):
).json()
self.assertNotEqual(result['Incoming Headers'].get('x-forwarded-for', '').split(', ')[0], '1.2.3.4')
balancer_url = json.loads(self.computer_partition.getConnectionParameterDict()['_'])['default-auth']
with self.assertRaisesRegexp(Exception, "certificate required"):
with self.assertRaisesRegex(Exception, "certificate required"):
requests.get(
balancer_url,
headers={'X-Forwarded-For': '1.2.3.4'},
......@@ -833,8 +828,8 @@ class TestServerTLSProvidedCertificate(BalancerTestCase):
# type: () -> dict
server_caucase = cls.getManagedResource('server_caucase', CaucaseService)
server_certificate = cls.getManagedResource('server_certificate', CaucaseCertificate)
server_certificate.request(six.ensure_text(cls._ipv4_address), server_caucase)
parameter_dict = super(TestServerTLSProvidedCertificate, cls)._getInstanceParameterDict()
server_certificate.request(cls._ipv4_address, server_caucase)
parameter_dict = super()._getInstanceParameterDict()
with open(server_certificate.cert_file) as f:
parameter_dict['ssl']['cert'] = f.read()
with open(server_certificate.key_file) as f:
......@@ -855,13 +850,13 @@ class TestClientTLS(BalancerTestCase):
# type: () -> dict
frontend_caucase1 = cls.getManagedResource('frontend_caucase1', CaucaseService)
certificate1 = cls.getManagedResource('client_certificate1', CaucaseCertificate)
certificate1.request(u'client_certificate1', frontend_caucase1)
certificate1.request('client_certificate1', frontend_caucase1)
frontend_caucase2 = cls.getManagedResource('frontend_caucase2', CaucaseService)
certificate2 = cls.getManagedResource('client_certificate2', CaucaseCertificate)
certificate2.request(u'client_certificate2', frontend_caucase2)
certificate2.request('client_certificate2', frontend_caucase2)
parameter_dict = super(TestClientTLS, cls)._getInstanceParameterDict()
parameter_dict = super()._getInstanceParameterDict()
parameter_dict['ssl-authentication-dict'] = {
'default': True,
}
......@@ -936,11 +931,11 @@ class TestClientTLS(BalancerTestCase):
process = pexpect.spawnu("faketime +1day %s" % caucase_updater)
process.logfile = DebugLogFile()
process.expect(u"Got new CRL.*Next wake-up at.*")
process.expect("Got new CRL.*Next wake-up at.*")
process.terminate()
process.wait()
with self.assertRaisesRegexp(Exception, 'certificate revoked'):
with self.assertRaisesRegex(Exception, 'certificate revoked'):
_make_request()
......@@ -952,10 +947,7 @@ class TestPathBasedRouting(BalancerTestCase):
@classmethod
def _getInstanceParameterDict(cls):
# type: () -> dict
parameter_dict = super(
TestPathBasedRouting,
cls,
)._getInstanceParameterDict()
parameter_dict = super()._getInstanceParameterDict()
parameter_dict['zope-family-dict'][
'second'
] = parameter_dict['zope-family-dict'][
......@@ -985,7 +977,7 @@ class TestPathBasedRouting(BalancerTestCase):
published_dict = json.loads(self.computer_partition.getConnectionParameterDict()['_'])
scheme = 'scheme'
netloc = 'example.com:8080'
prefix = '/VirtualHostBase/' + scheme + '//' + six.moves.urllib.parse.quote(
prefix = '/VirtualHostBase/' + scheme + '//' + urllib.parse.quote(
netloc,
safe='',
)
......@@ -1009,7 +1001,7 @@ class TestPathBasedRouting(BalancerTestCase):
# test will need to be updated accordingly.
self.assertEqual(
requests.get(
six.moves.urllib.parse.urljoin(published_dict[family], prefix + vhr + path),
urllib.parse.urljoin(published_dict[family], prefix + vhr + path),
verify=False,
).json()['Path'],
expected_path,
......
software/erp5/test/test/test_erp5.py
View file @ f5929506
......@@ -25,7 +25,6 @@
#
##############################################################################
from __future__ import absolute_import
import contextlib
import glob
......@@ -42,9 +41,8 @@ import unittest
import psutil
import requests
import six
import six.moves.urllib.parse
import six.moves.xmlrpc_client
import urllib.parse
import xmlrpc.client
import urllib3
from slapos.testing.utils import CrontabMixin
......@@ -53,7 +51,7 @@ from . import ERP5InstanceTestCase, setUpModule
setUpModule # pyflakes
class TestPublishedURLIsReachableMixin(object):
class TestPublishedURLIsReachableMixin:
"""Mixin that checks that default page of ERP5 is reachable.
"""
......@@ -61,7 +59,7 @@ class TestPublishedURLIsReachableMixin(object):
# We access ERP5 trough a "virtual host", which should make
# ERP5 produce URLs using https://virtual-host-name:1234/virtual_host_root
# as base.
virtual_host_url = six.moves.urllib.parse.urljoin(
virtual_host_url = urllib.parse.urljoin(
base_url,
'/VirtualHostBase/https/virtual-host-name:1234/{}/VirtualHostRoot/_vh_virtual_host_root/'
.format(site_id))
......@@ -89,7 +87,7 @@ class TestPublishedURLIsReachableMixin(object):
# login page can be rendered and contain the text "ERP5"
r = session.get(
six.moves.urllib.parse.urljoin(base_url, '{}/login_form'.format(site_id)),
urllib.parse.urljoin(base_url, f'{site_id}/login_form'),
verify=verify,
allow_redirects=False,
)
......@@ -184,7 +182,7 @@ class TestBalancerPorts(ERP5InstanceTestCase):
}
def checkValidHTTPSURL(self, url):
parsed = six.moves.urllib.parse.urlparse(url)
parsed = urllib.parse.urlparse(url)
self.assertEqual(parsed.scheme, 'https')
self.assertTrue(parsed.hostname)
self.assertTrue(parsed.port)
......@@ -194,16 +192,16 @@ class TestBalancerPorts(ERP5InstanceTestCase):
param_dict = self.getRootPartitionConnectionParameterDict()
for family_name in ('family1', 'family2'):
self.checkValidHTTPSURL(
param_dict['family-{family_name}'.format(family_name=family_name)])
param_dict[f'family-{family_name}'])
self.checkValidHTTPSURL(
param_dict['family-{family_name}-v6'.format(family_name=family_name)])
param_dict[f'family-{family_name}-v6'])
def test_published_test_runner_url(self):
# each family's also a list of test test runner URLs, by default 3 per family
param_dict = self.getRootPartitionConnectionParameterDict()
for family_name in ('family1', 'family2'):
family_test_runner_url_list = param_dict[
'{family_name}-test-runner-url-list'.format(family_name=family_name)]
f'{family_name}-test-runner-url-list']
self.assertEqual(3, len(family_test_runner_url_list))
for url in family_test_runner_url_list:
self.checkValidHTTPSURL(url)
......@@ -221,16 +219,16 @@ class TestBalancerPorts(ERP5InstanceTestCase):
# normal access on ipv4 and ipv6 and test runner access on ipv4 only
with self.slap.instance_supervisor_rpc as supervisor:
all_process_info = supervisor.getAllProcessInfo()
process_info, = [p for p in all_process_info if p['name'].startswith('haproxy-')]
process_info, = (p for p in all_process_info if p['name'].startswith('haproxy-'))
haproxy_master_process = psutil.Process(process_info['pid'])
haproxy_worker_process, = haproxy_master_process.children()
self.assertEqual(
sorted([socket.AF_INET] * 4 + [socket.AF_INET6] * 2),
sorted([
sorted(
c.family
for c in haproxy_worker_process.connections()
if c.status == 'LISTEN'
]))
))
class TestSeleniumTestRunner(ERP5InstanceTestCase, TestPublishedURLIsReachableMixin):
......@@ -296,7 +294,7 @@ class TestDisableTestRunner(ERP5InstanceTestCase, TestPublishedURLIsReachableMix
# Haproxy only listen on two ports, there is no haproxy ports allocated for test runner
with self.slap.instance_supervisor_rpc as supervisor:
all_process_info = supervisor.getAllProcessInfo()
process_info, = [p for p in all_process_info if p['name'].startswith('haproxy')]
process_info, = (p for p in all_process_info if p['name'].startswith('haproxy'))
haproxy_master_process = psutil.Process(process_info['pid'])
haproxy_worker_process, = haproxy_master_process.children()
self.assertEqual(
......@@ -366,7 +364,7 @@ class TestZopeNodeParameterOverride(ERP5InstanceTestCase, TestPublishedURLIsReac
zodb["pool-timeout"] = "10m"
storage["storage"] = "root"
storage["server"] = zeo_addr
with open('%s/etc/zope-%s.conf' % (partition, zope)) as f:
with open(f'{partition}/etc/zope-{zope}.conf') as f:
conf = list(map(str.strip, f.readlines()))
i = conf.index("<zodb_db root>") + 1
conf = iter(conf[i:conf.index("</zodb_db>", i)])
......@@ -376,12 +374,12 @@ class TestZopeNodeParameterOverride(ERP5InstanceTestCase, TestPublishedURLIsReac
if line == '</zeoclient>':
break
checkParameter(line, storage)
for k, v in six.iteritems(storage):
for k, v in storage.items():
self.assertIsNone(v, k)
del storage
else:
checkParameter(line, zodb)
for k, v in six.iteritems(zodb):
for k, v in zodb.items():
self.assertIsNone(v, k)
partition = self.getComputerPartitionPath('zope-a')
......@@ -436,19 +434,19 @@ class TestWatchActivities(ERP5InstanceTestCase):
env=dict(os.environ,
PATH=os.pathsep.join([tmpdir, os.environ['PATH']])),
stderr=subprocess.STDOUT,
universal_newlines=True,
text=True,
)
except subprocess.CalledProcessError as e:
self.fail(e.output)
self.assertIn(' dict ', output)
class ZopeSkinsMixin(object):
class ZopeSkinsMixin:
"""Mixins with utility methods to test zope behaviors.
"""
@classmethod
def _setUpClass(cls):
super(ZopeSkinsMixin, cls)._setUpClass()
super()._setUpClass()
param_dict = cls.getRootPartitionConnectionParameterDict()
with cls.getXMLRPCClient() as erp5_xmlrpc_client:
# wait for ERP5 to be ready (TODO: this should probably be a promise)
......@@ -456,8 +454,8 @@ class ZopeSkinsMixin(object):
time.sleep(1)
try:
erp5_xmlrpc_client.getTitle()
except (six.moves.xmlrpc_client.ProtocolError,
six.moves.xmlrpc_client.Fault):
except (xmlrpc.client.ProtocolError,
xmlrpc.client.Fault):
pass
else:
break
......@@ -470,7 +468,7 @@ class ZopeSkinsMixin(object):
path is joined with urllib.parse.urljoin to the URL of the portal.
"""
param_dict = cls.getRootPartitionConnectionParameterDict()
parsed = six.moves.urllib.parse.urlparse(param_dict['family-' + family_name])
parsed = urllib.parse.urlparse(param_dict['family-' + family_name])
base_url = parsed._replace(
netloc='{}:{}@{}:{}'.format(
param_dict['inituser-login'],
......@@ -480,7 +478,7 @@ class ZopeSkinsMixin(object):
),
path=param_dict['site-id'] + '/',
).geturl()
return six.moves.urllib_parse.urljoin(base_url, path)
return urllib.parse.urljoin(base_url, path)
@classmethod
@contextlib.contextmanager
......@@ -489,16 +487,12 @@ class ZopeSkinsMixin(object):
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
erp5_xmlrpc_client = six.moves.xmlrpc_client.ServerProxy(
erp5_xmlrpc_client = xmlrpc.client.ServerProxy(
cls._getAuthenticatedZopeUrl(''),
context=ssl_context,
)
# BBB use as a context manager only on python3
if sys.version_info < (3, ):
with erp5_xmlrpc_client:
yield erp5_xmlrpc_client
else:
with erp5_xmlrpc_client:
yield erp5_xmlrpc_client
@classmethod
def _addPythonScript(cls, script_id, params, body):
......@@ -507,7 +501,7 @@ class ZopeSkinsMixin(object):
try:
custom.manage_addProduct.PythonScripts.manage_addPythonScript(
script_id)
except six.moves.xmlrpc_client.ProtocolError as e:
except xmlrpc.client.ProtocolError as e:
if e.errcode != 302:
raise
getattr(custom, script_id).ZPythonScriptHTML_editAction(
......@@ -542,7 +536,7 @@ class ZopeTestMixin(ZopeSkinsMixin, CrontabMixin):
@classmethod
def _setUpClass(cls):
super(ZopeTestMixin, cls)._setUpClass()
super()._setUpClass()
cls.zope_base_url = cls._getAuthenticatedZopeUrl('')
param_dict = cls.getRootPartitionConnectionParameterDict()
cls.zope_deadlock_debugger_url = cls._getAuthenticatedZopeUrl(
......@@ -565,7 +559,7 @@ class ZopeTestMixin(ZopeSkinsMixin, CrontabMixin):
raise ValueError("Unknown mode: %s" % mode)
''',
)
cls.zope_verify_activity_processing_url = six.moves.urllib_parse.urljoin(
cls.zope_verify_activity_processing_url = urllib.parse.urljoin(
cls.zope_base_url,
'ERP5Site_verifyActivityProcessing',
)
......@@ -578,7 +572,7 @@ class ZopeTestMixin(ZopeSkinsMixin, CrontabMixin):
return log("hello %s" % name)
''',
)
cls.zope_log_message_url = six.moves.urllib_parse.urljoin(
cls.zope_log_message_url = urllib.parse.urljoin(
cls.zope_base_url,
'ERP5Site_logMessage',
)
......@@ -593,18 +587,18 @@ class ZopeTestMixin(ZopeSkinsMixin, CrontabMixin):
return "done"
''',
)
cls.zope_long_request_url = six.moves.urllib_parse.urljoin(
cls.zope_long_request_url = urllib.parse.urljoin(
cls.zope_base_url,
'ERP5Site_executeLongRequest',
)
def setUp(self):
super(ZopeTestMixin, self).setUp()
super().setUp()
# run logrotate a first time so that it create state files
self._executeCrontabAtDate('logrotate', '2000-01-01')
def tearDown(self):
super(ZopeTestMixin, self).tearDown()
super().tearDown()
# reset logrotate status
logrotate_status = os.path.join(
self.getComputerPartitionPath('zope-default'),
......@@ -859,11 +853,11 @@ class TestZopeWSGI(ZopeTestMixin, ERP5InstanceTestCase):
@unittest.expectedFailure
def test_long_request_log_rotation(self):
super(TestZopeWSGI, self).test_long_request_log_rotation(self)
super().test_long_request_log_rotation()
@unittest.expectedFailure
def test_basic_authentication_user_in_access_log(self):
super(TestZopeWSGI, self).test_basic_authentication_user_in_access_log(self)
super().test_basic_authentication_user_in_access_log()
class TestZopePublisherTimeout(ZopeSkinsMixin, ERP5InstanceTestCase):
......@@ -902,7 +896,7 @@ class TestZopePublisherTimeout(ZopeSkinsMixin, ERP5InstanceTestCase):
@classmethod
def _setUpClass(cls):
super(TestZopePublisherTimeout, cls)._setUpClass()
super()._setUpClass()
cls._addPythonScript(
'ERP5Site_doSlowRequest',
'',
......
software/erp5/test/test/test_mariadb.py
View file @ f5929506
##############################################################################
# coding: utf-8
#
# Copyright (c) 2018 Nexedi SA and Contributors. All Rights Reserved.
#
......@@ -26,11 +25,10 @@
#
##############################################################################
from __future__ import absolute_import
import os
import json
import glob
import six.moves.urllib.parse
import urllib.parse
import socket
import sys
import time
......@@ -39,7 +37,7 @@ import datetime
import subprocess
import gzip
from backports import lzma
import lzma
import MySQLdb
from slapos.testing.utils import CrontabMixin
......@@ -84,7 +82,7 @@ class MariaDBTestCase(ERP5InstanceTestCase):
# type: () -> MySQLdb.connections.Connection
connection_parameter_dict = json.loads(
self.computer_partition.getConnectionParameterDict()['_'])
db_url = six.moves.urllib.parse.urlparse(connection_parameter_dict['database-list'][0])
db_url = urllib.parse.urlparse(connection_parameter_dict['database-list'][0])
self.assertEqual('mysql', db_url.scheme)
self.assertTrue(db_url.path.startswith('/'))
......@@ -208,7 +206,7 @@ class TestMariaDB(MariaDBTestCase):
"""
select * from test_utf8_collation where col1 = "a"
""")
self.assertEqual(((u'à',),), cnx.store_result().fetch_row(maxrows=2))
self.assertEqual((('à',),), cnx.store_result().fetch_row(maxrows=2))
class TestMroonga(MariaDBTestCase):
......@@ -232,7 +230,7 @@ class TestMroonga(MariaDBTestCase):
SELECT mroonga_normalize("ABCDあぃうぇ㍑")
""")
# XXX this is returned as bytes by mroonga/mariadb (this might be a bug)
self.assertEqual(((u'abcdあぃうぇリットル'.encode('utf-8'),),),
self.assertEqual((('abcdあぃうぇリットル'.encode(),),),
cnx.store_result().fetch_row(maxrows=2))
if 0:
......@@ -245,7 +243,7 @@ class TestMroonga(MariaDBTestCase):
"""
SELECT mroonga_normalize("aBｃＤあぃウェ㍑", "NormalizerMySQLUnicodeCIExceptKanaCIKanaWithVoicedSoundMark")
""")
self.assertEqual(((u'ABCDあぃうぇ㍑'.encode('utf-8'),),),
self.assertEqual((('ABCDあぃうぇ㍑'.encode(),),),
cnx.store_result().fetch_row(maxrows=2))
def test_mroonga_full_text_normalizer(self):
......@@ -282,7 +280,7 @@ class TestMroonga(MariaDBTestCase):
WHERE MATCH (content) AGAINST ("+ﾌﾞﾗｯｸ" IN BOOLEAN MODE)
""")
self.assertEqual(
((datetime.date(2013, 4, 23), u'ブラックコーヒーを飲んだ。'),),
((datetime.date(2013, 4, 23), 'ブラックコーヒーを飲んだ。'),),
cnx.store_result().fetch_row(maxrows=2),
)
......
software/erp5/test/test/test_wcfs.py
View file @ f5929506
......@@ -16,7 +16,6 @@
# See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options.
from __future__ import absolute_import
import json
import os.path
import unittest
......@@ -76,5 +75,5 @@ def lookupMount(zurl):
# readfile returns content of file @path.
def readfile(path):
with open(path, 'r') as f:
with open(path) as f:
return f.read()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7