Skip to content

Commit

Permalink
fix : testcase 8 version
Browse files Browse the repository at this point in the history 
fix testcase

fix test case

testcase fix
  • Loading branch information
@sean-k1
sean-k1 committed Sep 18, 2023
1 parent 84c0a32 commit bc1ee8c
Show file tree
Hide file tree
Showing 2 changed files with 38 additions and 25 deletions.
31 changes: 13 additions & 18 deletions pymysqlreplication/tests/test_basic.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -721,6 +721,8 @@ def test_delete_multiple_row_event(self):
self.assertEqual(event.rows[1]["values"]["data"], "World")

def test_ignore_decode_errors(self):
if self.isMySQL80AndMore():
self.skipTest("MYSQL 8 Version Pymysql Data Error Incorrect string value")
problematic_unicode_string = (
b'[{"text":"\xed\xa0\xbd \xed\xb1\x8d Some string"}]'
)
Expand All @@ -740,8 +742,7 @@ def test_ignore_decode_errors(self):
)
with self.assertRaises(UnicodeError):
event = self.stream.fetchone()
if event.table_map[event.table_id].column_name_flag:
data = event.rows[0]["values"]["data"]
data = event.rows[0]["values"]["data"]

# Initialize with ignore_decode_errors=True
self.stream = BinLogStreamReader(
Expand Down Expand Up @@ -1661,6 +1662,11 @@ def test_enum_and_set_column_charset(self):
)

def test_visibility(self):
mysql_version = self.getMySQLVersion()
version = float(mysql_version.rsplit(".", 1)[0])
version_detail = int(mysql_version.rsplit(".", 1)[1])
if not (version >= 8.0 and version_detail >= 23):
self.skipTest("Mysql version 8.0.23 - visibility supprot")
create_query = "CREATE TABLE test_visibility (name VARCHAR(50), secret_key VARCHAR(50) DEFAULT 'qwerty' INVISIBLE);"
insert_query = "INSERT INTO test_visibility VALUES('Audrey');"

Expand Down Expand Up @@ -1697,12 +1703,7 @@ def test_sync_drop_table_map_event_table_schema(self):

event = self.stream.fetchone()
self.assertIsInstance(event, TableMapEvent)
self.assertEqual(
event.table_obj.data["column_schemas"][0]["COLUMN_NAME"], "name"
)
self.assertEqual(
event.table_obj.data["column_schemas"][0]["COLUMN_COMMENT"], ""
)
self.assertEqual(event.table_obj.data["columns"][0].name, "name")
self.assertEqual(len(column_schemas), 0)

def test_sync_column_drop_event_table_schema(self):
Expand Down Expand Up @@ -1730,18 +1731,12 @@ def test_sync_column_drop_event_table_schema(self):
event = self.stream.fetchone()
self.assertIsInstance(event, TableMapEvent)
self.assertEqual(len(column_schemas), 2)
self.assertEqual(len(event.table_obj.data["column_schemas"]), 3)
self.assertEqual(len(event.table_obj.data["columns"]), 3)
self.assertEqual(column_schemas[0][0], "drop_column1")
self.assertEqual(column_schemas[1][0], "drop_column3")
self.assertEqual(
event.table_obj.data["column_schemas"][0]["COLUMN_NAME"], "drop_column1"
)
self.assertEqual(
event.table_obj.data["column_schemas"][1]["COLUMN_NAME"], "drop_column2"
)
self.assertEqual(
event.table_obj.data["column_schemas"][2]["COLUMN_NAME"], "drop_column3"
)
self.assertEqual(event.table_obj.data["columns"][0].name, "drop_column1")
self.assertEqual(event.table_obj.data["columns"][1].name, "drop_column2")
self.assertEqual(event.table_obj.data["columns"][2].name, "drop_column3")

def tearDown(self):
self.execute("SET GLOBAL binlog_row_metadata='MINIMAL';")
Expand Down
32 changes: 25 additions & 7 deletions pymysqlreplication/tests/test_data_type.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -386,6 +386,10 @@ def test_zero_date(self):
self.assertEqual(event.rows[0]["values"]["test2"], None)

def test_zero_month(self):
if not self.isMySQL57():
self.skipTest(
"Not supported in this version of MySQL 8"
) # pymysql.err.OperationalError
self.set_sql_mode()
create_query = "CREATE TABLE test (id INTEGER, test DATE, test2 DATE);"
insert_query = "INSERT INTO test (id, test2) VALUES(1, '2015-00-21')"
Expand All @@ -395,6 +399,10 @@ def test_zero_month(self):
self.assertEqual(event.rows[0]["values"]["test2"], None)

def test_zero_day(self):
if not self.isMySQL57():
self.skipTest(
"Not supported in this version of MySQL 8"
) # pymysql.err.OperationalError
self.set_sql_mode()
create_query = "CREATE TABLE test (id INTEGER, test DATE, test2 DATE);"
insert_query = "INSERT INTO test (id, test2) VALUES(1, '2015-05-00')"
Expand Down Expand Up @@ -463,6 +471,10 @@ def test_datetime(self):
)

def test_zero_datetime(self):
if not self.isMySQL57():
self.skipTest(
"Not supported in this version of MySQL 8"
) # pymysql.err.OperationalError Invalid default value for 'test'
self.set_sql_mode()
create_query = (
"CREATE TABLE test (id INTEGER, test DATETIME NOT NULL DEFAULT 0);"
Expand All @@ -473,6 +485,10 @@ def test_zero_datetime(self):
self.assertEqual(event.rows[0]["values"]["test"], None)

def test_broken_datetime(self):
if not self.isMySQL57():
self.skipTest(
"Not supported in this version of MySQL 8"
) # pymysql.err.OperationalError Incorrect datetime value: '2013-00-00 00:00:00' for column 'test'
self.set_sql_mode()
create_query = "CREATE TABLE test (test DATETIME NOT NULL);"
insert_query = "INSERT INTO test VALUES('2013-00-00 00:00:00')"
Expand All @@ -483,8 +499,10 @@ def test_broken_datetime(self):
def test_year(self):
if self.isMySQL57():
# https://dev.mysql.com/doc/refman/5.7/en/migrating-to-year4.html
self.skipTest("YEAR(2) is unsupported in mysql 5.7")
create_query = "CREATE TABLE test (a YEAR(4), b YEAR(2))"
self.skipTest(
"YEAR(2) is unsupported in mysql 5.7"
) # pymysql.err.OperationalError: (1818, 'Supports only YEAR or YEAR(4) column.')
create_query = "CREATE TABLE test (a YEAR(4), b YEAR)"
insert_query = "INSERT INTO test VALUES(1984, 1984)"
event = self.create_and_insert_value(create_query, insert_query)
if event.table_map[event.table_id].column_name_flag:
Expand Down Expand Up @@ -533,31 +551,31 @@ def test_tiny_blob(self):
insert_query = "INSERT INTO test VALUES('Hello', 'World')"
event = self.create_and_insert_value(create_query, insert_query)
if event.table_map[event.table_id].column_name_flag:
self.assertEqual(event.rows[0]["values"]["test"], "Hello")
self.assertEqual(event.rows[0]["values"]["test"], b"Hello")
self.assertEqual(event.rows[0]["values"]["test2"], "World")

def test_medium_blob(self):
create_query = "CREATE TABLE test (test MEDIUMBLOB, test2 MEDIUMTEXT) CHARACTER SET latin1 COLLATE latin1_bin;"
insert_query = "INSERT INTO test VALUES('Hello', 'World')"
event = self.create_and_insert_value(create_query, insert_query)
if event.table_map[event.table_id].column_name_flag:
self.assertEqual(event.rows[0]["values"]["test"], "Hello")
self.assertEqual(event.rows[0]["values"]["test"], b"Hello")
self.assertEqual(event.rows[0]["values"]["test2"], "World")

def test_long_blob(self):
create_query = "CREATE TABLE test (test LONGBLOB, test2 LONGTEXT) CHARACTER SET latin1 COLLATE latin1_bin;"
insert_query = "INSERT INTO test VALUES('Hello', 'World')"
event = self.create_and_insert_value(create_query, insert_query)
if event.table_map[event.table_id].column_name_flag:
self.assertEqual(event.rows[0]["values"]["test"], "Hello")
self.assertEqual(event.rows[0]["values"]["test"], b"Hello")
self.assertEqual(event.rows[0]["values"]["test2"], "World")

def test_blob(self):
create_query = "CREATE TABLE test (test BLOB, test2 TEXT) CHARACTER SET latin1 COLLATE latin1_bin;"
insert_query = "INSERT INTO test VALUES('Hello', 'World')"
event = self.create_and_insert_value(create_query, insert_query)
if event.table_map[event.table_id].column_name_flag:
self.assertEqual(event.rows[0]["values"]["test"], "Hello")
self.assertEqual(event.rows[0]["values"]["test"], b"Hello")
self.assertEqual(event.rows[0]["values"]["test2"], "World")

def test_string(self):
Expand All @@ -571,7 +589,7 @@ def test_string(self):

def test_geometry(self):
create_query = "CREATE TABLE test (test GEOMETRY);"
insert_query = "INSERT INTO test VALUES(GeomFromText('POINT(1 1)'))"
insert_query = "INSERT INTO test VALUES(ST_GeomFromText('POINT(1 1)'))"
event = self.create_and_insert_value(create_query, insert_query)
if event.table_map[event.table_id].column_name_flag:
self.assertEqual(
Expand Down

0 comments on commit bc1ee8c

Please sign in to comment.