"""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()