path: root/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py

"""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 VersionDiffDryRunTests(unittest.TestCase):
    def test_dry_run_exits_zero_and_touches_nothing(self):
        with tempfile.TemporaryDirectory() as tmp:
            old = os.path.join(tmp, "old.bin")
            new = os.path.join(tmp, "new.bin")
            for path in (old, new):
                with open(path, "wb") as fh:
                    fh.write(NOP * 8)
            before = {}
            for name in ("old.bin", "new.bin"):
                path = os.path.join(tmp, name)
                with open(path, "rb") as fh:
                    before[name] = (os.path.getmtime(path), fh.read())
            out = os.path.join(tmp, "out")
            rc, stdout, _ = _run_main(
                version_diff_main,
                ["--old", old, "--new", new, "--out", out, "--dry-run", "--probe-instructions", "4"],
            )
            self.assertEqual(rc, 0)
            plan = json.loads(stdout[: stdout.find("\n[version-diff]")])
            self.assertEqual(plan["format"], "greeneye-version-diff-plan/1")
            self.assertEqual(plan["status"], "planned")
            self.assertFalse(os.path.exists(out), "dry-run must not create --out")
            for name, (mtime, data) in before.items():
                path = os.path.join(tmp, name)
                self.assertEqual(os.path.getmtime(path), mtime)
                with open(path, "rb") as fh:
                    self.assertEqual(fh.read(), data)

    def test_dry_run_rm610_refused(self):
        rc, _, err = _run_main(
            version_diff_main,
            ["--old", "rm610.elf", "--new", "new.elf", "--dry-run"],
        )
        self.assertEqual(rc, 3)
        self.assertIn("docs/23", err)


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