Skip to content

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

software/slapos-master/test: port the tests to python3

parent b833c90e
Expand all Show whitespace changes
Inline Side-by-side
Showing with 151 additions and 140 deletions
software/slapos-master/test/test/test_erp5.py
View file @ d0bb606d
......@@ -28,11 +28,11 @@
import os
import json
import glob
import urlparse
import urllib.parse
import socket
import time
import re
import BaseHTTPServer
import http.server
import multiprocessing
import subprocess
......@@ -44,7 +44,7 @@ from . import setUpModule
setUpModule # pyflakes
class TestPublishedURLIsReachableMixin(object):
class TestPublishedURLIsReachableMixin:
"""Mixin that checks that default page of ERP5 is reachable.
"""
......@@ -52,7 +52,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 = urlparse.urljoin(
virtual_host_url = urllib.parse.urljoin(
base_url,
'/VirtualHostBase/https/virtual-host-name:1234/{}/VirtualHostRoot/_vh_virtual_host_root/'
.format(site_id))
......@@ -72,7 +72,7 @@ class TestPublishedURLIsReachableMixin(object):
total=60,
backoff_factor=.5,
status_forcelist=(404, 500, 503))))
with session:
r = session.get(virtual_host_url, verify=verify, allow_redirects=False)
self.assertEqual(r.status_code, requests.codes.found)
# access on / are redirected to login form, with virtual host preserved
......@@ -80,7 +80,7 @@ class TestPublishedURLIsReachableMixin(object):
# login page can be rendered and contain the text "ERP5"
r = session.get(
urlparse.urljoin(base_url, '{}/login_form'.format(site_id)),
urllib.parse.urljoin(base_url, f'{site_id}/login_form'),
verify=verify,
allow_redirects=False,
)
......@@ -134,7 +134,7 @@ class TestJupyter(ERP5InstanceTestCase, TestPublishedURLIsReachableMixin):
param_dict = self.getRootPartitionConnectionParameterDict()
self.assertEqual(
'https://[%s]:8888/tree' % self._ipv6_address,
f'https://[{self._ipv6_address}]:8888/tree',
param_dict['jupyter-url']
)
......@@ -172,7 +172,7 @@ class TestApacheBalancerPorts(ERP5InstanceTestCase):
}
def checkValidHTTPSURL(self, url):
parsed = urlparse.urlparse(url)
parsed = urllib.parse.urlparse(url)
self.assertEqual(parsed.scheme, 'https')
self.assertTrue(parsed.hostname)
self.assertTrue(parsed.port)
......@@ -182,16 +182,16 @@ class TestApacheBalancerPorts(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)
......@@ -209,23 +209,23 @@ class TestApacheBalancerPorts(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'] == 'apache']
process_info, = (p for p in all_process_info if p['name'] == 'apache')
apache_process = psutil.Process(process_info['pid'])
self.assertEqual(
sorted([socket.AF_INET] * 4 + [socket.AF_INET6] * 2),
sorted([
sorted(
c.family
for c in apache_process.connections()
if c.status == 'LISTEN'
]))
))
def test_haproxy_listen(self):
# There is one haproxy per family
with self.slap.instance_supervisor_rpc as supervisor:
all_process_info = supervisor.getAllProcessInfo()
process_info, = [
process_info, = (
p for p in all_process_info if p['name'].startswith('haproxy-')
]
)
haproxy_process = psutil.Process(process_info['pid'])
self.assertEqual([socket.AF_INET, socket.AF_INET], [
c.family for c in haproxy_process.connections() if c.status == 'LISTEN'
......@@ -290,8 +290,8 @@ 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:
conf = map(str.strip, f.readlines())
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)])
for line in conf:
......@@ -300,23 +300,23 @@ class TestZopeNodeParameterOverride(ERP5InstanceTestCase, TestPublishedURLIsReac
if line == '</zeoclient>':
break
checkParameter(line, storage)
for k, v in storage.iteritems():
for k, v in storage.items():
self.assertIsNone(v, k)
del storage
else:
checkParameter(line, zodb)
for k, v in zodb.iteritems():
for k, v in zodb.items():
self.assertIsNone(v, k)
partition = self.getComputerPartitionPath('zope-a')
for zope in xrange(3):
for zope in range(3):
checkConf({
"cache-size-bytes": "20MB",
}, {
"cache-size": "50MB",
})
partition = self.getComputerPartitionPath('zope-bb')
for zope in xrange(5):
for zope in range(5):
checkConf({
"cache-size-bytes": "500MB" if zope else 1<<20,
}, {
......@@ -332,19 +332,20 @@ def popenCommunicate(command_list, input_=None, **kwargs):
popen.kill()
if popen.returncode != 0:
raise ValueError(
'Issue during calling %r, result was:\n%s' % (command_list, result))
f'Issue during calling {command_list!r}, result was:\n{result}')
return result
class TestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
class TestHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
response = {
response = json.dumps(
{
'Path': self.path,
'Incoming Headers': self.headers.dict
}
response = json.dumps(response, indent=2)
'Incoming Headers': {k.lower(): v for k, v in self.headers.items()},
},
indent=2,
).encode('utf-8')
self.end_headers()
self.wfile.write(response)
......@@ -352,7 +353,7 @@ class TestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
class TestDeploymentScriptInstantiation(ERP5InstanceTestCase):
"""This check deployment script like instantiation
Low level assertions are done here in roder to assure that
Low level assertions are done here in order to assure that
https://lab.nexedi.com/nexedi/slapos.package/blob/master/playbook/
slapos-master-standalone.yml
works correctly
......@@ -426,8 +427,8 @@ class TestDeploymentScriptInstantiation(ERP5InstanceTestCase):
1,
len(backend_apache_configuration_list)
)
backend_apache_configuration = open(
backend_apache_configuration_list[0]).read()
with open(backend_apache_configuration_list[0]) as f:
backend_apache_configuration = f.read()
self.assertIn(
'SSLVerifyClient require',
backend_apache_configuration
......@@ -452,7 +453,7 @@ class TestDeploymentScriptInstantiation(ERP5InstanceTestCase):
common_name = 'TEST-SSL-AUTH'
popenCommunicate([
'openssl', 'req', '-utf8', '-nodes', '-config', openssl_config, '-new',
'-keyout', key, '-out', csr, '-days', '3650'], '%s\n' % (common_name,),
'-keyout', key, '-out', csr, '-days', '3650'], f'{common_name}\n'.encode(),
stdin=subprocess.PIPE)
popenCommunicate([
'openssl', 'ca', '-utf8', '-days', '3650', '-batch', '-config',
......@@ -464,11 +465,14 @@ class TestDeploymentScriptInstantiation(ERP5InstanceTestCase):
ip, port = re.search(
r'.*http:\/\/(.*):(\d*)\/.*', portal_slap_line).groups()
port = int(port)
server = BaseHTTPServer.HTTPServer((ip, port), TestHandler)
server = http.server.HTTPServer((ip, port), TestHandler)
server_process = multiprocessing.Process(
target=server.serve_forever, name='HTTPServer')
server_process.start()
try:
self.addCleanup(server_process.terminate)
self.addCleanup(server_process.join, 10)
server.socket.close()
# assert that accessing the service endpoint results with certificate
# authentication and proper information extraction
result_json = requests.get(
......@@ -482,6 +486,3 @@ class TestDeploymentScriptInstantiation(ERP5InstanceTestCase):
'/erp5/portal_slap/',
result_json['Path']
)
finally:
server_process.join(10)
server_process.terminate()
software/slapos-master/test/test/test_mariadb.py
View file @ d0bb606d
##############################################################################
# coding: utf-8
#
# Copyright (c) 2018 Nexedi SA and Contributors. All Rights Reserved.
#
......@@ -29,7 +28,7 @@
import os
import json
import glob
import urlparse
import urllib.parse
import socket
import sys
import time
......@@ -38,7 +37,7 @@ import datetime
import subprocess
import gzip
from backports import lzma
import lzma
import MySQLdb
from slapos.testing.utils import CrontabMixin
......@@ -80,7 +79,7 @@ class MariaDBTestCase(ERP5InstanceTestCase):
def getDatabaseConnection(self):
connection_parameter_dict = json.loads(
self.computer_partition.getConnectionParameterDict()['_'])
db_url = urlparse.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('/'))
......@@ -91,6 +90,8 @@ class MariaDBTestCase(ERP5InstanceTestCase):
host=db_url.hostname,
port=db_url.port,
db=database_name,
use_unicode=True,
charset='utf8mb4'
)
......@@ -106,7 +107,7 @@ class TestCrontabs(MariaDBTestCase, CrontabMixin):
'mariadb-full',
'20500101000000.sql.gz',
),
'r') as dump:
'rt') as dump:
self.assertIn('CREATE TABLE', dump.read())
def test_logrotate_and_slow_query_digest(self):
......@@ -148,7 +149,7 @@ class TestCrontabs(MariaDBTestCase, CrontabMixin):
'slowquery_digest',
'slowquery_digest.txt-2050-01-01.xz',
)
with lzma.open(slow_query_report, 'r') as f:
with lzma.open(slow_query_report, 'rt') as f:
# this is the hash for our "select sleep(n)" slow query
self.assertIn("ID 0xF9A57DD5A41825CA", f.read())
......@@ -170,7 +171,7 @@ class TestCrontabs(MariaDBTestCase, CrontabMixin):
subprocess.check_output('faketime 2050-01-01 %s' % check_slow_query_promise_plugin['command'], shell=True)
self.assertEqual(
error_context.exception.output,
"""\
b"""\
Threshold is lower than expected:
Expected total queries : 1.0 and current is: 2
Expected slowest query : 0.1 and current is: 3
......@@ -220,7 +221,7 @@ class TestMroonga(MariaDBTestCase):
"""
SELECT mroonga_normalize("ABCDあぃうぇ㍑")
""")
self.assertEqual((('abcdあぃうぇリットル',),),
self.assertEqual((('abcdあぃうぇリットル'.encode(),),),
cnx.store_result().fetch_row(maxrows=2))
if 0:
......@@ -233,7 +234,7 @@ class TestMroonga(MariaDBTestCase):
"""
SELECT mroonga_normalize("aBｃＤあぃウェ㍑", "NormalizerMySQLUnicodeCIExceptKanaCIKanaWithVoicedSoundMark")
""")
self.assertEqual((('ABCDあぃうぇ㍑',),),
self.assertEqual((('ABCDあぃうぇ㍑'.encode(),),),
cnx.store_result().fetch_row(maxrows=2))
def test_mroonga_full_text_normalizer(self):
......@@ -321,7 +322,7 @@ class TestMroonga(MariaDBTestCase):
cnx = self.getDatabaseConnection()
with contextlib.closing(cnx):
cnx.query("SELECT mroonga_command('register token_filters/stem')")
self.assertEqual((('true',),), cnx.store_result().fetch_row(maxrows=2))
self.assertEqual(((b'true',),), cnx.store_result().fetch_row(maxrows=2))
cnx.query(
"""
CREATE TABLE memos (
......
software/slapos-master/test/test/test_wcfs.py
View file @ d0bb606d
......@@ -75,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