diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py | 639 |
1 files changed, 639 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py b/debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py new file mode 100644 index 0000000..93024d2 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py @@ -0,0 +1,639 @@ +"""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() |