TestCase resolved

test case resolved

Test Case resolved

Author: sean-k1

pymysqlreplication/constants/CHARSET.py

@@ -59,7 +59,7 @@ def by_name(self, name, dbms="mysql"):
 ) as f:
     f.readline()  # pass header
     for line in f:
-        lines = line.split(",")
+        lines = line.rstrip("\n").split(",")
         if len(lines) != 5:
             continue
 

pymysqlreplication/row_event.py

@@ -259,14 +259,15 @@ def charset_to_encoding(name):
 
     def __read_string(self, size, column):
         string = self.packet.read_length_coded_pascal_string(size)
+        origin_string = string
         if column.character_set_name is not None:
             encoding = self.charset_to_encoding(column.character_set_name)
             decode_errors = "ignore" if self._ignore_decode_errors else "strict"
             try:
                 string = string.decode(encoding, decode_errors)
             except LookupError:
-                # If python does not support Mysql encoding type ex)swe7 it will not decoding
-                string = string.decode(errors=decode_errors)
+                # python does not support Mysql encoding type ex)swe7 it will not decoding then Show origin string
+                string = origin_string
         return string
 
     def __read_bit(self, column):
@@ -688,13 +689,6 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.table_obj = Table(self.table_id, self.schema, self.table, self.columns)
         table_map[self.table_id] = self.table_obj
         self.optional_metadata = self._get_optional_meta_data()
-
-        # We exclude 'CHAR' and 'INTERVAL' as they map to 'TINY' and 'ENUM' respectively
-        self.reverse_field_type = {
-            v: k
-            for k, v in vars(FIELD_TYPE).items()
-            if isinstance(v, int) and k not in ["CHAR", "INTERVAL"]
-        }
         self._sync_column_info()
 
     def get_table(self):
@@ -810,7 +804,6 @@ def _sync_column_info(self):
         if not self.__optional_meta_data:
             # If optional_meta_data is False Do not sync Event Time Column Schemas
             return
-
         charset_pos = 0
         enum_or_set_pos = 0
         enum_pos = 0
@@ -825,11 +818,9 @@ def _sync_column_info(self):
             if self._is_character_column(column_type, dbms=self.dbms):
                 charset_id = self.optional_metadata.charset_collation_list[charset_pos]
                 charset_pos += 1
-
                 encode_name, collation_name, charset_name = find_charset(
-                    charset_id, dbms=self.dbms
+                    str(charset_id), dbms=self.dbms
                 )
-
                 self.columns[column_idx].collation_name = collation_name
                 self.columns[column_idx].character_set_name = encode_name
 
@@ -840,7 +831,7 @@ def _sync_column_info(self):
                 enum_or_set_pos += 1
 
                 encode_name, collation_name, charset_name = find_charset(
-                    charset_id, dbms=self.dbms
+                    str(charset_id), dbms=self.dbms
                 )
 
                 self.columns[column_idx].collation_name = collation_name
@@ -1049,9 +1040,6 @@ def _is_numeric_column(column_type):
             return True
         return False
 
-    def _get_field_type_key(self, field_type_value):
-        return self.reverse_field_type.get(field_type_value, None)
-
 
 def find_encoding(charset: CHARSET.Charset):
     encode = None

pymysqlreplication/tests/test_basic.py

@@ -605,6 +605,12 @@ def create_binlog_packet_wrapper(pkt):
 
 
 class TestMultipleRowBinLogStreamReader(base.PyMySQLReplicationTestCase):
+    def setUp(self):
+        super(TestMultipleRowBinLogStreamReader, self).setUp()
+        if self.isMySQL8014AndMore():
+            self.execute("SET GLOBAL binlog_row_metadata='FULL';")
+            self.execute("SET GLOBAL binlog_row_image='FULL';")
+
     def ignoredEvents(self):
         return [GtidEvent, PreviousGtidsEvent]
 
@@ -714,45 +720,40 @@ def test_delete_multiple_row_event(self):
             self.assertEqual(event.rows[1]["values"]["id"], 2)
             self.assertEqual(event.rows[1]["values"]["data"], "World")
 
-    # erase temporary
-    # def test_ignore_decode_errors(self):
-    #     problematic_unicode_string = (
-    #         b'[{"text":"\xed\xa0\xbd \xed\xb1\x8d Some string"}]'
-    #     )
-    #     self.stream.close()
-    #     self.execute("CREATE TABLE test (data VARCHAR(50) CHARACTER SET utf8mb4)")
-    #     self.execute_with_args(
-    #         "INSERT INTO test (data) VALUES (%s)", (problematic_unicode_string)
-    #     )
-    #     self.execute("COMMIT")
-    #
-    #     # Initialize with ignore_decode_errors=False
-    #     self.stream = BinLogStreamReader(
-    #         self.database,
-    #         server_id=1024,
-    #         only_events=(WriteRowsEvent,),
-    #         ignore_decode_errors=False,
-    #     )
-    #     event = self.stream.fetchone()
-    #     event = self.stream.fetchone()
-    #     with self.assertRaises(UnicodeError):
-    #         event = self.stream.fetchone()
-    #         if event.table_map[event.table_id].column_name_flag:
-    #             data = event.rows[0]["values"]["data"]
-    #
-    #     # Initialize with ignore_decode_errors=True
-    #     self.stream = BinLogStreamReader(
-    #         self.database,
-    #         server_id=1024,
-    #         only_events=(WriteRowsEvent,),
-    #         ignore_decode_errors=True,
-    #     )
-    #     self.stream.fetchone()
-    #     self.stream.fetchone()
-    #     event = self.stream.fetchone()
-    #     if event.table_map[event.table_id].column_name_flag:
-    #         data = event.rows[0]["values"]["data"]
-    #         self.assertEqual(data, '[{"text":"  Some string"}]')
+    def test_ignore_decode_errors(self):
+        problematic_unicode_string = (
+            b'[{"text":"\xed\xa0\xbd \xed\xb1\x8d Some string"}]'
+        )
+        self.stream.close()
+        self.execute("CREATE TABLE test (data VARCHAR(50) CHARACTER SET utf8mb4)")
+        self.execute_with_args(
+            "INSERT INTO test (data) VALUES (%s)", (problematic_unicode_string)
+        )
+        self.execute("COMMIT")
+
+        # Initialize with ignore_decode_errors=False
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=(WriteRowsEvent,),
+            ignore_decode_errors=False,
+        )
+        with self.assertRaises(UnicodeError):
+            event = self.stream.fetchone()
+            if event.table_map[event.table_id].column_name_flag:
+                data = event.rows[0]["values"]["data"]
+
+        # Initialize with ignore_decode_errors=True
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=(WriteRowsEvent,),
+            ignore_decode_errors=True,
+        )
+        event = self.stream.fetchone()
+        if event.table_map[event.table_id].column_name_flag:
+            data = event.rows[0]["values"]["data"]
+            self.assertEqual(data, '[{"text":"  Some string"}]')
 
     def test_drop_column(self):
         self.stream.close()
@@ -774,15 +775,15 @@ def test_drop_column(self):
         finally:
             self.resetBinLog()
 
-    @unittest.expectedFailure
     def test_alter_column(self):
+        if not self.isMySQL8014AndMore():
+            self.skipTest("Mysql version is under 8.0.14 - pass")
         self.stream.close()
         self.execute(
             "CREATE TABLE test_alter_column (id INTEGER(11), data VARCHAR(50))"
         )
         self.execute("INSERT INTO test_alter_column VALUES (1, 'A value')")
         self.execute("COMMIT")
-        # this is a problem only when column is added in position other than at the end
         self.execute(
             "ALTER TABLE test_alter_column ADD COLUMN another_data VARCHAR(50) AFTER id"
         )
@@ -796,16 +797,11 @@ def test_alter_column(self):
             server_id=1024,
             only_events=(WriteRowsEvent,),
         )
-        event = self.stream.fetchone()  # insert with two values
-        # both of these asserts fail because of issue underlying proble described in issue #118
-        # because it got table schema info after the alter table, it wrongly assumes the second
-        # column of the first insert is 'another_data'
-        # ER: {'id': 1, 'data': 'A value'}
-        # AR: {'id': 1, 'another_data': 'A value'}
-        self.assertIn("data", event.rows[0]["values"])
-        self.assertNot("another_data", event.rows[0]["values"])
+        event = self.stream.fetchone()
+        self.assertEqual(event.rows[0]["values"]["data"], "A value")
+        event = self.stream.fetchone()  # insert with three values
+        self.assertEqual(event.rows[0]["values"]["another_data"], "Another value")
         self.assertEqual(event.rows[0]["values"]["data"], "A value")
-        self.stream.fetchone()  # insert with three values
 
 
 class TestCTLConnectionSettings(base.PyMySQLReplicationTestCase):

pymysqlreplication/tests/test_data_type.py

@@ -5,6 +5,8 @@
 import sys
 import json
 
+from pymysqlreplication import BinLogStreamReader
+
 if sys.version_info < (2, 7):
     import unittest2 as unittest
 else:
@@ -34,6 +36,12 @@ def encode_value(v):
 
 
 class TestDataType(base.PyMySQLReplicationTestCase):
+    def setUp(self):
+        super(TestDataType, self).setUp()
+        if self.isMySQL8014AndMore():
+            self.execute("SET GLOBAL binlog_row_metadata='FULL';")
+            self.execute("SET GLOBAL binlog_row_image='FULL';")
+
     def ignoredEvents(self):
         return [GtidEvent, PreviousGtidsEvent]
 
@@ -104,20 +112,6 @@ def create_and_get_tablemap_event(self, bit):
 
         return event
 
-    def test_varbinary(self):
-        create_query = "CREATE TABLE test(b VARBINARY(4))"
-        insert_query = "INSERT INTO test VALUES(UNHEX('ff010000'))"
-        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"]["b"], b"\xff\x01\x00\x00")
-
-    def test_fixed_length_binary(self):
-        create_query = "CREATE TABLE test(b BINARY(4))"
-        insert_query = "INSERT INTO test VALUES(UNHEX('ff010000'))"
-        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"]["b"], b"\xff\x01\x00\x00")
-
     def test_decimal(self):
         create_query = "CREATE TABLE test (test DECIMAL(2,1))"
         insert_query = "INSERT INTO test VALUES(4.2)"
@@ -539,31 +533,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"], b"Hello")
+            self.assertEqual(event.rows[0]["values"]["test"], "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"], b"Hello")
+            self.assertEqual(event.rows[0]["values"]["test"], "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"], b"Hello")
+            self.assertEqual(event.rows[0]["values"]["test"], "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"], b"Hello")
+            self.assertEqual(event.rows[0]["values"]["test"], "Hello")
             self.assertEqual(event.rows[0]["values"]["test2"], "World")
 
     def test_string(self):
@@ -804,25 +798,6 @@ def test_encoding_utf8(self):
         if event.table_map[event.table_id].column_name_flag:
             self.assertMultiLineEqual(event.rows[0]["values"]["test"], string)
 
-    def test_zerofill(self):
-        create_query = "CREATE TABLE test ( \
-            test TINYINT UNSIGNED ZEROFILL DEFAULT NULL, \
-            test2 SMALLINT UNSIGNED ZEROFILL DEFAULT NULL, \
-            test3 MEDIUMINT UNSIGNED ZEROFILL DEFAULT NULL, \
-            test4 INT UNSIGNED ZEROFILL DEFAULT NULL, \
-            test5 BIGINT UNSIGNED ZEROFILL DEFAULT NULL \
-            )"
-        insert_query = (
-            "INSERT INTO test (test, test2, test3, test4, test5) VALUES(1, 1, 1, 1, 1)"
-        )
-        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"], "001")
-            self.assertEqual(event.rows[0]["values"]["test2"], "00001")
-            self.assertEqual(event.rows[0]["values"]["test3"], "00000001")
-            self.assertEqual(event.rows[0]["values"]["test4"], "0000000001")
-            self.assertEqual(event.rows[0]["values"]["test5"], "00000000000000000001")
-
     def test_partition_id(self):
         if not self.isMySQL80AndMore():
             self.skipTest("Not supported in this version of MySQL")
@@ -942,6 +917,24 @@ def test_mariadb_only_status_vars(self):
 
         self.assertEqual(event.query, create_query)
 
+    def test_varbinary(self):
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=(WriteRowsEvent,),
+            ignore_decode_errors=True,
+        )
+        create_query = "CREATE TABLE test(b VARBINARY(4))"
+        insert_query = "INSERT INTO test VALUES(UNHEX('ff010000'))"
+        self.execute(create_query)
+        self.execute(insert_query)
+        self.execute("COMMIT")
+
+        event = self.stream.fetchone()
+        if event.table_map[event.table_id].column_name_flag:
+            self.assertEqual(event.rows[0]["values"]["b"], b"\xff\x01\x00\x00")
+
 
 if __name__ == "__main__":
     unittest.main()