1 files changed, 291 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py b/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py
new file mode 100644
index 0000000..0f01c01
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py
@@ -0,0 +1,291 @@
+"""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,
+    HANDSHAKE_CONTRACT,
+    REPORT_FORMAT,
+    SHUTTLE_ANY,
+    TIMEOUT_INFINITE,
+    LibosPortModel,
+    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 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"))
+
+
+if __name__ == "__main__":
+    unittest.main()