a7464244c5
- Python: qudit, qudit_check, wasmq (an envelope's bytes, a path or an example's name), wasmq_verify (with the envelope, run again byte for byte) and wasmq_examples, the extension module running an envelope on wai-wasmq's own interpreter (no compiler in the wheel); Service.wasmq and Service.qudit run them on a service. Tests: both kinds run and verified here, a qudit run equal to `wai-quantum qudit run`'s bits and a wasmQ/1 job equal to `wai-quantum run`'s byte for byte, and both run on a service built from this checkout. - npm (wai-web, quantum_engine): QuantumEngine.qudit and QuantumEngine.wasmq (the interpreter, the only engine on wasm32), engine_qudit_check, engine_wasmq_verify, engine_wasmq_examples and engine_wasmq_example. The smoke test runs every example in Node, each sealed, held to its form and run again, a fault sealed and W013 refused. - Quantum Studio: the "Qudits" and "wasmQ" panes (closed until opened), in the browser and in the desktop app (its qudit, qudit_check, wasmq, wasmq_verify and wasmq_examples commands): the qudit counts by digits, amplitudes as integers, bounds, the seal checked and the request run again; an example envelope or a .wai file opened there, its outcome, calls, fuel, counts and energy, its seal and form checked and the envelope run again; a refusal with its code. The browser e2e drives each. - quantum-wasmq §11.9 and quantum-sim §9.10 name every surface. Tested on the build hosts: wai-web for wasm32-unknown-unknown and the npm smoke test in Node 26 (macOS arm64); the Python package's wheel (maturin 1.15.0), clippy -D warnings and pytest (95 passed, the widgets' node test skipped there) on Linux x86_64; the Studio's browser e2e (macOS arm64, STUDIO_SKIP_SERVICE=1), every qudit and wasmQ check passing. Co-Authored-By: Claude Opus 5.5 <[email protected]>
756 lines
38 KiB
Python
756 lines
38 KiB
Python
"""The engine in-process: each function, its answers and its refusals."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import pathlib
|
||
import math
|
||
import pickle
|
||
import xml.etree.ElementTree as ET
|
||
|
||
import pytest
|
||
|
||
import wai_quantum as wq
|
||
from conftest import BELL, REPO, bare
|
||
|
||
DEMO = wq.Key.demo()
|
||
|
||
|
||
# -- run ----------------------------------------------------------------------
|
||
|
||
|
||
def test_a_bell_pair_runs_on_the_stabilizer_method_and_its_receipt_verifies():
|
||
job = wq.run(BELL, shots=1000, seed=7, key=DEMO)
|
||
assert isinstance(job, wq.Job) and isinstance(job, dict)
|
||
assert job.status == "done" and job.method == "stabilizer" and job.accuracy == "exact"
|
||
assert set(job.counts) <= {"00", "11"} and job.shots == 1000
|
||
assert job["request"] == {"qasm": BELL, "shots": 1000, "seed": 7}
|
||
assert job.result["energy"] == {"unmetered": wq.UNMETERED}
|
||
v = wq.verify(job)
|
||
assert v and v.valid and v.key == DEMO.public_key and v.signer == DEMO.did
|
||
# The job is plain JSON: written and read back, it still verifies.
|
||
again = json.loads(json.dumps(job))
|
||
assert wq.verify(again).valid and job.verify().valid
|
||
|
||
|
||
def test_the_same_job_gives_the_same_answer():
|
||
a = wq.run(BELL, shots=500, seed=3, key=DEMO)
|
||
b = wq.run({"qasm": BELL, "shots": 500, "seed": 3}, key=DEMO)
|
||
assert bare(a["result"]) == bare(b["result"])
|
||
assert {k: v for k, v in a.items() if k != "result"} == {k: v for k, v in b.items() if k != "result"}
|
||
# Another seed samples other counts from the same distribution.
|
||
c = wq.run(BELL, shots=500, seed=4, key=DEMO)
|
||
assert c.counts != a.counts and set(c.counts) <= {"00", "11"}
|
||
|
||
|
||
def test_helpers_read_counts_frequencies_expectations_and_amplitudes():
|
||
src = "OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];"
|
||
job = wq.run(src, shots=2000, seed=5, observables=["Z0 Z1", {"pauli": "X0 X1", "coeff": 0.5}, "Z0"], amplitudes=["00", "11", "01"], key=DEMO)
|
||
assert job.shots == 2000
|
||
p = job.probabilities()
|
||
assert math.isclose(sum(p.values()), 1.0) and set(p) <= {"00", "11"}
|
||
e = job.expectations
|
||
assert e["Z0 Z1"] == pytest.approx(1.0) and e["X0 X1"] == pytest.approx(0.5) and abs(e["Z0"]) < 1e-12
|
||
a = job.amplitudes
|
||
assert abs(a["00"] - 2 ** -0.5) < 1e-6 and abs(a["11"] - 2 ** -0.5) < 1e-6 and a["01"] == 0
|
||
assert "sealed" in job.summary()
|
||
|
||
|
||
def test_an_arbitrary_angle_runs_on_an_f64_method_and_says_so():
|
||
# 0.3 rad is no dyadic multiple of 2π: the bit-exact method refuses it,
|
||
# and the planner sends it to the double-precision state vector.
|
||
src = "OPENQASM 3.0; qubit[1] q; ry(0.3) q[0];"
|
||
job = wq.run(src, observables=["Z0", "X0"], key=DEMO)
|
||
assert job.method == "statevector" and job.accuracy == "exact"
|
||
assert job.expectations["Z0"] == pytest.approx(math.cos(0.3), abs=1e-12)
|
||
assert job.expectations["X0"] == pytest.approx(math.sin(0.3), abs=1e-12)
|
||
with pytest.raises(wq.EngineError):
|
||
wq.run(src, observables=["Z0"], method="exact", key=DEMO)
|
||
|
||
|
||
def test_a_dyadic_circuit_runs_bit_exact_when_asked():
|
||
job = wq.run("OPENQASM 3.0; qubit[2] q; h q[0]; t q[0]; cx q[0], q[1];", method="exact", shots=100, key=DEMO)
|
||
assert job.method == "exact" and job.accuracy == "bit-exact" and len(job.result["state_hash"]) == 64
|
||
|
||
|
||
def test_a_noisy_run_names_its_method_and_its_noise_lowers_the_correlation():
|
||
src = "OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];"
|
||
clean = wq.run(src, observables=["Z0 Z1"], key=DEMO)
|
||
noisy = wq.run(src, observables=["Z0 Z1"], noise={"depolarizing_2q": 0.1}, method="density_matrix", key=DEMO)
|
||
assert noisy.method == "density_matrix" and noisy.accuracy == "exact"
|
||
# A two-qubit depolarizing channel of strength p after the CX scales ⟨Z0 Z1⟩ by 1 − 16p/15.
|
||
assert noisy.expectations["Z0 Z1"] == pytest.approx(clean.expectations["Z0 Z1"] * (1 - 16 * 0.1 / 15), abs=1e-9)
|
||
|
||
|
||
def test_a_dynamic_program_runs_branch_by_branch():
|
||
src = "OPENQASM 3.0; qubit[2] q; bit[2] c; h q[0]; c[0] = measure q[0]; if (c[0]) x q[1]; c[1] = measure q[1];"
|
||
job = wq.run(src, shots=400, seed=2, key=DEMO)
|
||
assert set(job.counts) <= {"00", "11"} and job.shots == 400
|
||
|
||
|
||
def test_a_key_may_be_given_as_a_key_a_seed_or_hex_and_defaults_to_the_users():
|
||
k = wq.Key.generate()
|
||
for given in (k, k.seed, k.seed.hex()):
|
||
assert wq.verify(wq.run(BELL, shots=10, key=given)).key == k.public_key
|
||
u = wq.run(BELL, shots=10)
|
||
assert wq.verify(u).key == wq.Key.user().public_key
|
||
with pytest.raises(ValueError):
|
||
wq.run(BELL, key=b"short")
|
||
with pytest.raises(TypeError):
|
||
wq.run(BELL, key=3.0)
|
||
|
||
|
||
def test_a_program_may_be_a_path(tmp_path):
|
||
f = tmp_path / "bell.qasm"
|
||
f.write_text(BELL)
|
||
assert wq.run(f, shots=10, seed=1, key=DEMO).counts == wq.run(BELL, shots=10, seed=1, key=DEMO).counts
|
||
assert wq.lint(f)["ok"] is True
|
||
|
||
|
||
# -- refusals -----------------------------------------------------------------
|
||
|
||
|
||
def test_a_refusal_raises_the_engines_message_with_its_line():
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.run("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n", key=DEMO)
|
||
assert e.value.line == 3 and "foo" in e.value.message and e.value.code == "Q103"
|
||
assert str(e.value) == e.value.message
|
||
# It survives pickling (a process pool carries it back).
|
||
back = pickle.loads(pickle.dumps(e.value))
|
||
assert (back.message, back.line, back.code) == (e.value.message, 3, "Q103")
|
||
with pytest.raises(wq.EngineError, match="asks for nothing"):
|
||
wq.run(BELL, shots=0, key=DEMO)
|
||
with pytest.raises(wq.EngineError, match="no such channel"):
|
||
wq.run(BELL, noise={"depolarising": 0.1}, key=DEMO)
|
||
|
||
|
||
def test_a_program_nested_past_the_readers_limit_raises_and_the_engine_goes_on():
|
||
# Nested thousands deep: refused with the reader's code, in this process.
|
||
deep = "OPENQASM 3.0;\nqubit q;\nrz(" + "(" * 5000 + "0.5" + ")" * 5000 + ") q;\n"
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.run(deep, key=DEMO)
|
||
assert e.value.code == "Q006"
|
||
assert wq.lint(deep)["ok"] is False
|
||
with pytest.raises(wq.EngineError):
|
||
wq.critique(deep)
|
||
blocks = "OPENQASM 3.0;\nqubit q;\n" + "{ " * 5000 + "x q;" + " }" * 5000 + "\n"
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.run(blocks, key=DEMO)
|
||
assert e.value.code == "Q006"
|
||
# The interpreter goes on: the next program runs.
|
||
assert wq.run(BELL, shots=10, seed=1, key=DEMO).status == "done"
|
||
|
||
|
||
def test_values_json_cannot_carry_are_refused_not_guessed():
|
||
with pytest.raises(ValueError, match="NaN"):
|
||
wq.run(BELL, options={"x": float("nan")}, key=DEMO)
|
||
with pytest.raises(ValueError, match="64 bits"):
|
||
wq.run(BELL, seed=2 ** 70, key=DEMO)
|
||
# Past 2^53 a double cannot hold it: the engine refuses it by its path.
|
||
with pytest.raises(wq.EngineError, match="seed"):
|
||
wq.run(BELL, seed=2 ** 53 + 1, key=DEMO)
|
||
with pytest.raises(TypeError, match="keys are strings"):
|
||
wq.run({"qasm": BELL, 1: 2}, key=DEMO)
|
||
with pytest.raises(TypeError, match="not a JSON value"):
|
||
wq.run(BELL, options={"x": object()}, key=DEMO)
|
||
|
||
|
||
# -- plan, lint, draw, layout -----------------------------------------------------
|
||
|
||
|
||
def test_plan_names_the_method_and_why_without_running():
|
||
p = wq.plan(BELL)
|
||
assert p["method"] == "stabilizer" and p["qubits"] == 2 and "Clifford" in p["why"]
|
||
assert set(p) == {"method", "why", "memory_bytes", "work", "qubits"}
|
||
assert wq.plan(BELL, method="statevector")["method"] == "statevector"
|
||
with pytest.raises(wq.EngineError):
|
||
wq.plan("qubit[1] q; bad q[0];")
|
||
|
||
|
||
def test_lint_gives_the_shape_or_the_problem_and_its_line():
|
||
assert wq.lint(BELL) == {"ok": True, "qubits": 2, "gates": 2, "bits": 2, "measured": 2, "dynamic_steps": None}
|
||
bad = wq.lint("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n")
|
||
assert bad["ok"] is False and bad["line"] == 3
|
||
|
||
|
||
def test_draw_writes_the_text_drawing_and_the_svg():
|
||
t = wq.draw(BELL)
|
||
assert t.splitlines()[0].startswith("q0: ") and "H" in t and "⊕" in t and "M1" in t
|
||
svg = wq.draw(BELL, format="svg")
|
||
root = ET.fromstring(svg)
|
||
assert root.tag == "{http://www.w3.org/2000/svg}svg"
|
||
items = [g for g in root.iter("{http://www.w3.org/2000/svg}g") if "qc-item" in g.get("class", "").split()]
|
||
assert len(items) == 4 and [g.get("data-op") for g in items if g.get("data-op")] == ["0", "1"]
|
||
assert wq.draw(BELL, format="svg") == svg
|
||
with pytest.raises(ValueError):
|
||
wq.draw(BELL, format="png")
|
||
with pytest.raises(ValueError):
|
||
wq.draw(BELL, hands=["person"])
|
||
with pytest.raises(wq.EngineError):
|
||
wq.draw("qubit[1] q; nope q[0];")
|
||
|
||
|
||
def test_svg_marks_each_gate_with_its_hand():
|
||
svg = wq.draw("OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];", format="svg", hands=["origin", "assistant"])
|
||
assert 'class="qc-item qc-by-origin" data-op="0"' in svg and 'class="qc-item qc-by-assistant" data-op="1"' in svg
|
||
assert ">the assistant: 1 gate</text>" in svg
|
||
|
||
|
||
def test_layout_is_the_drawing_as_data():
|
||
lay = wq.layout(BELL)
|
||
assert lay["qubits"] == 2 and lay["columns"][1][0] == {"cells": [[0, "●"], [1, "⊕"]], "joined": True, "op": 1}
|
||
|
||
|
||
# -- critique, improve, shortcuts, equivalent ------------------------------------
|
||
|
||
|
||
def test_critique_gives_measures_and_smells_and_refuses_what_does_not_read():
|
||
# A fix is offered for gates on lines of their own.
|
||
c = wq.critique("OPENQASM 3.0;\nqubit[1] q;\nh q[0];\nh q[0];\nx q[0];\n")
|
||
assert c["gates"] == 3 and [(s["kind"], s["lines"], s["fixable"]) for s in c["smells"]] == [("cancels", [3, 4], True)]
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.critique("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];")
|
||
assert e.value.line == 3
|
||
|
||
|
||
def test_improve_offers_only_variants_proved_equivalent():
|
||
v = wq.improve("OPENQASM 3.0;\nqubit[1] q;\nh q[0];\nh q[0];\nx q[0];\n")
|
||
best = v["candidates"][0]
|
||
assert best["measures"]["gates"] < v["base"]["gates"] and best["equivalence"]["verdict"] == "equivalent"
|
||
assert wq.equivalent(best["qasm"], "OPENQASM 3.0; qubit[1] q; x q[0];")["verdict"] == "equivalent"
|
||
|
||
|
||
def test_shortcuts_names_a_cheap_route_for_a_clifford_circuit():
|
||
s = wq.shortcuts(BELL)
|
||
assert s["ok"] is True and s["cheapest"]
|
||
|
||
|
||
def test_equivalent_decides_and_says_how():
|
||
assert wq.equivalent("qubit[1] q; h q[0]; h q[0];", "qubit[1] q;")["how"] == "tableau"
|
||
ex = wq.equivalent("qubit[1] q; t q[0]; t q[0];", "qubit[1] q; s q[0];")
|
||
assert ex["verdict"] == "equivalent" and ex["how"] == "exact"
|
||
assert wq.equivalent("qubit[1] q; x q[0];", "qubit[1] q; z q[0];")["verdict"] == "different"
|
||
assert wq.equivalent("qubit[1] q; x q[0];", "qubit[2] q;")["how"] == "shape"
|
||
assert wq.equivalent("qubit[1] q; nope q[0];", "qubit[1] q;")["verdict"] == "unchecked"
|
||
|
||
|
||
# -- receipts -------------------------------------------------------------------
|
||
|
||
|
||
def test_any_change_to_a_sealed_job_breaks_its_receipt():
|
||
job = wq.run(BELL, shots=100, seed=1, key=DEMO)
|
||
for path, value in ((("result", "counts", "00"), 1), (("request", "shots"), 99), (("result", "method"), "statevector"), (("result", "energy", "unmetered"), "x")):
|
||
forged = json.loads(json.dumps(job))
|
||
d = forged
|
||
for k in path[:-1]:
|
||
d = d[k]
|
||
d[path[-1]] = value
|
||
assert not wq.verify(forged), path
|
||
no_receipt = json.loads(json.dumps(job))
|
||
del no_receipt["result"]["receipt"]
|
||
# A result with no receipt is a verdict, that reason alone, as on every surface.
|
||
v = wq.verify(no_receipt)
|
||
assert not v.valid and v.reasons == ["no_receipt"]
|
||
|
||
|
||
def test_seal_and_verify_round_trip_and_the_digest_is_the_receipts():
|
||
k = wq.Key.generate()
|
||
req = {"qasm": BELL, "shots": 10}
|
||
res = {"method": "stabilizer", "counts": {"00": 6, "11": 4}, "value": 0.1}
|
||
r = wq.seal(req, res, key=k)
|
||
assert r["alg"] == "ed25519" and r["canon"] == "rfc8785" and r["key"] == k.public_key
|
||
assert r["digest"] == wq.digest(req, res)
|
||
sealed = {"request": req, "result": dict(res, receipt=r)}
|
||
assert wq.verify(sealed).valid
|
||
# A receipt already on the result is not part of what is sealed.
|
||
assert wq.seal(req, dict(res, receipt={"old": 1}), key=k) == r
|
||
|
||
|
||
def test_verify_reads_a_dict_json_text_or_a_file(tmp_path):
|
||
job = wq.run(BELL, shots=10, key=DEMO)
|
||
f = tmp_path / "job.json"
|
||
f.write_text(json.dumps(job))
|
||
assert wq.verify(job) and wq.verify(json.dumps(job)) and wq.verify(f) and wq.verify(str(f))
|
||
|
||
|
||
def test_canonical_json_is_rfc_8785():
|
||
# Integers and strings: the standard library writes the same text when it
|
||
# sorts keys and drops whitespace.
|
||
v = {"z": [3, -2, {"b": True, "a": None}], "a": "x\n\u0001\"\\é", "m": 9007199254740991}
|
||
assert wq.canonical(v) == json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
|
||
# Numbers as ECMAScript writes them; keys by UTF-16 code units.
|
||
assert wq.canonical({"b": 1.0, "a": [1e21, 1e-7, 0.000001, -0.0, 123e-20]}) == b'{"a":[1e+21,1e-7,0.000001,0,1.23e-18],"b":1}'
|
||
assert wq.canonical({"": 1, "\U0001f600": 2}) == '{"\U0001f600":2,"":1}'.encode()
|
||
|
||
|
||
def test_every_number_reaches_the_engine_as_the_double_it_is():
|
||
"""The engine's ECMAScript number vectors, each sent from Python as a
|
||
float: the canonical text is the vector's, so the float crossed into the
|
||
engine bit for bit."""
|
||
path = REPO / "wai" / "wai-quantum" / "src" / "engine" / "testdata" / "es-number-tostring.txt"
|
||
n = 0
|
||
for line in path.read_text().splitlines():
|
||
bits, want = line.split(" ", 1)
|
||
x = _from_bits(int(bits, 16))
|
||
if math.isnan(x) or math.isinf(x):
|
||
continue
|
||
assert wq.canonical(x) == want.encode(), bits
|
||
n += 1
|
||
assert n > 3000
|
||
|
||
|
||
def _from_bits(b: int) -> float:
|
||
import struct
|
||
|
||
return struct.unpack(">d", b.to_bytes(8, "big"))[0]
|
||
|
||
|
||
# -- keys -------------------------------------------------------------------------
|
||
|
||
|
||
def test_keys_name_their_signer_and_files_are_the_command_lines_form(tmp_path):
|
||
assert DEMO.is_demo and DEMO.did.startswith("did:key:z6Mk") and len(DEMO.public_key) == 64
|
||
k = wq.Key.generate()
|
||
p = k.save(tmp_path / "k")
|
||
assert p.read_text() == k.seed.hex() + "\n" and (p.stat().st_mode & 0o777) == 0o600 or __import__("os").name == "nt"
|
||
assert wq.Key.from_file(p) == k and wq.Key.from_hex(k.seed.hex()) == k
|
||
with pytest.raises(FileExistsError):
|
||
k.save(p)
|
||
(tmp_path / "bad").write_text("not a key")
|
||
with pytest.raises(ValueError):
|
||
wq.Key.from_file(tmp_path / "bad")
|
||
|
||
|
||
def test_the_users_key_is_made_once_and_a_named_file_must_exist(tmp_path, monkeypatch):
|
||
a = wq.Key.user()
|
||
assert a.path is not None and a.path.exists()
|
||
assert wq.Key.user() == a
|
||
monkeypatch.setenv("WAI_QUANTUM_KEY", str(tmp_path / "missing"))
|
||
with pytest.raises(FileNotFoundError):
|
||
wq.Key.user()
|
||
|
||
|
||
# -- the catalogue, the history, Bloch vectors --------------------------------------
|
||
|
||
|
||
def test_methods_and_accuracy_classes_are_the_catalogues():
|
||
ms = {m["method"]: m for m in wq.methods()}
|
||
assert {"exact", "statevector", "stabilizer", "pauli", "mps", "density_matrix", "trajectories"} <= set(ms)
|
||
assert ms["exact"]["accuracy"] == "bit-exact"
|
||
assert set(wq.accuracy_classes()) >= {"bit-exact", "exact", "bounded", "estimate", "heuristic"}
|
||
|
||
|
||
def test_a_history_records_versions_and_blames_each_gate():
|
||
bell = "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n"
|
||
k = wq.Key.generate()
|
||
h = wq.history("start", {"source": bell, "by": {"kind": "origin", "name": "example"}, "at": 1}, key=k)
|
||
h = wq.history("propose", {"history": h["history"], "parent": "v0", "proposed": bell + "x q[1];\n", "accepted": [0], "by": {"kind": "assistant", "name": "m"}, "at": 2}, key=k)
|
||
b = wq.history("blame", {"history": h["history"], "version": "v1"})
|
||
assert b["gates"] == ["origin", "origin", "assistant"]
|
||
log = wq.history("log", {"history": h["history"]})
|
||
assert log["check"]["ok"] is True
|
||
with pytest.raises(wq.EngineError):
|
||
wq.history("nope", {})
|
||
|
||
|
||
def test_bloch_vectors_are_the_engines_expectation_values():
|
||
cases = [
|
||
("qubit[1] q;", (0, 0, 1)),
|
||
("qubit[1] q; x q[0];", (0, 0, -1)),
|
||
("qubit[1] q; h q[0];", (1, 0, 0)),
|
||
("qubit[1] q; h q[0]; s q[0];", (0, 1, 0)),
|
||
("qubit[1] q; ry(0.7) q[0];", (math.sin(0.7), 0, math.cos(0.7))),
|
||
("qubit[1] q; rx(1.1) q[0];", (0, -math.sin(1.1), math.cos(1.1))),
|
||
]
|
||
for src, want in cases:
|
||
b = wq.bloch("OPENQASM 3.0; " + src, key=DEMO)
|
||
(q,) = b["qubits"]
|
||
assert (q["x"], q["y"], q["z"]) == pytest.approx(want, abs=1e-12), src
|
||
assert q["length"] == pytest.approx(1.0, abs=1e-12)
|
||
# Entangled: each qubit alone is maximally mixed.
|
||
ent = wq.bloch("OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];", key=DEMO)
|
||
assert [q["length"] for q in ent["qubits"]] == pytest.approx([0, 0], abs=1e-12)
|
||
assert ent["method"] == "stabilizer" and ent["job"].verify().valid
|
||
# Noise shortens the vector: dephasing p scales x by 1 − 2p.
|
||
noisy = wq.bloch("OPENQASM 3.0; qubit[1] q; h q[0];", noise={"dephasing": 0.1}, method="density_matrix", key=DEMO)
|
||
assert noisy["qubits"][0]["x"] == pytest.approx(1 - 2 * 0.1, abs=1e-9)
|
||
# The method bounds its values: each vector carries the bounds.
|
||
bounded = wq.bloch("OPENQASM 3.0; qubit[1] q; ry(0.4) q[0];", method="pauli", key=DEMO)
|
||
assert bounded["accuracy"] == "bounded" and len(bounded["qubits"][0]["bound"]) == 3
|
||
assert [q["qubit"] for q in wq.bloch("OPENQASM 3.0; qubit[3] q;", qubits=[2], key=DEMO)["qubits"]] == [2]
|
||
|
||
|
||
# -- estimate -----------------------------------------------------------------
|
||
|
||
LADDER = "OPENQASM 3.0;\nqubit[4] q;\nh q;\nctrl(3) @ x q[0], q[1], q[2], q[3];\nrz(0.3) q[0];\nt q[1];\n"
|
||
|
||
|
||
def test_an_estimate_counts_the_program_seals_every_figure_and_re_runs():
|
||
e = wq.estimate(LADDER, qubit="us-e3", budget=1e-4, max_slowdown=64, key=DEMO)
|
||
r = e["result"]
|
||
assert r["format"] == "wai.quantum.estimate/1"
|
||
c = r["counts"]
|
||
# A gate under three controls: an ancilla and three Toffolis.
|
||
assert (c["logical_qubits"], c["ancillas"], c["toffolis"], c["t_gates"], c["rotations"], c["measurements"]) == (5, 1, 3, 1, 1, 4)
|
||
assert r["epsilon"]["each"] == pytest.approx(1e-4 / 3) and r["epsilon"]["bound"] <= r["epsilon"]["synthesis"]
|
||
assert r["estimate"]["receipt"]["kind"] == "quantum-resource" and r["calibration"]["name"] == "exact_matching"
|
||
f = r["frontier"]
|
||
assert len(f) >= 2 and all(a["physical_qubits"] > b["physical_qubits"] for a, b in zip(f, f[1:]))
|
||
assert sum(1 for p in f if p.get("headline")) == 1
|
||
# Its seal checks as every job's does; the estimate check also re-runs it.
|
||
assert wq.verify(e).valid
|
||
v = wq.verify_estimate(e)
|
||
assert v["valid"] and v["signed"] and v["receipts"] and v["reproduced"] and v["signer"] == DEMO.did
|
||
forged = json.loads(json.dumps(e))
|
||
forged["result"]["estimate"]["physical_qubits"] += 1
|
||
assert not wq.verify_estimate(forged)["valid"]
|
||
# The same request, the same answer.
|
||
assert wq.estimate(LADDER, qubit="us-e3", budget=1e-4, max_slowdown=64, key=DEMO) == e
|
||
|
||
|
||
def test_an_estimate_on_the_reaction_clock_runs_each_toffoli_in_a_tick():
|
||
steps = wq.estimate(LADDER, qubit="us-e3", budget=1e-4, max_slowdown=1, key=DEMO)["result"]["estimate"]
|
||
e = wq.estimate(LADDER, qubit="us-e3", budget=1e-4, max_slowdown=1, key=DEMO, reaction_us=100)
|
||
r = e["result"]
|
||
assert r["options"]["clock"] == {"reaction_us": 100.0}
|
||
assert r["estimate"]["clock"] == {"reaction_us": 100.0, "tick_us": 100.0, "toffolis": 3}
|
||
assert "clock" not in steps and r["estimate"]["runtime_s"] < steps["runtime_s"]
|
||
assert r["estimate"]["receipt"]["clock"]["kind"] == "reaction"
|
||
assert wq.verify_estimate(e)["valid"]
|
||
# A reaction includes a measurement, and this qubit's takes 100 µs.
|
||
with pytest.raises(wq.EngineError):
|
||
wq.estimate(LADDER, qubit="us-e3", budget=1e-4, max_slowdown=1, key=DEMO, reaction_us=10)
|
||
|
||
|
||
def test_an_estimate_reads_and_writes_the_interchange_documents():
|
||
doc = wq.estimate(LADDER, export="counts", key=DEMO)["result"]["export"]
|
||
assert doc == {"numQubits": 5, "tCount": 1, "rotationCount": 1, "rotationDepth": 1, "cczCount": 3, "ccixCount": 0, "measurementCount": 4}
|
||
back = wq.estimate(counts=doc, key=DEMO)
|
||
assert back["result"]["program"]["origin"] == "counts" and wq.verify_estimate(back)["valid"]
|
||
routine = wq.estimate(LADDER, export="routine", name="ladder", key=DEMO)["result"]["export"]
|
||
assert routine["version"] == "v1" and routine["program"]["name"] == "ladder"
|
||
assert wq.estimate(routine=json.dumps(routine), key=DEMO)["result"]["program"]["origin"] == "routine"
|
||
|
||
|
||
def test_an_estimate_refuses_with_its_code():
|
||
with pytest.raises(wq.EngineError, match=r"\[QE08\]"):
|
||
wq.estimate(LADDER, qubit="nowhere", key=DEMO)
|
||
with pytest.raises(wq.EngineError, match=r"\[QE07\]"):
|
||
wq.estimate(counts={"tCount": 3}, key=DEMO)
|
||
with pytest.raises(TypeError):
|
||
wq.estimate(key=DEMO)
|
||
|
||
|
||
# -- verdicts and repairs -------------------------------------------------------
|
||
|
||
|
||
def test_a_run_with_an_intent_carries_its_verdict_sealed():
|
||
job = wq.run(BELL, shots=100, intent={"distribution": {"00": 0.5, "11": 0.5}}, key=DEMO)
|
||
v = job.verdict
|
||
assert (v["level"], v["holds"], v["intent"][0]["how"]) == ("proof", True, "statevector")
|
||
assert v["statement"].startswith("Checked as proof")
|
||
assert wq.verify(job).valid
|
||
forged = json.loads(json.dumps(job))
|
||
forged["result"]["verdict"]["level"] = "evidence"
|
||
assert not wq.verify(forged).valid
|
||
# The same verdict read back; another intent worked out here, not sealed.
|
||
again = wq.verdict(job)
|
||
assert again["level"] == "proof" and again["sealed"] is True
|
||
other = wq.verdict(job, {"distribution": {"00": 1.0}})
|
||
assert (other["level"], other["holds"], other["sealed"]) == ("proof", False, False)
|
||
# A run that asks for no verdict carries none.
|
||
assert wq.run(BELL, shots=10, key=DEMO).verdict is None
|
||
assert wq.run(BELL, shots=10, verdict=True, key=DEMO).verdict["level"] == "executed"
|
||
|
||
|
||
def test_a_refusal_carries_its_code_and_a_refused_job_its_verdict():
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.run("OPENQASM 3.0; qubit[1] q; rx(0.3) q[0];", method="stabilizer", key=DEMO)
|
||
assert e.value.code == "J011"
|
||
v = wq.verdict({"request": {"qasm": "OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n"}, "error": "line 3: unknown gate foo [Q103]"})
|
||
assert (v["level"], v["refusal"]["code"]) == ("refused", "Q103")
|
||
|
||
|
||
def test_repair_fixes_the_flagged_line_and_rejects_a_change_to_another():
|
||
r = wq.repair("OPENQASM 3.0;\nqubit[2] q\nh q[0];\n")
|
||
assert r["qasm"] == "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\n" and r["repairs"][0]["code"] == "Q004"
|
||
assert r["verdict"]["level"] == "parsed"
|
||
w = "OPENQASM 3.0;\nqubit[1] q;\nbit[1] c;\nc[0] = measure q[0];\nwhile (c[0]) {\n x q[0];\n c[0] = measure q[0];\n}\n"
|
||
assert wq.repair(w)["needs_person"][0]["code"] == "Q310"
|
||
assert wq.repair(w, max_iterations=3)["ok"] is True
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.repair("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n", proposed="OPENQASM 3.0;\nqubit[2] q;\nh q[0];\n")
|
||
assert e.value.code == "J021"
|
||
taken = wq.repair("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n", proposed="OPENQASM 3.0;\nqubit[1] q;\nh q[0];\n")
|
||
assert taken["ok"] is True and taken["repairs"][0]["by"] == "proposal"
|
||
|
||
|
||
def test_a_sampled_intent_is_evidence_and_never_proof():
|
||
job = wq.run(BELL, shots=10, seed=7, intent={"sampled": {"00": 0.5, "11": 0.5}, "test": "g_test", "alpha": 0.01, "shots": 2000}, key=DEMO)
|
||
v = job.verdict
|
||
assert (v["level"], v["holds"], v["intent"][0]["strength"], v["intent"][0]["how"]) == ("evidence", True, "evidence", "g_test")
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.run(BELL, shots=10, intent={"sampled": {"00": 0.5, "11": 0.5}, "test": "g_test", "shots": 2000}, key=DEMO)
|
||
assert e.value.code == "J020" and "alpha" in e.value.message
|
||
|
||
|
||
# -- energy-landscape jobs -----------------------------------------------------
|
||
|
||
|
||
def test_an_energy_landscape_job_is_sealed_and_verified_by_its_form():
|
||
talks = next(e for e in wq.landscape_examples() if e["name"] == "talks")
|
||
assert talks["expect"] == "works"
|
||
for op, extra in (("run", {"seed": "7"}), ("replay", {"seeds": 3}), ("sweep", {"seeds": 2, "temperatures": [0, 1.5]}), ("compare", {"seeds": 2})):
|
||
job = wq.landscape(talks["program"], op, key=DEMO, **extra)
|
||
assert job["result"]["schema"] == "wai.quantum.landscape/1", op
|
||
v = wq.verify(job)
|
||
assert v.valid and v.reasons == [], (op, dict(v))
|
||
assert v["landscape"]["op"] == op
|
||
run = wq.landscape(talks["program"], seed="7", key=DEMO)["result"]
|
||
assert (run["verdict"], run["energy"], run["member"]["id"], run["joules"]["kind"]) == ("solved", -10, "cpu:reference", "priced")
|
||
# The same request, the same bits, given its seed as text or as a number.
|
||
assert wq.landscape(talks["program"], seed=7, key=DEMO)["result"]["replay"] == run["replay"]
|
||
# An answer stated from another member, sealed again: its form is broken, and verify says so.
|
||
job = wq.landscape(talks["program"], seed="7", key=DEMO)
|
||
result = {k: v for k, v in job["result"].items() if k != "receipt"}
|
||
result["member"] = dict(result["member"], parity="bit-exact")
|
||
result["receipt"] = wq.seal(job["request"], result, key=DEMO)
|
||
v = wq.verify({"request": job["request"], "result": result})
|
||
assert not v.valid and v.reasons == ["landscape_form"], dict(v)
|
||
|
||
|
||
def test_a_landscape_program_the_reader_refuses_names_its_line(tmp_path):
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.landscape("vars 2\noppos 0 1\nsolve\n")
|
||
assert str(e.value).startswith("Line 2: ") or "Line 2: " in str(e.value)
|
||
with pytest.raises(wq.EngineError):
|
||
wq.landscape("vars 2\nagree 0 1\nminimize\n", "examples")
|
||
path = tmp_path / "chain.epuml"
|
||
path.write_text("vars 3\nagree 0 1\nagree 1 2\nfield 0 1\nminimize\n")
|
||
assert wq.landscape(path, key=DEMO)["result"]["state"] == "+++"
|
||
|
||
|
||
# -- qudit jobs and wasmQ envelopes -------------------------------------------------
|
||
|
||
QUTRITS = {"d": 3, "qudits": 2, "gates": [{"gate": "fourier", "qudits": [0]}, {"gate": "csum", "qudits": [0, 1]}], "shots": 1000, "seed": 7}
|
||
|
||
|
||
def test_a_qudit_job_is_sealed_and_runs_again_to_the_same_bits(tmp_path):
|
||
job = wq.qudit(QUTRITS, amplitudes=["00", "11"], key=DEMO)
|
||
r = job["result"]
|
||
assert r["schema"] == "wai.quantum.qudit/1" and set(r["counts"]) <= {"00", "11", "22"} and sum(r["counts"].values()) == 1000
|
||
# 1/sqrt(3) at 2^-30, the nearest whole number (quantum-sim.md section 9.2).
|
||
assert r["amplitudes"]["00"] == {"re": 619925131, "im": 0}
|
||
assert set(r["energy"]) == {"unmetered"} and r["energy"]["unmetered"]
|
||
assert wq.verify(job).valid
|
||
c = wq.qudit_check(job)
|
||
assert c["verified"] and c["reproduced"] and c["differences"] == [], c
|
||
# Its counts changed and sealed again: the seal holds, the run again does not.
|
||
forged = {k: v for k, v in r.items() if k != "receipt"}
|
||
forged["counts"] = {"00": 1000}
|
||
forged["receipt"] = wq.seal(job["request"], forged, key=DEMO)
|
||
c = wq.qudit_check({"request": job["request"], "result": forged})
|
||
assert c["signature"] and not c["verified"] and any(d.startswith("counts") for d in c["differences"]), c
|
||
# The same request from a file, the flags in place of its own.
|
||
path = tmp_path / "bell.json"
|
||
path.write_text(json.dumps(QUTRITS))
|
||
again = wq.qudit(path, shots=10, key=DEMO)["result"]
|
||
assert again["shots"] == 10 and again["state"] == r["state"]
|
||
for bad, code in (({"d": 5, "qudits": 1, "gates": []}, "J091"), ({"d": 2, "qudits": 30, "gates": []}, "J092"), ({"d": 3, "qudits": 1, "gates": [{"gate": "swap", "qudits": [0]}]}, "J090")):
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.qudit(bad)
|
||
assert code in str(e.value), (bad, str(e.value))
|
||
|
||
|
||
def test_every_wasmq_example_runs_here_and_verifies_by_its_form_and_its_run():
|
||
examples = wq.wasmq_examples()
|
||
assert {"surface3-decode64", "v2-t-toffoli", "fault-fuel"} <= {e["name"] for e in examples}
|
||
for e in examples:
|
||
job = wq.wasmq(e["envelope"], key=DEMO)
|
||
assert job["result"]["profile"] == e["profile"] and set(job) == {"status", "method", "request", "result"}, e["name"]
|
||
v = wq.verify(job)
|
||
assert v.valid and v["wasmq"]["problems"] == [] and v["wasmq"]["notes"] == [], (e["name"], dict(v))
|
||
w = wq.wasmq_verify(job, e["envelope"])
|
||
assert w["valid"] and w["bundle"] is True and w["reproduced"] is True and w["signer"] == DEMO.did, (e["name"], w)
|
||
if e["profile"] == "wai.quantum.wasmq/2":
|
||
run = job["result"]["energy"]["run"]
|
||
assert set(run) == {"unmetered"} and run["unmetered"], run
|
||
fault = wq.wasmq("fault-fuel", key=DEMO)
|
||
assert (fault["status"], fault["result"]["fault"]["reason"]) == ("fault", "fuel_exhausted")
|
||
assert wq.verify(fault).valid
|
||
# Another envelope is not the one the receipt binds.
|
||
other = wq.wasmq_verify(fault, "v2-t-toffoli")
|
||
assert other["bundle"] is False and not other["valid"], other
|
||
|
||
|
||
def test_a_wasmq_envelope_is_read_from_a_file_and_one_the_profile_refuses_says_its_code(tmp_path):
|
||
e = next(x for x in wq.wasmq_examples() if x["name"] == "v2-feedback-rotation")
|
||
path = tmp_path / "rotation.wai"
|
||
path.write_bytes(e["envelope"])
|
||
assert wq.wasmq(path, key=DEMO) == wq.wasmq(e["envelope"], key=DEMO) == wq.wasmq("v2-feedback-rotation", key=DEMO)
|
||
with pytest.raises(wq.EngineError) as err:
|
||
wq.wasmq(b"WAI1 not an envelope")
|
||
assert "W013" in str(err.value)
|
||
with pytest.raises(wq.EngineError):
|
||
wq.wasmq("no-such-example")
|
||
|
||
|
||
# -- error correction and mitigation ------------------------------------------
|
||
|
||
|
||
def test_a_code_comes_with_its_distance_certified_and_sealed():
|
||
c = wq.qec_code("color:5", key=DEMO)
|
||
r = c["result"]
|
||
assert (r["code"]["n"], r["code"]["k"]) == (19, 1)
|
||
cert = r["certificate"]
|
||
assert cert["distance"] == 5 and cert["x"]["distance"] == cert["z"]["distance"] == 5
|
||
assert len(cert["z"]["witness"]) == 5 and "color" in r["code"]["decoders"]
|
||
assert wq.verify(c).valid
|
||
# The certificate checks against its code, as a sealed answer or alone; a
|
||
# changed one does not.
|
||
for given in (c, cert, json.dumps(cert)):
|
||
check = wq.qec_code("color:5", certificate=given, key=DEMO)["result"]["check"]
|
||
assert check["verified"] and check["witnesses"] and check["search_reproduced"], check
|
||
forged = json.loads(json.dumps(cert))
|
||
forged["z"]["witness"] = forged["z"]["witness"][:-1]
|
||
assert not wq.qec_code("color:5", certificate=forged, key=DEMO)["result"]["check"]["verified"]
|
||
# The same request, the same answer; a search stopped short says so.
|
||
assert wq.qec_code("color:5", key=DEMO) == c
|
||
assert wq.qec_code("surface:5", certify=False, key=DEMO)["result"].get("certificate") is None
|
||
|
||
|
||
def test_every_error_below_half_the_distance_leaves_the_logical_state_unharmed():
|
||
# The distance-5 colour code: every error of weight 1 or 2, decoded by
|
||
# restriction to matching, leaves a product of stabilizers.
|
||
n = 19
|
||
errors = [[q] for q in range(n)] + [[a, b] for a in range(n) for b in range(a + 1, n)]
|
||
for e in errors:
|
||
r = wq.qec_decode("color:5", error=e, key=DEMO)["result"]
|
||
assert r["explains"] and not r["residual"]["logical"], (e, r["residual"])
|
||
# A syndrome alone decodes to a correction that explains it.
|
||
r = wq.qec_decode("surface:3", syndrome=[1], decoder="matching", key=DEMO)["result"]
|
||
assert r["explains"] and r["syndrome"] == [1]
|
||
|
||
|
||
def test_a_circuits_detectors_are_decoded_and_its_shots_counted():
|
||
one = wq.qec_decode(memory=(3, 3, 0.004), shots=2000, seed=3, decoder="correlated", key=DEMO)
|
||
r = one["result"]
|
||
assert r["shots"] == 2000 and 0 <= r["failures"] < 200
|
||
assert r["stderr"] == pytest.approx(math.sqrt(r["rate"] * (1 - r["rate"]) / 2000))
|
||
assert wq.verify(one).valid
|
||
assert wq.qec_decode(memory={"distance": 3, "rounds": 3, "p": 0.004}, shots=2000, seed=3, decoder="correlated", key=DEMO) == one
|
||
# A window over the rounds reads the same model.
|
||
w = wq.qec_decode(memory=(3, 4, 0.001), detectors=[], decoder="window", window=(2, 2), key=DEMO)["result"]
|
||
assert w["predicted"] == [] and len(w["detail"]["windows"]) >= 2
|
||
|
||
|
||
def test_a_threshold_measures_rates_with_their_errors():
|
||
t = wq.qec_threshold("color", distances=[3, 5, 7], p=[0.02], shots=2000, key=DEMO)
|
||
r = t["result"]
|
||
assert [pt["distance"] for pt in r["points"]] == [3, 5, 7]
|
||
assert all(pt["shots"] == 2000 and "stderr" in pt for pt in r["points"])
|
||
assert r["below_threshold"][0]["distance_helps"] is True
|
||
assert wq.verify(t).valid
|
||
|
||
|
||
def test_error_correction_refuses_with_its_code():
|
||
with pytest.raises(wq.EngineError, match=r"\[J009\]"):
|
||
wq.qec_code("hexagon:5", key=DEMO)
|
||
with pytest.raises(wq.EngineError, match=r"\[J009\]"):
|
||
wq.qec_threshold(shots=5_000_000, key=DEMO)
|
||
with pytest.raises(TypeError):
|
||
wq.qec_decode(memory=(3, 3), key=DEMO)
|
||
|
||
|
||
def test_a_mitigated_run_carries_each_value_raw_and_mitigated_with_its_assumptions():
|
||
ghz = "OPENQASM 3.0;\nqubit[3] q;\nh q[0];\ncx q[0], q[1];\ncx q[1], q[2];\nrz(0.3) q[2];\nrx(0.2) q[0];\n"
|
||
noise = {"depolarizing_2q": 0.02, "depolarizing_1q": 0.004}
|
||
job = wq.run(ghz, observables=["Z0 Z1"], noise=noise, mitigate={"method": "zne", "shots": 4000}, key=DEMO)
|
||
m = job["result"]["mitigation"]
|
||
assert m["method"] == "zne" and m["assumptions"]["bias"] and m["assumptions"]["variance"]
|
||
row = m["observables"][0]
|
||
assert abs(row["mitigated"] - row["ideal"]) < abs(row["raw"] - row["ideal"]) and row["stderr"] > 0
|
||
assert job.verify().valid
|
||
forged = json.loads(json.dumps(job))
|
||
forged["result"]["mitigation"]["observables"][0]["mitigated"] = 1.0
|
||
assert not wq.verify(forged).valid
|
||
# A method's name alone takes its defaults; mitigation without noise is refused.
|
||
assert wq.run(ghz, observables=["Z0 Z1"], noise=noise, mitigate="vd", key=DEMO)["result"]["mitigation"]["method"] == "vd"
|
||
with pytest.raises(wq.EngineError, match=r"\[J008\]"):
|
||
wq.run(ghz, observables=["Z0"], mitigate="zne", key=DEMO)
|
||
|
||
|
||
def test_every_receipt_of_the_corpus_gets_its_verdict_here():
|
||
"""The text of each job: read as the provider protocol's section 8.2 reads
|
||
it, and held to every rule of section 8.4, as the command line holds it."""
|
||
corpus = pathlib.Path(__file__).resolve().parents[2] / "quantum-conformance" / "receipts"
|
||
manifest = json.loads((corpus / "manifest.json").read_text())
|
||
for e in manifest["entries"]:
|
||
text = (corpus / e["file"]).read_bytes()
|
||
v = wq.verify(text)
|
||
assert sorted(v.reasons) == sorted(e["reasons"]), e["file"]
|
||
assert v.valid is (not e["reasons"]), e["file"]
|
||
|
||
|
||
def test_a_dict_is_held_to_every_rule_it_can_show():
|
||
corpus = pathlib.Path(__file__).resolve().parents[2] / "quantum-conformance" / "receipts"
|
||
for name, reason in [("crafted-provider-measured-simulator", "provider_class"), ("crafted-number-inexact", "inexact_number")]:
|
||
job = json.loads((corpus / "jobs" / (name + ".json")).read_text())
|
||
v = wq.verify(job)
|
||
assert not v.valid and reason in v.reasons, name
|
||
# Text and a path are read as text: a member named twice is found.
|
||
twice = corpus / "jobs" / "crafted-read-duplicate.json"
|
||
assert not wq.verify(str(twice)).valid
|
||
assert wq.verify(str(twice)).reasons == ["unreadable"]
|
||
assert wq.verify(twice.read_text()).reasons == ["unreadable"]
|
||
|
||
|
||
def test_a_dict_is_judged_as_its_text_is():
|
||
"""What a dict can show gets the verdict every surface gives: a body nested deeper
|
||
than a job's text holds is unreadable, and a number past 64 bits in a member nothing
|
||
seals is a number like any other. Text that begins with a byte order mark is read as
|
||
text, and is unreadable, not taken for a path."""
|
||
corpus = pathlib.Path(__file__).resolve().parents[2] / "quantum-conformance" / "receipts" / "jobs"
|
||
job = json.loads((corpus / "cli-wasmq-run.json").read_text())
|
||
assert wq.verify(job).valid, "the positive control"
|
||
nested = 0
|
||
for _ in range(128):
|
||
nested = [nested]
|
||
deep = dict(job, note=nested)
|
||
assert wq.verify(deep).reasons == ["unreadable"]
|
||
assert wq.verify(dict(job, id=2 ** 64 + 1)).valid
|
||
assert wq.verify("\ufeff" + json.dumps(job)).reasons == ["unreadable"]
|
||
|
||
|
||
def test_every_str_and_dict_gets_a_verdict(tmp_path, monkeypatch):
|
||
"""verify never raises for a str or a dict: a str is a path when it names a file and
|
||
the job's text otherwise, whatever it begins with; a lone surrogate, in text or in a
|
||
dict json.loads made of its escape, is unreadable; a dict that is no JSON at all is
|
||
unreadable. A path that begins with `[` is still a path."""
|
||
corpus = pathlib.Path(__file__).resolve().parents[2] / "quantum-conformance" / "receipts" / "jobs"
|
||
text = (corpus / "cli-wasmq-run.json").read_text()
|
||
for given, want in [
|
||
("null", ["not_a_job"]), ("5", ["not_a_job"]), ('"job"', ["not_a_job"]), ("", ["unreadable"]),
|
||
(" \r\n\t", ["unreadable"]), ("\x0c" + text, ["unreadable"]), ("\u00a0" + text, ["unreadable"]),
|
||
("\u2028" + text, ["unreadable"]), ("\x00" + text, ["unreadable"]), (" " * 5000 + "x", ["unreadable"]),
|
||
("signature VALID\n", ["unreadable"]), ('{"status": "\ud800"}', ["unreadable"]),
|
||
("no-such-job.json", ["unreadable"]),
|
||
]:
|
||
assert wq.verify(given).reasons == want, (given[:40], wq.verify(given).reasons)
|
||
lone = json.loads(text.replace('"status":"ok"', '"status":"\\ud800"', 1))
|
||
assert lone["result"]["status"] == "\ud800", "json.loads makes a lone surrogate of its escape"
|
||
assert wq.verify(lone).reasons == ["unreadable"], "a dict with a lone surrogate"
|
||
assert wq.verify({"request": {}, "result": {"x": {1, 2}}}).reasons == ["unreadable"], "a dict that is no JSON"
|
||
monkeypatch.chdir(tmp_path)
|
||
(tmp_path / "[run].json").write_text(text)
|
||
assert wq.verify("[run].json").valid, "a path that begins with ["
|
||
|
||
|
||
def test_a_set_s_text_is_judged_as_a_set():
|
||
"""A genuine set, given as text, a path or bytes, is valid."""
|
||
corpus = pathlib.Path(__file__).resolve().parents[2] / "quantum-conformance" / "receipts" / "jobs"
|
||
for name in ("cli-set-programs.json", "cli-set-sweep.json"):
|
||
path = corpus / name
|
||
assert wq.verify(json.loads(path.read_text())).valid, name
|
||
assert wq.verify(str(path)).valid, (name, wq.verify(str(path)))
|
||
assert wq.verify(path.read_bytes()).valid, name
|