ElebobaandClaude Opus 5.5
fa40be3d94
CI / ruff · pytest (pull_request) Successful in 25s
- Engine.handle registered via register_engine; lexicon-first safety (no LLM on ACUTE), fixed crisis script once per open crisis, POSSIBLE → C-SSRS screen - FSM registry (@state/@on_enter/@trigger) with auto-loaded flows; core flow: new → consent → age_gate → module onboarding → idle, crisis, consent_revoked - reusable scale session (buttons, extract_item + clarify, item 9 / C-SSRS buttons-only, crisis flags), execution of monitoring dialog actions, texts in config/texts/dialog.yaml Refs #10 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
486 lines
22 KiB
Python
486 lines
22 KiB
Python
"""Dialog engine (#10): consent, age gate, crisis without LLM, C-SSRS on POSSIBLE, scale sessions, revocation.
|
||
|
||
Two in-memory SQLite databases (identity / clinical), offline mock LLM wrapped in a spy that records every task
|
||
call, and a fake monitoring service (lane B stub) that records calls and returns scripted actions.
|
||
"""
|
||
|
||
from collections.abc import AsyncIterator
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from sqlalchemy import select
|
||
|
||
from app.config import get_settings
|
||
from app.core import clock
|
||
from app.core.rules import Action
|
||
from app.core.safety import crisis_reply
|
||
from app.core.types import AlertLevel, ScrubbedText
|
||
from app.db import identity_repo, repo, session
|
||
from app.db.bootstrap import init_db
|
||
from app.db.clinical import ItemResponse, Message, SafetyPlan
|
||
from app.dialog.contracts import Event, Reply, get_engine
|
||
from app.dialog.engine import Engine
|
||
from app.dialog.fsm import Ctx, Step, on_enter
|
||
from app.dialog.texts import load_dialog_texts
|
||
from app.integrations.base import VerifiedPerson, integrations
|
||
from app.llm.gateway import LLMGateway
|
||
from app.llm.mock import MockTasks
|
||
from app.llm.schemas import RiskVerdict
|
||
|
||
TOKEN = "a" * 32
|
||
TEXTS = load_dialog_texts()
|
||
MEM = "sqlite+aiosqlite:///:memory:"
|
||
|
||
|
||
class SpyTasks(MockTasks):
|
||
"""Offline mock that records which LLM tasks were called (to prove the crisis path never calls the LLM)."""
|
||
|
||
def __init__(self, risk: str | None = None) -> None:
|
||
self.calls: list[str] = []
|
||
self._risk = risk
|
||
|
||
async def extract_item(self, scale_item_id: str, item_text: str, max_score: int, answer: ScrubbedText) -> Any:
|
||
self.calls.append("extract_item")
|
||
return await super().extract_item(scale_item_id, item_text, max_score, answer)
|
||
|
||
async def extract_checkin(self, text: ScrubbedText) -> Any:
|
||
self.calls.append("extract_checkin")
|
||
return await super().extract_checkin(text)
|
||
|
||
async def risk_classify(self, text: ScrubbedText) -> RiskVerdict | None:
|
||
self.calls.append("risk_classify")
|
||
if self._risk is not None:
|
||
return RiskVerdict(level=self._risk, confidence=0.9) # type: ignore[arg-type]
|
||
return await super().risk_classify(text)
|
||
|
||
async def phrase(self, intent: str, payload: ScrubbedText, base_text: str) -> str | None:
|
||
self.calls.append("phrase")
|
||
return base_text
|
||
|
||
|
||
class FakeMonitoring:
|
||
def __init__(self, actions: list[Action] | None = None, in_crisis: bool = True) -> None:
|
||
self.calls = 0
|
||
self.actions = actions or []
|
||
self.in_crisis = in_crisis
|
||
|
||
async def after_data_change(self, token: str) -> list[Action]:
|
||
self.calls += 1
|
||
out, self.actions = self.actions, []
|
||
return out
|
||
|
||
async def is_in_crisis(self, token: str) -> bool:
|
||
return self.in_crisis
|
||
|
||
|
||
@pytest.fixture
|
||
async def dbs() -> AsyncIterator[None]:
|
||
await session.configure(identity_url=MEM, clinical_url=MEM)
|
||
await init_db()
|
||
async with session.get_identity_session() as s:
|
||
await identity_repo.create_patient(
|
||
s,
|
||
token=TOKEN,
|
||
display_name="Тестовый Пациент",
|
||
snils="000-000-000 00",
|
||
clinic_id="c1",
|
||
doctor_id="doc-test",
|
||
now=clock.now(),
|
||
birth_year=1992,
|
||
sex="m",
|
||
)
|
||
async with session.get_clinical_session() as s:
|
||
await repo.upsert_profile(s, token=TOKEN, doctor_id="doc-test", modules=[], sex="m")
|
||
yield
|
||
await session.configure() # back to default URLs for other test modules
|
||
|
||
|
||
@pytest.fixture
|
||
def spy() -> SpyTasks:
|
||
return SpyTasks()
|
||
|
||
|
||
@pytest.fixture
|
||
def mon() -> FakeMonitoring:
|
||
return FakeMonitoring()
|
||
|
||
|
||
@pytest.fixture
|
||
def engine(spy: SpyTasks, mon: FakeMonitoring) -> Engine:
|
||
return Engine(gateway=LLMGateway(spy), monitoring=mon)
|
||
|
||
|
||
# ----------------------------------------------------------------------------------------------- helpers
|
||
|
||
|
||
async def text(e: Engine, t: str) -> list[Reply]:
|
||
return await e.handle(TOKEN, Event(type="text", text=t))
|
||
|
||
|
||
async def press(e: Engine, data: str) -> list[Reply]:
|
||
return await e.handle(TOKEN, Event(type="button", button=data))
|
||
|
||
|
||
async def system(e: Engine, kind: str, **params: Any) -> list[Reply]:
|
||
return await e.handle(TOKEN, Event(type="system", system=kind, params=params)) # type: ignore[arg-type]
|
||
|
||
|
||
async def set_state(state: str, **data: Any) -> None:
|
||
async with session.get_clinical_session() as s:
|
||
await repo.set_dialog_state(s, token=TOKEN, state=state, data={"consent": True, **data}, now=clock.now())
|
||
|
||
|
||
async def get_state() -> tuple[str, dict[str, Any]]:
|
||
async with session.get_clinical_session() as s:
|
||
row = await repo.get_dialog_state(s, TOKEN)
|
||
assert row is not None
|
||
return row.state, dict(row.data)
|
||
|
||
|
||
async def stored_messages() -> list[Message]:
|
||
async with session.get_clinical_session() as s:
|
||
return list((await s.scalars(select(Message).where(Message.token == TOKEN).order_by(Message.id))).all())
|
||
|
||
|
||
async def identity_patient() -> Any:
|
||
async with session.get_identity_session() as s:
|
||
return await identity_repo.get_patient(s, TOKEN)
|
||
|
||
|
||
async def set_birth_year(year: int | None) -> None:
|
||
async with session.get_identity_session() as s:
|
||
patient = await identity_repo.get_patient(s, TOKEN)
|
||
assert patient is not None
|
||
patient.birth_year = year
|
||
|
||
|
||
def buttons(reply: Reply) -> list[str]:
|
||
return [b.data for row in reply.buttons for b in row]
|
||
|
||
|
||
# A fake module onboarding entry point, to check the hand-over without depending on #36.
|
||
@on_enter("zztest.onboarding.start")
|
||
async def _zz_onboarding(ctx: Ctx) -> Step:
|
||
return Step(replies=[Reply(text="zz onboarding")])
|
||
|
||
|
||
# ----------------------------------------------------------------------------------------------- registration
|
||
|
||
|
||
async def test_get_engine_returns_real_engine() -> None:
|
||
assert isinstance(get_engine(), Engine)
|
||
|
||
|
||
async def test_consent_accept_goes_to_idle(dbs: None, engine: Engine) -> None:
|
||
r = await engine.handle(TOKEN, Event(type="start"))
|
||
assert "152-ФЗ" in r[0].text and buttons(r[0]) == ["consent:yes", "consent:no"]
|
||
assert (await get_state())[0] == "consent"
|
||
|
||
r = await press(engine, "consent:yes")
|
||
state, data = await get_state()
|
||
# Birth year 1992 is known from registration → adult without asking; no module flow → idle.
|
||
assert state == "idle" and data["consent"] is True
|
||
assert r[-1].text.startswith(TEXTS.idle.welcome[:20]) and "consent:revoke" in buttons(r[-1])
|
||
patient = await identity_patient()
|
||
assert patient.consent_at is not None and patient.consent_ver == TEXTS.consent.version
|
||
|
||
|
||
async def test_connect_code_verified(dbs: None, engine: Engine, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
class Ident:
|
||
async def verify(self, auth_code: str) -> VerifiedPerson | None:
|
||
return VerifiedPerson(token=TOKEN, doctor_id="doc-test") if auth_code == "GOOD" else None
|
||
|
||
monkeypatch.setattr(integrations, "identity", Ident())
|
||
r = await engine.handle(TOKEN, Event(type="start", params={"code": "BAD"}))
|
||
assert r[0].text == TEXTS.consent.code_invalid and (await get_state())[0] == "new"
|
||
r = await engine.handle(TOKEN, Event(type="start", params={"code": "GOOD"}))
|
||
assert buttons(r[0]) == ["consent:yes", "consent:no"]
|
||
|
||
|
||
async def test_consent_refused_no_processing(dbs: None, engine: Engine, spy: SpyTasks) -> None:
|
||
await engine.handle(TOKEN, Event(type="start"))
|
||
r = await press(engine, "consent:no")
|
||
assert r[0].text == TEXTS.consent.refused and (await get_state())[0] == "refused"
|
||
r = await text(engine, "мне плохо, не сплю, позвоните 89161234567")
|
||
assert r[0].text == TEXTS.consent.refused
|
||
assert spy.calls == [] # nothing reached the LLM
|
||
assert await stored_messages() == [] # nothing stored without consent
|
||
assert (await identity_patient()).consent_at is None
|
||
r = await press(engine, "consent:restart")
|
||
assert buttons(r[0]) == ["consent:yes", "consent:no"]
|
||
|
||
|
||
async def test_minor_gets_child_helpline_and_stop(dbs: None, engine: Engine, spy: SpyTasks) -> None:
|
||
await set_birth_year(None)
|
||
await engine.handle(TOKEN, Event(type="start"))
|
||
r = await press(engine, "consent:yes")
|
||
assert buttons(r[-1]) == ["age:adult", "age:minor"]
|
||
r = await text(engine, "2012")
|
||
assert "8-800-2000-122" in r[0].text
|
||
state, data = await get_state()
|
||
assert state == "minor" and data["consent"] is False
|
||
assert (await identity_patient()).revoked_at is not None # a minor's consent is withdrawn
|
||
calls = len(spy.calls)
|
||
r = await text(engine, "а что дальше?")
|
||
assert "8-800-2000-122" in r[0].text and len(spy.calls) == calls
|
||
|
||
|
||
async def test_age_gate_button_adult(dbs: None, engine: Engine) -> None:
|
||
await set_birth_year(None)
|
||
await engine.handle(TOKEN, Event(type="start"))
|
||
await press(engine, "consent:yes")
|
||
r = await text(engine, "не скажу")
|
||
assert r[0].text.startswith(TEXTS.age_gate.unclear)
|
||
await press(engine, "age:adult")
|
||
assert (await get_state())[0] == "idle"
|
||
|
||
|
||
async def test_hand_over_to_module_onboarding(dbs: None, engine: Engine) -> None:
|
||
async with session.get_clinical_session() as s:
|
||
await repo.upsert_profile(s, token=TOKEN, modules=["zztest"])
|
||
await engine.handle(TOKEN, Event(type="start"))
|
||
r = await press(engine, "consent:yes")
|
||
assert r[-1].text == "zz onboarding"
|
||
assert (await get_state())[0] == "zztest.onboarding.start"
|
||
|
||
|
||
# ----------------------------------------------------------------------------------------------- safety
|
||
|
||
|
||
async def test_acute_sends_script_without_any_llm_call(
|
||
dbs: None, engine: Engine, spy: SpyTasks, mon: FakeMonitoring
|
||
) -> None:
|
||
await set_state("idle")
|
||
r = await text(engine, "хочу покончить с собой")
|
||
assert len(r) == 1 and r[0].kind == "crisis"
|
||
assert r[0].text == crisis_reply(["suicide"]).text # verbatim fixed script
|
||
assert spy.calls == [] # lexicon ACUTE → risk_classify is not even called
|
||
state, data = await get_state()
|
||
assert state == "crisis" and data["crisis"]["open"] is True
|
||
event = data["safety_events"][-1]
|
||
assert event["level"] == "acute" and event["categories"] == ["suicide"] and event["handled"] == "crisis_script"
|
||
assert mon.calls == 1 # lane B gets the chance to escalate
|
||
|
||
|
||
async def test_repeat_crisis_does_not_duplicate_script(dbs: None, engine: Engine, spy: SpyTasks) -> None:
|
||
await set_state("idle")
|
||
script = (await text(engine, "хочу покончить с собой"))[0].text
|
||
r = await text(engine, "не хочу жить")
|
||
assert r[0].kind == "crisis" and r[0].text != script
|
||
assert r[0].text.startswith(TEXTS.crisis.repeat) and "112" in r[0].text
|
||
assert spy.calls == []
|
||
r = await text(engine, "ладно")
|
||
assert r[0].text == TEXTS.crisis.hold
|
||
_, data = await get_state()
|
||
assert [e["handled"] for e in data["safety_events"]] == ["crisis_script", "crisis_repeat"]
|
||
|
||
|
||
async def test_doctor_ack_leaves_crisis(dbs: None, engine: Engine) -> None:
|
||
await set_state("idle")
|
||
await text(engine, "хочу покончить с собой")
|
||
r = await system(engine, "doctor_ack")
|
||
assert r[0].text == TEXTS.crisis.closed
|
||
state, data = await get_state()
|
||
assert state == "idle" and data["crisis"]["open"] is False
|
||
# A new acute message after the crisis was closed gets the full script again.
|
||
r = await text(engine, "хочу покончить с собой")
|
||
assert r[0].text == crisis_reply(["suicide"]).text
|
||
|
||
|
||
async def test_crisis_left_when_monitoring_says_resolved(dbs: None, spy: SpyTasks) -> None:
|
||
mon = FakeMonitoring(in_crisis=False)
|
||
e = Engine(gateway=LLMGateway(spy), monitoring=mon)
|
||
await set_state("crisis", crisis={"open": True})
|
||
r = await text(e, "мне получше")
|
||
assert r[0].text == TEXTS.crisis.closed and (await get_state())[0] == "idle"
|
||
|
||
|
||
async def test_possible_starts_cssrs_and_returns(dbs: None, engine: Engine, spy: SpyTasks) -> None:
|
||
await set_state("idle")
|
||
r = await text(engine, "всё бессмысленно, устал от всего")
|
||
assert "risk_classify" in spy.calls
|
||
assert (await get_state())[0] == "scale.cssrs_screen"
|
||
assert r[-1].kind == "question" and [b.label for row in r[-1].buttons for b in row] == ["Нет", "Да"]
|
||
assert TEXTS.scale.risk_screen_intro in r[-1].text
|
||
|
||
r = await text(engine, "нет") # C-SSRS is buttons-only: free text is never sent to extract_item
|
||
assert r[0].text.startswith(TEXTS.scale.buttons_only) and "extract_item" not in spy.calls
|
||
|
||
for i in range(1, 7):
|
||
r = await press(engine, f"scale:cssrs_screen:cssrs_{i}:0")
|
||
state, data = await get_state()
|
||
assert state == "idle" and "cssrs_screen" not in data.get("scales", {})
|
||
assert r[0].text == TEXTS.scale.done
|
||
async with session.get_clinical_session() as s:
|
||
(res,) = await repo.list_scale_results(s, TOKEN, "cssrs_screen")
|
||
assert res.severity == "none" and res.scheduled is False and res.complete
|
||
assert data["safety_events"][-1]["handled"] == "risk_screen"
|
||
|
||
|
||
async def test_cssrs_high_goes_to_crisis(dbs: None, engine: Engine) -> None:
|
||
await set_state("idle")
|
||
await text(engine, "всё бессмысленно, устал от всего")
|
||
await press(engine, "scale:cssrs_screen:cssrs_1:1")
|
||
await press(engine, "scale:cssrs_screen:cssrs_2:1")
|
||
await press(engine, "scale:cssrs_screen:cssrs_3:0")
|
||
r = await press(engine, "scale:cssrs_screen:cssrs_4:1") # intent → high: the session ends at once
|
||
assert any(x.kind == "crisis" and x.text == crisis_reply([]).text for x in r)
|
||
assert (await get_state())[0] == "crisis"
|
||
|
||
|
||
async def test_llm_signal_can_raise_risk(dbs: None, mon: FakeMonitoring) -> None:
|
||
spy = SpyTasks(risk="possible")
|
||
e = Engine(gateway=LLMGateway(spy), monitoring=mon)
|
||
await set_state("idle")
|
||
await text(e, "обычный день") # lexicon NONE, LLM says possible → max() = POSSIBLE
|
||
state, data = await get_state()
|
||
assert state == "scale.cssrs_screen" and data["safety_events"][-1]["source"] == "llm"
|
||
|
||
|
||
# ----------------------------------------------------------------------------------------------- scale session
|
||
|
||
|
||
async def test_scale_session_buttons_free_text_and_clarify(dbs: None, engine: Engine, spy: SpyTasks) -> None:
|
||
await set_state("idle")
|
||
r = await system(engine, "scale_due", scale="phq9", scheduled=True)
|
||
assert (await get_state())[0] == "scale.phq9"
|
||
assert r[0].kind == "question" and len(buttons(r[0])) == 4 and "1/9" in r[0].text
|
||
|
||
await press(engine, "scale:phq9:phq9_1:1")
|
||
r = await text(engine, "почти каждый день") # mock extract: confident score 3
|
||
assert "3/9" in r[0].text
|
||
r = await text(engine, "ну такое") # no cue → low confidence → clarify, nothing stored
|
||
assert r[0].text.startswith(TEXTS.scale.clarify) and "3/9" in r[0].text
|
||
r = await press(engine, "scale:phq9:phq9_1:2") # stale button from an earlier question → re-ask
|
||
assert "3/9" in r[0].text
|
||
for i in range(3, 9):
|
||
await press(engine, f"scale:phq9:phq9_{i}:0")
|
||
r = await text(engine, "нет") # item 9 is buttons-only
|
||
assert r[0].text.startswith(TEXTS.scale.buttons_only)
|
||
assert spy.calls.count("extract_item") == 2
|
||
r = await press(engine, "scale:phq9:phq9_9:0")
|
||
assert r[0].text == TEXTS.scale.done and (await get_state())[0] == "idle"
|
||
|
||
async with session.get_clinical_session() as s:
|
||
(res,) = await repo.list_scale_results(s, TOKEN, "phq9")
|
||
items = (await s.scalars(select(ItemResponse).order_by(ItemResponse.id))).all()
|
||
assert (res.total, res.item9, res.complete, res.scheduled) == (4, 0, True, True)
|
||
assert [(i.item_id, i.source) for i in items[:2]] == [("phq9_1", "button"), ("phq9_2", "llm")]
|
||
assert items[1].score == 3 and items[1].confidence is not None
|
||
|
||
|
||
async def test_phq9_item9_positive_crisis_then_cssrs(dbs: None, engine: Engine, spy: SpyTasks) -> None:
|
||
await set_state("idle")
|
||
await system(engine, "scale_due", scale="phq9", scheduled=False)
|
||
for i in range(1, 9):
|
||
await press(engine, f"scale:phq9:phq9_{i}:0")
|
||
r = await press(engine, "scale:phq9:phq9_9:2")
|
||
kinds = [x.kind for x in r]
|
||
assert "crisis" in kinds and r[kinds.index("crisis")].text == crisis_reply([]).text
|
||
assert r[-1].kind == "question" and "cssrs" not in r[-1].text # C-SSRS q1 asked after the script
|
||
state, data = await get_state()
|
||
assert state == "scale.cssrs_screen" and data["crisis"]["open"] is True
|
||
assert data["scales"]["cssrs_screen"]["return_state"] == "crisis"
|
||
async with session.get_clinical_session() as s:
|
||
(res,) = await repo.list_scale_results(s, TOKEN, "phq9")
|
||
assert res.item9 == 2 and res.scheduled is False and res.flags == {"phq9_item9_positive": True}
|
||
assert "extract_item" not in spy.calls
|
||
# A repeated acute phrase during the C-SSRS: short text, no second script, the screen continues.
|
||
r = await text(engine, "не хочу жить")
|
||
assert r[0].text.startswith(TEXTS.crisis.repeat) and (await get_state())[0] == "scale.cssrs_screen"
|
||
|
||
|
||
# ----------------------------------------------------------------------------------------------- monitoring actions
|
||
|
||
|
||
async def test_monitoring_actions_executed(dbs: None, spy: SpyTasks) -> None:
|
||
mon = FakeMonitoring(
|
||
actions=[
|
||
Action(type="alert", level=AlertLevel.HIGH, code="x"),
|
||
Action(type="show_safety_plan", code="phq9_item9"),
|
||
Action(type="schedule_scale", code="mood_low_3d", params={"scale": "phq9", "scheduled": False}),
|
||
]
|
||
)
|
||
e = Engine(gateway=LLMGateway(spy), monitoring=mon)
|
||
async with session.get_clinical_session() as s:
|
||
s.add(SafetyPlan(token=TOKEN, steps=["Позвонить брату Олегу", "Прогулка"], updated_at=clock.now()))
|
||
await set_state("idle")
|
||
r = await press(e, "menu:contact_doctor") # records a contact request → monitoring runs
|
||
assert r[0].text == TEXTS.idle.contact_doctor_ack
|
||
plan = next(x for x in r if x.kind == "safety_plan")
|
||
assert "Позвонить брату Олегу" in plan.text
|
||
assert r[-1].kind == "question" # unscheduled PHQ-9 started right away (idle = resting)
|
||
state, data = await get_state()
|
||
assert state == "scale.phq9" and data["scales"]["phq9"]["scheduled"] is False
|
||
assert data["contact_requests"]
|
||
# Stored outbound texts are scrubbed (the plan names a relative).
|
||
assert not any("Олег" in m.text_scrubbed for m in await stored_messages())
|
||
|
||
|
||
async def test_suggest_visit_and_med_info_phrased(dbs: None, spy: SpyTasks) -> None:
|
||
mon = FakeMonitoring(
|
||
actions=[
|
||
Action(type="med_info", code="alcohol_3d", params={"topic": "alcohol_interaction"}),
|
||
Action(type="suggest_visit", code="alcohol_3d", params={"reason": "alcohol"}),
|
||
]
|
||
)
|
||
e = Engine(gateway=LLMGateway(spy), monitoring=mon)
|
||
await set_state("idle")
|
||
r = await press(e, "menu:contact_doctor")
|
||
assert any("алкогол" in x.text for x in r) and buttons(r[-1]) == ["menu:visit"]
|
||
|
||
|
||
# ------------------------------------------------------------------------------------ consent revocation, privacy
|
||
|
||
|
||
async def test_revoke_consent_then_silence(dbs: None, engine: Engine, spy: SpyTasks) -> None:
|
||
await set_state("idle")
|
||
r = await press(engine, "consent:revoke")
|
||
assert r[0].text == TEXTS.revoked.confirm
|
||
state, data = await get_state()
|
||
assert state == "consent_revoked" and data["consent"] is False and data["delete_requested"] is True
|
||
assert (await identity_patient()).revoked_at is not None
|
||
before = len(await stored_messages())
|
||
assert await text(engine, "привет, как дела?") == []
|
||
assert await system(engine, "daily_checkin") == []
|
||
assert spy.calls == [] and len(await stored_messages()) == before
|
||
r = await engine.handle(TOKEN, Event(type="start")) # coming back = asking for consent again
|
||
assert buttons(r[0]) == ["consent:yes", "consent:no"]
|
||
|
||
|
||
async def test_revoke_by_text_command(dbs: None, engine: Engine) -> None:
|
||
await set_state("idle")
|
||
r = await text(engine, "Отозвать согласие")
|
||
assert r[0].text == TEXTS.revoked.confirm and (await get_state())[0] == "consent_revoked"
|
||
|
||
|
||
async def test_messages_stored_scrubbed(dbs: None, engine: Engine) -> None:
|
||
await set_state("idle")
|
||
r = await text(engine, "Перезвоните мне на 8 916 123-45-67, сегодня обычный день")
|
||
assert r and buttons(r[0]) # neutral support reply with the idle menu
|
||
msgs = await stored_messages()
|
||
inbound = [m.text_scrubbed for m in msgs if m.direction == "in"]
|
||
assert inbound and "[PHONE]" in inbound[0]
|
||
assert not any("123-45-67" in m.text_scrubbed or "916" in m.text_scrubbed for m in msgs)
|
||
assert any(m.direction == "out" for m in msgs)
|
||
|
||
|
||
async def test_db_unavailable_still_answers_crisis(engine: Engine, tmp_path: Any) -> None:
|
||
await session.configure(identity_url=MEM, clinical_url=f"sqlite+aiosqlite:///{tmp_path}/missing/dir/x.db")
|
||
try:
|
||
r = await text(engine, "хочу покончить с собой")
|
||
assert r[0].kind == "crisis" and "112" in r[0].text
|
||
assert crisis_reply(["suicide"]).doctor_notified not in r[0].text # nobody could be notified
|
||
r = await text(engine, "привет")
|
||
assert r[0].text == TEXTS.unavailable
|
||
finally:
|
||
await session.configure()
|
||
|
||
|
||
def test_dialog_texts_avoid_forbidden_words() -> None:
|
||
raw = (get_settings().config_dir / "texts" / "dialog.yaml").read_text(encoding="utf-8").lower()
|
||
body = "\n".join(line for line in raw.splitlines() if not line.lstrip().startswith("#"))
|
||
for word in ("учёт", "учет", "контрол", "надзор", "диагноз:"):
|
||
assert word not in body
|
||
assert "152-фз" in body and "8-800-2000-122" in body
|