93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from decimal import Decimal
|
||
|
|
import hashlib
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.integrations.p1 import IntegrityStatus, P1ParseError, TelegramFramer, parse_telegram
|
||
|
|
|
||
|
|
|
||
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
||
|
|
|
||
|
|
|
||
|
|
def _fixture(name: str) -> bytes:
|
||
|
|
return (FIXTURES / name).read_bytes()
|
||
|
|
|
||
|
|
|
||
|
|
def test_parser_preserves_fixture_quality_and_uses_only_fingerprints_for_identifiers():
|
||
|
|
frame = _fixture("warmtelink_p1_7n1.txt")
|
||
|
|
telegram = parse_telegram(frame)
|
||
|
|
|
||
|
|
assert telegram.integrity is IntegrityStatus.UNVERIFIABLE
|
||
|
|
assert telegram.frame_length == len(frame)
|
||
|
|
assert [channel.number for channel in telegram.channels] == [1, 2]
|
||
|
|
assert telegram.channels[0].readings[0].value == Decimal("5.900")
|
||
|
|
assert telegram.channels[1].readings[0].unit == "GJ"
|
||
|
|
assert all(field.raw_values == ("<redacted>",) for field in telegram.fields if "96.1" in field.code)
|
||
|
|
assert repr(telegram).find("<redacted>") >= 0
|
||
|
|
assert "KAM" not in repr(telegram)
|
||
|
|
|
||
|
|
|
||
|
|
def test_parser_hashes_identifiers_without_returning_them():
|
||
|
|
telegram_identifier = "secret-telegram-equipment-42"
|
||
|
|
channel_identifier = "secret-channel-equipment-42"
|
||
|
|
telegram = parse_telegram(
|
||
|
|
f")HEADER\n0-0:96.1.1({telegram_identifier})\n"
|
||
|
|
f"0-1:96.1.0({channel_identifier})\n!nope\n".encode()
|
||
|
|
)
|
||
|
|
|
||
|
|
telegram_fingerprint = hashlib.sha256(telegram_identifier.encode()).hexdigest()
|
||
|
|
channel_fingerprint = hashlib.sha256(channel_identifier.encode()).hexdigest()
|
||
|
|
assert telegram.equipment_fingerprint == telegram_fingerprint
|
||
|
|
assert telegram.channels[0].equipment_fingerprint == channel_fingerprint
|
||
|
|
rendered = repr(telegram)
|
||
|
|
assert telegram_identifier not in rendered
|
||
|
|
assert channel_identifier not in rendered
|
||
|
|
assert telegram_fingerprint not in rendered
|
||
|
|
assert channel_fingerprint not in rendered
|
||
|
|
|
||
|
|
|
||
|
|
def test_framer_handles_truncation_concatenation_and_variable_lengths_without_channel_mixup():
|
||
|
|
first = _fixture("warmtelink_p1_7n1.txt")
|
||
|
|
second = first.replace(b"(5.900*m3)", b"(12345.678*m3)")
|
||
|
|
framer = TelegramFramer()
|
||
|
|
|
||
|
|
assert framer.feed(first[:-4]) == []
|
||
|
|
frames = framer.feed(first[-4:] + second)
|
||
|
|
|
||
|
|
assert [parse_telegram(frame).channels[0].readings[0].value for frame in frames] == [
|
||
|
|
Decimal("5.900"),
|
||
|
|
Decimal("12345.678"),
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def test_parser_tolerates_bad_text_without_exposing_it_and_rejects_missing_footer():
|
||
|
|
telegram = parse_telegram(b")HEADER\n0-1:24.2.1(1.000*m3)\n\xff\n!not-hex\n")
|
||
|
|
assert telegram.integrity is IntegrityStatus.UNVERIFIABLE
|
||
|
|
assert telegram.channels[0].readings[0].value == Decimal("1.000")
|
||
|
|
|
||
|
|
with pytest.raises(P1ParseError, match="footer marker") as exc_info:
|
||
|
|
parse_telegram(b"secret raw telegram")
|
||
|
|
assert "secret" not in str(exc_info.value)
|
||
|
|
|
||
|
|
|
||
|
|
def test_parser_error_does_not_expose_equipment_identifiers():
|
||
|
|
first_identifier = "private-equipment-id-one"
|
||
|
|
second_identifier = "private-equipment-id-two"
|
||
|
|
|
||
|
|
with pytest.raises(P1ParseError) as exc_info:
|
||
|
|
parse_telegram(
|
||
|
|
f")HEADER\n0-0:96.1.1({first_identifier})\n"
|
||
|
|
f"0-1:96.1.0({second_identifier})\n".encode()
|
||
|
|
)
|
||
|
|
|
||
|
|
assert first_identifier not in str(exc_info.value)
|
||
|
|
assert second_identifier not in str(exc_info.value)
|
||
|
|
|
||
|
|
|
||
|
|
def test_parser_keeps_standard_invalid_crc_invalid():
|
||
|
|
frame = _fixture("dsmr_p1_valid.txt")
|
||
|
|
assert parse_telegram(frame[:-5] + b"0000\n").integrity is IntegrityStatus.INVALID
|