9dfa7915ee
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.
1048 lines
48 KiB
Python
1048 lines
48 KiB
Python
#!/usr/bin/env python3
|
|
"""Test this audio system's readiness to talk to the UmbraTrans controller.
|
|
|
|
Writes two artefacts and prints a summary:
|
|
|
|
* ``audio_controller_test_report.json`` -- the machine-readable result
|
|
* ``audio_controller_sync.log`` -- one JSON object per line, per step
|
|
|
|
Every request is logged with its endpoint, status and duration, and **no secret
|
|
is ever logged** -- not in the report, not in the log, not in an error message.
|
|
Header values are reduced to a presence flag before they reach a log line.
|
|
|
|
What this deliberately does not do: invent an endpoint. The brief anticipated
|
|
``/api/audio/ingest`` and friends; if none exists, that is recorded as
|
|
``MISSING_INGEST_ENDPOINT`` with a proposed contract, and the tests that depend
|
|
on it report BLOCKED rather than PASS or FAIL. A test that cannot run has not
|
|
passed.
|
|
|
|
python3 scripts/audio_controller_connect.py --report-dir /tmp/connect
|
|
|
|
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
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
#: The controller as found by discovery. Plain HTTP: the port answers uvicorn
|
|
#: with no TLS, so the ``https://`` in the original brief cannot connect and
|
|
#: ``UMBRA_AUDIO_CONTROLLER_VERIFY`` has nothing to verify. Recorded, not worked
|
|
#: around -- see ``transport`` in the report.
|
|
DEFAULT_BASE_URL = os.environ.get("UMBRA_AUDIO_CONTROLLER_URL", "http://192.168.123.14:8442")
|
|
|
|
INGEST_CANDIDATES = (
|
|
# Confirmed from the controller's own route table: agent_router, mounted at
|
|
# /api/node, exposes POST /audio/import. This path was MISSING from this list
|
|
# originally, and the report said MISSING_INGEST_ENDPOINT on the strength of
|
|
# guessed paths while the route table was one SSH command away. Guessing is
|
|
# not discovery; see t04_ingest_exists.
|
|
"/api/node/audio/import",
|
|
"/api/audio/ingest",
|
|
"/api/projects/import/audio",
|
|
"/api/import/audio",
|
|
"/api/audio/dataset",
|
|
"/api/projects/audio-dataset",
|
|
"/api/manifest/ingest",
|
|
"/api/compute/audio/import",
|
|
)
|
|
|
|
PROBE_PATHS = (
|
|
"/api/health", "/healthz", "/health", "/openapi.json", "/api/openapi.json",
|
|
"/docs", "/api/setup/status", "/api/projects", "/api/artifacts", "/api/workspace",
|
|
"/api/nodes", "/api/jobs", "/api/audit", "/api/me", "/api/auth/me",
|
|
"/api/definitions", "/api/ingest", "/api/datasets",
|
|
)
|
|
|
|
|
|
def now_iso() -> str:
|
|
stamp = datetime.now(UTC).replace(microsecond=0)
|
|
return stamp.isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
class SyncLog:
|
|
"""One JSON object per line. A log you cannot parse is not a log."""
|
|
|
|
def __init__(self, path: Path) -> None:
|
|
self.path = path
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.handle = path.open("w", encoding="utf-8")
|
|
self.entries = 0
|
|
|
|
def write(self, phase: str, action: str, **fields: Any) -> None:
|
|
entry = {
|
|
"timestamp": now_iso(),
|
|
"phase": phase,
|
|
"action": action,
|
|
**{k: v for k, v in fields.items()},
|
|
}
|
|
self.handle.write(json.dumps(entry, sort_keys=True, ensure_ascii=False) + "\n")
|
|
self.handle.flush()
|
|
self.entries += 1
|
|
|
|
def close(self) -> None:
|
|
self.handle.close()
|
|
|
|
|
|
@dataclass
|
|
class Result:
|
|
test_id: str
|
|
name: str
|
|
status: str # PASS | FAIL | BLOCKED | INFO
|
|
detail: str
|
|
evidence: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {"id": self.test_id, "name": self.name, "status": self.status,
|
|
"detail": self.detail, "evidence": self.evidence}
|
|
|
|
|
|
class Client:
|
|
def __init__(self, base_url: str, log: SyncLog) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.log = log
|
|
# A self-signed certificate would need an override; recorded if set, and
|
|
# never silently applied. Plain HTTP needs no context at all.
|
|
self.verify = os.environ.get("UMBRA_AUDIO_CONTROLLER_VERIFY", "true").lower() != "false"
|
|
self.context: ssl.SSLContext | None = None
|
|
if self.base_url.startswith("https") and not self.verify:
|
|
self.context = ssl.create_default_context()
|
|
self.context.check_hostname = False
|
|
self.context.verify_mode = ssl.CERT_NONE
|
|
|
|
def request(self, method: str, path: str, *, body: Any = None,
|
|
headers: dict[str, str] | None = None,
|
|
phase: str = "probe") -> tuple[int, Any, int]:
|
|
"""Returns (status, parsed_body_or_text, duration_ms). Never raises."""
|
|
|
|
url = self.base_url + path
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
request = urllib.request.Request(url, data=data, method=method)
|
|
for key, value in (headers or {}).items():
|
|
request.add_header(key, value)
|
|
if data is not None:
|
|
request.add_header("Content-Type", "application/json")
|
|
|
|
started = time.monotonic()
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=15,
|
|
context=self.context) as response:
|
|
status, raw = response.status, response.read()
|
|
except urllib.error.HTTPError as error:
|
|
status, raw = error.code, error.read()
|
|
except Exception as error: # noqa: BLE001 - connectivity is a result
|
|
duration = int((time.monotonic() - started) * 1000)
|
|
self.log.write(phase, method, endpoint=path,
|
|
status=None, duration_ms=duration,
|
|
result_hash=None,
|
|
error={"code": type(error).__name__,
|
|
"message": str(error)[:200],
|
|
"retryable": True,
|
|
"suggested_fix": "check the controller is running and reachable"})
|
|
failure = {"error": {"code": type(error).__name__,
|
|
"message": str(error)[:200]}}
|
|
return 0, failure, duration
|
|
|
|
duration = int((time.monotonic() - started) * 1000)
|
|
try:
|
|
parsed: Any = json.loads(raw)
|
|
except ValueError:
|
|
parsed = raw[:400].decode("utf-8", "replace")
|
|
# Header values are never logged; only whether one was present.
|
|
self.log.write(phase, method, endpoint=path, status=status,
|
|
duration_ms=duration,
|
|
credentials_present=bool(headers),
|
|
request_summary=_summarise(body),
|
|
result_hash=_hash(parsed))
|
|
return status, parsed, duration
|
|
|
|
def authenticate(self, log: SyncLog) -> tuple[dict[str, str], str]:
|
|
"""Node key, then service token. Returns (headers, how).
|
|
|
|
The node credential goes out as ``Authorization: Bearer <key>`` because
|
|
that is what ``require_node`` parses: it rejects anything that does not
|
|
start with ``bearer `` (routes_compute.py:477) and never looks at an
|
|
``X-Node-Key`` header. This used to send ``X-Node-Key``, so a perfectly
|
|
valid key was reported as a rejected credential.
|
|
"""
|
|
|
|
node_key = os.environ.get("UMBRA_AUDIO_NODE_KEY")
|
|
token = os.environ.get("UMBRA_AUDIO_API_TOKEN")
|
|
if node_key:
|
|
log.write("auth", "try_node_key", endpoint="GET /api/node/agent",
|
|
note="Authorization: Bearer (node credential)")
|
|
return {"Authorization": f"Bearer {node_key}"}, "node_bearer"
|
|
if token:
|
|
log.write("auth", "try_service_token", endpoint="/api/auth/me",
|
|
note="Authorization: Bearer")
|
|
return {"Authorization": f"Bearer {token}"}, "service_token_bearer"
|
|
log.write("auth", "no_credentials",
|
|
note="no node key or service token configured on this machine")
|
|
return {}, "none"
|
|
|
|
|
|
def _summarise(body: Any) -> Any:
|
|
"""A description of a request, small enough to log and free of secrets."""
|
|
|
|
if body is None:
|
|
return None
|
|
if isinstance(body, str):
|
|
# A plain-text body is usually the thing you need to read — "Internal
|
|
# Server Error", or a handler's own message. Reporting it as <str> threw
|
|
# away exactly the detail the caller was looking for.
|
|
return body[:400]
|
|
if isinstance(body, dict):
|
|
return {k: (f"<{len(v)} items>" if isinstance(v, (list, dict)) else v)
|
|
for k, v in body.items()
|
|
if k not in {"token", "api_key", "password", "secret"}}
|
|
return f"<{type(body).__name__}>"
|
|
|
|
|
|
def _hash(value: Any) -> str | None:
|
|
import hashlib
|
|
try:
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True,
|
|
default=str).encode()).hexdigest()[:16]
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# tests
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def t01_health(client: Client) -> Result:
|
|
status, body, _ = client.request("GET", "/api/health")
|
|
ok = status == 200 and isinstance(body, dict) and body.get("status") == "ok"
|
|
return Result("T01", "Health", "PASS" if ok else "FAIL",
|
|
"controller reports status=ok" if ok else f"health returned {status}",
|
|
{"status_code": status, "body": body if ok else str(body)[:200]})
|
|
|
|
|
|
def t02_auth(client: Client, headers: dict[str, str], how: str) -> Result:
|
|
"""Prove the credential works on an endpoint that accepts its kind.
|
|
|
|
`/api/auth/me` resolves a browser *session*, so a node credential gets 401
|
|
there no matter how valid it is. Testing a node key against it reported a
|
|
working credential as rejected. A node credential authenticates on
|
|
`/api/node/*`, which is also CSRF-exempt, so a GET there is a clean probe.
|
|
"""
|
|
|
|
if not headers:
|
|
return Result("T02", "Auth", "BLOCKED",
|
|
"no credential is configured on this machine; "
|
|
"an operator must supply a node key or a scoped service token",
|
|
{"headers_tried": [], "how": how})
|
|
|
|
node_bearer = how == "node_bearer"
|
|
endpoint = "/api/node/agent" if node_bearer else "/api/auth/me"
|
|
status, body, _ = client.request("GET", endpoint, headers=headers)
|
|
|
|
if node_bearer:
|
|
# require_node() answers 401 for a bad/absent key, and the scope check
|
|
# answers 403 only *after* the credential resolved. So 403 here means
|
|
# authenticated with insufficient scope, which is still authentication.
|
|
ok = status in (200, 403)
|
|
detail = (
|
|
f"node credential accepted on {endpoint} ({status})" if ok else
|
|
f"node credential rejected on {endpoint} ({status})"
|
|
)
|
|
else:
|
|
ok = status == 200
|
|
detail = (
|
|
f"authenticated with {how} on {endpoint}" if ok else
|
|
f"credential rejected ({status})"
|
|
)
|
|
|
|
return Result("T02", "Auth", "PASS" if ok else "FAIL", detail,
|
|
{"status_code": status, "mechanism": how,
|
|
"endpoint": endpoint, "secret_logged": False,
|
|
"scope_note": ("403 on a node route is the scope check after "
|
|
"a successful credential resolve, not a "
|
|
"rejection" if node_bearer else None)})
|
|
|
|
|
|
def t03_version(client: Client) -> Result:
|
|
status, body, _ = client.request("GET", "/api/health")
|
|
controller_version = (body or {}).get("version") if isinstance(body, dict) else None
|
|
here = "1.0.0 (bundle format)"
|
|
compatible = bool(controller_version)
|
|
return Result("T03", "Version compatibility", "PASS" if compatible else "FAIL",
|
|
f"controller {controller_version}; this side speaks "
|
|
f"bundle format {here}. No versioned API exists to negotiate "
|
|
f"against, so there is nothing to be incompatible with yet.",
|
|
{"controller_version": controller_version,
|
|
"controller_service": ((body or {}).get("service")
|
|
if isinstance(body, dict) else None),
|
|
"client_bundle_format": 1,
|
|
"hard_incompatibility": False})
|
|
|
|
|
|
def t04_ingest_exists(client: Client,
|
|
headers: dict[str, str] | None = None) -> Result:
|
|
"""Probe every candidate with both GET and POST.
|
|
|
|
Probing only with GET is how a POST-only ingest route looked absent for a
|
|
whole session: the real path is `POST /api/node/audio/import`, which answers
|
|
GET with 404 and POST with 405, so a GET-only sweep concluded it was not
|
|
there. A 404 to the wrong method is not evidence of absence.
|
|
|
|
The rule that survived two wrong attempts, and the reason it is narrower
|
|
than "anything but 404":
|
|
|
|
* On a SAFE method (GET), 404 means not routed and anything else means the
|
|
route table matched. This holds everywhere.
|
|
* On an UNSAFE method (POST), the controller checks CSRF in **middleware
|
|
before routing** (app.py:246-248) and exempts only
|
|
``CSRF_EXEMPT_PREFIXES = ("/api/node",)``. So a POST to any other /api
|
|
path comes back 403 csrf_token_missing whether or not a route exists —
|
|
that 403 is the security layer talking and is not evidence of anything.
|
|
My earlier probe read exactly such a 403 on
|
|
``/api/compute/audio/import`` as a live route; there is no such route.
|
|
|
|
A 401 on a /api/node path, by contrast, is strong: that prefix is CSRF-
|
|
exempt, so the request routed, then ``require_node`` refused it.
|
|
"""
|
|
|
|
# Mirror of the controller's CSRF_EXEMPT_PREFIXES (app.py:89). If the
|
|
# controller changes this, this list is what makes a POST probe trustworthy.
|
|
# Probe WITH the credential when one exists. Probing unauthenticated made a
|
|
# working route look permanently gated: every run reported 401 and the
|
|
# verdict stayed AUTH_REQUIRED even after auth was configured and T02
|
|
# passed. An unauthenticated probe answers "is there a door", never "does it
|
|
# open".
|
|
auth_headers = headers or {}
|
|
|
|
csrf_exempt_prefixes = ("/api/node",)
|
|
|
|
probed: dict[str, dict[str, Any]] = {}
|
|
routed: tuple[str, int] | None = None
|
|
csrf_masked: str | None = None
|
|
for path in INGEST_CANDIDATES:
|
|
results: dict[str, Any] = {}
|
|
for method in ("GET", "POST"):
|
|
status, _, _ = client.request(
|
|
method, path, body={} if method == "POST" else None,
|
|
headers=auth_headers)
|
|
results[method] = status
|
|
if status is None:
|
|
continue
|
|
unsafe = method not in ("GET", "HEAD", "OPTIONS")
|
|
if unsafe and status == 403 and not path.startswith(csrf_exempt_prefixes):
|
|
# Middleware answered before routing. Not evidence either way.
|
|
results[f"{method}_note"] = (
|
|
"403 csrf_token_missing from middleware that runs before "
|
|
"routing; proves nothing about whether a route exists")
|
|
csrf_masked = csrf_masked or f"{method} {path}"
|
|
continue
|
|
if status != 404 and routed is None:
|
|
routed = (f"{method} {path}", status)
|
|
probed[path] = results
|
|
|
|
if routed is not None:
|
|
which, status = routed
|
|
serving = status in (200, 201)
|
|
gated = status in (401, 403)
|
|
if serving:
|
|
why = "responding normally."
|
|
elif gated:
|
|
why = ("the route resolved and then refused the request, which is "
|
|
"the credential doing its job, not the path going missing.")
|
|
elif status == 422:
|
|
why = ("the request body failed schema validation — routing and "
|
|
"request parsing both work; this probe sends an empty object "
|
|
"on purpose, so 422 is the expected answer and is not a "
|
|
"fault.")
|
|
elif status == 405:
|
|
why = "the path exists but is registered for another method."
|
|
elif status >= 500:
|
|
why = ("the route resolved and then failed inside its own body — "
|
|
"the path exists and the handler is reachable.")
|
|
else:
|
|
why = "the path exists and rejected the request."
|
|
detail = f"{which} is routed ({status}). {why}"
|
|
evidence: dict[str, Any] = {
|
|
"probed": probed, "routed": which, "status": status,
|
|
"reachable": serving, "auth_gated": gated,
|
|
"schema_rejected": status == 422,
|
|
"handler_error": status >= 500,
|
|
"note": ("Routed is not the same as reachable, and reachable is not "
|
|
"the same as working. A 401/403 is a live route behind a "
|
|
"credential, a 422 is a live route parsing a body, and a "
|
|
"5xx is a live route failing on this payload. None of them "
|
|
"is an absent endpoint."),
|
|
}
|
|
if csrf_masked:
|
|
evidence["csrf_masked_probe"] = (
|
|
f"{csrf_masked} returned 403 from middleware that runs before "
|
|
"routing, so it was not counted as a route")
|
|
return Result("T04", "Ingest endpoint exists", "PASS", detail, evidence)
|
|
|
|
csrf_note = (
|
|
{"csrf_masked_probe": (
|
|
f"{csrf_masked} returned 403 from middleware that runs before "
|
|
"routing, so it was not counted as a route")}
|
|
if csrf_masked else {}
|
|
)
|
|
return Result("T04", "Ingest endpoint exists", "FAIL",
|
|
"no ingest path responded to GET or POST. NOTE: this is probed "
|
|
"paths only, not the controller's route table. Before treating "
|
|
"this as absence, enumerate the routes the controller actually "
|
|
"registers - include_router(...) in its app module - because a "
|
|
"guessed path that 404s proves only that the guess was wrong. "
|
|
"That mistake was made here once.",
|
|
{"probed": probed, "finding": "UNCONFIRMED_INGEST_ENDPOINT",
|
|
"how_to_confirm": "read the controller's router registrations, "
|
|
"not its 404s",
|
|
**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:
|
|
"""Actually run a dry run, when there is a bundle to run it against.
|
|
|
|
This used to return BLOCKED unconditionally with "depends on an ingest
|
|
endpoint; see T04" — which stayed true after the endpoint came up and after
|
|
a credential arrived, so the most important test in the file was the one
|
|
least willing to run. A dry run cannot write: ``import_bundle`` returns at
|
|
``if dry_run`` before it ever reaches the store, so executing it is safe by
|
|
construction and this test now behaves like it believes that.
|
|
"""
|
|
|
|
if not bundle_dir:
|
|
return Result("T05", "Dry-run validation", "BLOCKED",
|
|
"no --bundle given, so there is nothing to validate "
|
|
"against; pass --bundle to run a real dry run",
|
|
{"depends_on": None, "ran": False})
|
|
|
|
envelope, records, problem = _load_bundle(bundle_dir)
|
|
if problem:
|
|
return Result("T05", "Dry-run validation", "BLOCKED", problem,
|
|
{"ran": False})
|
|
|
|
status, body, duration = _dry_run_once(client, bundle_dir, headers)
|
|
|
|
summary = _summarise(body)
|
|
evidence: dict[str, Any] = {
|
|
"status_code": status,
|
|
"duration_ms": duration,
|
|
"records_sent": len(records),
|
|
"dry_run": True,
|
|
"confirm": None,
|
|
"wrote": False,
|
|
"body": summary,
|
|
"note": ("dry_run short-circuits before store.commit(), so this "
|
|
"request cannot write. Nothing was pushed."),
|
|
}
|
|
|
|
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']}, "
|
|
f"content_hash round-trips={evidence['hash_round_trips']}",
|
|
evidence)
|
|
|
|
detail = (f"dry run rejected ({status}); "
|
|
+ (str(summary)[:300] if summary else "empty body"))
|
|
if status >= 500:
|
|
detail += (
|
|
" — a 5xx on a valid-looking payload is the handler failing inside "
|
|
"its own validation rather than reporting a client error"
|
|
)
|
|
return Result("T05", "Dry-run validation", "FAIL", detail, evidence)
|
|
|
|
|
|
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(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",
|
|
"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:
|
|
"""The brief asked for a structured error with an actionable code.
|
|
|
|
The interesting part is that the controller does have its own envelope:
|
|
``{"error": {"code", "message"}}``. Probing with an unsupported method is
|
|
enough to see it, and it is worth seeing -- a service-to-service caller
|
|
needs to key off ``error.code`` rather than parsing prose. The probe also
|
|
revealed that **CSRF is enforced**, which is a real constraint on any future
|
|
browser-authenticated path and is recorded rather than left as a surprise.
|
|
"""
|
|
|
|
status, body, _ = client.request("DELETE", "/api/health")
|
|
raw = body.get("error") if isinstance(body, dict) else None
|
|
envelope: dict[str, Any] = raw if isinstance(raw, dict) else {}
|
|
structured = bool(envelope.get("code")) and bool(envelope.get("message"))
|
|
|
|
if structured:
|
|
detail = (f"error envelope carries code={envelope['code']!r} and an "
|
|
f"actionable message; probe returned {status}")
|
|
else:
|
|
detail = f"no structured error envelope; probe returned {status}: {str(body)[:160]}"
|
|
|
|
return Result("T09", "Error handling", "PASS" if structured else "FAIL",
|
|
detail,
|
|
{"status_code": status, "body": body,
|
|
"error_code": ((envelope or {}).get("code")
|
|
if isinstance(envelope, dict) else None),
|
|
"csrf_required": status == 403
|
|
and isinstance(envelope, dict)
|
|
and envelope.get("code") == "csrf_token_missing",
|
|
"note": ("CSRF is enforced on this origin. A machine-to-machine "
|
|
"client should use a bearer/node-key path rather than "
|
|
"cookie auth, or the CSRF requirement becomes a "
|
|
"per-request concern.")})
|
|
|
|
|
|
def t10_payload_readiness(bundle_records: int | None, bundle_bytes: int | None) -> Result:
|
|
if bundle_records is None:
|
|
return Result("T10", "Large payload readiness", "BLOCKED",
|
|
"no bundle was built in this run", {})
|
|
return Result("T10", "Large payload readiness", "PASS",
|
|
f"the dataset is {bundle_records} records / "
|
|
f"{bundle_bytes} bytes, which is small enough for a single "
|
|
f"request; no chunking or streaming strategy is needed yet",
|
|
{"records": bundle_records, "bytes": bundle_bytes,
|
|
"max_records_per_request": bundle_records,
|
|
"chunking_required": False,
|
|
"strategy": "single request, with content_hash for idempotency"})
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# proposed contract
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def _readiness(
|
|
routed_ingest: Result | None,
|
|
missing_ingest: bool,
|
|
t05: Result | None = None,
|
|
) -> tuple[str, list[str]]:
|
|
"""Name the controller's actual ingest state, not a nearby one.
|
|
|
|
This ran through three names and every one of them was wrong about the
|
|
thing it described, in the same direction: each described the day it was
|
|
written rather than the state it was reading.
|
|
|
|
* absent - guessed paths all 404'd, reported as MISSING. Wrong: the route
|
|
existed and was probed with GET only.
|
|
* needs restart - the route answered 405, reported as needing a restart.
|
|
Right at the time, then kept being reported after the restart happened.
|
|
* and a 401 was read as a dead end when it is a live route doing its job.
|
|
|
|
The states are separated by evidence the run actually holds: whether a path
|
|
resolved at all, whether the credential was the reason it refused, and
|
|
whether the handler came back.
|
|
|
|
A real dry run outranks every transport observation. Once T05 has actually
|
|
run a bundle, "is the route up" is settled and the thing worth naming is
|
|
what the controller said about the data — which is where the remaining
|
|
problem lives.
|
|
"""
|
|
|
|
base = [
|
|
"The controller's own request is for a metadata bundle, validated by its "
|
|
"own validator, handed over as a file. That is done.",
|
|
"Its request also says the accepted manifest format comes after it "
|
|
"receives the bundle, so T05-T08 are sequenced, not broken.",
|
|
]
|
|
|
|
if t05 is not None and t05.status == "FAIL":
|
|
return ("DRY_RUN_REJECTED", [
|
|
f"A real dry run was executed against the full bundle and the "
|
|
f"controller rejected it: {t05.detail}",
|
|
"The endpoint is live, authenticated and reachable — this is now a "
|
|
"question about the data contract, not about connectivity.",
|
|
"dry_run was true and confirm was null, so nothing could have been "
|
|
"written even had it succeeded. Nothing was pushed.",
|
|
"The controller has been told the failing message verbatim, along "
|
|
"with the exact line that produces a different content_hash for the "
|
|
"same bundle.",
|
|
] + base)
|
|
|
|
if t05 is not None and t05.status == "PASS":
|
|
return ("DRY_RUN_OK_CONFIRMATION_REQUIRED", [
|
|
f"The controller accepted the bundle in dry-run mode: {t05.detail}",
|
|
"A dry run is not an import. Committing needs dry_run false and "
|
|
"confirm set to the content_hash the controller returned, plus an "
|
|
"explicit operator instruction.",
|
|
"Nothing has been pushed.",
|
|
] + base)
|
|
|
|
if routed_ingest is None:
|
|
return ("INGEST_PATH_NOT_CONFIRMED", [
|
|
"No ingest path responded to GET or POST, so a dataset cannot be "
|
|
"pushed; nothing was pushed. This is UNCONFIRMED: confirm against "
|
|
"the controller's route table before treating it as absent. A 404 "
|
|
"to a guessed path is evidence about the guess, not about the "
|
|
"controller.",
|
|
] + base)
|
|
|
|
evidence = routed_ingest.evidence
|
|
where = evidence["routed"]
|
|
status = evidence["status"]
|
|
|
|
if evidence.get("auth_gated"):
|
|
return ("INGEST_ROUTED_AUTH_REQUIRED", [
|
|
f"The ingest route {where} is registered and resolved the request "
|
|
f"({status}); it then refused for want of a credential, which is the "
|
|
"auth check inside the route running as designed.",
|
|
"A 401/403 here is the strongest short of a 200 that the route is "
|
|
"live: routing happens before the dependency, so the path matched.",
|
|
"Nothing was pushed. T02/T05-T08 stay BLOCKED until a credential "
|
|
"with dataset:ingest is configured for this run.",
|
|
] + base)
|
|
|
|
if evidence.get("handler_error"):
|
|
return ("INGEST_ROUTED_HANDLER_ERROR", [
|
|
f"The ingest route {where} is registered, authenticated and "
|
|
f"reachable ({status}): the request routed, passed the credential "
|
|
"check, and failed inside the handler.",
|
|
"T04 probes with an empty JSON body on purpose — enough to prove "
|
|
"the path resolves and the handler runs, and nothing that could "
|
|
"write. A 5xx from that body means the handler is reachable and "
|
|
"rejecting an empty payload, which is expected; the verdict names "
|
|
"the reachable-and-failing state rather than calling the endpoint "
|
|
"absent.",
|
|
"Nothing was pushed. The empty probe body is not a bundle, so this "
|
|
"says nothing about whether a real one would import — supply "
|
|
"--bundle and T05 answers that directly.",
|
|
] + base)
|
|
|
|
if evidence.get("schema_rejected"):
|
|
return ("INGEST_ROUTED_AWAITING_BUNDLE", [
|
|
f"The ingest route {where} is registered, authenticated and parsing "
|
|
f"request bodies ({status} against an empty probe object).",
|
|
"422 here is the request schema doing its job on a body that is "
|
|
"deliberately incomplete. The endpoint is ready for a real bundle; "
|
|
"no bundle was sent in this run because none was supplied.",
|
|
"Nothing was pushed.",
|
|
] + base)
|
|
|
|
if not evidence.get("reachable"):
|
|
return ("INGEST_ROUTED_NOT_SERVING", [
|
|
f"The ingest route {where} is registered but answered {status}. "
|
|
"Nothing was pushed.",
|
|
] + base)
|
|
|
|
return ("READY_FOR_PUSH", [
|
|
f"The ingest route {where} answered {status} normally.",
|
|
"A dry run still has to succeed before any commit, and a commit still "
|
|
"requires explicit confirmation.",
|
|
] + base)
|
|
|
|
|
|
def proposed_contract() -> dict[str, Any]:
|
|
return {
|
|
"note": ("A proposal only. No endpoint was invented, probed into "
|
|
"existence, or assumed to exist; this is what the audio side "
|
|
"would send if the controller chooses to add it."),
|
|
"openapi_paths": {
|
|
"/api/audio/ingest": {
|
|
"post": {
|
|
"summary": "Accept an audio-system metadata dataset",
|
|
"requestBody": {
|
|
"required": True,
|
|
"content": {"application/json": {
|
|
"schema": {"$ref": "#/components/schemas/AudioDatasetEnvelope"},
|
|
}},
|
|
},
|
|
"responses": {
|
|
"200": {"description": "Dry run; validation report, nothing written"},
|
|
"201": {"description": "Accepted; import_id and audit_id returned"},
|
|
"400": {"description": "Structured validation error", "content":
|
|
{"application/json": {"schema": {
|
|
"$ref": "#/components/schemas/Error"}}}},
|
|
"409": {"description": ("Idempotency key already seen "
|
|
"with a different hash")},
|
|
},
|
|
}
|
|
}
|
|
},
|
|
"components": {
|
|
"schemas": {
|
|
"AudioDatasetEnvelope": {
|
|
"type": "object",
|
|
"required": ["bundle_format_version", "source", "content_hash", "dry_run"],
|
|
"properties": {
|
|
"bundle_format_version": {"type": "integer", "const": 1},
|
|
"source": {
|
|
"type": "object",
|
|
"required": ["system", "node_id"],
|
|
"properties": {
|
|
"system": {"type": "string"},
|
|
"node_id": {"type": "string"},
|
|
"product_version": {"type": "string"},
|
|
"database_schema_fingerprint": {"type": "string"},
|
|
},
|
|
},
|
|
"content_hash": {
|
|
"type": "string",
|
|
"description": ("sha256 of the manifest bytes. The server "
|
|
"uses it as the idempotency key so a "
|
|
"re-send of unchanged data is a no-op."),
|
|
},
|
|
"dry_run": {"type": "boolean", "default": True},
|
|
"records": {"type": "array", "items": {"type": "object"}},
|
|
},
|
|
},
|
|
"ValidationReport": {
|
|
"type": "object",
|
|
"required": ["valid", "counts", "errors", "warnings"],
|
|
"properties": {
|
|
"valid": {"type": "boolean"},
|
|
"counts": {"type": "object"},
|
|
"errors": {"type": "array", "items": {"type": "object"}},
|
|
"warnings": {"type": "array", "items": {"type": "object"}},
|
|
"proposed_changes": {"type": "array", "items": {"type": "object"}},
|
|
},
|
|
},
|
|
"Error": {
|
|
"type": "object",
|
|
"required": ["code", "message"],
|
|
"properties": {"code": {"type": "string"},
|
|
"message": {"type": "string"},
|
|
"retryable": {"type": "boolean"}},
|
|
},
|
|
}
|
|
},
|
|
"invariants_the_controller_should_enforce": [
|
|
"content_hash is the idempotency key; a repeat with the same hash is a no-op.",
|
|
"A dry run must not write. It returns counts and errors, and nothing else.",
|
|
"Records are keyed by (source.system, source.node_id, id) so a re-import "
|
|
"updates rather than duplicates.",
|
|
"Nothing is deleted on re-import; a record absent from a later dataset is "
|
|
"marked superseded, not removed.",
|
|
"Chapter identity uses the database row id, never chapter_index or a "
|
|
"filename-derived number.",
|
|
"Speaker identity is not merged by the server. Case-variant and alias names "
|
|
"are surfaced as warnings for a human to decide.",
|
|
],
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument("--report-dir", default=".", help="where to write the two artefacts")
|
|
parser.add_argument("--bundle", default=None,
|
|
help="a built bundle directory, to report its size for T10")
|
|
parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
|
|
args = parser.parse_args()
|
|
|
|
out = Path(args.report_dir)
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
log = SyncLog(out / "audio_controller_sync.log")
|
|
log.write("start", "run", base_url=args.base_url, note="no secret is logged")
|
|
|
|
client = Client(args.base_url, log)
|
|
|
|
# --- discovery ---------------------------------------------------
|
|
discovery: dict[str, Any] = {"base_url": args.base_url, "probes": {}}
|
|
for path in PROBE_PATHS:
|
|
status, body, _ = client.request("GET", path, phase="discovery")
|
|
discovery["probes"][path] = {
|
|
"status": status,
|
|
"body": body if status == 200 else str(body)[:160],
|
|
}
|
|
log.write("discovery", "complete", endpoints_found=sum(
|
|
1 for p in discovery["probes"].values() if p["status"] in (200, 401, 403)))
|
|
|
|
headers, how = client.authenticate(log)
|
|
|
|
bundle_records = bundle_bytes = None
|
|
if args.bundle:
|
|
manifest = Path(args.bundle) / "manifest.jsonl"
|
|
if manifest.exists():
|
|
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_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),
|
|
]
|
|
for result in results:
|
|
log.write("test", result.test_id, name=result.name, status=result.status)
|
|
|
|
passing = [r for r in results if r.status == "PASS"]
|
|
failed = [r for r in results if r.status == "FAIL"]
|
|
blocked = [r for r in results if r.status == "BLOCKED"]
|
|
|
|
missing_ingest = any(r.evidence.get("finding") == "UNCONFIRMED_INGEST_ENDPOINT"
|
|
for r in results)
|
|
routed_ingest = next((r for r in results if r.test_id == "T04"
|
|
and r.evidence.get("routed")), None)
|
|
transport = "plain HTTP (uvicorn); no TLS on this port" \
|
|
if client.base_url.startswith("http") else "HTTPS"
|
|
|
|
verdict, rationale = _readiness(routed_ingest, missing_ingest, t05_result)
|
|
|
|
report: dict[str, Any] = {
|
|
"report_version": 1,
|
|
"generated_at": now_iso(),
|
|
"source_system": {"system": "umbraaudio", "node_id": "umbraaudio-m4"},
|
|
"transport": {
|
|
"base_url": args.base_url,
|
|
"tls": transport,
|
|
"tls_verify_override": os.environ.get("UMBRA_AUDIO_CONTROLLER_VERIFY"),
|
|
"note": ("The brief specified https:// on this port. The port answers "
|
|
"plain HTTP, so the TLS handshake fails with a protocol-version "
|
|
"alert and there is nothing for a verify override to do. "
|
|
"Reported rather than worked around."),
|
|
},
|
|
"authentication": {
|
|
"mechanism_available": how,
|
|
"node_key_configured": bool(os.environ.get("UMBRA_AUDIO_NODE_KEY")),
|
|
"service_token_configured": bool(os.environ.get("UMBRA_AUDIO_API_TOKEN")),
|
|
"secrets_recorded_anywhere": False,
|
|
},
|
|
"discovery": discovery,
|
|
"tests": [r.as_dict() for r in results],
|
|
"summary": {
|
|
"counts": {"pass": len(passing), "fail": len(failed),
|
|
"blocked": len(blocked)},
|
|
"passing": [r.test_id for r in passing],
|
|
"failing": [r.test_id for r in failed],
|
|
"blocked_ids": [r.test_id for r in blocked],
|
|
"blocking_findings": ["UNCONFIRMED_INGEST_ENDPOINT"] if missing_ingest else [],
|
|
},
|
|
"readiness": {
|
|
"verdict": verdict,
|
|
"rationale": rationale,
|
|
},
|
|
"proposed_ingest_contract": proposed_contract() if missing_ingest else None,
|
|
"confirmation_gate": {
|
|
"required_before_any_push": True,
|
|
"current_state": "NOT CONFIRMED - nothing has been pushed",
|
|
"instruction": ("A non-dry-run import requires explicit operator "
|
|
"confirmation. This run made no write of any kind."),
|
|
},
|
|
}
|
|
|
|
(out / "audio_controller_test_report.json").write_text(
|
|
json.dumps(report, indent=2, sort_keys=True) + "\n")
|
|
log.write("end", "run", passed=len(passing), failed=len(failed), blocked=len(blocked))
|
|
log.close()
|
|
|
|
print(f"{'ID':4} {'STATUS':8} {'NAME'}")
|
|
print("-" * 52)
|
|
for result in results:
|
|
print(f"{result.test_id:4} {result.status:8} {result.name}")
|
|
print("-" * 52)
|
|
print(f"pass {len(passing)} fail {len(failed)} blocked {len(blocked)}")
|
|
print(f"report: {out / 'audio_controller_test_report.json'}")
|
|
print(f"log: {out / 'audio_controller_sync.log'} ({log.entries} entries)")
|
|
print(f"verdict: {report['readiness']['verdict']}")
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|