PRE-M8-T02: add read-only WarmteLink serial probe

tests/test_p1_probe.py

@@ -1,7 +1,16 @@
 from decimal import Decimal
+import io
 from pathlib import Path
 
-from scripts.p1_probe import IntegrityStatus, TelegramFramer, parse_telegram
+import serial
+
+from scripts.p1_probe import (
+    IntegrityStatus,
+    TelegramFramer,
+    build_parser,
+    parse_telegram,
+    run_probe,
+)
 
 
 FIXTURES = Path(__file__).parent / "fixtures"
@@ -74,3 +83,152 @@ def test_crc_mismatch_is_invalid_when_standard_framing_is_present():
     telegram = parse_telegram(invalid)
 
     assert telegram.integrity is IntegrityStatus.INVALID
+
+
+class FakeSerial:
+    def __init__(self, chunks: list[bytes]) -> None:
+        self.chunks = iter(chunks)
+        self.closed = False
+
+    def read(self, _size: int) -> bytes:
+        return next(self.chunks, b"")
+
+    def close(self) -> None:
+        self.closed = True
+
+
+def _clock(values: list[float]):
+    ticks = iter(values)
+    return lambda: next(ticks, values[-1])
+
+
+def test_cli_uses_measured_7n1_defaults_and_allows_overrides():
+    parser = build_parser()
+
+    defaults = parser.parse_args(["--device", "/dev/serial/by-id/example"])
+    overridden = parser.parse_args(
+        [
+            "--device",
+            "/dev/test",
+            "--baudrate",
+            "9600",
+            "--bytesize",
+            "8",
+            "--parity",
+            "E",
+            "--stopbits",
+            "2",
+            "--duration",
+            "1",
+        ]
+    )
+
+    assert (defaults.baudrate, defaults.bytesize, defaults.parity, defaults.stopbits) == (115200, 7, "N", 1)
+    assert (overridden.baudrate, overridden.bytesize, overridden.parity, overridden.stopbits) == (
+        9600,
+        8,
+        "E",
+        2,
+    )
+    assert "locally measured" in parser.format_help()
+
+
+def test_probe_reads_fake_chunks_writes_exact_raw_bytes_and_closes_device(tmp_path):
+    frame = _fixture("warmtelink_p1_7n1.txt")
+    fake = FakeSerial([frame[:17], frame[17:]])
+    raw_path = tmp_path / "capture.bin"
+    args = build_parser().parse_args(
+        ["--device", "/dev/serial/by-id/fake", "--duration", "3", "--raw-output", str(raw_path)]
+    )
+    created: dict[str, object] = {}
+
+    def serial_factory(**kwargs):
+        created.update(kwargs)
+        return fake
+
+    output = io.StringIO()
+    result = run_probe(
+        args,
+        serial_factory=serial_factory,
+        clock=_clock([0, 0.1, 0.2, 0.3, 4]),
+        output=output,
+    )
+
+    assert result == 0
+    assert fake.closed
+    assert raw_path.read_bytes() == frame
+    assert created == {
+        "port": "/dev/serial/by-id/fake",
+        "baudrate": 115200,
+        "bytesize": 7,
+        "parity": "N",
+        "stopbits": 1,
+        "timeout": 1,
+    }
+    assert "unverifiable" in output.getvalue()
+    assert "0-1:24.2.1" in output.getvalue()
+
+
+def test_show_changes_filters_unchanged_fields_and_reports_cadence():
+    frame = _fixture("warmtelink_p1_7n1.txt")
+    changed = frame.replace(b"(5.900*m3)", b"(5.901*m3)")
+    fake = FakeSerial([frame + changed])
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "2", "--show-changes"])
+    output = io.StringIO()
+
+    result = run_probe(
+        args,
+        serial_factory=lambda **_kwargs: fake,
+        clock=_clock([0, 0.1, 0.2, 1.2, 3]),
+        output=output,
+    )
+
+    assert result == 0
+    assert "cadence=1.0s" in output.getvalue()
+    assert "changed fields: 1" in output.getvalue()
+    assert fake.closed
+
+
+def test_probe_diagnoses_permission_errors_without_suggesting_root():
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"])
+    errors = io.StringIO()
+
+    def serial_factory(**_kwargs):
+        raise serial.SerialException("[Errno 13] Permission denied: '/dev/fake'")
+
+    assert run_probe(args, serial_factory=serial_factory, error_output=errors) == 1
+    assert "dialout" in errors.getvalue()
+    assert "run it as root" in errors.getvalue()
+
+
+def test_probe_diagnoses_busy_or_disconnected_device_and_closes_after_read_error():
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"])
+    errors = io.StringIO()
+
+    def busy_factory(**_kwargs):
+        raise serial.SerialException("[Errno 16] Device or resource busy: '/dev/fake'")
+
+    assert run_probe(args, serial_factory=busy_factory, error_output=errors) == 1
+    assert "Close the program" in errors.getvalue()
+
+    class DisconnectingSerial(FakeSerial):
+        def read(self, _size: int) -> bytes:
+            raise serial.SerialException("[Errno 5] device disconnected")
+
+    fake = DisconnectingSerial([])
+    errors = io.StringIO()
+    assert run_probe(args, serial_factory=lambda **_kwargs: fake, error_output=errors) == 1
+    assert fake.closed
+    assert "Check the cable" in errors.getvalue()
+
+
+def test_probe_closes_device_when_interrupted():
+    class InterruptingSerial(FakeSerial):
+        def read(self, _size: int) -> bytes:
+            raise KeyboardInterrupt
+
+    fake = InterruptingSerial([])
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"])
+
+    assert run_probe(args, serial_factory=lambda **_kwargs: fake) == 0
+    assert fake.closed