291 lines
9.3 KiB
Python
291 lines
9.3 KiB
Python
from decimal import Decimal
|
|
import hashlib
|
|
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.frame_length == len(frame)
|
|
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.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():
|
|
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_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()
|
|
|
|
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
|