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

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