1 files changed, 224 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_trace.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_trace.py
new file mode 100644
index 0000000..96934e9
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_trace.py
@@ -0,0 +1,224 @@
+"""Native R4 trace tests: collection, filters, session blob, analyzer, CLI.
+
+The engine tests are pure (no QEMU, no demo build needed); the session/CLI
+tests skip cleanly when the demo ELF is absent.
+"""
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import tempfile
+import unittest
+
+import greeneye as ge
+from greeneye.analyze import render_summary, render_summary_from
+from greeneye.errors import RspTimeout
+from sim.machine import Machine
+from sim.trace import Trace
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.join(HERE, "..")
+DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
+
+# lui a0,0x80000; addi a1,zero,0x2a; sw a1,0(a0); lw a2,0(a0); ebreak
+PROGRAM = bytes.fromhex("37 05 00 80 93 05 a0 02 23 20 b5 00 03 26 05 00 73 00 10 00")
+BASE = 0x80000000
+
+
+def demo_available() -> bool:
+    return os.path.exists(DEMO_ELF)
+
+
+def records(payload: bytes) -> list[dict]:
+    return [json.loads(line) for line in payload.decode().splitlines() if line.strip()]
+
+
+def read_blob(session_dir: str) -> bytes:
+    with open(os.path.join(session_dir, "blobs", "trace.jsonl"), "rb") as fh:
+        return fh.read()
+
+
+class TraceEngineTests(unittest.TestCase):
+    def _machine(self):
+        machine = Machine()
+        machine.load_segments([(BASE, PROGRAM)], BASE)
+        return machine
+
+    def test_exec_and_access_lines(self):
+        trace = Trace()
+        reason, _, _, _ = self._machine().run(100, trace=trace)
+        self.assertEqual(reason, 3)  # R_EBREAK, stopped at the trap
+        self.assertEqual(trace.counts(), {
+            "exec": 5, "read": 1, "write": 1, "events": 7, "dropped": 0,
+        })
+        lines = records(trace.payload())
+        self.assertEqual(lines[-1]["t"], "summary")
+        self.assertEqual(lines[-1]["insns"], 5)
+        self.assertEqual(lines[-1]["mem"], 2)
+        self.assertEqual(
+            [(r["t"], r["pc"]) for r in lines[:-1] if r["t"] == "exec"],
+            [("exec", "0x80000000"), ("exec", "0x80000004"), ("exec", "0x80000008"),
+             ("exec", "0x8000000c"), ("exec", "0x80000010")],
+        )
+        write = next(r for r in lines if r["t"] == "write")
+        read = next(r for r in lines if r["t"] == "read")
+        self.assertEqual((write["addr"], write["size"], write["pc"], write["vt"]), ("0x80000000", 4, "0x80000008", 3))
+        self.assertEqual((read["addr"], read["size"], read["pc"], read["vt"]), ("0x80000000", 4, "0x8000000c", 4))
+
+    def test_address_filters_and_exec_off(self):
+        trace = Trace(exec_enabled=False, start=BASE + 8, end=BASE + 8)
+        self._machine().run(100, trace=trace)
+        lines = records(trace.payload())
+        self.assertEqual([r["t"] for r in lines], ["write", "summary"])
+        self.assertEqual(lines[0]["pc"], "0x80000008")  # the in-range store only
+        self.assertFalse(lines[1]["exec"])
+        self.assertEqual(lines[1]["start"], "0x80000008")
+
+    def test_ring_budget_is_deterministic(self):
+        first = Trace(max_events=3)
+        second = Trace(max_events=3)
+        self._machine().run(100, trace=first)
+        self._machine().run(100, trace=second)
+        self.assertEqual(first.payload(), second.payload())
+        self.assertEqual(first.counts()["events"], 7)
+        self.assertEqual(first.counts()["dropped"], 4)
+
+    def test_trace_none_path_unchanged(self):
+        machine = self._machine()
+        machine.run(100)  # the default fast path still works untouched
+        self.assertEqual(machine.x[10], 0x80000000)
+        self.assertEqual(machine.x[11], 0x2A)
+        self.assertEqual(machine.x[12], 0x2A)
+
+
+@unittest.skipUnless(demo_available(), "demo firmware unavailable")
+class SessionTraceTests(unittest.TestCase):
+    def _recorded_trace(self, trace) -> tuple[str, bytes]:
+        session_dir = tempfile.mkdtemp(prefix="greeneye-trace-")
+        session = ge.launch(kernel=DEMO_ELF, backend="native", record=session_dir,
+                            name="unittest-sim-trace", trace=trace)
+        try:
+            try:
+                session.client.continue_(timeout=0.02)  # exactly 20k instructions
+            except RspTimeout:
+                pass
+        finally:
+            session.close()
+        return session_dir, read_blob(session_dir)
+
+    def test_blob_recorded_and_analyzed(self):
+        session_dir, blob = self._recorded_trace({"exec": True})
+        meta = json.load(open(os.path.join(session_dir, "session.json"), encoding="utf-8"))
+        self.assertEqual(meta["trace_blob"], "blobs/trace.jsonl")
+        self.assertTrue(meta["trace"]["exec"])
+        lines = blob.decode().splitlines()
+        self.assertTrue(lines[0].startswith('{"t":"exec"'))
+        self.assertIn('"t":"summary"', lines[-1])
+        events = [json.loads(line) for line in open(os.path.join(session_dir, "events.jsonl"), encoding="utf-8")]
+        self.assertTrue(any(e.get("type") == "blob_saved" for e in events))
+        self.assertFalse(any(e.get("type") in ("exec", "read", "write") for e in events),
+                         "trace volume must stay out of events.jsonl")
+        report = render_summary(session_dir)
+        self.assertIn("## Execution trace", report)
+        self.assertIn("unique PCs", report)
+        self.assertIn("### Accessed pages", report)
+
+    def test_trace_is_deterministic(self):
+        _, first = self._recorded_trace(True)
+        _, second = self._recorded_trace(True)
+        self.assertEqual(first, second)
+
+    def test_exec_off_records_accesses_only(self):
+        session_dir, blob = self._recorded_trace({"exec": False})
+        lines = blob.decode().splitlines()
+        self.assertFalse(any('"t":"exec"' in line for line in lines))
+        self.assertTrue(any('"t":"read"' in line or '"t":"write"' in line for line in lines))
+        report = render_summary(session_dir)
+        self.assertIn("exec=off", report)
+
+    def test_launch_trace_implies_recording(self):
+        old = os.environ.get("GREENEYE_DUMPS")
+        os.environ["GREENEYE_DUMPS"] = tempfile.mkdtemp(prefix="greeneye-trace-dumps-")
+        try:
+            session = ge.launch(kernel=DEMO_ELF, backend="native", name="unittest-sim-trace-auto", trace=True)
+            try:
+                directory = session.recorder.dir
+                self.assertIsNotNone(directory)
+            finally:
+                session.close()
+            self.assertTrue(os.path.exists(os.path.join(directory, "blobs", "trace.jsonl")))
+        finally:
+            if old is None:
+                os.environ.pop("GREENEYE_DUMPS", None)
+            else:
+                os.environ["GREENEYE_DUMPS"] = old
+
+    def test_render_from_folded_events(self):
+        meta = {"name": "synthetic", "format": "greeneye-session/1", "trace": {"exec": True}}
+        events = [
+            {"type": "exec", "pc": 0x80000010, "vt": 1},
+            {"type": "exec", "pc": 0x80000010, "vt": 2},
+            {"type": "read", "pc": 0x80000010, "addr": 0x80000100, "size": 4},
+            {"type": "write", "pc": 0x80000014, "addr": 0x80000200, "size": 8},
+            {"type": "access", "pc": 0x80000018, "addr": 0x80000200, "size": 4, "rw": "w"},
+        ]
+        report = render_summary_from(meta, events)
+        self.assertIn("## Execution trace", report)
+        self.assertIn("unique PCs: 1", report)
+        self.assertIn("1 reads (1 pages)", report)
+        self.assertIn("2 writes (1 pages)", report)
+        self.assertIn("| `0x80000000` | 1 | 2 |", report)
+
+
+@unittest.skipUnless(demo_available(), "demo firmware unavailable")
+class CliTraceTests(unittest.TestCase):
+    def test_info_reports_native(self):
+        from greeneye import cli
+
+        out = io.StringIO()
+        with contextlib.redirect_stdout(out):
+            code = cli.main(["info"])
+        self.assertEqual(code, 0)
+        self.assertIn("native simulator: available", out.getvalue())
+        self.assertIn("native trace: available", out.getvalue())
+
+    def test_record_native_trace_cli(self):
+        from greeneye import cli
+
+        session_dir = tempfile.mkdtemp(prefix="greeneye-trace-cli-")
+        out = io.StringIO()
+        with contextlib.redirect_stdout(out):
+            code = cli.main([
+                "record", "--kernel", DEMO_ELF, "--backend", "native",
+                "--duration", "0.05", "--trace", "--record", session_dir,
+            ])
+        self.assertEqual(code, 0)
+        summary = open(os.path.join(session_dir, "summary.md"), encoding="utf-8").read()
+        self.assertIn("## Execution trace", summary)
+        lines = read_blob(session_dir).decode().splitlines()
+        self.assertTrue(any('"t":"exec"' in line for line in lines))
+
+    def test_record_trace_address_filter_cli(self):
+        from greeneye import cli
+
+        session_dir = tempfile.mkdtemp(prefix="greeneye-trace-cli-filter-")
+        with contextlib.redirect_stdout(io.StringIO()):
+            code = cli.main([
+                "record", "--kernel", DEMO_ELF, "--backend", "native",
+                "--duration", "0.05", "--trace", "--trace-exec", "off",
+                "--trace-start", "0x80000000", "--trace-end", "0x800000ff",
+                "--record", session_dir,
+            ])
+        self.assertEqual(code, 0)
+        for line in read_blob(session_dir).decode().splitlines():
+            record = json.loads(line)
+            if record["t"] in ("read", "write"):
+                pc = int(record["pc"], 16)
+                self.assertLessEqual(pc, 0x800000FF)
+            self.assertNotEqual(record["t"], "exec")
+
+
+if __name__ == "__main__":
+    unittest.main()