"""libos port receive-path model tests (sim/libos_port.py, docs/51 section 7). Hermetic: a fake machine (register file only) and a fake base syscall model; no image, no boot, no machine. The tests pin the port-syscall ABI observed on the real 570.144 image (send args a0-a3, receive args a4-a7, wait id t1, timeout t2, return a0/a4/a5/a6), the async receive post, the boot handshake send, the host doorbell wake's external-interrupt handoff, and the strict "never complete an uncompleted receive" rule. Offline only: no hardware, kernel module, mdev, guest, network or QEMU. Run:: cd open-iov/debug/risc-v-gsp-rm-vm python3 -m unittest tests.test_libos_port -v """ from __future__ import annotations import json import os import sys 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 sim.libos import INTERRUPT_ARM_EXTERNAL, TASK_PORT_SEND_RECV_WAIT # noqa: E402 from sim.libos_port import ( # noqa: E402 EXTINTR_WAKE_CONTRACT, EXTERNAL_INTERRUPT_XCAUSE, HANDSHAKE_CONTRACT, NETWORK_REPORT_FORMAT, REPORT_FORMAT, SHUTTLE_ANY, SPURIOUS_XEPC, TIMEOUT_INFINITE, LibosPortModel, LibosPortNetwork, role_for, ) class FakeMachine: """Register file only: the port model never touches memory.""" def __init__(self): self.x = [0] * 32 self.insns = 0 self.pc = 0 def set(self, index: int, value: int) -> None: self.x[index] = value & 0xFFFFFFFFFFFFFFFF class FakeMemoryMachine(FakeMachine): """Register file plus flat memory for the routing model tests.""" def __init__(self, size: int = 0x10000): super().__init__() self.ram = bytearray(size) self.ram_base = 0 self.ram_size = size def read_mem(self, addr: int, size: int) -> bytes: if addr < 0 or addr + size > self.ram_size: raise ValueError("out of range") return bytes(self.ram[addr:addr + size]) def write_mem(self, addr: int, data: bytes) -> None: if addr < 0 or addr + len(data) > self.ram_size: raise ValueError("out of range") self.ram[addr:addr + len(data)] = data class FakeBase: """Minimal base syscall model: handles arm, records and stops the rest.""" name = "fake-libos" def __init__(self): self.calls: list[int] = [] self.armed_mask = 0 self.stopped: dict | None = None def handle(self, machine, pc: int = 0): number = int(machine.x[5]) if number == INTERRUPT_ARM_EXTERNAL: self.armed_mask |= int(machine.x[10]) return True, 0 self.calls.append(number) self.stopped = {"num": number, "name": f"sys_{number}", "pc": int(pc)} return False, None def _port_call(machine, *, send=None, recv=None, wait=None, timeout=TIMEOUT_INFINITE) -> None: """Write one ``TASK_PORT_SEND_RECV_WAIT`` argument set (explicit fields).""" machine.set(5, TASK_PORT_SEND_RECV_WAIT) # t0 for index, value in enumerate(send or (0, 0, 0, 0), start=10): machine.set(index, value) for index, value in enumerate(recv or (0, 0, 0, 0), start=14): machine.set(index, value) machine.set(6, 0 if wait is None else wait) # t1 machine.set(7, timeout) # t2 class PortSurfaceTests(unittest.TestCase): """Role table and argument contract.""" def test_observed_roles(self): self.assertEqual(role_for(6), "extintr") self.assertEqual(role_for(7), "notification") self.assertEqual(role_for(8), "request") self.assertEqual(role_for(12), "handshake") self.assertEqual(role_for(13), "event") self.assertEqual(role_for(99), "data") def test_async_post_keeps_receive_args_and_returns_ok(self): model = LibosPortModel(base=FakeBase()) machine = FakeMachine() _port_call(machine, recv=(10, 6, 0, 0)) handled, result = model.handle(machine, pc=0x4018038) self.assertTrue(handled) self.assertEqual(result, 0) self.assertEqual(len(model.posts), 1) self.assertEqual(model.posts[0]["shuttle"], 10) self.assertEqual(model.posts[0]["role"], "extintr") # The kernel returns LIBOS_OK through a0; the wrapper masks a0 only. self.assertEqual(machine.x[14], 10) self.assertEqual(machine.x[15], 6) def test_send_completes_with_the_transcribe_contract(self): model = LibosPortModel(base=FakeBase()) machine = FakeMachine() _port_call(machine, send=(1, 12, 0, 0, ), wait=1) handled, result = model.handle(machine, pc=0x4019CD4) self.assertTrue(handled) self.assertEqual(result, 0) self.assertEqual(len(model.sends), 1) self.assertEqual(model.sends[0]["contract"], HANDSHAKE_CONTRACT) self.assertEqual(model.sends[0]["role"], "handshake") self.assertEqual(machine.x[14], 0) # completed size self.assertEqual(machine.x[15], 0) # originating task self.assertEqual(machine.x[16], 1) # completed shuttle self.assertEqual(machine.x[17], 0) def test_blocking_wait_stops_without_completion(self): model = LibosPortModel(base=FakeBase()) machine = FakeMachine() _port_call(machine, recv=(9, 8, 0x4015070, 0x808)) model.handle(machine, pc=0x4018038) _port_call(machine, wait=SHUTTLE_ANY) handled, result = model.handle(machine, pc=0x4019DF4) self.assertFalse(handled) self.assertIsNone(result) self.assertIsNotNone(model.stopped) self.assertEqual(model.stopped["reason"], "port-wait-blocks") self.assertEqual(model.stopped["num"], TASK_PORT_SEND_RECV_WAIT) self.assertEqual(model.waits[-1]["reason"], "port-wait-blocks") def test_unmodelled_syscall_is_recorded_and_stopped(self): base = FakeBase() model = LibosPortModel(base=base) machine = FakeMachine() machine.set(5, 6) # task_state_query machine.set(10, 0) handled, result = model.handle(machine, pc=0x401A46C) self.assertFalse(handled) self.assertIsNone(result) self.assertEqual(model.stopped["num"], 6) self.assertEqual(model.stopped["reason"], "unmodelled-syscall") self.assertEqual(base.calls, [6]) class DoorbellWakeTests(unittest.TestCase): """The host wake's external-interrupt handoff (RTOS report contract).""" def test_kick_before_post_and_arm_is_delivered_at_post(self): base = FakeBase() model = LibosPortModel(base=base) machine = FakeMachine() event = model.host_kick(queue=0, value=0) self.assertIsNone(event["delivered"]) self.assertTrue(event["pending"]) # Arm first, then post the extintr receive: the pending wake fills it. machine.set(5, INTERRUPT_ARM_EXTERNAL) machine.set(10, 0xFF) model.handle(machine, pc=0x4018054) _port_call(machine, recv=(10, 6, 0, 0)) model.handle(machine, pc=0x4018038) self.assertEqual(model.kicks_pending, 0) self.assertEqual([entry["shuttle"] for entry in model.completed], [10]) self.assertEqual(model.completed[0]["payload_size"], 0) self.assertEqual(model.completed[0]["origin_task"], 0) self.assertEqual(model.completed[0]["contract"], EXTINTR_WAKE_CONTRACT) def test_wake_requires_an_arm(self): model = LibosPortModel(base=FakeBase()) machine = FakeMachine() _port_call(machine, recv=(10, 6, 0, 0)) model.handle(machine, pc=0x4018038) event = model.host_kick() self.assertIsNone(event["delivered"]) self.assertEqual(model.kicks_pending, 1) self.assertEqual(model.completed, []) # Without interrupts armed the wait still blocks (no fabricated wake). _port_call(machine, wait=SHUTTLE_ANY) handled, _ = model.handle(machine, pc=0x4019DF4) self.assertFalse(handled) def test_arm_seeds_from_the_attached_base_model(self): base = FakeBase() base.armed_mask = 0xFF model = LibosPortModel(base=base) self.assertEqual(model.armed_mask, 0xFF) machine = FakeMachine() model.host_kick() _port_call(machine, recv=(10, 6, 0, 0)) model.handle(machine, pc=0x4018038) self.assertEqual(model.kicks_pending, 0) self.assertEqual([entry["shuttle"] for entry in model.completed], [10]) def test_wait_returns_the_wake_with_the_argument_contract(self): model = LibosPortModel(base=FakeBase()) machine = FakeMachine() model.host_kick() _port_call(machine, recv=(10, 6, 0, 0)) model.handle(machine, pc=0x4018038) machine.set(5, INTERRUPT_ARM_EXTERNAL) machine.set(10, 0xFF) model.handle(machine, pc=0x4018054) _port_call(machine, wait=SHUTTLE_ANY) handled, result = model.handle(machine, pc=0x4019DF4) self.assertTrue(handled) self.assertEqual(result, 0) self.assertEqual(machine.x[14], 0) # completed size self.assertEqual(machine.x[15], 0) # originating task self.assertEqual(machine.x[16], 10) # completed shuttle self.assertEqual(model.waits[-1]["completed"], 10) self.assertEqual(model.waits[-1]["role"], "extintr") def test_one_wake_completes_one_receive_only(self): model = LibosPortModel(base=FakeBase()) machine = FakeMachine() model.host_kick() _port_call(machine, recv=(10, 6, 0, 0)) model.handle(machine, pc=0x4018038) machine.set(5, INTERRUPT_ARM_EXTERNAL) machine.set(10, 0xFF) model.handle(machine, pc=0x4018054) self.assertEqual(len(model.completed), 1) # Consume the wake, re-post and wait again: nothing completes. _port_call(machine, wait=SHUTTLE_ANY) handled, _ = model.handle(machine, pc=0x4019DF4) self.assertTrue(handled) _port_call(machine, recv=(10, 6, 0, 0)) model.handle(machine, pc=0x4018C8) _port_call(machine, wait=SHUTTLE_ANY) handled, _ = model.handle(machine, pc=0x4019DF4) self.assertFalse(handled) self.assertEqual(model.completed, []) class StateTests(unittest.TestCase): """Deterministic, serialisable report state.""" def _run(self) -> dict: model = LibosPortModel(base=FakeBase()) machine = FakeMachine() model.host_kick(queue=0, value=0) _port_call(machine, recv=(10, 6, 0, 0)) model.handle(machine, pc=0x4018038) machine.set(5, INTERRUPT_ARM_EXTERNAL) machine.set(10, 0xFF) model.handle(machine, pc=0x4018054) _port_call(machine, recv=(9, 8, 0x4015070, 0x808)) model.handle(machine, pc=0x4018038) _port_call(machine, wait=SHUTTLE_ANY) model.handle(machine, pc=0x4019DF4) return model.state() def test_state_is_byte_identical_across_runs(self): first = self._run() second = self._run() self.assertEqual(first["format"], REPORT_FORMAT) self.assertEqual(json.dumps(first, sort_keys=True), json.dumps(second, sort_keys=True)) self.assertEqual(first["armed_mask"], 0xFF) self.assertEqual(first["host_kicks"], 1) self.assertEqual(first["kicks_pending"], 0) self.assertEqual([post["role"] for post in first["posts"]], ["extintr", "request"]) self.assertEqual(first["completed"], []) self.assertEqual(first["waits"][-1]["completed"], 10) def test_state_has_no_host_timestamps(self): state = self._run() def keys(value): if isinstance(value, dict): for key, item in value.items(): yield key yield from keys(item) elif isinstance(value, list): for item in value: yield from keys(item) for key in keys(state): self.assertNotIn(str(key), ("t", "ts", "timestamp", "time")) class SyncPortTests(unittest.TestCase): """Synchronous receive semantics (waitId != 0 must block, not fake OK).""" def test_sync_recv_without_completion_blocks(self): model = LibosPortModel(base=FakeBase()) machine = FakeMachine() _port_call(machine, recv=(9, 8, 0x100, 0x10), wait=9) handled, result = model.handle(machine, pc=0x4019B00) self.assertFalse(handled) self.assertIsNone(result) self.assertEqual(model.stopped["reason"], "sync-recv-blocks") self.assertEqual(model.waits[-1]["reason"], "sync-recv-blocks") self.assertEqual(model.waits[-1]["shuttle"], 9) def test_sync_recv_returns_a_queued_completion(self): model = LibosPortModel(base=FakeBase()) machine = FakeMachine() model.host_kick() _port_call(machine, recv=(10, 6, 0, 0)) model.handle(machine, pc=0x4018038) machine.set(5, INTERRUPT_ARM_EXTERNAL) machine.set(10, 0xFF) model.handle(machine, pc=0x4018054) _port_call(machine, recv=(9, 8, 0x200, 0x10), wait=9) handled, result = model.handle(machine, pc=0x4019B00) self.assertTrue(handled) self.assertEqual(result, 0) self.assertEqual(machine.x[16], 10) self.assertEqual(model.waits[-1]["completed"], 10) class NetworkModelTests(unittest.TestCase): """The two-task routing model (opt-in, additive).""" def _network(self): return LibosPortNetwork(base=FakeBase(), tasks=("init", "rm"), current="init") def test_state_format_is_the_network_schema(self): model = self._network() self.assertEqual(model.state()["format"], NETWORK_REPORT_FORMAT) self.assertEqual(model.state()["tasks"], ["init", "rm"]) def test_cross_task_send_copies_the_firmware_payload(self): model = self._network() machine = FakeMemoryMachine() model.switch("init", pc=0) _port_call(machine, recv=(9, 8, 0x1000, 0x808)) model.handle(machine, pc=0x4019B00) machine.write_mem(0x2000, b"\x11\x22\x33\x44") model.switch("rm", pc=0) _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100), wait=2) handled, _ = model.handle(machine, pc=0x5AD507C) self.assertFalse(handled) # the sync receive blocks self.assertEqual(model.stopped["reason"], "sync-recv-blocks") completions = model.task_completions["init"] self.assertEqual(len(completions), 1) self.assertEqual(completions[0]["payload_size"], 4) self.assertEqual(completions[0]["origin_task"], 1) self.assertEqual(machine.read_mem(0x1000, 4), b"\x11\x22\x33\x44") self.assertEqual(model.cross_task_payloads, [{"port": 8, "size": 4, "from": "rm", "to": "init", "kind": "send"}]) def test_reply_completes_the_sender_receive_and_resume_is_idempotent(self): model = self._network() machine = FakeMemoryMachine() model.switch("init", pc=0) _port_call(machine, recv=(9, 8, 0x1000, 0x808)) model.handle(machine, pc=0x4019B00) machine.write_mem(0x2000, b"\xab\xcd\xef\x01") model.switch("rm", pc=0) _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100), wait=2) model.handle(machine, pc=0x5AD507C) model.stopped = None model.switch("init", pc=0) _port_call(machine, wait=SHUTTLE_ANY) handled, _ = model.handle(machine, pc=0x4019DF4) self.assertTrue(handled) self.assertEqual(machine.x[16], 9) machine.write_mem(0x1000, b"\x7f") _port_call(machine, send=(9, 0, 0x1000, 1), wait=9) handled, _ = model.handle(machine, pc=0x401829C) self.assertTrue(handled) model.stopped = None model.switch("rm", pc=0) _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100), wait=2) handled, _ = model.handle(machine, pc=0x5AD507C) self.assertTrue(handled) self.assertEqual(machine.x[16], 2) self.assertEqual(machine.read_mem(0x3000, 1), b"\x7f") # the resumed receive did not re-post a stale entry self.assertNotIn(2, model.task_recvs["rm"]) self.assertEqual(model.cross_task_payloads[-1], {"port": 0, "size": 1, "from": "init", "to": "rm", "kind": "reply"}) def test_task_state_query_serves_the_modelled_trap_state(self): model = self._network() machine = FakeMemoryMachine() model.switch("init", pc=0) model.host_kick() _port_call(machine, recv=(10, 6, 0, 0)) model.handle(machine, pc=0x4018038) machine.set(5, INTERRUPT_ARM_EXTERNAL) machine.set(10, 0xFF) model.handle(machine, pc=0x4018054) _port_call(machine, wait=SHUTTLE_ANY) model.handle(machine, pc=0x4019DF4) machine.set(5, 6) machine.set(10, 1) machine.set(11, 0x4000) handled, result = model.handle(machine, pc=0x401A46C) self.assertTrue(handled) self.assertEqual(result, 0) state = machine.read_mem(0x4000, 288) self.assertEqual(int.from_bytes(state[256:264], "little"), SPURIOUS_XEPC) self.assertEqual(int.from_bytes(state[264:272], "little"), EXTERNAL_INTERRUPT_XCAUSE) self.assertEqual(model.queries[-1]["task_index"], 1) def test_memory_query_and_shuttle_reset_are_services(self): model = self._network() machine = FakeMemoryMachine() machine.set(5, 5) machine.set(10, 0xFF) machine.set(11, 0x1000) handled, result = model.handle(machine, pc=0x5AC44FC) self.assertTrue(handled) self.assertEqual(result, 0) self.assertEqual(machine.x[12], 0x1000) self.assertEqual(machine.x[14], 1) machine.set(5, 7) machine.set(10, 1) handled, result = model.handle(machine, pc=0x5AB7D8C) self.assertTrue(handled) self.assertEqual(result, 0) self.assertEqual([s["service"] for s in model.services], ["memory_query", "shuttle_reset"]) def test_network_state_is_byte_identical_across_runs(self): def scenario(): model = self._network() machine = FakeMemoryMachine() model.switch("init", pc=0) _port_call(machine, recv=(9, 8, 0x1000, 0x808)) model.handle(machine, pc=0x4019B00) model.switch("rm", pc=0) _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100), wait=2) model.handle(machine, pc=0x5AD507C) return model.state() first = scenario() second = scenario() self.assertEqual(json.dumps(first, sort_keys=True), json.dumps(second, sort_keys=True)) if __name__ == "__main__": unittest.main()