282 lines
12 KiB
Python
282 lines
12 KiB
Python
"""End-to-end server-only MAX flow, real SQLite transactions and mocked network."""
|
|
|
|
from datetime import timedelta
|
|
from unittest.mock import AsyncMock
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import select
|
|
|
|
from app.channels import max_bot
|
|
from app.channels.max_client import MaxApiError
|
|
from app.config import get_settings
|
|
from app.core import clock
|
|
from app.db import repo, session
|
|
from app.db.bootstrap import init_db
|
|
from app.db.clinical import ScaleResult
|
|
from app.db.intake import QuestionnaireRecord
|
|
from app.db.max_notifications import MaxBinding, MaxDelivery, MaxReceipt
|
|
from app.dialog.contracts import Reply
|
|
from app.integrations import max_store as store
|
|
from app.integrations.base import MemoryOutbox, integrations
|
|
from app.main import app
|
|
|
|
TOKEN = "synthetic-patient"
|
|
SECRET = "synthetic-webhook-secret-" * 2
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
async def database(monkeypatch):
|
|
for key, value in {
|
|
"MAX_ENABLED": "true",
|
|
"MAX_BOT_TOKEN": "fake-token",
|
|
"MAX_BOT_USERNAME": "test_bot",
|
|
"MAX_WEBHOOK_SECRET": SECRET,
|
|
"MAX_WEBHOOK_URL": "https://api.example.org/api/v1/webhooks/max",
|
|
"MAX_SITE_URL": "https://example.org",
|
|
}.items():
|
|
monkeypatch.setenv(key, value)
|
|
get_settings.cache_clear()
|
|
max_bot.texts.cache_clear()
|
|
await session.configure(identity_url="sqlite+aiosqlite:///:memory:", clinical_url="sqlite+aiosqlite:///:memory:")
|
|
await init_db()
|
|
async with session.get_clinical_session() as s:
|
|
await repo.upsert_profile(s, token=TOKEN, doctor_id="")
|
|
await repo.set_dialog_state(
|
|
s, token=TOKEN, state="idle", data={"consent": True, "adult": True}, now=clock.now()
|
|
)
|
|
yield
|
|
await max_bot.stop()
|
|
await session.configure()
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def event(kind="bot_started", text="", uid=100, *, stamp=1000):
|
|
user = {"user_id": uid, "is_bot": False, "first_name": "Synthetic"}
|
|
if kind == "message_created":
|
|
return {
|
|
"update_type": kind,
|
|
"timestamp": stamp,
|
|
"message": {
|
|
"sender": user,
|
|
"recipient": {"chat_type": "dialog", "chat_id": 123},
|
|
"body": {"mid": f"mid-{stamp}", "text": text},
|
|
},
|
|
}
|
|
return {"update_type": kind, "timestamp": stamp, "user": user, "chat_id": 123, "payload": text}
|
|
|
|
|
|
async def connect():
|
|
async with session.get_identity_session() as s:
|
|
code, _ = await store.issue_code(s, TOKEN, clock.now(), 600)
|
|
await max_bot.handle_update(event(text=code))
|
|
return code
|
|
|
|
|
|
async def test_http_connect_webhook_duplicate_notification_and_stop():
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
auth = {"Authorization": f"Bearer {TOKEN}"}
|
|
linked = await client.post("/api/v1/me/notifications/max/link", headers=auth, json={"consent": True})
|
|
code = parse_qs(urlsplit(linked.json()["url"]).query)["start"][0]
|
|
headers = {"X-Max-Bot-Api-Secret": SECRET}
|
|
for _ in range(2):
|
|
assert (
|
|
await client.post("/api/v1/webhooks/max", headers=headers, json=event(text=code))
|
|
).status_code == 200
|
|
assert (await client.get("/api/v1/me/notifications/max", headers=auth)).json()["connected"]
|
|
async with session.get_identity_session() as s:
|
|
assert len((await s.scalars(select(MaxReceipt))).all()) == 1
|
|
assert len((await s.scalars(select(MaxDelivery))).all()) == 1
|
|
sender = AsyncMock()
|
|
assert await max_bot.deliver_one(sender)
|
|
assert sender.send.call_args.args == (100, max_bot.texts().connected)
|
|
await client.post("/api/v1/webhooks/max", headers=headers, json=event("message_created", "/stop", stamp=2000))
|
|
assert not (await client.get("/api/v1/me/notifications/max", headers=auth)).json()["connected"]
|
|
|
|
|
|
async def test_unbound_start_help_and_invalid_code_are_generic():
|
|
for e in (event(), event("message_created", "private patient text", stamp=2000), event(text="expired", stamp=3000)):
|
|
await max_bot.handle_update(e)
|
|
async with session.get_identity_session() as s:
|
|
rows = (await s.scalars(select(MaxDelivery))).all()
|
|
assert {r.kind for r in rows} == {"not_connected", "help", "invalid"}
|
|
assert all(r.token is None for r in rows)
|
|
assert not (await s.scalars(select(MaxBinding))).all()
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["bot_stopped", "dialog_removed", "message_created"])
|
|
async def test_stopping_cancels_pending_confirmation_and_reminders(kind):
|
|
await connect()
|
|
await max_bot.MaxOutbox(MemoryOutbox()).push(TOKEN, [Reply(text="private scheduled question")])
|
|
await max_bot.handle_update(event(kind, "/stop", stamp=2000))
|
|
async with session.get_identity_session() as s:
|
|
assert not (await s.get(MaxBinding, TOKEN)).enabled
|
|
rows = (await s.scalars(select(MaxDelivery).where(MaxDelivery.token == TOKEN))).all()
|
|
assert rows and all(r.state == "cancelled" for r in rows)
|
|
sender = AsyncMock()
|
|
await max_bot.deliver_one(sender)
|
|
assert sender.send.call_count == (1 if kind == "message_created" else 0)
|
|
|
|
|
|
async def test_private_patient_content_stays_in_fallback_only():
|
|
await connect()
|
|
fallback = MemoryOutbox()
|
|
wrapper = max_bot.MaxOutbox(fallback)
|
|
replies = [Reply(text="SYNTHETIC confidential result 42", kind="question")]
|
|
await wrapper.push(TOKEN, replies)
|
|
await wrapper.push(TOKEN, replies)
|
|
assert fallback.drain(TOKEN) == replies * 2
|
|
async with session.get_identity_session() as s:
|
|
rows = (await s.scalars(select(MaxDelivery).where(MaxDelivery.kind == "notification"))).all()
|
|
assert len(rows) == 1
|
|
assert "confidential" not in repr(rows[0].__dict__)
|
|
sender = AsyncMock()
|
|
while await max_bot.deliver_one(sender):
|
|
pass
|
|
assert any(c.args[1] == max_bot.texts().notification for c in sender.send.call_args_list)
|
|
assert all("confidential" not in c.args[1] and TOKEN not in str(c.kwargs) for c in sender.send.call_args_list)
|
|
|
|
|
|
async def test_revoked_consent_prevents_binding_and_delivery():
|
|
await connect()
|
|
async with session.get_clinical_session() as s:
|
|
await repo.set_dialog_state(
|
|
s, token=TOKEN, state="idle", data={"consent": False, "adult": True}, now=clock.now()
|
|
)
|
|
sender = AsyncMock()
|
|
await max_bot.deliver_one(sender)
|
|
sender.send.assert_not_called()
|
|
async with session.get_identity_session() as s:
|
|
await store.disconnect(s, TOKEN)
|
|
code, _ = await store.issue_code(s, TOKEN, clock.now(), 600)
|
|
await max_bot.handle_update(event(text=code, stamp=2000))
|
|
async with session.get_identity_session() as s:
|
|
assert not (await s.get(MaxBinding, TOKEN)).enabled
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"bad",
|
|
[
|
|
{},
|
|
{"update_type": "other", "timestamp": 1},
|
|
event() | {"timestamp": True},
|
|
event() | {"user": {"user_id": True, "is_bot": False}},
|
|
event() | {"user": {"user_id": 100, "is_bot": True}},
|
|
event() | {"payload": {"wrong": "type"}},
|
|
{"update_type": "message_created", "timestamp": 1, "message": {}},
|
|
],
|
|
)
|
|
async def test_malformed_or_bot_events_are_ignored(bad):
|
|
await max_bot.handle_update(bad)
|
|
async with session.get_identity_session() as s:
|
|
assert not (await s.scalars(select(MaxReceipt))).all()
|
|
|
|
|
|
async def test_group_cannot_claim_a_patient_link():
|
|
async with session.get_identity_session() as s:
|
|
code, _ = await store.issue_code(s, TOKEN, clock.now(), 600)
|
|
update = event("message_created", "/start " + code)
|
|
update["message"]["recipient"]["chat_type"] = "chat"
|
|
await max_bot.handle_update(update)
|
|
async with session.get_identity_session() as s:
|
|
assert await s.get(MaxBinding, TOKEN) is None
|
|
assert await store.code_token(s, code, clock.now()) == TOKEN
|
|
|
|
|
|
async def test_429_respects_retry_deadline_and_then_succeeds(monkeypatch):
|
|
await max_bot.handle_update(event())
|
|
now = clock.now()
|
|
monkeypatch.setattr(clock, "now", lambda: now)
|
|
sender = AsyncMock()
|
|
sender.send.side_effect = MaxApiError("http_429", retryable=True, retry_after=120)
|
|
assert await max_bot.deliver_one(sender)
|
|
async with session.get_identity_session() as s:
|
|
row = (await s.scalars(select(MaxDelivery))).one()
|
|
assert row.state == "queued" and row.next_attempt_at == now + timedelta(seconds=120)
|
|
assert not await max_bot.deliver_one(sender)
|
|
now += timedelta(seconds=121)
|
|
sender.send.side_effect = None
|
|
assert await max_bot.deliver_one(sender)
|
|
async with session.get_identity_session() as s:
|
|
row = (await s.scalars(select(MaxDelivery))).one()
|
|
assert row.state == "sent" and row.attempts == 2
|
|
|
|
|
|
async def test_permanent_error_does_not_retry():
|
|
await max_bot.handle_update(event())
|
|
sender = AsyncMock()
|
|
sender.send.side_effect = MaxApiError("http_403")
|
|
await max_bot.deliver_one(sender)
|
|
assert not await max_bot.deliver_one(sender)
|
|
async with session.get_identity_session() as s:
|
|
assert (await s.scalars(select(MaxDelivery))).one().state == "failed"
|
|
|
|
|
|
@pytest.mark.parametrize("source", ["scale", "questionnaire"])
|
|
async def test_completed_test_produces_one_neutral_notification(source):
|
|
await connect()
|
|
sender = AsyncMock()
|
|
await max_bot.deliver_one(sender)
|
|
sender.reset_mock()
|
|
async with session.get_clinical_session() as s:
|
|
now = clock.now()
|
|
if source == "scale":
|
|
s.add(ScaleResult(token=TOKEN, ts=now, scale="phq9", total=1, severity="synthetic", complete=True))
|
|
else:
|
|
s.add(
|
|
QuestionnaireRecord(
|
|
id="synthetic-form",
|
|
token=TOKEN,
|
|
kind="intake",
|
|
version="test",
|
|
policy_version="test",
|
|
started_at=now,
|
|
completed_at=now,
|
|
day=now.date(),
|
|
status="complete",
|
|
answers={},
|
|
details={},
|
|
)
|
|
)
|
|
for _ in range(2):
|
|
async with session.get_identity_session() as s:
|
|
await store.collect_completed(s, clock.now())
|
|
assert await max_bot.deliver_one(sender)
|
|
assert not await max_bot.deliver_one(sender)
|
|
sender.send.assert_awaited_once_with(
|
|
100, max_bot.texts().completed, site_url="https://example.org", button=max_bot.texts().open_site
|
|
)
|
|
|
|
|
|
async def test_private_start_command_accepts_issued_code():
|
|
async with session.get_identity_session() as s:
|
|
code, _ = await store.issue_code(s, TOKEN, clock.now(), 600)
|
|
await max_bot.handle_update(event("message_created", "/start " + code))
|
|
async with session.get_identity_session() as s:
|
|
assert (await s.get(MaxBinding, TOKEN)).user_id == 100
|
|
|
|
|
|
async def test_disabled_lifecycle_never_constructs_client(monkeypatch):
|
|
monkeypatch.setenv("MAX_ENABLED", "false")
|
|
get_settings.cache_clear()
|
|
monkeypatch.setattr(max_bot, "MaxClient", lambda _: pytest.fail("disabled MAX must not create a network client"))
|
|
before = integrations.outbox
|
|
await max_bot.start()
|
|
await max_bot.handle_update(event())
|
|
await max_bot.stop()
|
|
assert integrations.outbox is before
|
|
|
|
|
|
async def test_lifecycle_wraps_once_restores_and_closes(monkeypatch):
|
|
sender = AsyncMock()
|
|
monkeypatch.setattr(max_bot, "MaxClient", lambda _: sender)
|
|
before = integrations.outbox
|
|
await max_bot.start()
|
|
first = integrations.outbox
|
|
await max_bot.start()
|
|
assert integrations.outbox is first and first.fallback is before
|
|
await max_bot.stop()
|
|
assert integrations.outbox is before
|
|
sender.close.assert_awaited_once()
|