software/erp5/test: add python3 support

Also fix a few typos and type annotations

software/erp5/test/setup.py

@@ -24,6 +24,7 @@
 # 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'
@@ -52,7 +53,6 @@ setup(name=name,
         'cryptography',
         'pexpect',
         'pyOpenSSL',
-        'typing; python_version<"3"',
       ],
       test_suite='test',
     )

software/erp5/test/test/__init__.py

@@ -25,6 +25,7 @@
 #
 ##############################################################################
 
+from __future__ import absolute_import
 import json
 import os
 

software/erp5/test/test/test_erp5.py

@@ -25,10 +25,11 @@
 #
 ##############################################################################
 
+from __future__ import absolute_import
 import os
 import json
 import glob
-import urlparse
+import six.moves.urllib.parse
 import socket
 import time
 
@@ -37,6 +38,9 @@ import requests
 
 from . import ERP5InstanceTestCase
 from . import setUpModule
+import six
+from six.moves import map
+from six.moves import range
 setUpModule # pyflakes
 
 
@@ -48,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 = six.moves.urllib.parse.urljoin(
         base_url,
         '/VirtualHostBase/https/virtual-host-name:1234/{}/VirtualHostRoot/_vh_virtual_host_root/'
         .format(site_id))
@@ -76,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)),
+        six.moves.urllib.parse.urljoin(base_url, '{}/login_form'.format(site_id)),
         verify=verify,
         allow_redirects=False,
     )
@@ -119,6 +123,7 @@ class TestMedusa(ERP5InstanceTestCase, TestPublishedURLIsReachableMixin):
   def getInstanceParameterDict(cls):
     return {'_': json.dumps({'wsgi': False})}
 
+
 class TestJupyter(ERP5InstanceTestCase, TestPublishedURLIsReachableMixin):
   """Test ERP5 Jupyter notebook
   """
@@ -143,6 +148,7 @@ class TestJupyter(ERP5InstanceTestCase, TestPublishedURLIsReachableMixin):
       [result.status_code, result.is_redirect, result.headers['Location']]
     )
 
+
 class TestBalancerPorts(ERP5InstanceTestCase):
   """Instantiate with two zope families, this should create for each family:
    - a balancer entry point with corresponding haproxy
@@ -169,7 +175,7 @@ class TestBalancerPorts(ERP5InstanceTestCase):
     }
 
   def checkValidHTTPSURL(self, url):
-    parsed = urlparse.urlparse(url)
+    parsed = six.moves.urllib.parse.urlparse(url)
     self.assertEqual(parsed.scheme, 'https')
     self.assertTrue(parsed.hostname)
     self.assertTrue(parsed.port)
@@ -254,7 +260,7 @@ class TestSeleniumTestRunner(ERP5InstanceTestCase, TestPublishedURLIsReachableMi
     with open(config_file.strip()) as f:
       self.assertEqual(
           f.read(),
-          json.dumps(json.loads(self.getInstanceParameterDict()['_'])['test-runner']))
+          json.dumps(json.loads(self.getInstanceParameterDict()['_'])['test-runner'], sort_keys=True))
 
 
 class TestDisableTestRunner(ERP5InstanceTestCase, TestPublishedURLIsReachableMixin):
@@ -270,8 +276,8 @@ class TestDisableTestRunner(ERP5InstanceTestCase, TestPublishedURLIsReachableMix
     """
     # self.computer_partition_root_path is the path of root partition.
     # we want to assert that no scripts exist in any partition.
-    bin_programs = map(os.path.basename,
-      glob.glob(self.computer_partition_root_path + "/../*/bin/*"))
+    bin_programs = list(map(os.path.basename,
+      glob.glob(self.computer_partition_root_path + "/../*/bin/*")))
 
     self.assertTrue(bin_programs) # just to check the glob was correct.
     self.assertNotIn('runUnitTest', bin_programs)
@@ -352,7 +358,7 @@ class TestZopeNodeParameterOverride(ERP5InstanceTestCase, TestPublishedURLIsReac
       storage["storage"] = "root"
       storage["server"] = zeo_addr
       with open('%s/etc/zope-%s.conf' % (partition, zope)) as f:
-        conf = map(str.strip, f.readlines())
+        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:
@@ -361,23 +367,23 @@ class TestZopeNodeParameterOverride(ERP5InstanceTestCase, TestPublishedURLIsReac
             if line == '</zeoclient>':
               break
             checkParameter(line, storage)
-          for k, v in storage.iteritems():
+          for k, v in six.iteritems(storage):
             self.assertIsNone(v, k)
           del storage
         else:
           checkParameter(line, zodb)
-      for k, v in zodb.iteritems():
+      for k, v in six.iteritems(zodb):
         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,
         }, {

software/erp5/test/test/test_mariadb.py

@@ -26,10 +26,12 @@
 #
 ##############################################################################
 
+from __future__ import absolute_import
+from __future__ import unicode_literals
 import os
 import json
 import glob
-import urlparse
+import six.moves.urllib.parse
 import socket
 import sys
 import time
@@ -60,6 +62,7 @@ class MariaDBTestCase(ERP5InstanceTestCase):
 
   @classmethod
   def _getInstanceParameterDict(cls):
+    # type: () -> dict
     return {
         'tcpv4-port': 3306,
         'max-connection-count': 5,
@@ -75,12 +78,14 @@ class MariaDBTestCase(ERP5InstanceTestCase):
 
   @classmethod
   def getInstanceParameterDict(cls):
+    # type: () -> dict
     return {'_': json.dumps(cls._getInstanceParameterDict())}
 
   def getDatabaseConnection(self):
+    # type: () -> MySQLdb.connections.Connection
     connection_parameter_dict = json.loads(
         self.computer_partition.getConnectionParameterDict()['_'])
-    db_url = urlparse.urlparse(connection_parameter_dict['database-list'][0])
+    db_url = six.moves.urllib.parse.urlparse(connection_parameter_dict['database-list'][0])
     self.assertEqual('mysql', db_url.scheme)
 
     self.assertTrue(db_url.path.startswith('/'))
@@ -91,12 +96,14 @@ class MariaDBTestCase(ERP5InstanceTestCase):
         host=db_url.hostname,
         port=db_url.port,
         db=database_name,
+        use_unicode=True,
     )
 
 
 class TestCrontabs(MariaDBTestCase, CrontabMixin):
 
   def test_full_backup(self):
+    # type: () -> None
     self._executeCrontabAtDate('mariadb-backup', '2050-01-01')
     with gzip.open(
         os.path.join(
@@ -106,10 +113,11 @@ 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):
+    # type: () -> None
     # slow query digest needs to run after logrotate, since it operates on the rotated
     # file, so this tests both logrotate and slow query digest.
 
@@ -148,7 +156,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 +178,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
@@ -179,6 +187,7 @@ Expected slowest query : 0.1 and current is: 3
 
 class TestMariaDB(MariaDBTestCase):
   def test_utf8_collation(self):
+    # type: () -> None
     cnx = self.getDatabaseConnection()
     with contextlib.closing(cnx):
       cnx.query(
@@ -204,6 +213,7 @@ class TestMariaDB(MariaDBTestCase):
 
 class TestMroonga(MariaDBTestCase):
   def test_mroonga_plugin_loaded(self):
+    # type: () -> None
     cnx = self.getDatabaseConnection()
     with contextlib.closing(cnx):
       cnx.query("show plugins")
@@ -213,6 +223,7 @@ class TestMroonga(MariaDBTestCase):
           plugins)
 
   def test_mroonga_normalize_udf(self):
+    # type: () -> None
     # example from https://mroonga.org/docs/reference/udf/mroonga_normalize.html#usage
     cnx = self.getDatabaseConnection()
     with contextlib.closing(cnx):
@@ -220,7 +231,7 @@ class TestMroonga(MariaDBTestCase):
           """
           SELECT mroonga_normalize("ABCDあぃうぇ㍑")
           """)
-      self.assertEqual((('abcdあぃうぇリットル',),),
+      self.assertEqual((('abcdあぃうぇリットル'.encode('utf-8'),),),
                        cnx.store_result().fetch_row(maxrows=2))
 
       if 0:
@@ -233,10 +244,11 @@ class TestMroonga(MariaDBTestCase):
             """
             SELECT mroonga_normalize("aBｃＤあぃウェ㍑", "NormalizerMySQLUnicodeCIExceptKanaCIKanaWithVoicedSoundMark")
             """)
-        self.assertEqual((('ABCDあぃうぇ㍑',),),
+        self.assertEqual((('ABCDあぃうぇ㍑'.encode('utf-8'),),),
                          cnx.store_result().fetch_row(maxrows=2))
 
   def test_mroonga_full_text_normalizer(self):
+    # type: () -> None
     # example from https://mroonga.org//docs/tutorial/storage.html#how-to-specify-the-normalizer
     cnx = self.getDatabaseConnection()
     with contextlib.closing(cnx):
@@ -274,6 +286,7 @@ class TestMroonga(MariaDBTestCase):
       )
 
   def test_mroonga_full_text_normalizer_TokenBigramSplitSymbolAlphaDigit(self):
+    # type: () -> None
     # Similar to as ERP5's testI18NSearch with erp5_full_text_mroonga_catalog
     cnx = self.getDatabaseConnection()
     with contextlib.closing(cnx):
@@ -317,11 +330,12 @@ class TestMroonga(MariaDBTestCase):
       self.assertEqual(((1,),), cnx.store_result().fetch_row(maxrows=2))
 
   def test_mroonga_full_text_stem(self):
+    # type: () -> None
     # example from https://mroonga.org//docs/tutorial/storage.html#how-to-specify-the-token-filters
     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/erp5/test/test/test_wcfs.py

@@ -16,6 +16,7 @@
 # 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