diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_gmmu_debug.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_gmmu_debug.py | 484 |
1 files changed, 484 insertions, 0 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() |