Modify test structure to support multiple database (julien-duponchelle#502)

 Co-authored-by: mikaniz <k515jina@gmail.com> Co-authored-by: starcat37 <starcat37@korea.ac.kr> Co-authored-by: mjs <munmun2004@naver.com>

Author: 4 people

.github/workflows/pytest.yml

@@ -1,6 +1,7 @@
 name: PyTest
 on: [push, pull_request]
-
+env:
+ PYTEST_SKIP_OPTION: "not test_no_trailing_rotate_event and not test_end_log_pos and not test_query_event_latin1"
 jobs:
  test:
  strategy:
@@ -29,22 +30,34 @@ jobs:
  docker compose create
  docker compose start
  echo "wait mysql server"
- 
+
  while :
  do
- if mysql -h 127.0.0.1 --user=root --execute "SELECT version();" 2>&1 >/dev/null && mysql -h 127.0.0.1 --port=3307  --user=root --execute "SELECT version();" 2>&1 >/dev/null; then
+ if mysql -h 127.0.0.1 --user=root --execute "SELECT version();" 2>&1 >/dev/null && mysql -h 127.0.0.1 --port=3307 --user=root --execute "SELECT version();" 2>&1 >/dev/null; then
  break
  fi
  sleep 1
  done
- 
+
  echo "run pytest"
 
  - name: Install dependencies
  run: |
  pip install .
  pip install pytest
 
- - name: Run test suite
- run: |
- pytest -k "not test_no_trailing_rotate_event and not test_end_log_pos"
+ - name: Run tests for mysql-5
+ working-directory: pymysqlreplication/tests
+ run: pytest -k "$PYTEST_SKIP_OPTION" --db=mysql-5
+
+ - name: Run tests for mysql-5-ctl
+ working-directory: pymysqlreplication/tests
+ run: pytest -k "$PYTEST_SKIP_OPTION" --db=mysql-5-ctl
+
+ - name: Run tests for mysql-8
+ working-directory: pymysqlreplication/tests
+ run: pytest -k "$PYTEST_SKIP_OPTION" --db=mysql-8
+
+ - name: Run tests for mariadb-10
+ working-directory: pymysqlreplication/tests
+ run: pytest -k "$PYTEST_SKIP_OPTION" -m mariadb --db=mariadb-10

pymysqlreplication/tests/base.py

@@ -2,17 +2,35 @@
 import copy
 from pymysqlreplication import BinLogStreamReader
 import os
+import json
+import pytest
+
 import unittest
 
+
+def get_databases():
+ databases = {}
+ with open(
+ os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
+ ) as f:
+ databases = json.load(f)
+ return databases
+
+
 base = unittest.TestCase
 
 
 class PyMySQLReplicationTestCase(base):
  def ignoredEvents(self):
  return []
 
- def setUp(self, charset="utf8"):
- # default
+ @pytest.fixture(autouse=True)
+ def setUpDatabase(self, get_db):
+ databases = get_databases()
+ # For local testing, set the get_dbms parameter to one of the following values: 'mysql-5', 'mysql-8', mariadb-10'.
+ # This value should correspond to the desired database configuration specified in the 'config.json' file.
+ self.database = databases[get_db]
+ """
  self.database = {
  "host": os.environ.get("MYSQL_5_7") or "localhost",
  "user": "root",
@@ -22,7 +40,10 @@ def setUp(self, charset="utf8"):
  "charset": charset,
  "db": "pymysqlreplication_test",
  }
+ """
 
+ def setUp(self, charset="utf8"):
+ # default
  self.conn_control = None
  db = copy.copy(self.database)
  db["db"] = None
@@ -122,62 +143,3 @@ def bin_log_basename(self):
  bin_log_basename = cursor.fetchone()[0]
  bin_log_basename = bin_log_basename.split("/")[-1]
  return bin_log_basename
-
-
-class PyMySQLReplicationMariaDbTestCase(PyMySQLReplicationTestCase):
- def setUp(self):
- # default
- self.database = {
- "host": os.environ.get("MARIADB_10_6") or "localhost",
- "user": "root",
- "passwd": "",
- "port": int(os.environ.get("MARIADB_10_6_PORT") or 3308),
- "use_unicode": True,
- "charset": "utf8",
- "db": "pymysqlreplication_test",
- }
-
- self.conn_control = None
- db = copy.copy(self.database)
- db["db"] = None
- self.connect_conn_control(db)
- self.execute("DROP DATABASE IF EXISTS pymysqlreplication_test")
- self.execute("CREATE DATABASE pymysqlreplication_test")
- db = copy.copy(self.database)
- self.connect_conn_control(db)
- self.stream = None
- self.resetBinLog()
-
- def bin_log_basename(self):
- cursor = self.execute("SELECT @@log_bin_basename")
- bin_log_basename = cursor.fetchone()[0]
- bin_log_basename = bin_log_basename.split("/")[-1]
- return bin_log_basename
-
-
-class PyMySQLReplicationVersion8TestCase(PyMySQLReplicationTestCase):
- def setUp(self):
- super().setUp()
- # default
- self.database = {
- "host": os.environ.get("MYSQL_8_0") or "localhost",
- "user": "root",
- "passwd": "",
- "port": int(os.environ.get("MYSQL_8_0_PORT") or 3309),
- "use_unicode": True,
- "charset": "utf8",
- "db": "pymysqlreplication_test",
- }
-
- self.conn_control = None
- db = copy.copy(self.database)
- db["db"] = None
- self.connect_conn_control(db)
- self.execute("DROP DATABASE IF EXISTS pymysqlreplication_test")
- self.execute("CREATE DATABASE pymysqlreplication_test")
- db = copy.copy(self.database)
- self.connect_conn_control(db)
- self.stream = None
- self.resetBinLog()
- self.isMySQL80AndMore()
- self.__is_mariaDB = None

pymysqlreplication/tests/config.json

@@ -0,0 +1,38 @@
+{
+ "mysql-5": {
+ "host": "localhost",
+ "user": "root",
+ "passwd": "",
+ "port": 3306,
+ "use_unicode": true,
+ "charset": "utf8",
+ "db": "pymysqlreplication_test"
+ },
+ "mysql-5-ctl": {
+ "host": "localhost",
+ "user": "root",
+ "passwd": "",
+ "port": 3307,
+ "use_unicode": true,
+ "charset": "utf8",
+ "db": "pymysqlreplication_test"
+ },
+ "mariadb-10": {
+ "host": "localhost",
+ "user": "root",
+ "passwd": "",
+ "port": 3308,
+ "use_unicode": true,
+ "charset": "utf8",
+ "db": "pymysqlreplication_test"
+ },
+ "mysql-8": {
+ "host": "localhost",
+ "user": "root",
+ "passwd": "",
+ "port": 3309,
+ "use_unicode": true,
+ "charset": "utf8",
+ "db": "pymysqlreplication_test"
+ }
+}

pymysqlreplication/tests/conftest.py

@@ -0,0 +1,10 @@
+import pytest
+
+
+def pytest_addoption(parser):
+ parser.addoption("--db", action="store", default="mysql-5")
+
+
+@pytest.fixture
+def get_db(request):
+ return request.config.getoption("--db")

pymysqlreplication/tests/test_basic.py

@@ -1,8 +1,5 @@
-import copy
 import io
-import os
 import time
-import pymysql
 import unittest
 
 from pymysqlreplication.tests import base
@@ -13,6 +10,7 @@
 from pymysqlreplication.row_event import *
 from pymysqlreplication.packet import BinLogPacketWrapper
 from pymysql.protocol import MysqlPacket
+import pytest
 
 __all__ = [
  "TestBasicBinLogStreamReader",
@@ -826,42 +824,22 @@ def test_alter_column(self):
 
 
 class TestCTLConnectionSettings(base.PyMySQLReplicationTestCase):
- def setUp(self):
+ def setUp(self, charset="utf8"):
  super().setUp()
- self.stream.close()
- ctl_db = copy.copy(self.database)
- ctl_db["db"] = None
- ctl_db["port"] = int(os.environ.get("MYSQL_5_7_CTL_PORT") or 3307)
- ctl_db["host"] = os.environ.get("MYSQL_5_7_CTL") or "localhost"
- self.ctl_conn_control = pymysql.connect(**ctl_db)
- self.ctl_conn_control.cursor().execute(
- "DROP DATABASE IF EXISTS pymysqlreplication_test"
- )
- self.ctl_conn_control.cursor().execute(
- "CREATE DATABASE pymysqlreplication_test"
- )
- self.ctl_conn_control.close()
- ctl_db["db"] = "pymysqlreplication_test"
- self.ctl_conn_control = pymysql.connect(**ctl_db)
  self.stream = BinLogStreamReader(
  self.database,
- ctl_connection_settings=ctl_db,
  server_id=1024,
  only_events=(WriteRowsEvent,),
  )
 
- def tearDown(self):
- super().tearDown()
- self.ctl_conn_control.close()
-
  def test_separate_ctl_settings_no_error(self):
  self.execute("CREATE TABLE test (id INTEGER(11))")
  self.execute("INSERT INTO test VALUES (1)")
  self.execute("DROP TABLE test")
  self.execute("COMMIT")
- self.ctl_conn_control.cursor().execute("CREATE TABLE test (id INTEGER(11))")
- self.ctl_conn_control.cursor().execute("INSERT INTO test VALUES (1)")
- self.ctl_conn_control.cursor().execute("COMMIT")
+ self.conn_control.cursor().execute("CREATE TABLE test (id INTEGER(11))")
+ self.conn_control.cursor().execute("INSERT INTO test VALUES (1)")
+ self.conn_control.cursor().execute("COMMIT")
  try:
  self.stream.fetchone()
  except Exception as e:
@@ -1322,7 +1300,13 @@ def tearDown(self):
  super(TestStatementConnectionSetting, self).tearDown()
 
 
-class TestMariadbBinlogStreamReader(base.PyMySQLReplicationMariaDbTestCase):
+@pytest.mark.mariadb
+class TestMariadbBinlogStreamReader(base.PyMySQLReplicationTestCase):
+ def setUp(self):
+ super().setUp()
+ if not self.isMariaDB():
+ self.skipTest("Skipping the entire class for MariaDB")
+
  def test_binlog_checkpoint_event(self):
  self.stream.close()
  self.stream = BinLogStreamReader(
@@ -1353,7 +1337,13 @@ def test_binlog_checkpoint_event(self):
  self.assertEqual(event.filename, self.bin_log_basename() + ".000001")
 
 
-class TestMariadbBinlogStreamReader2(base.PyMySQLReplicationMariaDbTestCase):
+@pytest.mark.mariadb
+class TestMariadbBinlogStreamReader2(base.PyMySQLReplicationTestCase):
+ def setUp(self):
+ super().setUp()
+ if not self.isMariaDB():
+ self.skipTest("Skipping the entire class for MariaDB")
+
  def test_annotate_rows_event(self):
  query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
  self.execute(query)
@@ -1498,7 +1488,8 @@ def test_query_event_latin1(self):
  assert event.query == r"CREATE TABLE test_latin1_\xd6\xc6\xdb (a INT)"
 
 
-class TestOptionalMetaData(base.PyMySQLReplicationVersion8TestCase):
+@pytest.mark.mariadb
+class TestOptionalMetaData(base.PyMySQLReplicationTestCase):
  def setUp(self):
  super(TestOptionalMetaData, self).setUp()
  self.stream.close()

pymysqlreplication/tests/test_data_type.py

@@ -942,7 +942,7 @@ def test_varbinary(self):
  self.assertEqual(event.rows[0]["values"]["b"], b"\xff\x01\x00\x00")
 
 
-class TestDataTypeVersion8(base.PyMySQLReplicationVersion8TestCase):
+class TestDataTypeVersion8(base.PyMySQLReplicationTestCase):
  def ignoredEvents(self):
  return [GtidEvent, PreviousGtidsEvent]