104 lines
4.7 KiB
Python
104 lines
4.7 KiB
Python
"""MAX endpoints authenticate before reading bindings or accepting events."""
|
|
|
|
import sys
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
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.integrations import max_store
|
|
from app.main import app
|
|
|
|
BASE = "/api/v1/me/notifications/max"
|
|
HOOK = "/api/v1/webhooks/max"
|
|
TOKEN = "synthetic-patient"
|
|
AUTH = {"Authorization": f"Bearer {TOKEN}"}
|
|
SECRET = "test-secret-" * 4
|
|
|
|
|
|
@pytest.fixture
|
|
async def client(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" + HOOK,
|
|
"MAX_SITE_URL": "https://example.org",
|
|
}.items():
|
|
monkeypatch.setenv(key, value)
|
|
get_settings.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()
|
|
)
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
|
yield c
|
|
await session.configure()
|
|
get_settings.cache_clear()
|
|
|
|
|
|
@pytest.mark.parametrize("auth", [{}, {"Authorization": "Bearer unknown"}, {"Authorization": "Basic test"}])
|
|
async def test_auth_required(client, auth):
|
|
assert (await client.get(BASE, headers=auth)).status_code == 401
|
|
assert (await client.post(BASE + "/link", headers=auth, json={"consent": True})).status_code == 401
|
|
assert (await client.delete(BASE, headers=auth)).status_code == 401
|
|
|
|
|
|
async def test_link_requires_consent_and_hides_patient_credential(client):
|
|
for body in ({}, {"consent": False}, {"consent": True, "user_id": 100}):
|
|
assert (await client.post(BASE + "/link", headers=AUTH, json=body)).status_code == 422
|
|
response = await client.post(BASE + "/link", headers=AUTH, json={"consent": True})
|
|
assert response.status_code == 200
|
|
assert TOKEN not in response.text
|
|
url = urlsplit(response.json()["url"])
|
|
assert url.netloc == "max.ru" and url.path == "/test_bot"
|
|
code = parse_qs(url.query)["start"][0]
|
|
async with session.get_identity_session() as s:
|
|
assert await max_store.bind(s, code, 100, clock.now()) == TOKEN
|
|
assert (await client.get(BASE, headers=AUTH)).json()["connected"] is True
|
|
assert (await client.delete(BASE, headers=AUTH)).json()["connected"] is False
|
|
|
|
|
|
async def test_revoke_consent_prevents_linking(client):
|
|
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()
|
|
)
|
|
assert (await client.post(BASE + "/link", headers=AUTH, json={"consent": True})).status_code == 403
|
|
|
|
|
|
async def test_disabled_service_still_allows_disconnect(client, monkeypatch):
|
|
monkeypatch.setenv("MAX_ENABLED", "false")
|
|
get_settings.cache_clear()
|
|
assert (await client.get(BASE, headers=AUTH)).json()["available"] is False
|
|
assert (await client.post(BASE + "/link", headers=AUTH, json={"consent": True})).status_code == 503
|
|
assert (await client.post(HOOK, json={})).status_code == 503
|
|
assert (await client.delete(BASE, headers=AUTH)).status_code == 200
|
|
|
|
|
|
async def test_webhook_secret_checked_before_body_or_adapter(client):
|
|
assert (await client.post(HOOK, content="not-json")).status_code == 403
|
|
assert (await client.post(HOOK, headers={"X-Max-Bot-Api-Secret": "wrong"}, json={})).status_code == 403
|
|
headers = {"X-Max-Bot-Api-Secret": SECRET}
|
|
assert (await client.post(HOOK, headers=headers, content="not-json")).status_code == 400
|
|
assert (await client.post(HOOK, headers=headers, json=[])).status_code == 422
|
|
assert (await client.post(HOOK, headers=headers, content=" " * 65537)).status_code == 413
|
|
|
|
|
|
async def test_webhook_storage_failure_is_retryable_without_identity_details(client, monkeypatch):
|
|
handler = AsyncMock(side_effect=OperationalError("private SQL", {"token": TOKEN}, Exception("private")))
|
|
monkeypatch.setitem(sys.modules, "app.channels.max_bot", SimpleNamespace(handle_update=handler))
|
|
response = await client.post(HOOK, headers={"X-Max-Bot-Api-Secret": SECRET}, json={})
|
|
assert response.status_code == 503 and TOKEN not in response.text and "private" not in response.text
|