path: root/debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py

"""Host-traffic message-loop transport tests (sim/host_traffic.py, docs/51).

All fixtures are synthetic and fully offline: no hardware, no kernel module,
no QEMU, no guest, no network.  The tests drive the modelled doorbell, the
msgq rings, the queue-element framing and the stub dispatcher against a small
in-process Machine (or a bare region) and assert framing, ordering and
determinism only.  The dispatcher handlers are stubs: nothing here claims
firmware behaviour.

Run::

    python3 -m unittest tests.test_host_traffic -v
"""
from __future__ import annotations

import json
import os
import sys
import tempfile
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
if TOOL_ROOT not in sys.path:
    sys.path.insert(0, TOOL_ROOT)

from greeneye.record import Recorder, load_events  # noqa: E402
from sim.devices import DeviceRegistry, FalconBlock  # noqa: E402
from sim.host_traffic import (  # noqa: E402
    DEMO_FORMAT,
    DOORBELL_BASE,
    DOORBELL_SIZE,
    GSP_MSG_QUEUE_ELEMENT_HDR_SIZE,
    GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
    GSP_RM_ALLOC,
    GSP_RM_CONTROL,
    MSGQ_FLAGS_SWAP_RX,
    MSGQ_ENTRY_OFF,
    MSGQ_RX_HDR_OFF,
    NV_ERR_NOT_SUPPORTED,
    NV_OK,
    REPLAY_FORMAT,
    RPC_FN_ECHO,
    RPC_FN_STATUS_QUERY,
    RPC_HEADER_BYTES,
    RPC_HEADER_VERSION,
    RPC_SIGNATURE,
    SHARED_MEM_BASE,
    STATUS_MAGIC,
    HostTraffic,
    HostTrafficError,
    MsgqQueue,
    RegionMemory,
    SharedQueues,
    build_queue_element,
    build_rpc_message,
    compute_message_checksum,
    load_stimulus,
    main,
    parse_queue_element,
    parse_rpc_message,
    replay_stimulus,
    run_demo,
)
from sim.machine import R_EBREAK, Machine  # noqa: E402

FALCON_BASE = 0x2000000000110000
FALCON_DEBUGINFO = 0x94

# lui t0,0x20000; slli t0,t0,32; lui t1,0x111; addi t1,t1,-0x400;
# add t0,t0,t1; lw a0,0(t0); ebreak  -- read doorbell queue 0.
PROGRAM_READ_DOORBELL = bytes.fromhex(
    "b70200209392020237131100130303c0b382620003a5020073001000"
)
# lui t0,0x20000; slli t0,t0,32; lui t1,0x111; addi t1,t1,-0x3f0;
# add t0,t0,t1; sw a1,0(t0); ebreak  -- write doorbell queue 2.
PROGRAM_WRITE_DOORBELL = bytes.fromhex(
    "b70200209392020237131100130303c1b382620023a0b20073001000"
)


def _sample_capture_log() -> str:
    payload = (
        (0xC2000006).to_bytes(4, "little")
        + (0xABCD0080).to_bytes(4, "little")
        + (0x2080012B).to_bytes(4, "little")
        + bytes(4)
    )
    return "\n".join(
        [
            "[ 1.0] NVRM: GE-GOLDEN-BEGIN",
            "[ 1.1] NVRM: GE-CTRL cmd=0x2080012b hClient=0xc2000006 "
            "hObject=0xabcd0080 size=0x8 paramp=0x0 ret=0x0",
            "[ 1.1] NVRM: GE-CTRL 000:" + "".join(f" {b:02x}" for b in bytes(8)),
            "[ 1.2] NVRM: GE-SEND fn=0x0000004c len=0x00000028 ret=0x0",
            "[ 1.2] NVRM: GE-SEND 000:" + "".join(f" {b:02x}" for b in payload),
            "[ 1.3] NVRM: GE-GOLDEN-END status=0x0",
        ]
    ) + "\n"


class FramingTests(unittest.TestCase):
    def test_rpc_constants_match_capture_tool(self):
        from harness.rm_capture import (
            GSP_RM_ALLOC as capture_alloc,
            GSP_RM_CONTROL as capture_control,
            RPC_HEADER_BYTES as capture_header_bytes,
        )

        self.assertEqual(RPC_HEADER_BYTES, capture_header_bytes)
        self.assertEqual(GSP_RM_CONTROL, capture_control)
        self.assertEqual(GSP_RM_ALLOC, capture_alloc)
        self.assertEqual(RPC_HEADER_BYTES, 32)

    def test_rpc_message_roundtrip(self):
        params = bytes(range(64))
        raw = build_rpc_message(RPC_FN_ECHO, params, sequence=7)
        self.assertEqual(len(raw), RPC_HEADER_BYTES + len(params))
        message = parse_rpc_message(raw)
        self.assertEqual(message.header_version, RPC_HEADER_VERSION)
        self.assertEqual(message.signature, RPC_SIGNATURE)
        self.assertEqual(message.function, RPC_FN_ECHO)
        self.assertEqual(message.sequence, 7)
        self.assertEqual(message.rpc_result, NV_OK)
        self.assertEqual(message.length, len(raw))
        self.assertEqual(message.params, params)
        self.assertEqual(message.to_bytes(), raw)

    def test_rpc_bad_signature_is_refused(self):
        raw = bytearray(build_rpc_message(RPC_FN_ECHO, b"\x01", sequence=1))
        raw[4:8] = (0xDEADBEEF).to_bytes(4, "little")
        with self.assertRaises(HostTrafficError):
            parse_rpc_message(bytes(raw))

    def test_rpc_short_and_overlong_length_is_refused(self):
        raw = build_rpc_message(RPC_FN_ECHO, b"\x01\x02", sequence=1)
        with self.assertRaises(HostTrafficError):
            parse_rpc_message(raw[: RPC_HEADER_BYTES - 1])
        overlong = bytearray(raw)
        overlong[8:12] = (len(raw) + 16).to_bytes(4, "little")
        with self.assertRaises(HostTrafficError):
            parse_rpc_message(bytes(overlong))

    def test_queue_element_layout(self):
        params = bytes(range(16))
        rpc = build_rpc_message(RPC_FN_ECHO, params, sequence=3)
        element = build_queue_element(rpc, sequence=9)
        self.assertEqual(GSP_MSG_QUEUE_ELEMENT_HDR_SIZE, 0x30)
        self.assertEqual(element[0x30:0x34], rpc[:4])
        self.assertEqual(element[0x38:0x3C], len(rpc).to_bytes(4, "little"))
        self.assertEqual(element[0x34:0x38], RPC_SIGNATURE.to_bytes(4, "little"))
        self.assertEqual(element[0x24:0x28], (9).to_bytes(4, "little"))
        self.assertEqual(element[0x28:0x2C], (1).to_bytes(4, "little"))
        self.assertEqual(element[0x2C:0x30], bytes(4))
        self.assertEqual(element[0x50:], params)
        prefix, parsed = parse_queue_element(element)
        self.assertEqual(prefix["seq_num"], 9)
        self.assertEqual(prefix["elem_count"], 1)
        self.assertEqual(parsed.function, RPC_FN_ECHO)
        self.assertEqual(parsed.params, params)

    def test_queue_element_checksum_verifies_and_detects_tamper(self):
        rpc = build_rpc_message(RPC_FN_ECHO, bytes(range(32)), sequence=1)
        element = build_queue_element(rpc, sequence=1)
        self.assertEqual(compute_message_checksum(element), 0)
        tampered = bytearray(element)
        tampered[0x50] ^= 0x01
        self.assertNotEqual(compute_message_checksum(bytes(tampered)), 0)
        with self.assertRaises(HostTrafficError):
            parse_queue_element(bytes(tampered))

    def test_multi_element_message_roundtrip(self):
        params = bytes((i * 7) & 0xFF for i in range(5000))
        rpc = build_rpc_message(RPC_FN_ECHO, params, sequence=1)
        element = build_queue_element(rpc, sequence=1)
        self.assertEqual(element[0x28:0x2C], (2).to_bytes(4, "little"))
        prefix, parsed = parse_queue_element(element)
        self.assertEqual(prefix["elem_count"], 2)
        self.assertEqual(parsed.params, params)
        host = HostTraffic()
        host.submit_rpc(RPC_FN_ECHO, params, sequence=1)
        self.assertEqual(host.poll(), 1)
        reply = host.read_reply()
        self.assertEqual(reply["params"], params)
        self.assertEqual(host.queues.status.header().write_ptr, 2)
        host.close()


class MsgqTests(unittest.TestCase):
    def test_queue_header_layout_matches_msgq_arithmetic(self):
        mem = RegionMemory(SHARED_MEM_BASE, 0x40000)
        queue = MsgqQueue(mem, SHARED_MEM_BASE, 0x40000)
        queue.format()
        header = queue.header()
        self.assertEqual(header.version, 0)
        self.assertEqual(header.size, 0x40000)
        self.assertEqual(header.msg_size, GSP_MSG_QUEUE_ELEMENT_SIZE_MIN)
        self.assertEqual(header.msg_count, (0x40000 - MSGQ_ENTRY_OFF) // 0x1000)
        self.assertEqual(header.write_ptr, 0)
        self.assertEqual(header.flags, MSGQ_FLAGS_SWAP_RX)
        self.assertEqual(header.rx_hdr_off, MSGQ_RX_HDR_OFF)
        self.assertEqual(header.entry_off, MSGQ_ENTRY_OFF)
        self.assertEqual(MSGQ_RX_HDR_OFF, 32)
        self.assertEqual(MSGQ_ENTRY_OFF, 48)

    def test_write_publish_and_read_back_entry(self):
        mem = RegionMemory(SHARED_MEM_BASE, 0x40000)
        queue = MsgqQueue(mem, SHARED_MEM_BASE, 0x40000)
        queue.format()
        payload = bytes(range(256))
        queue.write_entry(0, payload)
        self.assertEqual(queue.read_entry(0, 256), payload)
        queue.publish(1)
        self.assertEqual(queue.write_ptr, 1)
        queue.set_read(1)
        self.assertEqual(queue.read_ptr, 1)

    def test_shared_queues_cross_link_rx_headers(self):
        queues = SharedQueues(RegionMemory(SHARED_MEM_BASE, 0x80000))
        queues.format()
        self.assertEqual(queues.command.rx_hdr_addr, queues.status.tx_base + MSGQ_RX_HDR_OFF)
        self.assertEqual(queues.status.rx_hdr_addr, queues.command.tx_base + MSGQ_RX_HDR_OFF)
        queues.command.set_read(5)
        raw = queues.mem.read(queues.status.tx_base + MSGQ_RX_HDR_OFF, 4)
        self.assertEqual(int.from_bytes(raw, "little"), 5)
        queues.status.set_read(2)
        raw = queues.mem.read(queues.command.tx_base + MSGQ_RX_HDR_OFF, 4)
        self.assertEqual(int.from_bytes(raw, "little"), 2)

    def test_submit_and_take_roundtrip(self):
        queues = SharedQueues(RegionMemory(SHARED_MEM_BASE, 0x80000))
        queues.format()
        element = build_queue_element(
            build_rpc_message(RPC_FN_ECHO, b"\xaa\xbb", sequence=4), sequence=0
        )
        first = queues.submit_element(element)
        self.assertEqual(first["index"], 0)
        self.assertEqual(queues.command.write_ptr, 1)
        taken = queues.take_command()
        self.assertIsNotNone(taken)
        self.assertEqual(taken["element"], element)
        self.assertEqual(queues.take_command(), None)
        published = queues.publish_element(element)
        self.assertEqual(published["index"], 0)
        status = queues.take_status()
        self.assertEqual(status["element"], element)
        self.assertEqual(queues.take_status(), None)


class DoorbellTests(unittest.TestCase):
    def test_model_records_writes(self):
        host = HostTraffic()
        host.doorbell.ring(0, 0)
        host.doorbell.ring(0, 0x11)
        host.doorbell.ring(2, 0x22)
        self.assertEqual(host.doorbell.head(0), 0x11)
        self.assertEqual(host.doorbell.head(2), 0x22)
        self.assertEqual(host.doorbell.kicks, 3)
        state = host.doorbell.state()
        self.assertEqual(len(state["writes"]), 3)
        self.assertEqual(state["writes"][-1]["queue"], 2)
        with self.assertRaises(ValueError):
            host.doorbell.ring(8, 0)
        host.close()

    def test_block_serves_guest_read(self):
        host = HostTraffic()
        host.doorbell.ring(1, 0xCAFEF00D)
        block = host.doorbell_block()
        self.assertEqual(block.read(0, 4), 0)
        self.assertEqual(block.read(8, 4), 0xCAFEF00D)
        self.assertIsNone(block.read(8 * 8, 4))
        self.assertEqual(block.base, DOORBELL_BASE)
        self.assertEqual(block.size, DOORBELL_SIZE)
        host.close()

    def test_registry_overlay_beats_wider_falcon_window(self):
        registry = DeviceRegistry()
        registry.add(FalconBlock(), FALCON_BASE, 0x1000)
        host = HostTraffic()
        host.doorbell.ring(0, 0x1234)
        registry.add(host.doorbell_block())
        handled, value = registry.read(DOORBELL_BASE, 4)
        self.assertTrue(handled)
        self.assertEqual(value, 0x1234)
        handled, value = registry.read(FALCON_BASE + FALCON_DEBUGINFO, 4)
        self.assertTrue(handled)
        self.assertEqual(value, 0)
        host.close()

    def test_guest_load_reads_doorbell(self):
        machine = Machine(memory_mb=8, ram_base=0)
        machine.load_segments([(0, PROGRAM_READ_DOORBELL)], entry=0)
        host = HostTraffic()
        host.doorbell.ring(0, 0x5A5A5A5A)
        registry = DeviceRegistry()
        registry.add(host.doorbell_block())
        machine.devices = registry
        reason, _pc, _a, _b = machine.run(100)
        self.assertEqual(reason, R_EBREAK)
        self.assertEqual(machine.x[10], 0x5A5A5A5A)
        host.close()

    def test_guest_store_is_recorded(self):
        machine = Machine(memory_mb=8, ram_base=0)
        machine.load_segments([(0, PROGRAM_WRITE_DOORBELL)], entry=0)
        machine.apply_reset_regs({"a1": 0x77})
        host = HostTraffic()
        registry = DeviceRegistry()
        registry.add(host.doorbell_block())
        machine.devices = registry
        reason, _pc, _a, _b = machine.run(100)
        self.assertEqual(reason, R_EBREAK)
        self.assertEqual(host.doorbell.head(2), 0x77)
        host.close()


class TrafficTests(unittest.TestCase):
    def test_submit_writes_ring_and_rings_doorbell(self):
        host = HostTraffic()
        raw = build_rpc_message(RPC_FN_ECHO, b"\x01\x02\x03", sequence=5)
        result = host.submit(raw)
        self.assertEqual(result["function"], RPC_FN_ECHO)
        self.assertEqual(result["sequence"], 5)
        self.assertEqual(result["index"], 0)
        self.assertEqual(result["write_ptr"], 1)
        self.assertEqual(host.queues.command.write_ptr, 1)
        self.assertEqual(host.doorbell.kicks, 1)
        self.assertEqual(host.doorbell.head(0), 0)
        taken = host.queues.take_command()
        self.assertIsNotNone(taken)
        _prefix, parsed = parse_queue_element(taken["element"])
        self.assertEqual(parsed.params, b"\x01\x02\x03")
        host.close()

    def test_poll_dispatches_echo(self):
        host = HostTraffic()
        host.submit_rpc(RPC_FN_ECHO, b"\xaa\xbb\xcc", sequence=1)
        self.assertEqual(host.poll(), 1)
        self.assertEqual(host.poll(), 0)
        reply = host.read_reply()
        self.assertIsNotNone(reply)
        self.assertEqual(reply["function"], RPC_FN_ECHO)
        self.assertEqual(reply["sequence"], 1)
        self.assertEqual(reply["result"], NV_OK)
        self.assertEqual(reply["params"], b"\xaa\xbb\xcc")
        self.assertEqual(host.read_reply(), None)
        host.close()

    def test_unknown_function_gets_not_supported(self):
        host = HostTraffic()
        host.submit_rpc(0x999, b"\x01", sequence=2)
        self.assertEqual(host.poll(), 1)
        reply = host.read_reply()
        self.assertEqual(reply["function"], 0x999)
        self.assertEqual(reply["result"], NV_ERR_NOT_SUPPORTED)
        self.assertEqual(reply["params"], b"")
        dispatched = [
            event for event in host.transcript()["events"]
            if event["type"] == "dispatch"
        ]
        self.assertEqual(dispatched[0]["handler"], "stub:unsupported")
        host.close()

    def test_status_query_reply_is_deterministic(self):
        replies = []
        for _ in range(2):
            host = HostTraffic()
            host.submit_rpc(RPC_FN_ECHO, b"\x01", sequence=1)
            host.submit_rpc(RPC_FN_STATUS_QUERY, b"", sequence=2)
            host.poll()
            first = host.read_reply()
            second = host.read_reply()
            replies.append((first["params"], second["params"]))
            host.close()
        self.assertEqual(replies[0], replies[1])
        self.assertEqual(replies[0][1][:4], STATUS_MAGIC)

    def test_empty_poll_and_read_reply(self):
        host = HostTraffic()
        self.assertEqual(host.poll(), 0)
        self.assertIsNone(host.read_reply())
        host.close()

    def test_run_cycles_is_bounded(self):
        machine = Machine(memory_mb=32)
        host = HostTraffic(machine=machine)
        host.submit_rpc(RPC_FN_ECHO, b"\x01", sequence=1)
        result = host.run_cycles(64)
        self.assertEqual(result["budget"], 64)
        self.assertEqual(result["insns"], machine.insns)
        self.assertEqual(result["dispatched"], 1)
        self.assertEqual(result["replies"], 1)
        with self.assertRaises(ValueError):
            host.run_cycles(0)
        with self.assertRaises(ValueError):
            host.run_cycles(-1)
        host.close()

    def test_oversized_message_is_refused(self):
        host = HostTraffic()
        with self.assertRaises(HostTrafficError):
            host.submit_rpc(RPC_FN_ECHO, bytes(GSP_MSG_QUEUE_ELEMENT_SIZE_MIN * 16),
                            sequence=1)
        host.close()

    def test_transcript_is_deterministic(self):
        def run():
            host = HostTraffic()
            host.submit_rpc(RPC_FN_ECHO, b"\x10\x20", sequence=1)
            host.submit_rpc(RPC_FN_STATUS_QUERY, b"", sequence=2)
            host.submit_rpc(0x999, b"", sequence=3)
            host.poll()
            while host.read_reply() is not None:
                pass
            transcript = host.transcript()
            host.close()
            return transcript

        first, second = run(), run()
        self.assertEqual(first, second)
        self.assertEqual(first["format"], "greeneye-host-traffic-log/1")
        self.assertEqual(
            [event["type"] for event in first["events"]],
            ["submit", "submit", "submit", "dispatch", "dispatch", "dispatch",
             "reply", "reply", "reply"],
        )


class ReplayTests(unittest.TestCase):
    def test_replay_stimulus_in_order(self):
        stimulus = {
            "format": "greeneye-rm-rpc-stimulus/1",
            "messages": [
                {
                    "seq": 0,
                    "kind": "send",
                    "fn": RPC_FN_ECHO,
                    "params_hex": "0102",
                    "truncated": False,
                },
                {
                    "seq": 1,
                    "kind": "send",
                    "fn": RPC_FN_ECHO,
                    "params_hex": "0304",
                    "truncated": False,
                },
            ],
        }
        host = HostTraffic()
        report = replay_stimulus(host, stimulus)
        self.assertEqual(report["format"], REPLAY_FORMAT)
        self.assertEqual(report["submitted"], 2)
        self.assertEqual(report["skipped"], [])
        self.assertEqual([pair["reply"]["params"] for pair in report["pairs"]],
                         [b"\x01\x02", b"\x03\x04"])
        host.close()

    def test_replay_strict_rejects_truncated(self):
        stimulus = {
            "format": "greeneye-rm-rpc-stimulus/1",
            "messages": [
                {
                    "seq": 0,
                    "kind": "control",
                    "params_hex": "00",
                    "truncated": True,
                }
            ],
        }
        host = HostTraffic()
        with self.assertRaises(HostTrafficError):
            replay_stimulus(host, stimulus, strict=True)
        host.close()

    def test_replay_lenient_skips_truncated(self):
        stimulus = {
            "format": "greeneye-rm-rpc-stimulus/1",
            "messages": [
                {
                    "seq": 0,
                    "kind": "send",
                    "fn": RPC_FN_ECHO,
                    "params_hex": "01",
                    "truncated": False,
                },
                {
                    "seq": 1,
                    "kind": "control",
                    "params_hex": "00",
                    "truncated": True,
                },
            ],
        }
        host = HostTraffic()
        report = replay_stimulus(host, stimulus, strict=False)
        self.assertEqual(report["submitted"], 1)
        self.assertEqual(len(report["skipped"]), 1)
        self.assertIn("truncated", report["skipped"][0]["reason"])
        host.close()

    def test_control_stimulus_maps_to_control_function(self):
        stimulus = {
            "format": "greeneye-rm-rpc-stimulus/1",
            "messages": [
                {
                    "seq": 0,
                    "kind": "control",
                    "params_hex": "01020304",
                    "truncated": False,
                }
            ],
        }
        host = HostTraffic()
        report = replay_stimulus(host, stimulus)
        self.assertEqual(report["pairs"][0]["reply"]["function"], GSP_RM_CONTROL)
        self.assertEqual(report["pairs"][0]["reply"]["result"], NV_ERR_NOT_SUPPORTED)
        host.close()

    def test_load_stimulus_from_capture_document(self):
        from harness.rm_capture import (
            capture_to_dict,
            decode_text,
            stimulus_from_capture,
        )

        capture = decode_text(_sample_capture_log(), source="fixture")
        self.assertIsNotNone(stimulus_from_capture(capture))
        with tempfile.TemporaryDirectory() as tmp:
            capture_path = os.path.join(tmp, "capture.json")
            with open(capture_path, "w", encoding="utf-8") as fh:
                json.dump(capture_to_dict(capture), fh)
            loaded = load_stimulus(capture_path)
        self.assertEqual(loaded["format"], "greeneye-rm-rpc-stimulus/1")
        self.assertEqual(len(loaded["messages"]), 2)

    def test_load_stimulus_accepts_inline_document(self):
        stimulus = {"format": "greeneye-rm-rpc-stimulus/1", "messages": []}
        self.assertEqual(load_stimulus(stimulus), stimulus)
        with self.assertRaises(HostTrafficError):
            load_stimulus({"format": "not-a-stimulus"})


class SessionTests(unittest.TestCase):
    def test_recorder_events_and_blob(self):
        with tempfile.TemporaryDirectory() as tmp:
            recorder = Recorder(tmp, name="host-traffic-test", meta={"kind": "test"})
            host = HostTraffic(recorder=recorder)
            host.submit_rpc(RPC_FN_ECHO, b"\x01", sequence=1)
            host.poll()
            host.read_reply()
            host.close()
            recorder.close()
            events = load_events(tmp)
            types = [event["type"] for event in events]
            self.assertIn("host_traffic_config", types)
            self.assertIn("doorbell", types)
            self.assertIn("host_submit", types)
            self.assertIn("host_dispatch", types)
            self.assertIn("host_reply_read", types)
            blob_path = os.path.join(tmp, "blobs", "host-traffic.json")
            self.assertTrue(os.path.isfile(blob_path))
            with open(blob_path, "r", encoding="utf-8") as fh:
                payload = json.load(fh)
            self.assertEqual(payload["format"], "greeneye-host-traffic-log/1")


class DemoTests(unittest.TestCase):
    def test_demo_end_to_end_and_deterministic(self):
        first, second = run_demo(), run_demo()
        self.assertTrue(first["ok"], first.get("checks"))
        self.assertEqual(first["format"], DEMO_FORMAT)
        self.assertEqual(first["replies"], second["replies"])
        self.assertEqual(first["transcript"], second["transcript"])
        self.assertEqual(first["control"]["bytes"], 512)
        self.assertEqual(first["control"]["reply_result"], NV_ERR_NOT_SUPPORTED)
        self.assertTrue(all(reply["read"] for reply in first["replies"]))

    def test_demo_writes_artifacts(self):
        with tempfile.TemporaryDirectory() as tmp:
            result = run_demo(out_dir=tmp)
            self.assertTrue(result["ok"], result.get("checks"))
            self.assertTrue(os.path.isfile(os.path.join(tmp, "blobs", "host-traffic.json")))
            self.assertTrue(os.path.isfile(os.path.join(tmp, "host-traffic-demo.json")))
            self.assertTrue(os.path.isfile(os.path.join(tmp, "session.json")))

    def test_cli_demo(self):
        import contextlib
        import io

        with tempfile.TemporaryDirectory() as tmp:
            out = io.StringIO()
            with contextlib.redirect_stdout(out):
                code = main(["demo", "--out", tmp])
            self.assertEqual(code, 0)
            payload = json.loads(out.getvalue())
            self.assertEqual(payload["format"], DEMO_FORMAT)
            self.assertTrue(payload["ok"])

    def test_cli_replay(self):
        import contextlib
        import io

        stimulus = {
            "format": "greeneye-rm-rpc-stimulus/1",
            "messages": [
                {
                    "seq": 0,
                    "kind": "send",
                    "fn": RPC_FN_ECHO,
                    "params_hex": "0102",
                    "truncated": False,
                }
            ],
        }
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "stimulus.json")
            with open(path, "w", encoding="utf-8") as fh:
                json.dump(stimulus, fh)
            out = io.StringIO()
            with contextlib.redirect_stdout(out):
                code = main(["replay", "--stimulus", path, "--out", tmp])
            self.assertEqual(code, 0)
            payload = json.loads(out.getvalue())
            self.assertEqual(payload["format"], REPLAY_FORMAT)

    def test_cli_requires_a_subcommand(self):
        import contextlib
        import io

        err = io.StringIO()
        with contextlib.redirect_stderr(err):
            code = main([])
        self.assertEqual(code, 2)


if __name__ == "__main__":
    unittest.main()