umbragit/UmbraAudio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Probe suite: T06/T07 now test, T08 names its real blocker

Browse Source 
The dry run passes since routes_compute.py:1792 was aligned with read_bundle
(separators + ensure_ascii=False), so three tests had to catch up. All three
carried narratives written when the bundle was rejected, and each kept asserting
a reason that had stopped being true — a BLOCKED test is a finding, so a stale
BLOCKED reason is a finding about the wrong world.

T06 re-runs the dry run and compares responses, proving validation is
deterministic. It says plainly that re-import suppressing stored rows is NOT
covered: that path needs rows that persist, and the import still builds
InMemoryAudioImportStore.

T07 reconciles the controller's counts against the bytes sent. 8884 records in,
8883 accounted for, plus the book record it does not count — so a controller
that silently dropped a record kind now fails here instead of passing quietly.

T08 keeps BLOCKED, but for the reason that is actually true: no Postgres
AudioImportStore adapter exists, so even a successful commit would return a
uuid4 project_id while translation_utterances stayed empty. Its text names
InMemoryAudioImportStore on purpose, so wiring an adapter fails a test and
forces the verdict to be re-decided.

Suite: 9 PASS / 0 FAIL / 1 BLOCKED (T08). Tests 30 -> 40.
This commit is contained in:
B.D. Murphy
2026-10-16 00:27:24 -07:00
parent 0909d050d9
commit 9dfa7915ee
2 changed files with 351 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

234
scripts/audio_controller_connect.py
View File

@@ -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] = {

158
tests/test_audio_controller_connect.py
View File

@@ -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