M8-T07: extract reusable P1 parser

tests/test_p1_probe.py

@@ -1,4 +1,5 @@
 from decimal import Decimal
+import hashlib
 import io
 from pathlib import Path
 
@@ -29,7 +30,7 @@ def test_standard_dsmr_fixture_has_a_valid_frame_and_crc():
 
     telegram = parse_telegram(frame)
     assert telegram.integrity is IntegrityStatus.VALID
-    assert telegram.header == b"/ISk5\\2MT382-1000"
+    assert telegram.frame_length == len(frame)
     assert telegram.timestamp == "240822120000S"
 
 
@@ -73,7 +74,7 @@ def test_missing_header_and_non_hex_footer_are_unverifiable():
     telegram = parse_telegram(b")HEADER\n0-0:1.0.0(240822120000S)\n!zzzz\n")
 
     assert telegram.integrity is IntegrityStatus.UNVERIFIABLE
-    assert telegram.footer == b"zzzz"
+    assert telegram.frame_length == len(b")HEADER\n0-0:1.0.0(240822120000S)\n!zzzz\n")
 
 
 def test_crc_mismatch_is_invalid_when_standard_framing_is_present():
@@ -189,6 +190,61 @@ def test_show_changes_filters_unchanged_fields_and_reports_cadence():
     assert fake.closed
 
 
+def test_probe_stdout_redacts_equipment_identifiers():
+    identifier = "private-equipment-id"
+    frame = (
+        f")HEADER\n0-0:96.1.1({identifier})\n0-1:96.1.0({identifier})\n"
+        "0-1:24.2.1(1.000*m3)\n!nope\n"
+    ).encode()
+    fake = FakeSerial([frame])
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"])
+    output = io.StringIO()
+
+    assert run_probe(args, serial_factory=lambda **_kwargs: fake, clock=_clock([0, 0.1, 2]), output=output) == 0
+    assert identifier not in output.getvalue()
+    assert "<redacted>" in output.getvalue()
+
+
+def test_show_changes_detects_redacted_identifier_changes_without_leaking_them():
+    first_telegram_identifier = "private-telegram-id-one"
+    first_channel_identifier = "private-channel-id-one"
+    second_telegram_identifier = "private-telegram-id-two"
+    second_channel_identifier = "private-channel-id-two"
+
+    def frame(telegram_identifier: str, channel_identifier: str) -> bytes:
+        return (
+            f")HEADER\n0-0:96.1.1({telegram_identifier})\n"
+            f"0-1:96.1.0({channel_identifier})\n"
+            "0-1:24.2.1(1.000*m3)\n!nope\n"
+        ).encode()
+
+    first_frame = frame(first_telegram_identifier, first_channel_identifier)
+    second_frame = frame(second_telegram_identifier, second_channel_identifier)
+    fake = FakeSerial([first_frame + second_frame])
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "2", "--show-changes"])
+    output = io.StringIO()
+
+    assert run_probe(
+        args,
+        serial_factory=lambda **_kwargs: fake,
+        clock=_clock([0, 0.1, 0.2, 1.2, 3]),
+        output=output,
+    ) == 0
+    rendered = output.getvalue()
+    assert "changed fields: 2" in rendered
+    assert "<redacted>" in rendered
+    identifiers = (
+        first_telegram_identifier,
+        first_channel_identifier,
+        second_telegram_identifier,
+        second_channel_identifier,
+    )
+    assert all(identifier not in rendered for identifier in identifiers)
+    assert all(hashlib.sha256(identifier.encode()).hexdigest() not in rendered for identifier in identifiers)
+    assert all(identifier not in repr(parse_telegram(first_frame)) for identifier in identifiers[:2])
+    assert all(identifier not in repr(parse_telegram(second_frame)) for identifier in identifiers[2:])
+
+
 def test_probe_diagnoses_permission_errors_without_suggesting_root():
     args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"])
     errors = io.StringIO()