test(order): verify rejected XLSX imports preserve scoped database facts

scripts/client-xlsx-address-probe.test.py

@@ -11,6 +11,34 @@ spec.loader.exec_module(probe)
 
 
 class AddressProbeTests(unittest.TestCase):
+    def test_snapshot_requires_complete_tables_and_row_counts(self):
+        rows = [{'table': t, 'count': 0} for t in probe.SNAPSHOT_TABLES]
+        before = probe.snapshot_summary(rows)
+        for bad in [rows[:-1], rows + [rows[0]], rows + [{'table': probe.SNAPSHOT_TABLES[0], 'hash': 'a' * 64}]]:
+            with self.assertRaises(AssertionError):
+                probe.snapshot_summary(bad)
+        changed = copy.deepcopy(rows)
+        changed[0]['count'] = 1
+        changed.append({'table': probe.SNAPSHOT_TABLES[0], 'hash': 'a' * 64})
+        after = probe.snapshot_summary(changed)
+        self.assertNotEqual(before, after)
+        changed[-1]['hash'] = 'b' * 64
+        self.assertNotEqual(after, probe.snapshot_summary(changed))
+
+    def test_snapshot_sql_is_read_only_complete_and_customer_scoped(self):
+        columns = {t: ['id', 'tenant_id', 'deleted', 'version'] for t in probe.SNAPSHOT_TABLES}
+        sql = probe.snapshot_sql(columns, ['123', '456'])
+        self.assertIn('START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY', sql)
+        self.assertEqual(sql.count('tenant_id IN (123,456)'), 14)
+        self.assertNotIn('deleted=0', sql)
+        self.assertIn('CAST(`version` AS CHAR)', sql)
+        for ids in [['123', '123'], ['123 OR 1=1', '456'], ['123']]:
+            with self.assertRaises(AssertionError):
+                probe.snapshot_sql(columns, ids)
+        columns['ord_order'] = ['id', 'unsafe`field']
+        with self.assertRaises(AssertionError):
+            probe.snapshot_sql(columns, ['123', '456'])
+
     def response(self):
         return {'code': '200', 'data': {'mode': 'PARTIAL', 'sourceFileId': 's', 'errorReportFileId': 'e',
                 'result': {'total': 1, 'succeeded': 0, 'failed': 1, 'items': [