thejayman77/trikkeSensors
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

preserve hardened USB smoke evidence

Browse Source
This commit is contained in:
Jay
2026-08-17 17:37:39 -04:00
parent 73e5680fc3
commit 00f52ecf0f
7 changed files with 93 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_trikke_protocol.py
+45 -21
View File
@@ -215,6 +215,15 @@ class ProtocolContractTest(unittest.TestCase):
"timing_anomalies": 8,
"gaps": [],
},
"direct_usb_73e5680.trk": {
"sha256": "fd34bb3bf8f92a64960024f1287e553c03076ff3714fe629ec629bec81ddf821",
"sample_count": 1184,
"first_sequence": 0,
"last_sequence": 1183,
"max_dropped": 0,
"timing_anomalies": 2,
"gaps": [],
},
}
for name, contract in expected.items():
@@ -285,29 +294,44 @@ class ProtocolContractTest(unittest.TestCase):
self.assertEqual(contract["gaps"], gaps)
def test_exact_usb_raw_wire_evidence(self) -> None:
wire = (
ROOT / "tests" / "fixtures" / "direct_usb_3c95f3d.wire"
).read_bytes()
validated = (
ROOT / "tests" / "fixtures" / "direct_usb_3c95f3d.trk"
).read_bytes()
self.assertEqual(
"3bdaeadff7962c6eac48c4ebeda285c8eb359e439d5e1728104add2009122c03",
hashlib.sha256(wire).hexdigest(),
)
expected = {
"direct_usb_3c95f3d": {
"sha256": "3bdaeadff7962c6eac48c4ebeda285c8eb359e439d5e1728104add2009122c03",
"frames": 214,
"skipped": 563,
},
"direct_usb_73e5680": {
"sha256": "82d6d17bbf0729e9bfc53f337ec9adf70bcb5e5898b039685eda5dfa19cf4eea",
"frames": 151,
"skipped": 3589,
},
}
for stem, contract in expected.items():
with self.subTest(fixture=stem):
wire = (
ROOT / "tests" / "fixtures" / f"{stem}.wire"
).read_bytes()
validated = (
ROOT / "tests" / "fixtures" / f"{stem}.trk"
).read_bytes()
self.assertEqual(
contract["sha256"], hashlib.sha256(wire).hexdigest()
)
parser = StreamParser()
frames = []
for offset in range(0, len(wire), 113):
frames.extend(parser.feed(wire[offset : offset + 113]))
parser = StreamParser()
frames = []
for offset in range(0, len(wire), 113):
frames.extend(parser.feed(wire[offset : offset + 113]))
self.assertEqual(214, len(frames))
self.assertEqual(validated, b"".join(frame.raw for frame in frames))
self.assertEqual(0, parser.startup_crc_errors)
self.assertEqual(0, parser.crc_errors)
self.assertEqual(1, parser.header_errors)
self.assertEqual(563, parser.skipped_bytes)
self.assertEqual(0, parser.buffered_bytes)
self.assertEqual(contract["frames"], len(frames))
self.assertEqual(
validated, b"".join(frame.raw for frame in frames)
)
self.assertEqual(0, parser.startup_crc_errors)
self.assertEqual(0, parser.crc_errors)
self.assertEqual(1, parser.header_errors)
self.assertEqual(contract["skipped"], parser.skipped_bytes)
self.assertEqual(0, parser.buffered_bytes)
if __name__ == "__main__":