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

software/slapos-master/test/test/test_erp5.py

@@ -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))
@@ -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,
     )
@@ -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,11 +332,11 @@ 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)
 
@@ -426,8 +426,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 +452,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,24 +464,24 @@ 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:
-      # assert that accessing the service endpoint results with certificate
-      # authentication and proper information extraction
-      result_json = requests.get(
-        self.getRootPartitionConnectionParameterDict()['family-service'],
-        verify=False, cert=(cert, key)).json()
-      self.assertEqual(
-        common_name,
-        result_json['Incoming Headers']['remote-user']
-      )
-      self.assertEqual(
-        '/erp5/portal_slap/',
-        result_json['Path']
-      )
-    finally:
-      server_process.join(10)
-      server_process.terminate()
+    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(
+      self.getRootPartitionConnectionParameterDict()['family-service'],
+      verify=False, cert=(cert, key)).json()
+    self.assertEqual(
+      common_name,
+      result_json['Incoming Headers']['remote-user']
+    )
+    self.assertEqual(
+      '/erp5/portal_slap/',
+      result_json['Path']
+    )

software/slapos-master/test/test/test_mariadb.py

 ##############################################################################
-# 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

@@ -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()