File size: 12,945 Bytes
00bd0c6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 | import pytest
from session_management.session_manager import SessionManager
@pytest.fixture
def manager():
"""Creates SessionManager in test mode (uses Test_PatientDB) for each test."""
sm = SessionManager(test_mode=True)
sm.db.clear_db()
yield sm
sm.db.clear_db()
@pytest.fixture
def patient(manager):
"""Create a test patient and return their pid."""
pid = manager.db.create_patient("Test Patient", "2000-01-01", "F")
return pid
@pytest.fixture
def session(manager, patient):
"""Start a session for the test patient and return (sid, pid)."""
sid = manager.start_session(patient)
return sid, patient
#─────── SESSION LIFECYCLE ────────────────────────────────────────────────────────────────────────────────
class TestSessionLifecycle:
def test_start_session_returns_sid(self, manager, patient):
sid = manager.start_session(patient)
assert sid is not None
def test_start_session_creates_active_session(self, manager, patient):
sid = manager.start_session(patient)
assert sid in manager.active_sessions
def test_start_session_stores_correct_patient_id(self, manager, patient):
sid = manager.start_session(patient)
state = manager.get_session(sid)
assert state.patient_id == patient
def test_get_session_returns_state(self, manager, session):
sid, _ = session
state = manager.get_session(sid)
assert state is not None
def test_get_session_invalid_id_returns_none(self, manager):
assert manager.get_session(9999) is None
def test_end_session_removes_from_active_sessions(self, manager, session):
sid, _ = session
manager.end_session(sid)
assert sid not in manager.active_sessions
def test_end_session_persists_symptoms_to_db(self, manager, session):
sid, pid = session
manager.add_validated_evidence(sid, symptoms=["persistent cough"])
manager.end_session(sid)
evidence = manager.db.get_patient_evidence(pid)
assert "persistent cough" in evidence["symptoms"]
def test_end_session_persists_risk_factors_to_db(self, manager, session):
sid, pid = session
manager.add_validated_evidence(sid, risk_factors=["tobacco smoking"])
manager.end_session(sid)
evidence = manager.db.get_patient_evidence(pid)
assert "tobacco smoking" in evidence["risk_factors"]
def test_end_session_persists_imaging_to_db(self, manager, session):
sid, pid = session
manager.add_validated_evidence(sid, imaging_detections=["pleural effusion"])
manager.end_session(sid)
evidence = manager.db.get_patient_evidence(pid)
assert "pleural effusion" in evidence["image_detections"]
def test_multiple_sessions_are_independent(self, manager, patient):
sid1 = manager.start_session(patient)
sid2 = manager.start_session(patient)
assert sid1 != sid2
assert sid1 in manager.active_sessions
assert sid2 in manager.active_sessions
#─────── EVIDENCE — TEXT, IMAGING, PATIENT RECORDS ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
class TestEvidenceManagement:
def test_add_text_evidence_to_current_loop(self, manager, session):
sid, _ = session
manager.add_text_evidence(sid, ["cough", "fatigue"])
evidence = manager.get_current_loop_evidence(sid)
assert "cough" in evidence["text_evidence"]
assert "fatigue" in evidence["text_evidence"]
def test_add_imaging_evidence_to_current_loop(self, manager, session):
sid, _ = session
manager.add_imaging_evidence(sid, ["lung opacity"])
evidence = manager.get_current_loop_evidence(sid)
assert "lung opacity" in evidence["imaging_evidence"]
def test_add_patient_records_evidence_to_current_loop(self, manager, session):
sid, _ = session
manager.add_patient_records_evidence(sid, ["tobacco smoking"])
evidence = manager.get_current_loop_evidence(sid)
assert "tobacco smoking" in evidence["patient_records"]
def test_add_validated_evidence_symptoms(self, manager, session):
sid, _ = session
manager.add_validated_evidence(sid, symptoms=["haemoptysis"])
validated = manager.get_validated_evidence(sid)
assert "haemoptysis" in validated["symptoms"]
def test_add_validated_evidence_risk_factors(self, manager, session):
sid, _ = session
manager.add_validated_evidence(sid, risk_factors=["asbestos"])
validated = manager.get_validated_evidence(sid)
assert "asbestos" in validated["risk_factors"]
def test_add_validated_evidence_imaging(self, manager, session):
sid, _ = session
manager.add_validated_evidence(sid, imaging_detections=["consolidation"])
validated = manager.get_validated_evidence(sid)
assert "consolidation" in validated["imaging_detections"]
def test_add_validated_evidence_accumulates_across_calls(self, manager, session):
sid, _ = session
manager.add_validated_evidence(sid, symptoms=["cough"])
manager.add_validated_evidence(sid, symptoms=["weight loss"])
validated = manager.get_validated_evidence(sid)
assert "cough" in validated["symptoms"]
assert "weight loss" in validated["symptoms"]
def test_add_validated_evidence_no_duplicates_in_session(self, manager, session):
sid, _ = session
manager.add_validated_evidence(sid, symptoms=["cough"])
manager.add_validated_evidence(sid, symptoms=["cough"])
validated = manager.get_validated_evidence(sid)
assert validated["symptoms"].count("cough") == 1
#──────── COMBINED EVIDENCE ────────────────────────────────────────────────────────────────────────────────────────────────────
class TestCombinedEvidence:
def test_get_combined_current_loop_evidence(self, manager, session):
sid, _ = session
manager.add_text_evidence(sid, ["cough"])
manager.add_imaging_evidence(sid, ["lung opacity"])
manager.add_patient_records_evidence(sid, ["tobacco smoking"])
combined = manager.get_combined_current_loop_evidence(sid)
assert "cough" in combined
assert "lung opacity" in combined
assert "tobacco smoking" in combined
def test_get_combined_session_evidence_no_duplicates(self, manager, session):
sid, _ = session
manager.add_validated_evidence(sid, symptoms=["cough"])
combined = manager.get_combined_session_evidence(sid, ["cough"])
assert combined.count("cough") == 1
def test_get_combined_session_evidence_merges_existing_and_validated(self, manager, session):
sid, _ = session
manager.add_validated_evidence(sid, symptoms=["weight loss"])
combined = manager.get_combined_session_evidence(sid, ["cough"])
assert "cough" in combined
assert "weight loss" in combined
#──────── CLEAR CURRENT LOOP EVIDENCE ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
class TestClearEvidence:
def test_clear_removes_text_evidence(self, manager, session):
sid, _ = session
manager.add_text_evidence(sid, ["cough"])
manager.clear_current_loop_evidence(sid)
evidence = manager.get_current_loop_evidence(sid)
assert len(evidence["text_evidence"]) == 0
def test_clear_removes_imaging_evidence(self, manager, session):
sid, _ = session
manager.add_imaging_evidence(sid, ["lung opacity"])
manager.clear_current_loop_evidence(sid)
evidence = manager.get_current_loop_evidence(sid)
assert len(evidence["imaging_evidence"]) == 0
def test_clear_preserves_patient_records(self, manager, session):
sid, _ = session
manager.add_patient_records_evidence(sid, ["tobacco smoking"])
manager.clear_current_loop_evidence(sid)
evidence = manager.get_current_loop_evidence(sid)
assert "tobacco smoking" in evidence["patient_records"]
#──────── RISK ASSIGNMENT ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
class TestRiskAssignment:
def test_assign_risk_sets_risk_level(self, manager, session):
sid, _ = session
manager.assign_risk(sid, "7")
state = manager.get_session(sid)
assert state.risk_level == "7"
def test_assign_risk_sets_flag(self, manager, session):
sid, _ = session
manager.assign_risk(sid, "7")
state = manager.get_session(sid)
assert state.is_risk_assigned is True
def test_assign_risk_can_be_updated(self, manager, session):
sid, _ = session
manager.assign_risk(sid, "3")
manager.assign_risk(sid, "6")
state = manager.get_session(sid)
assert state.risk_level == "6"
#──────── X-RAY EVIDENCE ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
class TestXray:
def test_add_xray_sets_path(self, manager, session):
sid, _ = session
manager.add_xray(sid, "/path/xray.jpg")
state = manager.get_session(sid)
assert state.x_ray_image_file == "/path/xray.jpg"
def test_add_xray_sets_uploaded_flag(self, manager, session):
sid, _ = session
manager.add_xray(sid, "/path/xray.jpg")
state = manager.get_session(sid)
assert state.x_ray_uploaded is True
#───── PATIENT EVIDENCE RETRIEVAL ───────────────────────────────────────────────────────────────────────────────
class TestGetPatientEvidence:
def test_get_patient_evidence_empty_for_new_patient(self, manager, patient):
evidence = manager.get_patient_evidence(patient)
assert evidence["symptoms"] == []
assert evidence["risk_factors"] == []
assert evidence["imaging"] == []
def test_get_patient_evidence_reflects_persisted_data(self, manager, session):
sid, pid = session
manager.add_validated_evidence(sid, symptoms=["cough"])
manager.end_session(sid)
evidence = manager.get_patient_evidence(pid)
assert "cough" in evidence["symptoms"]
#─────── HAS SUFFICIENT EVIDENCE ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
class TestHasSufficientEvidence:
def test_sufficient_evidence_true(self, manager):
assert manager.has_sufficient_evidence(["cough", "fever", "jaundice", "lower back pain"]) is True
def test_sufficient_evidence_exact_minimum(self, manager):
assert manager.has_sufficient_evidence(["cough", "fever", "jaundice"]) is True
def test_sufficient_evidence_false(self, manager):
assert manager.has_sufficient_evidence(["cough", "fever"]) is False
def test_sufficient_evidence_empty(self, manager):
assert manager.has_sufficient_evidence([]) is False |