from __future__ import annotations import contextlib import io import os import sys import unittest import wave from pathlib import Path from unittest.mock import patch REPO_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(REPO_ROOT / "scribe")) # Hermetic config: mock ASR + offline LLM, and do not read the real .env. 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__") from fastapi.testclient import TestClient import carepath.main as main_module from carepath.main import app, get_pipeline, get_settings def _silent_wav_bytes(seconds: int = 1) -> bytes: buffer = io.BytesIO() with contextlib.closing(wave.open(buffer, "wb")) as writer: writer.setnchannels(1) writer.setsampwidth(2) writer.setframerate(16000) writer.writeframes(b"\x00\x00" * 16000 * seconds) return buffer.getvalue() class UploadGuardrailTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: get_settings.cache_clear() get_pipeline.cache_clear() cls.client = TestClient(app) def setUp(self) -> None: get_settings.cache_clear() main_module._rate_limit_events.clear() main_module._global_rate_limit_events.clear() def _post_wav(self, headers: dict[str, str] | None = None): return self.client.post( "/api/v1/soap-notes", files={"audio": ("demo.wav", _silent_wav_bytes(), "audio/wav")}, headers=headers or {}, ) def test_valid_wav_is_accepted(self) -> None: response = self._post_wav() self.assertEqual(response.status_code, 200, response.text) self.assertTrue(response.json()["soap"]["review_required"]) def test_unsupported_file_type_is_rejected(self) -> None: response = self.client.post( "/api/v1/soap-notes", files={"audio": ("notes.txt", b"not audio", "text/plain")}, ) self.assertEqual(response.status_code, 400) self.assertIn("Unsupported file type", response.json()["detail"]) def test_oversized_upload_is_rejected(self) -> None: with patch.object(main_module, "MAX_UPLOAD_BYTES", 1024): response = self.client.post( "/api/v1/soap-notes", files={"audio": ("big.wav", b"\x00" * 4096, "audio/wav")}, ) self.assertEqual(response.status_code, 400) self.assertIn("too large", response.json()["detail"]) def test_startup_warms_pipeline(self) -> None: calls: list[bool] = [] class FakePipeline: def warmup(self) -> dict[str, object]: calls.append(True) return {"asr": {"ready": True}} with patch.object(main_module, "get_pipeline", return_value=FakePipeline()): main_module._warmup_pipeline() self.assertEqual(calls, [True]) def test_disallowed_origin_is_rejected_when_allowlist_set(self) -> None: with patch.dict( os.environ, {"CORS_ORIGINS": "https://carepath-omega.vercel.app"}, clear=False, ): get_settings.cache_clear() response = self.client.options( "/api/v1/soap-notes", headers={ "Origin": "https://example.invalid", "Access-Control-Request-Method": "POST", }, ) self.assertEqual(response.status_code, 400) self.assertIn("Disallowed CORS origin", response.text) def test_same_origin_is_allowed_when_cross_origin_allowlist_is_set(self) -> None: with patch.dict( os.environ, {"CORS_ORIGINS": "https://carepath-omega.vercel.app"}, clear=False, ): get_settings.cache_clear() response = self.client.get( "/api/v1/health", headers={ "Host": "carepath.example.test", "Origin": "https://carepath.example.test", }, ) self.assertEqual(response.status_code, 200, response.text) def test_fourth_request_from_same_ip_is_rate_limited(self) -> None: with patch.dict( os.environ, { "TEAM_CODE": "", "SOAP_RATE_LIMIT_PER_IP_HOUR": "3", "SOAP_RATE_LIMIT_PER_IP_DAY": "10", "SOAP_RATE_LIMIT_GLOBAL_DAY": "100", }, clear=False, ): get_settings.cache_clear() headers = {"X-Forwarded-For": "203.0.113.10"} responses = [self._post_wav(headers=headers) for _ in range(4)] self.assertEqual( [response.status_code for response in responses[:3]], [200, 200, 200], ) self.assertEqual(responses[3].status_code, 429) self.assertIn("Retry-After", responses[3].headers) self.assertIn("giới hạn demo", responses[3].json()["detail"]["message"]) def test_team_code_bypasses_limits(self) -> None: with patch.dict( os.environ, { "TEAM_CODE": "letmein", "SOAP_RATE_LIMIT_PER_IP_HOUR": "1", "SOAP_RATE_LIMIT_PER_IP_DAY": "1", "SOAP_RATE_LIMIT_GLOBAL_DAY": "1", }, clear=False, ): get_settings.cache_clear() headers = {"X-Forwarded-For": "203.0.113.11", "X-Team-Code": "letmein"} responses = [self._post_wav(headers=headers) for _ in range(3)] self.assertEqual( [response.status_code for response in responses], [200, 200, 200], ) def test_wrong_team_code_is_limited_like_anonymous(self) -> None: with patch.dict( os.environ, { "TEAM_CODE": "letmein", "SOAP_RATE_LIMIT_PER_IP_HOUR": "1", "SOAP_RATE_LIMIT_PER_IP_DAY": "10", "SOAP_RATE_LIMIT_GLOBAL_DAY": "100", }, clear=False, ): get_settings.cache_clear() headers = {"X-Forwarded-For": "203.0.113.12", "X-Team-Code": "wrong"} first = self._post_wav(headers=headers) second = self._post_wav(headers=headers) self.assertEqual(first.status_code, 200, first.text) self.assertEqual(second.status_code, 429) def test_global_daily_cap_limits_all_ips(self) -> None: with patch.dict( os.environ, { "TEAM_CODE": "", "SOAP_RATE_LIMIT_PER_IP_HOUR": "10", "SOAP_RATE_LIMIT_PER_IP_DAY": "10", "SOAP_RATE_LIMIT_GLOBAL_DAY": "2", }, clear=False, ): get_settings.cache_clear() first = self._post_wav(headers={"X-Forwarded-For": "203.0.113.13"}) second = self._post_wav(headers={"X-Forwarded-For": "203.0.113.14"}) third = self._post_wav(headers={"X-Forwarded-For": "203.0.113.15"}) self.assertEqual(first.status_code, 200, first.text) self.assertEqual(second.status_code, 200, second.text) self.assertEqual(third.status_code, 429) self.assertIn("giới hạn sử dụng trong ngày", third.json()["detail"]["message"]) if __name__ == "__main__": unittest.main()