implement partial format description event fields

pymysqlreplication/event.py

@@ -368,10 +368,25 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.mysql_version_str = self.packet.read(50).rstrip(b"\0").decode()
         numbers = self.mysql_version_str.split("-")[0]
         self.mysql_version = tuple(map(int, numbers.split(".")))
+        self.created = struct.unpack("<I", self.packet.read(4))[0]
+        self.common_header_len = struct.unpack("<B", self.packet.read(1))[0]
+        offset = 4 + 2 + 50 + 1 # created + binlog_version + mysql_version_str + common_header_len
+        checksum_algorithm = 1
+        checksum = 4
+        n = event_size - offset - self.common_header_len - checksum_algorithm - checksum
+        self.post_header_len = struct.unpack(f"<{n}B", self.packet.read(n))
+        self.server_version_split = struct.unpack("<3B", self.packet.read(3))
+        self.number_of_event_types = struct.unpack("<B", self.packet.read(1))[0]
 
     def _dump(self):
         print(f"Binlog version: {self.binlog_version}")
-        print(f"MySQL version: {self.mysql_version_str}")
+        print(f"mysql version: {self.mysql_version_str}")
+        print(f"Created: {self.created}")
+        print(f"Common header length: {self.common_header_len}")
+        print(f"Post header length: {self.post_header_len}")
+        print(f"Server version split: {self.server_version_split}")
+        print(f"Number of event types: {self.number_of_event_types}")
+
 
 
 class StopEvent(BinLogEvent):

pymysqlreplication/tests/test_basic.py

@@ -626,6 +626,29 @@ def test_json_update(self):
             ),
             self.assertEqual(event.rows[0]["after_values"]["setting"], {b"btn": True}),
 
+    def test_format_description_event(self):
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            blocking=False,
+            only_events=[FormatDescriptionEvent],
+        )
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, FormatDescriptionEvent)
+        self.assertIsInstance(event.binlog_version, tuple)
+        self.assertIsInstance(event.mysql_version_str, str)
+        self.assertTrue(
+            event.mysql_version_str.startswith("5.") or event.mysql_version_str.startswith("8."))  # Example check
+        self.assertIsInstance(event.common_header_len, int)
+        self.assertIsInstance(event.post_header_len, tuple)
+        self.assertIsInstance(event.mysql_version, tuple)
+        self.assertEqual(len(event.mysql_version), 3)
+        self.assertIsInstance(event.server_version_split, tuple)
+        self.assertEqual(len(event.server_version_split), 3)
+        self.assertIsInstance(event.number_of_event_types, int)
+
 
 class TestMultipleRowBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def setUp(self):
@@ -1447,6 +1470,30 @@ def test_gtid_list_event(self):
         self.assertEqual(event.event_type, 163)
         self.assertEqual(event.gtid_list[0].gtid, "0-1-15")
 
+    def test_format_description_event(self):
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            blocking=False,
+            only_events=[FormatDescriptionEvent],
+            is_mariadb=True,
+        )
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, FormatDescriptionEvent)
+        self.assertIsInstance(event.binlog_version, tuple)
+        self.assertIsInstance(event.mysql_version_str, str)
+        self.assertTrue(
+            event.mysql_version_str.startswith("10."))
+        self.assertIsInstance(event.common_header_len, int)
+        self.assertIsInstance(event.post_header_len, tuple)
+        self.assertIsInstance(event.mysql_version, tuple)
+        self.assertEqual(len(event.mysql_version), 3)
+        self.assertIsInstance(event.server_version_split, tuple)
+        self.assertEqual(len(event.server_version_split), 3)
+        self.assertIsInstance(event.number_of_event_types, int)
+
 
 class TestRowsQueryLogEvents(base.PyMySQLReplicationTestCase):
     def setUp(self):