File size: 10,396 Bytes
9e551a1 | 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 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 | # Tensor Roll — Recursive CUDA-Q Model Quantizer
# Copyright (C) 2026 SnapKitty Collective
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Tests for the backend-honesty chunk: doctor, IR, conformance, and the
integration-state audit. Every test asserts measured behavior — no
fabricated numbers. Fast: the whole file runs in seconds."""
import hashlib
import json
import os
import sys
import tempfile
import unittest
import numpy as np
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from tensor_roll import doctor as DR
from tensor_roll import ir as IR
from tensor_roll import conformance as CF
from tensor_roll import boundary as BD
class TestDoctor(unittest.TestCase):
def test_report_structure(self):
rep = DR.doctor_report()
for key in ("capabilities", "sources", "execution", "gpu",
"integration"):
self.assertIn(key, rep)
cap = rep["capabilities"]
self.assertEqual(cap["CPU"]["state"], "AVAILABLE")
self.assertEqual(cap["NumPy"]["state"], "AVAILABLE")
# This machine: no GPU, no CUDA-Q toolchain, no Q# toolchain.
self.assertEqual(cap["CUDA"]["state"], "UNAVAILABLE")
self.assertEqual(cap["CUDA-Q"]["state"], "UNAVAILABLE")
self.assertEqual(cap["Q#"]["state"], "UNAVAILABLE")
self.assertEqual(rep["gpu"], "NONE")
for label in ("CUDA", "CUDA-Q", "Q#"):
self.assertTrue(cap[label].get("reason"),
f"{label} must carry a reason, not just a state")
def test_cpu_execution_passes(self):
rep = DR.doctor_report()
cpu = rep["execution"]["CPU"]
self.assertEqual(cpu["state"], "EXECUTED")
self.assertEqual(cpu["result"], "PASS")
self.assertGreater(cpu["matmul_ms"], 0)
def test_others_not_executed(self):
rep = DR.doctor_report()
for label in ("CUDA", "CUDA-Q", "Q#"):
rec = rep["execution"][label]
self.assertEqual(rec["state"], "NOT EXECUTED")
self.assertTrue(rec["reason"])
self.assertIn(rec["source"], ("VERIFIED", "PRESENT"))
def test_sources_verified(self):
for label, rec in DR.verify_sources().items():
self.assertIn(rec["state"], ("VERIFIED", "PRESENT"), label)
# On this box all three sources carry the AGPL header and the
# expected identifiers, so VERIFIED is the honest state.
states = {l: r["state"] for l, r in DR.verify_sources().items()}
self.assertEqual(states, {"CUDA source": "VERIFIED",
"CUDA-Q source": "VERIFIED",
"Q# source": "VERIFIED"})
def test_render_shape(self):
text = DR.render_doctor(DR.doctor_report())
expected = "\n".join([
"Tensor Roll Backend Report",
"CPU AVAILABLE",
"NumPy AVAILABLE",
"CUDA UNAVAILABLE",
"CUDA-Q UNAVAILABLE",
"Q# UNAVAILABLE",
"GPU NONE",
"CUDA source VERIFIED",
"CUDA-Q source VERIFIED",
"Q# source VERIFIED",
"Execution:",
"CPU PASS",
"CUDA NOT EXECUTED",
"CUDA-Q NOT EXECUTED",
"Q# NOT EXECUTED",
"",
])
self.assertEqual(text, expected)
class TestIR(unittest.TestCase):
def _plan(self):
return {
"w1": {"decision": "QUANTIZE",
"cost_params": {"bits": 8}},
"w2": {"decision": "QUANTUM-ENCODED",
"cost_params": {"n_qubits": 2,
"angles": [0.3, 0.6],
"shots": 64, "seed": 1}},
"w3": {"decision": "PRESERVE", "cost_params": {}},
}
def test_build_ir_structure(self):
ops = IR.build_ir(self._plan())
by_out = [o.op for o in ops]
# w1: LOAD ROLL QUANTIZE EMIT
self.assertEqual(by_out[:4], ["LOAD", "ROLL", "QUANTIZE", "EMIT"])
# w2: LOAD ROLL ROTATE ENTANGLE MEASURE RECONSTRUCT EMIT
self.assertEqual(by_out[4:11],
["LOAD", "ROLL", "ROTATE", "ENTANGLE", "MEASURE",
"RECONSTRUCT", "EMIT"])
# w3: LOAD ROLL EMIT (pass-through, decision in meta)
self.assertEqual(by_out[11:], ["LOAD", "ROLL", "EMIT"])
self.assertEqual(ops[-1].meta["decision"], "PRESERVE")
def test_json_roundtrip(self):
ops = IR.build_ir(self._plan())
back = IR.from_json(IR.to_json(ops))
self.assertEqual([o.op for o in back], [o.op for o in ops])
self.assertEqual([o.params for o in back], [o.params for o in ops])
self.assertEqual([o.meta for o in back], [o.meta for o in ops])
self.assertEqual([o.inputs for o in back], [o.inputs for o in ops])
self.assertEqual([o.outputs for o in back], [o.outputs for o in ops])
def test_lower_cpu_gemm_numeric(self):
A = [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]]
B = [[7.0, 8.0], [9.0, 10.0], [11.0, 12.0]]
ops = [
IR.IROp("LOAD", [], ["A"], {"data": A, "shape": [2, 3]}, {}),
IR.IROp("LOAD", [], ["B"], {"data": B, "shape": [3, 2]}, {}),
IR.IROp("GEMM", ["A", "B"], ["C"], {}, {}),
IR.IROp("EMIT", ["C"], ["out"], {}, {}),
]
recs = IR.lower(ops, "cpu")
self.assertTrue(all(r["state"] == "EXECUTED" for r in recs))
gemm = recs[2]
self.assertEqual(gemm["backend"], "cpu-numpy")
want = np.array(A) @ np.array(B)
self.assertEqual(gemm["detail"]["shape"], [2, 2])
self.assertAlmostEqual(gemm["detail"]["checksum"], float(want.sum()))
emit = recs[3]
self.assertEqual(emit["detail"]["sha256"],
hashlib.sha256(want.tobytes()).hexdigest())
def test_lower_cpu_quantum_family(self):
ops = [
IR.IROp("LOAD", [], ["x"],
{"data": [0.5, -0.25, 0.75, 0.1], "shape": [4]}, {}),
IR.IROp("ROTATE", ["x"], ["xr"],
{"angles": [0.2, 0.4]}, {}),
IR.IROp("ENTANGLE", ["xr"], ["xe"], {}, {}),
IR.IROp("MEASURE", ["xe"], ["xm"],
{"shots": 128, "seed": 3}, {}),
IR.IROp("RECONSTRUCT", ["xm"], ["xhat"], {}, {}),
IR.IROp("EMIT", ["xhat"], ["out"], {}, {}),
]
recs = IR.lower(ops, "cpu")
self.assertTrue(all(r["state"] == "EXECUTED" for r in recs),
[r for r in recs if r["state"] != "EXECUTED"])
self.assertEqual(recs[1]["backend"], "qsim-classical")
self.assertEqual(recs[3]["detail"]["shots"], 128)
def test_lower_cudaq_not_executed(self):
ok, _ = DR.cudaq_toolchain()
if ok:
self.skipTest("real CUDA-Q toolchain present; "
"NOT-EXECUTED path does not apply")
ops = IR.build_ir(self._plan())
recs = IR.lower(ops, "cuda-q")
self.assertTrue(all(r["state"] == "NOT EXECUTED" for r in recs))
self.assertTrue(all(r["reason"] for r in recs))
self.assertTrue(all(r["source"] in ("VERIFIED", "PRESENT")
for r in recs))
rot = next(r for r in recs if r["op"] == "ROTATE")
self.assertEqual(rot["gate_trace_schema"], BD.TRACE_SCHEMA)
self.assertTrue(rot["gate_trace"])
self.assertIn("IR-specification only", rot["note"])
def test_lower_unknown_backend(self):
with self.assertRaises(ValueError):
IR.lower([], "tpu")
class TestConformance(unittest.TestCase):
def test_numpy_executed_and_matches(self):
with tempfile.TemporaryDirectory() as d:
doc = CF.run_conformance(d)
path = os.path.join(d, "conformance.json")
self.assertTrue(os.path.isfile(path))
with open(path) as f:
back = json.load(f)["backends"]
rec = back["numpy"]
self.assertEqual(rec["state"], "EXECUTED")
self.assertTrue(rec["match"])
self.assertTrue(all(rec["checks"].values()))
self.assertEqual(set(back), {"numpy", "cuda", "cuda-q", "qsharp"})
def test_others_not_executed_with_reasons(self):
with tempfile.TemporaryDirectory() as d:
doc = CF.run_conformance(d)
back = doc["backends"]
for name in ("cuda", "cuda-q", "qsharp"):
rec = back[name]
self.assertEqual(rec["state"], "NOT EXECUTED", name)
self.assertTrue(rec["reason"], name)
self.assertIsNone(rec["match"], name)
for key in ("state", "reason", "outputs", "expected",
"match", "measured"):
self.assertIn(key, rec, f"{name}.{key}")
def test_vectors_deterministic(self):
self.assertEqual(CF.test_vectors(), CF.test_vectors())
exp = CF.expected_outputs()
# expected C hard-coded matches a fresh NumPy computation
v = CF.test_vectors()
C = (np.array(v["matmul"]["A"]) @ np.array(v["matmul"]["B"])).tolist()
self.assertEqual(C, exp["matmul"]["C"])
class TestIntegrationState(unittest.TestCase):
def test_audited_value(self):
res = DR.integration_state()
self.assertEqual(res["state"], "file-mediated")
self.assertTrue(res["evidence"])
self.assertTrue(any("tensor_roll/boundary.py:" in e
for e in res["evidence"]))
# the audit must record that vendor execution here is source-only
self.assertTrue(any("source-only" in e for e in res["evidence"]))
def test_boundary_trace_still_roundtrips(self):
with tempfile.TemporaryDirectory() as d:
p = os.path.join(d, "t.json")
BD.write_trace(p, "tensor_roll_encode", 2, 64,
{"00": 40, "11": 24}, producer="test")
back = BD.read_trace(p)
self.assertEqual(back["counts"], {"00": 40, "11": 24})
with open(p, "a") as f:
f.write("corrupt")
with self.assertRaises(ValueError):
BD.read_trace(p)
if __name__ == "__main__":
unittest.main()
|