from __future__ import annotations import json import unittest from copy import deepcopy from types import SimpleNamespace from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from app.auth import get_current_user from app.routes import rips from test.fixtures.rips_cases import base_rips_options class FakeRipsService: def __init__(self) -> None: self.download_calls = 0 self.generated_payload = { "case_key": "case-1", "version": 1, "stored": True, "regenerated": False, "source": "generated", "export_status": "ready", "export_message": "El JSON puede pasar a revisión final.", "payload": {"numFactura": "FV1001", "usuarios": []}, "validation_report": {"status": "approved", "ready_for_cuv": True, "findings": []}, "field_trace_map": [], } self.template_payload = { "scope": { "level": "provider_payer", "provider_nit": "900123456", "payer_nit": "800999111", }, "options": deepcopy(base_rips_options()), "version": 3, "stored": True, "source": "case_scope", "resolved_from_level": "provider_payer", } def generate_case_rips(self, username: str, case_key: str, *, options, regen: bool, persist: bool): return SimpleNamespace(model_dump=lambda **_: deepcopy(self.generated_payload)) def get_latest_case_rips(self, username: str, case_key: str): if case_key == "missing": return None return SimpleNamespace(model_dump=lambda **_: deepcopy(self.generated_payload)) def download_case_rips(self, username: str, case_key: str): self.download_calls += 1 if case_key == "missing": return None return SimpleNamespace(model_dump=lambda **_: deepcopy(self.generated_payload)) def get_case_template(self, username: str, case_key: str): return SimpleNamespace(model_dump=lambda **_: deepcopy(self.template_payload)) def save_case_template(self, username: str, case_key: str, *, options, scope_level: str): payload = deepcopy(self.template_payload) payload["scope"]["level"] = scope_level return SimpleNamespace(model_dump=lambda **_: payload) class FakeDemoIdentityService: enabled = True def resolve_case_key(self, *, username: str, visible_case_key: str) -> str: if visible_case_key == "00001-paciente-01-caso-00001": return "case-1" return visible_case_key def project_case_identity( self, *, username: str, case_key: str, case_number: str = "", patient_id: str = "", patient_name: str = "", ) -> dict[str, str]: if case_key == "case-1": return { "case_key": "00001-paciente-01-caso-00001", "case_number": case_number, "patient_id": "00001", "patient_name": "paciente-01", "nombre_paciente": "paciente-01", } return { "case_key": case_key, "case_number": case_number, "patient_id": patient_id, "patient_name": patient_name, "nombre_paciente": patient_name, } class RipsRoutesTest(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.fake_services = SimpleNamespace(rips_service=FakeRipsService()) self.app = FastAPI() self.app.include_router(rips.router) self.app.state.services = self.fake_services async def _override_user(): return SimpleNamespace(username="tester") self.app.dependency_overrides[get_current_user] = _override_user self.transport = ASGITransport(app=self.app) def tearDown(self) -> None: self.app.dependency_overrides.clear() async def test_post_generates_case_rips(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/rips/cases/case-1", json={"options": base_rips_options(), "regen": False, "persist": True}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["case_key"], "case-1") self.assertEqual(response.json()["validation_report"]["status"], "approved") self.assertEqual(response.json()["export_status"], "ready") async def test_post_ignores_legacy_placeholder_catalog_rows(self) -> None: options = base_rips_options() options["medication_catalog"] = [ { "tipo_medicamento": "01", "unidad_medida": 1, "unidad_min_dispensa": 1, "dias_tratamiento": 1, "concentracion_medicamento": 0, "concepto_recaudo": "02", } ] options["other_service_catalog"] = [ { "source_section": "honorarios_medicos", "tipo_os": "04", } ] async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/rips/cases/case-1", json={"options": options, "regen": False, "persist": True}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["case_key"], "case-1") async def test_post_returns_400_for_invalid_non_placeholder_catalog_rows(self) -> None: options = base_rips_options() options["other_service_catalog"] = [ { "source_section": "honorarios_medicos", "nom_tecnologia_salud": "Honorarios cirujano", } ] async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/rips/cases/case-1", json={"options": options, "regen": False, "persist": True}, ) self.assertEqual(response.status_code, 400) self.assertEqual(response.json()["detail"][0]["loc"][-1], "match_key") async def test_get_returns_404_when_case_rips_is_missing(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/rips/cases/missing") self.assertEqual(response.status_code, 404) async def test_download_returns_json_attachment(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/rips/cases/case-1/download") self.assertEqual(response.status_code, 200) self.assertEqual(response.headers["content-type"], "application/json") self.assertIn("attachment; filename=", response.headers["content-disposition"]) payload = json.loads(response.text) self.assertEqual(payload["numFactura"], "FV1001") self.assertNotIn("case_key", payload) self.assertNotIn("validation_report", payload) self.assertEqual(self.fake_services.rips_service.download_calls, 1) async def test_get_template_returns_scope_and_options(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/rips/cases/case-1/template") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["scope"]["level"], "provider_payer") self.assertIn("user", response.json()["options"]) async def test_post_template_persists_scope_level(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/rips/cases/case-1/template", json={"options": base_rips_options(), "scope_level": "provider"}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["scope"]["level"], "provider")