preserve transport outage validation evidence

tests/test_trikke_protocol.py

@@ -1,3 +1,4 @@
+import hashlib
 import shutil
 import subprocess
 import sys
@@ -41,9 +42,13 @@ class ProtocolContractTest(unittest.TestCase):
             ],
             check=True,
         )
-        cls.encoded = subprocess.run(
-            [str(executable)], check=True, capture_output=True
-        ).stdout
+        fixture = subprocess.run([str(executable)], capture_output=True)
+        if fixture.returncode != 0:
+            stderr = fixture.stderr.decode(errors="replace").strip()
+            raise AssertionError(
+                f"protocol fixture exited {fixture.returncode}: {stderr}"
+            )
+        cls.encoded = fixture.stdout
 
     @classmethod
     def tearDownClass(cls) -> None:
@@ -131,6 +136,72 @@ class ProtocolContractTest(unittest.TestCase):
         self.assertEqual(3, len(frames))
         self.assertEqual(36 + 2 * 20 - 5, parser.buffered_bytes)
 
+    def test_hardware_outage_validation_artifacts(self) -> None:
+        expected = {
+            "forced_outage_3s.trk": {
+                "sha256": "01482816cdaa668e4681c33c8baa1df331d733b9bbcbc4f448ece25e88185ad6",
+                "sample_count": 2144,
+                "last_sequence": 2143,
+                "max_dropped": 0,
+                "gaps": [],
+            },
+            "forced_outage_7s.trk": {
+                "sha256": "2ea8a5742944bdebc13bec2ccdbceba75f0bb71e48c856b0f86285878e190cd3",
+                "sample_count": 1840,
+                "last_sequence": 1977,
+                "max_dropped": 138,
+                "gaps": [(511, 650, 1_390_000)],
+            },
+        }
+
+        for name, contract in expected.items():
+            with self.subTest(fixture=name):
+                data = (ROOT / "tests" / "fixtures" / name).read_bytes()
+                self.assertEqual(
+                    contract["sha256"], hashlib.sha256(data).hexdigest()
+                )
+
+                parser = StreamParser()
+                frames = []
+                for offset in range(0, len(data), 257):
+                    frames.extend(parser.feed(data[offset : offset + 257]))
+
+                self.assertEqual(0, parser.startup_crc_errors)
+                self.assertEqual(0, parser.crc_errors)
+                self.assertEqual(0, parser.header_errors)
+                self.assertEqual(0, parser.skipped_bytes)
+                self.assertEqual(0, parser.buffered_bytes)
+                self.assertTrue(frames)
+
+                samples = [sample for frame in frames for sample in frame.samples]
+                self.assertEqual(contract["sample_count"], len(samples))
+                self.assertEqual(0, samples[0].sequence)
+                self.assertEqual(contract["last_sequence"], samples[-1].sequence)
+                self.assertEqual(
+                    contract["max_dropped"],
+                    max(frame.dropped_sample_count for frame in frames),
+                )
+                self.assertEqual(
+                    0, max(frame.loop_overrun_count for frame in frames)
+                )
+                self.assertFalse(
+                    any(
+                        frame.flags & PACKET_FLAG_TIMESTAMP_DELTA_SATURATED
+                        for frame in frames
+                    )
+                )
+
+                gaps = [
+                    (
+                        left.sequence,
+                        right.sequence,
+                        right.timestamp_us - left.timestamp_us,
+                    )
+                    for left, right in zip(samples, samples[1:])
+                    if right.sequence != left.sequence + 1
+                ]
+                self.assertEqual(contract["gaps"], gaps)
+
 
 if __name__ == "__main__":
     unittest.main()