| """End-to-end smoke tests for the BioPatch API. |
| |
| Uses an in-memory SQLite shared across connections so the TestClient and the |
| WebSocket use the same data. The simulator and auto-seeder are disabled here |
| so we can drive the system deterministically. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import os |
| import sys |
| from pathlib import Path |
|
|
| os.environ["DATABASE_URL"] = "sqlite+pysqlite:///file:biopatch_test?mode=memory&cache=shared&uri=true" |
| os.environ["ENABLE_SIMULATOR"] = "false" |
| os.environ["AUTO_SEED"] = "false" |
| os.environ["JWT_SECRET"] = "test-secret-for-pytest-only-not-for-prod-use" |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) |
|
|
| import pytest |
| from fastapi.testclient import TestClient |
|
|
| from core.database import init_db |
| from main import app |
|
|
|
|
| @pytest.fixture(scope="module") |
| def client(): |
| init_db() |
| with TestClient(app) as c: |
| yield c |
|
|
|
|
| def _register_and_login(client: TestClient, role: str = "admin") -> str: |
| email = f"{role}@biopatch.io" |
| client.post( |
| "/api/auth/register", |
| json={ |
| "email": email, |
| "full_name": f"Test {role.title()}", |
| "password": "TestPassword!1", |
| "role": role, |
| }, |
| ) |
| resp = client.post( |
| "/api/auth/login/json", |
| json={"email": email, "password": "TestPassword!1"}, |
| ) |
| assert resp.status_code == 200, resp.text |
| return resp.json()["access_token"] |
|
|
|
|
| def test_health(client: TestClient) -> None: |
| r = client.get("/api/health") |
| assert r.status_code == 200 |
| assert r.json()["status"] == "ok" |
|
|
|
|
| def test_auth_required_on_patients(client: TestClient) -> None: |
| r = client.get("/api/patients") |
| assert r.status_code == 401 |
|
|
|
|
| def test_full_patient_flow(client: TestClient) -> None: |
| token = _register_and_login(client, role="admin") |
| h = {"Authorization": f"Bearer {token}"} |
|
|
| |
| r = client.post( |
| "/api/patients", |
| json={ |
| "mrn": "MRN-TEST-1", |
| "full_name": "Test Patient", |
| "age": 60, |
| "sex": "M", |
| "ward": "TestWard", |
| "bed": "1", |
| "surgery_type": "Appendectomy", |
| "surgery_date": "2025-05-10T08:00:00Z", |
| }, |
| headers=h, |
| ) |
| assert r.status_code == 201, r.text |
| patient_id = r.json()["id"] |
|
|
| |
| r = client.post( |
| f"/api/patients/{patient_id}/sensors", |
| json={"device_id": "BP-TEST-001", "location": "incision"}, |
| headers=h, |
| ) |
| assert r.status_code == 201, r.text |
| sensor_id = r.json()["id"] |
|
|
| |
| r = client.post( |
| f"/api/patients/{patient_id}/vitals", |
| json={"temperature_c": 38.6, "ph": 8.1, "humidity": 88.0, "sensor_id": sensor_id}, |
| headers=h, |
| ) |
| assert r.status_code == 201, r.text |
| vital = r.json() |
| assert vital["severity"] == "critical" |
| assert vital["sepsis_risk"] >= 75 |
| assert vital["bacterial_signature"] is not None |
|
|
| |
| r = client.get("/api/patients", headers=h) |
| assert r.status_code == 200 |
| items = r.json() |
| target = next(p for p in items if p["id"] == patient_id) |
| assert target["latest_severity"] == "critical" |
| assert target["online_sensor_count"] == 1 |
|
|
| |
| r = client.get(f"/api/patients/{patient_id}/vitals?hours=1", headers=h) |
| assert r.status_code == 200 |
| assert len(r.json()) == 1 |
|
|
| |
| r = client.post( |
| f"/api/patients/{patient_id}/notes", |
| json={"body": "Started broad-spectrum antibiotics.", "category": "intervention", "severity": "warning"}, |
| headers=h, |
| ) |
| assert r.status_code == 201, r.text |
| assert r.json()["author_name"] == "Test Admin" |
|
|
| r = client.get(f"/api/patients/{patient_id}/notes", headers=h) |
| assert r.status_code == 200 |
| assert len(r.json()) == 1 |
|
|
|
|
| def test_role_enforcement_for_patient_create(client: TestClient) -> None: |
| token = _register_and_login(client, role="nurse") |
| h = {"Authorization": f"Bearer {token}"} |
| r = client.post( |
| "/api/patients", |
| json={ |
| "mrn": "MRN-TEST-NURSE", |
| "full_name": "Should Fail", |
| "age": 50, |
| "sex": "F", |
| "ward": "X", |
| "bed": "1", |
| "surgery_type": "X", |
| "surgery_date": "2025-05-10T08:00:00Z", |
| }, |
| headers=h, |
| ) |
| assert r.status_code == 403 |
|
|
|
|
| def test_websocket_snapshot_and_broadcast(client: TestClient) -> None: |
| token = _register_and_login(client, role="admin") |
|
|
| |
| with client.websocket_connect("/ws/stream") as ws: |
| msg = ws.receive_json() |
| assert msg["type"] == "error" |
|
|
| with client.websocket_connect(f"/ws/stream?token={token}") as ws: |
| snap = ws.receive_json() |
| assert snap["type"] == "snapshot" |
| assert "patients" in snap["data"] |
|
|
| |
| h = {"Authorization": f"Bearer {token}"} |
| patients = client.get("/api/patients", headers=h).json() |
| assert patients, "expected at least one patient from previous test" |
| pid = patients[0]["id"] |
| r = client.post( |
| f"/api/patients/{pid}/vitals", |
| json={"temperature_c": 34.0, "ph": 6.0, "humidity": 75.0}, |
| headers=h, |
| ) |
| assert r.status_code == 201 |
|
|
| seen_types: set[str] = set() |
| for _ in range(5): |
| event = ws.receive_json() |
| seen_types.add(event["type"]) |
| if "vitals" in seen_types: |
| break |
| assert "vitals" in seen_types |
|
|
