"""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()