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,
    image_tasks,
    missing_semantic_after_handoff,
    missing_semantic_after_second_task,
    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)
        with self.assertRaises(MsgqAttachError):
            run_attach(image="/nonexistent/gsp-image.bin",
                       second_task_budget=-1)
        with self.assertRaises(MsgqAttachError):
            run_attach(image="/nonexistent/gsp-image.bin",
                       second_task_budget=MAX_RUN_CYCLES + 1)


class TaskTableTests(unittest.TestCase):
    """Data-driven task discovery from the real image section table."""

    @unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "real GSP firmware absent")
    def test_two_tasks_are_discovered_with_entries_and_stacks(self):
        doc = image_tasks(DEFAULT_IMAGE)
        self.assertEqual(doc["format"], "greeneye-image-tasks/1")
        names = [task["name"] for task in doc["tasks"]]
        self.assertEqual(names, ["init", "rm"])
        for task in doc["tasks"]:
            self.assertGreaterEqual(task["entry"], task["text_base"])
            self.assertLess(task["entry"],
                            task["text_base"] + task["text_size"])
            self.assertGreater(task["stack_top"], task["stack_base"])
            self.assertGreater(task["text_size"], 0)
        rm = doc["tasks"][1]
        self.assertGreater(rm["text_size"], doc["tasks"][0]["text_size"])

    def test_absent_image_yields_an_empty_table(self):
        doc = image_tasks("/nonexistent/image.bin")
        self.assertEqual(doc, {"format": "greeneye-image-tasks/1", "tasks": []})

    def test_second_task_missing_semantic_is_precise(self):
        doc = image_tasks(DEFAULT_IMAGE) if os.path.isfile(DEFAULT_IMAGE) \
            else {"format": "greeneye-image-tasks/1",
                  "tasks": [{"name": "init"}, {"name": "rm"}]}
        missing = missing_semantic_after_second_task(
            tasks_doc=doc,
            schedule=[{"task": "init", "reason": "port-wait-blocks"},
                      {"task": "rm", "reason": "scheduled"}],
            frontier={"task": "rm", "reason": "second-task-slice-budget"},
            polls=[])
        self.assertEqual(missing["stage"], "second-task-frontier")
        self.assertEqual(missing["polls"], 0)
        self.assertEqual(missing["second_task"], "rm")
        self.assertGreaterEqual(len(missing["missing"]), 3)
        self.assertTrue(missing["next_step"])

    def test_report_carries_tasks_and_schedule_additively(self):
        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},
            tasks={"format": "greeneye-image-tasks/1",
                   "tasks": [{"name": "init"}, {"name": "rm"}]},
            schedule=[{"task": "rm", "reason": "scheduled"}],
        )
        self.assertEqual([task["name"] for task in report["tasks"]["tasks"]],
                         ["init", "rm"])
        self.assertEqual(report["schedule"][0]["task"], "rm")
        # the schema tag and the original keys are untouched
        self.assertEqual(report["format"], REPORT_FORMAT)
        self.assertIn("first_poll", report)
        self.assertIn("missing_semantic", report)


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).  The scheduled second task (the image-discovered RM task)
    completes its boot port handshake inside the bounded slice and then runs
    its own start-up past the cap, so the report is expected to take the
    honest ``wait-reached-no-poll`` frontier path -- this test accepts any of
    the allowed outcomes and asserts determinism, never success.  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"])
            self.assertEqual(report["port_model"]["format"],
                             "greeneye-msgq-port-network/1")
            self.assertEqual([task["name"] for task in report["tasks"]["tasks"]],
                             ["init", "rm"])
            self.assertTrue(report["schedule"])
            self.assertIn("rm", [step.get("task") for step in report["schedule"]])
            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") == "second-task-frontier":
                # The doorbell wake was consumed by the init task and the
                # discovered RM task completed the boot handshake before its
                # start-up ran past the bounded slice; the exact frontier is
                # named and no poll is claimed.
                self.assertIsNone(report["first_poll"])
                self.assertIsNotNone(report["missing_semantic"])
                self.assertEqual(report["missing_semantic"]["stage"],
                                 "second-task-frontier")
                completed = [wait.get("completed")
                             for wait in report["port_model"].get("waits", [])]
                self.assertIn(10, completed)
                payloads = report["port_model"].get("cross_task_payloads", [])
                self.assertTrue(any(item.get("from") == "rm"
                                    and item.get("to") == "init"
                                    for item in payloads))
        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["tasks"], second["tasks"])
        self.assertEqual(first["schedule"], second["schedule"])
        self.assertEqual(first["determinism"]["report_sha256"],
                         second["determinism"]["report_sha256"])

    def test_single_task_mode_keeps_the_handoff_classification(self):
        report = run_attach(image=DEFAULT_IMAGE, max_cycles=SMALL_BUDGET,
                            doorbell=True, second_task=False)
        self.assertIn(report["outcome"],
                      ("not-reached", "wait-reached-no-poll",
                       "wait-reached-poll-classified"))
        self.assertEqual(report["port_model"]["format"], "greeneye-msgq-port/1")
        self.assertFalse(report["schedule"])
        divergence = report["divergence"]
        if divergence and divergence.get("kind") == "task-state-query":
            self.assertIsNone(report["first_poll"])
            self.assertEqual(report["missing_semantic"]["stage"],
                             "after-extintr-handoff")


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