home-assistant-arkteos/tests/test_frame_extractor.py

"""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"],
    ]