File size: 5,582 Bytes
966d578 b3588b5 966d578 b3588b5 966d578 | 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 | import json
import os
import pathlib
import tempfile
import unittest
import zipfile
from quread.qec_bundle import build_sample_qec_bundle_bytes, parse_qec_bundle, sample_qec_bundle_filename
def _write_bundle(manifest: dict, artifacts=None) -> str:
artifacts = dict(artifacts or {})
fd, path = tempfile.mkstemp(prefix="qec_bundle_", suffix=".zip")
os.close(fd)
with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("manifest.json", json.dumps(manifest))
for name, content in artifacts.items():
zf.writestr(name, content)
return path
def _valid_manifest() -> dict:
return {
"bundle_version": "1.0",
"source": "nvidia_ising_decoding",
"code_family": "surface_code",
"experiment_name": "surface_d13_demo",
"distance": 13,
"n_rounds": 104,
"basis": "X",
"rotation": "O1",
"noise_model_label": "public_default",
"generated_by": "unit-test",
"timestamp": "2026-04-15T10:00:00Z",
"notes": "demo bundle",
"model": {
"variant": "fast",
"model_id": 1,
"checkpoint_name": "Ising-Decoder-SurfaceCode-1-Fast.pt",
},
"decoders": {
"baseline": {
"name": "pymatching",
"ler": 0.0123,
"latency_ms": 4.8,
"syndrome_density_before": 0.031,
"syndrome_density_after": 0.031,
"logical_failures": 123,
"num_samples": 10000,
},
"ai_predecoder_plus_baseline": {
"name": "ising_predecoder_plus_pymatching",
"ler": 0.0104,
"latency_ms": 3.2,
"syndrome_density_before": 0.031,
"syndrome_density_after": 0.011,
"logical_failures": 104,
"num_samples": 10000,
},
},
"artifacts": [
{"path": "artifacts/run.log", "kind": "log"},
{"path": "artifacts/model.onnx", "kind": "onnx"},
],
}
class QECBundleParserTest(unittest.TestCase):
def test_sample_bundle_bytes_parse_successfully(self):
parsed = parse_qec_bundle(
{"data": build_sample_qec_bundle_bytes(), "name": sample_qec_bundle_filename()}
)
self.assertEqual(parsed["source_name"], sample_qec_bundle_filename())
self.assertEqual(parsed["manifest"]["experiment_name"], "surface_d13_public_demo")
self.assertEqual(len(parsed["artifact_rows"]), 5)
self.assertIn("artifacts/run.log", parsed["preview_text"])
self.assertEqual(parsed["warnings"], [])
def test_valid_bundle_parses_and_normalizes(self):
path = _write_bundle(
_valid_manifest(),
{
"artifacts/run.log": "decoder run log",
"artifacts/model.onnx": "fake-onnx",
},
)
try:
parsed = parse_qec_bundle(path)
self.assertEqual(parsed["manifest"]["code_family"], "surface_code")
self.assertEqual(parsed["manifest"]["source"], "nvidia_ising_decoding")
self.assertEqual(len(parsed["artifact_rows"]), 2)
self.assertIn("manifest.json", parsed["preview_text"])
self.assertEqual(parsed["warnings"], [])
finally:
os.remove(path)
def test_missing_manifest_fails(self):
fd, path = tempfile.mkstemp(prefix="qec_bundle_missing_", suffix=".zip")
os.close(fd)
with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("artifacts/run.log", "hello")
try:
with self.assertRaisesRegex(ValueError, "manifest.json"):
parse_qec_bundle(path)
finally:
os.remove(path)
def test_missing_required_manifest_fields_fail(self):
manifest = _valid_manifest()
del manifest["distance"]
path = _write_bundle(manifest)
try:
with self.assertRaisesRegex(ValueError, "distance"):
parse_qec_bundle(path)
finally:
os.remove(path)
def test_unsupported_code_family_fails(self):
manifest = _valid_manifest()
manifest["code_family"] = "color_code"
path = _write_bundle(manifest)
try:
with self.assertRaisesRegex(ValueError, "surface_code"):
parse_qec_bundle(path)
finally:
os.remove(path)
def test_missing_optional_artifact_becomes_warning(self):
path = _write_bundle(
_valid_manifest(),
{
"artifacts/run.log": "decoder run log",
},
)
try:
parsed = parse_qec_bundle(path)
self.assertEqual(len(parsed["warnings"]), 1)
self.assertIn("Listed artifact missing from bundle", parsed["warnings"][0])
statuses = {row["path"]: row["status"] for row in parsed["artifact_rows"]}
self.assertEqual(statuses["artifacts/model.onnx"], "missing")
finally:
os.remove(path)
def test_mismatched_num_samples_adds_warning(self):
manifest = _valid_manifest()
manifest["decoders"]["ai_predecoder_plus_baseline"]["num_samples"] = 8000
path = _write_bundle(manifest)
try:
parsed = parse_qec_bundle(path)
self.assertTrue(any("sample counts differ" in warning for warning in parsed["warnings"]))
finally:
os.remove(path)
|