Open raw ↗#!/usr/bin/env python3
"""Programmatic verification test suite for HAD Digital MVP.
Validates the acceptance criteria defined in ORIGINAL_REQUEST.md:
1. Server launch and zero-dependency HTTP response
2. Static frontend assets serving (HTML, CSS, JS)
3. Patient authentication and session cookie issuance
4. Patient toxicity report submission (CTCAE auto-grading, SQLite persistence)
5. Clinician authentication and session management
6. Clinician care timeline viewing containing the submitted report
7. Clean server shutdown
Usage:
python verify_mvp.py [--exe PATH_TO_EXE] [--source] [--port PORT]
"""
import argparse
import http.client
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path
current_dir = Path(__file__).resolve().parent
if current_dir.name in ("05_Test", "04_Build"):
PROJECT_ROOT = current_dir.parent
elif current_dir.name == "explorer_survey_2":
PROJECT_ROOT = current_dir.parent.parent
else:
PROJECT_ROOT = current_dir
MVP_DIR = PROJECT_ROOT / "MVP"
class TestRunner:
def __init__(self, exe_path: str | None = None, use_source: bool = False, port: int = 8099):
self.exe_path = exe_path
self.use_source = use_source
self.port = port
self.host = "127.0.0.1"
self.server_proc = None
self.temp_dir = None
self.test_db_path = None
self.results = []
def log(self, tag: str, msg: str):
print(f"[{tag}] {msg}")
def record_result(self, step: str, passed: bool, details: str = ""):
status_str = "PASS" if passed else "FAIL"
print(f" [{status_str}] {step}: {details}")
self.results.append({"step": step, "passed": passed, "details": details})
def request(self, method: str, path: str, body: dict | None = None, cookie: str | None = None) -> tuple[int, dict, dict | str]:
conn = http.client.HTTPConnection(self.host, self.port, timeout=10)
headers = {"Content-Type": "application/json"}
if cookie:
headers["Cookie"] = cookie
payload = json.dumps(body) if body is not None else None
try:
conn.request(method, path, body=payload, headers=headers)
resp = conn.getresponse()
raw_body = resp.read().decode("utf-8")
resp_headers = dict(resp.getheaders())
try:
parsed_json = json.loads(raw_body)
except json.JSONDecodeError:
parsed_json = raw_body
return resp.status, resp_headers, parsed_json
finally:
conn.close()
def setup_environment(self):
self.temp_dir = tempfile.mkdtemp(prefix="had_test_")
self.test_db_path = os.path.join(self.temp_dir, "had_test.db")
self.log("SETUP", f"Isolated test database: {self.test_db_path}")
def start_server(self):
self.log("START", f"Launching server on {self.host}:{self.port}...")
env = os.environ.copy()
env["HAD_HOST"] = self.host
env["HAD_PORT"] = str(self.port)
env["HAD_DB_PATH"] = str(self.test_db_path)
env["HAD_DEBUG"] = "false"
if self.exe_path:
cmd = [str(self.exe_path), "--host", self.host, "--port", str(self.port)]
self.log("START", f"Executing standalone binary: {self.exe_path}")
self.server_proc = subprocess.Popen(
cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
else:
cmd = [sys.executable, str(MVP_DIR / "app.py"), "--host", self.host, "--port", str(self.port)]
self.log("START", f"Executing Python source: {MVP_DIR / 'app.py'}")
self.server_proc = subprocess.Popen(
cmd, env=env, cwd=str(MVP_DIR), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
# Wait for server readiness
max_attempts = 30
for i in range(max_attempts):
time.sleep(0.5)
# Check if process died
if self.server_proc.poll() is not None:
_, err = self.server_proc.communicate()
raise RuntimeError(f"Server process terminated prematurely with exit code {self.server_proc.returncode}: {err}")
try:
status, _, data = self.request("GET", "/api/whoami")
if status in (200, 401):
self.log("READY", f"Server responded successfully after {(i+1)*0.5:.1f}s (Status: {status})")
return
except (ConnectionRefusedError, OSError):
continue
raise TimeoutError(f"Server failed to start within {max_attempts * 0.5} seconds.")
def stop_server(self):
if self.server_proc:
self.log("CLEANUP", "Terminating server process...")
self.server_proc.terminate()
try:
self.server_proc.wait(timeout=5)
except subprocess.TimeoutExpired:
self.server_proc.kill()
self.server_proc.wait()
self.server_proc = None
if self.temp_dir and os.path.exists(self.temp_dir):
shutil.rmtree(self.temp_dir, ignore_errors=True)
def run_tests(self):
print("\n================== EXECUTING ACCEPTANCE TESTS ==================\n")
patient_cookie = None
clinician_cookie = None
submitted_report_id = None
# Test 1: Static HTML frontend serving
try:
status, _, body = self.request("GET", "/")
passed = (status == 200 and "<html" in str(body).lower() and "HAD Digital" in str(body))
self.record_result("R3.1 Local Frontend Index Serving", passed, f"Status: {status}")
except Exception as e:
self.record_result("R3.1 Local Frontend Index Serving", False, str(e))
# Test 2: Static CSS and JS assets
try:
status_css, _, body_css = self.request("GET", "/static/app.css")
status_js, _, body_js = self.request("GET", "/static/app.js")
passed = (status_css == 200 and status_js == 200 and len(body_css) > 1000 and len(body_js) > 1000)
self.record_result("R3.2 Static CSS/JS Bundle Assets", passed, f"CSS: {status_css}, JS: {status_js}")
except Exception as e:
self.record_result("R3.2 Static CSS/JS Bundle Assets", False, str(e))
# Test 3: Unauthenticated /whoami
try:
status, _, body = self.request("GET", "/api/whoami")
passed = (status in (200, 401) and isinstance(body, dict) and body.get("authenticated") is False)
self.record_result("R2.1 Unauthenticated State Check", passed, f"Status: {status}, Response: {body}")
except Exception as e:
self.record_result("R2.1 Unauthenticated State Check", False, str(e))
# Test 4: Patient Authentication
try:
status, headers, body = self.request("POST", "/api/login", {"username": "patient.durand", "password": "demo123"})
passed = (status == 200 and "user" in body and body["user"].get("role") == "patient")
raw_cookie = headers.get("set-cookie", headers.get("Set-Cookie", ""))
if raw_cookie:
patient_cookie = raw_cookie.split(";")[0]
self.record_result("R2.2 Patient Authentication", passed, f"Role: {body.get('user', {}).get('role')}, Cookie: {bool(patient_cookie)}")
except Exception as e:
self.record_result("R2.2 Patient Authentication", False, str(e))
# Test 5: Patient Toxicity Report Submission
try:
# We submit both multi-symptom payload and required fields
report_payload = {
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": 5, # Grade 2
"symptoms": {"nausea": 2, "fatigue": 3},
"notes": "Moderate nausea and fatigue post cycle 3",
}
status, _, body = self.request("POST", "/api/reports", report_payload, cookie=patient_cookie)
passed = (status in (200, 201) and "report_id" in body)
if passed:
submitted_report_id = body["report_id"]
grade = body.get("grading", {}).get("grade")
self.record_result("R2.3 Patient Toxicity Report Submission", True, f"Report ID: {submitted_report_id}, Grade: {grade}")
else:
self.record_result("R2.3 Patient Toxicity Report Submission", False, f"Status {status}: {body}")
except Exception as e:
self.record_result("R2.3 Patient Toxicity Report Submission", False, str(e))
# Test 6: Clinician Authentication
try:
status, headers, body = self.request("POST", "/api/login", {"username": "dr.martin", "password": "demo123"})
passed = (status == 200 and "user" in body and body["user"].get("role") == "oncologist")
raw_cookie = headers.get("set-cookie", headers.get("Set-Cookie", ""))
if raw_cookie:
clinician_cookie = raw_cookie.split(";")[0]
self.record_result("R2.4 Clinician Authentication", passed, f"Role: {body.get('user', {}).get('role')}")
except Exception as e:
self.record_result("R2.4 Clinician Authentication", False, str(e))
# Test 7: Clinician Care Timeline Viewing
try:
status, _, body = self.request("GET", "/api/timeline?patient_id=1", cookie=clinician_cookie)
events = body.get("events", []) if isinstance(body, dict) else []
has_report_event = any("Toxicity report" in e.get("title", "") or "nausea" in str(e).lower() for e in events)
passed = (status == 200 and has_report_event)
self.record_result("R2.5 Clinician Timeline Report Visibility", passed, f"Total Events: {len(events)}, Report Found: {has_report_event}")
except Exception as e:
self.record_result("R2.5 Clinician Timeline Report Visibility", False, str(e))
# Test 8: SQLite Database Persistence Check
try:
import sqlite3
conn = sqlite3.connect(self.test_db_path)
c = conn.cursor()
c.execute("SELECT COUNT(*) FROM toxicity_reports")
report_count = c.fetchone()[0]
c.execute("SELECT COUNT(*) FROM timeline_events")
timeline_count = c.fetchone()[0]
conn.close()
passed = (report_count > 0 and timeline_count > 0)
self.record_result("R1.1 SQLite Direct Persistence Verification", passed, f"Reports in DB: {report_count}, Timeline Events: {timeline_count}")
except Exception as e:
self.record_result("R1.1 SQLite Direct Persistence Verification", False, str(e))
print("\n================== VERIFICATION SUMMARY ==================\n")
total = len(self.results)
passed_count = sum(1 for r in self.results if r["passed"])
print(f"Total Tests: {total} | Passed: {passed_count} | Failed: {total - passed_count}")
for r in self.results:
mark = "[OK]" if r["passed"] else "[FAIL]"
print(f" {mark} {r['step']}")
all_passed = (total > 0 and passed_count == total)
if all_passed:
print("\n[SUCCESS] All acceptance criteria verified successfully!\n")
else:
print("\n[FAILURE] One or more acceptance criteria failed.\n")
return all_passed
def main():
parser = argparse.ArgumentParser(description="Programmatic verification test for HAD Digital MVP")
parser.add_argument("--exe", type=str, default=None, help="Path to compiled standalone .exe")
parser.add_argument("--source", action="store_true", help="Test against Python source code instead of .exe")
parser.add_argument("--port", type=int, default=8099, help="Port to run test server on (default 8099)")
args = parser.parse_args()
if not args.exe and not args.source:
# Default to dist/HAD_Digital.exe if it exists, else source
dist_exe = PROJECT_ROOT / "04_Build" / "dist" / "HAD_Digital.exe"
if not dist_exe.exists():
dist_exe = MVP_DIR / "dist" / "HAD Digital" / "HAD Digital.exe"
if dist_exe.exists():
args.exe = str(dist_exe)
else:
args.source = True
runner = TestRunner(exe_path=args.exe, use_source=args.source, port=args.port)
try:
runner.setup_environment()
runner.start_server()
success = runner.run_tests()
sys.exit(0 if success else 1)
finally:
runner.stop_server()
if __name__ == "__main__":
main()