sttil/Signal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
Signal/scripts/signal_e2e_test.py
Kisa 42a79c6682 add pitch assets, fix UI and backend bugs, add test utilities 
- add 3 signal pitch assets (workflow flowchart, market opportunity, win-win-win) as HTML + PNG
- fix doc_state_machine cascade and supplied sentinel flags
- fix CSVImport mapping review, WorklistTable colored bullets and expand row
- fix App.jsx auth gate, api.js CORS handling
- add signal E2E test script and stress test generator

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-19 12:52:29 -04:00

407 lines
18 KiB
Python
Raw Blame History

#!/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 from env var SIGNAL_API_KEY or hardcoded fallback.
"""
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", "4HKamqzn78Md-L-xj2OjZESVwZh2Otzc6iFRyqVZl7U")
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)
Reference in a new issue View git blame Copy permalink