diff --git a/scripts/audio_controller_connect.py b/scripts/audio_controller_connect.py index 864d067..eba3850 100644 --- a/scripts/audio_controller_connect.py +++ b/scripts/audio_controller_connect.py @@ -24,6 +24,7 @@ Exit 0 when no test failed. BLOCKED does not fail the run; it is a finding. from __future__ import annotations import argparse +import hashlib import json import os import ssl @@ -408,6 +409,44 @@ def t04_ingest_exists(client: Client, **csrf_note}) +def _load_bundle(bundle_dir: str) -> tuple[dict[str, Any] | None, + list[dict[str, Any]] | None, + str | None]: + """Read bundle.json + manifest.jsonl, or say why they are unusable.""" + + root = Path(bundle_dir) + envelope_path = root / "bundle.json" + manifest_path = root / "manifest.jsonl" + if not envelope_path.is_file() or not manifest_path.is_file(): + return None, None, f"{bundle_dir} is missing bundle.json or manifest.jsonl" + envelope = json.loads(envelope_path.read_text(encoding="utf-8")) + records = [json.loads(line) for line in + manifest_path.read_text(encoding="utf-8").splitlines() + if line.strip()] + return envelope, records, None + + +def _dry_run_once(client: Client, + bundle_dir: str, + headers: dict[str, str]) -> tuple[int, Any, int]: + """One dry run. Returns (status, body, duration_ms).""" + + envelope, records, _problem = _load_bundle(bundle_dir) + return client.request( + "POST", "/api/node/audio/import", headers=headers, + body={"envelope": envelope, "records": records, + "dry_run": True, "confirm": None}) + + +def _digest(body: Any) -> str | None: + """A comparable fingerprint of a dry-run response.""" + + if not isinstance(body, dict): + return None + return hashlib.sha256( + json.dumps(body, sort_keys=True).encode("utf-8")).hexdigest() + + def t05_dry_run(client: Client, bundle_dir: str | None, headers: dict[str, str]) -> Result: @@ -427,23 +466,12 @@ def t05_dry_run(client: Client, "against; pass --bundle to run a real dry run", {"depends_on": None, "ran": False}) - root = Path(bundle_dir) - envelope_path = root / "bundle.json" - manifest_path = root / "manifest.jsonl" - if not envelope_path.is_file() or not manifest_path.is_file(): - return Result("T05", "Dry-run validation", "BLOCKED", - f"{bundle_dir} is missing bundle.json or manifest.jsonl", + envelope, records, problem = _load_bundle(bundle_dir) + if problem: + return Result("T05", "Dry-run validation", "BLOCKED", problem, {"ran": False}) - envelope = json.loads(envelope_path.read_text(encoding="utf-8")) - records = [json.loads(line) for line in - manifest_path.read_text(encoding="utf-8").splitlines() - if line.strip()] - - status, body, duration = client.request( - "POST", "/api/node/audio/import", headers=headers, - body={"envelope": envelope, "records": records, - "dry_run": True, "confirm": None}) + status, body, duration = _dry_run_once(client, bundle_dir, headers) summary = _summarise(body) evidence: dict[str, Any] = { @@ -461,9 +489,22 @@ def t05_dry_run(client: Client, if status in (200, 201): evidence["reported_status"] = body.get("status") if isinstance(body, dict) else None evidence["content_hash"] = body.get("content_hash") if isinstance(body, dict) else None + evidence["counts"] = body.get("counts") if isinstance(body, dict) else None + evidence["warning_count"] = (len(body.get("warnings") or []) + if isinstance(body, dict) else None) + evidence["response_digest"] = _digest(body) + declared = (envelope or {}).get("content_hash", {}) + declared_value = declared.get("value") if isinstance(declared, dict) else None + evidence["declared_content_hash"] = declared_value + evidence["hash_round_trips"] = ( + evidence["content_hash"] is not None + and evidence["content_hash"] == declared_value + ) return Result("T05", "Dry-run validation", "PASS", f"controller accepted the bundle in dry-run mode ({status}); " - f"reported status={evidence['reported_status']}", evidence) + f"reported status={evidence['reported_status']}, " + f"content_hash round-trips={evidence['hash_round_trips']}", + evidence) detail = (f"dry run rejected ({status}); " + (str(summary)[:300] if summary else "empty body")) @@ -475,33 +516,144 @@ def t05_dry_run(client: Client, return Result("T05", "Dry-run validation", "FAIL", detail, evidence) -def t06_idempotency(client: Client) -> Result: - return Result("T06", "Idempotency check", "BLOCKED", - "needs T05 to produce a content_hash to compare against, and " - "T05 does not get one while the bundle is rejected. The " - "controller also reports its accepted manifest format only " - "after it receives a bundle, so this is sequenced, not broken.", - {"depends_on": "T05", - "client_side_equivalent": "the bundle build is deterministic: " - "the same project yields the same manifest.jsonl bytes and the " - "same content_hash, so re-shipping is detectable without a server"}) +def t06_idempotency(client: Client, + bundle_dir: str | None, + headers: dict[str, str], + t05: Result | None) -> Result: + """Two dry runs must agree; re-import behaviour still needs a real commit. + + Historically this returned a narrative written for the era when T05 was + rejected ("T05 does not get one while the bundle is rejected"), so it stayed + BLOCKED with a reason that had stopped being true. It now runs the dry run a + second time and compares responses. + + What this proves: the controller's validation is deterministic, so the same + bundle always produces the same plan and the same content_hash. That is one + half of idempotency. + + What it cannot prove: that re-importing the same hash is a no-op. That + assertion lives in ``import_bundle``'s ``last_content_hash`` check and only + means anything once rows exist, which needs a store that persists. See T08. + """ + + if not bundle_dir: + return Result("T06", "Idempotency check", "BLOCKED", + "no --bundle given, so there is nothing to re-run", + {"depends_on": "T05", "ran": False}) + if t05 is None or t05.status != "PASS": + return Result("T06", "Idempotency check", "BLOCKED", + "T05 did not pass, so there is no first response to " + "compare a second one against", + {"depends_on": "T05", "ran": False}) + + first_digest = t05.evidence.get("response_digest") + status, body, duration = _dry_run_once(client, bundle_dir, headers) + second_digest = _digest(body) + + evidence: dict[str, Any] = { + "ran": True, + "first_digest": first_digest, + "second_digest": second_digest, + "second_status_code": status, + "second_duration_ms": duration, + "second_content_hash": (body.get("content_hash") + if isinstance(body, dict) else None), + "wrote": False, + } + + if status not in (200, 201): + return Result("T06", "Idempotency check", "FAIL", + f"the second dry run was rejected ({status}) while the " + f"first passed, so validation is not repeatable", + evidence) + if first_digest != second_digest: + return Result("T06", "Idempotency check", "FAIL", + "two identical dry runs returned different bodies, so the " + "plan is not deterministic", + evidence) + + evidence["untested"] = ("re-import of an already-stored content_hash " + "returning no_change") + return Result("T06", "Idempotency check", "PASS", + "two consecutive dry runs returned byte-identical responses, " + "so the plan and content_hash are deterministic. Re-import " + "suppressing already-stored rows is NOT covered here: that " + "path needs rows that persist, and the import currently " + "builds an in-memory store", + evidence) -def t07_schema_contract(client: Client) -> Result: - return Result("T07", "Schema contract", "BLOCKED", - "the controller has not yet issued a manifest format to accept; " - "its request says the mapping and the accepted format come after " - "the bundle. A contract is proposed in the report, not assumed.", - {"depends_on": "T04", - "bundle_validates_locally": True}) +def t07_schema_contract(t05: Result | None, + bundle_records: int | None) -> Result: + """Did the controller's counts reconcile with the bytes we actually sent? + + This used to say "the controller has not yet issued a manifest format to + accept" — true before the first accepted dry run, stale afterwards. An + accepted response IS the format confirmation: the controller parsed all + 8,884 records and reported counts for them. + """ + + if t05 is None or t05.status != "PASS": + return Result("T07", "Schema contract", "BLOCKED", + "the controller has not yet accepted a bundle, so no " + "format has been confirmed by it", + {"depends_on": "T04", "bundle_validates_locally": True}) + + counts = t05.evidence.get("counts") or {} + accounted = sum(counts.get(k, 0) + for k in ("chapters", "speakers", "utterances", "artifacts")) + evidence: dict[str, Any] = { + "ran": True, + "controller_counts": counts, + "records_sent": bundle_records, + "records_accounted": accounted, + "note": ("the book record is the one kind the controller does not " + "report as a count, so records_sent - records_accounted should " + "be exactly 1"), + } + if bundle_records is None: + evidence["delta"] = None + return Result("T07", "Schema contract", "PASS", + "the controller accepted and parsed the manifest format", + evidence) + + delta = bundle_records - accounted + evidence["delta"] = delta + if delta != 1: + return Result("T07", "Schema contract", "FAIL", + f"the controller accounted for {accounted} of " + f"{bundle_records} records sent (delta {delta}, expected " + f"1 for the uncounted book record) — some record kind is " + f"being dropped silently", + evidence) + return Result("T07", "Schema contract", "PASS", + f"the controller parsed all {bundle_records} records: its " + f"counts account for {accounted}, plus 1 book record it does " + f"not count separately", + evidence) def t08_audit_round_trip(client: Client) -> Result: + """Read back what was written, from the controller's side. + + Its old text said "no ingest event can be created without an ingest + endpoint". The endpoint has existed since T04 went green, so that reason was + stale. The real blocker is that nothing is written: the import builds + ``InMemoryAudioImportStore``, whose ``commit`` appends to a list and mints + ``uuid4()`` ids, so there would be no row to read back even after a + successful commit. + """ + return Result("T08", "Bidirectional audit", "BLOCKED", - "no ingest event can be created without an ingest endpoint; " - "the controller does expose an audit table but only to " - "authenticated principals", - {"depends_on": "T04"}) + "needs a committed import to read back from. The endpoint " + "routes and validates (T04, T05), but routes_compute builds " + "InMemoryAudioImportStore, whose commit() writes to a list and " + "not to Postgres — so a commit would return a uuid4 project_id " + "while translation_utterances stayed empty, and there would be " + "nothing to audit. Blocked on a persistence adapter, not on " + "the endpoint", + {"depends_on": "T04", + "needs": "a Postgres AudioImportStore adapter"}) def t09_error_handling(client: Client) -> Result: @@ -803,14 +955,15 @@ def main() -> int: bundle_records = sum(1 for line in manifest.read_text().splitlines() if line.strip()) bundle_bytes = manifest.stat().st_size + t05_result = t05_dry_run(client, args.bundle, headers) results = [ t01_health(client), t02_auth(client, headers, how), t03_version(client), t04_ingest_exists(client, headers), - t05_dry_run(client, args.bundle, headers), - t06_idempotency(client), - t07_schema_contract(client), + t05_result, + t06_idempotency(client, args.bundle, headers, t05_result), + t07_schema_contract(t05_result, bundle_records), t08_audit_round_trip(client), t09_error_handling(client), t10_payload_readiness(bundle_records, bundle_bytes), @@ -829,7 +982,6 @@ def main() -> int: transport = "plain HTTP (uvicorn); no TLS on this port" \ if client.base_url.startswith("http") else "HTTPS" - t05_result = next((r for r in results if r.test_id == "T05"), None) verdict, rationale = _readiness(routed_ingest, missing_ingest, t05_result) report: dict[str, Any] = { diff --git a/tests/test_audio_controller_connect.py b/tests/test_audio_controller_connect.py index b8e2061..28ff8d4 100644 --- a/tests/test_audio_controller_connect.py +++ b/tests/test_audio_controller_connect.py @@ -516,3 +516,161 @@ def test_dry_run_passes_only_on_a_2xx(tmp_path: Path) -> None: assert passed.evidence["content_hash"] == "a" * 64 assert failed.status == "FAIL" assert failed.evidence["status_code"] == 500 + + +# --------------------------------------------------------------------------- +# What T06/T07/T08 claim, once a dry run can actually pass +# --------------------------------------------------------------------------- +# +# These tests exist because all three carried narratives written for the era +# when the bundle was rejected, and each kept asserting a reason that had +# stopped being true: T06 said "T05 does not get [a content_hash] while the +# bundle is rejected", T07 said the controller "has not yet issued a manifest +# format", T08 said there was no ingest endpoint. Every one of those was once +# accurate, and every one survived its own refutation. A BLOCKED test is a +# finding, so a stale BLOCKED reason is a finding about the wrong world. + + +def _write_bundle(tmp_path: Path, records: int = 1) -> None: + (tmp_path / "bundle.json").write_text( + json.dumps({"bundle_format_version": 1, + "content_hash": {"value": "0" * 64}}), encoding="utf-8") + (tmp_path / "manifest.jsonl").write_text( + "".join(json.dumps({"id": f"b-{i}", "kind": "book"}) + "\n" + for i in range(records)), encoding="utf-8") + + +def _passing_t05(digest: str = "d" * 64, + counts: dict[str, int] | None = None) -> Any: + return tool.Result( + "T05", "Dry-run validation", "PASS", "accepted", + {"status_code": 200, "reported_status": "dry_run", + "content_hash": "0" * 64, "response_digest": digest, + "counts": counts if counts is not None + else {"chapters": 1, "speakers": 0, "utterances": 0, "artifacts": 0}}) + + +def test_t06_does_not_run_a_canned_reason_when_t05_failed(tmp_path: Path) -> None: + """The old bug: T06 narrated why it could not run, whatever T05 said.""" + + _write_bundle(tmp_path) + rejected = tool.Result("T05", "Dry-run validation", "FAIL", "500", {}) + + result = tool.t06_idempotency(FakeClient({}), str(tmp_path), {}, rejected) + + assert result.status == "BLOCKED" + assert result.evidence.get("ran") is False + + +def test_t06_blocked_without_a_bundle_or_a_first_response() -> None: + assert tool.t06_idempotency(FakeClient({}), None, {}, None).evidence["ran"] is False + assert tool.t06_idempotency( + FakeClient({}), None, {}, _passing_t05()).evidence["ran"] is False + + +def test_t06_passes_when_two_dry_runs_agree(tmp_path: Path) -> None: + _write_bundle(tmp_path) + body = {"status": "dry_run", "content_hash": "0" * 64, + "counts": {"chapters": 1}} + first = tool.t05_dry_run( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): body}), str(tmp_path), {}) + + second = tool.t06_idempotency( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): body}), + str(tmp_path), {}, first) + + assert first.status == "PASS" + assert second.status == "PASS" + assert second.evidence["first_digest"] == second.evidence["second_digest"] + + +def test_t06_fails_when_two_identical_dry_runs_disagree(tmp_path: Path) -> None: + """Non-determinism means the plan is not a function of the bundle.""" + + _write_bundle(tmp_path) + first = _passing_t05(digest="d" * 64) + + drifted = tool.t06_idempotency( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): {"status": "dry_run", + "content_hash": "1" * 64}}), + str(tmp_path), {}, first) + + assert drifted.status == "FAIL" + + +def test_t06_names_what_it_cannot_prove(tmp_path: Path) -> None: + """Passing on determinism must not read as proof of re-import safety.""" + + _write_bundle(tmp_path) + body = {"status": "dry_run", "content_hash": "0" * 64} + first = tool.t05_dry_run( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): body}), str(tmp_path), {}) + + result = tool.t06_idempotency( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): body}), + str(tmp_path), {}, first) + + assert result.status == "PASS" + assert "no_change" not in result.detail.split(".")[0] + assert "NOT covered" in result.detail + assert "in-memory store" in result.detail + + +def test_t07_reconciles_counts_and_allows_for_the_uncounted_book_record() -> None: + """32+55+4319+4477 = 8883, and the 8884th record is the book.""" + + result = tool.t07_schema_contract( + _passing_t05(counts={"chapters": 32, "speakers": 55, + "utterances": 4319, "artifacts": 4477}), + 8884) + + assert result.status == "PASS" + assert result.evidence["delta"] == 1 + + +def test_t07_fails_when_the_controller_drops_records(tmp_path: Path) -> None: + """A controller that silently ignores a record kind would otherwise pass.""" + + result = tool.t07_schema_contract( + _passing_t05(counts={"chapters": 1, "speakers": 0, + "utterances": 0, "artifacts": 0}), + 5) + + assert result.status == "FAIL" + assert result.evidence["delta"] == 4 + + +def test_t07_blocked_until_a_bundle_is_accepted() -> None: + result = tool.t07_schema_contract( + tool.Result("T05", "Dry-run validation", "FAIL", "500", {}), 1) + + assert result.status == "BLOCKED" + + +def test_t08_no_longer_blames_an_endpoint_that_routs() -> None: + """The specific stale claim T08 used to make.""" + + result = tool.t08_audit_round_trip(FakeClient({})) + + assert result.status == "BLOCKED" + assert "without an ingest endpoint" not in result.detail + assert "depends on T04" not in result.detail + + +def test_t08_names_the_actual_blocker_so_a_fix_has_to_update_it() -> None: + """Wiring a persistence adapter must force this text to change. + + If someone implements a Postgres AudioImportStore, this test fails and + makes them re-decide T08 rather than leave a blocked verdict describing a + store that no longer exists. + """ + + result = tool.t08_audit_round_trip(FakeClient({})) + + assert "InMemoryAudioImportStore" in result.detail + assert "uuid4" in result.detail