Add incremental REG3 frame extractor

tests/test_frame_extractor.py

@@ -0,0 +1,167 @@
+"""Tests de l'extracteur incrémental REG3, indépendants de Home Assistant."""
+
+from pathlib import Path
+
+from custom_components.arkteos.frame_extractor import FrameExtractor
+
+
+FIXTURES_PATH = Path(__file__).with_name("fixtures")
+
+
+def _load_frames() -> dict[str, bytes]:
+    return {
+        "metadata": (FIXTURES_PATH / "metadata_95.bin").read_bytes(),
+        "frigo": (FIXTURES_PATH / "frigo_163.bin").read_bytes(),
+        "regulation": (FIXTURES_PATH / "regulation_227.bin").read_bytes(),
+    }
+
+
+def test_no_bytes() -> None:
+    extractor = FrameExtractor()
+    assert extractor.feed(b"") == []
+    assert extractor.buffered_bytes == b""
+
+
+def test_complete_metadata_frame() -> None:
+    metadata = _load_frames()["metadata"]
+    assert FrameExtractor().feed(metadata) == [metadata]
+
+
+def test_complete_frigo_frame() -> None:
+    frigo = _load_frames()["frigo"]
+    assert FrameExtractor().feed(frigo) == [frigo]
+
+
+def test_complete_regulation_frame() -> None:
+    regulation = _load_frames()["regulation"]
+    assert FrameExtractor().feed(regulation) == [regulation]
+
+
+def test_frame_split_in_two_calls() -> None:
+    frigo = _load_frames()["frigo"]
+    extractor = FrameExtractor()
+    assert extractor.feed(frigo[:80]) == []
+    assert extractor.feed(frigo[80:]) == [frigo]
+
+
+def test_frame_provided_byte_by_byte() -> None:
+    regulation = _load_frames()["regulation"]
+    extractor = FrameExtractor()
+    extracted = [frame for byte in regulation for frame in extractor.feed(bytes([byte]))]
+    assert extracted == [regulation]
+
+
+def test_multiple_concatenated_frames() -> None:
+    frames = _load_frames()
+    sequence = frames["metadata"] + frames["frigo"] + frames["regulation"]
+    assert FrameExtractor().feed(sequence) == [
+        frames["metadata"],
+        frames["frigo"],
+        frames["regulation"],
+    ]
+
+
+def test_order_different_from_observed_cycle() -> None:
+    frames = _load_frames()
+    sequence = frames["frigo"] + frames["metadata"] + frames["regulation"]
+    assert FrameExtractor().feed(sequence) == [
+        frames["frigo"],
+        frames["metadata"],
+        frames["regulation"],
+    ]
+
+
+def test_noise_before_signature() -> None:
+    metadata = _load_frames()["metadata"]
+    assert FrameExtractor().feed(b"parasites" + metadata) == [metadata]
+
+
+def test_signature_split_between_calls() -> None:
+    metadata = _load_frames()["metadata"]
+    extractor = FrameExtractor()
+    assert extractor.feed(b"parasites\x55") == []
+    assert extractor.buffered_bytes == b"\x55"
+    assert extractor.feed(metadata[1:]) == [metadata]
+
+
+def test_incoherent_type_header_resynchronizes() -> None:
+    metadata = _load_frames()["metadata"]
+    invalid = bytearray(metadata)
+    invalid[8] = 0x0A
+    assert FrameExtractor().feed(bytes(invalid) + metadata) == [metadata]
+
+
+def test_unknown_length_header_resynchronizes() -> None:
+    metadata = _load_frames()["metadata"]
+    invalid = bytearray(metadata)
+    invalid[10:12] = (81).to_bytes(2, "little")
+    assert FrameExtractor().feed(bytes(invalid) + metadata) == [metadata]
+
+
+def test_announced_size_above_limit_resynchronizes() -> None:
+    frames = _load_frames()
+    extractor = FrameExtractor(max_buffer_size=200)
+    assert extractor.feed(frames["regulation"] + frames["metadata"]) == [
+        frames["metadata"]
+    ]
+
+
+def test_false_signature_in_noise() -> None:
+    metadata = _load_frames()["metadata"]
+    false_header = b"\x55\x00" + b"\x00" * 10
+    assert FrameExtractor().feed(b"\x10" + false_header + metadata) == [metadata]
+
+
+def test_complete_frame_followed_by_incomplete_frame() -> None:
+    frames = _load_frames()
+    extractor = FrameExtractor()
+    assert extractor.feed(frames["metadata"] + frames["frigo"][:20]) == [
+        frames["metadata"]
+    ]
+    assert extractor.buffered_bytes == frames["frigo"][:20]
+    assert extractor.feed(frames["frigo"][20:]) == [frames["frigo"]]
+
+
+def test_variable_size_fragments() -> None:
+    frames = _load_frames()
+    sequence = frames["frigo"] + frames["metadata"] + frames["regulation"]
+    extractor = FrameExtractor()
+    extracted: list[bytes] = []
+    fragment_sizes = (1, 7, 31, 2, 83, 19, 11, 131)
+    offset = 0
+    for size in fragment_sizes:
+        extracted.extend(extractor.feed(sequence[offset : offset + size]))
+        offset += size
+    extracted.extend(extractor.feed(sequence[offset:]))
+    assert extracted == [frames["frigo"], frames["metadata"], frames["regulation"]]
+
+
+def test_buffer_is_empty_after_complete_extraction() -> None:
+    metadata = _load_frames()["metadata"]
+    extractor = FrameExtractor()
+    extractor.feed(metadata)
+    assert extractor.buffered_size == 0
+
+
+def test_incomplete_buffer_is_preserved() -> None:
+    metadata = _load_frames()["metadata"]
+    extractor = FrameExtractor()
+    assert extractor.feed(metadata[:10]) == []
+    assert extractor.buffered_bytes == metadata[:10]
+
+
+def test_repeated_invalid_data_terminates() -> None:
+    invalid_header = b"\x55\x00" + b"\xff" * 10
+    extractor = FrameExtractor()
+    assert extractor.feed(invalid_header * 100) == []
+    assert extractor.buffered_size == 0
+
+
+def test_sequence_with_all_real_fixtures() -> None:
+    frames = _load_frames()
+    sequence = frames["regulation"] + frames["frigo"] + frames["metadata"]
+    assert FrameExtractor().feed(sequence) == [
+        frames["regulation"],
+        frames["frigo"],
+        frames["metadata"],
+    ]