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