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