#!/usr/bin/env python3 """runner/exit-status mocked-transport tests — NOT live Ollama model-quality. Label: runner/exit-status mocked-transport (FAKE curl on PATH). Covers the 7 audit cases plus truncation (done_reason length/max_tokens). """ from __future__ import annotations import json import os import subprocess import sys import tempfile from pathlib import Path RUNNER = Path(__file__).resolve().parent / "runner.py" KIT = Path(__file__).resolve().parent.parent CASES = { "normal": { "body": json.dumps({"model": "fake", "response": "hello from model", "done": True, "done_reason": "stop"}), "http": "200", "rc": 0, }, "model_missing": { "body": json.dumps({"error": "model 'nope' not found"}), "http": "404", "rc": 0, }, "server_error": { "body": json.dumps({"error": "internal server error"}), "http": "500", "rc": 0, }, "done_false": { "body": json.dumps({"model": "fake", "response": "partial text still here", "done": False}), "http": "200", "rc": 0, }, "empty": { "body": json.dumps({"model": "fake", "response": "", "done": True, "done_reason": "stop"}), "http": "200", "rc": 0, }, "invalid_json": { "body": "NOT-JSON{{{", "http": "200", "rc": 0, }, "transport_fail": { "body": "", "http": "", "rc": 7, }, "truncated": { "body": json.dumps({"model": "fake", "response": "cut off mid-", "done": True, "done_reason": "length"}), "http": "200", "rc": 0, }, } def install_fake_curl(fake_dir: Path, scenario: str) -> None: cfg = CASES[scenario] body_file = fake_dir / "body.bin" body_file.write_bytes(cfg["body"].encode("utf-8")) http = cfg["http"] rc = cfg["rc"] # Fake curl: consume stdin, emit body + newline + http_code (curl --write-out style) script = f"""#!/bin/sh cat >/dev/null if [ "{rc}" -ne 0 ]; then echo "fake curl transport failure" >&2 exit {rc} fi cat "{body_file}" printf '\\n{http}' exit 0 """ curl_path = fake_dir / "curl" curl_path.write_text(script) curl_path.chmod(0o755) def run_ollama_with_fake(scenario: str, tmp: Path) -> tuple[int, dict]: fake_dir = tmp / f"fake-curl-{scenario}" fake_dir.mkdir(parents=True, exist_ok=True) install_fake_curl(fake_dir, scenario) out = tmp / f"out-{scenario}.md" env = os.environ.copy() env["PATH"] = str(fake_dir) + os.pathsep + env.get("PATH", "") proc = subprocess.run( [sys.executable, str(RUNNER), "ollama", "--model", "fake-model", "--prompt", "test prompt", "--out", str(out)], capture_output=True, text=True, env=env, cwd=str(KIT), ) payload = {} if proc.stdout.strip(): try: payload = json.loads(proc.stdout.strip().splitlines()[-1]) except json.JSONDecodeError: payload = {"_raw_stdout": proc.stdout[:500]} return proc.returncode, payload def test_transport_cases() -> list[tuple[str, bool, str]]: results = [] expectations = { "normal": (0, True), "model_missing": (2, False), "server_error": (2, False), "done_false": (2, False), "empty": (2, False), "invalid_json": (2, False), "transport_fail": (2, False), "truncated": (2, False), } with tempfile.TemporaryDirectory(prefix="rd-transport-") as td: tmp = Path(td) for name, (exp_rc, exp_ok) in expectations.items(): rc, payload = run_ollama_with_fake(name, tmp) ok = payload.get("ok") meta = payload.get("meta") or {} passed = (rc == exp_rc) and (ok is exp_ok) if name == "normal": passed = passed and meta.get("status") == "MODEL_OUTPUT_RECEIVED_UNREVIEWED" passed = passed and meta.get("review_status") == "UNREVIEWED" else: passed = passed and ok is False passed = passed and meta.get("review_status", "UNREVIEWED") == "UNREVIEWED" passed = passed and meta.get("status") not in ( "COMPLETE", "PASS", "MODEL_OUTPUT_RECEIVED_UNREVIEWED", ) if name == "model_missing": passed = passed and meta.get("status") == "BLOCKED_MODEL_UNAVAILABLE" if name == "truncated": passed = passed and meta.get("error") == "generation_truncated" if name == "done_false": passed = passed and meta.get("error") == "generation_not_done" detail = f"rc={rc} ok={ok} status={meta.get('status')} error={meta.get('error')}" results.append((name, passed, detail)) return results def test_build_review_inclusion() -> tuple[bool, str]: with tempfile.TemporaryDirectory(prefix="rd-buildrev-") as td: tmp = Path(td) out = tmp / "ai2-prompt.txt" proc = subprocess.run( [sys.executable, str(RUNNER), "build-review", "--sources-file", str(KIT / "input" / "synthetic-sources.json"), "--draft-file", str(KIT / "sample-output" / "ai1-structure.REDACTED.md"), "--restrictions-file", str(KIT / "input" / "restrictions.txt"), "--out", str(out)], capture_output=True, text=True, cwd=str(KIT), ) if proc.returncode != 0: return False, f"build-review exit {proc.returncode}: {proc.stderr[:300]}" text = out.read_text(encoding="utf-8") checks = { "sources_marker": "=== ORIGINAL LABELED SOURCES ===" in text, "draft_marker": "=== AI-1 DRAFT (SECONDARY EVIDENCE) ===" in text, "restrictions_marker": "=== CARRIED-FORWARD RESTRICTIONS ===" in text, "has_S1": '"id": "S1"' in text, "has_S2": '"id": "S2"' in text, "has_restriction_untested": "Monitoring remains Untested" in text, "has_restriction_data": "data-analysis is NOT cleared" in text, "has_waiting": "WAITING_ON_HUMAN" in text, "has_draft_content": len((KIT / "sample-output" / "ai1-structure.REDACTED.md").read_text()) > 0 and ( KIT / "sample-output" / "ai1-structure.REDACTED.md" ).read_text() in text, } missing = [k for k, v in checks.items() if not v] return (not missing), f"missing={missing} bytes={len(text)} stdout={proc.stdout.strip()[:200]}" def main(): print("TEST_TYPE: runner/exit-status mocked-transport (NOT live Ollama model-quality)") transport = test_transport_cases() all_ok = True for name, passed, detail in transport: mark = "PASS" if passed else "FAIL" if not passed: all_ok = False print(f" [{mark}] {name}: {detail}") br_ok, br_detail = test_build_review_inclusion() if not br_ok: all_ok = False print(f" [{'PASS' if br_ok else 'FAIL'}] build-review_inclusion: {br_detail}") n_pass = sum(1 for _, p, _ in transport if p) print( f"SUMMARY: {'ALL PASS' if all_ok else 'FAILURES'} — " f"{n_pass}/{len(transport)} transport + build-review={'PASS' if br_ok else 'FAIL'}" ) raise SystemExit(0 if all_ok else 1) if __name__ == "__main__": main()