seismo-relay/tests/test_micromate_framing.py

"""Framing tests for the Micromate (series-4) live protocol.

Every constant below is a **real frame**, lifted from
``bridges/captures/9-24-26 - micromate2/`` (UM12947, firmware 11.0CB) or from
``scratch/fake_unit.py``, which preserves a POLL probe reply.  Frames are
embedded as hex rather than read from disk because both ``bridges/captures/``
and ``tests/fixtures/`` are gitignored -- these tests must pass on a fresh
clone.

The few synthesised frames are marked ``SYNTH_`` and each says what it stands
in for and why a captured frame was not available.

Two of these tests exist because the first draft of
``docs/micromate_client_spec.md`` got the rule wrong, and both wrong rules
fail quietly -- a frame the unit ignores, or a checksum that reads as bad:

  * ``test_builder_matches_thor_byte_for_byte`` -- the escape set.  Escaping
    only 0x10 (the series-3 rule) reproduces 161 of Thor's 218 read frames.
  * ``test_checksum_is_plain_sum8_over_destuffed_payload`` -- the checksum.
    The DLE-aware variant disagrees with the wire on 55 of 251 responses.
"""
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.framing import (
    ACK,
    DLE,
    ETX,
    FLAGS_BLASTWARE,
    FLAGS_THOR,
    STX,
    MicromateFrame,
    MicromateFrameParser,
    build_request,
    checksum,
    stuff,
    unstuff,
)

# ── Captured response frames ──────────────────────────────────────────────────

# POLL probe reply, 19 B on the wire -- the shortest valid frame there is.
# Preserved in scratch/fake_unit.py, captured from UM12947 on 2026-09-24.
# Its payload[8:10] is 0x0030, the data length POLL then asks for.
RSP_POLL_PROBE = bytes.fromhex("0200c5a400000000000030000000000000009903")

# SUB 0x49 -> 0xB6, the cheap state read.  Carries a literal 0x02, escaped.
RSP_STATE = bytes.fromhex("0200c5b6000005000000000000000000001002e8000b007503")

# SUB 0x48 -> 0xB7, a path-addressed file read.  offset_hi 0x04 arrives escaped,
# so page_hi is only correct if the parser destuffs before indexing.
RSP_FILE_READ = bytes.fromhex("0200c5b70000100400000000000000010000000000018203")

# An 0x5A download chunk, 138 B on the wire.  THE checksum case: its payload
# holds literal 0x10 bytes, so plain SUM8 (0xC1, correct) and the DLE-aware
# variant (0x91) disagree.  Also holds a literal 0x41, which is NOT escaped.
RSP_CHUNK_WITH_DLE = bytes.fromhex(
    "0200c5a500007000003400000000000000e3fd1f10020f0f0e2e1e1fd4f2d4c3d2f00f000"
    "e003fe03fe200d6e2d12d2e1c4e101011d101f3d0f3e0f2efe23d2b2e3c0d3e0e2c4101e0"
    "d4d3e21f1d311e202fe2e2f1e3f100210d201002ee2e22d2f2f0f30fd3f0101000f0111f1"
    "ff01e011010011f0f1002b4d200e0f0302c4010020001fffedb2dc103"
)

# ── Captured request frames (Thor -> unit) ────────────────────────────────────

REQ_POLL       = bytes.fromhex("41021010005b000030000000000000000000009b03")
REQ_SERIAL     = bytes.fromhex("41021010001500000a000000000000000000002f03")
REQ_STATE      = bytes.fromhex("41021010004900ffff000000000000000000005703")
REQ_COMPLIANCE = bytes.fromhex("41021010001a00ffff000000000000000000002803")
REQ_ARM_EVENT  = bytes.fromhex("41021010009300ffff00000000000000000000a103")

# ⚠ The two frames that break a 0x10-only escaper.
# Scheduler enable: params[7] = 0x03, on the wire as `10 03`.
REQ_SCHED_ON = bytes.fromhex("41021010004700ffff00000000000000100300005803")
# Bulk download, first chunk of event 055d4a81: offset 0x0400 puts a literal
# 0x04 in offset_hi, on the wire as `10 04`.  EVERY download frame needs this.
REQ_DOWNLOAD_CHUNK0 = bytes.fromhex(
    "41021010005a00100400055d4a810000000000009b03"
)
# A later chunk of the same event: params[2:4] = 0x1000, doubled to `10 10`.
REQ_DOWNLOAD_CHUNK4 = bytes.fromhex(
    "41021010005a0010040000001010000000000000007e03"
)


# ── Stuffing ──────────────────────────────────────────────────────────────────

def test_escape_set_is_exactly_four_bytes():
    """0x02, 0x03, 0x04 and 0x10 -- and nothing else.

    ACK (0x41) in particular is NOT escaped; assuming it was reproduced only
    196 of 251 captured responses.
    """
    assert stuff(bytes([0x02, 0x03, 0x04, 0x10])) == bytes(
        [DLE, 0x02, DLE, 0x03, DLE, 0x04, DLE, 0x10]
    )
    for b in (0x00, 0x01, 0x05, 0x41, 0xC5, 0xFF):
        assert stuff(bytes([b])) == bytes([b]), f"0x{b:02x} must not be escaped"


def test_unstuff_is_uniform_with_no_inner_frame_carve_out():
    """`10 XX` -> `XX` for any XX -- the series-3 DLE+ETX exception is absent."""
    assert unstuff(bytes.fromhex("1003")) == b"\x03"
    assert unstuff(bytes.fromhex("1010")) == b"\x10"
    assert unstuff(bytes.fromhex("001002ff")) == bytes.fromhex("0002ff")


def test_stuff_unstuff_round_trips_over_every_byte_value():
    data = bytes(range(256))
    assert unstuff(stuff(data)) == data


def test_a_trailing_dle_is_held_not_dropped():
    """A DLE as the last byte of a chunk must not consume nothing and vanish."""
    assert unstuff(b"\xff\x10") == b"\xff\x10"


# ── Checksum ──────────────────────────────────────────────────────────────────

def test_checksum_is_plain_sum8_over_destuffed_payload():
    """⚠ Plain SUM8 -- do NOT exclude 0x10 bytes.

    RSP_CHUNK_WITH_DLE is a real frame whose payload holds literal 0x10 bytes.
    The wire says 0xC1; plain SUM8 gives 0xC1 and the DLE-aware variant used by
    series-3 `5A`/write frames gives 0x91.  55 of 251 captured responses
    disagree the same way.
    """
    payload = unstuff(RSP_CHUNK_WITH_DLE[1:-1])[:-1]
    chk_on_wire = unstuff(RSP_CHUNK_WITH_DLE[1:-1])[-1]

    assert DLE in payload, "this frame is only interesting if it holds a 0x10"
    assert checksum(payload) == chk_on_wire == 0xC1
    assert (sum(b for b in payload if b != DLE) & 0xFF) == 0x91  # the wrong rule


def test_every_captured_frame_validates():
    parser = MicromateFrameParser()
    frames = parser.feed(
        RSP_POLL_PROBE + RSP_STATE + RSP_FILE_READ + RSP_CHUNK_WITH_DLE
    )
    assert len(frames) == 4
    assert all(f.checksum_valid for f in frames)


# ── Request builder ───────────────────────────────────────────────────────────

@pytest.mark.parametrize(
    "wire, sub, offset, params",
    [
        (REQ_POLL,            0x5B, 0x0030, bytes(10)),
        (REQ_SERIAL,          0x15, 0x000A, bytes(10)),
        (REQ_STATE,           0x49, 0xFFFF, bytes(10)),
        (REQ_COMPLIANCE,      0x1A, 0xFFFF, bytes(10)),
        (REQ_ARM_EVENT,       0x93, 0xFFFF, bytes(10)),
        (REQ_SCHED_ON,        0x47, 0xFFFF, bytes.fromhex("00000000000000030000")),
        (REQ_DOWNLOAD_CHUNK0, 0x5A, 0x0400, bytes.fromhex("055d4a81000000000000")),
        (REQ_DOWNLOAD_CHUNK4, 0x5A, 0x0400, bytes.fromhex("00001000000000000000")),
    ],
    ids="poll serial state compliance arm sched_on dl_chunk0 dl_chunk4".split(),
)
def test_builder_matches_thor_byte_for_byte(wire, sub, offset, params):
    assert build_request(sub, offset, params) == wire


def test_a_0x10_in_params_needs_no_special_handling():
    """The spec's one open question.  Thor sends it; the wire doubles it."""
    frame = build_request(0x5A, 0x0400, bytes.fromhex("00001000000000000000"))
    assert bytes([DLE, DLE]) in frame
    assert frame == REQ_DOWNLOAD_CHUNK4


def test_builder_rejects_malformed_arguments():
    with pytest.raises(ValueError):
        build_request(0x5B, 0, bytes(9))
    with pytest.raises(ValueError):
        build_request(0x5B, 0x10000)
    with pytest.raises(ValueError):
        build_request(0x100)


# ── Parsing ───────────────────────────────────────────────────────────────────

def test_poll_probe_reply_fields():
    (f,) = MicromateFrameParser().feed(RSP_POLL_PROBE)
    assert f.sub == 0xA4
    assert f.request_sub == 0x5B
    assert f.flags == FLAGS_BLASTWARE
    assert f.firmware_line == "blastware"
    assert f.checksum_valid
    assert f.probe_length == 0x0030


def test_probe_length_is_a_uint16_not_a_byte():
    """⚠ Read as data[3] alone, SUB 0x1A's 0x082C (2092) reads as 44 -- 47x low.

    SYNTHESISED: no probe response survives in the captures on disk (the
    9-24-26 session uses single-step reads at offset 0xFFFF throughout, so it
    contains no probes at all).  The field position is taken from the captured
    POLL probe reply above, which does exercise it for real.
    """
    payload = bytes([0x00, FLAGS_BLASTWARE, 0xE5, 0x00, 0x00]) + bytes(
        [0x00, 0x00, 0x00, 0x08, 0x2C]
    )
    synth = bytes([STX]) + stuff(payload + bytes([checksum(payload)])) + bytes([ETX])

    (f,) = MicromateFrameParser().feed(synth)
    assert f.probe_length == 0x082C == 2092
    assert f.data[3] == 0x08, "the high byte is where a byte-wide read loses 2048"


def test_escaped_bytes_land_in_the_right_field():
    """RSP_FILE_READ's first data byte is 0x04, which arrives as `10 04`.

    Without destuffing it reads as 0x10 and every field after it is one byte
    late -- the failure mode that put `SUB 0x02` in the log as `SUB_10` for an
    afternoon.
    """
    (f,) = MicromateFrameParser().feed(RSP_FILE_READ)
    assert f.sub == 0xB7
    assert f.request_sub == 0x48
    assert (f.page_hi, f.page_lo) == (0x00, 0x00)
    assert f.data[0] == 0x04
    assert len(f.data) == 15, "one byte shorter than the wire suggests"
    assert f.checksum_valid


# The real 11.0BD POLL data section, UM20147 over USB 2026-09-30.  This
# replaces a synthesised frame -- the Thor firmware line was inference-only
# until this capture.
_POLL_BD_DATA = bytes.fromhex(
    "30000000000000000000000000005649"
    "6e7374616e74656c000600c3f04a00e4"
    "194f0052024d4d2f495345452f530058"
    "f406001f60c0755760c075"
)


def test_thor_firmware_line_survives_destuffing():
    """⚠ flags = 0x03 is ETX, so it arrives as `10 03`.

    A parser that does not destuff ends the frame at byte 2 on half the fleet.

    The data section is REAL (UM20147, 11.0BD); the frame wrapper is rebuilt,
    which is sound because the framing is verified 251/251 elsewhere in this
    file.  scratch/mm_frame_parse.py reported payload=64 for this frame, and
    the assertion below pins that, so the reconstruction is checked rather than
    assumed.
    """
    payload = bytes([0x00, FLAGS_THOR, 0xA4, 0x00, 0x00]) + _POLL_BD_DATA
    assert len(payload) == 64, "the parser reported payload=64 for this frame"
    wire = bytes([STX]) + stuff(payload + bytes([checksum(payload)])) + bytes([ETX])

    assert bytes([DLE, ETX]) in wire, "flags 0x03 must be escaped on the wire"
    (f,) = MicromateFrameParser().feed(wire)
    assert f.flags == FLAGS_THOR
    assert f.firmware_line == "thor"
    assert f.sub == 0xA4
    assert f.request_sub == 0x5B
    assert f.checksum_valid
    assert b"MM/ISEE/S\x00" in f.data, "the shorter BD model string"


def test_an_escaped_checksum_byte_is_read_correctly():
    """SYNTHESISED, but the behaviour is real: three captured responses have a
    checksum of 0x02/0x03/0x04 and all three escape it on the wire.  The
    shortest is 1,070 B (an 0x5A chunk in
    raw_s3_20260925_011403_Download_events_then_delete_1_event.bin, chk = 0x03),
    too long to embed for one byte's worth of assertion.
    """
    payload = bytes([0x00, FLAGS_BLASTWARE, 0xA4, 0x00, 0x00, 0x03])
    assert checksum(payload) == 0x03 + FLAGS_BLASTWARE + 0xA4 & 0xFF
    body = payload + bytes([checksum(payload)])
    synth = bytes([STX]) + stuff(body) + bytes([ETX])

    (f,) = MicromateFrameParser().feed(synth)
    assert f.chk_byte == checksum(payload)
    assert f.checksum_valid


def test_a_corrupted_checksum_still_yields_a_frame():
    """Flag it, do not swallow it -- a dropped frame looks like a dead unit."""
    broken = bytearray(RSP_POLL_PROBE)
    broken[-2] ^= 0xFF
    (f,) = MicromateFrameParser().feed(bytes(broken))
    assert f.sub == 0xA4
    assert not f.checksum_valid


def test_a_truncated_frame_yields_nothing():
    parser = MicromateFrameParser()
    assert parser.feed(RSP_POLL_PROBE[:-1]) == []
    assert parser.frames == []
    assert parser.bytes_fed == len(RSP_POLL_PROBE) - 1


def test_a_frame_too_short_to_hold_a_header_is_rejected():
    assert MicromateFrameParser().feed(bytes([STX, 0x00, 0xC5, 0xA4, ETX])) == []


def test_request_frames_are_not_mistaken_for_responses():
    """Feeding a bidirectional capture must yield only the unit's side."""
    parser = MicromateFrameParser()
    frames = parser.feed(REQ_POLL + RSP_POLL_PROBE + REQ_SERIAL)
    assert len(frames) == 1
    assert frames[0].request_sub == 0x5B


def test_leading_noise_is_discarded():
    """Cold-boot banners and the RV55's RING/CONNECT chatter precede frames."""
    noise = b"\r\nRING\r\n\r\nCONNECT\r\n" + bytes([ACK])
    (f,) = MicromateFrameParser().feed(noise + RSP_POLL_PROBE)
    assert f.sub == 0xA4
    assert f.checksum_valid


def test_frames_split_across_feeds_reassemble():
    """The transport hands over whatever the socket returned, DLE pairs and all."""
    whole = RSP_STATE + RSP_CHUNK_WITH_DLE
    for cut in (1, 2, 5, 17, 24, 25, 40, len(RSP_STATE), len(whole) - 1):
        parser = MicromateFrameParser()
        got = parser.feed(whole[:cut]) + parser.feed(whole[cut:])
        assert len(got) == 2, f"split at {cut} lost a frame"
        assert all(f.checksum_valid for f in got), f"split at {cut} broke a checksum"


def test_reset_clears_partial_state():
    parser = MicromateFrameParser()
    parser.feed(RSP_POLL_PROBE[:6])
    parser.reset()
    assert parser.bytes_fed == 0
    (f,) = parser.feed(RSP_POLL_PROBE)
    assert f.checksum_valid


def test_firmware_line_of_an_unknown_flags_byte():
    f = MicromateFrame(sub=0xA4, flags=0x99, page_hi=0, page_lo=0,
                       data=b"", checksum_valid=True)
    assert f.firmware_line == "unknown"
    assert f.probe_length is None


# ── Whole-session round trip ──────────────────────────────────────────────────

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


@pytest.mark.skipif(
    not _CAPTURES.is_dir(),
    reason="capture directory is gitignored; present only on a dev box",
)
def test_whole_captured_sessions_parse_with_no_bad_checksums():
    """Belt-and-braces against the real bytes when they happen to be here.

    scratch/mm_frame_parse.py reported zero bad checksums on these sessions
    only because it accepts a frame matching *either* checksum rule.  This
    asserts the single rule holds across all of them.
    """
    total = 0
    for path in sorted(_CAPTURES.rglob("raw_s3_*.bin")):
        parser = MicromateFrameParser()
        frames = parser.feed(path.read_bytes())
        assert frames, f"{path.name}: no frames parsed"
        bad = [f for f in frames if not f.checksum_valid]
        assert not bad, f"{path.name}: {len(bad)} bad checksums"
        total += len(frames)
    assert total == 251, f"expected 251 response frames across the corpus, got {total}"


@pytest.mark.skipif(
    not _CAPTURES.is_dir(),
    reason="capture directory is gitignored; present only on a dev box",
)
def test_builder_reproduces_every_captured_read_frame():
    """218/218.  This is the test that would have caught the escape-set error."""
    checked = 0
    for path in sorted(_CAPTURES.rglob("raw_bw_*.bin")):
        blob = path.read_bytes()
        i = 0
        while i < len(blob):
            if not (blob[i] == ACK and i + 1 < len(blob) and blob[i + 1] == STX):
                i += 1
                continue
            j = i + 2
            body = bytearray()
            while j < len(blob):
                if blob[j] == DLE and j + 1 < len(blob):
                    body.append(blob[j + 1])
                    j += 2
                    continue
                if blob[j] == ETX:
                    break
                body.append(blob[j])
                j += 1
            payload = bytes(body[:-1])
            if len(payload) == 16:      # a read frame; writes carry a data section
                sub = payload[2]
                offset = (payload[4] << 8) | payload[5]
                assert build_request(sub, offset, payload[6:16]) == blob[i:j + 1], (
                    f"{path.name} @0x{i:04x} SUB=0x{sub:02x} offset=0x{offset:04x}"
                )
                checked += 1
            i = j + 1
    assert checked == 218, f"expected 218 read frames, checked {checked}"