home-automation/tests/test_p1_probe.py

from decimal import Decimal
import io
from pathlib import Path

import serial

from scripts.p1_probe import (
    IntegrityStatus,
    TelegramFramer,
    build_parser,
    parse_telegram,
    run_probe,
)


FIXTURES = Path(__file__).parent / "fixtures"


def _fixture(name: str) -> bytes:
    return (FIXTURES / name).read_bytes()


def test_standard_dsmr_fixture_has_a_valid_frame_and_crc():
    frame = _fixture("dsmr_p1_valid.txt")
    framer = TelegramFramer()

    assert framer.feed(frame[:23]) == []
    assert framer.feed(frame[23:]) == [frame]

    telegram = parse_telegram(frame)
    assert telegram.integrity is IntegrityStatus.VALID
    assert telegram.header == b"/ISk5\\2MT382-1000"
    assert telegram.timestamp == "240822120000S"


def test_warmtelink_fixture_is_unverifiable_but_parses_channels_by_obis_code():
    telegram = parse_telegram(_fixture("warmtelink_p1_7n1.txt"))

    assert telegram.integrity is IntegrityStatus.UNVERIFIABLE
    assert "missing standard" in telegram.integrity_reason
    assert [channel.number for channel in telegram.channels] == [1, 2]
    values = {
        channel.number: [(field.value, field.unit) for field in channel.readings]
        for channel in telegram.channels
    }
    assert values[1] == [(Decimal("5.900"), "m3")]
    assert values[2] == [(Decimal("0.017"), "GJ")]


def test_arbitrary_chunk_boundaries_and_field_order_do_not_change_parsing():
    frame = _fixture("warmtelink_p1_7n1.txt")
    framer = TelegramFramer()
    frames = []
    for byte in frame:
        frames.extend(framer.feed(bytes([byte])))

    assert frames == [frame]
    assert parse_telegram(frames[0]).channels == parse_telegram(frame).channels


def test_unknown_fields_are_preserved_verbatim():
    frame = b")HEADER\n9-9:9.9.9(opaque)(still-opaque)\n!nope\n"

    telegram = parse_telegram(frame)

    assert len(telegram.fields) == 1
    assert telegram.fields[0].code == "9-9:9.9.9"
    assert telegram.fields[0].raw_values == ("opaque", "still-opaque")
    assert telegram.fields[0].value is None


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"


def test_crc_mismatch_is_invalid_when_standard_framing_is_present():
    frame = _fixture("dsmr_p1_valid.txt")
    invalid = frame[:-5] + b"0000\n"

    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