test: use explicit charset as utf8 is an alias for utf8mb3 form MariaDB 10.6.

product/ZMySQLDA/connectionAdd.dtml

@@ -56,7 +56,7 @@
     <dd>
       The connection string used for Z MySQL Database Connection is of the form:
       <br />
-      <code>[%ssl_name] [*lock] [+/-][database][@host[:port]] [user [password [unix_socket]]]</code>
+      <code>[%ssl_name] [*lock] [!isolation-level] [+/-][database][@host[:port]] [user [password [unix_socket]]]</code>
       <br />
       or typically:
       <br />
@@ -103,6 +103,12 @@
       you'll get an error in the logs, and inconsistent data. In this
       respect, it's equivalent to transactions turned off.
     </dd>
+    <dd>
+      !<em>isolation_level</em> at the begining of the connection string
+      will set the transaction isolation level in each transaction. The
+      value should be one of REPEATABLE-READ, READ-COMMITTED,
+      READ-UNCOMMITTED or SERIALIZABLE.
+    </dd>
     <dd>
       Transactions are highly recommended. Using a named lock in
       conjunctions with transactions is probably pointless.

product/ZMySQLDA/db.py

@@ -242,7 +242,7 @@ class DB(TM):
         self._use_TM = transactional or self._mysql_lock
 
     def _parse_connection_string(self):
-        self._mysql_lock = self._try_transactions = None
+        self._mysql_lock = self._try_transactions = self._isolation_level = None
         self._kw_args = kwargs = {'conv': self.conv}
         items = self._connection.split()
         if not items:
@@ -260,6 +260,8 @@ class DB(TM):
             del items[0]
         if items[0][0] == "*":
             self._mysql_lock = items.pop(0)[1:]
+        if items[0][0] == "!":
+            self._isolation_level = items.pop(0)[1:]
         db = items.pop(0)
         if '@' in db:
             db, host = db.split('@', 1)
@@ -489,6 +491,8 @@ class DB(TM):
         try:
             self._transaction_begun = True
             if self._transactions:
+                if self._isolation_level:
+                    self._query("SET TRANSACTION ISOLATION LEVEL %s" % self._isolation_level.replace('-', ' '))
                 self._query("BEGIN", allow_reconnect=True)
             if self._mysql_lock:
                 self._query("SELECT GET_LOCK('%s',0)" % self._mysql_lock, allow_reconnect=not self._transactions)

product/ZMySQLDA/tests/testTableStructureMigration.py

@@ -66,13 +66,13 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
             """\
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL,
         `b` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""))
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""))
     self.query("SELECT a, b FROM X")
 
   def test_remove_column(self):
@@ -82,12 +82,12 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL,
         `b` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
       CREATE TABLE `X` (
         `b` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""))
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""))
     self.query("SELECT b FROM X")
     with self.assertRaisesRegex(OperationalError,
                                  "Unknown column 'a' in 'field list'"):
@@ -99,12 +99,12 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
             """\
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
       CREATE TABLE `X` (
         `b` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""))
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""))
     self.query("SELECT b FROM X")
     with self.assertRaisesRegex(OperationalError,
                                  "Unknown column 'a' in 'field list'"):
@@ -116,12 +116,12 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
             """\
       CREATE TABLE `X` (
         `a` varchar(10) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""))
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""))
     # insterting '1' will be casted as int
     self.query("INSERT INTO X VALUES ('1')")
     self.assertEqual((1,), self.query("SELECT a FROM X")[1][0])
@@ -132,12 +132,12 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
             """\
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
       CREATE TABLE `X` (
         `a` int(11) DEFAULT 123
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""))
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""))
     self.query("INSERT INTO X VALUES ()")
     self.assertEqual((123,), self.query("SELECT a FROM X")[1][0])
 
@@ -147,12 +147,12 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
             """\
       CREATE TABLE `X` (
         `a` int(11) NOT NULL COMMENT 'old comment'
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
       CREATE TABLE `X` (
         `a` int(11) NOT NULL COMMENT 'new comment'
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""))
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""))
     self.assertEqual(
         ('a', 'new comment'),
         self.query(
@@ -170,13 +170,13 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
             """\
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL,
         KEY `idx_a` (`a`)
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""))
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""))
     self.query("SELECT * FROM X USE INDEX (`idx_a`)")
 
   def test_remove_index(self):
@@ -186,12 +186,12 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL,
         KEY `idx_a` (`a`)
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
       CREATE TABLE `X` (
         `a` int(11) DEFAULT NULL
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""))
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""))
     with self.assertRaisesRegex(OperationalError,
                                  "Key 'idx_a' doesn't exist in table 'X'"):
       self.query("SELECT * FROM X USE INDEX (`idx_a`)")
@@ -204,14 +204,14 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
         `drop` int(11) DEFAULT NULL,
         `alter` int(11) DEFAULT NULL,
         KEY `CASE` (`drop`)
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
       CREATE TABLE `table` (
         `and` int(11) DEFAULT NULL,
         `alter` varchar(255) CHARACTER SET cp1250 COLLATE cp1250_croatian_ci DEFAULT 'BETWEEN',
         KEY `use` (`alter`)
-      ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         table_name='table')
     self.query(
         "SELECT `alter`, `and` FROM `table` USE INDEX (`use`)")
@@ -222,12 +222,12 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
             """\
             CREATE TABLE `X` (
               `a` int(11) DEFAULT NULL
-            ) ENGINE=MyISAM DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""),
+            ) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""),
         dedent(
             """\
             CREATE TABLE `X` (
               `a` int(11) DEFAULT NULL
-            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci"""))
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"""))
     self.assertEqual(
         ('X', 'InnoDB'),
         self.query(
@@ -245,12 +245,12 @@ class TestTableStructureMigrationTestCase(ERP5TypeTestCase):
             """\
             CREATE TABLE `X` (
               `a` int(11) DEFAULT NULL
-            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='old comment'"""),
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='old comment'"""),
         dedent(
             """\
             CREATE TABLE `X` (
               `a` int(11) DEFAULT NULL
-            ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='new comment'"""))
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='new comment'"""))
     self.assertEqual(
         ('X', 'new comment'),
         self.query(