"""Tests for version_diff / ci_release (roadmap H4.1, planned docs/21). Pure-function tests need neither QEMU nor the demo; the self-diff integration test boots the demo twice and skips cleanly when qemu-system-riscv64 or the demo ELF is absent. """ 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, "..")) DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") if TOOL_ROOT not in sys.path: sys.path.insert(0, TOOL_ROOT) from harness.ci_release import main as ci_release_main # noqa: E402 from harness.common import SymbolDB, qemu_available # noqa: E402 from harness.version_diff import ( # noqa: E402 LOAD_ADDR, Rm610GateError, VersionDiffOptions, check_rm610_gate, corpus_diff, corpus_section, image_diff, inventory_diff, inventory_records, is_rm610_label, load_any_target, load_symbol_sources, load_three_column_export, looks_like_three_column_export, main as version_diff_main, rm610_approved, run_version_diff, shape_payload, ) NOP = b"\x13\x00\x00\x00" # addi x0, x0, 0 def _run_main(main_fn, argv: list[str]) -> tuple[int, str, str]: out, err = io.StringIO(), io.StringIO() with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): rc = main_fn(argv) return rc, out.getvalue(), err.getvalue() def _write(path: str, text: str) -> str: with open(path, "w", encoding="utf-8") as fh: fh.write(text) return path class Rm610GateTests(unittest.TestCase): def test_label_detection(self): for label in ("rm610", "RM610", "rm-610-fw.bin", "gsp_rm_610.elf", "release-RM610"): self.assertTrue(is_rm610_label(label), label) for label in ("570.144", "uproc_demo", "gsp-570.211.fwimage.bin", ""): self.assertFalse(is_rm610_label(label), label) def test_refused_by_default(self): with self.assertRaises(Rm610GateError) as ctx: check_rm610_gate(["rm610"], environ={}) message = str(ctx.exception) self.assertIn("docs/23", message) self.assertIn("RM610-APPROVED", message) self.assertIn("no target file was opened", message) def test_token_and_env_approve(self): check_rm610_gate(["rm610"], approval_token="RM610-APPROVED", environ={}) check_rm610_gate(["rm_610"], environ={"GREENEYE_RM610_APPROVED": "1"}) self.assertTrue(rm610_approved("RM610-APPROVED", {})) self.assertTrue(rm610_approved(None, {"GREENEYE_RM610_APPROVED": "1"})) with self.assertRaises(Rm610GateError): check_rm610_gate(["rm610"], approval_token="rm610-approved", environ={}) with self.assertRaises(Rm610GateError): check_rm610_gate(["rm610"], environ={"GREENEYE_RM610_APPROVED": "0"}) def test_other_labels_are_not_gated(self): check_rm610_gate(["570.144", "uproc_demo"], environ={}) class InventoryDiffTests(unittest.TestCase): def test_matching_discipline(self): records_a = [ {"name": "keep", "addr": 0x100, "size": 16, "source": "elf"}, {"name": "gone", "addr": 0x180, "size": 8, "source": "elf"}, {"name": "FUN_00000200", "addr": 0x200, "size": 12, "source": "elf"}, ] records_b = [ {"name": "keep", "addr": 0x110, "size": 16, "source": "elf"}, {"name": "newfn", "addr": 0x190, "size": 8, "source": "elf"}, {"name": "FUN_00000210", "addr": 0x210, "size": 12, "source": "elf"}, ] diff = inventory_diff(records_a, records_b) self.assertEqual(diff["shift"], 0x10) self.assertEqual([row["name"] for row in diff["moved"]], ["keep"]) self.assertEqual(diff["matched_shifted"], 1) self.assertEqual(diff["totals"]["added"], 1) self.assertEqual(diff["totals"]["removed"], 1) self.assertEqual(diff["added"][0]["name"], "newfn") self.assertEqual(diff["removed"][0]["name"], "gone") def test_resize_and_self_diff(self): records = [ {"name": "a", "addr": 0x10, "size": 4, "source": "elf"}, {"name": "FUN_00000020", "addr": 0x20, "size": 4, "source": "elf"}, ] resized = [dict(row) for row in records] resized[1]["size"] = 8 diff = inventory_diff(records, resized) self.assertEqual(diff["resized_anonymous"][0]["new_size"], 8) zero = inventory_diff(records, records) self.assertEqual(zero["totals"]["added"], 0) self.assertEqual(zero["totals"]["removed"], 0) self.assertEqual(zero["totals"]["moved"], 0) self.assertEqual(zero["totals"]["resized_named"], 0) class SymbolSourceTests(unittest.TestCase): def test_three_column_export(self): with tempfile.TemporaryDirectory() as tmp: path = _write( os.path.join(tmp, "symbols.tsv"), "addr\tname\tsize\n0x80000010\trpc_ping\t16\n0x80000020\trpc_sum\t32\n", ) self.assertTrue(looks_like_three_column_export(path)) db = SymbolDB() self.assertEqual(load_three_column_export(db, path, bias=0), 2) self.assertEqual(db.get("rpc_ping").size, 16) self.assertEqual(db.get("rpc_sum").addr, 0x80000020) def test_names_tsv_is_not_mistaken_for_export(self): with tempfile.TemporaryDirectory() as tmp: path = _write(os.path.join(tmp, "names.tsv"), "0x80000010\trpc_ping\tghidra\n") self.assertFalse(looks_like_three_column_export(path)) @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") def test_sources_merge_on_demo(self): loaded = load_any_target(DEMO_ELF) self.assertTrue(loaded.ok) db, sources, _ = load_symbol_sources(loaded) self.assertTrue(any(source.startswith("elf:") for source in sources)) self.assertTrue(len(inventory_records(db)) > 0) class TargetShapeTests(unittest.TestCase): def test_missing_and_incompatible(self): missing = load_any_target("/nonexistent/nope.elf") self.assertEqual(missing.kind, "missing") self.assertFalse(missing.ok) with tempfile.TemporaryDirectory() as tmp: bad = os.path.join(tmp, "bad.elf") with open(bad, "wb") as fh: fh.write(b"\x7fELF" + b"\x00" * 48) loaded = load_any_target(bad) self.assertEqual(loaded.kind, "unsupported") self.assertIn("EI_CLASS", loaded.reason or "") def test_raw_target_shape(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "raw.bin") with open(path, "wb") as fh: fh.write(NOP * 16) loaded = load_any_target(path) self.assertEqual(loaded.kind, "raw") self.assertEqual(loaded.bias, LOAD_ADDR) payload = shape_payload(loaded) self.assertEqual(payload["span"], 64) self.assertEqual(payload["sections"], []) self.assertEqual(payload["loads"][0]["vaddr"], 0) @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") def test_self_shape_and_inventory_zero(self): old, new = load_any_target(DEMO_ELF), load_any_target(DEMO_ELF) image = image_diff(old, new, max_rows=40) self.assertEqual(image["status"], "ok") self.assertTrue(image["same"]) old_db, _, _ = load_symbol_sources(old) new_db, _, _ = load_symbol_sources(new) diff = inventory_diff(inventory_records(old_db), inventory_records(new_db)) self.assertEqual(diff["shift"], 0) self.assertEqual(diff["totals"]["added"], 0) self.assertEqual(diff["totals"]["removed"], 0) self.assertEqual(diff["totals"]["moved"], 0) self.assertEqual(diff["totals"]["resized_named"], 0) class CorpusTests(unittest.TestCase): def test_corpus_diff(self): with tempfile.TemporaryDirectory() as tmp: a = _write( os.path.join(tmp, "a.tsv"), "addr\tfile_addr\tsize\tname\tbucket\tline\tstatus\n" "0x1000\t0x80000000\t16\trpc_ping\tb\t0\tok\n", ) b = _write( os.path.join(tmp, "b.tsv"), "addr\tfile_addr\tsize\tname\tbucket\tline\tstatus\n" "0x1000\t0x80000010\t16\trpc_ping\tb\t0\tok\n", ) diff = corpus_diff(a, b) self.assertEqual(diff["shift"], 0x10) self.assertEqual(diff["matched_named"], 1) def test_not_supplied(self): options = VersionDiffOptions(old="old.elf", new="new.elf", out="/tmp/unused") section = corpus_section(options, options.out) self.assertEqual(section["status"], "skipped") self.assertIn("not supplied", section["reason"]) class RunVersionDiffTests(unittest.TestCase): def test_missing_target_skips_cleanly(self): with tempfile.TemporaryDirectory() as tmp: new = os.path.join(tmp, "new.bin") with open(new, "wb") as fh: fh.write(NOP * 16) out = os.path.join(tmp, "out") options = VersionDiffOptions(old=os.path.join(tmp, "missing.elf"), new=new, out=out, probe_instructions=8) report = run_version_diff(options) for name in ("image", "symbols", "probe", "behavior"): self.assertEqual(report["sections"][name]["status"], "skipped", name) self.assertTrue(report["sections"][name]["reason"], name) self.assertIn("not supplied", report["sections"]["corpus"]["reason"]) self.assertTrue(os.path.exists(os.path.join(out, "version-diff.json"))) self.assertTrue(os.path.exists(os.path.join(out, "version-diff.md"))) def test_gate_refused_before_any_file_open(self): with tempfile.TemporaryDirectory() as tmp: out = os.path.join(tmp, "out") rc, _, err = _run_main( version_diff_main, ["--old", os.path.join(tmp, "rm610.elf"), "--new", os.path.join(tmp, "new.elf"), "--out", out], ) self.assertEqual(rc, 3) self.assertIn("docs/23", err) self.assertFalse(os.path.exists(out), "the gate must fire before --out is created") def test_gate_passes_with_token(self): with tempfile.TemporaryDirectory() as tmp: new = os.path.join(tmp, "new.bin") with open(new, "wb") as fh: fh.write(NOP * 16) out = os.path.join(tmp, "out") rc, _, _ = _run_main( version_diff_main, ["--old", os.path.join(tmp, "rm610.bin"), "--new", new, "--release", "rm610", "--out", out, "--approval-token", "RM610-APPROVED", "--probe-instructions", "4"], ) self.assertEqual(rc, 0) self.assertTrue(os.path.exists(os.path.join(out, "version-diff.json"))) class CiReleaseTests(unittest.TestCase): def test_dry_run_exits_zero_and_touches_nothing(self): with tempfile.TemporaryDirectory() as tmp: out = os.path.join(tmp, "bundle") rc, stdout, _ = _run_main( ci_release_main, ["--old", os.path.join(tmp, "old.elf"), "--new", os.path.join(tmp, "new.elf"), "--out", out, "--dry-run"], ) self.assertEqual(rc, 0) self.assertIn("version_diff", stdout) self.assertIn("portability_matrix", stdout) self.assertIn("mutation_smoke", stdout) self.assertFalse(os.path.exists(out)) def test_dry_run_rm610_refused(self): rc, _, err = _run_main( ci_release_main, ["--old", "rm610.elf", "--new", "new.elf", "--dry-run"], ) self.assertEqual(rc, 3) self.assertIn("docs/23", err) def test_help_is_clean(self): for main_fn in (ci_release_main, version_diff_main): with contextlib.redirect_stdout(io.StringIO()) as buf, self.assertRaises(SystemExit) as ctx: main_fn(["--help"]) self.assertEqual(ctx.exception.code, 0) self.assertIn("usage", buf.getvalue()) @unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed") @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") class VersionDiffQemuTests(unittest.TestCase): def test_self_diff_zero_inventory_and_behavior(self): with tempfile.TemporaryDirectory(prefix="vdiff-self-") as tmp: options = VersionDiffOptions( old=DEMO_ELF, new=DEMO_ELF, out=tmp, probe_instructions=200, behavior_duration=0.2, max_rows=20, ) report = run_version_diff(options) self.assertTrue(report["sections"]["image"]["same"]) symbols = report["sections"]["symbols"] self.assertEqual(symbols["status"], "ok") self.assertEqual(symbols["shift"], 0) self.assertEqual(symbols["totals"]["added"], 0) self.assertEqual(symbols["totals"]["removed"], 0) self.assertEqual(symbols["totals"]["moved"], 0) self.assertEqual(symbols["totals"]["resized_named"], 0) probe = report["sections"]["probe"] self.assertEqual(probe["status"], "ok") self.assertFalse(probe["changed"]) behavior = report["sections"]["behavior"] self.assertEqual(behavior["status"], "ok") self.assertTrue(behavior["all_zero"], behavior["deltas"]) for rows in behavior["deltas"].values(): for row in rows: self.assertEqual(row["delta"], 0) self.assertTrue(os.path.exists(os.path.join(tmp, "version-diff.md"))) with open(os.path.join(tmp, "version-diff.json"), encoding="utf-8") as fh: self.assertEqual(json.load(fh)["format"], "greeneye-version-diff/1") if __name__ == "__main__": unittest.main()