"""Application CI: exercise HTTP clients and assert protected filesystem effects. Build the Rust binaries first. This runner owns temporary state and subprocesses; it is optional verification tooling, not a prerequisite for using the apps. """ from contextlib import contextmanager import concurrent.futures import json import os from pathlib import Path import subprocess import tempfile import time import urllib.error import urllib.request BIN = Path(os.environ.get("CHIO_APP_BIN_DIR", "target/debug")).resolve() @contextmanager def host(binary, port, state=None, **extra): with tempfile.TemporaryDirectory(prefix="chio-application-") as temporary: directory = Path(state or temporary) with (directory / "host.log").open("a") as log: process = subprocess.Popen([str(BIN / binary)], env={**os.environ, "CHIO_APP_STATE": str(directory), **extra}, stdout=log, stderr=log) try: for _ in range(200): if process.poll() is not None: raise AssertionError((directory / "host.log").read_text()) try: urllib.request.urlopen(f"http://127.0.0.1:{port}/", timeout=.2).close() break except (OSError, urllib.error.URLError): time.sleep(.1) else: raise AssertionError("Application did not start") yield directory finally: process.terminate() try: process.wait(timeout=20) except subprocess.TimeoutExpired: process.kill() process.wait() raise AssertionError("host required forced shutdown") assert process.returncode == 0, (process.returncode, (directory / "host.log").read_text()) def api(port, token, path, body=None, expected=200): request = urllib.request.Request(f"http://127.0.0.1:{port}{path}", data=None if body is None else json.dumps(body).encode(), headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}) try: response = urllib.request.urlopen(request, timeout=15) except urllib.error.HTTPError as error: response = error with response: content = response.read().decode() assert response.status == expected, (path, response.status, content) if path == "/export": return [json.loads(line) for line in content.splitlines()] try: return json.loads(content) except json.JSONDecodeError: assert expected >= 400, content return {"error": content} def approvals(): with host("approval-desk", 4310, APPROVAL_TTL_SECONDS="2") as state: agent = (state / "agent-token.txt").read_text() mara = (state / "mara-token.txt").read_text() def proposal(sku="sku-301", percent=80, customer="basalt"): return api(4310, agent, "/proposals", {"sku": sku, "percent": percent, "customer": customer})["approval_id"] def decision(id, choice, **changes): preview = api(4310, mara, f"/proposals/{id}") return {"decision": choice, "parameter_hash": preview["pending"]["parameter_hash"], **changes} first = proposal() assert not list((state / "price-changes").iterdir()) body = decision(first, "approve") api(4310, agent, f"/proposals/{first}/decision", body, 401) api(4310, mara, f"/proposals/{first}/decision", {**body, "parameter_hash": "changed"}, 409) api(4310, mara, f"/proposals/{first}/decision", {**body, "arguments": {"percent": 85}}, 422) allowed = api(4310, mara, f"/proposals/{first}/decision", body) assert allowed["verdict"] == "Allow", allowed assert json.loads((state / "price-changes" / f"{first}.json").read_text())["price"] == "28.00" api(4310, mara, f"/proposals/{first}/decision", body, 409) denied_id = proposal(percent=90) denied = api(4310, mara, f"/proposals/{denied_id}/decision", decision(denied_id, "reject")) assert denied["verdict"] == "Deny" and denied["receipt_id"] != allowed["receipt_id"] assert not (state / "price-changes" / f"{denied_id}.json").exists() expired = proposal(percent=75) expired_decision = decision(expired, "approve") time.sleep(2.1) api(4310, mara, f"/proposals/{expired}/decision", expired_decision, 409) assert not (state / "price-changes" / f"{expired}.json").exists() old_key = (state / "kernel-public-key.txt").read_text() revoked = api(4310, mara, "/agent/revoke", {}) revoked_id = proposal(percent=70) refused = api(4310, mara, f"/proposals/{revoked_id}/decision", decision(revoked_id, "approve")) assert refused["verdict"] == "Deny" and not (state / "price-changes" / f"{revoked_id}.json").exists() rotated = api(4310, mara, "/agent/rotate", {}) assert rotated["old_capability_id"] == revoked["revoked_capability_id"] assert rotated["capability_id"] != rotated["old_capability_id"] and rotated["kernel_key"] == old_key new_id = proposal(percent=60) renewed = api(4310, mara, f"/proposals/{new_id}/decision", decision(new_id, "approve")) assert renewed["verdict"] == "Allow" and (state / "price-changes" / f"{new_id}.json").exists() print("PASS approvals: exact approved price change; rejection, expiry, mutation, an unauthorized approver, and replay change no price") print("PASS rotation: revoked authority cannot publish a price; fresh agent authority can; the kernel verification key is unchanged") def customers(): with host("customer-service", 4311) as state: tokens = {name: (state / f"{name}-token.txt").read_text() for name in ("priya", "tomas", "operator")} priya = api(4311, tokens["priya"], "/quotes", {"sku": "sku-118", "percent": 10}) tomas = api(4311, tokens["tomas"], "/quotes", {"sku": "sku-204", "percent": 25}) assert priya["verdict"] == tomas["verdict"] == "Allow", (priya, tomas) assert priya["receipt"]["tenant_id"] == "harbor-goods" and tomas["receipt"]["tenant_id"] == "ostrander" api(4311, tokens["priya"], "/receipts/" + tomas["receipt_id"], expected=404) assert api(4311, tokens["tomas"], "/receipts/" + tomas["receipt_id"])["id"] == tomas["receipt_id"] api(4311, tokens["priya"], "/receipts?tenant=ostrander", expected=409) api(4311, tokens["priya"], "/quotes", {"sku": "sku-118", "percent": 10, "tenant": "ostrander"}, 422) assert {row["tenant_id"] for row in api(4311, tokens["priya"], "/export")} == {"harbor-goods"} assert {row["tenant_id"] for row in api(4311, tokens["tomas"], "/export")} == {"ostrander"} assert api(4311, tokens["operator"], "/receipts")["total"] == 2 assert api(4311, tokens["priya"], "/quotes")["quotes"][0]["price"] == "21.60" assert api(4311, tokens["tomas"], "/quotes")["quotes"][0]["price"] == "43.50" print("PASS tenancy: authenticated quotes, separate accounts, exact receipt reads, scoped exports, and separate operator access") def terminal(token, id): for _ in range(160): result = api(4312, token, f"/jobs/{id}") if result["result"] is not None: return result time.sleep(.1) raise AssertionError("The repricing job never reached a terminal outcome") def repricing(): for mode, delay in [("refuse", None), ("complete", 0), ("complete", 1.4), ("fail", None), ("uncertain", None)]: with host("repricing-service", 4312) as state: token = (state / "client-token.txt").read_text() api(4312, token, "/jobs", {"percent": 140, "mode": mode}, 400) id = api(4312, token, "/jobs", {"percent": 10, "mode": mode})["job_id"] if delay is not None: time.sleep(delay) api(4312, token, f"/jobs/{id}/cancel", {}) result = terminal(token, id) if delay == 0: assert result["result"].get("status") == "cancelled_before_dispatch", result assert api(4312, token, "/budget")["available"] == 200 else: assert result["result"]["receipt_id"], result if mode == "refuse" or delay == 0: assert not result["started"] and not result["published"], result elif mode == "uncertain": assert result["published"], result else: assert result["started"] and not result["published"], result print("PASS repricing", mode, "cancel_delay=", delay, "verdict=", result["result"].get("verdict"), "published=", result["published"]) with tempfile.TemporaryDirectory(prefix="chio-shared-budget-") as persistent: with host("repricing-service", 4312, persistent) as state: token = (state / "client-token.txt").read_text() with concurrent.futures.ThreadPoolExecutor(max_workers=3) as workers: ids = list(workers.map(lambda _: api(4312, token, "/jobs", {"percent": 10, "mode": "complete"})["job_id"], range(3))) results = [terminal(token, id) for id in ids] assert sum(result["published"] for result in results) == 2, results assert len({result["result"]["receipt_id"] for result in results}) == 3 before = api(4312, token, "/budget") assert before["spent"] == 160 and before["available"] == 40, before with host("repricing-service", 4312, persistent): after = api(4312, token, "/budget") assert after == before, (before, after) id = api(4312, token, "/jobs", {"percent": 10, "mode": "complete"})["job_id"] assert not terminal(token, id)["started"] print("PASS shared allowance: three concurrent workers, two price sheets, exact receipts, exhaustion, and an unchanged balance after restart") def shutdown_active_job(): with tempfile.TemporaryDirectory(prefix="chio-repricing-shutdown-") as persistent: with host("repricing-service", 4312, persistent) as state: token = (state / "client-token.txt").read_text() id = api(4312, token, "/jobs", {"percent": 10, "mode": "complete"})["job_id"] for _ in range(100): if api(4312, token, f"/jobs/{id}")["started"]: break time.sleep(.05) else: raise AssertionError("the repricing job never entered execution") # Exiting the owned host sends SIGTERM while a worker is active. result = json.loads((Path(persistent) / "jobs" / f"{id}.result.json").read_text()) assert result.get("receipt_id"), result assert not (Path(persistent) / "price-sheets" / f"{id}.csv").exists() print("PASS SIGTERM: active repricing job cancelled, final receipt retained, no price sheet, host joined cleanly") if __name__ == "__main__": approvals() customers() repricing() shutdown_active_job()