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

"""Message-queue attach tests (sim/msgq_attach.py, docs/51 section 7).

Hermetic tests drive the pure classification/report functions with crafted
inputs -- no image, no boot, no machine.  One image test runs the real 570.144
firmware with a small bounded budget when the image is present and asserts
honest, deterministic outcomes only: it never requires the message loop to be
reached, because the tool must report a first divergence rather than fake one.

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_msgq_attach -v
"""
from __future__ import annotations

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.host_traffic import (  # noqa: E402
    DOORBELL_BASE,
    DOORBELL_SIZE,
    GSP_MSG_QUEUE_ELEMENT_HDR_SIZE,
    GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
    MSGQ_ENTRY_OFF,
    MSGQ_RX_HDR_OFF,
    MSGQ_TX_HEADER_BYTES,
    SHARED_MEM_BASE,
    HostTraffic,
)
from sim.msgq_attach import (  # noqa: E402
    DEFAULT_IMAGE,
    MAX_RUN_CYCLES,
    REPORT_FORMAT,
    MsgqAttachError,
    RingGeometry,
    assemble_report,
    classify_ring_access,
    missing_semantic_after_handoff,
    ring_geometry_from_host,
    run_attach,
)

#: Small total budget: boot-to-B3 needs ~474k instructions; the rest is a
#: bounded continuation tail.  Two runs stay well under the test time cap.
SMALL_BUDGET = 600_000


def _geometry() -> RingGeometry:
    return RingGeometry(
        shared_base=SHARED_MEM_BASE,
        cmd_base=SHARED_MEM_BASE,
        stat_base=SHARED_MEM_BASE + 0x40000,
        cmd_size=0x40000,
        stat_size=0x40000,
        msg_size=GSP_MSG_QUEUE_ELEMENT_SIZE_MIN,
        entry_off=MSGQ_ENTRY_OFF,
        rx_hdr_off=MSGQ_RX_HDR_OFF,
        doorbell_base=DOORBELL_BASE,
        doorbell_size=DOORBELL_SIZE,
    )


def _image() -> dict:
    return {"path": "/nonexistent/image.bin", "sha256": "0" * 64, "size": 1}


def _profile() -> dict:
    return {"name": "crafted", "path": "/nonexistent/profile.json"}


def _boot(reached: bool) -> dict:
    return {
        "ran": True,
        "skipped": False,
        "rung": "B3" if reached else "B1",
        "message_loop_wait_reached": reached,
        "instructions_executed": 474043 if reached else 1000,
        "stop": {"reason": "ecall" if reached else "budget",
                 "pc": 0x4018038 if reached else 0x4009000},
    }


def _assemble(reached: bool, polls, divergence) -> dict:
    return assemble_report(
        image=_image(),
        profile=_profile(),
        boot=_boot(reached),
        attach={"cmd_base": SHARED_MEM_BASE},
        doorbell={"requested": True},
        polls=polls,
        divergence=divergence,
        missing_mmio={"count": 0, "first": None},
        determinism={"instructions": 474043 if reached else 1000},
    )


class ClassifierTests(unittest.TestCase):
    """Pure ring-access classification (no machine, no boot)."""

    def test_geometry_from_host_matches_the_attached_layout(self):
        host = HostTraffic()
        try:
            geometry = ring_geometry_from_host(host)
        finally:
            host.close()
        self.assertEqual(geometry.cmd_base, SHARED_MEM_BASE)
        self.assertEqual(geometry.stat_base, SHARED_MEM_BASE + 0x40000)
        self.assertEqual(geometry.msg_size, GSP_MSG_QUEUE_ELEMENT_SIZE_MIN)
        self.assertEqual(geometry.entry_off, MSGQ_ENTRY_OFF)
        self.assertEqual(geometry.rx_hdr_off, MSGQ_RX_HDR_OFF)
        self.assertEqual(geometry.doorbell_base, DOORBELL_BASE)
        self.assertEqual(geometry.doorbell_size, DOORBELL_SIZE)

    def test_queue_head_classification(self):
        geometry = _geometry()
        poll = classify_ring_access(geometry.doorbell_base + 8, 4, 0x11, geometry)
        self.assertTrue(poll["in_ring"])
        self.assertEqual(poll["kind"], "queue-head")
        self.assertEqual(poll["queue"], 1)
        self.assertEqual(poll["register"], "NV_PGSP_QUEUE_HEAD(1)")
        self.assertEqual(poll["window"], "PRIV+0x110c00")
        self.assertEqual(poll["value"], 0x11)

    def test_tx_header_classification(self):
        geometry = _geometry()
        poll = classify_ring_access(geometry.cmd_base + 16, 4, 0, geometry)
        self.assertTrue(poll["in_ring"])
        self.assertEqual(poll["kind"], "tx-header")
        self.assertEqual(poll["buffer"], "command")
        self.assertEqual(poll["field"], "writePtr")
        self.assertEqual(poll["field_offset"], 16)
        self.assertEqual(MSGQ_TX_HEADER_BYTES, 32)

    def test_rx_header_classification_carries_the_cross_link(self):
        geometry = _geometry()
        poll = classify_ring_access(geometry.stat_base + MSGQ_RX_HDR_OFF, 4, 0,
                                    geometry)
        self.assertEqual(poll["kind"], "rx-header")
        self.assertEqual(poll["buffer"], "status")
        self.assertEqual(poll["cross_linked_owner"], "command")
        self.assertEqual(poll["semantics"], "readPtr")

    def test_entry_classification_and_slot(self):
        geometry = _geometry()
        addr = (geometry.cmd_base + MSGQ_ENTRY_OFF
                + 3 * GSP_MSG_QUEUE_ELEMENT_SIZE_MIN + 0x40)
        poll = classify_ring_access(addr, 4, 0xAB, geometry)
        self.assertEqual(poll["kind"], "entry")
        self.assertEqual(poll["buffer"], "command")
        self.assertEqual(poll["entry_index"], 3)
        self.assertEqual(poll["slot_offset"], 0x40)
        self.assertEqual(poll["region"], "entry-payload")
        header = classify_ring_access(
            geometry.cmd_base + MSGQ_ENTRY_OFF + 0x28, 4, 0, geometry)
        self.assertEqual(header["region"], "entry-header")
        self.assertEqual(GSP_MSG_QUEUE_ELEMENT_HDR_SIZE, 0x30)

    def test_non_ring_access_is_other(self):
        geometry = _geometry()
        poll = classify_ring_access(0xDEAD, 4, 0, geometry)
        self.assertFalse(poll["in_ring"])
        self.assertEqual(poll["kind"], "other")


class ReportTests(unittest.TestCase):
    """Pure report assembly: schema, outcomes and honest divergence."""

    def test_not_reached_keeps_divergence_and_no_poll(self):
        geometry = _geometry()
        divergence = {"kind": "budget",
                      "reason": "message-loop port wait not reached within budget"}
        report = _assemble(False, [], divergence)
        self.assertEqual(report["format"], REPORT_FORMAT)
        self.assertEqual(report["outcome"], "not-reached")
        self.assertFalse(report["ok"])
        self.assertIsNone(report["first_poll"])
        self.assertEqual(report["subsequent_polls"], [])
        self.assertEqual(report["poll_count"], 0)
        self.assertTrue(report["divergence"]["reason"])
        self.assertEqual(report["determinism"]["report_sha256"],
                         _assemble(False, [], divergence)["determinism"]["report_sha256"])
        self.assertEqual(geometry.doorbell_base, DOORBELL_BASE)

    def test_reached_with_poll_is_classified(self):
        geometry = _geometry()
        poll = classify_ring_access(geometry.doorbell_base, 4, 0, geometry)
        poll["pc"] = 0x4019000
        report = _assemble(True, [poll], None)
        self.assertEqual(report["outcome"], "wait-reached-poll-classified")
        self.assertTrue(report["ok"])
        self.assertEqual(report["first_poll"]["kind"], "queue-head")
        self.assertEqual(report["first_poll"]["pc"], 0x4019000)
        self.assertEqual(report["subsequent_polls"], [])
        self.assertIsNone(report["divergence"])

    def test_reached_without_poll_is_honest(self):
        report = _assemble(True, [], None)
        self.assertEqual(report["outcome"], "wait-reached-no-poll")
        self.assertTrue(report["ok"])
        self.assertIsNone(report["first_poll"])

    def test_subsequent_polls_are_bounded_to_sixteen(self):
        geometry = _geometry()
        polls = []
        for index in range(20):
            poll = classify_ring_access(geometry.cmd_base + 8, 4, index, geometry)
            poll["pc"] = 0x4018000 + index
            polls.append(poll)
        report = _assemble(True, polls, None)
        self.assertEqual(report["poll_count"], 20)
        self.assertEqual(report["first_poll"]["value"], 0)
        self.assertEqual(len(report["subsequent_polls"]), 16)
        self.assertEqual(report["subsequent_polls"][-1]["value"], 16)

    def test_absent_image_is_a_clean_skip(self):
        report = run_attach(image="/nonexistent/gsp-image.bin",
                            max_cycles=SMALL_BUDGET)
        self.assertEqual(report["format"], REPORT_FORMAT)
        self.assertEqual(report["outcome"], "skipped")
        self.assertFalse(report["ok"])
        self.assertTrue(report["boot"]["reason"])

    def test_bad_budget_is_refused(self):
        with self.assertRaises(MsgqAttachError):
            run_attach(image="/nonexistent/gsp-image.bin", max_cycles=0)
        with self.assertRaises(MsgqAttachError):
            run_attach(image="/nonexistent/gsp-image.bin",
                       max_cycles=MAX_RUN_CYCLES + 1)
        with self.assertRaises(MsgqAttachError):
            run_attach(image="/nonexistent/gsp-image.bin", max_cycles=True)


class PortModelReportTests(unittest.TestCase):
    """The additive port_model / missing_semantic report fields."""

    def test_defaults_are_additive_and_honest(self):
        report = _assemble(True, [], None)
        self.assertEqual(report["port_model"], {"enabled": False})
        self.assertIsNone(report["missing_semantic"])
        self.assertEqual(report["format"], REPORT_FORMAT)

    def test_port_model_state_is_kept_verbatim(self):
        state = {"enabled": True, "format": "greeneye-msgq-port/1",
                 "armed_mask": 0xFF, "host_kicks": 1}
        report = assemble_report(
            image=_image(),
            profile=_profile(),
            boot=_boot(True),
            attach={},
            doorbell={"requested": True},
            polls=[],
            divergence=None,
            missing_mmio={"count": 0, "first": None},
            determinism={"instructions": 1},
            port_model=state,
            missing_semantic={"stage": "after-extintr-handoff"},
        )
        self.assertTrue(report["port_model"]["enabled"])
        self.assertEqual(report["port_model"]["armed_mask"], 0xFF)
        self.assertEqual(report["missing_semantic"]["stage"],
                         "after-extintr-handoff")

    def test_missing_semantic_after_handoff_is_precise(self):
        stopped = {"num": 6, "name": "task_state_query", "pc": 0x401A46C,
                   "reason": "unmodelled-syscall"}
        missing = missing_semantic_after_handoff(stopped=stopped, polls=[])
        self.assertEqual(missing["stage"], "after-extintr-handoff")
        self.assertEqual(missing["stop"]["num"], 6)
        self.assertEqual(missing["polls"], 0)
        self.assertGreaterEqual(len(missing["missing"]), 3)
        self.assertTrue(missing["next_step"])
        self.assertTrue(all(isinstance(item, str) for item in missing["observed"]))


@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "real GSP firmware absent")
class ImageTests(unittest.TestCase):
    """Bounded real-firmware run: deterministic, honest outcome only.

    Limitation: the budget is deliberately small (600k instructions; boot-to-B3
    needs ~474k).  If the firmware needs more, the report is expected to take
    the honest ``not-reached`` path with a non-empty divergence reason -- this
    test accepts either outcome and asserts only determinism, never success.
    With the default port model the run advances past the host wake to the
    external-interrupt dispatch and reports the exact unmodelled semantic
    (``task-state-query``); a fabricated poll is never accepted.
    """

    def test_bounded_run_is_deterministic_and_honest(self):
        first = run_attach(image=DEFAULT_IMAGE, max_cycles=SMALL_BUDGET,
                           doorbell=True)
        second = run_attach(image=DEFAULT_IMAGE, max_cycles=SMALL_BUDGET,
                            doorbell=True)
        allowed = ("not-reached", "wait-reached-no-poll",
                   "wait-reached-poll-classified")
        for report in (first, second):
            self.assertEqual(report["format"], REPORT_FORMAT)
            self.assertIn(report["outcome"], allowed)
            self.assertGreater(report["image"]["size"], 0)
            self.assertEqual(len(report["image"]["sha256"]), 64)
            self.assertTrue(report["port_model"]["enabled"])
            if not report["boot"]["message_loop_wait_reached"]:
                self.assertIsNotNone(report["divergence"])
                self.assertTrue(report["divergence"]["reason"])
            if report["first_poll"] is not None:
                self.assertTrue(report["first_poll"]["in_ring"])
                self.assertTrue(report["first_poll"]["kind"] != "other")
            divergence = report["divergence"]
            if divergence and divergence.get("kind") == "task-state-query":
                # The port model delivered the doorbell wake (extintr shuttle
                # 10) and stopped at the libos trap/report path; the missing
                # semantic is named, and no poll is claimed.
                self.assertIsNone(report["first_poll"])
                self.assertIsNotNone(report["missing_semantic"])
                self.assertEqual(report["missing_semantic"]["stage"],
                                 "after-extintr-handoff")
                completed = [wait.get("completed")
                             for wait in report["port_model"].get("waits", [])]
                self.assertIn(10, completed)
        self.assertEqual(first["boot"]["message_loop_wait_reached"],
                         second["boot"]["message_loop_wait_reached"])
        self.assertEqual(first["boot"]["instructions_executed"],
                         second["boot"]["instructions_executed"])
        self.assertEqual(first["outcome"], second["outcome"])
        self.assertEqual(first["first_poll"], second["first_poll"])
        self.assertEqual(first["divergence"], second["divergence"])
        self.assertEqual(first["continuation"], second["continuation"])
        self.assertEqual(first["port_model"], second["port_model"])
        self.assertEqual(first["missing_semantic"], second["missing_semantic"])
        self.assertEqual(first["determinism"]["report_sha256"],
                         second["determinism"]["report_sha256"])


if __name__ == "__main__":
    unittest.main()