Spaces:
Running
Running
File size: 7,011 Bytes
79b0bef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | """
tests/test_api.py
ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
Integration tests for the FastAPI application.
Uses FastAPI's TestClient (which runs requests in-process, no
server needed) with an in-memory SQLite database so tests are
fully isolated and fast.
ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient
from src.db.connection import _reset_for_testing, create_all_tables
@pytest.fixture(scope="module", autouse=True)
def test_database():
"""Set up an isolated in-memory SQLite database for the test module."""
_reset_for_testing("sqlite:///:memory:")
create_all_tables()
yield
_reset_for_testing("sqlite:///:memory:")
@pytest.fixture(scope="module")
def client():
"""Return a TestClient for the FastAPI app."""
from src.api.main import app
return TestClient(app, raise_server_exceptions=True)
# ββ Health check ββββββββββββββββββββββββββββββββββββββββββββββββββ
class TestHealth:
def test_health_returns_ok(self, client):
resp = client.get("/health")
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "ok"
assert "version" in data
assert "database" in data
def test_root_redirects_to_docs(self, client):
resp = client.get("/", follow_redirects=False)
assert resp.status_code in (301, 302, 307, 308)
assert "/docs" in resp.headers.get("location", "")
# ββ Notes list ββββββββββββββββββββββββββββββββββββββββββββββββββββ
class TestNotesList:
def test_list_notes_returns_paginated_response(self, client):
resp = client.get("/notes")
assert resp.status_code == 200
data = resp.json()
assert "total" in data
assert "items" in data
assert "limit" in data
assert "offset" in data
def test_list_notes_default_pagination(self, client):
resp = client.get("/notes")
data = resp.json()
assert data["limit"] == 50
assert data["offset"] == 0
def test_list_notes_invalid_limit_rejected(self, client):
resp = client.get("/notes?limit=0")
assert resp.status_code == 422
def test_note_not_found_returns_404(self, client):
resp = client.get("/notes/99999")
assert resp.status_code == 404
# ββ Note analysis βββββββββββββββββββββββββββββββββββββββββββββββββ
class TestAnalyse:
def _mock_ner_pipeline(self):
from src.nlp.ner import Entity
mock = MagicMock()
mock.extract.return_value = [
Entity(text="hypertension", label="DISEASE",
start=10, end=22, confidence=None),
Entity(text="metformin", label="MEDICATION",
start=30, end=39, confidence=None),
]
return mock
def test_analyse_returns_expected_shape(self, client):
with patch("src.api.routes.notes._get_ner",
return_value=self._mock_ner_pipeline()), patch("src.api.routes.notes._get_mapper", return_value=None), patch("src.api.routes.notes._get_classifier", return_value=None):
resp = client.post("/notes/analyse", json={
"text": "Patient has hypertension and takes metformin.",
"include_icd10": False,
"include_severity": False,
})
assert resp.status_code == 200
data = resp.json()
assert "entities" in data
assert "entity_counts" in data
assert "word_count" in data
assert "text_length" in data
def test_analyse_short_text_rejected(self, client):
resp = client.post("/notes/analyse", json={"text": "hi"})
assert resp.status_code == 422
def test_analyse_blank_text_rejected(self, client):
resp = client.post("/notes/analyse", json={"text": " "})
assert resp.status_code == 422
def test_analyse_entity_counts_match_entities(self, client):
with patch("src.api.routes.notes._get_ner",
return_value=self._mock_ner_pipeline()), patch("src.api.routes.notes._get_mapper", return_value=None), patch("src.api.routes.notes._get_classifier", return_value=None):
resp = client.post("/notes/analyse", json={
"text": "Patient has hypertension and takes metformin daily.",
"include_icd10": False,
"include_severity": False,
})
data = resp.json()
counts = data["entity_counts"]
total = sum(counts.values())
assert total == len(data["entities"])
# ββ ICD-10 lookup βββββββββββββββββββββββββββββββββββββββββββββββββ
class TestICD10:
def test_icd_lookup_missing_mapper_returns_empty(self, client):
with patch("src.api.routes.icd._get_mapper", return_value=None):
resp = client.post("/icd/lookup", json={
"text": "hypertension", "top_k": 3
})
assert resp.status_code == 200
assert resp.json()["matches"] == []
def test_icd_lookup_short_text_rejected(self, client):
resp = client.post("/icd/lookup", json={"text": "x"})
assert resp.status_code == 422
def test_icd_top_returns_list(self, client):
resp = client.get("/icd/top")
assert resp.status_code == 200
assert isinstance(resp.json(), list)
# ββ Entities ββββββββββββββββββββββββββββββββββββββββββββββββββββββ
class TestEntities:
def test_top_entities_returns_expected_shape(self, client):
resp = client.get("/entities/top")
assert resp.status_code == 200
data = resp.json()
assert "items" in data
assert "limit" in data
def test_top_entities_label_filter_accepted(self, client):
resp = client.get("/entities/top?label=DISEASE&limit=10")
assert resp.status_code == 200
def test_cooccurrence_returns_list(self, client):
resp = client.get("/entities/cooccurrence")
assert resp.status_code == 200
assert isinstance(resp.json(), list)
|