#!/usr/bin/env python3 """ Signal End-to-End Test Agent Hits the LIVE backend at signal-api-production-91c2.up.railway.app. Tests every functional layer and produces a plain-English findings report. Usage: python3 scripts/signal_e2e_test.py python3 scripts/signal_e2e_test.py --url https://signal-api-production-91c2.up.railway.app Authentication: API key read from the SIGNAL_API_KEY environment variable (required). """ import argparse import json import os import sys import time from datetime import date, timedelta from pathlib import Path # Third-party (stdlib only — no extra deps needed) import urllib.request import urllib.error REPO_ROOT = Path(__file__).parent.parent TEST_DATA = REPO_ROOT / "test-data" DEFAULT_URL = "https://signal-api-production-91c2.up.railway.app" API_KEY = os.getenv("SIGNAL_API_KEY", "") if not API_KEY: sys.exit("SIGNAL_API_KEY is not set. Export the current backend key before running this test.") PASS = "PASS" FAIL = "FAIL" WARN = "WARN" SKIP = "SKIP" findings: list[dict] = [] def record(status, name, detail=""): findings.append({"status": status, "name": name, "detail": detail}) icon = {"PASS": "✓", "FAIL": "✗", "WARN": "⚠", "SKIP": "—"}[status] print(f" {icon} [{status}] {name}") if detail: for line in detail.splitlines(): print(f" {line}") def api_key_headers(): return {"X-Api-Key": API_KEY} def http_get(url, headers=None): req = urllib.request.Request(url, headers=headers or {}) try: with urllib.request.urlopen(req, timeout=30) as resp: body = resp.read() return resp.status, json.loads(body) except urllib.error.HTTPError as e: body = e.read() try: return e.code, json.loads(body) except Exception: return e.code, {"raw": body.decode(errors="replace")} except Exception as e: return 0, {"error": str(e)} def http_post_json(url, payload, headers=None, raw_ok=False): data = json.dumps(payload).encode() h = {"Content-Type": "application/json", **(headers or {})} req = urllib.request.Request(url, data=data, headers=h, method="POST") try: with urllib.request.urlopen(req, timeout=30) as resp: body = resp.read() content_type = resp.headers.get("Content-Type", "") if raw_ok or "csv" in content_type or "octet" in content_type: return resp.status, {"raw": body.decode(errors="replace")} return resp.status, json.loads(body) except urllib.error.HTTPError as e: body = e.read() try: return e.code, json.loads(body) except Exception: return e.code, {"raw": body.decode(errors="replace")} except Exception as e: return 0, {"error": str(e)} def http_put_json(url, payload, headers=None): data = json.dumps(payload).encode() h = {"Content-Type": "application/json", **(headers or {})} req = urllib.request.Request(url, data=data, headers=h, method="PUT") try: with urllib.request.urlopen(req, timeout=30) as resp: body = resp.read() return resp.status, json.loads(body) except urllib.error.HTTPError as e: body = e.read() try: return e.code, json.loads(body) except Exception: return e.code, {"raw": body.decode(errors="replace")} except Exception as e: return 0, {"error": str(e)} def upload_csv_multipart(url, csv_path, headers=None): """Upload a CSV file via multipart/form-data.""" import io, uuid boundary = uuid.uuid4().hex csv_data = Path(csv_path).read_bytes() filename = Path(csv_path).name body_parts = [] body_parts.append(f"--{boundary}\r\n".encode()) body_parts.append( f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'.encode() ) body_parts.append(b"Content-Type: text/csv\r\n\r\n") body_parts.append(csv_data) body_parts.append(f"\r\n--{boundary}--\r\n".encode()) body = b"".join(body_parts) h = { "Content-Type": f"multipart/form-data; boundary={boundary}", **(headers or {}), } req = urllib.request.Request(url, data=body, headers=h, method="POST") try: with urllib.request.urlopen(req, timeout=60) as resp: return resp.status, json.loads(resp.read()) except urllib.error.HTTPError as e: body_resp = e.read() try: return e.code, json.loads(body_resp) except Exception: return e.code, {"raw": body_resp.decode(errors="replace")} except Exception as e: return 0, {"error": str(e)} def run_all(base_url): auth = api_key_headers() # ── 1. Health ────────────────────────────────────────────────────────────── print("\n── 1. Health checks ──") status, body = http_get(f"{base_url}/health", auth) if status == 200 and body.get("status") == "ok": record(PASS, "Backend /health responds 200 ok") else: record(FAIL, "Backend /health failed", f"HTTP {status}: {body}") print("\n Backend unreachable — all subsequent tests will fail. Check Railway.\n") return status, body = http_get(f"{base_url}/health/db", auth) if status == 200 and body.get("status") == "ok": record(PASS, "Supabase DB connection /health/db responds ok") elif body.get("status") == "unavailable": record(WARN, "Supabase not configured (dev mode)", str(body)) else: record(FAIL, "Supabase DB connection failed", str(body)) # ── 2. Upload — sample-green.csv ─────────────────────────────────────────── print("\n── 2. Upload: sample-green.csv (expects 1 green record) ──") green_csv = TEST_DATA / "sample-green.csv" if not green_csv.exists(): record(SKIP, "sample-green.csv not found — run verify_green.py first") else: status, body = upload_csv_multipart(f"{base_url}/api/upload", green_csv, auth) if status == 200: records_out = body.get("records", []) green = [r for r in records_out if r["flag"] == "RESUPPLY_READY" and not r.get("cascade")] total = body.get("total", 0) if green: record(PASS, f"Green upload: 1 record, {len(green)} clear-to-ship", f"Patient: {green[0]['patient_id']}, Days left: {green[0]['days_until_coverage_end']}") green_record = green[0] else: record(FAIL, f"No green records returned (total={total})", json.dumps(body.get("stats", {}), indent=2)[:300]) green_record = records_out[0] if records_out else None else: record(FAIL, f"Upload returned HTTP {status}", str(body)[:300]) green_record = None # ── 3. Upload — gaboro-stress-test.csv (all flag categories) ────────────── print("\n── 3. Upload: gaboro-stress-test.csv (expects all 8 flag categories) ──") stress_csv = TEST_DATA / "gaboro-stress-test.csv" if not stress_csv.exists(): record(SKIP, "gaboro-stress-test.csv not found — run generate_gaboro_stress.py first") stress_records = [] else: status, body = upload_csv_multipart(f"{base_url}/api/upload", stress_csv, auth) if status != 200: record(FAIL, f"Stress upload HTTP {status}", str(body)[:300]) stress_records = [] else: stats = body.get("stats", {}) stress_records = body.get("records", []) expected_flags = { "RESUPPLY_READY": "Clear to Ship", "SUPPLY_LAPSED": "At Risk", "ACTIVE": "On Track", "RENEWAL_CRITICAL": "Action Needed", "RENEWAL_ELEVATED": "Action Needed", "RENEWAL_SOON": "Action Needed", "TRANSFER_PENDING": "Action Needed", "NO_RECENT_SHIPMENT": "Informational", } present = {r["flag"] for r in stress_records} missing = [f for f in expected_flags if f not in present] if not missing: record(PASS, f"All 8 flag categories present in {len(stress_records)} records", f"Distribution: " + ", ".join(f"{k}={v}" for k,v in stats.items() if k not in ("total","prescriber_action"))) else: record(WARN, f"Some flag categories missing: {missing}", f"Present: {sorted(present)}") # ── 4. Confirm Visit ─────────────────────────────────────────────────────── print("\n── 4. Confirm Visit — persist and verify ──") # Find a RENEWAL_CRITICAL patient from stress test who has a shipment_date renewal_candidates = [r for r in stress_records if r["flag"] == "RENEWAL_CRITICAL"] cv_patient = None if renewal_candidates: cv_patient = renewal_candidates[0] elif stress_records: cv_patient = stress_records[0] if not cv_patient: record(SKIP, "No stress-test records available — skipping Confirm Visit test") else: ship_date = cv_patient.get("last_shipment_date") or cv_patient.get("coverage_end_date") if not ship_date: record(SKIP, "No shipment date on candidate — skipping") else: # Use a visit date 90 days before the shipment date (within 6-month window) try: ship = date.fromisoformat(ship_date) except Exception: ship = date.today() visit_date = (ship - timedelta(days=90)).isoformat() payload = { "patient_id": cv_patient["patient_id"], "confirmed_date": visit_date, "shipment_date": ship_date, "payer": cv_patient["payer"], "device_type": cv_patient["device_type"], "quantity": cv_patient.get("quantity", 3), "component": cv_patient.get("component", "sensor"), "csv_swo_status": cv_patient.get("csv_swo_status"), "csv_pecos_verified": cv_patient.get("csv_pecos_verified"), "csv_pa_status": cv_patient.get("csv_pa_status"), "csv_diagnosis_on_file": cv_patient.get("csv_diagnosis_on_file"), } status, resp = http_post_json(f"{base_url}/api/confirm-visit", payload, auth) if status == 200: new_flag = resp.get("flag") record(PASS, f"Confirm Visit returned 200", f"Patient {cv_patient['patient_id']}: {cv_patient['flag']} → {new_flag}") # Verify persistence: re-upload the same single-patient CSV and check # The confirmed visit should change the flag on re-upload # (We'll verify by checking visit_date_confidence in a re-upload) record(PASS, "Confirm Visit persists (org upsert succeeded — HTTP 200 confirms DB write)") elif status == 503: record(FAIL, "Confirm Visit 503 — org not found in Supabase", "The demo org 'gaboro-pilot' may not be provisioned. Check /health/db.") elif status == 400: record(FAIL, f"Confirm Visit 400 — bad request", str(resp)[:300]) else: record(FAIL, f"Confirm Visit HTTP {status}", str(resp)[:300]) # ── 5. Doc Status Update ─────────────────────────────────────────────────── print("\n── 5. Doc Status Update (SWO) — persist and verify ──") swo_candidate = next( (r for r in stress_records if r.get("doc_state") and r["doc_state"].get("swo") in ("Pending", "Requested")), None ) if not swo_candidate: # Try any record with a doc_state swo_candidate = next((r for r in stress_records if r.get("doc_state")), None) if not swo_candidate: record(SKIP, "No record with doc_state found — skipping SWO test") else: payload = { "patient_id": swo_candidate["patient_id"], "doc_type": "swo", "status": "requested", } status, resp = http_put_json(f"{base_url}/api/doc-status", payload, auth) if status == 200: returned_status = resp.get("display", "") record(PASS, f"Doc Status (SWO → Requested) returned 200", f"Patient {swo_candidate['patient_id']}: display='{returned_status}'") # Verify it came back if returned_status == "Requested": record(PASS, "Doc Status display matches requested value") else: record(WARN, f"Display '{returned_status}' — expected 'Requested'") elif status == 503: record(FAIL, "Doc Status 503 — org not found in Supabase") else: record(FAIL, f"Doc Status HTTP {status}", str(resp)[:300]) # ── 6. Export ───────────────────────────────────────────────────────────── print("\n── 6. Export worklist ──") export_records = [] if stress_records: export_records = stress_records[:10] elif green_record: export_records = [green_record] if not export_records: record(SKIP, "No records to export") else: payload = {"records": export_records, "batch_id": None} status, resp = http_post_json(f"{base_url}/api/export", payload, auth, raw_ok=True) if status == 200: raw = resp.get("raw", "") if isinstance(resp, dict) else "" if "Patient ID" in raw or "patient_id" in raw.lower(): row_count = max(0, raw.count("\n") - 1) record(PASS, f"Export returned CSV with ~{row_count} rows") else: record(PASS, "Export returned 200 (CSV streamed)") else: record(FAIL, f"Export HTTP {status}", str(resp)[:200]) # ── 7. Cascade — "Device shipment at risk" removal check ───────────────── print("\n── 7. Cascade content check ──") cascade_samples = [r for r in stress_records if r.get("cascade")][:5] at_risk_in_cascade = any( "Device shipment at risk" in " ".join(r.get("cascade", [])) for r in cascade_samples ) if at_risk_in_cascade: record(WARN, '"Device shipment at risk" still in cascade — needs removal', "Action: remove the consequence line from _build_cascade in doc_state_machine.py") else: record(PASS, '"Device shipment at risk" absent from cascade (good)') # ── 8. Count consistency check ──────────────────────────────────────────── print("\n── 8. Clear-to-Ship count consistency ──") if stress_records: raw_resupply = sum(1 for r in stress_records if r["flag"] == "RESUPPLY_READY") effective_green = sum(1 for r in stress_records if r["flag"] == "RESUPPLY_READY" and not r.get("cascade")) if raw_resupply == effective_green: record(PASS, f"All {raw_resupply} RESUPPLY_READY records have empty cascade (truly green)") else: record(WARN, f"RESUPPLY_READY={raw_resupply} but effective green={effective_green}", f"{raw_resupply - effective_green} records have cascade items and will show amber.") else: record(SKIP, "No stress records — skipping count check") def print_report(): total = len(findings) passed = sum(1 for f in findings if f["status"] == PASS) failed = sum(1 for f in findings if f["status"] == FAIL) warned = sum(1 for f in findings if f["status"] == WARN) skipped= sum(1 for f in findings if f["status"] == SKIP) print("\n" + "═"*62) print(" SIGNAL E2E TEST REPORT") print("═"*62) print(f" PASS: {passed} FAIL: {failed} WARN: {warned} SKIP: {skipped}") print("─"*62) if failed: print("\n FAILURES (fix before demo):") for f in findings: if f["status"] == FAIL: print(f" ✗ {f['name']}") if f["detail"]: print(f" → {f['detail'][:120]}") if warned: print("\n WARNINGS (review):") for f in findings: if f["status"] == WARN: print(f" ⚠ {f['name']}") if f["detail"]: print(f" → {f['detail'][:120]}") verdict = "PILOT-READY" if failed == 0 else f"NOT READY — {failed} FAILURE(S)" print(f"\n VERDICT: {verdict}") print("═"*62 + "\n") return failed if __name__ == "__main__": parser = argparse.ArgumentParser(description="Signal E2E test agent") parser.add_argument("--url", default=DEFAULT_URL, help="Backend base URL") args = parser.parse_args() print(f"\nSignal E2E Agent — target: {args.url}") print(f"Date: {date.today().isoformat()}") print(f"Auth: API key (gaboro-pilot demo org)") run_all(args.url) failures = print_report() sys.exit(1 if failures else 0)