MCP FHIR Server

""" Comprehensive tests for FHIR MCP Server. These tests validate: 1. FHIR resource validation before server submission 2. Read/write permission enforcement (security) 3. Correct HTTP endpoint routing based on resourceType 4. Error handling and actionable error messages for LLM 5. Authentication header inclusion 6. Create vs Update logic (POST vs PUT) """ import pytest import respx import httpx from unittest.mock import Mock, patch import json # Import after patching to control config import server @pytest.fixture def valid_patient_resource(): """Sample valid Patient resource for testing.""" return { "resourceType": "Patient", "name": [{"family": "Smith", "given": ["John"]}], "gender": "male", "birthDate": "1980-01-01", } @pytest.fixture def valid_patient_with_id(): """Sample valid Patient resource with ID for update testing.""" return { "resourceType": "Patient", "id": "patient-123", "name": [{"family": "Smith", "given": ["John"]}], "gender": "male", } class TestWriteFHIRResource: """Tests for write_fhir_resource tool.""" @pytest.mark.asyncio @respx.mock async def test_create_valid_resource_without_id(self, valid_patient_resource): """Test creating a new resource (POST) when no ID is provided.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): mock_response = valid_patient_resource.copy() mock_response["id"] = "new-patient-123" route = respx.post( "http://test-fhir.com/fhir/Patient" ).mock(return_value=httpx.Response(201, json=mock_response)) result = await server.write_fhir_resource.fn( valid_patient_resource ) assert route.called assert "Success" in result assert "new-patient-123" in result @pytest.mark.asyncio @respx.mock async def test_update_valid_resource_with_id(self, valid_patient_with_id): """Test updating an existing resource (PUT) when ID is provided.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): route = respx.put( "http://test-fhir.com/fhir/Patient/patient-123" ).mock( return_value=httpx.Response( 200, json=valid_patient_with_id ) ) result = await server.write_fhir_resource.fn( valid_patient_with_id ) assert route.called assert "Success" in result assert "patient-123" in result @pytest.mark.asyncio async def test_invalid_resource_validation_fails(self, valid_patient_resource): """Test that invalid resources are rejected with validation errors.""" error_msg = "Missing required field: identifier" with patch("server.validate_fhir_resource", return_value=(False, error_msg)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): result = await server.write_fhir_resource.fn(valid_patient_resource) assert "Validation failed" in result assert "Missing required field: identifier" in result @pytest.mark.asyncio async def test_missing_resource_type(self): """Test error when resourceType is missing.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): resource = {"name": [{"family": "Smith"}]} # No resourceType result = await server.write_fhir_resource.fn(resource) assert "Error" in result assert "resourceType" in result @pytest.mark.asyncio async def test_write_permission_denied(self, valid_patient_resource): """Test that write operations are blocked when FHIR_ALLOW_WRITE is false.""" with patch.object(server, "FHIR_ALLOW_WRITE", False): result = await server.write_fhir_resource.fn(valid_patient_resource) assert "Error" in result assert "not allowed" in result.lower() @pytest.mark.asyncio @respx.mock async def test_read_permission_hides_response_body(self, valid_patient_resource): """Test that response body is hidden when FHIR_ALLOW_READ is false.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object(server, "FHIR_ALLOW_READ", False): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): mock_response = valid_patient_resource.copy() mock_response["id"] = "secret-123" respx.post("http://test-fhir.com/fhir/Patient").mock( return_value=httpx.Response(201, json=mock_response) ) result = await server.write_fhir_resource.fn( valid_patient_resource ) assert "Success" in result # Should NOT contain the full resource data assert '"resourceType"' not in result @pytest.mark.asyncio @respx.mock async def test_server_validation_error_400(self, valid_patient_resource): """Test handling of server validation errors (400).""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): error_response = { "resourceType": "OperationOutcome", "issue": [ { "severity": "error", "code": "invalid", "diagnostics": "Invalid birthDate format", } ], } respx.post("http://test-fhir.com/fhir/Patient").mock( return_value=httpx.Response(400, json=error_response) ) result = await server.write_fhir_resource.fn( valid_patient_resource ) assert "400" in result assert "Invalid birthDate format" in result @pytest.mark.asyncio @respx.mock async def test_authentication_token_included(self, valid_patient_resource): """Test that auth token is included in request headers.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object(server, "FHIR_ALLOW_READ", False): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): with patch.object( server, "FHIR_AUTH_TOKEN", "test-token-123" ): route = respx.post( "http://test-fhir.com/fhir/Patient" ).mock( return_value=httpx.Response(201, json={"id": "123"}) ) await server.write_fhir_resource.fn( valid_patient_resource ) assert route.called request = route.calls.last.request assert "Authorization" in request.headers assert ( request.headers["Authorization"] == "Bearer test-token-123" ) @pytest.mark.asyncio @respx.mock async def test_create_resource_with_builder_id(self, valid_patient_resource): """Test creating a resource with builder_id parameter.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): mock_response = valid_patient_resource.copy() mock_response["id"] = "new-patient-123" route = respx.post( "http://test-fhir.com/fhir/Patient" ).mock(return_value=httpx.Response(201, json=mock_response)) result = await server.write_fhir_resource.fn( valid_patient_resource, {"Zus-Account": "builder-123"} ) assert route.called assert "Success" in result assert "new-patient-123" in result # Verify that the Zus-Account header is set correctly request = route.calls.last.request assert request.headers["Zus-Account"] == "builder-123" # Verify that the request body does NOT include builderID request_body = json.loads(request.content) assert "builderID" not in request_body @pytest.mark.asyncio @respx.mock async def test_create_resource_without_builder_id(self, valid_patient_resource): """Test creating a resource without builder_id parameter (should not include Zus-Account header).""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): mock_response = valid_patient_resource.copy() mock_response["id"] = "new-patient-123" route = respx.post( "http://test-fhir.com/fhir/Patient" ).mock(return_value=httpx.Response(201, json=mock_response)) result = await server.write_fhir_resource.fn( valid_patient_resource ) assert route.called assert "Success" in result assert "new-patient-123" in result # Verify that the Zus-Account header is NOT set request = route.calls.last.request assert "Zus-Account" not in request.headers @pytest.mark.asyncio @respx.mock async def test_update_resource_with_builder_id(self, valid_patient_with_id): """Test updating a resource with builder_id parameter.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): route = respx.put( "http://test-fhir.com/fhir/Patient/patient-123" ).mock( return_value=httpx.Response(200, json=valid_patient_with_id) ) result = await server.write_fhir_resource.fn( valid_patient_with_id, {"Zus-Account": "builder-456"} ) assert route.called assert "Success" in result assert "patient-123" in result # Verify that the Zus-Account header is set correctly request = route.calls.last.request assert request.headers["Zus-Account"] == "builder-456" # Verify that the request body does NOT include builderID request_body = json.loads(request.content) assert "builderID" not in request_body @pytest.mark.asyncio @respx.mock async def test_create_observation_with_builder_id(self): """Test creating an Observation resource with builder_id parameter.""" observation_resource = { "resourceType": "Observation", "status": "final", "code": { "coding": [{"system": "http://loinc.org", "code": "33747-0", "display": "Temperature"}] }, "valueQuantity": { "value": 98.6, "unit": "F", "system": "http://unitsofmeasure.org", "code": "[degF]" } } with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): mock_response = observation_resource.copy() mock_response["id"] = "obs-123" route = respx.post( "http://test-fhir.com/fhir/Observation" ).mock(return_value=httpx.Response(201, json=mock_response)) result = await server.write_fhir_resource.fn( observation_resource, {"Zus-Account": "builder-789"} ) assert route.called assert "Success" in result assert "obs-123" in result # Verify that the Zus-Account header is set correctly request = route.calls.last.request assert request.headers["Zus-Account"] == "builder-789" # Verify that the request body does NOT include builderID request_body = json.loads(request.content) assert "builderID" not in request_body @pytest.mark.asyncio async def test_builder_id_does_not_modify_original_resource(self, valid_patient_resource): """Test that providing builder_id does not modify the original resource object.""" original_resource = valid_patient_resource.copy() with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): with patch.object(server, "FHIR_ALLOW_READ", False): with patch.object( server, "FHIR_BASE_URL", "http://test-fhir.com/fhir" ): with respx.mock: respx.post("http://test-fhir.com/fhir/Patient").mock( return_value=httpx.Response(201, json={"id": "123"}) ) await server.write_fhir_resource.fn( valid_patient_resource, {"Zus-Account": "builder-123"} ) # Verify original resource is unchanged assert "builderID" not in original_resource assert original_resource == valid_patient_resource class TestReadFHIRResource: """Tests for read_fhir_resource tool.""" @pytest.mark.asyncio @respx.mock async def test_successful_read(self, valid_patient_with_id): """Test successfully reading a resource.""" with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "FHIR_BASE_URL", "http://test-fhir.com/fhir"): route = respx.get("http://test-fhir.com/fhir/Patient/patient-123").mock( return_value=httpx.Response(200, json=valid_patient_with_id) ) result = await server.read_fhir_resource.fn("Patient", "patient-123") assert route.called assert "Patient" in result assert "patient-123" in result @pytest.mark.asyncio async def test_read_permission_denied(self): """Test that read operations are blocked when FHIR_ALLOW_READ is false.""" with patch.object(server, "FHIR_ALLOW_READ", False): result = await server.read_fhir_resource.fn("Patient", "123") assert "Error" in result assert "not allowed" in result.lower() @pytest.mark.asyncio @respx.mock async def test_resource_not_found_404(self): """Test 404 error when resource doesn't exist.""" with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "FHIR_BASE_URL", "http://test-fhir.com/fhir"): respx.get("http://test-fhir.com/fhir/Patient/nonexistent").mock( return_value=httpx.Response(404, text="Not found") ) result = await server.read_fhir_resource.fn("Patient", "nonexistent") assert "Error" in result assert "not found" in result.lower() @pytest.mark.asyncio @respx.mock async def test_read_resource_with_builder_id(self, valid_patient_with_id): """Test reading a resource with builder_id parameter.""" with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "FHIR_BASE_URL", "http://test-fhir.com/fhir"): route = respx.get("http://test-fhir.com/fhir/Patient/patient-123").mock( return_value=httpx.Response(200, json=valid_patient_with_id) ) result = await server.read_fhir_resource.fn("Patient", "patient-123", {"Zus-Account": "builder-123"}) assert route.called assert "Patient" in result assert "patient-123" in result # Verify that the Zus-Account header is set correctly request = route.calls.last.request assert request.headers["Zus-Account"] == "builder-123" @pytest.mark.asyncio @respx.mock async def test_read_resource_without_builder_id(self, valid_patient_with_id): """Test reading a resource without builder_id parameter.""" with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "FHIR_BASE_URL", "http://test-fhir.com/fhir"): route = respx.get("http://test-fhir.com/fhir/Patient/patient-123").mock( return_value=httpx.Response(200, json=valid_patient_with_id) ) result = await server.read_fhir_resource.fn("Patient", "patient-123") assert route.called assert "Patient" in result assert "patient-123" in result # Verify that the Zus-Account header is NOT set request = route.calls.last.request assert "Zus-Account" not in request.headers class TestSearchFHIRResources: """Tests for search_fhir_resources tool.""" @pytest.mark.asyncio @respx.mock async def test_search_without_params(self): """Test searching without parameters returns all resources.""" with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "FHIR_BASE_URL", "http://test-fhir.com/fhir"): bundle = { "resourceType": "Bundle", "type": "searchset", "total": 2, "entry": [ {"resource": {"resourceType": "Patient", "id": "1"}}, {"resource": {"resourceType": "Patient", "id": "2"}}, ], } route = respx.get("http://test-fhir.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle) ) result = await server.search_fhir_resources.fn("Patient") assert route.called assert "Found 2 Patient" in result @pytest.mark.asyncio @respx.mock async def test_search_with_params(self): """Test searching with query parameters.""" with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "FHIR_BASE_URL", "http://test-fhir.com/fhir"): bundle = { "resourceType": "Bundle", "type": "searchset", "total": 1, "entry": [{"resource": {"resourceType": "Patient", "id": "1"}}], } route = respx.get("http://test-fhir.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle) ) params = {"name": "Smith", "gender": "female"} result = await server.search_fhir_resources.fn("Patient", params) assert route.called request = route.calls.last.request assert "name" in str(request.url) assert "Smith" in str(request.url) @pytest.mark.asyncio async def test_search_permission_denied(self): """Test that search operations are blocked when FHIR_ALLOW_READ is false.""" with patch.object(server, "FHIR_ALLOW_READ", False): result = await server.search_fhir_resources.fn("Patient") assert "Error" in result assert "not allowed" in result.lower() @pytest.mark.asyncio @respx.mock async def test_search_with_builder_id(self): """Test searching with builder_id parameter.""" with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "FHIR_BASE_URL", "http://test-fhir.com/fhir"): bundle = { "resourceType": "Bundle", "type": "searchset", "total": 1, "entry": [{"resource": {"resourceType": "Patient", "id": "1"}}], } route = respx.get("http://test-fhir.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle) ) result = await server.search_fhir_resources.fn("Patient", None, {"Zus-Account": "builder-123"}) assert route.called assert "Found 1 Patient" in result # Verify that the Zus-Account header is set correctly request = route.calls.last.request assert request.headers["Zus-Account"] == "builder-123" @pytest.mark.asyncio @respx.mock async def test_search_without_builder_id(self): """Test searching without builder_id parameter.""" with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "FHIR_BASE_URL", "http://test-fhir.com/fhir"): bundle = { "resourceType": "Bundle", "type": "searchset", "total": 1, "entry": [{"resource": {"resourceType": "Patient", "id": "1"}}], } route = respx.get("http://test-fhir.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle) ) result = await server.search_fhir_resources.fn("Patient") assert route.called assert "Found 1 Patient" in result # Verify that the Zus-Account header is NOT set request = route.calls.last.request assert "Zus-Account" not in request.headers class TestGetFHIRConfig: """Tests for get_fhir_config tool.""" @pytest.mark.asyncio async def test_config_display(self): """Test that configuration is correctly displayed.""" with patch.object(server, "FHIR_BASE_URL", "http://custom-fhir.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "FHIR_ALLOW_WRITE", False): with patch.object(server, "FHIR_AUTH_TOKEN", "secret-token"): with patch.object(server, "_allowed_methods_set", None): result = await server.get_fhir_config.fn() assert "http://custom-fhir.com/fhir" in result assert "Read Allowed: True" in result assert "Write Allowed: False" in result assert "Configured" in result @pytest.mark.asyncio async def test_config_with_allowed_methods(self): """Test configuration display with HTTP methods configured.""" with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "_allowed_methods_set", {"GET", "POST"}): result = await server.get_fhir_config.fn() assert "GET, POST" in result assert "overrides" in result.lower() class TestHTTPMethodsConfiguration: """Tests for HTTP methods configuration (FHIR_ALLOWED_METHODS).""" @pytest.mark.asyncio async def test_allowed_methods_overrides_write_permission( self, valid_patient_resource ): """Test that FHIR_ALLOWED_METHODS takes precedence over FHIR_ALLOW_WRITE.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", True): # Write is allowed with patch.object( server, "_allowed_methods_set", {"GET"} ): # But only GET in methods result = await server.write_fhir_resource.fn( valid_patient_resource ) assert "Error" in result assert "POST is not allowed" in result assert "GET" in result @pytest.mark.asyncio async def test_allowed_methods_permits_specific_write_method( self, valid_patient_resource ): """Test that specific write methods can be allowed via FHIR_ALLOWED_METHODS.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object(server, "FHIR_ALLOW_WRITE", False): # Write disabled with patch.object( server, "_allowed_methods_set", {"POST"} ): # But POST allowed with patch.object( server, "FHIR_BASE_URL", "http://test.com/fhir" ): with patch.object(server, "FHIR_ALLOW_READ", False): import respx with respx.mock: respx.post("http://test.com/fhir/Patient").mock( return_value=httpx.Response( 201, json={"id": "123"} ) ) result = await server.write_fhir_resource.fn( valid_patient_resource ) assert "Success" in result @pytest.mark.asyncio async def test_allowed_methods_blocks_put_but_allows_post( self, valid_patient_with_id, valid_patient_resource ): """Test that PUT can be blocked while POST is allowed.""" with patch("server.validate_fhir_resource", return_value=(True, None)): with patch.object(server, "compiled_validator", Mock()): with patch.object( server, "_allowed_methods_set", {"POST"} ): # Only POST with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", False): # PUT should fail result_put = await server.write_fhir_resource.fn( valid_patient_with_id ) assert "Error" in result_put assert "PUT is not allowed" in result_put # POST should succeed import respx with respx.mock: respx.post("http://test.com/fhir/Patient").mock( return_value=httpx.Response(201, json={"id": "123"}) ) result_post = await server.write_fhir_resource.fn( valid_patient_resource ) assert "Success" in result_post @pytest.mark.asyncio async def test_allowed_methods_overrides_read_permission(self): """Test that FHIR_ALLOWED_METHODS overrides FHIR_ALLOW_READ.""" with patch.object(server, "FHIR_ALLOW_READ", True): # Read allowed with patch.object( server, "_allowed_methods_set", {"POST"} ): # But only POST in methods result = await server.read_fhir_resource.fn("Patient", "123") assert "Error" in result assert "GET is not allowed" in result assert "POST" in result class TestGetPatientZusUpid: """Tests for get_patient_zus_upid tool.""" @pytest.fixture def patient_with_zus_upid(self): """Sample Patient resource with Zus UPID identifier.""" return { "resourceType": "Patient", "id": "patient-123", "name": [{"family": "Smith", "given": ["John"]}], "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-12345" }, { "system": "http://hospital.example.org/patient-ids", "value": "hospital-123" } ] } @pytest.fixture def patient_without_zus_upid(self): """Sample Patient resource without Zus UPID identifier.""" return { "resourceType": "Patient", "id": "patient-456", "name": [{"family": "Doe", "given": ["Jane"]}], "identifier": [ { "system": "http://hospital.example.org/patient-ids", "value": "hospital-456" } ] } @pytest.mark.asyncio @respx.mock async def test_get_zus_upid_success(self, patient_with_zus_upid): """Test successful retrieval of Zus UPID.""" with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", True): with patch("zus_extensions.FHIR_BASE_URL", "http://test.com/fhir"): # Mock FHIR search response bundle_response = { "resourceType": "Bundle", "total": 1, "entry": [ { "resource": patient_with_zus_upid } ] } respx.get("http://test.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle_response) ) result = await server.get_patient_zus_upid.fn("John", "Smith") assert "Zus UPID: zus-upid-12345" in result @pytest.mark.asyncio @respx.mock async def test_get_zus_upid_with_builder_id(self, patient_with_zus_upid): """Test successful retrieval of Zus UPID with builder ID filter.""" with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", True): with patch("zus_extensions.FHIR_BASE_URL", "http://test.com/fhir"): # Mock FHIR search response bundle_response = { "resourceType": "Bundle", "total": 1, "entry": [ { "resource": patient_with_zus_upid } ] } respx.get("http://test.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle_response) ) result = await server.get_patient_zus_upid.fn("John", "Smith", "builder-123") assert "Zus UPID: zus-upid-12345" in result @pytest.mark.asyncio @respx.mock async def test_get_zus_upid_no_patients_found(self): """Test when no patients are found.""" with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", True): with patch("zus_extensions.FHIR_BASE_URL", "http://test.com/fhir"): # Mock empty FHIR search response bundle_response = { "resourceType": "Bundle", "total": 0, "entry": [] } respx.get("http://test.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle_response) ) result = await server.get_patient_zus_upid.fn("John", "Smith") assert "No Patient found with name 'John Smith'" in result @pytest.mark.asyncio @respx.mock async def test_get_zus_upid_no_zus_upid_identifier(self, patient_without_zus_upid): """Test when patient found but no Zus UPID identifier.""" with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", True): with patch("zus_extensions.FHIR_BASE_URL", "http://test.com/fhir"): # Mock FHIR search response bundle_response = { "resourceType": "Bundle", "total": 1, "entry": [ { "resource": patient_without_zus_upid } ] } respx.get("http://test.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle_response) ) result = await server.get_patient_zus_upid.fn("Jane", "Doe") assert "No Zus UPID found for Patient(s) with name 'Jane Doe'" in result @pytest.mark.asyncio @respx.mock async def test_get_zus_upid_multiple_patients(self): """Test when multiple patients with Zus UPIDs are found.""" with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", True): with patch("zus_extensions.FHIR_BASE_URL", "http://test.com/fhir"): # Mock FHIR search response with multiple patients bundle_response = { "resourceType": "Bundle", "total": 2, "entry": [ { "resource": { "resourceType": "Patient", "id": "patient-1", "name": [{"family": "Smith", "given": ["John"]}], "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-1" } ] } }, { "resource": { "resourceType": "Patient", "id": "patient-2", "name": [{"family": "Smith", "given": ["Johnny"]}], "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-2" } ] } } ] } respx.get("http://test.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle_response) ) result = await server.get_patient_zus_upid.fn("John", "Smith") assert "Zus UPID: zus-upid-1 (Best match: John Smith)" in result assert "zus-upid-1" in result assert "zus-upid-2" in result @pytest.mark.asyncio async def test_get_zus_upid_read_not_allowed(self): """Test when read operations are not allowed.""" with patch.object(server, "FHIR_ALLOW_READ", False): result = await server.get_patient_zus_upid.fn("John", "Smith") assert "Error" in result assert "Read operations are not allowed" in result @pytest.mark.asyncio async def test_get_zus_upid_method_not_allowed(self): """Test when GET method is not allowed.""" with patch.object(server, "FHIR_ALLOW_READ", True): with patch.object(server, "_allowed_methods_set", {"POST"}): result = await server.get_patient_zus_upid.fn("John", "Smith") assert "Error" in result assert "GET is not allowed" in result assert "POST" in result @pytest.mark.asyncio @respx.mock async def test_get_zus_upid_name_matching_best_match(self): """Test name matching when multiple patients found with clear best match.""" with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", True): with patch("zus_extensions.FHIR_BASE_URL", "http://test.com/fhir"): # Mock FHIR search response with multiple patients bundle_response = { "resourceType": "Bundle", "total": 3, "entry": [ { "resource": { "resourceType": "Patient", "id": "patient-1", "name": [{"family": "Smith", "given": ["John"]}], # Exact match "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-1" } ] } }, { "resource": { "resourceType": "Patient", "id": "patient-2", "name": [{"family": "Smith", "given": ["Johnny"]}], # Partial match "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-2" } ] } }, { "resource": { "resourceType": "Patient", "id": "patient-3", "name": [{"family": "Smith", "given": ["Jonathan"]}], # No match "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-3" } ] } } ] } respx.get("http://test.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle_response) ) result = await server.get_patient_zus_upid.fn("John", "Smith") # Should return the best match (exact match) assert "Zus UPID: zus-upid-1" in result assert "Best match: John Smith" in result assert "Other matches found:" in result assert "zus-upid-2" in result # Johnny Smith should be mentioned @pytest.mark.asyncio @respx.mock async def test_get_zus_upid_name_matching_no_good_match(self): """Test when multiple patients found but no clear name match.""" with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", True): with patch("zus_extensions.FHIR_BASE_URL", "http://test.com/fhir"): # Mock FHIR search response with patients that don't match well bundle_response = { "resourceType": "Bundle", "total": 2, "entry": [ { "resource": { "resourceType": "Patient", "id": "patient-1", "name": [{"family": "Johnson", "given": ["Mike"]}], # No match "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-1" } ] } }, { "resource": { "resourceType": "Patient", "id": "patient-2", "name": [{"family": "Williams", "given": ["Jane"]}], # No match "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-2" } ] } } ] } respx.get("http://test.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle_response) ) result = await server.get_patient_zus_upid.fn("John", "Smith") # Should return all patients since no good match assert "Found 2 Patient(s) with Zus UPID(s) (no clear name match):" in result assert "zus-upid-1" in result assert "zus-upid-2" in result @pytest.mark.asyncio @respx.mock async def test_get_zus_upid_name_matching_partial_match(self): """Test name matching with partial matches.""" with patch.object(server, "FHIR_BASE_URL", "http://test.com/fhir"): with patch.object(server, "FHIR_ALLOW_READ", True): with patch("zus_extensions.FHIR_BASE_URL", "http://test.com/fhir"): # Mock FHIR search response with partial matches bundle_response = { "resourceType": "Bundle", "total": 2, "entry": [ { "resource": { "resourceType": "Patient", "id": "patient-1", "name": [{"family": "Smith", "given": ["Johnny"]}], # Partial match "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-1" } ] } }, { "resource": { "resourceType": "Patient", "id": "patient-2", "name": [{"family": "Smith", "given": ["Johnson"]}], # Partial match "identifier": [ { "system": "https://zusapi.com/fhir/identifier/universal-id", "value": "zus-upid-2" } ] } } ] } respx.get("http://test.com/fhir/Patient").mock( return_value=httpx.Response(200, json=bundle_response) ) result = await server.get_patient_zus_upid.fn("John", "Smith") # Should return the best partial match (Johnny matches John) assert "Zus UPID: zus-upid-1" in result assert "Best match: Johnny Smith" in result def test_calculate_name_similarity_exact_match(self): """Test name similarity calculation with exact match.""" patient_name = {"family": "Smith", "given": ["John"]} score = server.calculate_name_similarity(patient_name, "John", "Smith") assert score == 1.0 def test_calculate_name_similarity_partial_match(self): """Test name similarity calculation with partial match.""" patient_name = {"family": "Smith", "given": ["Johnny"]} score = server.calculate_name_similarity(patient_name, "John", "Smith") assert abs(score - 0.88) < 0.001 # 0.6 (family) + 0.4 * 0.7 (partial given) def test_calculate_name_similarity_no_match(self): """Test name similarity calculation with no match.""" patient_name = {"family": "Johnson", "given": ["Mike"]} score = server.calculate_name_similarity(patient_name, "John", "Smith") assert score == 0.0 def test_calculate_name_similarity_family_only_match(self): """Test name similarity calculation with family name only match.""" patient_name = {"family": "Smith", "given": ["Mike"]} score = server.calculate_name_similarity(patient_name, "John", "Smith") assert score == 0.6 # Only family name matches def test_calculate_name_similarity_empty_name(self): """Test name similarity calculation with empty name.""" patient_name = {} score = server.calculate_name_similarity(patient_name, "John", "Smith") assert score == 0.0