open-standards/wai/wai-quantum-py/tests/conftest.py
Transaction Science c4bda95cca wai-quantum 0.7.0 integrated: the generated material on the final tree, the pins it moves, and what replaying every job found
Generated on the final tree: the views page, rebuilt (802 495 bytes, under
1 MiB) from its sources; the notebooks' stored outputs; the agent docs. The
pins that move: wai-web's course check names its engine now, so its digest
moves (374b400e... to 969b2c1c..., and with every release from here), and the
page's wasm32 count is 18 (native 26, as pinned). wai-quantum-qpu's library
count is 34, as the cherry-picked pin says.

What replaying each workflow's jobs found, each fixed here:
- the views page's one line carried the estimate pane's English note, "no
  other beats on both", which the comparative-prose check reads across the
  line; the note says "none of the others beats on both";
- the docs runner's broken.qasm is broken on purpose, so the editors check
  requires it to fail to parse, as the intent set's broken programs;
- the command line's parse test named the service's submit and task in a
  build without the service; those checks are a test of their own, built
  with it;
- the set-only Merkle helpers warned in a build without receipts;
- a one-element loop in the service's strict test;
- the energy job looked for the live counter's line at the start of a line,
  where --nocapture puts it after the test's name;
- the Python test comparing a run on the service with one here now leaves out
  the provenance the service adds, and checks it is there.

The release heading's caller list is the final API check's: the frame text's
new fields and variant, and the request's compile passes and mitigation.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-28 08:43:06 -04:00

134 lines
4.1 KiB
Python

"""Fixtures: a fresh key directory per test, and the other surfaces to hold the
package to (the command line and the job service, built from this checkout).
``WAI_QUANTUM_CLI`` and ``WAI_QUANTUM_SERVICE_BIN`` name the binaries. Without
them the tests that need them are skipped, unless ``WAI_QUANTUM_REQUIRE_ALL=1``
(as CI sets it), which fails them instead: a skipped cross-check is not a
passed one.
"""
from __future__ import annotations
import json
import os
import shutil
import socket
import subprocess
import time
import urllib.request
from pathlib import Path
import pytest
HERE = Path(__file__).resolve().parent
PACKAGE = HERE.parent
REPO = PACKAGE.parent.parent
REQUIRE_ALL = os.environ.get("WAI_QUANTUM_REQUIRE_ALL") == "1"
BELL = "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n"
def needed(what: str, why: str):
"""Skip, or fail when everything is required."""
if REQUIRE_ALL:
pytest.fail(f"{what} is required (WAI_QUANTUM_REQUIRE_ALL=1): {why}")
pytest.skip(f"{what}: {why}")
def binary(env: str) -> str:
p = os.environ.get(env)
if not p or not Path(p).is_file():
needed(env, "names no binary")
return p
@pytest.fixture(autouse=True)
def own_config(tmp_path, monkeypatch):
"""Each test keeps its keys in a directory of its own, never the user's."""
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
monkeypatch.delenv("WAI_QUANTUM_KEY", raising=False)
monkeypatch.delenv("WAI_QUANTUM_SERVICE", raising=False)
@pytest.fixture(scope="session")
def cli() -> str:
return binary("WAI_QUANTUM_CLI")
@pytest.fixture
def run_cli(cli, tmp_path):
"""The command line, run with arguments (and a program on a file when
given): (exit status, standard output)."""
def go(*args: str, stdin: str = None, check: bool = True):
p = subprocess.run([cli, *args], input=stdin, capture_output=True, text=True, timeout=300)
if check and p.returncode != 0:
raise AssertionError(f"wai-quantum {' '.join(args)} exited {p.returncode}: {p.stderr}")
return p.returncode, p.stdout
return go
def free_port() -> int:
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@pytest.fixture(scope="session")
def service_url(tmp_path_factory):
"""The job service built from this checkout, serving on a port of its own:
one worker, so a second job waits in the queue."""
exe = binary("WAI_QUANTUM_SERVICE_BIN")
d = tmp_path_factory.mktemp("service")
port = free_port()
env = dict(
os.environ,
WAI_QSVC_ADDR=f"127.0.0.1:{port}",
WAI_QSVC_KEY=str(d / "service.key"),
# The store in the test's own scratch, never the person's home.
WAI_QSVC_STORE=str(d / "store"),
WAI_QSVC_BUDGET_GB="4",
WAI_QSVC_JOB_GB="2",
WAI_QSVC_WORKERS="1",
WAI_QSVC_MIN_FREE_PCT="0",
WAI_QSVC_CLIENT_ACTIVE="8",
WAI_QSVC_CLIENT_BURST="10000",
)
log = open(d / "service.log", "w")
p = subprocess.Popen([exe, "serve"], env=env, stdout=log, stderr=subprocess.STDOUT)
url = f"http://127.0.0.1:{port}"
for _ in range(100):
try:
with urllib.request.urlopen(f"{url}/v1/health", timeout=2) as r:
if json.loads(r.read()).get("ok"):
break
except OSError:
time.sleep(0.2)
else:
p.kill()
raise RuntimeError(f"the service did not start: {(d / 'service.log').read_text()}")
yield url
p.terminate()
try:
p.wait(10)
except subprocess.TimeoutExpired:
p.kill()
log.close()
def bare(result: dict) -> dict:
"""A result without what differs from host to host by design: its time,
its host's word on energy, the provenance a host adds (which names the
host), and its receipt (which seals them)."""
r = dict(result)
for k in ("elapsed_ms", "energy", "provenance", "receipt"):
r.pop(k, None)
return r
def node() -> str:
exe = os.environ.get("WAI_QUANTUM_NODE") or shutil.which("node")
if not exe:
needed("node", "runs the widgets' front ends")
return exe