"""Tests for the host sender: pacing, chunking, and the status back-channel.

The plan's acceptance run says "MAME with a tap interface: run vidsend.py for
200 frames at 30 fps".  That run cannot happen in this repository: the MAME
driver creates no ``device_network_interface``, so the emulated LANCE is not
attached to the host at all (ruling R11).  What can be tested here is every
layer below the socket, and that is what this file does:

* the *schedule* -- frame ``n`` leaves at ``t0 + n/fps``, measured on an
  injected clock rather than by sleeping for real;
* the *wire form* -- a frame is chunked at 1450 bytes, only the last datagram is
  padded to the 60-byte Ethernet minimum, and the pieces reassemble to exactly
  the bytes ``vidsrc.StripSource`` produced;
* the *status* path -- a status frame is recognised in an Ethernet frame,
  decoded field by field, and ``drops``/``crc_err`` decide the exit status;
* the *loop* -- a fake counterparty that reassembles datagrams, counts frames
  and answers with status frames, so a 200-frame run's verdict is exercised
  without a NIC or a terminal.

No test here touches a socket, and none of them claims anything about whether a
frame would arrive on a real wire.
"""
import io
import struct

import pytest

from harness.udpstream import make_tiles, vidsend
from harness.udpstream.frame import (BROADCAST, ETHERTYPE, HEADER_SIZE,
                                     KIND_DICT_CHUNK, KIND_STATUS, TERMINAL_MAC,
                                     ethernet_frame, pack_header, pack_status,
                                     unpack_header)
from harness.udpstream.vidsrc import CHUNK, ROW_BYTES, StripSource

#: The host's own address.  Locally administered, so it cannot be mistaken for
#: the terminal's (08:00:2b:16:4b:5a, TERMINAL_MAC) in a datagram assertion.
HOST_MAC = b"\x02\x00\x00\x00\x00\x01"


class FakeClock:
    """A monotonic clock that only moves when something sleeps on it."""

    def __init__(self):
        self.t = 0.0
        self.sleeps = []

    def now(self):
        return self.t

    def sleep(self, seconds):
        self.sleeps.append(seconds)
        self.t += seconds


class RecordingTransport:
    """Stands in for the NIC: records each datagram and when it was sent."""

    def __init__(self, clock):
        self.clock = clock
        self.datagrams = []
        self.times = []

    def __call__(self, data):
        self.datagrams.append(data)
        self.times.append(self.clock.now())


def make_sender(clock, transport, **kw):
    kw.setdefault("rows", 8)
    kw.setdefault("stretch", 128)
    kw.setdefault("fps", 10.0)
    kw.setdefault("status_wait", 0.0)
    kw.setdefault("out", io.StringIO())
    kw.setdefault("err", io.StringIO())
    kw.setdefault("src", HOST_MAC)
    return vidsend.Sender(transport=transport, clock=clock.now,
                          sleep=clock.sleep, **kw)


def status_frame(frames_rx=0, crc_err=0, drops=0, csr0=0x20F2, heap_free_kb=2048,
                 frame_no=0, body=None):
    body = pack_status(frames_rx, crc_err, drops, csr0, heap_free_kb) \
        if body is None else body
    hdr = pack_header(KIND_STATUS, frame_no, len(body), 0, 0, payload=body)
    return ethernet_frame(hdr + body, src=TERMINAL_MAC)


# --- pacing ---------------------------------------------------------------

def test_frame_n_leaves_at_t0_plus_n_over_fps():
    """The schedule, not "about 10 a second": exactly t0 + n/fps."""
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_sender(clock, transport, fps=10.0, frames=5)
    sender.run()      # the exit status is the status verdict, tested below
    assert transport.times == pytest.approx([0.0, 0.1, 0.2, 0.3, 0.4])
    # And the sleeps are the differences: the first frame does not sleep.
    assert clock.sleeps == pytest.approx([0.1, 0.1, 0.1, 0.1])


def test_pacing_does_not_accumulate_drift():
    """A slow frame is followed by a shorter sleep, not by a later schedule.

    ``t0 + n/fps`` is absolute, so the sender catches up instead of falling
    behind by however long the previous frame took.
    """
    clock = FakeClock()
    sent_at = []

    def transport(data):
        sent_at.append(clock.now())
        clock.t += 0.03                      # this frame took 30 ms to send

    sender = make_sender(clock, transport, fps=10.0, frames=4)
    sender.run()      # the exit status is the status verdict, tested below
    assert sent_at == pytest.approx([0.0, 0.1, 0.2, 0.3])


def test_fps_zero_sends_unpaced():
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_sender(clock, transport, fps=0.0, frames=3)
    sender.run()      # the exit status is the status verdict, tested below
    assert transport.times == [0.0, 0.0, 0.0]
    assert clock.sleeps == []


def test_a_csv_line_every_thirty_frames():
    clock = FakeClock()
    transport = RecordingTransport(clock)
    out = io.StringIO()
    sender = make_sender(clock, transport, fps=0.0, frames=65, out=out)
    sender.run()      # the exit status is the status verdict, tested below
    lines = out.getvalue().strip().splitlines()
    assert lines[0] == vidsend.CSV_HEADER == "frame,elapsed_s,fps,bytes"
    # 30 and 60, not a line per frame and not 0 (nothing sent yet).  stdout
    # carries the table only; the prose (status, verdict) goes to stderr.
    assert [ln.split(",")[0] for ln in lines[1:]] == ["30", "60"]
    frame_col, elapsed_col, _fps, bytes_col = lines[1].split(",")
    assert int(frame_col) == 30 and float(elapsed_col) == 0.0
    assert int(bytes_col) > 0


# --- the wire form --------------------------------------------------------

def test_only_the_last_datagram_is_short_and_none_is_padded_mid_frame():
    """128 rows = 20,480 B: 15 datagrams at 1450, the last one alone short.

    Each datagram carries the 14-byte Ethernet header in front of its chunk,
    so a middle datagram is exactly CHUNK + 14 bytes; only the last one may be
    padded to the minimum and here it is already well over it.
    """
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_sender(clock, transport, rows=128, stretch=8, frames=1)
    sender.run()      # the exit status is the status verdict, tested below
    dgrams = transport.datagrams
    assert len(dgrams) == 15
    assert [len(d) for d in dgrams[:-1]] == [CHUNK + 14] * 14
    assert len(dgrams[-1]) == 14 + (HEADER_SIZE + 128 * ROW_BYTES - 14 * CHUNK)


def test_datagrams_reassemble_to_the_frame_the_encoder_built():
    """What the receiver gets back is byte for byte vidsrc's frame."""
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_sender(clock, transport, rows=8, stretch=128, frames=1)
    sender.run()      # the exit status is the status verdict, tested below
    body = b"".join(d[14:] for d in transport.datagrams)
    want = StripSource(8, 128).encode(0, vidsend.bars(8, 0))
    assert body == b"".join(want)
    h = unpack_header(body)
    assert h["crc_ok"] is True and h["kind"] == 1 and h["rows"] == 8


def test_a_datagram_carries_the_ethernet_header_and_padding():
    """The smallest valid strip (8 rows x 128) is 1296 bytes: one datagram."""
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_sender(clock, transport, rows=8, stretch=128, frames=1)
    sender.run()      # the exit status is the status verdict, tested below
    (d,) = transport.datagrams
    assert len(d) == 14 + HEADER_SIZE + 8 * ROW_BYTES
    assert d[:6] == BROADCAST and d[6:12] == HOST_MAC
    assert d[12:14] == struct.pack(">H", ETHERTYPE)


def test_a_short_last_datagram_is_padded_and_a_mid_one_is_not():
    """The padding rule, stated where it can bite: Ethernet needs 60 bytes.

    The smallest valid strip is 1296 bytes; chunked at 1255 the second (and
    last) datagram carries 41 bytes of frame, so it must be padded to 60 while
    the 1255-byte chunk before it must not be.  Padding a middle chunk would
    insert bytes into the frame body.
    """
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_sender(clock, transport, rows=8, stretch=128, frames=1,
                         chunk=1255)
    sender.run()      # the exit status is the status verdict, tested below
    sizes = [len(d) for d in transport.datagrams]
    assert sizes == [1255 + 14, 60]          # 1296 - 1255 = 41 bytes of frame
    frame = b"".join(d[14:] for d in transport.datagrams)
    assert frame[:1296] == b"".join(
        StripSource(8, 128).encode(0, vidsend.bars(8, 0)))
    assert frame[1296:] == bytes(5)          # the pad, and only at the end


def test_a_chunk_smaller_than_the_ring_can_hold_is_rejected():
    """A whole strip frame split over more than 16 datagrams can never arrive.

    ``--chunk 1000`` puts a 128-row frame (20,496 B) in 21 datagrams; the
    terminal's RX ring holds 16, so the payload would drop the frame.  The
    constructor refuses it and names the smallest legal chunk.
    """
    clock = FakeClock()
    transport = RecordingTransport(clock)
    with pytest.raises(ValueError) as e:
        make_sender(clock, transport, rows=128, stretch=8, chunk=1000)
    assert "at least 1281" in str(e.value)
    assert "16-descriptor" in str(e.value)


def test_a_chunk_that_fits_the_ring_is_accepted():
    """The boundary: 1,281 B * 16 = 20,496, exactly the 128-row frame."""
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_sender(clock, transport, rows=128, stretch=8, chunk=1281,
                         frames=1)
    sender.run()
    assert len(transport.datagrams) == 16


def test_frame_numbers_wrap_at_16_bits_and_bodies_differ_per_frame():
    """frame_no is a u16 field; the pattern must move or a stream is not one."""
    assert vidsend.bars(8, 0) != vidsend.bars(8, 1)
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_sender(clock, transport, fps=0.0, frames=2)
    sender.frame_offset = 0xFFFE            # next two frames straddle the wrap
    sender.run()      # the exit status is the status verdict, tested below
    nos = [unpack_header(d[14:])["frame_no"] for d in transport.datagrams]
    assert nos == [0xFFFE, 0xFFFF]


# --- the status path ------------------------------------------------------

def test_a_status_frame_is_decoded_from_an_ethernet_frame():
    d = vidsend.decode_status_frame(status_frame(frames_rx=200, csr0=0x20F2,
                                                 heap_free_kb=2048))
    assert d is not None
    assert d["kind"] == KIND_STATUS and d["crc_ok"] is True
    assert d["frames_rx"] == 200 and d["drops"] == 0 and d["crc_err"] == 0
    assert d["csr0"] == 0x20F2 and d["heap_free_kb"] == 2048


def test_a_strip_frame_is_not_mistaken_for_a_status():
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_sender(clock, transport, fps=0.0, frames=1)
    sender.run()      # the exit status is the status verdict, tested below
    assert vidsend.decode_status_frame(transport.datagrams[0]) is None


def test_a_status_frame_with_a_corrupt_header_is_flagged_not_decoded():
    bad = bytearray(status_frame(frames_rx=1))
    bad[14 + 8] ^= 0x01                     # rows byte, inside the CRC prefix
    d = vidsend.decode_status_frame(bytes(bad))
    assert d is not None and d["crc_ok"] is False


def test_a_status_in_a_foreign_ethertype_is_ignored():
    body = pack_status(1, 0, 0, 0, 0)
    hdr = pack_header(KIND_STATUS, 0, len(body), 0, 0)
    data = (BROADCAST + TERMINAL_MAC + struct.pack(">H", 0x88B6) + hdr + body)
    assert vidsend.decode_status_frame(data) is None


# --- the verdict ----------------------------------------------------------

def run_with_statuses(*status_bytes, frames=3):
    clock = FakeClock()
    transport = RecordingTransport(clock)
    pending = list(status_bytes)

    def recv():
        out, pending[:] = list(pending), []
        return out

    out, err = io.StringIO(), io.StringIO()
    sender = vidsend.Sender(transport=transport, receive=recv, src=HOST_MAC,
                            clock=clock.now, sleep=clock.sleep, frames=frames,
                            fps=0.0, status_wait=0.0, rows=8, stretch=128,
                            out=out, err=err)
    code = sender.run()
    return sender, code, out.getvalue() + err.getvalue()


def test_exit_zero_only_when_drops_and_crc_err_are_clean():
    _, code, text = run_with_statuses(status_frame(frames_rx=3))
    assert code == 0
    assert "drops=0" in text and "crc_err=0" in text

    _, code, _ = run_with_statuses(status_frame(frames_rx=3, drops=1))
    assert code == 1

    _, code, _ = run_with_statuses(status_frame(frames_rx=3, crc_err=2))
    assert code == 1


def test_exit_is_non_zero_when_no_status_arrives():
    """No status means the two conditions were never observed, so no PASS."""
    _, code, text = run_with_statuses()
    assert code == 1
    assert "no status" in text.lower()


def test_a_status_with_a_bad_header_crc_does_not_count():
    bad = bytearray(status_frame(frames_rx=3))
    bad[14 + 3] ^= 0x01                     # kind byte, inside the CRC prefix
    _, code, text = run_with_statuses(bytes(bad))
    assert code == 1
    assert "crc" in text.lower()


# --- the whole loop, with a fake counterparty -----------------------------

class FakeTerminal:
    """Reassembles the sender's datagrams and answers with status frames.

    This is the sender's counterparty, not a VT1200 model: it counts what
    arrived and every 64 accepted frames hands back the status frame the
    payload would send.  It exists to exercise vidsend.py's loop, not to stand
    in for the terminal -- the terminal's own status emission is checked in
    MAME (VT1200_MODE=stream) and decoded by vidstream_status_check.py.
    """

    def __init__(self, every=64, drops=0, crc_err=0):
        self.every = every
        self.drops = drops
        self.crc_err = crc_err
        self.buf = b""
        self.frames_rx = 0
        self.reported = []
        self.pending = []

    def send(self, data):
        if data[12:14] != struct.pack(">H", ETHERTYPE):
            return
        self.buf += data[14:]
        if len(self.buf) < HEADER_SIZE:
            return
        plen = struct.unpack_from("<H", self.buf, 6)[0]
        if len(self.buf) < HEADER_SIZE + plen:
            return
        frame, self.buf = self.buf[:HEADER_SIZE + plen], b""
        h = unpack_header(frame)
        assert h["crc_ok"], "the sender built a frame the parser rejects"
        self.frames_rx += 1
        if self.frames_rx % self.every == 0:
            self.reported.append(self.frames_rx)
            self.pending.append(status_frame(
                frames_rx=self.frames_rx, drops=self.drops,
                crc_err=self.crc_err, frame_no=len(self.reported)))

    def recv(self):
        out, self.pending = self.pending, []
        return out


def test_two_hundred_frames_get_three_periodic_statuses_and_a_clean_verdict():
    """The brief's run on the layer below the socket.

    200 frames at 30 fps, a status every 64 accepted: 64, 128, 192, each with
    drops=0 and crc_err=0, so the verdict is PASS.  The brief's final reading,
    ``frames_rx == 200``, needs one more thing the sender cannot assume: the
    terminal only reports on its 64th frame, so the last report is 192 unless
    the terminal is told the stream ended -- in MAME that is the RX_EOF cell,
    see vidstream.lua and the task-6 report.
    """
    clock = FakeClock()
    term = FakeTerminal()
    out, err = io.StringIO(), io.StringIO()
    sender = vidsend.Sender(transport=term.send, receive=term.recv,
                            src=HOST_MAC, clock=clock.now,
                            sleep=clock.sleep, rows=8, stretch=128,
                            frames=200, fps=30.0, status_wait=0.0, out=out,
                            err=err)
    code = sender.run()
    assert code == 0
    assert term.frames_rx == 200
    assert term.reported == [64, 128, 192]
    assert [s["frames_rx"] for s in sender.statuses] == [64, 128, 192]
    # The schedule: 200 frames at 30 fps take 199/30 s of virtual time.
    assert clock.now() == pytest.approx(199 / 30.0)
    assert "PASS" in err.getvalue()


def test_a_terminal_reporting_drops_makes_the_verdict_fail():
    """drops is the terminal's number, not something the sender invents."""
    clock = FakeClock()
    term = FakeTerminal(drops=1)
    out, err = io.StringIO(), io.StringIO()
    sender = vidsend.Sender(transport=term.send, receive=term.recv,
                            src=HOST_MAC, clock=clock.now,
                            sleep=clock.sleep, rows=8, stretch=128,
                            frames=64, fps=0.0, status_wait=0.0, out=out,
                            err=err)
    assert sender.run() == 1
    assert "drops=1" in err.getvalue()
    assert "FAIL" in err.getvalue()


# --- tile mode: the cache is the stream ------------------------------------
#
# ``--tiles`` sends what ``make_tiles.py`` cached: the 32 KB dictionary in
# chunks, then one frame per clip frame -- a TILES_FULL keyframe on the cadence
# (and whenever the terminal's cumulative ``drops`` rises), a TILES_DELTA
# otherwise.  Nothing here touches a socket; the counterparty is the transport
# callable, exactly as for the strip tests above.

@pytest.fixture(scope="module")
def tile_cache(tmp_path_factory):
    """A three-frame cache: enough for a keyframe and two deltas."""
    directory = tmp_path_factory.mktemp("tiles")
    tiles = make_tiles.build(make_tiles.synthetic_clip(3))
    make_tiles.write_cache(directory, tiles, clip="synthetic", every=60,
                           seed=0)
    return make_tiles.read_cache(directory)


def make_tile_sender(cache, clock, transport, **kw):
    kw.setdefault("fps", 0.0)
    kw.setdefault("status_wait", 0.0)
    kw.setdefault("out", io.StringIO())
    kw.setdefault("err", io.StringIO())
    kw.setdefault("src", HOST_MAC)
    return vidsend.TileSender(cache=cache, transport=transport,
                              clock=clock.now, sleep=clock.sleep, **kw)


def tile_frames(datagrams):
    """The VS frames among ``datagrams``, reassembled (dictionary first)."""
    return [d[14:] for d in datagrams]


def test_tile_mode_rejects_a_chunk_too_small_for_a_full_screen_delta(tile_cache):
    """A 1,280-tile delta is 3,858 B; at 200 B/datagram that is 20 > 16."""
    clock = FakeClock()
    transport = RecordingTransport(clock)
    with pytest.raises(ValueError) as e:
        make_tile_sender(tile_cache, clock, transport, chunk=200)
    assert "at least 242" in str(e.value)
    assert "full-screen delta" in str(e.value)


def test_tile_mode_uploads_the_dictionary_before_the_first_frame(tile_cache):
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_tile_sender(tile_cache, clock, transport, frames=1)
    sender.run()
    chunks = []
    for data in transport.datagrams:
        body = data[14:]
        h = unpack_header(body)
        if h["kind"] != KIND_DICT_CHUNK:
            break
        chunks.append(body)
    assert len(chunks) == 23, "32 KB at 1,448 body bytes is 23 chunks"
    offsets = [struct.unpack_from("<H", c, HEADER_SIZE)[0] for c in chunks]
    assert offsets[0] == 0
    assert offsets == sorted(offsets) and len(set(offsets)) == len(offsets)
    rebuilt = b"".join(c[HEADER_SIZE + 2:] for c in chunks)
    assert rebuilt == tile_cache.dictionary


def test_tile_mode_sends_the_cached_plan_on_the_wire(tile_cache):
    clock = FakeClock()
    transport = RecordingTransport(clock)
    sender = make_tile_sender(tile_cache, clock, transport, frames=3)
    sender.run()
    # The dictionary goes first (23 chunks); the clip's frames follow, one
    # datagram each at the default chunk size.
    frames = [d[14:] for d in transport.datagrams[23:]]
    assert len(frames) == len(tile_cache.records)
    for frame, record in zip(frames, tile_cache.records):
        h = unpack_header(frame)
        assert h["crc_ok"] and h["kind"] == record.kind
        assert h["payload_len"] == len(record.payload)
        assert frame[HEADER_SIZE:] == record.payload
        assert h["rows"] == 0 and h["stretch"] == 0


def test_tile_mode_forces_a_keyframe_when_drops_rise(tile_cache):
    """Carried from Task 5: the cumulative drops must reach keyframe_due."""
    clock = FakeClock()
    transport = RecordingTransport(clock)
    seen = []

    def receive():
        # A status arrives after the second frame, reporting one drop.
        if len(seen) == 2:
            return [status_frame(frames_rx=2, drops=1)]
        return []

    sender = make_tile_sender(tile_cache, clock, transport, frames=3,
                              receive=receive)
    # Count the frames the sender builds, before the transport sees them.
    original = sender.frame_plan
    def counting(n):
        kind, payload, _changed, _keyframe = original(n)
        seen.append(kind)
        return kind, payload, _changed, _keyframe
    sender.frame_plan = counting
    sender.run()
    assert seen == [make_tiles.KIND_FULL, make_tiles.KIND_DELTA,
                    make_tiles.KIND_FULL]
    assert sender.keyframes == [0, 2]
    assert sender.seen_drops == 1


def test_tile_mode_keeps_the_cadence_without_drops(tile_cache):
    clock = FakeClock()
    transport = RecordingTransport(clock)
    kinds = []
    sender = make_tile_sender(tile_cache, clock, transport, frames=3, every=2)
    original = sender.frame_plan
    def counting(n):
        kind, payload, _changed, _keyframe = original(n)
        kinds.append(kind)
        return kind, payload, _changed, _keyframe
    sender.frame_plan = counting
    sender.run()
    assert kinds == [make_tiles.KIND_FULL, make_tiles.KIND_DELTA,
                     make_tiles.KIND_FULL]


def test_tile_mode_prints_one_csv_row_per_frame(tile_cache):
    clock = FakeClock()
    transport = RecordingTransport(clock)
    out = io.StringIO()
    sender = make_tile_sender(tile_cache, clock, transport, frames=3, out=out)
    sender.run()
    lines = out.getvalue().strip().splitlines()
    assert lines[0] == vidsend.CSV_HEADER_TILES
    assert [ln.split(",")[0] for ln in lines[1:]] == ["0", "1", "2"]
    row = lines[2].split(",")
    assert row[4] == "delta"
    assert int(row[5]) == len(tile_cache.records[1].payload)
    assert int(row[6]) == tile_cache.records[1].changed
    assert row[7] == "0"


def test_tile_mode_verdict_needs_a_clean_status(tile_cache):
    clock = FakeClock()
    transport = RecordingTransport(clock)
    err = io.StringIO()
    sender = make_tile_sender(tile_cache, clock, transport, frames=1, err=err)
    assert sender.run() == 1
    assert "no status frame received" in err.getvalue()

    clock = FakeClock()
    transport = RecordingTransport(clock)
    out, err = io.StringIO(), io.StringIO()
    sender = make_tile_sender(tile_cache, clock, transport, frames=1, out=out,
                              err=err, receive=lambda: [status_frame(
                                  frames_rx=1, drops=0, crc_err=0)])
    assert sender.run() == 0
    assert "PASS" in err.getvalue()