jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889"""Offline contract for the deployment-side admission gate.
The previous experiment deployed an image no receipt covered, and nothingrefused it. These cases pin the refusals, not the happy path: the gate is onlyworth anything if it says no to everything it has not seen pass.
python3 tests/admission_contract.py"""import jsonimport pathlibimport subprocessimport sysimport tempfile
ADMIT = pathlib.Path(__file__).resolve().parent.parent / "scripts" / "admit"DIGEST = "sha256:" + "a" * 64OTHER = "sha256:" + "b" * 64
def verify(receipts_dir, digest): return subprocess.run( [str(ADMIT), "verify", digest], capture_output=True, text=True, env={"PATH": "/usr/bin:/bin:/usr/local/bin", "STREAM_RECEIPTS_DIR": str(receipts_dir)}, )
def write_receipt(receipts_dir, name, *, digest, suites): (receipts_dir / name).write_text( json.dumps( { "schema": "stream-admission/1", "image": {"tag": "atcr.io/zat.dev/stream:test", "registry_digest": digest}, "revisions": {"stream": "0" * 40}, "suites": suites, } ) )
ALL_PASS = {"unit-debug": "pass", "archive-contract": "pass"}
def case(label, condition): if not condition: print(f"FAIL: {label}") sys.exit(1) print(f" ok: {label}")
with tempfile.TemporaryDirectory() as tmp: receipts = pathlib.Path(tmp)
# Nothing admitted yet: the gate must refuse, not default open. r = verify(receipts, DIGEST) case("empty receipts dir rejects", r.returncode != 0 and "REJECT" in r.stdout)
write_receipt(receipts, "abc1234.json", digest=DIGEST, suites=ALL_PASS) r = verify(receipts, DIGEST) case("fully passing receipt admits its own digest", r.returncode == 0 and "ADMITTED" in r.stdout)
# The core of the e1926f3 failure: a *different* artifact must not inherit # another artifact's receipt. r = verify(receipts, OTHER) case("a different digest is rejected", r.returncode != 0 and "REJECT" in r.stdout)
# A receipt that never reached publish has no digest to stand behind. write_receipt(receipts, "unpublished.json", digest=None, suites=ALL_PASS) r = verify(receipts, "null") case("unpublished receipt admits nothing", r.returncode != 0)
# A skipped suite is not a passed suite. Silence must not read as success. skipped = receipts / "skipped" skipped.mkdir() write_receipt(skipped, "def5678.json", digest=DIGEST, suites={**ALL_PASS, "powerloss-oracle": "skipped"}) r = verify(skipped, DIGEST) case("a skipped suite blocks admission", r.returncode != 0 and "non-passing" in r.stdout)
failed = receipts / "failed" failed.mkdir() write_receipt(failed, "aaa1111.json", digest=DIGEST, suites={**ALL_PASS, "differential-oracle": "fail"}) r = verify(failed, DIGEST) case("a failed suite blocks admission", r.returncode != 0 and "non-passing" in r.stdout)
print("ADMISSION GATE CONTRACT PASS")