387 lines
16 KiB
Python
387 lines
16 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
import asyncio
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
|
|
# Add project root to sys.path
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from app.ai.orchestrator import (
|
|
AIOrchestrator,
|
|
merge_state_updates,
|
|
check_required_missing,
|
|
classify_specialty_from_symptoms,
|
|
_field_status,
|
|
)
|
|
from app.services.contextual_normalizer import (
|
|
normalize_duration_from_message,
|
|
detect_previous_question_intent,
|
|
)
|
|
from app.services.completion_service import CompletionService
|
|
from app.schemas.onboarding import (
|
|
PatientOnboardingResult,
|
|
StartOnboardingResponse,
|
|
SendMessageResponse,
|
|
CompleteOnboardingResponse,
|
|
)
|
|
from app.schemas.voice import VoiceTurnResponse
|
|
|
|
|
|
def get_mock_session():
|
|
session = MagicMock()
|
|
session.session_id = "ONB-FLOW-TEST-001"
|
|
session.patient_id = "PAT-FLOW-001"
|
|
session.patient_name = "Ramesh Kumar"
|
|
session.age = 35
|
|
session.gender = "male"
|
|
session.category_type = "general"
|
|
session.category_code = "general_medicine"
|
|
session.category_name = "General Medicine"
|
|
session.language = "ta-IN"
|
|
session.stage = "greeting"
|
|
session.status = "in_progress"
|
|
return session
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 1: Patient Provides All Information in One Message
|
|
# =====================================================================
|
|
def test_1_patient_provides_all_info_in_one_message():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
msg = "எனக்கு diabetes இருக்கு, metformin 500 mg தினமும் சாப்பிடுறேன், தலைவலி மூணு நாளா இருக்கு, penicillin allergy இருக்கு."
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res = await orchestrator.process(mock_session, state_mock, [], msg)
|
|
merged = merge_state_updates({}, res["state_updates"])
|
|
|
|
assert len(merged.get("medical_history", [])) >= 1
|
|
assert len(merged.get("medications", [])) >= 1
|
|
assert len(merged.get("allergies", [])) >= 1
|
|
assert len(merged.get("symptoms", [])) >= 1
|
|
assert res.get("completed") is True
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 2: Patient Gives Symptoms First
|
|
# =====================================================================
|
|
def test_2_patient_gives_symptoms_first():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
msg = "எனக்கு தலைவலி இருக்கு."
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res = await orchestrator.process(mock_session, state_mock, [], msg)
|
|
assert res.get("completed") is False
|
|
assert len(res.get("reply", "")) > 5
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 3: Patient Gives Duration (No Fake Symptom Created)
|
|
# =====================================================================
|
|
def test_3_patient_gives_duration_no_fake_symptom():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
state_mock = MagicMock(state_json=json.dumps({
|
|
"reason_for_visit": "தலைவலி",
|
|
"symptoms": [{"name": "தலைவலி"}]
|
|
}))
|
|
|
|
res = await orchestrator.process(
|
|
mock_session, state_mock,
|
|
[{"role": "patient", "message": "தலைவலி இருக்கு", "stage": "greeting"}],
|
|
"மூணு நாளா இருக்கு."
|
|
)
|
|
merged = merge_state_updates({"symptoms": [{"name": "தலைவலி"}]}, res["state_updates"])
|
|
syms = merged.get("symptoms", [])
|
|
|
|
assert len(syms) == 1
|
|
assert syms[0]["name"] == "தலைவலி"
|
|
assert syms[0]["duration"] == "3 days"
|
|
assert res.get("completed") is False
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 4: Tamil Tanglish Duration
|
|
# =====================================================================
|
|
def test_4_tamil_tanglish_duration():
|
|
assert normalize_duration_from_message("10 டேஸா இருக்கு.") == "10 days"
|
|
assert normalize_duration_from_message("10 days-ah இருக்கு.") == "10 days"
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 5: Telugu Duration Support
|
|
# =====================================================================
|
|
def test_5_telugu_duration():
|
|
assert normalize_duration_from_message("మూడు రోజుల నుంచి ఉంది") == "3 days"
|
|
assert normalize_duration_from_message("3 రోజుల నుండి ఉంది") == "3 days"
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 6: Kannada Duration Support
|
|
# =====================================================================
|
|
def test_6_kannada_duration():
|
|
assert normalize_duration_from_message("ಮೂರು ದಿನದಿಂದ ಇದೆ") == "3 days"
|
|
assert normalize_duration_from_message("3 ದಿನದಿಂದ ಇದೆ") == "3 days"
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 7: Medication First (No Fake Symptom Created)
|
|
# =====================================================================
|
|
def test_7_medication_first_no_fake_symptom():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res = await orchestrator.process(mock_session, state_mock, [], "Metformin 500 mg தினமும் சாப்பிடுறேன்.")
|
|
updates = res["state_updates"]
|
|
|
|
meds = updates.get("medications") or []
|
|
syms = updates.get("symptoms") or []
|
|
|
|
if isinstance(meds, dict):
|
|
meds = meds.get("items", [])
|
|
|
|
assert len(meds) == 1
|
|
assert meds[0]["name"] == "Metformin"
|
|
assert len(syms) == 0
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 8: Cross-Language Symptom Deduplication
|
|
# =====================================================================
|
|
def test_8_cross_language_symptom_deduplication():
|
|
state1 = merge_state_updates({}, {"symptoms": [{"name": "தலைவலி"}]})
|
|
state2 = merge_state_updates(state1, {"symptoms": [{"name": "headache", "duration": "3 days"}]})
|
|
syms = state2.get("symptoms", [])
|
|
|
|
assert len(syms) == 1
|
|
assert syms[0]["duration"] == "3 days"
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 9: Progressive Enrichment
|
|
# =====================================================================
|
|
def test_9_progressive_enrichment():
|
|
s1 = merge_state_updates({}, {"medical_history": [{"condition": "Diabetes"}]})
|
|
s2 = merge_state_updates(s1, {"medical_history": [{"condition": "Diabetes", "duration": "8 years"}]})
|
|
mh = s2.get("medical_history", [])
|
|
|
|
assert len(mh) == 1
|
|
assert mh[0]["condition"] == "Diabetes"
|
|
assert mh[0]["duration"] == "8 years"
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 10: Patient Correction
|
|
# =====================================================================
|
|
def test_10_patient_correction():
|
|
s1 = merge_state_updates({}, {"symptoms": [{"name": "headache", "duration": "3 days"}]})
|
|
s2 = merge_state_updates(s1, {"symptoms": [{"name": "headache", "duration": "10 days"}]})
|
|
syms = s2.get("symptoms", [])
|
|
|
|
assert len(syms) == 1
|
|
assert syms[0]["duration"] == "10 days"
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 11: Emergency Override (Spoken Tamil)
|
|
# =====================================================================
|
|
def test_11_emergency_override_tamil():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res = await orchestrator.process(mock_session, state_mock, [], "மூச்சு விட முடியல")
|
|
|
|
assert ("மருத்துவமனை" in res["reply"] or "அவசர" in res["reply"])
|
|
assert res.get("completed") is True
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 12: Mental Health Crisis Override (Spoken Tamil)
|
|
# =====================================================================
|
|
def test_12_mental_health_crisis_override_tamil():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res = await orchestrator.process(mock_session, state_mock, [], "நான் என்னையே காயப்படுத்திக்கணும்னு தோணுது")
|
|
|
|
assert "14416" in res["reply"] or "Tele-MANAS" in res["reply"]
|
|
assert res.get("completed") is True
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 13: Nothing Else Handling
|
|
# =====================================================================
|
|
def test_13_nothing_else_handling():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
ready_state = {
|
|
"reason_for_visit": "fever",
|
|
"symptoms": [{"name": "fever", "duration": "2 days"}],
|
|
"medical_history": {"status": "confirmed_none", "items": []},
|
|
"medications": {"status": "confirmed_none", "items": []},
|
|
"allergies": [],
|
|
}
|
|
state_mock = MagicMock(state_json=json.dumps(ready_state))
|
|
|
|
res = await orchestrator.process(
|
|
mock_session, state_mock,
|
|
[{"role": "assistant", "message": "Do you have any allergies?", "stage": "allergies"}],
|
|
"வேற எதுவும் இல்லை"
|
|
)
|
|
|
|
assert res.get("completed") is True
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 14: Mental Health Specialty Focus
|
|
# =====================================================================
|
|
def test_14_mental_health_specialty():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
mock_session.category_code = "psychiatry"
|
|
orchestrator = AIOrchestrator()
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res = await orchestrator.process(mock_session, state_mock, [], "எனக்கு ரொம்ப stress, தூக்கம் சரியா வரல, இரண்டு வாரமா இப்படித்தான்.")
|
|
reply = res["reply"]
|
|
|
|
assert not any(k in reply for k in ["இருமல்", "பிபி", "வயிற்று வலி"])
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 15: Cardiology Specialty Focus
|
|
# =====================================================================
|
|
def test_15_cardiology_specialty():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
mock_session.category_code = "cardiology"
|
|
orchestrator = AIOrchestrator()
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res = await orchestrator.process(mock_session, state_mock, [], "நெஞ்சு படபடன்னு இருக்கு, நடக்கும்போது மூச்சு வாங்குது.")
|
|
syms = res["state_updates"].get("symptoms", [])
|
|
|
|
assert len(syms) >= 2
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 16: Multiple Symptoms Extraction
|
|
# =====================================================================
|
|
def test_16_multiple_symptoms_extraction():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res = await orchestrator.process(mock_session, state_mock, [], "தலைவலி, கை கால் வலி, தலைசுற்றல், நெஞ்சு படபடப்பு.")
|
|
syms = res["state_updates"].get("symptoms", [])
|
|
|
|
assert len(syms) >= 3
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 17: No False Symptom from Medication Statement
|
|
# =====================================================================
|
|
def test_17_no_false_symptom_from_medication():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res = await orchestrator.process(mock_session, state_mock, [], "நான் Metformin சாப்பிடுறேன்.")
|
|
syms = res["state_updates"].get("symptoms") or []
|
|
|
|
assert len(syms) == 0
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 18: Session Isolation Verification
|
|
# =====================================================================
|
|
def test_18_session_isolation():
|
|
state_a = merge_state_updates({}, {"symptoms": [{"name": "headache"}]})
|
|
state_b = merge_state_updates({}, {"symptoms": [{"name": "fever"}]})
|
|
|
|
assert state_a["symptoms"][0]["name"] == "headache"
|
|
assert state_b["symptoms"][0]["name"] == "fever"
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 19 & 20: API Response & Voice Turn Compatibility
|
|
# =====================================================================
|
|
def test_19_20_api_backward_compatibility():
|
|
res = VoiceTurnResponse(
|
|
success=True,
|
|
session_id="SESS-001",
|
|
patient_transcript="hello",
|
|
ai_text="hi",
|
|
stage="greeting",
|
|
completed=False,
|
|
audio_url="/v1/voice-onboarding/audio/123",
|
|
response_time_ms=100.0,
|
|
response_time_sec=0.1
|
|
)
|
|
assert res.success is True
|
|
assert res.completed is False
|
|
|
|
|
|
# =====================================================================
|
|
# TEST 21: Exact Bug Regression Test (Multi-Symptom + Duration Turn 2)
|
|
# =====================================================================
|
|
def test_21_exact_bug_regression():
|
|
async def run():
|
|
mock_session = get_mock_session()
|
|
orchestrator = AIOrchestrator()
|
|
msg1 = "எனக்கு ரொம்ப தலைவலி, கை கால் வலியாக இருக்குது. என் நெஞ்சு அப்பப்ப படபடன்னு இருக்குது. தலை சுத்துது, தலை பாரமாக இருக்குது."
|
|
state_mock = MagicMock(state_json=json.dumps({}))
|
|
|
|
res1 = await orchestrator.process(mock_session, state_mock, [], msg1)
|
|
s1 = merge_state_updates({}, res1["state_updates"])
|
|
state_mock2 = MagicMock(state_json=json.dumps(s1))
|
|
|
|
res2 = await orchestrator.process(
|
|
mock_session, state_mock2,
|
|
[{"role": "patient", "message": msg1, "stage": "greeting"}, {"role": "assistant", "message": res1["reply"], "stage": res1["next_stage"]}],
|
|
"பிரச்சனை வந்து ஒரு மூணு நாளா இருக்கு."
|
|
)
|
|
|
|
# MUST NOT complete after duration answer alone
|
|
assert res2.get("completed") is False
|
|
assert "எத்தனை நாளா" not in res2["reply"]
|
|
|
|
asyncio.run(run())
|