Open raw ↗"""Tier 1: Feature Coverage & Nominal Contract Tests.
Validates that 100% of core features and REST API endpoints behave strictly
per ORIGINAL_REQUEST.md, PROJECT.md, and 03_Architecture/API_Contract.md:
- Multi-Role Authentication & Session Management (>= 5 tests)
- Patient Toxicity Reporting (>= 5 tests)
- CTCAE Automated Grading Engine (>= 5 tests)
- Tiered Alert Routing & Acknowledgment (>= 5 tests)
- Care Coordination Timeline Feed (>= 5 tests)
- Supplementary Features: Patients, Guidance, Messaging, Export (>= 5 tests)
"""
import pytest
@pytest.mark.tier1
class TestAuthFeature:
"""Feature 1: Role-Based Authentication & Session Management."""
def test_patient_login_success(self, api_client):
resp = api_client.login("patient.durand", "demo123")
assert resp.status_code == 200
data = resp.json()
assert data.get("user", {}).get("role") == "patient"
assert data.get("user", {}).get("username") == "patient.durand"
assert "HAD_SESSION" in resp.headers.get("Set-Cookie", "")
def test_oncologist_login_success(self, api_client):
resp = api_client.login("dr.martin", "demo123")
assert resp.status_code == 200
data = resp.json()
assert data.get("user", {}).get("role") == "oncologist"
assert data.get("user", {}).get("username") == "dr.martin"
def test_had_nurse_login_success(self, api_client):
resp = api_client.login("inf.moret", "demo123")
assert resp.status_code == 200
data = resp.json()
assert data.get("user", {}).get("role") == "had_nurse"
assert data.get("user", {}).get("username") == "inf.moret"
def test_admin_login_success(self, api_client):
resp = api_client.login("admin", "admin123")
assert resp.status_code == 200
data = resp.json()
assert data.get("user", {}).get("role") == "admin"
def test_invalid_credentials_rejected(self, api_client):
resp = api_client.login("patient.durand", "incorrect_password")
assert resp.status_code == 401
data = resp.json()
assert "error" in data or "Invalid" in data.get("message", "")
def test_whoami_unauthenticated(self, api_client):
resp = api_client.get("/api/whoami")
# Unauthenticated whoami returns 401 or 200 with authenticated: False
assert resp.status_code in (200, 401)
data = resp.json()
assert data.get("authenticated") is False
def test_whoami_authenticated(self, patient_client):
resp = patient_client.get("/api/whoami")
assert resp.status_code == 200
data = resp.json()
assert data.get("authenticated") is True
assert data.get("user", {}).get("username") == "patient.durand"
def test_logout_clears_session(self, api_client):
login_resp = api_client.login("patient.durand", "demo123")
assert login_resp.status_code == 200
logout_resp = api_client.logout()
assert logout_resp.status_code == 200
# Subsequent check verifies unauthenticated state
check_resp = api_client.get("/api/whoami")
assert check_resp.status_code in (200, 401)
assert check_resp.json().get("authenticated") is False
@pytest.mark.tier1
class TestToxicityReportingFeature:
"""Feature 2: Patient Toxicity Reporting."""
def test_submit_nausea_report(self, patient_client):
payload = {
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": 2,
"notes": "Mild nausea managed with liquids",
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
data = resp.json()
assert "report_id" in data
assert data["report_id"] > 0
assert "grading" in data
def test_submit_fever_report_constitutional(self, patient_client):
payload = {
"patient_id": 1,
"symptom_id": "fever",
"symptom_category": "constitutional",
"severity_score": 3,
"lab_values": {"temp_celsius": 38.6},
"notes": "Temperature 38.6C measured at 14:00",
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
data = resp.json()
assert data.get("report_id") is not None
def test_submit_neuropathy_report(self, patient_client):
payload = {
"patient_id": 1,
"symptom_id": "peripheral_neuropathy",
"symptom_category": "neurologic",
"severity_score": 1,
"notes": "Mild tingling in fingertips after cold exposure",
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
def test_list_reports_by_patient(self, patient_client):
resp = patient_client.get("/api/reports?patient_id=1")
assert resp.status_code == 200
data = resp.json()
assert "reports" in data
assert len(data["reports"]) >= 1
# Verify structure
first = data["reports"][0]
assert "id" in first
assert "symptom_id" in first
assert "patient_id" in first
def test_report_preserves_reporter_attribution(self, patient_client, session_server):
payload = {
"patient_id": 1,
"symptom_id": "fatigue",
"symptom_category": "constitutional",
"severity_score": 2,
"notes": "Moderate fatigue attribution test",
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
report_id = resp.json()["report_id"]
rows = session_server.query_db(
"SELECT reporter_id, notes FROM toxicity_reports WHERE id = ?", (report_id,)
)
assert len(rows) == 1
assert rows[0]["reporter_id"] == patient_client.current_user["id"]
assert rows[0]["notes"] == payload["notes"]
def test_unauthenticated_report_submission_rejected(self, api_client):
payload = {
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": 2,
}
resp = api_client.post("/api/reports", json=payload)
assert resp.status_code == 401
@pytest.mark.tier1
class TestCTCAEGradingFeature:
"""Feature 3: CTCAE Automated Grading Engine."""
def test_ctcae_grade_1_nausea(self, patient_client):
payload = {
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": 2,
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
grading = resp.json().get("grading", {})
assert grading.get("grade") == 1
assert grading.get("provisional") is False
def test_ctcae_grade_2_nausea_provisional(self, patient_client):
payload = {
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": 5,
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
grading = resp.json().get("grading", {})
assert grading.get("grade") == 2
# Grades >= 2 must be flagged provisional pending clinician review
assert grading.get("provisional") is True
def test_ctcae_grade_3_diarrhea(self, patient_client):
payload = {
"patient_id": 1,
"symptom_id": "diarrhea",
"symptom_category": "gastrointestinal",
"grading_inputs": {"stools_increase_per_day": 8},
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
grading = resp.json().get("grading", {})
assert grading.get("grade") == 3
assert grading.get("provisional") is True
def test_ctcae_grade_3_fever(self, patient_client):
payload = {
"patient_id": 1,
"symptom_id": "fever",
"symptom_category": "constitutional",
"lab_values": {"temp_celsius": 40.2},
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
grading = resp.json().get("grading", {})
assert grading.get("grade") == 3
assert grading.get("provisional") is True
def test_ctcae_list_grades_endpoint(self, patient_client):
resp = patient_client.get("/api/grades?patient_id=1")
assert resp.status_code == 200
data = resp.json()
assert "grades" in data
assert len(data["grades"]) >= 1
grade_entry = data["grades"][0]
assert "grade" in grade_entry
assert "criteria" in grade_entry
assert "provisional" in grade_entry
def test_ctcae_list_symptoms(self, api_client):
resp = api_client.get("/api/symptoms")
assert resp.status_code == 200
symptoms = resp.json().get("symptoms", [])
assert len(symptoms) >= 10
symptom_ids = [s["symptom_id"] for s in symptoms]
assert "nausea" in symptom_ids
assert "vomiting" in symptom_ids
assert "diarrhea" in symptom_ids
assert "fever" in symptom_ids
assert "peripheral_neuropathy" in symptom_ids
@pytest.mark.tier1
class TestAlertRoutingFeature:
"""Feature 4: Tiered Alert Routing & Acknowledgment."""
def test_grade_1_routine_no_alert_row(self, patient_client, session_server):
# Grade 1 is routine: no row created in alerts table
payload = {
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": 1,
"notes": "Routine grade 1 nausea",
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
report_id = resp.json()["report_id"]
# Alert returned in response is None for Grade 1
assert resp.json().get("alert") is None
# Verify directly in SQLite alerts table
alert_rows = session_server.query_db(
"SELECT * FROM alerts WHERE report_id = ?", (report_id,)
)
assert len(alert_rows) == 0
def test_grade_2_urgent_creates_alert(self, patient_client, session_server):
payload = {
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": 5,
"notes": "Urgent grade 2 nausea",
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
report_id = resp.json()["report_id"]
alert = resp.json().get("alert")
assert alert is not None
assert alert.get("alert_type") == "urgent"
assert alert.get("severity") == "medium"
# Direct DB verification
alert_rows = session_server.query_db(
"SELECT * FROM alerts WHERE report_id = ?", (report_id,)
)
assert len(alert_rows) == 1
assert alert_rows[0]["alert_type"] == "urgent"
def test_grade_3_emergency_creates_alert(self, patient_client, session_server):
payload = {
"patient_id": 1,
"symptom_id": "fever",
"symptom_category": "constitutional",
"lab_values": {"temp_celsius": 40.5},
"notes": "High fever emergency",
}
resp = patient_client.post("/api/reports", json=payload)
assert resp.status_code == 201
report_id = resp.json()["report_id"]
alert = resp.json().get("alert")
assert alert is not None
assert alert.get("alert_type") == "emergency"
assert alert.get("severity") in ("high", "critical")
def test_list_alerts_endpoint(self, oncologist_client):
resp = oncologist_client.get("/api/alerts?patient_id=1")
assert resp.status_code == 200
data = resp.json()
assert "alerts" in data
assert isinstance(data["alerts"], list)
def test_acknowledge_alert_workflow(self, nurse_client, session_server):
# Create an alert first
cur = session_server.execute_db(
"INSERT INTO alerts (patient_id, alert_type, severity, title, message, status) "
"VALUES (1, 'urgent', 'medium', 'Test Alert', 'Test Message', 'pending')"
)
alert_rows = session_server.query_db(
"SELECT id FROM alerts WHERE status = 'pending' ORDER BY id DESC LIMIT 1"
)
assert len(alert_rows) > 0
alert_id = alert_rows[0]["id"]
# Clinician acknowledges alert
resp = nurse_client.post(f"/api/alerts/{alert_id}/acknowledge")
assert resp.status_code == 200
assert "acknowledged" in resp.json().get("message", "").lower()
# Direct DB check confirms acknowledged status
updated = session_server.query_db("SELECT status, acknowledged_by FROM alerts WHERE id = ?", (alert_id,))
assert updated[0]["status"] == "acknowledged"
assert updated[0]["acknowledged_by"] == nurse_client.current_user["id"]
def test_acknowledge_invalid_alert_id(self, nurse_client):
resp = nurse_client.post("/api/alerts/99999/acknowledge")
assert resp.status_code == 404
@pytest.mark.tier1
class TestCareTimelineFeature:
"""Feature 5: Care Coordination Timeline."""
def test_timeline_event_created_on_report_submission(self, patient_client, oncologist_client):
unique_note = "Timeline test unique note 98765"
payload = {
"patient_id": 1,
"symptom_id": "fatigue",
"symptom_category": "constitutional",
"severity_score": 2,
"notes": unique_note,
}
rep_resp = patient_client.post("/api/reports", json=payload)
assert rep_resp.status_code == 201
# Query timeline as oncologist
tl_resp = oncologist_client.get("/api/timeline?patient_id=1")
assert tl_resp.status_code == 200
events = tl_resp.json().get("events", [])
assert len(events) >= 1
event_titles = [e.get("title", "") for e in events]
assert any("fatigue" in t.lower() for t in event_titles)
def test_timeline_events_ordered_descending(self, oncologist_client):
resp = oncologist_client.get("/api/timeline?patient_id=1")
assert resp.status_code == 200
events = resp.json().get("events", [])
if len(events) >= 2:
dates = [e["event_date"] for e in events]
# Descending order check
for i in range(len(dates) - 1):
assert dates[i] >= dates[i + 1]
def test_timeline_filtered_by_patient(self, oncologist_client):
resp = oncologist_client.get("/api/timeline?patient_id=2")
assert resp.status_code == 200
events = resp.json().get("events", [])
for e in events:
assert e["patient_id"] == 2
def test_timeline_limit_query_param(self, oncologist_client):
resp = oncologist_client.get("/api/timeline?patient_id=1&limit=2")
assert resp.status_code == 200
events = resp.json().get("events", [])
assert len(events) <= 2
def test_clinician_can_view_patient_timeline(self, oncologist_client):
resp = oncologist_client.get("/api/timeline?patient_id=1")
assert resp.status_code == 200
assert "events" in resp.json()
def test_unauthenticated_timeline_rejected(self, api_client):
resp = api_client.get("/api/timeline?patient_id=1")
assert resp.status_code == 401
@pytest.mark.tier1
class TestSupplementaryFeatures:
"""Feature 6: Roster, Detail, Guidance, Messaging, and Export."""
def test_patient_roster_listing(self, oncologist_client):
resp = oncologist_client.get("/api/patients")
assert resp.status_code == 200
patients = resp.json().get("patients", [])
assert len(patients) >= 3
mrns = [p["mrn"] for p in patients]
assert "MRN-2024-001" in mrns
def test_patient_detail_view(self, oncologist_client):
resp = oncologist_client.get("/api/patients/1")
assert resp.status_code == 200
data = resp.json()
assert data["id"] == 1
assert data["first_name"] == "Jeanne"
assert data["last_name"] == "Durand"
assert "active_treatment_plan" in data or "cancer_type" in data
def test_french_guidance_lookup(self, api_client):
resp = api_client.get("/api/guidance?symptom_id=nausea&grade=2")
assert resp.status_code == 200
data = resp.json()
assert data.get("symptom_id") == "nausea"
assert data.get("grade") == 2
assert "guidance" in data
guidance = data["guidance"]
assert "patient_message" in guidance
assert "actions" in guidance
def test_export_summary_endpoint(self, oncologist_client):
resp = oncologist_client.get("/api/export/summary?patient_id=1")
assert resp.status_code == 200
data = resp.json()
assert "patient" in data
assert "treatment_plans" in data
assert "toxicity_summary" in data
assert data["patient"]["mrn"] == "MRN-2024-001"
def test_messaging_between_users(self, patient_client, nurse_client):
# Patient sends message to nurse
msg_payload = {
"recipient_role": "had_nurse",
"patient_id": 1,
"subject": "Question concerning nausea meds",
"body": "Bonjour, should I take ondansetron with food?",
"message_type": "direct",
}
send_resp = patient_client.post("/api/messages", json=msg_payload)
assert send_resp.status_code == 201
msg_id = send_resp.json().get("message_id")
assert msg_id is not None
# Nurse lists messages
list_resp = nurse_client.get("/api/messages?patient_id=1")
assert list_resp.status_code == 200
messages = list_resp.json().get("messages", [])
assert any(m["id"] == msg_id for m in messages)