"""Reading a photographed Vietnamese medical document into confirmable turns. The point of this path: harm for limited-English-proficiency patients concentrates at medication reconciliation and discharge, which happen on paper. The vision model only transcribes -- the risk engine, not the model, decides whether a line is dangerous. """ from __future__ import annotations import io import json import os import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch REPO_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(REPO_ROOT / "scribe")) sys.path.insert(0, str(REPO_ROOT / "interpreter")) os.environ.setdefault("ASR_PROVIDER", "mock") os.environ.setdefault("ALLOW_MOCK_ASR", "true") os.environ.setdefault("LLM_PROVIDER", "offline") os.environ.setdefault("MEDICAL_LEXICON_PATH", str(REPO_ROOT / "data" / "medical_lexicon.json")) os.environ.setdefault("CAREPATH_ENV_FILE", "__tests_no_env__") # setdefault, never assignment: these modules share a process, and pinning a # mode here leaks into every other test module that imports the app. os.environ.setdefault("PROVIDER_MODE", "mock") _TMP_DB = Path(tempfile.mkdtemp(prefix="carepath_docs_")) / "docs.db" os.environ["DATABASE_URL"] = f"sqlite:///{_TMP_DB.as_posix()}" from fastapi.testclient import TestClient import app.config as interpreter_config import app.db as interpreter_db from app.providers.base import ProviderOutputError from app.providers.ckey import CKeyChatClient, parse_document_lines, read_document_lines from app.providers.registry import read_document from app.config import Settings from carepath.main import app, get_pipeline, get_settings PNG_1PX = ( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06" b"\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01\x00\x00\x05" b"\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82" ) class FakeResponse: def __init__(self, content: str) -> None: self._body = json.dumps({"choices": [{"message": {"content": content}}]}).encode("utf-8") def read(self) -> bytes: return self._body def __enter__(self) -> "FakeResponse": return self def __exit__(self, *exc: object) -> None: return None class RecordingOpener: def __init__(self, content: str) -> None: self.content = content self.requests: list[dict] = [] def __call__(self, request, timeout=None): self.requests.append(json.loads(request.data.decode("utf-8"))) return FakeResponse(self.content) class ParseTests(unittest.TestCase): def test_parses_lines(self) -> None: lines = parse_document_lines('{"lines": ["Amoxicillin 500mg", " ngày 2 lần "]}') self.assertEqual(lines, ["Amoxicillin 500mg", "ngày 2 lần"]) def test_unreadable_document_yields_nothing(self) -> None: """Fail closed: better zero lines than a guess at what a prescription says.""" self.assertEqual(parse_document_lines('{"lines": []}'), []) def test_accepts_fenced_json(self) -> None: self.assertEqual(parse_document_lines('```json\n{"lines": ["a"]}\n```'), ["a"]) def test_rejects_prose_or_extra_keys(self) -> None: for bad in [ '{"lines": ["a"], "summary": "a prescription"}', '{"text": "Amoxicillin"}', '{"lines": "Amoxicillin"}', '{"lines": [{"text": "a"}]}', "The document says Amoxicillin 500mg", ]: with self.assertRaises(ProviderOutputError, msg=bad): parse_document_lines(bad) class VisionRequestTests(unittest.TestCase): def test_image_is_sent_as_a_data_uri_alongside_the_prompt(self) -> None: opener = RecordingOpener('{"lines": ["Amoxicillin 500mg"]}') client = CKeyChatClient( api_key="sk-test", base_url="https://example.test/v1", model="gpt-5.4", opener=opener, backoff_seconds=0, ) lines = read_document_lines(client, PNG_1PX, "image/png") self.assertEqual(lines, ["Amoxicillin 500mg"]) content = opener.requests[0]["messages"][1]["content"] self.assertEqual(content[0]["type"], "text") self.assertEqual(content[1]["type"], "image_url") self.assertTrue(content[1]["image_url"]["url"].startswith("data:image/png;base64,")) system = opener.requests[0]["messages"][0]["content"] self.assertIn("Do not translate", system) class ModeTests(unittest.TestCase): def test_modes_without_vision_fail_closed(self) -> None: """Silence would read as 'the document was blank'. Raise instead.""" for mode in ("mock", "cloud"): with self.assertRaises(RuntimeError, msg=mode): read_document(Settings(provider_mode=mode), PNG_1PX, "image/png") def test_demo_mode_replays_a_scripted_prescription(self) -> None: lines = read_document(Settings(provider_mode="demo"), PNG_1PX, "image/png") self.assertTrue(any("Amoxicillin" in line for line in lines)) class EndpointTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: get_settings.cache_clear() get_pipeline.cache_clear() interpreter_config.get_settings.cache_clear() interpreter_db.set_engine(None) def _visit(self, client) -> str: return client.post("/api/sessions", json={"consent": {"ok": True}}).json()["session_id"] def _upload(self, client, visit_id, data=PNG_1PX, filename="donthuoc.png", ctype="image/png"): return client.post( f"/api/v1/visits/{visit_id}/documents", files={"image": (filename, io.BytesIO(data), ctype)}, ) def test_document_lines_become_gated_turns(self) -> None: prescription = read_document(Settings(provider_mode="demo"), PNG_1PX, "image/png") with TestClient(app) as client: visit_id = self._visit(client) # Patch the reader rather than pinning PROVIDER_MODE for the process. with patch("carepath.main.read_document", return_value=prescription): response = self._upload(client, visit_id) self.assertEqual(response.status_code, 200, response.text) turns = response.json() self.assertTrue(turns) self.assertTrue(all(turn["speaker"] == "document" for turn in turns)) # The risk engine, not the vision model, decides this is risky. dosed = [ turn for turn in turns if any(span["kind"] == "dose_number" for span in turn["risk_spans"]) ] self.assertTrue(dosed, "a prescription line with a dose should be detected") self.assertEqual(dosed[0]["status"], "awaiting_confirm") # They are ordinary turns, so they appear in the visit transcript. transcript = client.get(f"/api/sessions/{visit_id}/transcript").json() self.assertEqual(len(transcript), len(turns)) def test_sample_header_reads_the_scripted_document_whatever_was_uploaded(self) -> None: """The public demo's sample path, which the page labels as scripted. `demo` mode never looks at the bytes. That is honest for a sample and would be a fabrication for a visitor's own prescription, so the header exists to make the distinction explicit at the call site. """ with TestClient(app) as client: visit_id = self._visit(client) response = client.post( f"/api/v1/visits/{visit_id}/documents", files={"image": ("anything.png", io.BytesIO(PNG_1PX), "image/png")}, headers={"X-CarePath-Sample": "1"}, ) self.assertEqual(response.status_code, 200, response.text) turns = response.json() self.assertTrue(any("Amoxicillin" in turn["source_text"] for turn in turns)) def test_sample_mode_still_gates_the_risky_lines(self) -> None: """Scripted translation, real gate. If the demo showed a dose straight to the patient pane it would be advertising the opposite of the product. """ with TestClient(app) as client: visit_id = self._visit(client) turns = client.post( f"/api/v1/visits/{visit_id}/documents", files={"image": ("anything.png", io.BytesIO(PNG_1PX), "image/png")}, headers={"X-CarePath-Sample": "1"}, ).json() dosed = [ turn for turn in turns if any(span["kind"] == "dose_number" for span in turn["risk_spans"]) ] self.assertTrue(dosed, "the scripted prescription must carry a dose line") for turn in dosed: self.assertEqual(turn["status"], "awaiting_confirm") def test_without_the_header_the_configured_mode_still_applies(self) -> None: """The header opts in. It must not become the default by accident: this process runs PROVIDER_MODE=mock, which has no vision and fails closed. """ with TestClient(app) as client: visit_id = self._visit(client) self.assertEqual(self._upload(client, visit_id).status_code, 502) def test_rejects_a_non_image_upload(self) -> None: with TestClient(app) as client: visit_id = self._visit(client) response = self._upload( client, visit_id, data=b"not an image", filename="notes.txt", ctype="text/plain" ) self.assertEqual(response.status_code, 400) def test_rejects_an_empty_upload(self) -> None: with TestClient(app) as client: visit_id = self._visit(client) self.assertEqual(self._upload(client, visit_id, data=b"").status_code, 400) def test_unknown_visit_is_404(self) -> None: with TestClient(app) as client: self.assertEqual(self._upload(client, "nope").status_code, 404) def test_reader_failure_is_502_and_creates_no_turns(self) -> None: with TestClient(app) as client: visit_id = self._visit(client) with patch("carepath.main.read_document", side_effect=RuntimeError("gateway down")): response = self._upload(client, visit_id) self.assertEqual(response.status_code, 502) self.assertEqual(client.get(f"/api/sessions/{visit_id}/transcript").json(), []) def test_failure_message_does_not_leak_internals(self) -> None: with TestClient(app) as client: visit_id = self._visit(client) with patch( "carepath.main.read_document", side_effect=RuntimeError("sk-secret leaked and Tôi bị dị ứng"), ): response = self._upload(client, visit_id) self.assertNotIn("sk-secret", response.text) self.assertNotIn("dị ứng", response.text) if __name__ == "__main__": unittest.main()