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