Open raw ↗#!/usr/bin/env python3
"""HAD Digital MVP - Standalone Programmatic Verification Script.
Zero external test dependencies: uses only Python standard library modules
(http.client, json, os, shutil, socket, sqlite3, subprocess, sys, tempfile, time, urllib.parse).
Validates the complete acceptance criteria per ORIGINAL_REQUEST.md:
1. Automated launch of application (Python source or standalone .exe) on an ephemeral port
2. Startup health check and unauthenticated access rejection
3. Patient authentication (patient.durand / demo123) and session cookie issuance
4. Patient toxicity report submission with automated CTCAE grading
5. Direct SQLite persistence query verifying insertion in toxicity_reports and toxicity_grades
6. Clinician authentication (dr.martin / demo123)
7. Clinician care timeline verification confirming patient's submitted report is visible
8. Clean process termination and resource cleanup
Usage:
python verify_mvp.py --source
python verify_mvp.py --exe
python verify_mvp.py --mode source [--app-path PATH]
python verify_mvp.py --mode exe [--exe-path PATH]
"""
import argparse
import http.client
import json
import os
import shutil
import socket
import sqlite3
import subprocess
import sys
import tempfile
import time
from pathlib import Path
class VerificationRunner:
def __init__(self, mode="source", target_path=None, timeout=30):
self.mode = mode
self.target_path = target_path
self.timeout = timeout
self.port = self._find_free_port()
self.temp_dir = tempfile.mkdtemp(prefix="had_verify_")
self.temp_db = os.path.join(self.temp_dir, "verify_had.db")
self.proc = None
self.results = []
# Determine project root and paths
current_dir = Path(__file__).resolve().parent
if current_dir.name == "05_Test":
self.project_root = current_dir.parent
else:
self.project_root = current_dir
self.mvp_dir = self.project_root / "MVP"
def _find_free_port(self) -> int:
"""Allocate an ephemeral port on 127.0.0.1."""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("127.0.0.1", 0))
port = s.getsockname()[1]
s.close()
return port
def log_step(self, step_id: str, step_name: str, passed: bool, message: str = ""):
"""Log test step result and track pass/fail state."""
status = "PASS" if passed else "FAIL"
tag = f"[{status}]"
print(f" {tag:<8} Step {step_id}: {step_name}")
if message:
print(f" Details: {message}")
self.results.append({
"step_id": step_id,
"name": step_name,
"status": status,
"passed": passed,
"message": message,
})
if not passed:
raise RuntimeError(f"Step {step_id} failed: {step_name} - {message}")
def start_server(self):
"""Launch Python source or standalone binary on ephemeral port with isolated database."""
env = os.environ.copy()
env["HAD_DB_PATH"] = str(self.temp_db)
env["HAD_PORT"] = str(self.port)
env["HAD_HOST"] = "127.0.0.1"
env["HAD_DEBUG"] = "false"
if self.mode == "exe":
default_exe = self.project_root / "dist" / "HAD Digital.exe"
if not default_exe.exists():
# Fallback to MVP/dist/HAD Digital/HAD Digital.exe
alt_exe = self.mvp_dir / "dist" / "HAD Digital" / "HAD Digital.exe"
if alt_exe.exists():
default_exe = alt_exe
exe_file = Path(self.target_path) if self.target_path else default_exe
if not exe_file.exists():
raise FileNotFoundError(f"Standalone executable not found at: {exe_file}")
cmd = [str(exe_file), "--port", str(self.port)]
cwd = str(exe_file.parent)
print(f"[VERIFY] Launching standalone executable: {exe_file} on port {self.port}")
else:
default_app = self.mvp_dir / "app.py"
app_file = Path(self.target_path) if self.target_path else default_app
if not app_file.exists():
raise FileNotFoundError(f"Python application script not found at: {app_file}")
cmd = [sys.executable, str(app_file), "--port", str(self.port)]
cwd = str(self.project_root)
print(f"[VERIFY] Launching Python server: {app_file} on port {self.port}")
self.proc = subprocess.Popen(
cmd, cwd=cwd, env=env,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
# Health ping loop: wait for server to bind and respond
start_time = time.time()
online = False
while time.time() - start_time < self.timeout:
time.sleep(0.3)
# Check if process terminated prematurely
if self.proc.poll() is not None:
stdout, stderr = self.proc.communicate()
raise RuntimeError(
f"Server process terminated unexpectedly with code {self.proc.returncode}.\n"
f"Stdout: {stdout}\nStderr: {stderr}"
)
try:
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=2)
conn.request("GET", "/api/whoami")
resp = conn.getresponse()
conn.close()
# 401 (unauthenticated) or 200 (auth status) means server is online and responding
if resp.status in (200, 401):
online = True
break
except (OSError, http.client.HTTPException):
pass
if not online:
raise TimeoutError(f"Server failed to start on port {self.port} within {self.timeout}s")
elapsed = time.time() - start_time
self.log_step(
"1a", "Server Launch & Health Ping",
True, f"Process PID {self.proc.pid} responding on port {self.port} in {elapsed:.2f}s"
)
def http_request(
self, method: str, path: str, body: dict = None, cookie: str = None
) -> tuple[int, dict, str]:
"""Perform HTTP request using http.client standard library."""
conn = http.client.HTTPConnection("127.0.0.1", 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")
set_cookie = resp.getheader("Set-Cookie")
try:
data = json.loads(raw_body)
except json.JSONDecodeError:
data = {"raw": raw_body}
return resp.status, data, set_cookie
finally:
conn.close()
def run_verification(self) -> int:
"""Execute full end-to-end verification sequence."""
print("======================================================================")
print(" HAD DIGITAL MVP - ACCEPTANCE CRITERIA VERIFICATION RUNNER ")
print("======================================================================")
print(f" Mode: {self.mode.upper()}")
print(f" Ephemeral Port: {self.port}")
print(f" Isolated DB: {self.temp_db}")
print("----------------------------------------------------------------------")
try:
# 1. Launch application and verify health
self.start_server()
# 2. Verify unauthenticated access rejection
# Unauthenticated access to /api/patients or protected endpoints must be rejected
st, data, _ = self.http_request("GET", "/api/patients")
self.log_step(
"2a", "Unauthenticated Access Rejection",
st == 401, f"Protected endpoint /api/patients correctly rejected with HTTP {st}"
)
# Also verify /api/whoami reports unauthenticated
st, data, _ = self.http_request("GET", "/api/whoami")
is_unauth = (st == 401) or (st == 200 and not data.get("authenticated", False))
self.log_step(
"2b", "Unauthenticated Session Verification",
is_unauth, f"/api/whoami returned HTTP {st} with {data}"
)
# 3. Patient Authentication (patient.durand / demo123)
# Negative test: invalid credentials
st, bad_data, _ = self.http_request(
"POST", "/api/login", {"username": "patient.durand", "password": "wrongpassword"}
)
self.log_step(
"3a", "Invalid Credential Rejection",
st == 401, f"Invalid password rejected with HTTP {st}"
)
# Positive test: valid patient credentials
st, login_data, cookie = self.http_request(
"POST", "/api/login", {"username": "patient.durand", "password": "demo123"}
)
user_info = login_data.get("user", {})
patient_auth_ok = (
st == 200
and user_info.get("role") == "patient"
and cookie is not None
and "HAD_SESSION=" in cookie
)
self.log_step(
"3b", "Patient Authentication",
patient_auth_ok,
f"Logged in as '{user_info.get('username')}' ({user_info.get('role')}), session cookie received"
)
patient_cookie = cookie.split(";")[0]
# Verify whoami now returns patient user
st, whoami_data, _ = self.http_request("GET", "/api/whoami", cookie=patient_cookie)
whoami_ok = st == 200 and whoami_data.get("authenticated") is True
self.log_step(
"3c", "Patient Session Validation (/api/whoami)",
whoami_ok, f"Active session confirmed for user ID {whoami_data.get('user', {}).get('id')}"
)
# 4. Patient Toxicity Report Submission
report_payload = {
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": 2,
"notes": "Moderate nausea post-infusion day 3, managed with oral liquids",
}
st, report_data, _ = self.http_request(
"POST", "/api/reports", report_payload, cookie=patient_cookie
)
report_id = report_data.get("report_id")
grading = report_data.get("grading", {})
report_ok = (
st == 201
and report_id is not None
and grading.get("grade") is not None
)
self.log_step(
"4a", "Patient Toxicity Report Submission",
report_ok,
f"Report ID {report_id} created with CTCAE Grade {grading.get('grade')} ({grading.get('criteria')})"
)
# 5. Direct SQLite Persistence Query
conn = sqlite3.connect(self.temp_db)
cur = conn.cursor()
cur.execute(
"SELECT id, patient_id, symptom_id, severity_score, notes FROM toxicity_reports WHERE id = ?",
(report_id,)
)
rep_row = cur.fetchone()
cur.execute(
"SELECT id, report_id, grade, criteria, provisional FROM toxicity_grades WHERE report_id = ?",
(report_id,)
)
grade_row = cur.fetchone()
conn.close()
db_ok = (
rep_row is not None
and rep_row[0] == report_id
and rep_row[1] == 1
and rep_row[2] == "nausea"
and grade_row is not None
and grade_row[1] == report_id
)
self.log_step(
"5a", "Direct SQLite Persistence Verification",
db_ok,
f"DB toxicity_reports row: {rep_row} | toxicity_grades row: {grade_row}"
)
# 6. Clinician Authentication (dr.martin / demo123)
st, clin_data, clin_cookie_raw = self.http_request(
"POST", "/api/login", {"username": "dr.martin", "password": "demo123"}
)
clin_user = clin_data.get("user", {})
clin_ok = (
st == 200
and clin_user.get("role") == "oncologist"
and clin_cookie_raw is not None
)
self.log_step(
"6a", "Clinician Authentication (dr.martin)",
clin_ok,
f"Logged in as Dr. Martin (role: '{clin_user.get('role')}')"
)
clinician_cookie = clin_cookie_raw.split(";")[0]
# 7. Clinician Care Timeline Verification
st, tl_data, _ = self.http_request(
"GET", "/api/timeline?patient_id=1", cookie=clinician_cookie
)
events = tl_data.get("events", [])
matching_event = None
for event in events:
title = event.get("title", "").lower()
desc = event.get("description", "").lower()
if "nausea" in title or "nausea" in desc:
matching_event = event
break
timeline_ok = st == 200 and matching_event is not None
self.log_step(
"7a", "Clinician Care Timeline Verification",
timeline_ok,
f"Found matching event on patient timeline: '{matching_event.get('title') if matching_event else 'None'}' "
f"among {len(events)} total events"
)
# Clinician list reports endpoint verification
st, rep_list_data, _ = self.http_request(
"GET", "/api/reports?patient_id=1", cookie=clinician_cookie
)
reports_list = rep_list_data.get("reports", [])
found_report = any(r.get("id") == report_id for r in reports_list)
self.log_step(
"7b", "Clinician Reports List Verification",
st == 200 and found_report,
f"Report ID {report_id} verified in clinician reports list ({len(reports_list)} reports)"
)
print("----------------------------------------------------------------------")
print(" ALL VERIFICATION STEPS PASSED SUCCESSFULLY!")
print("======================================================================")
return 0
except Exception as ex:
print("----------------------------------------------------------------------")
print(f" [ERROR] Verification failed: {ex}")
print("======================================================================")
return 1
finally:
self.cleanup()
def cleanup(self):
"""Clean up background processes and temporary directories."""
if self.proc:
try:
self.proc.terminate()
self.proc.wait(timeout=5)
self.log_step("8a", "Clean Process Shutdown", True, f"PID {self.proc.pid} terminated cleanly")
except Exception:
try:
self.proc.kill()
self.proc.wait(timeout=2)
self.log_step("8a", "Process Shutdown via Kill", True, "Process killed")
except Exception as e:
print(f" [WARN] Cleanup exception: {e}")
# Clean up temporary database directory
if os.path.exists(self.temp_dir):
shutil.rmtree(self.temp_dir, ignore_errors=True)
def main():
parser = argparse.ArgumentParser(
description="HAD Digital MVP Acceptance Criteria Programmatic Verification"
)
# Support both --source / --exe flags and --mode source / --mode exe
parser.add_argument("--source", action="store_true", help="Test Python source (MVP/app.py)")
parser.add_argument("--exe", action="store_true", help="Test standalone executable")
parser.add_argument("--mode", choices=["source", "exe"], default=None, help="Mode: source or exe")
parser.add_argument("--app-path", default=None, help="Custom path to Python app script")
parser.add_argument("--exe-path", default=None, help="Custom path to standalone .exe")
parser.add_argument("--timeout", type=int, default=30, help="Startup timeout in seconds")
args = parser.parse_args()
# Determine mode
if args.exe:
mode = "exe"
target_path = args.exe_path
elif args.source:
mode = "source"
target_path = args.app_path
elif args.mode:
mode = args.mode
target_path = args.exe_path if mode == "exe" else args.app_path
else:
mode = "source"
target_path = args.app_path
runner = VerificationRunner(mode=mode, target_path=target_path, timeout=args.timeout)
exit_code = runner.run_verification()
sys.exit(exit_code)
if __name__ == "__main__":
main()