1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
"""Tests for the green-eye automation harness (roadmap R5/R6, H4.1/H4.3).
The integration tests drive the two harness CLIs as subprocesses against the
bundled demo firmware and skip cleanly when qemu-system-riscv64 or the demo
ELF is absent. The pure-function tests need neither.
"""
from __future__ import annotations
import json
import os
import subprocess
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")
MATRIX = os.path.join(TOOL_ROOT, "harness", "portability_matrix.py")
MUTATION = os.path.join(TOOL_ROOT, "harness", "mutation_harness.py")
if TOOL_ROOT not in sys.path:
sys.path.insert(0, TOOL_ROOT)
from harness.common import qemu_available # noqa: E402
from harness.mutation_harness import Mutation, expected_for, expected_ok, generate_mutations # noqa: E402
from harness.portability_matrix import counter_delta # noqa: E402
def _run(args: list[str], timeout: float = 600.0) -> subprocess.CompletedProcess:
return subprocess.run(
[sys.executable] + args,
capture_output=True,
text=True,
cwd=TOOL_ROOT,
timeout=timeout,
)
class HarnessUnitTests(unittest.TestCase):
def test_generate_mutations_deterministic(self):
first = generate_mutations(9, seed=5)
second = generate_mutations(9, seed=5)
self.assertEqual([m.to_json() for m in first], [m.to_json() for m in second])
self.assertEqual(
{m.strategy for m in first},
{"cmd-invalid", "arg-boundary", "bit-flip", "buffer-length"},
)
self.assertNotEqual([m.to_json() for m in first], [m.to_json() for m in generate_mutations(9, seed=6)])
def test_memory_region_strategy(self):
mutations = generate_mutations(5, seed=1, region=(0x80001000, 64))
region = [m for m in mutations if m.strategy == "mem-region"]
self.assertEqual(len(region), 1)
self.assertEqual(region[0].region["addr"], 0x80001000)
self.assertTrue(0 <= region[0].region["offset"] < 64)
def test_demo_oracle(self):
before = {"result": 0x30, "last_rpc": 5}
ping = Mutation(0, "bit-flip", cmd=1, arg0=0x1234)
self.assertEqual(expected_for(ping, before, 0x80000800)["value"], 0x1234 ^ 0xBEEF)
fill = Mutation(1, "arg-boundary", cmd=3, arg1=1 << 40)
self.assertEqual(expected_for(fill, before, 0)["value"], 256)
invalid = Mutation(2, "cmd-invalid", cmd=13)
after = {"result": (~13) & ((1 << 64) - 1), "last_rpc": 13}
self.assertTrue(expected_ok(expected_for(invalid, before, 0), invalid, before, after))
idle = Mutation(3, "cmd-invalid", cmd=0)
unchanged = {"result": 0x30, "last_rpc": 5}
self.assertTrue(expected_ok(expected_for(idle, before, 0), idle, before, unchanged))
def test_counter_delta_zero_for_identical_profiles(self):
rows = counter_delta({"rpc_ping": 2}, {"rpc_ping": 2})
self.assertTrue(all(row["delta"] == 0 for row in rows))
@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed")
@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)")
class HarnessIntegrationTests(unittest.TestCase):
def test_portability_matrix_self_comparison(self):
with tempfile.TemporaryDirectory(prefix="harness-matrix-") as out:
result = _run(
[
MATRIX,
"--target",
DEMO_ELF,
"--target",
DEMO_ELF,
"--out",
out,
"--probe-instructions",
"200",
"--scenario-duration",
"0.2",
]
)
self.assertEqual(result.returncode, 0, result.stderr[-2000:])
with open(os.path.join(out, "portability-report.json"), encoding="utf-8") as fh:
report = json.load(fh)
self.assertEqual([t["status"] for t in report["targets"]], ["ok", "ok"])
self.assertEqual(len(report["comparisons"]), 1)
comparison = report["comparisons"][0]
self.assertTrue(comparison["all_zero"], comparison["deltas"])
for rows in comparison["deltas"].values():
for row in rows:
self.assertEqual(row["delta"], 0)
self.assertTrue(os.path.exists(os.path.join(out, "portability-report.md")))
self.assertTrue(all(t["scenario"]["event_counts"] for t in report["targets"]))
def test_mutation_run_and_seed_determinism(self):
with tempfile.TemporaryDirectory(prefix="harness-mut-a-") as out_a, tempfile.TemporaryDirectory(
prefix="harness-mut-b-"
) as out_b:
base = [
MUTATION,
"--kernel",
DEMO_ELF,
"--iterations",
"12",
"--seed",
"7",
"--timeout",
"0.3",
"--settle",
"0.1",
]
first = _run(base + ["--out", out_a])
self.assertEqual(first.returncode, 0, first.stderr[-2000:])
second = _run(base + ["--out", out_b])
self.assertEqual(second.returncode, 0, second.stderr[-2000:])
findings_path = os.path.join(out_a, "findings.jsonl")
self.assertTrue(os.path.exists(findings_path))
with open(findings_path, encoding="utf-8") as fh:
findings = [json.loads(line) for line in fh if line.strip()]
self.assertEqual(len(findings), 12)
self.assertTrue({finding["outcome"] for finding in findings})
self.assertTrue(all("mutation" in finding and "detail" in finding for finding in findings))
self.assertTrue(os.path.exists(os.path.join(out_a, "summary.md")))
with open(findings_path, encoding="utf-8") as fa, open(
os.path.join(out_b, "findings.jsonl"), encoding="utf-8"
) as fb:
self.assertEqual(fa.read(), fb.read(), "the same seed must replay byte-for-byte")
if __name__ == "__main__":
unittest.main()
|
