Open raw ↗#!/usr/bin/env python3
"""HAD Digital MVP - Empirical Stress Test Harness."""
import argparse
import concurrent.futures
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 StressHarness:
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_stress_")
self.db_path = os.path.join(self.temp_dir, "stress_had.db")
self.proc = None
self.failures = []
self.passed_assertions = 0
current_dir = Path(__file__).resolve().parent
self.project_root = current_dir.parent if current_dir.name == "05_Test" else current_dir
self.mvp_dir = self.project_root / "MVP"
def _find_free_port(self) -> int:
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(self, section: str, msg: str):
print(f"[{section}] {msg}")
def assert_true(self, condition: bool, test_name: str, details: str = ""):
if condition:
self.passed_assertions += 1
print(f" [PASS] {test_name}")
if details:
print(f" {details}")
else:
fail_msg = f"{test_name} FAILED: {details}"
self.failures.append(fail_msg)
print(f" [FAIL] {fail_msg}")
def start_server(self, custom_db: str = None) -> int:
db_to_use = custom_db or self.db_path
env = os.environ.copy()
env["HAD_DB_PATH"] = str(db_to_use)
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"
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: {exe_file}")
cmd = [str(exe_file), "--port", str(self.port)]
cwd = str(exe_file.parent)
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"App script not found: {app_file}")
cmd = [sys.executable, str(app_file), "--port", str(self.port)]
cwd = str(self.project_root)
self.proc = subprocess.Popen(
cmd, cwd=cwd, env=env,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
start = time.time()
online = False
while time.time() - start < self.timeout:
time.sleep(0.3)
if self.proc.poll() is not None:
out, err = self.proc.communicate()
raise RuntimeError(f"Server died on start (code {self.proc.returncode}):\n{err}\n{out}")
try:
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=2)
conn.request("GET", "/api/whoami")
resp = conn.getresponse()
conn.close()
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")
return self.proc.pid
def stop_server(self, force_kill: bool = False):
if self.proc:
pid = self.proc.pid
try:
# On Windows, PyInstaller onefile runs bootloader + child process. Use taskkill /T to kill process tree.
if sys.platform == "win32":
subprocess.run(["taskkill", "/F", "/T", "/PID", str(pid)], capture_output=True, check=False)
if force_kill:
self.proc.kill()
else:
self.proc.terminate()
self.proc.wait(timeout=5)
except Exception:
try:
self.proc.kill()
self.proc.wait(timeout=2)
except Exception:
pass
self.proc = None
def http_request(self, method: str, path: str, body: dict = None, cookie: str = None) -> tuple[int, dict, str]:
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 = resp.read().decode("utf-8")
set_cookie = resp.getheader("Set-Cookie")
try:
data = json.loads(raw)
except json.JSONDecodeError:
data = {"raw": raw}
return resp.status, data, set_cookie
finally:
conn.close()
def login(self, username, password) -> str:
st, data, cookie = self.http_request("POST", "/api/login", {"username": username, "password": password})
if st == 200 and cookie:
return cookie.split(";")[0]
return None
# =========================================================================
# CHALLENGE 1: Multi-Patient Toxicity Reporting & Direct SQLite Row Assertion
# =========================================================================
def test_challenge_1_multi_patient_reporting(self):
self.log("CHALLENGE 1", "Multi-Patient Toxicity Reporting & Direct SQLite Row Assertion")
cookie_p1 = self.login("patient.durand", "demo123")
cookie_p2 = self.login("patient.moreau", "demo123")
cookie_p3 = self.login("patient.laurent", "demo123")
self.assert_true(cookie_p1 and cookie_p2 and cookie_p3, "All 3 patient logins succeeded")
# Patient 1 submits multi-symptom dictionary report
p1_payload = {
"patient_id": 1,
"symptoms": {"nausea": 5, "fatigue": 2},
"notes": "Patient 1 multi-symptom batch report"
}
st1, data1, _ = self.http_request("POST", "/api/reports", p1_payload, cookie=cookie_p1)
self.assert_true(st1 == 201, "Patient 1 multi-symptom submission HTTP 201", f"Response: {data1}")
self.assert_true(len(data1.get("grades", [])) == 2, "Patient 1 returned 2 graded symptoms", f"{data1.get('grades')}")
self.assert_true(data1.get("alert", {}).get("alert_type") == "urgent", "Patient 1 Grade 2 nausea triggered urgent alert")
# Patient 2 submits single symptom report with proper input for diarrhea
p2_payload = {
"patient_id": 2,
"symptom_id": "diarrhea",
"symptom_category": "gastrointestinal",
"grading_inputs": {"stools_increase_per_day": 2},
"notes": "Patient 2 mild diarrhea"
}
st2, data2, _ = self.http_request("POST", "/api/reports", p2_payload, cookie=cookie_p2)
p2_rid = data2.get("report_id")
self.assert_true(st2 == 201, "Patient 2 single-symptom submission HTTP 201")
self.assert_true(data2.get("grading", {}).get("grade") == 1, "Patient 2 diarrhea correctly graded Grade 1")
self.assert_true(data2.get("alert") is None, "Patient 2 Grade 1 produces no alert")
# Patient 3 submits hematologic report
p3_payload = {
"patient_id": 3,
"symptom_id": "neutropenia",
"symptom_category": "hematologic",
"lab_values": {"anc_mm3": 350},
"notes": "Patient 3 febrile neutropenia test"
}
st3, data3, _ = self.http_request("POST", "/api/reports", p3_payload, cookie=cookie_p3)
p3_rid = data3.get("report_id")
self.assert_true(st3 == 201, "Patient 3 lab submission HTTP 201")
self.assert_true(data3.get("grading", {}).get("grade") == 3, "Patient 3 anc=350 graded Grade 3")
self.assert_true(data3.get("alert", {}).get("alert_type") == "emergency", "Patient 3 Grade 3 triggered emergency alert")
# Direct SQLite Row Assertions
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
# Verify toxicity_reports rows
cur.execute("SELECT id, patient_id, symptom_id FROM toxicity_reports WHERE patient_id IN (1, 2, 3) ORDER BY id DESC LIMIT 4")
recent_reps = cur.fetchall()
self.assert_true(len(recent_reps) == 4, f"SQLite: Found 4 toxicity_reports rows for patients 1,2,3 (got {len(recent_reps)})")
# Verify toxicity_grades rows by exact report_id
cur.execute("SELECT grade, provisional FROM toxicity_grades WHERE report_id = ?", (p2_rid,))
p2_gr = cur.fetchone()
self.assert_true(p2_gr == (1, 0), f"SQLite: Patient 2 report {p2_rid} is Grade 1, provisional=0 (got {p2_gr})")
cur.execute("SELECT grade, provisional FROM toxicity_grades WHERE report_id = ?", (p3_rid,))
p3_gr = cur.fetchone()
self.assert_true(p3_gr == (3, 1), f"SQLite: Patient 3 report {p3_rid} is Grade 3, provisional=1 (got {p3_gr})")
# Verify alerts
cur.execute("SELECT alert_type, severity FROM alerts WHERE report_id = ?", (p3_rid,))
p3_alert = cur.fetchone()
self.assert_true(p3_alert == ("emergency", "high"), f"SQLite: Alert for report {p3_rid} is emergency/high: {p3_alert}")
# Verify timeline events
cur.execute("SELECT patient_id, title FROM timeline_events WHERE patient_id IN (1, 2, 3) ORDER BY id DESC LIMIT 4")
tl_rows = cur.fetchall()
self.assert_true(len(tl_rows) >= 4, f"SQLite: Timeline events created for all submitted reports ({len(tl_rows)} found)")
conn.close()
# =========================================================================
# CHALLENGE 2: Multi-Role Access Control Matrix
# =========================================================================
def test_challenge_2_role_permissions(self):
self.log("CHALLENGE 2", "Multi-Role Access Control Matrix (Oncologist, HAD Nurse, Community Nurse, Patient, Admin)")
# 1. Unauthenticated client probes
protected_get_endpoints = [
"/api/patients",
"/api/patients/1",
"/api/reports?patient_id=1",
"/api/grades?patient_id=1",
"/api/alerts",
"/api/timeline?patient_id=1",
"/api/treatment-plan?patient_id=1",
"/api/messages",
"/api/export/summary?patient_id=1",
"/api/audit-log",
]
all_unauth_rejected = True
for ep in protected_get_endpoints:
st, _, _ = self.http_request("GET", ep)
if st != 401:
all_unauth_rejected = False
self.assert_true(False, f"Unauthenticated GET {ep} rejected", f"Got status {st}")
self.assert_true(all_unauth_rejected, "All protected GET endpoints reject unauthenticated access with 401")
# Protected POST endpoints
st, _, _ = self.http_request("POST", "/api/reports", {"patient_id": 1, "symptom_id": "nausea"})
self.assert_true(st == 401, "Unauthenticated POST /api/reports rejected with 401")
st, _, _ = self.http_request("POST", "/api/messages", {"body": "test message"})
self.assert_true(st == 401, "Unauthenticated POST /api/messages rejected with 401")
st, _, _ = self.http_request("POST", "/api/chat", {"message": "test chat"})
self.assert_true(st == 401, "Unauthenticated POST /api/chat rejected with 401")
# 2. Patient Role (patient.durand)
cookie_patient = self.login("patient.durand", "demo123")
st, _, _ = self.http_request("GET", "/api/audit-log", cookie=cookie_patient)
self.assert_true(st == 403, "Patient role forbidden from /api/audit-log (403)")
st, p_data, _ = self.http_request("GET", "/api/patients", cookie=cookie_patient)
self.assert_true(st == 200 and len(p_data.get("patients", [])) == 1, "Patient role /api/patients lists only self")
# 3. HAD Nurse Role (inf.moret)
cookie_had_nurse = self.login("inf.moret", "demo123")
st, _, _ = self.http_request("GET", "/api/audit-log", cookie=cookie_had_nurse)
self.assert_true(st == 403, "HAD Nurse role forbidden from /api/audit-log (403)")
st, n_data, _ = self.http_request("GET", "/api/patients", cookie=cookie_had_nurse)
self.assert_true(st == 200 and len(n_data.get("patients", [])) >= 1, "HAD Nurse role /api/patients lists active patients")
st, al_data, _ = self.http_request("GET", "/api/alerts", cookie=cookie_had_nurse)
self.assert_true(st == 200 and "alerts" in al_data, "HAD Nurse can access /api/alerts")
# 4. Community Nurse Role (inf.dubois)
cookie_comm_nurse = self.login("inf.dubois", "demo123")
st, _, _ = self.http_request("GET", "/api/audit-log", cookie=cookie_comm_nurse)
self.assert_true(st == 403, "Community Nurse forbidden from /api/audit-log (403)")
st, cn_data, _ = self.http_request("GET", "/api/patients", cookie=cookie_comm_nurse)
self.assert_true(st == 200 and len(cn_data.get("patients", [])) >= 1, "Community Nurse /api/patients lists patients")
st, tl_data, _ = self.http_request("GET", "/api/timeline?patient_id=1", cookie=cookie_comm_nurse)
self.assert_true(st == 200 and "events" in tl_data, "Community Nurse can access /api/timeline")
# 5. Oncologist Role (dr.martin)
cookie_onco = self.login("dr.martin", "demo123")
st, _, _ = self.http_request("GET", "/api/audit-log", cookie=cookie_onco)
self.assert_true(st == 403, "Oncologist role forbidden from /api/audit-log (403)")
st, exp_data, _ = self.http_request("GET", "/api/export/summary?patient_id=1", cookie=cookie_onco)
self.assert_true(st == 200 and "patient" in exp_data, "Oncologist can access /api/export/summary")
# 6. Admin Role (admin)
cookie_admin = self.login("admin", "admin123")
st, audit_data, _ = self.http_request("GET", "/api/audit-log", cookie=cookie_admin)
self.assert_true(st == 200 and "audit_log" in audit_data, "Admin role authorized for /api/audit-log (200)")
self.assert_true(len(audit_data.get("audit_log", [])) > 0, "Admin audit log contains logged events")
# =========================================================================
# CHALLENGE 3: High-Volume and Rapid Concurrent Sequential Submissions
# =========================================================================
def test_challenge_3_concurrency_stress(self):
self.log("CHALLENGE 3", "High-Volume and Rapid Concurrent Sequential Submissions (SQLite WAL Concurrency)")
cookie_p1 = self.login("patient.durand", "demo123")
cookie_onco = self.login("dr.martin", "demo123")
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM toxicity_reports WHERE patient_id = 1")
initial_count = cur.fetchone()[0]
conn.close()
num_writers = 20
reports_per_writer = 2
total_reports = num_writers * reports_per_writer
writer_errors = []
reader_errors = []
def submit_worker(worker_id):
for i in range(reports_per_writer):
payload = {
"patient_id": 1,
"symptom_id": "fatigue",
"symptom_category": "constitutional",
"severity_score": 2,
"notes": f"Concurrent stress thread {worker_id} report {i}"
}
try:
st, data, _ = self.http_request("POST", "/api/reports", payload, cookie=cookie_p1)
if st != 201:
writer_errors.append(f"Worker {worker_id}-{i} returned {st}: {data}")
except Exception as ex:
writer_errors.append(f"Worker {worker_id}-{i} exception: {ex}")
def reader_worker(reader_id):
for _ in range(5):
try:
st1, _, _ = self.http_request("GET", "/api/reports?patient_id=1", cookie=cookie_onco)
st2, _, _ = self.http_request("GET", "/api/timeline?patient_id=1", cookie=cookie_onco)
if st1 != 200 or st2 != 200:
reader_errors.append(f"Reader {reader_id} returned {st1}/{st2}")
except Exception as ex:
reader_errors.append(f"Reader {reader_id} exception: {ex}")
time.sleep(0.05)
start_time = time.time()
with concurrent.futures.ThreadPoolExecutor(max_workers=25) as executor:
writer_futures = [executor.submit(submit_worker, w) for w in range(num_writers)]
reader_futures = [executor.submit(reader_worker, r) for r in range(5)]
concurrent.futures.wait(writer_futures + reader_futures)
elapsed = time.time() - start_time
self.assert_true(
len(writer_errors) == 0,
f"Zero write errors across {total_reports} concurrent submissions in {elapsed:.2f}s",
f"Errors: {writer_errors[:3]}"
)
self.assert_true(
len(reader_errors) == 0,
"Zero read errors during concurrent SQLite writes",
f"Errors: {reader_errors[:3]}"
)
# Direct DB verification
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM toxicity_reports WHERE patient_id = 1")
final_count = cur.fetchone()[0]
self.assert_true(
final_count == initial_count + total_reports,
f"Exact row count match in SQLite: {final_count} == {initial_count} + {total_reports}"
)
# SQLite integrity check
cur.execute("PRAGMA integrity_check")
integrity = cur.fetchone()[0]
self.assert_true(integrity == "ok", f"SQLite PRAGMA integrity_check: {integrity}")
conn.close()
# =========================================================================
# CHALLENGE 4: CTCAE Rule Evaluation Across Boundary Scores & Malformed Payloads
# =========================================================================
def test_challenge_4_ctcae_boundaries_and_malformed(self):
self.log("CHALLENGE 4", "CTCAE Rule Evaluation Across Boundary Scores (0,1,2,3,4,5) & Malformed Payloads")
cookie_p1 = self.login("patient.durand", "demo123")
# Nausea boundary test cases: (score, expected_grade, expected_provisional, expected_alert_tier)
nausea_boundaries = [
(0, None, True, None), # Score 0: Below threshold
(1, 1, False, None), # Score 1: Lower bound Grade 1
(3, 1, False, None), # Score 3: Upper bound Grade 1
(4, 2, True, "urgent"), # Score 4: Lower bound Grade 2
(6, 2, True, "urgent"), # Score 6: Upper bound Grade 2
(7, 3, True, "emergency"), # Score 7: Lower bound Grade 3
(8, 3, True, "emergency"), # Score 8: Upper bound Grade 3
(9, 4, True, "emergency"), # Score 9: Lower bound Grade 4
(10, 4, True, "emergency"), # Score 10: Upper bound Grade 4
(11, 5, True, "emergency"), # Score 11: Grade 5
(12, None, True, None), # Score 12: Out of bounds high
(-2, None, True, None), # Score -2: Negative
(3.5, None, True, None), # Float gap between Grade 1 (3) and Grade 2 (4)
]
all_nausea_boundaries_ok = True
for score, exp_grade, exp_prov, exp_alert in nausea_boundaries:
payload = {
"patient_id": 1,
"symptom_id": "nausea",
"severity_score": score,
"notes": f"Boundary test score {score}"
}
st, data, _ = self.http_request("POST", "/api/reports", payload, cookie=cookie_p1)
if st != 201:
all_nausea_boundaries_ok = False
self.assert_true(False, f"Nausea score {score} HTTP 201", f"Got status {st}")
continue
grading = data.get("grading", {})
actual_grade = grading.get("grade")
if actual_grade != exp_grade:
all_nausea_boundaries_ok = False
self.assert_true(False, f"Nausea score {score} grade expectation", f"Expected {exp_grade}, got {actual_grade}")
self.assert_true(all_nausea_boundaries_ok, f"All {len(nausea_boundaries)} CTCAE nausea boundary scores evaluated correctly")
# Hematologic boundary tests (neutropenia anc_mm3)
neutro_cases = [
(1200, 1),
(800, 2),
(350, 3),
(100, 4),
(25, 5),
]
all_neutro_ok = True
for anc, exp_g in neutro_cases:
payload = {
"patient_id": 1,
"symptom_id": "neutropenia",
"lab_values": {"anc_mm3": anc},
"notes": f"Neutropenia ANC={anc}"
}
st, data, _ = self.http_request("POST", "/api/reports", payload, cookie=cookie_p1)
actual_g = data.get("grading", {}).get("grade")
if actual_g != exp_g:
all_neutro_ok = False
self.assert_true(False, f"Neutropenia ANC={anc} grade", f"Expected {exp_g}, got {actual_g}")
self.assert_true(all_neutro_ok, "All hematologic CTCAE neutropenia lab boundaries (Grades 1-5) match")
# Malformed payloads
# Empty payload
st, data, _ = self.http_request("POST", "/api/reports", {}, cookie=cookie_p1)
self.assert_true(st == 400, "Empty report payload rejected with HTTP 400")
# Missing symptom_id
st, data, _ = self.http_request("POST", "/api/reports", {"patient_id": 1, "severity_score": 2}, cookie=cookie_p1)
self.assert_true(st == 400, "Missing symptom_id rejected with HTTP 400")
# Non-numeric severity_score
st, data, _ = self.http_request(
"POST", "/api/reports",
{"patient_id": 1, "symptom_id": "nausea", "severity_score": "extremely_severe"},
cookie=cookie_p1
)
self.assert_true(st == 201 and data.get("grading", {}).get("grade") is None, "Non-numeric severity_score handled gracefully (no 500 crash)")
# Unknown symptom ID
st, data, _ = self.http_request(
"POST", "/api/reports",
{"patient_id": 1, "symptom_id": "unknown_exotic_syndrome_123", "severity_score": 3},
cookie=cookie_p1
)
self.assert_true(st == 201 and data.get("grading", {}).get("grade") is None, "Unknown symptom_id handled gracefully (no 500 crash)")
# Massive notes payload (25KB)
huge_notes = "A" * 25000
st, data, _ = self.http_request(
"POST", "/api/reports",
{"patient_id": 1, "symptom_id": "fatigue", "severity_score": 2, "notes": huge_notes},
cookie=cookie_p1
)
self.assert_true(st == 201, "25KB large notes payload accepted and stored cleanly")
# SQL Injection in notes
sqli_notes = "Normal notes'; DROP TABLE users; --"
st, data, _ = self.http_request(
"POST", "/api/reports",
{"patient_id": 1, "symptom_id": "fatigue", "severity_score": 1, "notes": sqli_notes},
cookie=cookie_p1
)
self.assert_true(st == 201, "SQL injection in notes payload accepted as literal string")
# Verify users table still intact
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM users")
user_count = cur.fetchone()[0]
self.assert_true(user_count > 0, f"Users table intact after SQL injection probe ({user_count} users)")
conn.close()
# =========================================================================
# CHALLENGE 5: Persistence Verification Across Server Process Restart
# =========================================================================
def test_challenge_5_persistence_across_restart(self):
self.log("CHALLENGE 5", "Persistence Verification Across Server Process Kill & Relaunch")
# Phase 1: Pre-restart submissions
cookie_p1 = self.login("patient.durand", "demo123")
cookie_nurse = self.login("inf.moret", "demo123")
# 1. Submit pre-restart toxicity reports
st_a, data_a, _ = self.http_request(
"POST", "/api/reports",
{"patient_id": 1, "symptom_id": "nausea", "severity_score": 5, "notes": "Pre-restart Nausea Gr 2"},
cookie=cookie_p1
)
rid_a = data_a.get("report_id")
st_b, data_b, _ = self.http_request(
"POST", "/api/reports",
{"patient_id": 1, "symptom_id": "vomiting", "grading_inputs": {"episodes_per_24h": 7}, "notes": "Pre-restart Vomiting Gr 3"},
cookie=cookie_p1
)
rid_b = data_b.get("report_id")
# 2. Send pre-restart message
msg_text = "Pre-restart urgent coordination message from HAD Nurse"
st_msg, data_msg, _ = self.http_request(
"POST", "/api/messages",
{"recipient_id": 2, "patient_id": 1, "subject": "Pre-restart check", "body": msg_text},
cookie=cookie_nurse
)
mid = data_msg.get("message_id")
self.assert_true(
st_a == 201 and st_b == 201 and st_msg == 201,
"Pre-restart reports and message submitted successfully",
f"Report IDs: {rid_a}, {rid_b}; Message ID: {mid}"
)
# Direct DB check before kill
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM toxicity_reports WHERE id IN (?, ?)", (rid_a, rid_b))
self.assert_true(cur.fetchone()[0] == 2, "Pre-restart: 2 reports exist in SQLite")
conn.close()
# Phase 2: ABRUPT PROCESS KILL
self.log("RESTART", f"Simulating abrupt crash/kill of process PID {self.proc.pid}...")
old_pid = self.proc.pid
self.stop_server(force_kill=True)
time.sleep(1.0)
self.assert_true(self.proc is None, f"Old server process PID {old_pid} killed")
# Verify port is released / server is unreachable
port_closed = False
try:
c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=1)
c.request("GET", "/api/whoami")
c.getresponse()
c.close()
except (OSError, http.client.HTTPException):
port_closed = True
self.assert_true(port_closed, f"Port {self.port} verified closed after process kill")
# Phase 3: RELAUNCH SERVER POINTING TO SAME DB
self.log("RESTART", f"Relaunching server pointing to existing database: {self.db_path}...")
new_pid = self.start_server(custom_db=self.db_path)
self.assert_true(new_pid is not None and new_pid != old_pid, f"New server process successfully running with PID {new_pid}")
# Phase 4: POST-RESTART VERIFICATION
# 1. Clinician logs in to new process
cookie_onco_new = self.login("dr.martin", "demo123")
self.assert_true(cookie_onco_new is not None, "Clinician successfully authenticated on restarted server")
# 2. Check reports persist
st_reps, reps_data, _ = self.http_request("GET", "/api/reports?patient_id=1", cookie=cookie_onco_new)
self.assert_true(st_reps == 200, "Post-restart /api/reports returned HTTP 200")
persisted_report_ids = [r["id"] for r in reps_data.get("reports", [])]
self.assert_true(
rid_a in persisted_report_ids and rid_b in persisted_report_ids,
"Both pre-crash reports (rid_a, rid_b) persist intact in report roster"
)
# 3. Check timeline persists
st_tl, tl_data, _ = self.http_request("GET", "/api/timeline?patient_id=1", cookie=cookie_onco_new)
self.assert_true(st_tl == 200, "Post-restart /api/timeline returned HTTP 200")
tl_desc = " ".join([e.get("description", "") for e in tl_data.get("events", [])])
self.assert_true(
"Nausea" in tl_desc or "Grade 2" in tl_desc,
"Pre-crash timeline events persist and are queryable post-restart"
)
# 4. Check message persists
st_msgs, msgs_data, _ = self.http_request("GET", "/api/messages?patient_id=1&limit=200", cookie=cookie_onco_new)
self.assert_true(st_msgs == 200, "Post-restart /api/messages returned HTTP 200")
persisted_messages = [m["body"] for m in msgs_data.get("messages", [])]
self.assert_true(
msg_text in persisted_messages,
"Pre-crash clinical care team message persists intact"
)
# 5. Post-restart write test: database accepts new writes seamlessly
cookie_p1_new = self.login("patient.durand", "demo123")
st_post, data_post, _ = self.http_request(
"POST", "/api/reports",
{"patient_id": 1, "symptom_id": "fatigue", "severity_score": 2, "notes": "Post-restart report test"},
cookie=cookie_p1_new
)
self.assert_true(st_post == 201, "New report accepted on restarted server (HTTP 201)")
rid_post = data_post.get("report_id")
# 6. Direct SQLite check on restarted database
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
cur.execute("SELECT id, symptom_id, severity_score FROM toxicity_reports WHERE id = ?", (rid_post,))
post_row = cur.fetchone()
self.assert_true(post_row is not None and post_row[0] == rid_post, "Post-restart row written to SQLite successfully")
cur.execute("PRAGMA integrity_check")
integrity = cur.fetchone()[0]
self.assert_true(integrity == "ok", f"Post-restart database PRAGMA integrity_check: {integrity}")
conn.close()
# =========================================================================
# RUN ALL CHALLENGES
# =========================================================================
def run_all(self) -> int:
print("=" * 75)
print(f" HAD DIGITAL MVP - EMPIRICAL CHALLENGER STRESS HARNESS ({self.mode.upper()})")
print(f" Target: {self.target_path or 'Default ' + self.mode}")
print(f" Ephemeral Port: {self.port}")
print(f" Isolated DB: {self.db_path}")
print("=" * 75)
start_time = time.time()
try:
self.start_server()
self.log("SERVER", f"Server online (PID {self.proc.pid}) on port {self.port}")
self.test_challenge_1_multi_patient_reporting()
self.test_challenge_2_role_permissions()
self.test_challenge_3_concurrency_stress()
self.test_challenge_4_ctcae_boundaries_and_malformed()
self.test_challenge_5_persistence_across_restart()
except Exception as ex:
self.log("ERROR", f"Unhandled harness exception: {ex}")
self.failures.append(f"Harness exception: {ex}")
import traceback
traceback.print_exc()
finally:
self.stop_server()
if os.path.exists(self.temp_dir):
shutil.rmtree(self.temp_dir, ignore_errors=True)
elapsed = time.time() - start_time
print("-" * 75)
print(f" SUMMARY FOR {self.mode.upper()}:")
print(f" Passed Assertions: {self.passed_assertions}")
print(f" Failures: {len(self.failures)}")
print(f" Execution Time: {elapsed:.2f}s")
print("=" * 75)
if self.failures:
print("\n[FAILED TESTS LOG]:")
for f in self.failures:
print(f" - {f}")
return 1
else:
print(f"\n[VERDICT: APPROVE] All empirical challenges passed for {self.mode.upper()}!")
return 0
def main():
parser = argparse.ArgumentParser(description="HAD Digital MVP Empirical Stress Harness")
parser.add_argument("--source", action="store_true", help="Run against Python source (MVP/app.py)")
parser.add_argument("--exe", action="store_true", help="Run against dist/HAD Digital.exe")
parser.add_argument("--all", action="store_true", help="Run against BOTH source and dist/HAD Digital.exe")
parser.add_argument("--app-path", default=None, help="Custom path to app.py")
parser.add_argument("--exe-path", default=None, help="Custom path to HAD Digital.exe")
args = parser.parse_args()
modes = []
if args.all or (not args.source and not args.exe):
modes = [("source", args.app_path), ("exe", args.exe_path)]
elif args.source:
modes.append(("source", args.app_path))
elif args.exe:
modes.append(("exe", args.exe_path))
overall_rc = 0
for mode, target in modes:
harness = StressHarness(mode=mode, target_path=target)
rc = harness.run_all()
if rc != 0:
overall_rc = rc
sys.exit(overall_rc)
if __name__ == "__main__":
main()