Align % formatting in Cursor.executemany() with Cursor.execute() (#721)

ports the changes from PyMySQL/PyMySQL#450

also, fix the test case for detecting whether executemany() just called execute() multiple times

CHANGES.txt

@@ -16,6 +16,7 @@ To be included in 1.0.0 (unreleased)
 * Remove deprecated no_delay argument #702
 * Support PyMySQL up to version 1.0.2 #643
 * Bump minimal PyMySQL version to 1.0.0 #713
+* Align % formatting in Cursor.executemany() with Cursor.execute(), literal % now need to be doubled in Cursor.executemany() #714
 
 
 0.0.22 (2021-11-14)

aiomysql/cursors.py

@@ -14,7 +14,7 @@
 # https://github.com/PyMySQL/PyMySQL/blob/master/pymysql/cursors.py#L11-L18
 
 #: Regular expression for :meth:`Cursor.executemany`.
-#: executemany only suports simple bulk insert.
+#: executemany only supports simple bulk insert.
 #: You can use it to load large dataset.
 RE_INSERT_VALUES = re.compile(
     r"\s*((?:INSERT|REPLACE)\s.+\sVALUES?\s+)" +
@@ -274,7 +274,7 @@ async def executemany(self, query, args):
 
         m = RE_INSERT_VALUES.match(query)
         if m:
-            q_prefix = m.group(1)
+            q_prefix = m.group(1) % ()
             q_values = m.group(2).rstrip()
             q_postfix = m.group(3) or ''
             assert q_values[0] == '(' and q_values[-1] == ')'

tests/test_cursor.py

@@ -2,8 +2,6 @@
 
 import pytest
 
-from pymysql.err import OperationalError
-
 from aiomysql import ProgrammingError, Cursor, InterfaceError
 from aiomysql.cursors import RE_INSERT_VALUES
 
@@ -339,11 +337,6 @@ async def test_execute_percentage(connection_creator):
         await cur.execute(q, (3, 4))
 
 
-@pytest.mark.xfail(
-    reason="https://github.com/aio-libs/aiomysql/pull/545",
-    raises=OperationalError,
-    strict=True,
-)
 @pytest.mark.run_loop
 async def test_executemany_percentage(connection_creator):
     # %% in column set
@@ -359,5 +352,5 @@ async def test_executemany_percentage(connection_creator):
 
         assert RE_INSERT_VALUES.match(q) is not None
         await cur.executemany(q, [(3, 4), (5, 6)])
-        assert cur._last_executed.endswith("(3, 4),(5, 6)"), \
+        assert cur._last_executed.endswith(b"(3, 4),(5, 6)"), \
             "executemany with %% not in one query"