diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_gmmu_debug.py | 484 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_gmmu_fbif_trace.py | 381 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_gmmu_oracle.py | 397 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_gmmu_pte.py | 534 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_mpu.py | 177 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py | 28 |
6 files changed, 1995 insertions, 6 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_gmmu_debug.py b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_debug.py new file mode 100644 index 0000000..d9046be --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_debug.py @@ -0,0 +1,484 @@ +"""Tests for the GMMU/FBIF static<->dynamic join (harness/gmmu_debug.py, docs/44). + +All fixtures are synthetic: no firmware, no simulator run, no QEMU, no +hardware. The end-to-end demo test builds its own fixture in a temporary +directory and checks the frozen threshold result. + +Run:: + + python3 -m unittest tests.test_gmmu_debug -v +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import sys +import tempfile +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 harness.gmmu_debug import ( # noqa: E402 + DEBUG_REPORT_FORMAT, + HITS_HEADER, + MANUAL_STATIC_BASELINE_SECONDS, + TRACE_FORMAT, + WORKFLOW_THRESHOLD_FRACTION, + GmmuInputError, + build_debug_report, + load_sim_trace, + load_symbols, + main, + run_demo, + run_report, + write_hits_tsv, +) + + +def _write_json(path: str, payload) -> str: + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh) + return path + + +def _symbols_json(path: str) -> str: + return _write_json( + path, + { + "format": "greeneye-symbols/1", + "symbols": [ + {"addr": 0x4001000, "name": "gmmu_map_alloc", "size": 0x120}, + {"addr": 0x4002000, "name": "fbif_setup", "size": 0x80}, + ], + }, + ) + + +def _events() -> list[dict]: + return [ + {"seq": 0, "pc": "0x4002040", "kind": "write", "window": "falcon", + "offset": "0x624", "name": "FBIF_CTL", "value": "0x1"}, + {"seq": 1, "pc": "0x4002050", "kind": "read", "window": "falcon", + "offset": "0x8", "name": "IRQSTAT", "value": "0x0"}, + {"seq": 2, "pc": "0x4001010", "kind": "write", "window": "gmmu", + "offset": "0x0", "name": "GMMU_PDB_LO", "value": "0x400000"}, + {"seq": 3, "pc": "0x4001020", "kind": "write", "window": "gmmu", + "offset": "0x0", "name": "GMMU_PDB_LO", "value": "0x800000"}, + {"seq": 4, "pc": "0x4009000", "kind": "write", "window": "gmmu", + "offset": "0x10", "name": "GMMU_MODE", "value": "0x7"}, + ] + + +def _g1_document() -> dict: + """The nested shape emitted by sim/devices/observe.py:build_trace.""" + return { + "schema": TRACE_FORMAT, + "image_sha256": "ab" * 32, + "profile": "tu104-570.144", + "profile_name": "fixture", + "symbols": "symbols.json", + "summary": {"windows": 1, "accesses": 2, "retained_accesses": 2, + "dropped_accesses": 0}, + "windows": [ + { + "label": "falcon-observe", + "base": "0x300110000", + "size": "0x1000", + "mode": "observe-only", + "evidence": "header", + "counts": {"r": 1, "w": 1, "served": 0, "declined": 2}, + "registers": [ + { + "offset": "0x624", + "name": "FBIF_CTL", + "access": "rw", + "counts": {"r": 0, "w": 1, "served": 0}, + "accesses": [ + {"addr": "0x300110624", "offset": "0x624", "size": 4, + "rw": "w", "value": "0x1", "pc": "0x4002040", "vt": 123, + "window": "falcon-observe", "register": "FBIF_CTL", + "served": False, "pc_symbol": "fbif_setup"}, + ], + } + ], + "unclassified": { + "counts": {"r": 1, "w": 0, "served": 0}, + "accesses": [ + {"addr": "0x300110008", "offset": "0x8", "size": 4, + "rw": "r", "value": None, "pc": "0x4002050", "vt": 123, + "window": "falcon-observe", "register": None, + "served": False, "pc_symbol": "fbif_setup"}, + ], + }, + } + ], + } + + +def _run_main(argv: list[str]) -> tuple[int, str, str]: + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = main(argv) + return rc, out.getvalue(), err.getvalue() + + +class TraceIngestTests(unittest.TestCase): + def test_trace_tagged_and_aliases(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_json( + os.path.join(tmp, "t.json"), + { + "format": TRACE_FORMAT, + "events": [ + {"pc": 0x4002040, "rw": "w", "block": "falcon", + "reg_offset": 0x624, "register": "FBIF_CTL", "data": 1}, + ], + }, + ) + trace = load_sim_trace(path) + self.assertEqual(trace.format, TRACE_FORMAT) + obs = trace.observations[0] + self.assertEqual((obs.kind, obs.window, obs.offset, obs.name, obs.value), + ("write", "falcon", 0x624, "FBIF_CTL", 1)) + + def test_jsonl_fallback(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "t.jsonl") + with open(path, "w", encoding="utf-8") as fh: + fh.write(json.dumps({"format": TRACE_FORMAT}) + "\n") + fh.write(json.dumps({"t": "write", "addr": "0x624", "size": 4, "pc": "0x4002040"}) + "\n") + fh.write(json.dumps({"t": "summary", "insns": 1}) + "\n") + trace = load_sim_trace(path) + self.assertEqual(len(trace.observations), 1) + self.assertEqual(trace.observations[0].addr, 0x624) + + def test_empty_event_list_is_legal(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_json(os.path.join(tmp, "t.json"), {"format": TRACE_FORMAT, "events": []}) + trace = load_sim_trace(path) + self.assertEqual(trace.observations, []) + + def test_untagged_object_without_events_is_clean_error(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_json(os.path.join(tmp, "t.json"), {"hello": "world"}) + with self.assertRaises(GmmuInputError): + load_sim_trace(path) + + def test_missing_trace_is_clean_error(self): + with self.assertRaises(GmmuInputError): + load_sim_trace("/does/not/exist.json") + + def test_session_directory_ingest(self): + with tempfile.TemporaryDirectory() as tmp: + blobs = os.path.join(tmp, "blobs") + os.makedirs(blobs) + _write_json(os.path.join(tmp, "session.json"), {"format": "greeneye-session/1"}) + _write_json( + os.path.join(blobs, "gmmu-fbif-trace.json"), + {"format": TRACE_FORMAT, "events": _events()}, + ) + trace = load_sim_trace(tmp) + self.assertEqual(len(trace.observations), len(_events())) + + def test_g1_nested_document_flattens(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_json(os.path.join(tmp, "t.json"), _g1_document()) + trace = load_sim_trace(path) + self.assertEqual(trace.format, TRACE_FORMAT) + self.assertEqual(len(trace.observations), 2) + self.assertEqual(trace.source["profile"], "tu104-570.144") + self.assertEqual(trace.source["summary"]["dropped_accesses"], 0) + self.assertEqual(trace.source["summary"]["retained_accesses"], 2) + write_obs = trace.observations[0] + self.assertEqual(write_obs.kind, "write") + self.assertEqual(write_obs.name, "FBIF_CTL") + self.assertEqual(write_obs.window, "falcon-observe") + self.assertEqual(write_obs.offset, 0x624) + self.assertEqual(write_obs.addr, 0x300110624) + self.assertEqual(write_obs.pc, 0x4002040) + self.assertEqual(write_obs.vt, 123) + self.assertIs(write_obs.handled, False) + self.assertEqual(write_obs.function, "fbif_setup") + read_obs = trace.observations[1] + self.assertEqual(read_obs.kind, "read") + self.assertEqual(read_obs.offset, 0x8) + self.assertIsNone(read_obs.value) + + def test_g1_nested_document_report(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_json(os.path.join(tmp, "t.json"), _g1_document()) + report = build_debug_report(load_sim_trace(path), None) + self.assertEqual(report["summary"]["functions"], 1) + function = report["functions"][0] + self.assertEqual(function["function"], "fbif_setup") + registers = {r["register"]: r for r in function["registers"]} + self.assertEqual(registers["FBIF_CTL"]["write_count"], 1) + self.assertEqual(registers["0x8"]["read_count"], 1) + + +class ObserveProducerIntegrationTests(unittest.TestCase): + """Ingest a document produced by the real sim/devices/observe.py module. + + The producer is in-tree, so the import is unconditional: if it breaks, + the G1<->G3 interface has regressed and this test must fail, not skip. + """ + + def test_ingest_real_build_trace_document(self): + from sim.devices.observe import ObservedWindow, build_trace + window = ObservedWindow( + base=0x300110000, + size=0x1000, + label="integration", + registers={"0x624": {"name": "FBIF_CTL", "width": 4, "access": "rw"}}, + ) + self.assertFalse(window.write(0x624, 4, 1, pc=0x4002040)) + self.assertIsNone(window.read(0x8, 4, pc=0x4002050)) + registry = type("Registry", (), {"devices": [window]})() + document = build_trace( + registry, image_sha256="cd" * 32, profile="tu104-570.144" + ) + self.assertEqual(document.get("schema"), TRACE_FORMAT) + with tempfile.TemporaryDirectory() as tmp: + path = _write_json(os.path.join(tmp, "t.json"), document) + trace = load_sim_trace(path) + self.assertEqual(trace.format, TRACE_FORMAT) + self.assertEqual(len(trace.observations), 2) + names = {obs.name for obs in trace.observations} + self.assertIn("FBIF_CTL", names) + write_obs = next(obs for obs in trace.observations if obs.kind == "write") + self.assertEqual(write_obs.offset, 0x624) + self.assertEqual(write_obs.value, 1) + + +class SymbolLoadingTests(unittest.TestCase): + def test_symbols_json_db(self): + with tempfile.TemporaryDirectory() as tmp: + db = load_symbols(_symbols_json(os.path.join(tmp, "s.json"))) + self.assertEqual(db.resolve(0x4002040).name, "fbif_setup") + + def test_symbols_plain_map_json(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_json(os.path.join(tmp, "s.json"), {"0x4001000": "gmmu_map_alloc"}) + db = load_symbols(path) + self.assertEqual(db.get("gmmu_map_alloc").addr, 0x4001000) + + def test_symbols_tsv(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "s.tsv") + with open(path, "w", encoding="utf-8") as fh: + fh.write("0x4001000\tgmmu_map_alloc\n0x4002000\tfbif_setup\n") + db = load_symbols(path) + self.assertEqual(db.resolve(0x4002000).name, "fbif_setup") + + def test_missing_symbols_is_clean_error(self): + with self.assertRaises(GmmuInputError): + load_symbols("/does/not/exist.tsv") + + +class ReportTests(unittest.TestCase): + def _report(self, events=None, symbols=None): + with tempfile.TemporaryDirectory() as tmp: + trace_path = _write_json( + os.path.join(tmp, "t.json"), + {"format": TRACE_FORMAT, "events": events if events is not None else _events()}, + ) + trace = load_sim_trace(trace_path) + return build_debug_report(trace, symbols) + + def test_report_shape_and_resolution(self): + with tempfile.TemporaryDirectory() as tmp: + symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json"))) + report = self._report(symbols=symbols) + self.assertEqual(report["format"], DEBUG_REPORT_FORMAT) + summary = report["summary"] + self.assertEqual(summary["functions"], 3) + self.assertEqual(summary["writes"], 4) + self.assertEqual(summary["reads"], 1) + names = [f["function"] for f in report["functions"]] + self.assertEqual(names, ["gmmu_map_alloc", "fbif_setup", "sub_0x4009000"]) + fbif = report["functions"][1] + self.assertEqual(fbif["addr"], 0x4002000) + registers = {r["register"]: r for r in fbif["registers"]} + self.assertEqual(registers["FBIF_CTL"]["write_count"], 1) + self.assertEqual(registers["FBIF_CTL"]["first_pc"], 0x4002040) + self.assertEqual(registers["FBIF_CTL"]["last_pc"], 0x4002040) + self.assertEqual(registers["IRQSTAT"]["read_count"], 1) + unknown = report["functions"][2] + self.assertEqual(unknown["function"], "sub_0x4009000") + self.assertEqual(unknown["addr"], 0x4009000) + + def test_sequence_order_and_last_pc(self): + with tempfile.TemporaryDirectory() as tmp: + symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json"))) + report = self._report(symbols=symbols) + alloc = report["functions"][0] + pdb = next(r for r in alloc["registers"] if r["register"] == "GMMU_PDB_LO") + self.assertEqual(pdb["write_count"], 2) + self.assertEqual(pdb["values"], [0x400000, 0x800000]) + self.assertEqual(pdb["first_pc"], 0x4001010) + self.assertEqual(pdb["last_pc"], 0x4001020) + + def test_explicit_function_field_wins(self): + events = [ + {"seq": 0, "pc": "0x4009000", "kind": "write", "window": "gmmu", + "offset": "0x10", "name": "GMMU_MODE", "value": "0x7", + "function": "fbif_setup"}, + ] + with tempfile.TemporaryDirectory() as tmp: + symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json"))) + report = self._report(events=events, symbols=symbols) + self.assertEqual(report["functions"][0]["function"], "fbif_setup") + self.assertEqual(report["functions"][0]["addr"], 0x4002000) + + def test_no_symbols_gives_sub_labels(self): + report = self._report(symbols=None) + self.assertTrue(all(f["function"].startswith("sub_") for f in report["functions"])) + + def test_events_without_pc_are_counted_not_attributed(self): + events = [{"seq": 0, "kind": "write", "window": "gmmu", "offset": "0x10", + "name": "GMMU_MODE", "value": "0x7"}] + report = self._report(events=events, symbols=None) + self.assertEqual(report["summary"]["unknown_pc_events"], 1) + self.assertEqual(report["summary"]["functions"], 0) + + def test_empty_trace(self): + report = self._report(events=[], symbols=None) + self.assertEqual(report["summary"]["functions"], 0) + self.assertEqual(report["summary"]["registers"], 0) + self.assertEqual(report["functions"], []) + + +class HitsTsvTests(unittest.TestCase): + def test_hits_tsv_is_markgreeneye_compatible(self): + with tempfile.TemporaryDirectory() as tmp: + symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json"))) + trace = load_sim_trace( + _write_json( + os.path.join(tmp, "t.json"), + {"format": TRACE_FORMAT, "events": _events()}, + ) + ) + report = build_debug_report(trace, symbols) + hits_path = write_hits_tsv(report, os.path.join(tmp, "greeneye_hits.tsv")) + with open(hits_path, "r", encoding="utf-8") as fh: + lines = [line.rstrip("\n") for line in fh] + self.assertEqual(lines[0], HITS_HEADER) + self.assertTrue(lines[0].startswith("addr")) + rows = [line.split("\t") for line in lines[1:]] + self.assertEqual(len(rows), 3) + for cols in rows: + self.assertGreaterEqual(len(cols), 2) + int(cols[0].replace("0x", ""), 16) + self.assertTrue(cols[1].isdigit()) + first = rows[0] + self.assertEqual(first[0], "0x4001000") + self.assertEqual(first[2], "gmmu_map_alloc") + self.assertIn("write", first[3]) + addrs = {int(cols[0], 16) for cols in rows} + self.assertIn(0x4009000, addrs) + + def test_hits_tsv_empty_report(self): + with tempfile.TemporaryDirectory() as tmp: + report = {"format": DEBUG_REPORT_FORMAT, "functions": []} + hits_path = write_hits_tsv(report, os.path.join(tmp, "greeneye_hits.tsv")) + with open(hits_path, "r", encoding="utf-8") as fh: + self.assertEqual(fh.read().strip(), HITS_HEADER) + + +class ArtifactTests(unittest.TestCase): + def test_run_report_writes_artifacts(self): + with tempfile.TemporaryDirectory() as tmp: + symbols_path = _symbols_json(os.path.join(tmp, "s.json")) + trace_path = _write_json( + os.path.join(tmp, "t.json"), + {"format": TRACE_FORMAT, "events": _events()}, + ) + out = os.path.join(tmp, "out") + report = run_report(trace_path, symbols_path, out) + for key in ("report_json", "report_md", "hits_tsv", "ghidra_command"): + self.assertIn(key, report["artifacts"]) + self.assertTrue(os.path.isfile(report["artifacts"]["report_json"])) + self.assertTrue(os.path.isfile(report["artifacts"]["report_md"])) + self.assertTrue(os.path.isfile(report["artifacts"]["hits_tsv"])) + self.assertIn("MarkGreenEyeHits.java", report["artifacts"]["ghidra_command"]) + with open(report["artifacts"]["report_json"], "r", encoding="utf-8") as fh: + loaded = json.load(fh) + self.assertEqual(loaded["summary"]["functions"], 3) + with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh: + md = fh.read() + self.assertIn("Per function", md) + self.assertIn("Ghidra annotation", md) + + +class DemoTests(unittest.TestCase): + def test_demo_end_to_end_and_frozen_threshold(self): + with tempfile.TemporaryDirectory() as tmp: + out = io.StringIO() + with contextlib.redirect_stdout(out): + result = run_demo(tmp) + self.assertIn("RESULT_JSON:", out.getvalue()) + self.assertTrue(result["pass"]) + self.assertLessEqual(result["measured_seconds"], result["threshold_seconds"]) + self.assertEqual( + result["threshold_seconds"], + round(MANUAL_STATIC_BASELINE_SECONDS * WORKFLOW_THRESHOLD_FRACTION, 4), + ) + self.assertIn("estimate", result["manual_baseline_label"]) + self.assertEqual(result["summary"]["functions"], 4) + self.assertEqual(result["summary"]["writes"], 11) + self.assertEqual(result["summary"]["reads"], 2) + for key in ("report_json", "report_md", "hits_tsv"): + self.assertTrue(os.path.isfile(result["artifacts"][key])) + self.assertTrue(os.path.isfile(os.path.join(tmp, "demo-result.json"))) + + +class CliTests(unittest.TestCase): + def test_report_cli(self): + with tempfile.TemporaryDirectory() as tmp: + symbols_path = _symbols_json(os.path.join(tmp, "s.json")) + trace_path = _write_json( + os.path.join(tmp, "t.json"), + {"format": TRACE_FORMAT, "events": _events()}, + ) + out = os.path.join(tmp, "out") + rc, stdout, _stderr = _run_main( + ["report", "--trace", trace_path, "--symbols", symbols_path, "--out", out] + ) + self.assertEqual(rc, 0) + payload = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1]) + self.assertEqual(payload["summary"]["functions"], 3) + self.assertTrue(os.path.isfile(os.path.join(out, "greeneye_hits.tsv"))) + + def test_report_cli_missing_trace_exits_2(self): + with tempfile.TemporaryDirectory() as tmp: + rc, _stdout, stderr = _run_main( + ["report", "--trace", "/nope.json", "--out", os.path.join(tmp, "out")] + ) + self.assertEqual(rc, 2) + self.assertIn("trace not found", stderr) + + def test_empty_trace_cli_writes_report(self): + with tempfile.TemporaryDirectory() as tmp: + trace_path = _write_json( + os.path.join(tmp, "t.json"), {"format": TRACE_FORMAT, "events": []} + ) + out = os.path.join(tmp, "out") + rc, stdout, _stderr = _run_main(["report", "--trace", trace_path, "--out", out]) + self.assertEqual(rc, 0) + payload = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1]) + self.assertEqual(payload["summary"]["functions"], 0) + self.assertTrue(os.path.isfile(os.path.join(out, "gmmu-debug-report.md"))) + + def test_no_command_prints_help(self): + rc, stdout, _stderr = _run_main([]) + self.assertEqual(rc, 2) + self.assertIn("usage", stdout.lower()) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_gmmu_fbif_trace.py b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_fbif_trace.py new file mode 100644 index 0000000..7b43b64 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_fbif_trace.py @@ -0,0 +1,381 @@ +"""GMMU/FBIF observer tests: no-fake-read, no semantic change, trace schema. + +Every assertion is tied to the evidence documented in +``sim/devices/observe.py`` and ``docs/42-GMMU-FBIF-OBSERVABILITY.md``. The +real-firmware runs are bounded (<= 20k instructions) and assert observed +numbers only: adding observer windows must not change instructions, stop +PC/reason, unknown MMIO, unhandled-in-window counts or served accesses. +""" +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sim.devices import ( # noqa: E402 + ObservedWindow, + PmcBlock, + TRACE_SCHEMA, + build_trace, + collect_observations, + load_profile, + parse_window_spec, + registry_from_profile, + windows_from_section, +) +from sim.devices.base import DeviceRegistry # noqa: E402 +from sim.machine import Machine # noqa: E402 + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +DEFAULT_IMAGE = os.path.normpath( + os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", + "gsp-570.144.fwimage.bin")) +DEFAULT_PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json") + +PMC_BASE = 0x2000000000000000 +PMC_BOOT_42 = PMC_BASE + 0xA00 + + +class ObserveWindowUnitTests(unittest.TestCase): + def test_observe_only_declines_and_records(self): + reg = DeviceRegistry() + dev = ObservedWindow(base=0x1000, size=0x100, label="unit", + mode="observe-only", + registers={0x10: "REG_A"}) + reg.add(dev) + value = reg.read(0x1010, 4, pc=0x4000) + self.assertEqual(value, (False, 0)) + self.assertEqual(reg.unhandled, 1) + self.assertEqual(dev.counts["r"], 1) + self.assertEqual(dev.counts["declined"], 1) + entry = dev.accesses[0] + self.assertEqual(entry["addr"], 0x1010) + self.assertEqual(entry["offset"], 0x10) + self.assertEqual(entry["size"], 4) + self.assertEqual(entry["rw"], "r") + self.assertIsNone(entry["value"]) # never invent a read value + self.assertFalse(entry["served"]) + self.assertEqual(entry["pc"], 0x4000) + self.assertEqual(entry["register"], "REG_A") + self.assertEqual(entry["window"], "unit") + self.assertIsNone(entry["vt"]) # no clock supplied + self.assertFalse(reg.write(0x1010, 4, 0xAB, pc=0x4004)) + self.assertEqual(dev.accesses[1]["value"], 0xAB) + self.assertFalse(dev.accesses[1]["served"]) + + def test_storage_reads_back_written_values_only(self): + dev = ObservedWindow(base=0x2000, size=0x100, label="st", mode="storage", + registers={ + 0x10: {"name": "SPARE", "access": "rw"}, + 0x14: {"name": "RO", "access": "ro"}, + }) + reg = DeviceRegistry() + reg.add(dev) + # Never-written storage register: declined, no invented reset value. + self.assertEqual(reg.read(0x2010, 4), (False, 0)) + self.assertTrue(reg.write(0x2010, 4, 0xCAFE)) + self.assertEqual(reg.read(0x2010, 4), (True, 0xCAFE)) + # Read-only register: writes declined, reads have no value source. + self.assertFalse(reg.write(0x2014, 4, 1)) + self.assertEqual(reg.read(0x2014, 4), (False, 0)) + # Undeclared offset inside the window stays declined. + self.assertEqual(reg.read(0x2018, 4), (False, 0)) + # Wrong width is not served. + self.assertEqual(reg.read(0x2010, 8), (False, 0)) + self.assertEqual(dev.written[0x10], 0xCAFE) + self.assertEqual(dev.counts["served"], 2) # one write + one read-back + + def test_storage_field_decode_in_trace(self): + dev = ObservedWindow(base=0x3000, size=0x100, label="fields", + mode="storage", + registers={0x10: {"name": "CFG", + "fields": {"LOW": [3, 0], + "HIGH": [7, 4]}}}) + reg = DeviceRegistry() + reg.add(dev) + reg.write(0x3010, 4, 0x0053, pc=0x100) + reg.read(0x3010, 4, pc=0x104) + doc = build_trace(reg, image_sha256="0" * 64, profile="p.json", + profile_name="p") + win = doc["windows"][0] + self.assertEqual(win["label"], "fields") + self.assertEqual(win["mode"], "storage") + reg_entry = win["registers"][0] + self.assertEqual(reg_entry["name"], "CFG") + self.assertEqual(reg_entry["counts"], {"r": 1, "w": 1, "served": 2}) + write, read = reg_entry["accesses"] + self.assertEqual(write["value"], "0x53") + self.assertEqual(write["decoded"], {"LOW": 3, "HIGH": 5}) + self.assertEqual(read["value"], "0x53") + + def test_windows_from_section_expands_count_and_stride(self): + section = { + "mode": "observe-only", + "registers": {"0x0": "REG0"}, + "windows": [ + {"base": "0x1000", "size": "0x100", "label": "single"}, + {"base": "0x2000", "size": "0x100", "label": "many", + "count": 3, "stride": "0x1000"}, + {"base": "0x3000", "size": "0x100", "label": "override", + "mode": "storage"}, + ], + } + windows = windows_from_section(section) + self.assertEqual([w["label"] for w in windows], + ["single", "many-0", "many-1", "many-2", "override"]) + self.assertEqual([w["base"] for w in windows], + [0x1000, 0x2000, 0x3000, 0x4000, 0x3000]) + self.assertEqual(windows[1]["registers"], {"0x0": "REG0"}) + self.assertEqual(windows[-1]["mode"], "storage") + + def test_parse_window_spec(self): + win = parse_window_spec("0x2000000000100000:0x1000:adhoc") + self.assertEqual((win["base"], win["size"], win["label"], win["mode"]), + (0x2000000000100000, 0x1000, "adhoc", "observe-only")) + win = parse_window_spec("0x1000:0x100:st:storage") + self.assertEqual(win["mode"], "storage") + with self.assertRaises(ValueError): + parse_window_spec("0x1000") + with self.assertRaises(ValueError): + parse_window_spec("0x1000:0x100:bad:wrong-mode") + + def test_state_round_trip(self): + dev = ObservedWindow(base=0x4000, size=0x100, label="rt", mode="storage", + registers={0x10: "SPARE"}) + dev.write(0x10, 4, 0x1234) + snap = dev.state() + other = ObservedWindow(base=0x4000, size=0x100, label="rt", mode="storage", + registers={0x10: "SPARE"}) + other.load_state(snap) + self.assertEqual(other.state(), dev.state()) + self.assertEqual(other.written[0x10], 0x1234) + + def test_access_log_cap_is_reported_not_silent(self): + dev = ObservedWindow(base=0x6000, size=0x100, label="cap", + max_access_log=2, registers={0x10: "REG_A"}) + for value in (1, 2, 3, 4): + dev.write(0x10, 4, value) + doc = build_trace(type("Registry", (), {"devices": [dev]})()) + self.assertEqual(dev.counts["w"], 4) + win = doc["windows"][0] + self.assertEqual(win["counts"]["w"], 4) + self.assertEqual(win["retained_accesses"], 2) + self.assertEqual(win["dropped_accesses"], 2) + self.assertEqual(win["registers"][0]["counts"]["w"], 2) + summary = doc["summary"] + self.assertEqual(summary["accesses"], 4) + self.assertEqual(summary["retained_accesses"], 2) + self.assertEqual(summary["dropped_accesses"], 2) + + +class NoSemanticChangeTests(unittest.TestCase): + def test_observe_only_matches_no_window_lenient(self): + reg_a = DeviceRegistry() + reg_a.add(ObservedWindow(base=0x50000000, size=0x1000, label="obs")) + reg_b = DeviceRegistry() + a = Machine(memory_mb=1, ram_base=0, devices=reg_a, unknown_mmio="lenient") + b = Machine(memory_mb=1, ram_base=0, devices=reg_b, unknown_mmio="lenient") + self.assertEqual(a._mmio_load(0x50000010, 4, pc=0x40), + b._mmio_load(0x50000010, 4, pc=0x40)) + self.assertEqual(a.mmio_unknown, b.mmio_unknown) + self.assertEqual(a.mmio_log, b.mmio_log) + a._mmio_store(0x50000010, 4, 7, pc=0x44) + b._mmio_store(0x50000010, 4, 7, pc=0x44) + self.assertEqual(a.mmio_unknown, b.mmio_unknown) + self.assertEqual(a.mmio_log, b.mmio_log) + + def test_observe_only_matches_no_window_strict(self): + from sim.machine import Fault + + reg_a = DeviceRegistry() + reg_a.add(ObservedWindow(base=0x50000000, size=0x1000, label="obs")) + a = Machine(memory_mb=1, ram_base=0, devices=reg_a, unknown_mmio="fault") + b = Machine(memory_mb=1, ram_base=0, devices=DeviceRegistry(), + unknown_mmio="fault") + with self.assertRaises(Fault) as ca: + a._mmio_load(0x50000010, 4, pc=0x40) + with self.assertRaises(Fault) as cb: + b._mmio_load(0x50000010, 4, pc=0x40) + self.assertEqual(ca.exception.cause, cb.exception.cause) + self.assertEqual(ca.exception.tval, cb.exception.tval) + + def test_observer_does_not_shadow_a_modelled_window(self): + reg = DeviceRegistry() + pmc = PmcBlock(chip_id=0x164, base=PMC_BASE, size=0x1000) + reg.add(pmc) + obs = ObservedWindow(base=PMC_BOOT_42, size=0x10, label="overlay") + reg.add(obs) + machine = Machine(memory_mb=1, ram_base=0, devices=reg, + unknown_mmio="lenient") + value = machine._mmio_load(PMC_BOOT_42, 4, pc=0x400) + self.assertEqual(value, pmc.read(0xA00, 4)) + self.assertEqual(reg.unhandled, 0) + self.assertEqual(len(obs.accesses), 1) + self.assertFalse(obs.accesses[0]["served"]) + self.assertIsNone(obs.accesses[0]["value"]) + + def test_gmmu_wpr_overlap_leaves_underlying_path_identical(self): + # gmmu-wpr-lock (profile) overlaps the fwsec WPR2 window; the + # observer must record and decline, leaving whatever the underlying + # path does byte-for-byte identical to a run without the observer. + profile = load_profile(DEFAULT_PROFILE) + addr = 0x20000000001FA824 + + def probe(p): + machine = Machine(memory_mb=16, ram_base=0, unknown_mmio="lenient") + reg = registry_from_profile(p, machine=machine) + machine.devices = reg + value = machine._mmio_load(addr, 4, pc=0x40) + observations = collect_observations(reg) + return (machine.mmio_unknown, reg.unhandled, value, + {d.label: d.counts for d in observations}) + + with_observers = probe(profile) + stripped = json.loads(json.dumps(profile)) + stripped["devices"].pop("gmmu", None) + stripped["devices"].pop("fbif", None) + without_observers = probe(stripped) + self.assertEqual(with_observers[:3], without_observers[:3]) + self.assertEqual(with_observers[3]["gmmu-wpr-lock"]["declined"], 1) + self.assertEqual(without_observers[3], {}) + + +class ProfileAndBootTests(unittest.TestCase): + def test_profile_declares_gmmu_fbif_observe_windows(self): + profile = load_profile(DEFAULT_PROFILE) + self.assertIn("gmmu", profile["devices"]) + self.assertIn("fbif", profile["devices"]) + machine = Machine(memory_mb=16, ram_base=0, unknown_mmio="lenient") + reg = registry_from_profile(profile, machine=machine) + observers = collect_observations(reg) + self.assertEqual(len(observers), 23) + by_label = {d.label: d for d in observers} + self.assertEqual(by_label["gmmu-pri-mmu"].base, 0x2000000000100C80) + self.assertEqual(by_label["gmmu-pri-mmu"].size, 0x280) + self.assertEqual(by_label["gmmu-wpr-lock"].base, 0x20000000001FA7D8) + self.assertEqual(by_label["gmmu-hsmmu"].base, 0x20000000001FAC00) + self.assertEqual(by_label["fbpa"].base, 0x20000000009A0000) + self.assertEqual(by_label["fbpa-partition-15"].base, + 0x2000000000900000 + 15 * 0x4000) + self.assertEqual(by_label["fbpa-mc-2"].base, + 0x2000000000980000 + 2 * 0x4000) + for dev in observers: + self.assertEqual(dev.mode, "observe-only") + self.assertEqual(by_label["gmmu-pri-mmu"].registers[0x0]["name"], + "NV_PFB_PRI_MMU_CTRL") + self.assertEqual(by_label["fbpa"].registers[0x50]["name"], + "NV_PFB_FBPA_SPARE") + + def test_b2_numbers_unchanged_by_observer_windows(self): + from sim.gsp_boot import run_boot + + r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, + instructions=20_000, record=False, use_shim=True) + self.assertEqual(r.rung, "B2") + self.assertEqual(r.instructions, 20_000) + self.assertEqual(r.stop_reason, "budget") + self.assertEqual(r.mmio_unknown, 0) + self.assertEqual(r.device_unhandled, 0) + self.assertEqual(r.device_accesses, 15) + # The run reaches the WPR2 reads inside the gmmu-wpr-lock observer + # window (offsets 0x4C/0x50); the observer records them without + # serving, and the fwsec device still answers. + self.assertIsNotNone(r.trace_summary) + self.assertEqual(r.trace_summary["windows"], 23) + self.assertEqual(r.trace_summary["accesses"], 2) + self.assertEqual(r.trace_summary["served"], 0) + + # Same run with the observer sections removed: identical execution. + with open(DEFAULT_PROFILE, "r", encoding="utf-8") as fh: + profile = json.load(fh) + del profile["devices"]["gmmu"] + del profile["devices"]["fbif"] + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "no-observe.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(profile, fh) + base = run_boot(image=DEFAULT_IMAGE, profile_path=path, + instructions=20_000, record=False, use_shim=True) + self.assertEqual(base.instructions, r.instructions) + self.assertEqual(base.stop_reason, r.stop_reason) + self.assertEqual(base.stop_pc, r.stop_pc) + self.assertEqual(base.mmio_unknown, r.mmio_unknown) + self.assertEqual(base.device_unhandled, r.device_unhandled) + self.assertEqual(base.device_accesses, r.device_accesses) + self.assertEqual(base.registers, r.registers) + + def test_boot_trace_schema_and_determinism(self): + from sim.gsp_boot import run_boot + + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "trace-dir") + os.makedirs(path) + r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, + instructions=20_000, record=False, use_shim=True, + trace_out=path) + self.assertEqual(r.trace_path, + os.path.join(path, "gmmu-fbif-trace.json")) + with open(r.trace_path, "r", encoding="utf-8") as fh: + doc = json.load(fh) + self.assertEqual(doc["schema"], TRACE_SCHEMA) + self.assertEqual(doc["image_sha256"], r.image_sha256) + self.assertEqual(set(doc["summary"]), + {"windows", "windows_touched", "accesses", + "retained_accesses", "dropped_accesses", + "reads", "writes", "served", "registers_touched"}) + self.assertEqual(doc["summary"]["accesses"], 2) + self.assertEqual(doc["summary"]["retained_accesses"], 2) + self.assertEqual(doc["summary"]["dropped_accesses"], 0) + labels = [w["label"] for w in doc["windows"]] + self.assertIn("gmmu-pri-mmu", labels) + self.assertIn("fbpa", labels) + for win in doc["windows"]: + self.assertEqual(set(win), + {"label", "base", "size", "mode", "evidence", + "counts", "dropped_accesses", + "retained_accesses", "registers", + "unclassified"}) + self.assertEqual(win["mode"], "observe-only") + self.assertEqual(win["dropped_accesses"], 0) + self.assertTrue(win["base"].startswith("0x")) + r2 = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, + instructions=20_000, record=False, use_shim=True) + self.assertEqual(r.device_state_sha256, r2.device_state_sha256) + + def test_runtime_window_records_without_faking_and_resolves_symbols(self): + from sim.gsp_boot import run_boot + + symbols = {"format": "greeneye-symbols/1", "count": 1, + "symbols": [{"addr": "0x4009014", "name": "boot_probe", + "size": 64}]} + with tempfile.TemporaryDirectory() as tmp: + sym_path = os.path.join(tmp, "symbols.json") + trace_path = os.path.join(tmp, "trace.json") + with open(sym_path, "w", encoding="utf-8") as fh: + json.dump(symbols, fh) + r = run_boot( + image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, + instructions=20_000, record=False, use_shim=True, + observe_windows=["0x%x:0x10:pmc-probe" % PMC_BOOT_42], + symbols_path=sym_path, trace_out=trace_path) + # Two gmmu-wpr-lock reads (the WPR2 pair) plus the two BOOT_42 + # reads inside the runtime PMC probe window. + self.assertEqual(r.trace_summary["accesses"], 4) + self.assertEqual(r.trace_summary["windows"], 24) + with open(trace_path, "r", encoding="utf-8") as fh: + doc = json.load(fh) + self.assertEqual(doc["symbols"], sym_path) + win = [w for w in doc["windows"] if w["label"] == "pmc-probe"][0] + entry = win["unclassified"]["accesses"][0] + self.assertEqual(entry["addr"], "0x%x" % PMC_BOOT_42) + self.assertEqual(entry["rw"], "r") + self.assertIsNone(entry["value"]) # observe-only: no fake read + self.assertFalse(entry["served"]) + self.assertEqual(entry["pc_symbol"], "boot_probe") + self.assertIsInstance(entry["vt"], int) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_gmmu_oracle.py b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_oracle.py new file mode 100644 index 0000000..9211c81 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_oracle.py @@ -0,0 +1,397 @@ +"""Tests for the GMMU/FBIF oracle comparison (harness/gmmu_oracle.py, docs/44). + +All fixtures are synthetic and written to temporary directories: no oracle +capture, no hardware, no simulator run, no QEMU, no network. The comparison +is offline file-in/file-out. Tests that would need real inputs are written +to assert the clean error path instead. + +Run:: + + python3 -m unittest tests.test_gmmu_oracle -v +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import sys +import tempfile +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 harness.gmmu_debug import GmmuInputError, TRACE_FORMAT # noqa: E402 +from harness.gmmu_oracle import ( # noqa: E402 + ORACLE_FORMAT, + REPORT_FORMAT, + compare, + load_oracle, + load_sim_trace, + main, + print_capture_hints, + run_compare, +) + + +def _write_json(path: str, payload) -> str: + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh) + return path + + +def _oracle_items() -> list[dict]: + return [ + { + "name": "FBIF_CTL", + "window": "falcon", + "offset": "0x624", + "value": "0x1", + "source": "rm-readback", + "captured_at": "2026-10-03", + }, + { + "name": "GMMU_PDB_LO", + "window": "gmmu", + "offset": "0x0", + "value": "0x400000", + "source": "rm-readback", + "captured_at": "2026-10-03", + }, + ] + + +def _sim_events() -> list[dict]: + return [ + { + "seq": 0, + "pc": "0x4002040", + "kind": "write", + "window": "falcon", + "offset": "0x624", + "name": "FBIF_CTL", + "size": 4, + "value": "0x1", + }, + { + "seq": 1, + "pc": "0x4001010", + "kind": "write", + "window": "gmmu", + "offset": "0x0", + "name": "GMMU_PDB_LO", + "size": 4, + "value": "0x400000", + }, + ] + + +def _run_main(argv: list[str]) -> tuple[int, str, str]: + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = main(argv) + return rc, out.getvalue(), err.getvalue() + + +class IngestTests(unittest.TestCase): + def test_oracle_object_and_bare_list(self): + with tempfile.TemporaryDirectory() as tmp: + tagged = _write_json( + os.path.join(tmp, "o1.json"), + {"format": ORACLE_FORMAT, "captures": _oracle_items()}, + ) + bare = _write_json(os.path.join(tmp, "o2.json"), _oracle_items()) + self.assertEqual(load_oracle(tagged).format, ORACLE_FORMAT) + self.assertEqual(load_oracle(bare).format, "unlabeled-list") + self.assertEqual(len(load_oracle(tagged).items), 2) + self.assertEqual(load_oracle(tagged).items[0].value, 1) + self.assertEqual(load_oracle(tagged).items[1].offset, 0) + + def test_oracle_wrong_format_is_clean_error(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_json( + os.path.join(tmp, "o.json"), {"format": "something-else", "captures": []} + ) + with self.assertRaises(GmmuInputError): + load_oracle(path) + + def test_oracle_missing_file_is_clean_error(self): + with self.assertRaises(GmmuInputError): + load_oracle("/does/not/exist.json") + + def test_trace_aliases(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_json( + os.path.join(tmp, "t.json"), + { + "format": TRACE_FORMAT, + "events": [ + { + "pc": 0x4002040, + "rw": "w", + "block": "falcon", + "reg_offset": 0x624, + "register": "FBIF_CTL", + "value": 1, + } + ], + }, + ) + trace = load_sim_trace(path) + obs = trace.observations[0] + self.assertEqual(obs.kind, "write") + self.assertEqual(obs.window, "falcon") + self.assertEqual(obs.offset, 0x624) + self.assertEqual(obs.name, "FBIF_CTL") + self.assertEqual(obs.value, 1) + + def test_trace_missing_file_is_clean_error(self): + with self.assertRaises(GmmuInputError): + load_sim_trace("/does/not/exist.json") + + +class ComparisonTests(unittest.TestCase): + def _compare(self, items, events): + with tempfile.TemporaryDirectory() as tmp: + oracle_path = _write_json( + os.path.join(tmp, "oracle.json"), + {"format": ORACLE_FORMAT, "captures": items}, + ) + trace_path = _write_json( + os.path.join(tmp, "trace.json"), {"format": TRACE_FORMAT, "events": events} + ) + return compare(load_oracle(oracle_path), load_sim_trace(trace_path)) + + def test_full_agreement(self): + report = self._compare(_oracle_items(), _sim_events()) + summary = report["summary"] + self.assertEqual(summary["oracle_items"], 2) + self.assertEqual(summary["compared"], 2) + self.assertEqual(summary["agreed"], 2) + self.assertEqual(summary["mismatched"], 0) + self.assertEqual(summary["oracle_only"], 0) + self.assertEqual(summary["simulator_only"], 0) + self.assertEqual(summary["agreement_pct"], 100.0) + self.assertEqual(summary["coverage_pct"], 100.0) + self.assertTrue(all(row["status"] == "agree" for row in report["rows"])) + self.assertTrue(all(row["fidelity"] == "observed vs modelled" for row in report["rows"])) + + def test_mismatch_is_reported(self): + events = _sim_events() + events[0]["value"] = "0x2" + report = self._compare(_oracle_items(), events) + summary = report["summary"] + self.assertEqual(summary["mismatched"], 1) + self.assertEqual(summary["agreed"], 1) + self.assertEqual(summary["agreement_pct"], 50.0) + mismatch = next(row for row in report["rows"] if row["status"] == "mismatch") + self.assertEqual(mismatch["oracle_value"], 1) + self.assertEqual(mismatch["sim_value"], 2) + + def test_missing_simulator_observation(self): + report = self._compare(_oracle_items(), _sim_events()[:1]) + summary = report["summary"] + self.assertEqual(summary["oracle_only"], 1) + self.assertEqual(summary["coverage_pct"], 50.0) + self.assertEqual(len(report["unmatched_oracle"]), 1) + self.assertEqual(report["unmatched_oracle"][0]["register"], "GMMU_PDB_LO") + oracle_only = next(row for row in report["rows"] if row["status"] == "oracle-only") + self.assertIn("observed", oracle_only["fidelity"]) + + def test_missing_oracle_counterpart(self): + extra = { + "seq": 9, + "pc": "0x4003010", + "kind": "write", + "window": "gmmu", + "offset": "0x200", + "name": "PTE_DATA", + "value": "0x1", + } + report = self._compare(_oracle_items(), _sim_events() + [extra]) + self.assertEqual(report["summary"]["simulator_only"], 1) + self.assertEqual(len(report["unmatched_simulator"]), 1) + self.assertEqual(report["unmatched_simulator"][0]["register"], "PTE_DATA") + self.assertEqual(report["summary"]["simulator_match_pct"], 66.67) + + def test_name_fallback_when_oracle_has_no_offset(self): + items = [{"name": "IRQSTAT", "window": "falcon", "value": "0x0", "source": "rm"}] + events = [ + { + "seq": 0, + "pc": 0x4002050, + "kind": "read", + "window": "falcon", + "offset": "0x8", + "name": "IRQSTAT", + "value": "0x0", + } + ] + report = self._compare(items, events) + self.assertEqual(report["summary"]["agreed"], 1) + self.assertEqual(report["rows"][0]["matched_by"], "window-name") + + def test_no_value_is_not_agreement(self): + items = [{"name": "FBIF_CTL", "window": "falcon", "offset": "0x624"}] + report = self._compare(items, _sim_events()[:1]) + self.assertEqual(report["summary"]["no_value"], 1) + self.assertEqual(report["summary"]["agreement_pct"], None) + + def test_g1_nested_document_compares(self): + document = { + "schema": TRACE_FORMAT, + "image_sha256": "ab" * 32, + "profile": "tu104-570.144", + "windows": [ + { + "label": "falcon-observe", + "registers": [ + { + "offset": "0x624", + "name": "FBIF_CTL", + "access": "rw", + "accesses": [ + {"addr": "0x300110624", "offset": "0x624", + "size": 4, "rw": "w", "value": "0x1", + "pc": "0x4002040", "window": "falcon-observe", + "register": "FBIF_CTL", "served": False, + "pc_symbol": None}, + ], + } + ], + } + ], + } + with tempfile.TemporaryDirectory() as tmp: + oracle_path = _write_json( + os.path.join(tmp, "oracle.json"), + { + "format": ORACLE_FORMAT, + "captures": [ + {"name": "FBIF_CTL", "window": "falcon-observe", + "offset": "0x624", "value": "0x1", "source": "rm"} + ], + }, + ) + trace_path = _write_json(os.path.join(tmp, "trace.json"), document) + report = compare(load_oracle(oracle_path), load_sim_trace(trace_path)) + self.assertEqual(report["summary"]["oracle_items"], 1) + self.assertEqual(report["summary"]["agreed"], 1) + self.assertEqual(report["inputs"]["simulator"]["format"], TRACE_FORMAT) + + def test_empty_inputs(self): + report = self._compare([], []) + summary = report["summary"] + self.assertEqual(summary["oracle_items"], 0) + self.assertEqual(summary["compared"], 0) + self.assertEqual(summary["agreement_pct"], None) + self.assertEqual(summary["coverage_pct"], None) + self.assertEqual(report["rows"], []) + self.assertEqual(report["unmatched_simulator"], []) + + +class SessionIngestTests(unittest.TestCase): + def test_session_directory_with_blobs_trace(self): + with tempfile.TemporaryDirectory() as tmp: + session = os.path.join(tmp, "session") + blobs = os.path.join(session, "blobs") + os.makedirs(blobs) + _write_json( + os.path.join(session, "session.json"), + {"format": "greeneye-session/1", "meta": {}}, + ) + _write_json( + os.path.join(blobs, "gmmu-fbif-trace.json"), + {"format": TRACE_FORMAT, "events": _sim_events()}, + ) + trace = load_sim_trace(session) + self.assertEqual(len(trace.observations), 2) + self.assertEqual(trace.format, TRACE_FORMAT) + + def test_session_without_trace_is_clean_error(self): + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(GmmuInputError): + load_sim_trace(tmp) + + +class ReportArtifactTests(unittest.TestCase): + def test_run_compare_writes_json_and_markdown(self): + with tempfile.TemporaryDirectory() as tmp: + oracle_path = _write_json( + os.path.join(tmp, "oracle.json"), + {"format": ORACLE_FORMAT, "captures": _oracle_items()}, + ) + trace_path = _write_json( + os.path.join(tmp, "trace.json"), + {"format": TRACE_FORMAT, "events": _sim_events()}, + ) + out = os.path.join(tmp, "out") + report = run_compare(oracle_path, trace_path, out) + self.assertEqual(report["format"], REPORT_FORMAT) + self.assertTrue(os.path.isfile(report["artifacts"]["report_json"])) + self.assertTrue(os.path.isfile(report["artifacts"]["report_md"])) + with open(report["artifacts"]["report_json"], "r", encoding="utf-8") as fh: + loaded = json.load(fh) + self.assertEqual(loaded["summary"]["agreed"], 2) + with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh: + md = fh.read() + self.assertIn("oracle vs simulator report", md) + self.assertIn("No cycle/timing claim", md) + self.assertIn("Oracle items with no simulator observation", md) + + +class CliTests(unittest.TestCase): + def test_print_capture_hints_executes_nothing(self): + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + print_capture_hints(out) + text = out.getvalue() + self.assertIn("SUPERVISED, READ-ONLY, HUMAN-RUN ONLY", text) + self.assertIn("executes NOTHING", text) + self.assertIn("RUNBOOK P1/P7", text) + self.assertEqual(err.getvalue(), "") + + def test_print_capture_hints_cli(self): + rc, stdout, _stderr = _run_main(["--print-capture-hints"]) + self.assertEqual(rc, 0) + self.assertIn("rmharness/harness.sh", stdout) + self.assertIn("gmmu_oracle.py compare", stdout) + + def test_compare_cli_writes_report(self): + with tempfile.TemporaryDirectory() as tmp: + oracle_path = _write_json( + os.path.join(tmp, "oracle.json"), + {"format": ORACLE_FORMAT, "captures": _oracle_items()}, + ) + trace_path = _write_json( + os.path.join(tmp, "trace.json"), + {"format": TRACE_FORMAT, "events": _sim_events()}, + ) + out = os.path.join(tmp, "out") + rc, stdout, _stderr = _run_main( + ["compare", "--oracle", oracle_path, "--sim", trace_path, "--out", out] + ) + self.assertEqual(rc, 0) + payload = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1]) + self.assertEqual(payload["summary"]["agreed"], 2) + self.assertTrue(os.path.isfile(os.path.join(out, "gmmu-oracle-report.json"))) + + def test_compare_missing_inputs_exits_2(self): + with tempfile.TemporaryDirectory() as tmp: + rc, _stdout, stderr = _run_main( + ["compare", "--oracle", "/nope.json", "--sim", "/nope.json", "--out", tmp] + ) + self.assertEqual(rc, 2) + self.assertIn("oracle not found", stderr) + + def test_no_command_prints_help(self): + rc, stdout, _stderr = _run_main([]) + self.assertEqual(rc, 2) + self.assertIn("usage", stdout.lower()) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_gmmu_pte.py b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_pte.py new file mode 100644 index 0000000..2460f1f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_pte.py @@ -0,0 +1,534 @@ +"""Golden-vector tests for the clean-room TU104 GMMU PTE/PDE decoder. + +Every vector is constructed field-by-field from the bit positions documented +in ``sim/gmmu_pte.py`` (see ``docs/43-GMMU-PTE-DECODER.md``), then decoded and asserted field-by-field. +The tests also pin the explicit failure behaviour: reserved bits are +reported, unsupported encodings/levels error instead of guessing, empty and +truncated blobs are handled deterministically, and decode output is +repeatable. +""" +import contextlib +import io +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sim.gmmu_pte import ( # noqa: E402 + APERTURE_NAMES_PDE, + APERTURE_NAMES_PTE, + E_BAD_VALUE, + E_TRUNCATED, + E_UNSUPPORTED, + GmmuDecodeError, + KIND_VER2, + KIND_VER2_DUAL, + LEVELS, + PTE_KIND_NAMES, + SUPPORTED_LEVELS, + decode_dual_pde, + decode_entry, + decode_pde, + decode_pte, + decode_table, + main, +) + + +def make_pte(*, valid=1, aperture=2, volatile=1, encrypted=0, privilege=0, + read_only=0, atomic_disable=1, address=0, comptagline=0, + peer_id=0, kind=0x06): + value = 0 + value |= (valid & 1) << 0 + value |= (aperture & 3) << 1 + value |= (volatile & 1) << 3 + value |= (encrypted & 1) << 4 + value |= (privilege & 1) << 5 + value |= (read_only & 1) << 6 + value |= (atomic_disable & 1) << 7 + if aperture in (2, 3): + value |= ((address >> 12) & ((1 << 46) - 1)) << 8 + else: + value |= ((address >> 12) & ((1 << 25) - 1)) << 8 + value |= (peer_id & 7) << 33 + value |= (comptagline & ((1 << 20) - 1)) << 36 + value |= (kind & 0xFF) << 56 + return value + + +def make_pde(*, aperture=1, volatile=1, no_ats=0, address=0, peer_id=0): + value = 0 + value |= (aperture & 3) << 1 + value |= (volatile & 1) << 3 + value |= (no_ats & 1) << 5 + value |= ((address >> 12) & ((1 << 46) - 1)) << 8 + value |= (peer_id & 7) << 33 + return value + + +def make_dual(*, aperture_big=1, volatile_big=1, aperture_small=1, + volatile_small=1, address_big=0, address_small=0, + peer_big=0, peer_small=0, no_ats=0): + value = 0 + value |= (aperture_big & 3) << 1 + value |= (volatile_big & 1) << 3 + value |= ((address_big >> 8) & ((1 << 50) - 1)) << 4 + value |= (peer_big & 7) << 33 + value |= (aperture_small & 3) << 65 + value |= (volatile_small & 1) << 67 + value |= ((address_small >> 12) & ((1 << 46) - 1)) << 72 + value |= (peer_small & 7) << 97 + value |= (no_ats & 1) << 5 + return value + + +class GoldenVectorTests(unittest.TestCase): + """Field-by-field constructed vectors must round-trip exactly.""" + + def test_pte_system_coherent_round_trip(self): + pa = 0x123456789000 + value = make_pte(valid=1, aperture=2, volatile=1, encrypted=0, + privilege=1, read_only=1, atomic_disable=0, + address=pa, kind=0x06) + d = decode_pte(value) + self.assertEqual(d["role"], "pte") + self.assertTrue(d["valid"]) + self.assertEqual(d["aperture"], 2) + self.assertEqual(d["aperture_name"], "system_coherent_memory") + self.assertTrue(d["volatile"]) + self.assertFalse(d["encrypted"]) + self.assertTrue(d["privilege"]) + self.assertTrue(d["read_only"]) + self.assertFalse(d["writable"]) + self.assertFalse(d["atomic_disable"]) + self.assertEqual(d["address"], pa) + self.assertTrue(d["address_complete"]) + self.assertEqual(d["kind"], 0x06) + self.assertEqual(d["kind_name"], "GENERIC_MEMORY") + self.assertTrue(d["kind_known"]) + + def test_pte_system_non_coherent_round_trip(self): + pa = 0x2AAAAA0000 + value = make_pte(aperture=3, address=pa, read_only=0, + atomic_disable=1, kind=0x00) + d = decode_pte(value) + self.assertEqual(d["aperture_name"], "system_non_coherent_memory") + self.assertEqual(d["address"], pa) + self.assertTrue(d["writable"]) + self.assertTrue(d["atomic_disable"]) + self.assertEqual(d["kind_name"], "PITCH") + + def test_pte_video_aperture_round_trip(self): + pa_low = 0x0ABCDEF000 + value = make_pte(aperture=0, address=pa_low, comptagline=0x12345, + kind=0x06) + d = decode_pte(value) + self.assertEqual(d["aperture_name"], "video_memory") + self.assertEqual(d["address"], pa_low) + self.assertFalse(d["address_complete"]) + self.assertEqual(d["comptagline"], 0x12345) + self.assertEqual(d["peer_id"], 0) + + def test_pte_peer_aperture_round_trip(self): + pa_low = 0x0100000000 + value = make_pte(aperture=1, address=pa_low, peer_id=5, kind=0x06) + d = decode_pte(value) + self.assertEqual(d["aperture_name"], "peer_memory") + self.assertEqual(d["address"], pa_low) + self.assertFalse(d["address_complete"]) + self.assertEqual(d["peer_id"], 5) + + def test_pte_aperture_names_are_the_proven_four(self): + self.assertEqual( + APERTURE_NAMES_PTE, + {0: "video_memory", 1: "peer_memory", + 2: "system_coherent_memory", 3: "system_non_coherent_memory"}) + for aperture in range(4): + d = decode_pte(make_pte(aperture=aperture)) + self.assertEqual(d["aperture_name"], APERTURE_NAMES_PTE[aperture]) + + def test_pde_video_pointer_round_trip(self): + pa = 0x00000123456000 + value = make_pde(aperture=1, volatile=1, no_ats=0, address=pa) + d = decode_pde(value) + self.assertEqual(d["role"], "pde") + self.assertTrue(d["is_pde_pointer"]) + self.assertFalse(d["is_leaf"]) + self.assertEqual(d["aperture_name"], "video_memory") + self.assertTrue(d["volatile"]) + self.assertFalse(d["no_ats"]) + self.assertEqual(d["address"], pa) + self.assertFalse(d["address_complete"]) + + def test_pde_system_pointer_round_trip(self): + pa = 0x1FFFF0000000 + value = make_pde(aperture=2, volatile=0, no_ats=1, address=pa) + d = decode_pde(value) + self.assertEqual(d["aperture_name"], "system_coherent_memory") + self.assertFalse(d["volatile"]) + self.assertTrue(d["no_ats"]) + self.assertEqual(d["address"], pa) + self.assertTrue(d["address_complete"]) + + def test_pde_invalid_aperture_has_no_address(self): + d = decode_pde(make_pde(aperture=0, volatile=0)) + self.assertEqual(d["aperture_name"], "invalid") + self.assertIsNone(d["address"]) + self.assertFalse(d["address_complete"]) + + def test_pde_leaf_bit_is_rejected(self): + with self.assertRaises(GmmuDecodeError) as ctx: + decode_pde(1) + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + + def test_dual_pde_halves_round_trip(self): + addr_big = 0x0000001234560000 + addr_small = 0x0000000FEDCBA000 + value = make_dual(aperture_big=1, volatile_big=1, aperture_small=1, + volatile_small=1, address_big=addr_big, + address_small=addr_small) + d = decode_dual_pde(value) + self.assertEqual(d["role"], "dual_pde") + self.assertTrue(d["is_pde_pointer"]) + self.assertFalse(d["is_leaf"]) + self.assertFalse(d["no_ats"]) + self.assertEqual(d["big"]["aperture_name"], "video_memory") + self.assertTrue(d["big"]["volatile"]) + self.assertEqual(d["big"]["address"], addr_big) + self.assertFalse(d["big"]["address_complete"]) + self.assertEqual(d["small"]["aperture_name"], "video_memory") + self.assertTrue(d["small"]["volatile"]) + self.assertEqual(d["small"]["address"], addr_small) + self.assertFalse(d["small"]["address_complete"]) + + def test_dual_pde_no_ats_bit_does_not_perturb_big_address(self): + # The format names bit 5 NO_ATS inside the BIG address field; valid + # depth-3 allocations are 4 KB aligned, so address bit 9 is zero and + # the flag must decode without changing the reconstructed address. + addr_big = 0x0000001234560000 + d = decode_dual_pde(make_dual(address_big=addr_big, no_ats=1)) + self.assertTrue(d["no_ats"]) + self.assertEqual(d["big"]["address"], addr_big) + self.assertEqual(d["big"]["address_vid"] & (1 << 1), 1 << 1) + + def test_dual_pde_system_halves_are_complete(self): + value = make_dual(aperture_big=2, aperture_small=3, + address_big=0x1000000000, + address_small=0x2000000000) + d = decode_dual_pde(value) + self.assertEqual(d["big"]["aperture_name"], "system_coherent_memory") + self.assertTrue(d["big"]["address_complete"]) + self.assertEqual(d["big"]["address"], 0x1000000000) + self.assertEqual(d["small"]["aperture_name"], + "system_non_coherent_memory") + self.assertTrue(d["small"]["address_complete"]) + self.assertEqual(d["small"]["address"], 0x2000000000) + + def test_dual_pde_peer_ids_round_trip(self): + value = make_dual(aperture_big=1, aperture_small=1, peer_big=3, + peer_small=6) + d = decode_dual_pde(value) + self.assertEqual(d["big"]["peer_id"], 3) + self.assertEqual(d["small"]["peer_id"], 6) + + def test_pde_aperture_names_are_the_proven_four(self): + self.assertEqual( + APERTURE_NAMES_PDE, + {0: "invalid", 1: "video_memory", + 2: "system_coherent_memory", 3: "system_non_coherent_memory"}) + for aperture in range(4): + d = decode_pde(make_pde(aperture=aperture)) + self.assertEqual(d["aperture_name"], APERTURE_NAMES_PDE[aperture]) + + def test_known_kind_names_spot_values(self): + self.assertEqual(PTE_KIND_NAMES[0x06], "GENERIC_MEMORY") + self.assertEqual(PTE_KIND_NAMES[0x00], "PITCH") + self.assertEqual(PTE_KIND_NAMES[0x07], "INVALID") + self.assertEqual(PTE_KIND_NAMES[0x0F], "SMSKED_MESSAGE") + self.assertEqual(PTE_KIND_NAMES[0xFE], "GENERIC_16BX2") + self.assertEqual(len(PTE_KIND_NAMES), 243) + + def test_unknown_kind_field_is_reported_not_guessed(self): + value = make_pte(kind=0xFF) + d = decode_pte(value) + self.assertEqual(d["kind"], 0xFF) + self.assertIsNone(d["kind_name"]) + self.assertFalse(d["kind_known"]) + + def test_level_metadata(self): + self.assertEqual([lv["depth"] for lv in LEVELS], + list(SUPPORTED_LEVELS)) + self.assertEqual(decode_pte(make_pte(), level=3)["page_size"], "2m") + self.assertEqual(decode_pte(make_pte(), level=4)["page_size"], + "4k_or_64k") + self.assertEqual(decode_pte(make_pte())["level_candidates"], [3, 4]) + + +class ReservedBitTests(unittest.TestCase): + """Set bits with no named field must be reported, never dropped.""" + + def test_pde_all_bits_set_reports_exact_reserved_set(self): + d = decode_pde(((1 << 64) - 1) & ~1) # bit 0 clear = directory entry + self.assertEqual(d["reserved_bits"], [4, 6, 7] + list(range(54, 64))) + self.assertEqual(d["reserved_mask"], + sum(1 << b for b in d["reserved_bits"])) + + def test_pde_named_bits_are_not_reported(self): + d = decode_pde(make_pde(aperture=2, volatile=1, no_ats=1, + address=0x12345000)) + self.assertEqual(d["reserved_bits"], []) + + def test_dual_pde_all_bits_set_reports_exact_reserved_set(self): + d = decode_dual_pde((1 << 128) - 1) + expected = list(range(54, 65)) + list(range(68, 72)) + \ + list(range(118, 128)) + self.assertEqual(d["reserved_bits"], expected) + self.assertEqual(d["reserved_mask"], + sum(1 << b for b in expected)) + + def test_dual_pde_unnamed_small_half_bit_64_is_reported(self): + d = decode_dual_pde(1 << 64) + self.assertEqual(d["reserved_bits"], [64]) + + def test_pte_has_no_reserved_bits(self): + d = decode_pte((1 << 64) - 1) + self.assertEqual(d["reserved_bits"], []) + self.assertEqual(d["reserved_mask"], 0) + + +class ErrorBehaviourTests(unittest.TestCase): + """Unsupported requests error; they never guess.""" + + def test_unknown_encoding_kind_errors(self): + with self.assertRaises(GmmuDecodeError) as ctx: + decode_pte(0, pte_kind="legacy") + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + with self.assertRaises(GmmuDecodeError) as ctx: + decode_pde(0, pde_kind="ver1") + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + with self.assertRaises(GmmuDecodeError) as ctx: + decode_table(b"", base_pa=0, kind="nope") + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + + def test_wrong_role_for_encoding_errors(self): + with self.assertRaises(GmmuDecodeError) as ctx: + decode_pte(0, pte_kind=KIND_VER2_DUAL) + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + with self.assertRaises(GmmuDecodeError) as ctx: + decode_dual_pde(0, pde_kind=KIND_VER2) + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + + def test_unsupported_levels_error(self): + value = make_pde().to_bytes(8, "little") + for bad in (5, 99, -1, (0, 7)): + with self.assertRaises(GmmuDecodeError) as ctx: + decode_table(value, base_pa=0, levels=bad) + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + with self.assertRaises(GmmuDecodeError) as ctx: + decode_pte(0, level=0) + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + + def test_entry_size_must_match_kind(self): + with self.assertRaises(GmmuDecodeError) as ctx: + decode_table(b"\x00" * 8, base_pa=0, kind=KIND_VER2_DUAL) + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + with self.assertRaises(GmmuDecodeError) as ctx: + decode_table(b"\x00" * 16, base_pa=0, entry_size=16) + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + + def test_bad_values_error(self): + with self.assertRaises(GmmuDecodeError) as ctx: + decode_pte(1 << 64) + self.assertEqual(ctx.exception.code, E_BAD_VALUE) + with self.assertRaises(GmmuDecodeError) as ctx: + decode_dual_pde(-1) + self.assertEqual(ctx.exception.code, E_BAD_VALUE) + with self.assertRaises(GmmuDecodeError) as ctx: + decode_table(b"", base_pa=-1) + self.assertEqual(ctx.exception.code, E_BAD_VALUE) + + +class TableDecodeTests(unittest.TestCase): + def test_empty_blob_is_empty_list(self): + self.assertEqual(decode_table(b"", base_pa=0x1000), []) + self.assertEqual(decode_table(b"", base_pa=0x1000, kind=KIND_VER2_DUAL, + entry_size=16), []) + + def test_short_blob_errors(self): + for blob in (b"\x01", b"\x01\x02\x03", b"\x00" * 4, b"\x00" * 12): + with self.assertRaises(GmmuDecodeError) as ctx: + decode_table(blob, base_pa=0) + self.assertEqual(ctx.exception.code, E_TRUNCATED) + with self.assertRaises(GmmuDecodeError) as ctx: + decode_table(b"\x00" * 24, base_pa=0, kind=KIND_VER2_DUAL, + entry_size=16) + self.assertEqual(ctx.exception.code, E_TRUNCATED) + + def test_two_entry_table_round_trip_and_addresses(self): + pte = make_pte(aperture=2, address=0x40000000) + pde = make_pde(aperture=1, address=0x50000000) + blob = pte.to_bytes(8, "little") + pde.to_bytes(8, "little") + entries = decode_table(blob, base_pa=0x1000) + self.assertEqual(len(entries), 2) + self.assertEqual(entries[0]["index"], 0) + self.assertEqual(entries[0]["entry_pa"], 0x1000) + self.assertEqual(entries[0]["role"], "pte") + self.assertEqual(entries[0]["address"], 0x40000000) + self.assertEqual(entries[1]["index"], 1) + self.assertEqual(entries[1]["entry_pa"], 0x1008) + self.assertEqual(entries[1]["role"], "pde") + self.assertEqual(entries[1]["address"], 0x50000000) + + def test_auto_role_uses_bit_zero_and_reports_inference(self): + entries = decode_table(make_pte().to_bytes(8, "little"), + base_pa=0) + self.assertEqual(entries[0]["role"], "pte") + self.assertTrue(entries[0]["role_inferred"]) + self.assertIsNone(entries[0]["level"]) + self.assertEqual(entries[0]["level_candidates"], [3, 4]) + entries = decode_table(make_pde().to_bytes(8, "little"), base_pa=0) + self.assertEqual(entries[0]["role"], "pde") + self.assertTrue(entries[0]["role_inferred"]) + self.assertEqual(entries[0]["level_candidates"], [0, 1, 2, 3]) + + def test_pinned_level_selects_role(self): + pte = make_pte() + pde = make_pde() + leaf = decode_table(pte.to_bytes(8, "little"), base_pa=0, levels=3) + self.assertEqual((leaf[0]["role"], leaf[0]["level"]), ("pte", 3)) + self.assertEqual(leaf[0]["page_size"], "2m") + ptr = decode_table(pde.to_bytes(8, "little"), base_pa=0, levels=3) + self.assertEqual((ptr[0]["role"], ptr[0]["level"]), ("pde", 3)) + upper = decode_table(pde.to_bytes(8, "little"), base_pa=0, levels=1) + self.assertEqual((upper[0]["role"], upper[0]["level"]), ("pde", 1)) + page = decode_table(pte.to_bytes(8, "little"), base_pa=0, levels=4) + self.assertEqual((page[0]["role"], page[0]["level"]), ("pte", 4)) + + def test_level_sequence_narrows_candidates(self): + entries = decode_table(make_pte().to_bytes(8, "little"), base_pa=0, + levels=(3, 4)) + self.assertEqual(entries[0]["level_candidates"], [3, 4]) + entries = decode_table(make_pde().to_bytes(8, "little"), base_pa=0, + levels=(3, 4)) + self.assertEqual(entries[0]["level_candidates"], [3]) + + def test_dual_table_entry(self): + value = make_dual(address_big=0x1000, address_small=0x2000) + entries = decode_table(value.to_bytes(16, "little"), base_pa=0x2000, + kind=KIND_VER2_DUAL, entry_size=16) + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0]["role"], "dual_pde") + self.assertEqual(entries[0]["entry_pa"], 0x2000) + self.assertEqual(entries[0]["encoding"], KIND_VER2_DUAL) + self.assertFalse(entries[0]["role_inferred"]) + self.assertEqual(entries[0]["big"]["address"], 0x1000) + self.assertEqual(entries[0]["small"]["address"], 0x2000) + + def test_decode_entry_explicit_roles(self): + pte = decode_entry(make_pte(), role="pte") + self.assertEqual(pte["role"], "pte") + pde = decode_entry(make_pde(), role="pde") + self.assertEqual(pde["role"], "pde") + dual = decode_entry(make_dual(), kind=KIND_VER2_DUAL, role="dual_pde") + self.assertEqual(dual["role"], "dual_pde") + with self.assertRaises(GmmuDecodeError) as ctx: + decode_entry(0, role="bogus") + self.assertEqual(ctx.exception.code, E_UNSUPPORTED) + + def test_output_is_deterministic(self): + blob = (make_pte(aperture=2, address=0x1000).to_bytes(8, "little") + + make_pde(aperture=1, address=0x2000).to_bytes(8, "little") + + make_dual(address_big=0x3000, + address_small=0x4000).to_bytes(16, "little")) + first = decode_table(blob, base_pa=0x1000) + second = decode_table(blob, base_pa=0x1000) + self.assertEqual(first, second) + self.assertEqual(json.dumps(first, sort_keys=True), + json.dumps(second, sort_keys=True)) + + def test_blob_accepts_bytearray_and_memoryview(self): + pte = make_pte(aperture=2, address=0x1000).to_bytes(8, "little") + for blob in (bytearray(pte), memoryview(pte)): + entries = decode_table(blob, base_pa=0) + self.assertEqual(entries[0]["address"], 0x1000) + + +class CliTests(unittest.TestCase): + def test_value_json(self): + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + rc = main(["--value", hex(make_pte(aperture=2, address=0x4000)), + "--role", "pte", "--json"]) + self.assertEqual(rc, 0) + data = json.loads(buf.getvalue()) + self.assertEqual(len(data), 1) + self.assertEqual(data[0]["role"], "pte") + self.assertEqual(data[0]["address"], 0x4000) + + def test_value_level_is_honoured(self): + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + rc = main(["--value", hex(make_pte()), "--role", "pte", + "--levels", "3", "--json"]) + self.assertEqual(rc, 0) + data = json.loads(buf.getvalue()) + self.assertEqual(data[0]["level"], 3) + self.assertEqual(data[0]["page_size"], "2m") + err = io.StringIO() + with contextlib.redirect_stderr(err): + rc = main(["--value", hex(make_pte()), "--role", "pte", + "--levels", "3,4"]) + self.assertEqual(rc, 2) + self.assertIn(E_BAD_VALUE, err.getvalue()) + + def test_blob_json(self): + blob = (make_pte(aperture=2, address=0x1000).to_bytes(8, "little") + + make_pde(aperture=1, address=0x2000).to_bytes(8, "little")) + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "table.bin") + with open(path, "wb") as handle: + handle.write(blob) + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + rc = main(["--blob", path, "--base-pa", "0x1000", "--json"]) + self.assertEqual(rc, 0) + data = json.loads(buf.getvalue()) + self.assertEqual([e["role"] for e in data], ["pte", "pde"]) + self.assertEqual(data[1]["entry_pa"], 0x1008) + + def test_text_output_is_deterministic(self): + value = hex(make_pte(aperture=2, address=0x4000)) + outputs = [] + for _ in range(2): + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + rc = main(["--value", value, "--role", "pte"]) + self.assertEqual(rc, 0) + outputs.append(buf.getvalue()) + self.assertEqual(outputs[0], outputs[1]) + self.assertIn("role=pte", outputs[0]) + + def test_truncated_blob_exits_two(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "short.bin") + with open(path, "wb") as handle: + handle.write(b"\x01\x02\x03") + err = io.StringIO() + with contextlib.redirect_stderr(err): + rc = main(["--blob", path, "--base-pa", "0x1000"]) + self.assertEqual(rc, 2) + self.assertIn(E_TRUNCATED, err.getvalue()) + + def test_unsupported_level_exits_two(self): + err = io.StringIO() + with contextlib.redirect_stderr(err): + rc = main(["--value", "0x1", "--role", "pde", "--levels", "9"]) + self.assertEqual(rc, 2) + self.assertIn(E_UNSUPPORTED, err.getvalue()) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_mpu.py b/debug/risc-v-gsp-rm-vm/tests/test_mpu.py new file mode 100644 index 0000000..73ac81f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_mpu.py @@ -0,0 +1,177 @@ +"""MMU (NVMPU) model tests: CSR semantics, translation, machine integration. + +The model is exercised directly and through the machine's guest paths; the +real-firmware relocation-copy success is covered by +``tests/test_sim_devices.py`` (boot ladder). All fixtures are synthetic; no +hardware, module, guest or network is touched. +""" +from __future__ import annotations + +import unittest + +from sim.machine import Machine, R_EBREAK +from sim.mpu import ( + MSTATUS_VM_NVMPU, + MMPUATTR, + MMPUIDX, + MMPUPA, + MMPURNG, + MMPUVA, + RiscvMpu, +) + +KB = 1024 + + +def enable(mpu: RiscvMpu) -> None: + mpu.on_csr_write(0x300, MSTATUS_VM_NVMPU << 24) + + +def entry(mpu: RiscvMpu, index: int, va: int, pa: int, rng: int, attr: int = 0x600) -> None: + mpu.on_csr_write(MMPUIDX, index) + mpu.on_csr_write(MMPUPA, pa) + mpu.on_csr_write(MMPURNG, rng) + mpu.on_csr_write(MMPUATTR, attr) + mpu.on_csr_write(MMPUVA, va | 1) # VLD + + +class RiscvMpuTests(unittest.TestCase): + def test_disabled_does_not_translate(self): + mpu = RiscvMpu() + entry(mpu, 0, 0x80000000, 0x1000, 0x1000) + self.assertFalse(mpu.enabled) + self.assertIsNone(mpu.translate(0x80000000)) + + def test_enable_is_the_nvmpu_vm_mode(self): + mpu = RiscvMpu() + entry(mpu, 0, 0x80000000, 0x1000, 0x1000) + mpu.on_csr_write(0x300, 0x1E000000) + self.assertTrue(mpu.enabled) + # Bare mode disables translation again. + mpu.on_csr_write(0x300, 0) + self.assertFalse(mpu.enabled) + self.assertIsNone(mpu.translate(0x80000000)) + + def test_translate_range_and_offset(self): + mpu = RiscvMpu() + entry(mpu, 2, 0x400E000, 0x1000, 0x3000) + enable(mpu) + self.assertEqual(mpu.translate(0x400E000), 0x1000) + self.assertEqual(mpu.translate(0x400E800), 0x1800) + self.assertEqual(mpu.translate(0x4010FFF), 0x3FFF) + self.assertIsNone(mpu.translate(0x400DFFF)) + self.assertIsNone(mpu.translate(0x4011000)) + + def test_lowest_index_precedence(self): + mpu = RiscvMpu() + entry(mpu, 3, 0x1000, 0x9000, 0x1000) + entry(mpu, 0, 0x1000, 0x5000, 0x1000) + entry(mpu, 1, 0x1000, 0x7000, 0x1000) + enable(mpu) + self.assertEqual(mpu.translate(0x1000), 0x5000) + + def test_vld_clear_disables_entry(self): + mpu = RiscvMpu() + entry(mpu, 0, 0x80000000, 0x1000, 0x1000) + enable(mpu) + self.assertIsNotNone(mpu.translate(0x80000000)) + mpu.on_csr_write(MMPUIDX, 0) + mpu.on_csr_write(MMPUVA, 0) # VLD clear + self.assertIsNone(mpu.translate(0x80000000)) + + def test_index_field_is_six_bits(self): + mpu = RiscvMpu() + mpu.on_csr_write(MMPUIDX, 0x140) # 0x40 | 0x00 -> index 0 + entry(mpu, 0, 0x1000, 0x2000, 0x1000) + enable(mpu) + self.assertEqual(mpu.translate(0x1000), 0x2000) + + def test_state_and_reset(self): + mpu = RiscvMpu() + entry(mpu, 0, 0x80000000, 0x1000, 0x1000) + enable(mpu) + state = mpu.state() + self.assertTrue(state["enabled"]) + self.assertEqual(state["entries"][0]["index"], 0) + self.assertEqual(state["entries"][0]["pa"], 0x1000) + mpu.reset() + self.assertFalse(mpu.enabled) + self.assertIsNone(mpu.translate(0x80000000)) + + +class MachineMpuTests(unittest.TestCase): + def test_guest_load_store_through_mapping(self): + m = Machine(memory_mb=1, ram_base=0, unknown_mmio="fault") + m.write_mem(0x1000, b"\x11\x22\x33\x44\x55\x66\x77\x88") + entry(m.mpu, 0, 0x80000000, 0x1000, 0x1000) + enable(m.mpu) + self.assertEqual(m._guest_load(0x80000000, 8), + int.from_bytes(b"\x11\x22\x33\x44\x55\x66\x77\x88", "little")) + m._guest_store(0x80000004, 8, 0xAABBCCDD) + self.assertEqual(m.read_mem(0x1004, 4), b"\xdd\xcc\xbb\xaa") + # The debugger surface stays raw/strict: the mapped VA has no backing. + with self.assertRaises(Exception): + m.read_mem(0x80000000, 4) + + def test_instruction_fetch_is_translated(self): + m = Machine(memory_mb=1, ram_base=0, unknown_mmio="fault") + program = bytes.fromhex("13057000" "73001000") # addi a0,zero,7; ebreak + m.write_mem(0x2000, program) + entry(m.mpu, 0, 0x80002000, 0x2000, 0x1000) + enable(m.mpu) + m.pc = 0x80002000 # virtual entry; the backing lives at PA 0x2000 + reason, pc, _, _ = m.run(10) + self.assertEqual(reason, R_EBREAK) + self.assertEqual(m.x[10], 7) + self.assertEqual(pc, 0x80002004) + + def test_unmapped_fetch_still_faults(self): + m = Machine(memory_mb=1, ram_base=0, unknown_mmio="fault") + entry(m.mpu, 0, 0x80002000, 0x2000, 0x1000) + enable(m.mpu) + m.pc = 0x90000000 # no entry covers it, no backing RAM + reason, _, _, _ = m.run(1) + self.assertNotEqual(reason, R_EBREAK) + + +class Wpr2RelativeOffsetTests(unittest.TestCase): + def test_window_relative_offsets_read_the_same_pair(self): + from sim.devices.fwsec import Wpr2Block + + dev = Wpr2Block() + dev.set_range(0x2000000, 0x6000000) + self.assertEqual(dev.read(0x824, 4), dev.read(0x1FA824, 4)) + self.assertEqual(dev.read(0x828, 4), dev.read(0x1FA828, 4)) + self.assertIsNone(dev.read(0x820, 4)) + + +class BootShimArgsTests(unittest.TestCase): + def test_bootstrap_args_written_and_image_mirror_absent(self): + from sim.gsp_boot import _apply_boot_shim, DEFAULT_IMAGE, DEFAULT_PROFILE + from greeneye.elf import load_elf + from sim.devices import load_profile + + profile = load_profile(DEFAULT_PROFILE) + m = Machine(memory_mb=16, ram_base=0, unknown_mmio="lenient") + shim = dict(profile.get("boot_shim", {})) + shim["_image_path"] = DEFAULT_IMAGE + meta = {"enabled": True, "assumptions": []} + _apply_boot_shim(m, shim, load_elf(DEFAULT_IMAGE), meta) + args = profile["boot_shim"]["bootstrap_args"] + pa = int(args["pa"], 0) + import struct + + fields = struct.unpack("<6Q", m.read_mem(pa, 48)) + self.assertEqual(fields[0], int(args["imem_base_pa"], 0)) + self.assertEqual(fields[1], int(args["imem_size"], 0)) + self.assertEqual(fields[2], int(args["imem_bs_size"], 0)) + self.assertEqual(fields[3], int(args["dmem_base_pa"], 0)) + self.assertEqual(fields[4], int(args["dmem_size"], 0)) + self.assertEqual(fields[5], int(args["dmem_bs_size"], 0)) + # The DMEM image mirror hack is gone from this profile. + self.assertNotIn("image_at", shim) + self.assertNotIn("image_at", meta) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py index 8063ee9..13912b5 100644 --- a/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py +++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py @@ -428,18 +428,34 @@ class BootLadderTests(unittest.TestCase): self.assertEqual(r.mmio_unknown, 0) self.assertEqual(r.device_unhandled, 0) - def test_shim_reaches_relocation_b2(self): + def test_shim_reaches_message_loop_b3(self): + from sim.gsp_boot import run_boot + r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, + instructions=2_000_000, record=False, use_shim=True) + self.assertTrue(r.ran) + self.assertEqual(r.rung, "B3") + self.assertIn("relocation_copy", r.milestones) + self.assertEqual(r.mmio_unknown, 0) + self.assertEqual(r.device_unhandled, 0) + # The MMU model resolves the resident-section copy, the libos service + # path runs (DMA/pagetable registration) and the message-loop task + # blocks in the port wait -- deterministic, not faked. + self.assertEqual(r.stop_reason, "ecall") + self.assertEqual(r.stop_pc, 0x4018038) + stopped = r.libos.get("stopped") or {} + self.assertEqual(stopped.get("num"), 1) # task_port_send_recv_wait + + def test_shim_reaches_relocation_plus_mmu(self): + """The relocation copy reads its real backing through the MPU.""" from sim.gsp_boot import run_boot r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, instructions=20_000, record=False, use_shim=True) self.assertTrue(r.ran) - self.assertEqual(r.rung, "B2") self.assertIn("relocation_copy", r.milestones) self.assertEqual(r.mmio_unknown, 0) - # Known unknown: the DMEM runspace content the app relocates from is - # not reconstructed; the run stops there rather than faking it. - self.assertEqual(r.stop_reason, "fault") - self.assertEqual(r.stop_pc, 0x400EA0C) + # The run no longer stops at the old B2 fault; with a 20k budget it + # is mid-init and deterministic. + self.assertNotEqual(r.stop_pc, 0x400EA0C) def test_two_runs_identical(self): from sim.gsp_boot import run_boot |