seismo-relay/tests/test_micromate_protocol.py

"""Protocol-layer tests for the Micromate (series-4) live client.

The load-bearing assertion in here is not "our parser understands the device" —
it is **"the bytes we put on the wire are the bytes THOR puts on the wire."**
Every request constant below is lifted from
``bridges/captures/9-24-26 - micromate2/`` (UM12947, firmware 11.0CB), so a
passing test means a real unit has already answered exactly that frame.

Responses are replayed through a scripted transport.  No hardware, no network.
"""
from __future__ import annotations

import os
import sys
from pathlib import Path

import pytest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from micromate import protocol as P
from micromate.framing import ETX, STX, checksum, stuff
from micromate.protocol import (
    ACK_DATA_LEN,
    ChecksumError,
    MicromateProtocol,
    ShortRead,
    UnexpectedResponse,
    chunk_params,
    key_lo_params,
    key_params,
    token_params,
)

FLAGS_CB = 0xC5


# ── Test doubles ──────────────────────────────────────────────────────────────

class ScriptedTransport:
    """Hands back queued responses; records every byte written."""

    def __init__(self, responses: list[bytes] | None = None) -> None:
        self.queue = list(responses or [])
        self.written: list[bytes] = []
        self._connected = True

    # BaseTransport surface actually used by MicromateProtocol
    def connect(self) -> None:
        self._connected = True

    def disconnect(self) -> None:
        self._connected = False

    def is_connected(self) -> bool:
        return self._connected

    def write(self, data: bytes) -> None:
        self.written.append(data)

    def read(self, n: int) -> bytes:
        return self.queue.pop(0) if self.queue else b""


def frame(rsp_sub: int, data: bytes, *, flags: int = FLAGS_CB, page: int = 0) -> bytes:
    """Build a response frame the way a unit would."""
    payload = bytes([0x00, flags, rsp_sub, (page >> 8) & 0xFF, page & 0xFF]) + data
    return bytes([STX]) + stuff(payload + bytes([checksum(payload)])) + bytes([ETX])


def ack(rsp_sub: int) -> bytes:
    return frame(rsp_sub, bytes(ACK_DATA_LEN))


def proto(responses: list[bytes], **kw) -> tuple[MicromateProtocol, ScriptedTransport]:
    t = ScriptedTransport(responses)
    return MicromateProtocol(t, recv_timeout=0.5, **kw), t


# ── Captured THOR request frames ──────────────────────────────────────────────

REQ = {
    "poll":        bytes.fromhex("41021010005b000030000000000000000000009b03"),
    "serial":      bytes.fromhex("41021010001500000a000000000000000000002f03"),
    "state":       bytes.fromhex("41021010004900ffff000000000000000000005703"),
    "compliance":  bytes.fromhex("41021010001a00ffff000000000000000000002803"),
    "arm":         bytes.fromhex("41021010009300ffff00000000000000000000a103"),
    "setup_first": bytes.fromhex("41021010003f00ffff000000000000000000004d03"),
    "setup_next":  bytes.fromhex("41021010004000ffff000000000000000000004e03"),
}

# The complete 0x5A sequence for event 055d4a81 (4,076 bytes → 4 chunks), as
# THOR sent it.  Chunk 1's params hold a literal 0x04 and chunk 3's offset is
# the exact remainder.
REQ_CHUNKS_4A81 = [
    bytes.fromhex("41021010005a00100400055d4a810000000000009b03"),
    bytes.fromhex("41021010005a0010040000001004000000000000007203"),
    bytes.fromhex("41021010005a00100400000008000000000000007603"),
    bytes.fromhex("41021010005a001003ec00000c000000000000006503"),
]
SIZE_4A81 = 4076


# ── Params builders ───────────────────────────────────────────────────────────

def test_event_token_sits_at_params_7():
    """⚠ THOR sends 0xFE here; the protocol reference documents all-zero params.

    That reference entry describes our own browse probing, not THOR's.
    """
    assert token_params() == bytes.fromhex("00000000000000fe0000")


def test_event_record_takes_the_full_key_at_params_4():
    assert key_params(bytes.fromhex("055d4a81")) == bytes.fromhex("00000000055d4a810000")


def test_monitor_log_takes_only_the_low_half_of_the_key():
    """⚠ Inferred from one key value -- see key_lo_params' docstring."""
    assert key_lo_params(bytes.fromhex("055d4a81")) == bytes.fromhex("0000000000004a810000")


def test_chunk_params_switch_from_key_to_byte_offset():
    key = bytes.fromhex("055d4a81")
    assert chunk_params(key, 0) == bytes.fromhex("055d4a81000000000000")
    assert chunk_params(key, 1024) == bytes.fromhex("00000400000000000000")
    assert chunk_params(key, 13312) == bytes.fromhex("00003400000000000000")


@pytest.mark.parametrize("bad", [b"", b"\x01\x02\x03", b"\x01\x02\x03\x04\x05"])
def test_params_builders_reject_a_wrong_length_key(bad):
    for fn in (key_params, key_lo_params):
        with pytest.raises(ValueError):
            fn(bad)


# ── Each read emits the frame THOR emits ──────────────────────────────────────

@pytest.mark.parametrize(
    "name, method, rsp_sub, data_len",
    [
        ("poll",        "poll",              0xA4, 59),
        ("serial",      "read_serial",       0xEA, 21),
        ("state",       "read_state",        0xB6, 16),
        ("compliance",  "read_compliance_config", 0xE5, 2103),
        ("setup_first", "read_first_setup",  0xC0, 266),
        ("setup_next",  "read_next_setup",   0xBF, 266),
    ],
)
def test_reads_match_thors_wire_bytes(name, method, rsp_sub, data_len):
    p, t = proto([frame(rsp_sub, bytes(data_len))])
    getattr(p, method)()
    assert t.written == [REQ[name]]


def test_arm_event_matches_thors_wire_bytes():
    p, t = proto([ack(0x6C)])
    p.arm_event()
    assert t.written == [REQ["arm"]]


def test_poll_is_the_only_read_with_a_non_ffff_offset_besides_serial():
    """Reads are single-step at 0xFFFF; POLL and SERIAL are the exceptions."""
    assert set(P._OFFSETS) == {P.SUB_POLL, P.SUB_SERIAL}
    assert P._OFFSETS[P.SUB_POLL] == 0x0030
    assert P._OFFSETS[P.SUB_SERIAL] == 0x000A


# ── The chunk walk ────────────────────────────────────────────────────────────

def test_download_reproduces_thors_chunk_sequence_byte_for_byte():
    """The whole point of step 2.  Four chunks, 4,076 bytes, THOR's exact frames."""
    payload = bytes(range(256)) * 16          # 4096 B, we use the first 4076
    payload = payload[:SIZE_4A81]
    responses = []
    for i in range(4):
        want = min(P.THOR_CHUNK_SIZE, SIZE_4A81 - i * P.THOR_CHUNK_SIZE)
        body = payload[i * P.THOR_CHUNK_SIZE: i * P.THOR_CHUNK_SIZE + want]
        responses.append(frame(0xA5, bytes(11) + body, page=want // 256))

    p, t = proto(responses)
    got = p.read_event_file(bytes.fromhex("055d4a81"), SIZE_4A81,
                            chunk_size=P.THOR_CHUNK_SIZE)

    assert t.written == REQ_CHUNKS_4A81
    assert got == payload
    assert len(got) == SIZE_4A81


@pytest.mark.parametrize(
    "size, n_chunks, last_offset",
    [
        (4076, 4, 0x03EC), (11032, 11, 0x0318), (11502, 12, 0x00EE),
        (13424, 14, 0x0070), (8746, 9, 0x022A), (6092, 6, 0x03CC),
        (1024, 1, 0x0400), (1, 1, 0x0001), (1025, 2, 0x0001),
    ],
)
def test_chunk_count_and_final_offset(size, n_chunks, last_offset):
    """The first six rows are the six bench events, with THOR's real offsets."""
    responses = []
    for i in range(n_chunks):
        want = min(P.THOR_CHUNK_SIZE, size - i * P.THOR_CHUNK_SIZE)
        responses.append(frame(0xA5, bytes(11) + bytes(want)))

    p, t = proto(responses)
    p.read_event_file(bytes.fromhex("055d4a81"), size,
                      chunk_size=P.THOR_CHUNK_SIZE)

    assert len(t.written) == n_chunks
    # offset is payload[4:5] of the request; recover it from the built frame
    final = t.written[-1]
    assert final[7:9] in (
        bytes([last_offset >> 8, last_offset & 0xFF]),
        # a 0x02/0x03/0x04/0x10 high byte arrives escaped, shifting the pair
        bytes([0x10, last_offset >> 8]),
    )


def test_a_short_chunk_is_absorbed_and_the_remainder_refetched():
    """⚠ Changed 2026-10-02: a short chunk is no longer an error.

    It used to raise, on the principle that a silently short event is the failure
    mode this codebase keeps hitting.  But the device *legitimately* returns
    short — it clamps an over-large request to 16,384 B without saying so — and
    the real protection is tracking the offset by bytes received, which makes a
    short response cost one extra request instead of corrupting the file.  The
    total length is still checked, so a genuinely truncated event still raises.
    """
    p, t = proto([frame(0xA5, bytes(11) + b"\xaa" * 900),
                  frame(0xA5, bytes(11) + b"\xbb" * 124)])
    got = p.read_event_file(bytes.fromhex("055d4a81"), 1024,
                            chunk_size=P.THOR_CHUNK_SIZE)
    assert got == b"\xaa" * 900 + b"\xbb" * 124
    assert len(t.written) == 2, "the 124 B remainder was refetched"


def test_a_chunk_too_short_to_hold_its_header_raises():
    p, _ = proto([frame(0xA5, bytes(4))])
    with pytest.raises(ShortRead, match="too short to hold a chunk header"):
        p.read_event_file(bytes.fromhex("055d4a81"), 1024,
                          chunk_size=P.THOR_CHUNK_SIZE)


def test_download_rejects_a_nonsense_size():
    p, _ = proto([])
    with pytest.raises(ValueError):
        p.read_event_file(bytes.fromhex("055d4a81"), 0)


# ── The monitor-log walk ──────────────────────────────────────────────────────

def test_monitor_log_walk_ends_on_a_short_response():
    """⚠ Not a keyed read -- the same request repeated, device-side cursor.

    Eight records then an 11-byte ack, which is what the capture shows.
    """
    key = bytes.fromhex("055d4a81")
    responses = [frame(0xF5, bytes(297)) for _ in range(8)] + [ack(0xF5)]
    p, t = proto(responses)

    records = []
    while (rec := p.read_monitor_log_next(key)) is not None:
        records.append(rec)

    assert len(records) == 8
    assert len(t.written) == 9
    assert len(set(t.written)) == 1, "every request in the walk is identical"


# ── Error handling ────────────────────────────────────────────────────────────

def test_a_bad_checksum_raises_by_default():
    """⚠ Deliberately stricter than the Series III sibling.

    That one logs and continues because its parser cannot always tell an
    inner-frame delimiter from a checksum byte.  The Micromate rule is exact on
    251/251 captured frames, so a mismatch here means something real.
    """
    bad = bytearray(frame(0xA4, bytes(59)))
    bad[-2] ^= 0xFF
    p, _ = proto([bytes(bad)])
    with pytest.raises(ChecksumError, match="checksum mismatch"):
        p.poll()


def test_a_bad_checksum_can_be_downgraded_for_field_diagnosis():
    bad = bytearray(frame(0xA4, bytes(59)))
    bad[-2] ^= 0xFF
    p, _ = proto([bytes(bad)], strict_checksums=False)
    assert p.poll().sub == 0xA4


def test_the_wrong_response_sub_raises():
    p, _ = proto([frame(0xE0, bytes(19))])      # 0xE0 answers 0x1F, not 0x1E
    with pytest.raises(UnexpectedResponse, match="expected SUB 0xE1"):
        p.read_event_first()


def test_a_timeout_reports_how_many_bytes_arrived():
    """Separates "nothing came back" from "bytes arrived but never framed".

    Those have completely different causes -- and on a Micromate the second one
    is the signature of a modem forwarding a session it should not be.
    """
    p, _ = proto([])
    with pytest.raises(P.TimeoutError, match="0 bytes were received"):
        p.poll()

    unframed = b"\x02\x00\xc5\xa4garbage-no-terminator"
    p2, _ = proto([unframed])
    with pytest.raises(P.TimeoutError, match=f"{len(unframed)} bytes were received"):
        p2.poll()


def test_a_leftover_frame_is_discarded_rather_than_answered_with():
    """If a read returns two frames, the extra must not answer the NEXT request.

    Every exchange resets the parser before sending, so anything already
    buffered is treated as stale.  Delivering it would be the worse failure:
    `expected_sub` happens to catch a mismatched SUB, but a same-SUB leftover
    would sail through and return data for the wrong key.
    """
    # Both frames arrive while answering arm_event(); the 0xE1 is left over.
    p, _ = proto([ack(0x6C) + frame(0xE1, b"\xaa" * 19)])
    p.arm_event()

    # The next request gets no bytes of its own, so it must time out rather
    # than hand back the stale 0xE1.
    with pytest.raises(P.TimeoutError):
        p.read_event_first()


# ── Against the real capture, when it happens to be present ───────────────────

_CAPTURES = (
    Path(__file__).resolve().parents[1]
    / "bridges" / "captures" / "9-24-26 - micromate2"
)
_DOWNLOAD = "raw_bw_20260925_011403_Download_events_then_delete_1_event.bin"


@pytest.mark.skipif(
    not (_CAPTURES / _DOWNLOAD).is_file(),
    reason="capture is gitignored; present only on a dev box",
)
def test_every_captured_download_frame_is_one_we_would_have_sent():
    """Replay the real session: for each event, assert our chunk walk emits
    exactly the frames THOR emitted -- all 50-odd of them, six events."""
    from micromate.framing import ACK, DLE

    def destuffed_frames(blob: bytes, is_req: bool):
        i, n = 0, len(blob)
        while i < n:
            if is_req:
                if not (blob[i] == ACK and i + 1 < n and blob[i + 1] == STX):
                    i += 1
                    continue
                j = i + 2
            else:
                if blob[i] != STX:
                    i += 1
                    continue
                j = i + 1
            out = bytearray()
            while j < n:
                if blob[j] == DLE and j + 1 < n:
                    out.append(blob[j + 1])
                    j += 2
                    continue
                if blob[j] == ETX:
                    break
                out.append(blob[j])
                j += 1
            if len(out) >= 6:
                yield blob[i:j + 1], bytes(out[:-1])
            i = j + 1

    bw = list(destuffed_frames((_CAPTURES / _DOWNLOAD).read_bytes(), True))
    s3 = list(
        destuffed_frames(
            (_CAPTURES / _DOWNLOAD.replace("raw_bw", "raw_s3")).read_bytes(), False
        )
    )

    # Group THOR's 0x5A frames per event, taking each event's key+size from the
    # 1E/1F that preceded them.
    events, cur = [], None
    for (wire, req), (_, rsp) in zip(bw, s3):
        sub, data = req[2], rsp[5:]
        if sub in (0x1E, 0x1F) and len(data) >= 19:
            cur = {"key": data[11:15], "size": int.from_bytes(data[15:19], "big"),
                   "reqs": [], "rsps": []}
            if cur["size"]:
                events.append(cur)
        elif sub == 0x5A and cur is not None:
            cur["reqs"].append(wire)
            cur["rsps"].append(rsp)

    # The capture walks the chain twice (it deletes an event on the second
    # pass), so some 1E/1F hits carry a size but no download behind them.
    events = [e for e in events if e["reqs"]]
    assert len(events) == 6, f"expected 6 downloaded events, found {len(events)}"

    total = 0
    for e in events:
        p, t = proto([bytes([STX]) + stuff(r + bytes([checksum(r)])) + bytes([ETX])
                      for r in e["rsps"]])
        got = p.read_event_file(e["key"], e["size"],
                                chunk_size=P.THOR_CHUNK_SIZE)
        assert t.written == e["reqs"], (
            f"event {e['key'].hex()}: our {len(t.written)} frames differ from "
            f"THOR's {len(e['reqs'])}"
        )
        assert len(got) == e["size"]
        total += len(e["reqs"])

    assert total == 56, f"expected 56 download frames across the 6 events, saw {total}"


# ── Above 64 KB: the limit UM20147 already exceeds ────────────────────────────

def test_chunk_offsets_carry_past_64_kb():
    """⚠ The offset is a uint32 at params[0:4], not a uint16 at params[2:4].

    Every offset THOR was observed to send fits in two bytes (largest 0x3400),
    so params[0:2] was always zero and the field looks narrower than it is.
    Reading it as a uint16 caps a download at 65,536 B — and UM20147 holds a
    72,560 B event, so the cap is not hypothetical.

    This is the Series III 64 KB page-boundary bug in a new guise; that one is
    still open.  An address field whose high bytes are zero in every capture is
    not a narrow field, it is an untested one.
    """
    key = bytes.fromhex("055d4a83")
    # Below the old cap: byte-identical to the uint16 form, so nothing verified
    # against THOR's frames changes.
    for off in (1024, 13312, 65535):
        assert chunk_params(key, off) == bytes(2) + off.to_bytes(2, "big") + bytes(6)
    # Above it: the carry lands in params[1].
    assert chunk_params(key, 65536) == bytes.fromhex("00010000") + bytes(6)
    assert chunk_params(key, 71680) == bytes.fromhex("00011800") + bytes(6)


def test_a_chunk_offset_with_0x10_in_it_is_escaped_on_the_wire():
    """At offset 1 MiB params[1] is 0x10, which must go out as `10 10`.

    Nothing below 64 KB can produce this, so it only became reachable with the
    uint32 offset — and an unescaped 0x10 is the bug class that cost the
    Series III `5A` walk a release.
    """
    params = chunk_params(bytes(4), 1 << 20)
    assert params[:4] == bytes.fromhex("00100000")

    from micromate.framing import build_request
    frame = build_request(P.SUB_BULK_DOWNLOAD, 0x0400, params)
    assert bytes.fromhex("1010") in frame


def test_a_72kb_event_downloads_in_71_chunks():
    """UM20147's event 055d4a83, the one that broke the uint16 assumption.

    ✅ Confirmed against the real unit 2026-10-01: 71 chunks, exact size, with
    chunks 64-70 carrying 0x01 in params[1].  This test pins the arithmetic the
    hardware agreed with.
    """
    size = 72560
    n = 71
    responses = []
    for i in range(n):
        want = min(P.THOR_CHUNK_SIZE, size - i * P.THOR_CHUNK_SIZE)
        responses.append(frame(0xA5, bytes(11) + bytes(want)))

    p, t = proto(responses)
    got = p.read_event_file(bytes.fromhex("055d4a83"), size,
                            chunk_size=P.THOR_CHUNK_SIZE)

    assert len(got) == size
    assert len(t.written) == n
    # Chunk 64 is the first past the old cap; its params must carry the 0x01.
    wire = t.written[64]
    assert bytes.fromhex("000100") in wire, "the carry into params[1] is on the wire"


# ── The 16 KB ceiling, and the silent clamp ───────────────────────────────────

def test_default_chunk_size_is_the_measured_ceiling_not_thors():
    assert P.CHUNK_SIZE == 16384
    assert P.THOR_CHUNK_SIZE == 1024


def test_a_silently_clamped_response_is_absorbed_not_failed():
    """⚠ Ask for more than the device serves and it CLAMPS — silently.

    Measured on UM20147: a 32,768 B request returns exactly 16,384 B of correct
    data, no error.  A loop striding by a fixed chunk size would either fail on
    that or skip the bytes it never collected.  Driving by bytes received makes
    it a non-event: one more request.

    Here a 20,000 B event is fetched with chunk_size=16384 against a device
    pretending to clamp at 8192, so the walk must take 3 requests at
    offsets 0 / 8192 / 16384.
    """
    size, clamp = 20000, 8192
    payload = bytes(range(256)) * 100
    payload = payload[:size]

    served, responses = 0, []
    while served < size:
        n = min(clamp, size - served)
        responses.append(frame(0xA5, bytes(11) + payload[served:served + n]))
        served += n

    p, t = proto(responses)
    got = p.read_event_file(bytes.fromhex("055d4a83"), size, chunk_size=16384)

    assert got == payload
    assert len(t.written) == 3, "two clamped requests plus the remainder"


def test_a_chunk_that_returns_nothing_raises_rather_than_spinning():
    """A byte-driven loop must not loop forever on zero progress."""
    p, _ = proto([frame(0xA5, bytes(11))] * 4)
    with pytest.raises(ShortRead, match="got none"):
        p.read_event_file(bytes.fromhex("055d4a83"), 5000)


def test_the_72kb_event_now_takes_5_requests_not_71():
    size = 72560
    served, responses = 0, []
    while served < size:
        n = min(P.CHUNK_SIZE, size - served)
        responses.append(frame(0xA5, bytes(11) + bytes(n)))
        served += n

    p, t = proto(responses)
    got = p.read_event_file(bytes.fromhex("055d4a83"), size)
    assert len(got) == size
    assert len(t.written) == 5, "14x fewer round trips than THOR's 71"