ISSUES-14235 add mysql_5_7 and mysql_8_0 empty transaction tests

tests/integration/test_materialize_mysql_database/materialize_with_ddl.py

 import time
+import pymysql.cursors
 
 
 def check_query(clickhouse_node, query, result_set, retry_count=3, interval_seconds=3):
@@ -321,3 +322,38 @@ def alter_rename_table_with_materialize_mysql_database(clickhouse_node, mysql_no
 
     clickhouse_node.query("DROP DATABASE test_database")
     mysql_node.query("DROP DATABASE test_database")
+
+
+def query_event_with_empty_transaction(clickhouse_node, mysql_node, service_name):
+    mysql_node.query("CREATE DATABASE test_database")
+
+    mysql_node.query("RESET MASTER")
+    mysql_node.query("CREATE TABLE test_database.t1(a INT NOT NULL PRIMARY KEY, b VARCHAR(255) DEFAULT 'BEGIN')")
+    mysql_node.query("INSERT INTO test_database.t1(a) VALUES(1)")
+
+    clickhouse_node.query(
+        "CREATE DATABASE test_database ENGINE = MaterializeMySQL('{}:3306', 'test_database', 'root', 'clickhouse')".format(
+            service_name))
+
+    # Reject one empty GTID QUERY event with 'BEGIN' and 'COMMIT'
+    mysql_cursor = mysql_node.cursor(pymysql.cursors.DictCursor)
+    mysql_cursor.execute("SHOW MASTER STATUS")
+    (uuid, seqs) = mysql_cursor.fetchall()[0]["Executed_Gtid_Set"].split(":")
+    (seq_begin, seq_end) = seqs.split("-")
+    assert int(seq_begin) == 1
+    assert int(seq_end) == 3
+    next_gtid = uuid + ":" + str(int(seq_end) + 1)
+    mysql_node.query("SET gtid_next='" + next_gtid + "'")
+    mysql_node.query("BEGIN")
+    mysql_node.query("COMMIT")
+    mysql_node.query("SET gtid_next='AUTOMATIC'")
+
+    # Reject one 'BEGIN' QUERY event and 'COMMIT' XID event.
+    mysql_node.query("/* start */ begin /* end */")
+    mysql_node.query("INSERT INTO test_database.t1(a) VALUES(2)")
+    mysql_node.query("/* start */ commit /* end */")
+
+    check_query(clickhouse_node, "SELECT * FROM test_database.t1 ORDER BY a FORMAT TSV",
+                "1\tBEGIN\n2\tBEGIN\n")
+    clickhouse_node.query("DROP DATABASE test_database")
+    mysql_node.query("DROP DATABASE test_database")

tests/integration/test_materialize_mysql_database/test.py

@@ -120,37 +120,8 @@ def test_materialize_database_ddl_with_mysql_8_0(started_cluster, started_mysql_
     materialize_with_ddl.alter_rename_column_with_materialize_mysql_database(clickhouse_node, started_mysql_8_0, "mysql8_0")
     materialize_with_ddl.alter_modify_column_with_materialize_mysql_database(clickhouse_node, started_mysql_8_0, "mysql8_0")
 
-def test_materialize_database_ddl_with_empty_transaction(started_cluster, started_mysql_5_7):
-    mysql_node = started_mysql_5_7.alloc_connection()
-    mysql_node.query("CREATE DATABASE test_database")
-
-
-    mysql_node.query("RESET MASTER")
-    mysql_node.query("CREATE TABLE test_database.t1(a INT NOT NULL PRIMARY KEY, b VARCHAR(255) DEFAULT 'BEGIN')")
-    mysql_node.query("INSERT INTO test_database.t1(a) VALUES(1)")
-
-    clickhouse_node.query(
-        "CREATE DATABASE test_database ENGINE = MaterializeMySQL('{}:3306', 'test_database', 'root', 'clickhouse')".format(
-            "mysql5_7"))
-
-    # Reject one empty GTID QUERY event with 'BEGIN' and 'COMMIT'
-    mysql_cursor = mysql_node.cursor(pymysql.cursors.DictCursor)
-    mysql_cursor.execute("SHOW MASTER STATUS")
-    (uuid, seqs) = mysql_cursor.fetchall()[0]["Executed_Gtid_Set"].split(":")
-    (seq_begin, seq_end) = seqs.split("-")
-    assert int(seq_begin) == 1
-    assert int(seq_end) == 3
-    next_gtid = uuid + ":" + str(int(seq_end) + 1)
-    mysql_node.query("SET gtid_next='" + next_gtid + "'")
-    mysql_node.query("BEGIN")
-    mysql_node.query("COMMIT")
-    mysql_node.query("SET gtid_next='AUTOMATIC'")
-
-    # Reject one 'BEGIN' QUERY event and 'COMMIT' XID event.
-    mysql_node.query("/* start */ begin /* end */")
-    mysql_node.query("INSERT INTO test_database.t1(a) VALUES(2)")
-    mysql_node.query("/* start */ commit /* end */")
-
-    materialize_with_ddl.check_query(clickhouse_node, "SELECT * FROM test_database.t1 ORDER BY a FORMAT TSV", "1\tBEGIN\n2\tBEGIN\n")
-    clickhouse_node.query("DROP DATABASE test_database")
-    mysql_node.query("DROP DATABASE test_database")
+def test_materialize_database_ddl_with_empty_transaction_5_7(started_cluster, started_mysql_5_7):
+    materialize_with_ddl.query_event_with_empty_transaction(clickhouse_node, started_mysql_5_7.alloc_connection(), "mysql5_7")
+
+def test_materialize_database_ddl_with_empty_transaction_8_0(started_cluster, started_mysql_8_0):
+    materialize_with_ddl.query_event_with_empty_transaction(clickhouse_node, started_mysql_8_0.alloc_connection(), "mysql8_0")