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

"""Tests for the porting automation chain (docs/28, roadmap H4.1/R6).

Pure/CLI tests need neither QEMU nor the demo; the one-version sweep test
boots the demo under QEMU and skips cleanly when qemu-system-riscv64 or the
demo ELF is absent. Everything else runs from a temp directory and is
side-effect free.
"""
from __future__ import annotations

import contextlib
import io
import json
import os
import sys
import tempfile
import unittest
from unittest import mock

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 import port_driver_versions as pdv  # noqa: E402
from harness import port_new_device as pnd  # noqa: E402
from harness.common import qemu_available  # noqa: E402
from harness.version_diff import check_rm610_gate  # noqa: E402

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 _plan_from_stdout(stdout: str) -> dict:
    """The dry-run prints one JSON plan followed by [tool] status lines."""
    marker = stdout.find("\n[sweep]")
    if marker < 0:
        marker = stdout.find("\n[device]")
    return json.loads(stdout[:marker] if marker >= 0 else stdout)


class VersionHelperTests(unittest.TestCase):
    def test_parse_versions(self):
        self.assertEqual(pdv.parse_versions("540,550,560,570"), ["540", "550", "560", "570"])
        self.assertEqual(pdv.parse_versions(" 570.144 , 570.144 "), ["570.144"])
        self.assertEqual(pdv.parse_versions(""), [])
        with self.assertRaises(ValueError):
            pdv.parse_versions("540$")

    def test_version_ge_610(self):
        self.assertTrue(pdv.version_ge_610("610"))
        self.assertTrue(pdv.version_ge_610("610.1"))
        self.assertTrue(pdv.version_ge_610("620"))
        self.assertFalse(pdv.version_ge_610("570.144"))
        self.assertFalse(pdv.version_ge_610("609.99"))

    def test_default_cadence_is_documented_and_not_invented(self):
        self.assertIn("540", pdv.DEFAULT_VERSIONS)
        self.assertIn("570", pdv.DEFAULT_VERSIONS)
        self.assertNotIn("610", pdv.DEFAULT_VERSIONS)
        self.assertIn("DEFAULT_VERSIONS", pdv.__doc__ or "")

    def test_version_sort_key(self):
        values = sorted(["560", "570.144", "540", "550"], key=pdv.version_sort_key)
        self.assertEqual(values, ["540", "550", "560", "570.144"])

    def test_non_rm610_labels_pass_the_gate(self):
        check_rm610_gate(["570.144", "uproc_demo"], environ={})


class SweepDryRunTests(unittest.TestCase):
    def test_help_is_clean(self):
        for main_fn in (pdv.main, pnd.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())

    def test_help_documents_cron_and_agent_invocation(self):
        with contextlib.redirect_stdout(io.StringIO()) as buf, self.assertRaises(SystemExit):
            pdv.main(["--help"])
        text = buf.getvalue()
        self.assertIn("cron", text)
        self.assertIn("agent", text)
        self.assertIn("RM610-APPROVED", text)

    def test_dry_run_explicit_target_exits_zero_and_creates_nothing(self):
        with tempfile.TemporaryDirectory() as tmp:
            image = os.path.join(tmp, "gsp-570.144.bin")
            with open(image, "wb") as fh:
                fh.write(NOP * 4)
            out = os.path.join(tmp, "out")
            rc, stdout, _ = _run_main(
                pdv.main,
                ["--versions", "570.144", "--target", f"570.144={image}", "--out", out],
            )
            self.assertEqual(rc, 0)
            plan = _plan_from_stdout(stdout)
            self.assertEqual(plan["format"], pdv.PLAN_FORMAT)
            self.assertEqual(plan["found"], 1)
            self.assertEqual(plan["versions"][0]["status"], "found")
            self.assertFalse(os.path.exists(out), "dry-run must not create --out")

    def test_dry_run_missing_input_dir_is_clean(self):
        with tempfile.TemporaryDirectory() as tmp:
            out = os.path.join(tmp, "out")
            rc, stdout, _ = _run_main(
                pdv.main,
                ["--versions", "540,550", "--input-dir", os.path.join(tmp, "absent"), "--out", out],
            )
            self.assertEqual(rc, 0)
            plan = _plan_from_stdout(stdout)
            self.assertEqual(plan["missing"], 2)
            self.assertTrue(all(row["status"] == "no-input" for row in plan["versions"]))
            self.assertFalse(os.path.exists(out))

    def test_dry_run_rm610_refused_before_out(self):
        with tempfile.TemporaryDirectory() as tmp:
            out = os.path.join(tmp, "out")
            rc, _, err = _run_main(pdv.main, ["--versions", "610", "--out", out])
            self.assertEqual(rc, 3)
            self.assertIn("docs/23", err)
            self.assertFalse(os.path.exists(out))

    def test_dry_run_rm610_passes_with_token(self):
        rc, stdout, _ = _run_main(
            pdv.main, ["--versions", "610", "--approval-token", "RM610-APPROVED"]
        )
        self.assertEqual(rc, 0)
        self.assertEqual(_plan_from_stdout(stdout)["rm610_gate"]["status"], "passed")

    def test_config_file_drives_the_plan(self):
        with tempfile.TemporaryDirectory() as tmp:
            image = os.path.join(tmp, "gsp-550.bin")
            with open(image, "wb") as fh:
                fh.write(NOP * 4)
            config = os.path.join(tmp, "sweep.json")
            with open(config, "w", encoding="utf-8") as fh:
                json.dump({
                    "format": pdv.CONFIG_FORMAT,
                    "versions": ["550"],
                    "baseline": "550",
                    "targets": {"550": [image]},
                }, fh)
            rc, stdout, _ = _run_main(pdv.main, ["--config", config])
            self.assertEqual(rc, 0)
            plan = _plan_from_stdout(stdout)
            self.assertEqual(plan["baseline"], "550")
            self.assertEqual(plan["found"], 1)

    def test_config_rejects_unknown_keys(self):
        with tempfile.TemporaryDirectory() as tmp:
            config = os.path.join(tmp, "bad.json")
            with open(config, "w", encoding="utf-8") as fh:
                fh.write('{"versons": ["540"]}')
            with self.assertRaises(ValueError):
                pdv.load_config(config)


class SweepRunTests(unittest.TestCase):
    def test_help_lists_exit_interpretation(self):
        with contextlib.redirect_stdout(io.StringIO()) as buf, self.assertRaises(SystemExit):
            pdv.main(["--help"])
        self.assertIn("--run", buf.getvalue())

    def test_run_without_inputs_fails_cleanly(self):
        with tempfile.TemporaryDirectory() as tmp:
            out = os.path.join(tmp, "out")
            rc, stdout, err = _run_main(
                pdv.main,
                ["--run", "--versions", "540,550", "--input-dir", os.path.join(tmp, "absent"),
                 "--out", out],
            )
            self.assertEqual(rc, 1)
            self.assertIn("no baseline target", stdout)
            self.assertNotIn("Traceback", err)
            self.assertTrue(os.path.exists(os.path.join(out, "sweep-summary.json")))

    def test_completed_status_only_accepts_resumable_states(self):
        with tempfile.TemporaryDirectory() as tmp:
            for state, expected in (("ok", "ok"), ("partial", "partial"),
                                    ("error", None), ("no-input", None)):
                version_dir = os.path.join(tmp, state)
                os.makedirs(version_dir, exist_ok=True)
                with open(os.path.join(version_dir, "version.json"), "w", encoding="utf-8") as fh:
                    json.dump({"status": state}, fh)
                self.assertEqual(pdv.completed_status(version_dir), expected, state)

    def test_run_resumes_completed_versions(self):
        with tempfile.TemporaryDirectory() as tmp:
            out = os.path.join(tmp, "out")
            for version in ("1", "2"):
                version_dir = os.path.join(out, version)
                os.makedirs(version_dir)
                with open(os.path.join(version_dir, "version.json"), "w", encoding="utf-8") as fh:
                    json.dump({"status": "ok"}, fh)
            fake = {
                version: pdv.SweepTarget(version=version, paths=[os.path.join(tmp, f"{version}.bin")],
                                         kind="raw", source="explicit")
                for version in ("1", "2")
            }
            options = pdv.SweepOptions(versions=["1", "2"], out=out, run=True, step_timeout=5)
            with mock.patch.object(pdv, "resolve_targets", return_value=fake):
                report = pdv.run_sweep(options)
            self.assertEqual(report["failed"], [])
            self.assertEqual(report["ran"], 2)
            self.assertTrue(all(row.get("resumed") for row in report["versions"]))
            self.assertTrue(os.path.exists(os.path.join(out, "bundle-manifest.json")))


class DevicePortDryRunTests(unittest.TestCase):
    def test_dry_run_exits_zero_and_creates_nothing(self):
        with tempfile.TemporaryDirectory() as tmp:
            firmware = os.path.join(tmp, "gsp-570.144.bin")
            with open(firmware, "wb") as fh:
                fh.write(NOP * 4)
            out = os.path.join(tmp, "out")
            rc, stdout, _ = _run_main(
                pnd.main, ["--device", "rtx4090", "--firmware", firmware, "--out", out]
            )
            self.assertEqual(rc, 0)
            plan = _plan_from_stdout(stdout)
            self.assertEqual(plan["format"], pnd.PLAN_FORMAT)
            self.assertEqual(plan["profile"]["action"], "would-create-scaffold")
            self.assertTrue(plan["firmware"]["found"])
            self.assertFalse(os.path.exists(out))

    def test_dry_run_rm610_refused(self):
        rc, _, err = _run_main(pnd.main, ["--device", "rm610-box"])
        self.assertEqual(rc, 3)
        self.assertIn("docs/23", err)

    def test_dry_run_rm610_passes_with_env(self):
        with mock.patch.dict(os.environ, {"GREENEYE_RM610_APPROVED": "1"}):
            rc, stdout, _ = _run_main(pnd.main, ["--device", "rm610-box"])
        self.assertEqual(rc, 0)
        self.assertEqual(_plan_from_stdout(stdout)["rm610_gate"]["status"], "passed")

    def test_run_without_firmware_fails_cleanly(self):
        with tempfile.TemporaryDirectory() as tmp:
            out = os.path.join(tmp, "out")
            rc, _, err = _run_main(pnd.main, ["--run", "--device", "testdev", "--out", out])
            self.assertEqual(rc, 2)
            self.assertIn("no device firmware", err)
            self.assertNotIn("Traceback", err)

    def test_scaffold_profile_has_all_sections(self):
        profile = pnd.scaffold_profile("testdev")
        self.assertEqual(profile["format"], pnd.PROFILE_FORMAT)
        for key in ("pci", "bars", "fb", "engines", "interrupts", "rom", "physical_validation"):
            self.assertIn(key, profile)
        self.assertEqual(sorted(pnd.profile_missing_fields(profile)),
                         ["bars", "engines", "fb", "interrupts", "rom"])


@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed")
@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)")
class OneVersionSweepQemuTests(unittest.TestCase):
    def test_one_version_sweep_writes_the_evidence_bundle(self):
        with tempfile.TemporaryDirectory(prefix="port-sweep-") as tmp:
            out = os.path.join(tmp, "out")
            rc, stdout, err = _run_main(
                pdv.main,
                ["--run", "--versions", "demo", "--target", f"demo={DEMO_ELF}",
                 "--baseline", "demo", "--out", out,
                 "--probe-instructions", "16", "--probe-duration", "0.05",
                 "--scenario-duration", "0.05", "--max-stops", "50",
                 "--mutation-iterations", "1", "--mutation-seed", "1",
                 "--mutation-timeout", "0.05", "--mutation-settle", "0.01",
                 "--step-timeout", "180"],
            )
            self.assertEqual(rc, 0, err + stdout)
            version_dir = os.path.join(out, "demo")
            with open(os.path.join(version_dir, "version.json"), encoding="utf-8") as fh:
                version = json.load(fh)
            self.assertEqual(version["status"], "ok", version)
            self.assertEqual(pdv.completed_status(version_dir), "ok")
            for rel in ("version-diff/version-diff.json",
                        "portability-matrix/portability-report.json",
                        "mutation-smoke/findings.jsonl",
                        "bundle-manifest.json"):
                self.assertTrue(os.path.exists(os.path.join(version_dir, rel)), rel)
            with open(os.path.join(out, "sweep-summary.json"), encoding="utf-8") as fh:
                summary = json.load(fh)
            self.assertEqual(summary["failed"], [])
            self.assertEqual(summary["ran"], 1)
            self.assertTrue(summary["interpretation"].startswith("pass"))
            with open(os.path.join(version_dir, "bundle-manifest.json"), encoding="utf-8") as fh:
                manifest = json.load(fh)
            paths = {row["path"] for row in manifest["files"]}
            self.assertIn("version-diff/version-diff.json", paths)


if __name__ == "__main__":
    unittest.main()