Use wal2json non-row messages to detect that wal has moved on

Author: louis-pie

singer-connectors/tap-postgres/tests/unit/test_logical_replication.py

@@ -176,12 +176,38 @@ def test_consume_with_message_stream_in_payload_is_not_selected_expect_same_stat
 
         self.assertDictEqual({}, output)
 
-    def test_consume_with_non_row_action_returns_state_unchanged(self):
-        """B/C/T/M actions are silently skipped to allow WAL slot LSN to advance"""
+    def test_consume_with_truncate_action_returns_state_unchanged(self):
+        """Truncate (T) actions with schema/table keys are silently skipped"""
         output = logical_replication.consume_message(
             [{'tap_stream_id': 'myschema-mytable'}],
             {},
-            self.WalMessage(payload='{"action":"truncate", "schema": "myschema", "table": "mytable"}',
+            self.WalMessage(payload='{"action":"T", "schema": "myschema", "table": "mytable"}',
+                            data_start='some lsn'),
+            None,
+            {}
+        )
+
+        self.assertDictEqual({}, output)
+
+    def test_consume_with_begin_action_without_schema_table_returns_state_unchanged(self):
+        """Begin (B) messages from wal2json don't include schema/table keys — must not KeyError"""
+        output = logical_replication.consume_message(
+            [{'tap_stream_id': 'myschema-mytable'}],
+            {},
+            self.WalMessage(payload='{"action":"B","xid":12345,"timestamp":"2026-04-06 12:00:00.000000+00"}',
+                            data_start='some lsn'),
+            None,
+            {}
+        )
+
+        self.assertDictEqual({}, output)
+
+    def test_consume_with_commit_action_without_schema_table_returns_state_unchanged(self):
+        """Commit (C) messages from wal2json don't include schema/table keys — must not KeyError"""
+        output = logical_replication.consume_message(
+            [{'tap_stream_id': 'myschema-mytable'}],
+            {},
+            self.WalMessage(payload='{"action":"C","xid":12345,"timestamp":"2026-04-06 12:00:00.000000+00"}',
                             data_start='some lsn'),
             None,
             {}