diff --git a/gateway-api/src/gateway_api/controller.py b/gateway-api/src/gateway_api/controller.py index 4a17d08c..f186bcb2 100644 --- a/gateway-api/src/gateway_api/controller.py +++ b/gateway-api/src/gateway_api/controller.py @@ -18,6 +18,7 @@ from gateway_api.common.common import FlaskResponse from gateway_api.pds_search import PdsClient, PdsSearchResults +from gateway_api.sds_search import SdsClient, SdsSearchResults @dataclass @@ -44,62 +45,6 @@ def __str__(self) -> str: return self.message -@dataclass -class SdsSearchResults: - """ - Stub SDS search results dataclass. - - Replace this with the real one once it's implemented. - - :param asid: Accredited System ID. - :param endpoint: Endpoint URL associated with the organisation, if applicable. - """ - - asid: str - endpoint: str | None - - -class SdsClient: - """ - Stub SDS client for obtaining ASID from ODS code. - - Replace this with the real one once it's implemented. - """ - - SANDBOX_URL = "https://example.invalid/sds" - - def __init__( - self, - auth_token: str, - base_url: str = SANDBOX_URL, - timeout: int = 10, - ) -> None: - """ - Create an SDS client. - - :param auth_token: Authentication token to present to SDS. - :param base_url: Base URL for SDS. - :param timeout: Timeout in seconds for SDS calls. - """ - self.auth_token = auth_token - self.base_url = base_url - self.timeout = timeout - - def get_org_details(self, ods_code: str) -> SdsSearchResults | None: - """ - Retrieve SDS org details for a given ODS code. - - This is a placeholder implementation that always returns an ASID and endpoint. - - :param ods_code: ODS code to look up. - :returns: SDS search results or ``None`` if not found. - """ - # Placeholder implementation - return SdsSearchResults( - asid=f"asid_{ods_code}", endpoint="https://example-provider.org/endpoint" - ) - - class Controller: """ Orchestrates calls to PDS -> SDS -> GP provider. @@ -113,7 +58,7 @@ class Controller: def __init__( self, pds_base_url: str = PdsClient.SANDBOX_URL, - sds_base_url: str = "https://example.invalid/sds", + sds_base_url: str = SdsClient.SANDBOX_URL, nhsd_session_urid: str | None = None, timeout: int = 10, ) -> None: @@ -252,7 +197,7 @@ def _get_sds_details( - provider details (ASID + endpoint) - consumer details (ASID) - :param auth_token: Authorization token to use for SDS. + :param auth_token: Authorization token to use for SDS (used as API key). :param consumer_ods: Consumer organisation ODS code (from request headers). :param provider_ods: Provider organisation ODS code (from PDS). :returns: Tuple of (consumer_asid, provider_asid, provider_endpoint). @@ -260,7 +205,7 @@ def _get_sds_details( """ # SDS: Get provider details (ASID + endpoint) for provider ODS sds = SdsClient( - auth_token=auth_token, + api_key=auth_token, base_url=self.sds_base_url, timeout=self.timeout, ) diff --git a/gateway-api/src/gateway_api/provider_request.py b/gateway-api/src/gateway_api/provider_request.py index a628dbcf..3afe0e7f 100644 --- a/gateway-api/src/gateway_api/provider_request.py +++ b/gateway-api/src/gateway_api/provider_request.py @@ -26,7 +26,7 @@ from urllib.parse import urljoin from requests import HTTPError, Response, post -from stubs.stub_provider import stub_post +from stubs.stub_provider import GpProviderStub ARS_INTERACTION_ID = ( "urn:nhs:names:services:gpconnect:structured" @@ -43,7 +43,8 @@ # Direct all requests to the stub provider for steel threading in dev. # Replace with `from requests import post` for real requests. PostCallable = Callable[..., Response] - post: PostCallable = stub_post # type: ignore[no-redef] + _gp_provider_stub = GpProviderStub() + post: PostCallable = _gp_provider_stub.post # type: ignore[no-redef] class ExternalServiceError(Exception): diff --git a/gateway-api/src/gateway_api/sds_search.py b/gateway-api/src/gateway_api/sds_search.py new file mode 100644 index 00000000..a3183a87 --- /dev/null +++ b/gateway-api/src/gateway_api/sds_search.py @@ -0,0 +1,280 @@ +""" +SDS (Spine Directory Service) FHIR R4 device and endpoint lookup client. + +This module provides a client for querying the Spine Directory Service to retrieve: +- Device records (including ASID - Accredited System ID) +- Endpoint records (including endpoint URLs for routing) + +The client is structured similarly to :mod:`gateway_api.pds_search` and supports +stubbing for testing purposes. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any, Literal, cast + +import requests +from stubs.stub_sds import SdsFhirApiStub + +# Recursive JSON-like structure typing used for parsed FHIR bodies. +type ResultStructure = str | dict[str, "ResultStructure"] | list["ResultStructure"] +type ResultStructureDict = dict[str, ResultStructure] +type ResultList = list[ResultStructureDict] + +# Type for stub get method +type GetCallable = Callable[..., requests.Response] + + +class ExternalServiceError(Exception): + """ + Raised when the downstream SDS request fails. + + This module catches :class:`requests.HTTPError` thrown by + ``response.raise_for_status()`` and re-raises it as ``ExternalServiceError`` so + callers are not coupled to ``requests`` exception types. + """ + + +@dataclass +class SdsSearchResults: + """ + SDS lookup results containing ASID and endpoint information. + + :param asid: Accredited System ID extracted from the Device resource. + :param endpoint: Endpoint URL extracted from the Endpoint resource, or ``None`` + if no endpoint is available. + """ + + asid: str | None + endpoint: str | None + + +class SdsClient: + """ + Simple client for SDS FHIR R4 device and endpoint retrieval. + + The client supports: + + * :meth:`get_org_details` - Retrieves ASID and endpoint for an organization + + This method returns a :class:`SdsSearchResults` instance when data can be + extracted, otherwise ``None``. + + **Usage example**:: + + sds = SdsClient( + api_key="YOUR_API_KEY", + base_url="https://sandbox.api.service.nhs.uk/spine-directory/FHIR/R4", + ) + + result = sds.get_org_details("A12345") + + if result: + print(f"ASID: {result.asid}, Endpoint: {result.endpoint}") + """ + + # URLs for different SDS environments + SANDBOX_URL = "https://sandbox.api.service.nhs.uk/spine-directory/FHIR/R4" + INT_URL = "https://int.api.service.nhs.uk/spine-directory/FHIR/R4" + DEP_UAT_URL = "https://dep.api.service.nhs.uk/spine-directory/FHIR/R4" + PROD_URL = "https://api.service.nhs.uk/spine-directory/FHIR/R4" + + # FHIR identifier systems + ODS_SYSTEM = "https://fhir.nhs.uk/Id/ods-organization-code" + INTERACTION_SYSTEM = "https://fhir.nhs.uk/Id/nhsServiceInteractionId" + PARTYKEY_SYSTEM = "https://fhir.nhs.uk/Id/nhsMhsPartyKey" + ASID_SYSTEM = "https://fhir.nhs.uk/Id/nhsSpineASID" + + # SDS resource types + DEVICE: Literal["Device"] = "Device" + ENDPOINT: Literal["Endpoint"] = "Endpoint" + + # Default service interaction ID for GP Connect + DEFAULT_SERVICE_INTERACTION_ID = ( + "urn:nhs:names:services:gpconnect:fhir:rest:read:metadata-1" + ) + + def __init__( + self, + api_key: str, + base_url: str = SANDBOX_URL, + timeout: int = 10, + service_interaction_id: str | None = None, + ) -> None: + """ + Create an SDS client. + + :param api_key: API key for SDS authentication (header 'apikey'). + :param base_url: Base URL for the SDS API. Trailing slashes are stripped. + :param timeout: Default timeout in seconds for HTTP calls. + :param service_interaction_id: Service interaction ID to use for lookups. + If not provided, uses :attr:`DEFAULT_SERVICE_INTERACTION_ID`. + """ + self.api_key = api_key + self.base_url = base_url.rstrip("/") + self.timeout = timeout + self.service_interaction_id = ( + service_interaction_id or self.DEFAULT_SERVICE_INTERACTION_ID + ) + self.stub = SdsFhirApiStub() + + # Use stub for now - use environment variable once we have one + # TODO: Put this back to using the environment variable + # if os.environ.get("STUB_SDS", None): + self.get_method: GetCallable = self.stub.get + # else: + # self.get_method: GetCallable = requests.get + + def _build_headers(self, correlation_id: str | None = None) -> dict[str, str]: + """ + Build mandatory and optional headers for an SDS request. + + :param correlation_id: Optional ``X-Correlation-Id`` for cross-system tracing. + :return: Dictionary of HTTP headers for the outbound request. + """ + headers = { + "Accept": "application/fhir+json", + "apikey": self.api_key, + } + + if correlation_id: + headers["X-Correlation-Id"] = correlation_id + + return headers + + def get_org_details( + self, + ods_code: str, + correlation_id: str | None = None, + timeout: int | None = None, + ) -> SdsSearchResults | None: + """ + Retrieve ASID and endpoint for an organization by ODS code. + + This method performs two SDS queries: + 1. Query /Device to get the ASID for the organization + 2. Query /Endpoint to get the endpoint URL (if available) + + :param ods_code: ODS code of the organization to look up. + :param correlation_id: Optional correlation ID for tracing. + :param timeout: Optional per-call timeout in seconds. If not provided, + :attr:`timeout` is used. + :return: A :class:`SdsSearchResults` instance if data can be extracted, + otherwise ``None``. + :raises ExternalServiceError: If the HTTP request returns an error status. + """ + # Step 1: Get Device to obtain ASID + device_bundle = self._query_sds( + ods_code=ods_code, + correlation_id=correlation_id, + timeout=timeout, + querytype=self.DEVICE, + ) + + device = self._extract_first_entry(device_bundle) + if device is None: + return None + + asid = self._extract_identifier(device, self.ASID_SYSTEM) + party_key = self._extract_identifier(device, self.PARTYKEY_SYSTEM) + + # Step 2: Get Endpoint to obtain endpoint URL + endpoint_url: str | None = None + if party_key: + endpoint_bundle = self._query_sds( + ods_code=ods_code, + party_key=party_key, + correlation_id=correlation_id, + timeout=timeout, + querytype=self.ENDPOINT, + ) + endpoint = self._extract_first_entry(endpoint_bundle) + if endpoint: + address = endpoint.get("address") + if address: + endpoint_url = str(address).strip() + + return SdsSearchResults(asid=asid, endpoint=endpoint_url) + + def _query_sds( + self, + ods_code: str, + party_key: str | None = None, + correlation_id: str | None = None, + timeout: int | None = 10, + querytype: Literal["Device", "Endpoint"] = DEVICE, + ) -> ResultStructureDict: + """ + Query SDS /Device or /Endpoint endpoint. + + :param ods_code: ODS code to search for. + :param party_key: Party key to search for. + :param correlation_id: Optional correlation ID. + :param timeout: Optional timeout. + :return: Parsed JSON response as a dictionary. + :raises ExternalServiceError: If the request fails. + """ + headers = self._build_headers(correlation_id=correlation_id) + url = f"{self.base_url}/{querytype}" + + params: dict[str, Any] = { + "organization": f"{self.ODS_SYSTEM}|{ods_code}", + "identifier": [f"{self.INTERACTION_SYSTEM}|{self.service_interaction_id}"], + } + + if party_key is not None: + params["identifier"].append(f"{self.PARTYKEY_SYSTEM}|{party_key}") + + response = self.get_method( + url, + headers=headers, + params=params, + timeout=timeout or self.timeout, + ) + + # TODO: Post-steel-thread we probably want a raise_for_status() here + + body = response.json() + return cast("ResultStructureDict", body) + + # --------------- internal helpers for result extraction ----------------- + + @staticmethod + def _extract_first_entry( + bundle: ResultStructureDict, + ) -> ResultStructureDict | None: + """ + Extract the first Device resource from a Bundle. + + :param bundle: FHIR Bundle containing Device resources. + :return: First Device resource, or ``None`` if the bundle is empty. + """ + entries = cast("ResultList", bundle.get("entry", [])) + + # TODO: Post-steel-thread handle case where bundle contains no entries + + first_entry = entries[0] + return cast("ResultStructureDict", first_entry.get("resource", {})) + + def _extract_identifier( + self, device: ResultStructureDict, system: str + ) -> str | None: + """ + Extract an identifier value from a Device resource for a given system. + + :param device: Device resource dictionary. + :param system: The identifier system to look for. + :return: Identifier value if found, otherwise ``None``. + """ + identifiers = cast("ResultList", device.get("identifier", [])) + + for identifier in identifiers: + id_system = str(identifier.get("system", "")) + if id_system == system: + value = identifier.get("value") + if value: + return str(value).strip() + + return None diff --git a/gateway-api/src/gateway_api/test_controller.py b/gateway-api/src/gateway_api/test_controller.py index 3fc3ded4..e16e44a5 100644 --- a/gateway-api/src/gateway_api/test_controller.py +++ b/gateway-api/src/gateway_api/test_controller.py @@ -14,11 +14,9 @@ import gateway_api.controller as controller_module from gateway_api.app import app -from gateway_api.controller import ( - Controller, - SdsSearchResults, -) +from gateway_api.controller import Controller from gateway_api.get_structured_record.request import GetStructuredRecordRequest +from gateway_api.sds_search import SdsSearchResults if TYPE_CHECKING: from collections.abc import Generator @@ -77,18 +75,21 @@ class FakeSdsClient: def __init__( self, - auth_token: str | None = None, + api_key: str, base_url: str = "test_url", timeout: int = 10, + service_interaction_id: str | None = None, ) -> None: FakeSdsClient.last_init = { - "auth_token": auth_token, + "api_key": api_key, "base_url": base_url, "timeout": timeout, + "service_interaction_id": service_interaction_id, } - self.auth_token = auth_token + self.api_key = api_key self.base_url = base_url self.timeout = timeout + self.service_interaction_id = service_interaction_id self._org_details_by_ods: dict[str, SdsSearchResults | None] = {} def set_org_details( diff --git a/gateway-api/src/gateway_api/test_sds_search.py b/gateway-api/src/gateway_api/test_sds_search.py new file mode 100644 index 00000000..d023fb1e --- /dev/null +++ b/gateway-api/src/gateway_api/test_sds_search.py @@ -0,0 +1,389 @@ +""" +Unit tests for :mod:`gateway_api.sds_search`. +""" + +from __future__ import annotations + +from typing import Any + +import pytest +from stubs.stub_sds import SdsFhirApiStub + +from gateway_api.sds_search import SdsClient, SdsSearchResults + + +@pytest.fixture +def stub() -> SdsFhirApiStub: + """ + Create a stub backend instance. + + :return: A :class:`stubs.stub_sds.SdsFhirApiStub` instance. + """ + return SdsFhirApiStub() + + +@pytest.fixture +def mock_requests_get( + monkeypatch: pytest.MonkeyPatch, stub: SdsFhirApiStub +) -> dict[str, Any]: + """ + Patch ``SdsFhirApiStub`` so the SdsClient uses the test stub fixture. + + The fixture returns a "capture" dict recording the most recent request information. + + :param monkeypatch: Pytest monkeypatch fixture. + :param stub: Stub backend used to serve GET requests. + :param return: A capture dictionary containing the last call details. + """ + capture: dict[str, Any] = {} + + # Wrap the stub's get method to capture call parameters + original_stub_get = stub.get + + def _capturing_get( + url: str, + headers: dict[str, str] | None = None, + params: Any = None, + timeout: Any = None, + ) -> Any: + """ + Wrapper around stub.get that captures parameters. + + :param url: URL passed by the client. + :param headers: Headers passed by the client. + :param params: Query parameters. + :param timeout: Timeout. + :return: Response from the stub. + """ + headers = headers or {} + capture["url"] = url + capture["headers"] = dict(headers) + capture["params"] = params + capture["timeout"] = timeout + + return original_stub_get(url, headers, params, timeout) + + stub.get = _capturing_get # type: ignore[method-assign] + + # Monkeypatch SdsFhirApiStub so SdsClient uses our test stub + import gateway_api.sds_search as sds_module + + monkeypatch.setattr( + sds_module, + "SdsFhirApiStub", + lambda *args, **kwargs: stub, # NOQA ARG005 (maintain signature) + ) + + return capture + + +def test_sds_client_get_org_details_success( + stub: SdsFhirApiStub, # noqa: ARG001 + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test SdsClient can successfully look up organization details. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + client = SdsClient(api_key="test-key", base_url=SdsClient.SANDBOX_URL) + + result = client.get_org_details(ods_code="PROVIDER") + + assert result is not None + assert isinstance(result, SdsSearchResults) + assert result.asid == "asid_PROV" + assert result.endpoint is not None + + +def test_sds_client_get_org_details_with_endpoint( + stub: SdsFhirApiStub, + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test SdsClient retrieves endpoint when available. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + # Add a device with party key so we can get an endpoint + stub.upsert_device( + organization_ods="TESTORG", + service_interaction_id="urn:nhs:names:services:gpconnect:fhir:rest:read:metadata-1", + party_key="TESTORG-123456", + device={ + "resourceType": "Device", + "id": "test-device-id", + "identifier": [ + { + "system": "https://fhir.nhs.uk/Id/nhsSpineASID", + "value": "999999999999", + }, + { + "system": "https://fhir.nhs.uk/Id/nhsMhsPartyKey", + "value": "TESTORG-123456", + }, + ], + "owner": { + "identifier": { + "system": "https://fhir.nhs.uk/Id/ods-organization-code", + "value": "TESTORG", + } + }, + }, + ) + + stub.upsert_endpoint( + organization_ods="TESTORG", + service_interaction_id="urn:nhs:names:services:gpconnect:fhir:rest:read:metadata-1", + party_key="TESTORG-123456", + endpoint={ + "resourceType": "Endpoint", + "id": "test-endpoint-id", + "status": "active", + "address": "https://testorg.example.com/fhir", + "managingOrganization": { + "identifier": { + "system": "https://fhir.nhs.uk/Id/ods-organization-code", + "value": "TESTORG", + } + }, + "identifier": [ + { + "system": "https://fhir.nhs.uk/Id/nhsMhsPartyKey", + "value": "TESTORG-123456", + } + ], + }, + ) + + client = SdsClient(api_key="test-key", base_url=SdsClient.SANDBOX_URL) + result = client.get_org_details(ods_code="TESTORG") + + assert result is not None + assert result.asid == "999999999999" + assert result.endpoint == "https://testorg.example.com/fhir" + + +def test_sds_client_get_org_details_no_endpoint( + stub: SdsFhirApiStub, + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test SdsClient handles missing endpoint gracefully. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + # Add a device without a party key (so no endpoint will be found) + stub.upsert_device( + organization_ods="NOENDPOINT", + service_interaction_id="urn:nhs:names:services:gpconnect:fhir:rest:read:metadata-1", + party_key=None, + device={ + "resourceType": "Device", + "id": "noendpoint-device-id", + "identifier": [ + { + "system": "https://fhir.nhs.uk/Id/nhsSpineASID", + "value": "888888888888", + } + ], + "owner": { + "identifier": { + "system": "https://fhir.nhs.uk/Id/ods-organization-code", + "value": "NOENDPOINT", + } + }, + }, + ) + + client = SdsClient(api_key="test-key", base_url=SdsClient.SANDBOX_URL) + result = client.get_org_details(ods_code="NOENDPOINT") + + assert result is not None + assert result.asid == "888888888888" + assert result.endpoint is None + + +def test_sds_client_sends_correlation_id( + stub: SdsFhirApiStub, # noqa: ARG001 + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test that SdsClient sends X-Correlation-Id header when provided. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + client = SdsClient(api_key="test-key", base_url=SdsClient.SANDBOX_URL) + + correlation_id = "test-correlation-123" + client.get_org_details(ods_code="PROVIDER", correlation_id=correlation_id) + + # Check that the header was sent + assert mock_requests_get["headers"]["X-Correlation-Id"] == correlation_id + + +def test_sds_client_sends_apikey( + stub: SdsFhirApiStub, # noqa: ARG001 + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test that SdsClient sends apikey header. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + api_key = "my-secret-key" + client = SdsClient(api_key=api_key, base_url=SdsClient.SANDBOX_URL) + + client.get_org_details(ods_code="PROVIDER") + + # Check that the apikey header was sent + assert mock_requests_get["headers"]["apikey"] == api_key + + +def test_sds_client_timeout_parameter( + stub: SdsFhirApiStub, # noqa: ARG001 + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test that SdsClient passes timeout parameter to requests. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + client = SdsClient(api_key="test-key", base_url=SdsClient.SANDBOX_URL, timeout=30) + + client.get_org_details(ods_code="PROVIDER", timeout=60) + + # Check that the custom timeout was passed + assert mock_requests_get["timeout"] == 60 + + +def test_sds_client_custom_service_interaction_id( + stub: SdsFhirApiStub, + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test that SdsClient uses custom interaction ID when provided. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + custom_interaction = "urn:nhs:names:services:custom:CUSTOM123" + + # Add device with custom interaction ID + stub.upsert_device( + organization_ods="CUSTOMINT", + service_interaction_id=custom_interaction, + party_key=None, + device={ + "resourceType": "Device", + "id": "custom-device", + "identifier": [ + { + "system": "https://fhir.nhs.uk/Id/nhsSpineASID", + "value": "777777777777", + } + ], + "owner": { + "identifier": { + "system": "https://fhir.nhs.uk/Id/ods-organization-code", + "value": "CUSTOMINT", + } + }, + }, + ) + + client = SdsClient( + api_key="test-key", + base_url=SdsClient.SANDBOX_URL, + service_interaction_id=custom_interaction, + ) + + result = client.get_org_details(ods_code="CUSTOMINT") + + # Verify the custom interaction was used + params = mock_requests_get["params"] + assert any( + custom_interaction in str(ident) for ident in params.get("identifier", []) + ) + + # Verify we got the result + assert result is not None + assert result.asid == "777777777777" + + +def test_sds_client_builds_correct_device_query_params( + stub: SdsFhirApiStub, # noqa: ARG001 + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test that SdsClient builds Device query parameters correctly. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + client = SdsClient(api_key="test-key", base_url=SdsClient.SANDBOX_URL) + + client.get_org_details(ods_code="PROVIDER") + + params = mock_requests_get["params"] + + # Check organization parameter + assert ( + params["organization"] + == "https://fhir.nhs.uk/Id/ods-organization-code|PROVIDER" + ) + + # Check identifier list contains interaction ID + identifiers = params["identifier"] + assert isinstance(identifiers, list) + assert any( + "https://fhir.nhs.uk/Id/nhsServiceInteractionId|" in str(ident) + for ident in identifiers + ) + + +def test_sds_client_extract_asid_from_device( + stub: SdsFhirApiStub, # noqa: ARG001 + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test ASID extraction from Device resource. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + client = SdsClient(api_key="test-key", base_url=SdsClient.SANDBOX_URL) + + result = client.get_org_details(ods_code="PROVIDER") + + assert result is not None + assert result.asid is not None + assert result.asid == "asid_PROV" + + +def test_sds_client_extract_party_key_from_device( + stub: SdsFhirApiStub, # noqa: ARG001 + mock_requests_get: dict[str, Any], # noqa: ARG001 +) -> None: + """ + Test party key extraction and subsequent endpoint lookup. + + :param stub: SDS stub fixture. + :param mock_requests_get: Capture fixture for request details. + """ + # The default seeded PROVIDER device has a party key, which should trigger + # an endpoint lookup + client = SdsClient(api_key="test-key", base_url=SdsClient.SANDBOX_URL) + + # Need to seed the data correctly - let's use CONSUMER which has party key + result = client.get_org_details(ods_code="CONSUMER") + + # Should have found ASID but may not have endpoint depending on seeding + assert result is not None + assert result.asid == "asid_CONS" diff --git a/gateway-api/stubs/stubs/__init__.py b/gateway-api/stubs/stubs/__init__.py index e69de29b..2b22a081 100644 --- a/gateway-api/stubs/stubs/__init__.py +++ b/gateway-api/stubs/stubs/__init__.py @@ -0,0 +1,6 @@ +from .base_stub import StubBase +from .stub_pds import PdsFhirApiStub +from .stub_provider import GpProviderStub +from .stub_sds import SdsFhirApiStub + +__all__ = ["StubBase", "PdsFhirApiStub", "SdsFhirApiStub", "GpProviderStub"] diff --git a/gateway-api/stubs/stubs/base_stub.py b/gateway-api/stubs/stubs/base_stub.py new file mode 100644 index 00000000..f1b08070 --- /dev/null +++ b/gateway-api/stubs/stubs/base_stub.py @@ -0,0 +1,62 @@ +""" +Base class for FHIR API stubs. + +Provides common functionality for creating stub responses. +""" + +from __future__ import annotations + +import json +from abc import ABC, abstractmethod +from http.client import responses as http_responses +from typing import Any + +from requests import Response +from requests.structures import CaseInsensitiveDict + + +class StubBase(ABC): + """ + Abstract base class for FHIR API stubs. + + Provides common functionality for creating HTTP responses and defines + the interface that all stub implementations must provide. + """ + + @staticmethod + def _create_response( + status_code: int, + headers: dict[str, str], + json_data: dict[str, Any], + ) -> Response: + """ + Create a :class:`requests.Response` object for the stub. + """ + response = Response() + response.status_code = status_code + response.headers = CaseInsensitiveDict(headers) + response._content = json.dumps(json_data).encode("utf-8") # noqa: SLF001 to customise stub + response.encoding = "utf-8" + # Set a reason phrase for HTTP error handling + response.reason = http_responses.get(status_code, "Unknown") + return response + + @abstractmethod + def get( + self, url: str, headers: dict[str, str], params: dict[str, Any], timeout: int + ) -> Response: + """ + Handle HTTP GET requests for the stub. + """ + + @abstractmethod + def post( + self, + url: str, + headers: dict[str, Any], + data: str, + timeout: int, + ) -> Response: + """ + Handle HTTP POST requests for the stub. + """ diff --git a/gateway-api/stubs/stubs/stub_pds.py b/gateway-api/stubs/stubs/stub_pds.py index f8249295..a1b2b219 100644 --- a/gateway-api/stubs/stubs/stub_pds.py +++ b/gateway-api/stubs/stubs/stub_pds.py @@ -6,41 +6,18 @@ from __future__ import annotations -import json import re import uuid from datetime import datetime, timezone -from http.client import responses as http_responses -from typing import Any +from typing import TYPE_CHECKING, Any -from requests import Response -from requests.structures import CaseInsensitiveDict +from .base_stub import StubBase - -def _create_response( - status_code: int, - headers: dict[str, str], - json_data: dict[str, Any], -) -> Response: - """ - Create a :class:`requests.Response` object for the stub. - - :param status_code: HTTP status code. - :param headers: Response headers dictionary. - :param json_data: JSON body data. - :return: A :class:`requests.Response` instance. - """ - response = Response() - response.status_code = status_code - response.headers = CaseInsensitiveDict(headers) - response._content = json.dumps(json_data).encode("utf-8") # noqa: SLF001 - response.encoding = "utf-8" - # Set a reason phrase for HTTP error handling - response.reason = http_responses.get(status_code, "Unknown") - return response +if TYPE_CHECKING: + from requests import Response -class PdsFhirApiStub: +class PdsFhirApiStub(StubBase): """ Minimal in-memory stub for the PDS FHIR API, implementing only ``GET /Patient/{id}`` @@ -253,7 +230,9 @@ def get_patient( # ETag mirrors the "W/\"\"" shape and aligns to meta.versionId. headers_out["ETag"] = f'W/"{version_id}"' - return _create_response(status_code=200, headers=headers_out, json_data=patient) + return self._create_response( + status_code=200, headers=headers_out, json_data=patient + ) def get( self, @@ -287,6 +266,24 @@ def get( end_user_org_ods=end_user_org_ods, ) + def post( + self, + url: str, + headers: dict[str, Any], + data: Any, + timeout: int, + ) -> Response: + """ + Handle HTTP POST requests for the stub. + + :param url: Request URL. + :param headers: Request headers. + :param data: Request body data. + :param timeout: Request timeout in seconds. + :raises NotImplementedError: POST requests are not supported by this stub. + """ + raise NotImplementedError("POST requests are not supported by PdsFhirApiStub") + # --------------------------- # Internal helpers # --------------------------- @@ -356,9 +353,13 @@ def _bad_request( display=message, ) - @staticmethod def _operation_outcome( - *, status_code: int, headers: dict[str, str], spine_code: str, display: str + self, + *, + status_code: int, + headers: dict[str, str], + spine_code: str, + display: str, ) -> Response: """ Construct an OperationOutcome response body. @@ -388,6 +389,6 @@ def _operation_outcome( } ], } - return _create_response( + return self._create_response( status_code=status_code, headers=dict(headers), json_data=body ) diff --git a/gateway-api/stubs/stubs/stub_provider.py b/gateway-api/stubs/stubs/stub_provider.py index 2d0c96ba..4e9845a1 100644 --- a/gateway-api/stubs/stubs/stub_provider.py +++ b/gateway-api/stubs/stubs/stub_provider.py @@ -21,39 +21,14 @@ Request Body JSON (FHIR STU3 Parameters resource with patient NHS number. """ -import json from typing import Any -from gateway_api.common.common import json_str from requests import Response -from requests.structures import CaseInsensitiveDict +from .base_stub import StubBase -def _create_response( - status_code: int, - headers: dict[str, str] | CaseInsensitiveDict[str], - content: bytes, - reason: str = "", -) -> Response: - """ - Create a :class:`requests.Response` object for the stub. - - :param status_code: HTTP status code. - :param headers: Response headers dictionary. - :param content: Response body as bytes. - :param reason: HTTP reason phrase (e.g., "OK", "Bad Request"). - :return: A :class:`requests.Response` instance. - """ - response = Response() - response.status_code = status_code - response.headers = CaseInsensitiveDict(headers) - response._content = content # noqa: SLF001 - response.reason = reason - response.encoding = "utf-8" - return response - -class GpProviderStub: +class GpProviderStub(StubBase): """ A minimal in-memory stub for a Provider GP System FHIR API, implementing only accessRecordStructured to read basic @@ -118,36 +93,61 @@ def access_record_structured( returns: Response: The stub patient bundle wrapped in a Response object. """ + if trace_id == "invalid for test": + return self._create_response( + status_code=400, + headers={"Content-Type": "application/fhir+json"}, + json_data={ + "resourceType": "OperationOutcome", + "issue": [ + { + "severity": "error", + "code": "invalid", + "diagnostics": "Invalid for testing", + } + ], + }, + ) - stub_response = _create_response( + return self._create_response( status_code=200, - headers=CaseInsensitiveDict({"Content-Type": "application/fhir+json"}), - content=json.dumps(self.patient_bundle).encode("utf-8"), - reason="OK", + headers={"Content-Type": "application/fhir+json"}, + json_data=self.patient_bundle, ) - if trace_id == "invalid for test": - return _create_response( - status_code=400, - headers=CaseInsensitiveDict({"Content-Type": "application/fhir+json"}), - content=( - b'{"resourceType":"OperationOutcome","issue":[' - b'{"severity":"error","code":"invalid",' - b'"diagnostics":"Invalid for testing"}]}' - ), - reason="Bad Request", - ) + def post( + self, + url: str, # NOQA ARG001 # NOSONAR S1172 (unused in stub) + headers: dict[str, Any], + data: str, + timeout: int, # NOQA ARG001 # NOSONAR S1172 (unused in stub) + ) -> Response: + """ + Handle HTTP POST requests for the stub. - return stub_response + :param url: Request URL. + :param headers: Request headers. + :param data: Request body data. + :param timeout: Request timeout in seconds. + :return: A :class:`requests.Response` instance. + """ + trace_id = headers.get("Ssp-TraceID", "no-trace-id") + return self.access_record_structured(trace_id, data) + def get( + self, + url: str, + headers: dict[str, str], + params: dict[str, Any], + timeout: int, + ) -> Response: + """ + Handle HTTP GET requests for the stub. -def stub_post( - url: str, # NOQA ARG001 # NOSONAR S1172 (unused in stub) - headers: dict[str, Any], - data: json_str, - timeout: int, # NOQA ARG001 # NOSONAR S1172 (unused in stub) -) -> Response: - """A stubbed requests.post function that routes to the GPProviderStub.""" - _provider_stub = GpProviderStub() - trace_id = headers.get("Ssp-TraceID", "no-trace-id") - return _provider_stub.access_record_structured(trace_id, data) + :param url: Request URL. + :param headers: Request headers. + :param params: Query parameters. + :param timeout: Request timeout in seconds. + :raises NotImplementedError: GET requests are not supported by this stub. + """ + raise NotImplementedError("GET requests are not supported by GpProviderStub") diff --git a/gateway-api/stubs/stubs/stub_sds.py b/gateway-api/stubs/stubs/stub_sds.py new file mode 100644 index 00000000..16fc6d75 --- /dev/null +++ b/gateway-api/stubs/stubs/stub_sds.py @@ -0,0 +1,710 @@ +""" +In-memory SDS FHIR R4 API stub. + +The stub does **not** implement the full SDS API surface, nor full FHIR validation. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from .base_stub import StubBase + +if TYPE_CHECKING: + from requests import Response + + +class SdsFhirApiStub(StubBase): + """ + Minimal in-memory stub for the SDS FHIR API, implementing ``GET /Device`` + and ``GET /Endpoint`` + + Contract elements modelled from the SDS OpenAPI spec: + + * ``/Device`` requires query params: + - ``organization`` (required): ODS code with FHIR identifier prefix + - ``identifier`` (required, repeatable): Service interaction ID and/or party key + - ``manufacturing-organization`` (optional): Manufacturing org ODS code + * ``/Endpoint`` requires query param: + - ``identifier`` (required, repeatable): Service interaction ID and/or party key + - ``organization`` (optional): ODS code with FHIR identifier prefix + * ``X-Correlation-Id`` is optional and echoed back if supplied + * ``apikey`` header is required (but any value accepted in stub mode) + * Returns a FHIR Bundle with ``resourceType: "Bundle"`` and ``type: "searchset"`` + + See: + https://github.com/NHSDigital/spine-directory-service-api + """ + + ODS_SYSTEM = "https://fhir.nhs.uk/Id/ods-organization-code" + INTERACTION_SYSTEM = "https://fhir.nhs.uk/Id/nhsServiceInteractionId" + PARTYKEY_SYSTEM = "https://fhir.nhs.uk/Id/nhsMhsPartyKey" + ASID_SYSTEM = "https://fhir.nhs.uk/Id/nhsSpineASID" + CONNECTION_SYSTEM = ( + "https://terminology.hl7.org/CodeSystem/endpoint-connection-type" + ) + CODING_SYSTEM = "https://terminology.hl7.org/CodeSystem/endpoint-payload-type" + + GP_CONNECT_INTERACTION = ( + "urn:nhs:names:services:gpconnect:fhir:rest:read:metadata-1" + ) + CONNECTION_DISPLAY = "HL7 FHIR" + + def __init__(self) -> None: + """ + Create a new stub instance. + + :param strict_validation: If ``True``, enforce required query parameters and + apikey header. If ``False``, validation is relaxed. + """ + # Internal store: (org_ods, interaction_id, party_key) -> list[device_resource] + # party_key may be None if not specified + self._devices: dict[tuple[str, str, str | None], list[dict[str, Any]]] = {} + + # Internal store for endpoints: + # (org_ods, interaction_id, party_key) -> list[endpoint_resource] + # org_ods and/or interaction_id may be None since they're optional for + # endpoint queries + self._endpoints: dict[ + tuple[str | None, str | None, str | None], list[dict[str, Any]] + ] = {} + + # Seed some deterministic examples matching common test scenarios + self._seed_default_devices() + self._seed_default_endpoints() + + def _seed_default_devices(self) -> None: + """Seed the stub with some default Device records for testing.""" + # Define test device data as a list of parameters + device_data = [ + { + "org_ods": "PROVIDER", + "party_key": "PROVIDER-0000806", + "device_id": "F0F0E921-92CA-4A88-A550-2DBB36F703AF", + "asid": "asid_PROV", + "display": "Example NHS Trust", + }, + { + "org_ods": "CONSUMER", + "party_key": "CONSUMER-0000807", + "device_id": "C0C0E921-92CA-4A88-A550-2DBB36F703AF", + "asid": "asid_CONS", + "display": "Example Consumer Organisation", + }, + { + "org_ods": "A12345", + "party_key": "A12345-0000808", + "device_id": "A1A1E921-92CA-4A88-A550-2DBB36F703AF", + "asid": "asid_A12345", + "display": "Example GP Practice A12345", + }, + ] + + # Iterate through test data and create devices + for data in device_data: + self.upsert_device( + organization_ods=data["org_ods"], + service_interaction_id=self.GP_CONNECT_INTERACTION, + party_key=data["party_key"], + device=self._create_device_resource( + device_id=data["device_id"], + asid=data["asid"], + party_key=data["party_key"], + org_ods=data["org_ods"], + display=data["display"], + ), + ) + + def _seed_default_endpoints(self) -> None: + """Seed the stub with some default Endpoint records for testing.""" + # Define test endpoint data as a list of parameters + endpoint_data = [ + { + "org_ods": "PROVIDER", + "party_key": "PROVIDER-0000806", + "endpoint_id": "E0E0E921-92CA-4A88-A550-2DBB36F703AF", + "asid": "asid_PROV", + "address": "https://provider.example.com/fhir", + }, + { + "org_ods": "CONSUMER", + "party_key": "CONSUMER-0000807", + "endpoint_id": "E1E1E921-92CA-4A88-A550-2DBB36F703AF", + "asid": "asid_CONS", + "address": "https://consumer.example.com/fhir", + }, + { + "org_ods": "A12345", + "party_key": "A12345-0000808", + "endpoint_id": "E2E2E921-92CA-4A88-A550-2DBB36F703AF", + "asid": "asid_A12345", + "address": "https://a12345.example.com/fhir", + }, + ] + + # Iterate through test data and create endpoints + for data in endpoint_data: + self.upsert_endpoint( + organization_ods=data["org_ods"], + service_interaction_id=self.GP_CONNECT_INTERACTION, + party_key=data["party_key"], + endpoint=self._create_endpoint_resource( + endpoint_id=data["endpoint_id"], + asid=data["asid"], + party_key=data["party_key"], + org_ods=data["org_ods"], + address=data["address"], + ), + ) + + def _create_device_resource( + self, + device_id: str, + asid: str, + party_key: str, + org_ods: str, + display: str, + ) -> dict[str, Any]: + """Create a Device resource dictionary with the given parameters.""" + return { + "resourceType": "Device", + "id": device_id, + "identifier": [ + { + "system": self.ASID_SYSTEM, + "value": asid, + }, + { + "system": self.PARTYKEY_SYSTEM, + "value": party_key, + }, + ], + "owner": { + "identifier": { + "system": self.ODS_SYSTEM, + "value": org_ods, + }, + "display": display, + }, + } + + def _create_endpoint_resource( + self, + endpoint_id: str, + asid: str, + party_key: str, + org_ods: str, + address: str, + ) -> dict[str, Any]: + """Create an Endpoint resource dictionary with the given parameters.""" + return { + "resourceType": "Endpoint", + "id": endpoint_id, + "status": "active", + "connectionType": { + "system": self.CONNECTION_SYSTEM, + "code": "hl7-fhir-rest", + "display": self.CONNECTION_DISPLAY, + }, + "payloadType": [ + { + "coding": [ + { + "system": self.CODING_SYSTEM, + "code": "any", + "display": "Any", + } + ] + } + ], + "address": address, + "managingOrganization": { + "identifier": { + "system": self.ODS_SYSTEM, + "value": org_ods, + } + }, + "identifier": [ + { + "system": self.ASID_SYSTEM, + "value": asid, + }, + { + "system": self.PARTYKEY_SYSTEM, + "value": party_key, + }, + ], + } + + # --------------------------- + # Public API for tests + # --------------------------- + + def upsert_device( + self, + organization_ods: str, + service_interaction_id: str, + party_key: str | None, + device: dict[str, Any], + ) -> None: + """ + Insert or append a Device record in the stub store. + + Multiple devices can be registered for the same query combination (they will + all be returned in the Bundle.entry array). + + :param organization_ods: Organization ODS code. + :param service_interaction_id: Service interaction ID. + :param party_key: Optional MHS party key. + :param device: Device resource dictionary. + """ + key = (organization_ods, service_interaction_id, party_key) + if key not in self._devices: + self._devices[key] = [] + self._devices[key].append(device) + + def clear_devices(self) -> None: + """Clear all Device records from the stub.""" + self._devices.clear() + + def upsert_endpoint( + self, + organization_ods: str | None, + service_interaction_id: str | None, + party_key: str | None, + endpoint: dict[str, Any], + ) -> None: + """ + Insert or append an Endpoint record in the stub store. + + Multiple endpoints can be registered for the same query combination (they will + all be returned in the Bundle.entry array). + + :param organization_ods: Organization ODS code (optional for endpoints). + :param service_interaction_id: Service interaction ID (optional for endpoints). + :param party_key: Optional MHS party key. + :param endpoint: Endpoint resource dictionary. + """ + key = (organization_ods, service_interaction_id, party_key) + if key not in self._endpoints: + self._endpoints[key] = [] + self._endpoints[key].append(endpoint) + + def clear_endpoints(self) -> None: + """Clear all Endpoint records from the stub.""" + self._endpoints.clear() + + def get_device_bundle( + self, + url: str, # noqa: ARG002 # NOSONAR S1172 (ignored in stub) + headers: dict[str, str], + params: dict[str, Any], + timeout: int | None = None, # noqa: ARG002 # NOSONAR S1172 (ignored in stub) + ) -> Response: + """ + Implements ``GET /Device``. + + :param url: Request URL (expected to end with /Device). + :param headers: Request headers. Must include ``apikey``. + May include ``X-Correlation-Id``. + :param params: Query parameters dictionary. Must include ``organization`` and + ``identifier`` (list). + :param timeout: Timeout (ignored by the stub). + :return: A :class:`requests.Response` representing either: + * ``200`` with Bundle JSON (may be empty) + * ``400`` with error details for missing/invalid parameters + """ + headers = headers or {} + params = params or {} + + headers_out: dict[str, str] = {} + + # Echo correlation ID if provided + correlation_id = headers.get("X-Correlation-Id") + if correlation_id: + headers_out["X-Correlation-Id"] = correlation_id + + # Validate apikey header + if "apikey" not in headers: + return self._error_response( + status_code=400, + headers=headers_out, + message="Missing required header: apikey", + ) + + # Always validate required query parameters (not just in strict mode) + organization = params.get("organization") + identifier = params.get("identifier") + + if not organization: + return self._error_response( + status_code=400, + headers=headers_out, + message="Missing required query parameter: organization", + ) + if not identifier: + return self._error_response( + status_code=400, + headers=headers_out, + message="Missing required query parameter: identifier", + ) + + # Parse organization ODS code + org_ods = self._extract_param_value(organization, self.ODS_SYSTEM) + + # Parse identifier list (can be string or list) + # if isinstance(identifier, str): + identifier_list = [identifier] if isinstance(identifier, str) else identifier + # else: + # identifier_list = identifier + + service_interaction_id: str | None = None + party_key: str | None = None + + for ident in identifier_list: + if self.INTERACTION_SYSTEM in ident: + service_interaction_id = self._extract_param_value( + ident, self.INTERACTION_SYSTEM + ) + elif self.PARTYKEY_SYSTEM in ident: + party_key = self._extract_param_value(ident, self.PARTYKEY_SYSTEM) + + # Always validate service interaction ID is present + if not service_interaction_id: + return self._error_response( + status_code=400, + headers=headers_out, + message="identifier must include nhsServiceInteractionId", + ) + + # Look up devices + devices = self._lookup_devices( + org_ods=org_ods or "", + service_interaction_id=service_interaction_id or "", + party_key=party_key, + ) + + # Build FHIR Bundle response + bundle = self._build_bundle(devices) + + return self._create_response( + status_code=200, headers=headers_out, json_data=bundle + ) + + def get_endpoint_bundle( + self, + url: str, # noqa: ARG002 # NOSONAR S1172 (ignored in stub) + headers: dict[str, str] | None = None, + params: dict[str, Any] | None = None, + timeout: int | None = None, # noqa: ARG002 # NOSONAR S1172 (ignored in stub) + ) -> Response: + """ + Implements ``GET /Endpoint``. + + :param url: Request URL (expected to end with /Endpoint). + :param headers: Request headers. Must include ``apikey`. + May include ``X-Correlation-Id``. + :param params: Query parameters dictionary. Must include ``identifier`` (list). + ``organization`` is optional. + :param timeout: Timeout (ignored by the stub). + :return: A :class:`requests.Response` representing either: + * ``200`` with Bundle JSON (may be empty) + * ``400`` with error details for missing/invalid parameters + """ + headers = headers or {} + params = params or {} + + headers_out: dict[str, str] = {} + + # Echo correlation ID if provided + correlation_id = headers.get("X-Correlation-Id") + if correlation_id: + headers_out["X-Correlation-Id"] = correlation_id + + # Validate apikey header + if "apikey" not in headers: + return self._error_response( + status_code=400, + headers=headers_out, + message="Missing required header: apikey", + ) + + # Always validate required query parameters (not just in strict mode) + identifier = params.get("identifier") + organization = params.get("organization") + + if not identifier: + return self._error_response( + status_code=400, + headers=headers_out, + message="Missing required query parameter: identifier", + ) + + # Parse organization ODS code (optional) + org_ods: str | None = None + if organization: + org_ods = self._extract_param_value(organization, self.ODS_SYSTEM) + + # Parse identifier list (can be string or list) + if isinstance(identifier, str): + identifier = [identifier] + + service_interaction_id: str | None = None + party_key: str | None = None + + for ident in identifier or []: + if self.INTERACTION_SYSTEM in ident: + service_interaction_id = self._extract_param_value( + ident, self.INTERACTION_SYSTEM + ) + elif self.PARTYKEY_SYSTEM in ident: + party_key = self._extract_param_value(ident, self.PARTYKEY_SYSTEM) + + # Look up endpoints + endpoints = self._lookup_endpoints( + org_ods=org_ods, + service_interaction_id=service_interaction_id, + party_key=party_key, + ) + + # Build FHIR Bundle response + bundle = self._build_endpoint_bundle(endpoints) + + return self._create_response( + status_code=200, headers=headers_out, json_data=bundle + ) + + def get( + self, + url: str, + headers: dict[str, str], + params: dict[str, Any], + timeout: int = 10, + ) -> Response: + """ + Convenience method matching requests.get signature for easy monkeypatching. + + Routes to the appropriate handler based on the URL path. + + :param url: Request URL. + :param headers: Request headers. + :param params: Query parameters. + :param timeout: Timeout value. + :return: A :class:`requests.Response`. + """ + if "/Endpoint" in url: + return self.get_endpoint_bundle( + url=url, headers=headers, params=params, timeout=timeout + ) + return self.get_device_bundle( + url=url, headers=headers, params=params, timeout=timeout + ) + + def post( + self, + url: str, + headers: dict[str, Any], + data: Any, + timeout: int, + ) -> Response: + """ + Handle HTTP POST requests for the stub. + + :param url: Request URL. + :param headers: Request headers. + :param data: Request body data. + :param timeout: Request timeout in seconds. + :raises NotImplementedError: POST requests are not supported by this stub. + """ + raise NotImplementedError("POST requests are not supported by SdsFhirApiStub") + + # --------------------------- + # Internal helpers + # --------------------------- + + def _lookup_devices( + self, org_ods: str, service_interaction_id: str, party_key: str | None + ) -> list[dict[str, Any]]: + """ + Look up devices matching the query parameters. + + :param org_ods: Organization ODS code. + :param service_interaction_id: Service interaction ID. + :param party_key: Optional party key. + :return: List of matching Device resources. + """ + # Exact match with party key (or None) + key = (org_ods, service_interaction_id, party_key) + if key in self._devices: + return list(self._devices[key]) + + # If no party_key was provided (None), search for any entries with the + # same org+interaction + # This allows querying without knowing the party_key upfront + if party_key is None: + for stored_key, devices in self._devices.items(): + stored_org, stored_interaction, _ = stored_key + if ( + stored_org == org_ods + and stored_interaction == service_interaction_id + ): + return list(devices) + + # If party_key was provided but no exact match, try without party key + if party_key: + key_without_party = (org_ods, service_interaction_id, None) + if key_without_party in self._devices: + return list(self._devices[key_without_party]) + + return [] + + def _lookup_endpoints( + self, + org_ods: str | None, + service_interaction_id: str | None, + party_key: str | None, + ) -> list[dict[str, Any]]: + """ + Look up endpoints matching the query parameters. + + For /Endpoint, the query combinations are more flexible: + - organization + service_interaction_id + party_key + - organization + party_key + - organization + service_interaction_id + - service_interaction_id + party_key + + :param org_ods: Organization ODS code (optional). + :param service_interaction_id: Service interaction ID (optional). + :param party_key: Optional party key. + :return: List of matching Endpoint resources. + """ + results = [] + + # Try to find exact matches and partial matches + for key, endpoints in self._endpoints.items(): + stored_org, stored_interaction, stored_party = key + + # Check if the query parameters match + org_match = org_ods is None or stored_org is None or org_ods == stored_org + interaction_match = ( + service_interaction_id is None + or stored_interaction is None + or service_interaction_id == stored_interaction + ) + party_match = ( + party_key is None or stored_party is None or party_key == stored_party + ) + + # If all specified parameters match, include these endpoints + if org_match and interaction_match and party_match: + # But at least one must be non-None and match + has_match = ( + (org_ods and stored_org and org_ods == stored_org) + or ( + service_interaction_id + and stored_interaction + and service_interaction_id == stored_interaction + ) + or (party_key and stored_party and party_key == stored_party) + ) + if has_match: + results.extend(endpoints) + + return results + + def _build_bundle(self, devices: list[dict[str, Any]]) -> dict[str, Any]: + """ + Build a FHIR Bundle from a list of Device resources. + + :param devices: List of Device resources. + :return: FHIR Bundle dictionary. + """ + entries = [] + for device in devices: + device_id = device.get("id", "unknown") + entries.append( + { + "fullUrl": f"https://sandbox.api.service.nhs.uk/spine-directory/FHIR/R4/Device/{device_id}", + "resource": device, + "search": {"mode": "match"}, + } + ) + + return { + "resourceType": "Bundle", + "type": "searchset", + "total": len(devices), + "entry": entries, + } + + def _build_endpoint_bundle(self, endpoints: list[dict[str, Any]]) -> dict[str, Any]: + """ + Build a FHIR Bundle from a list of Endpoint resources. + + :param endpoints: List of Endpoint resources. + :return: FHIR Bundle dictionary. + """ + entries = [] + for endpoint in endpoints: + endpoint_id = endpoint.get("id", "unknown") + entries.append( + { + "fullUrl": f"https://sandbox.api.service.nhs.uk/spine-directory/FHIR/R4/Endpoint/{endpoint_id}", + "resource": endpoint, + "search": {"mode": "match"}, + } + ) + + return { + "resourceType": "Bundle", + "type": "searchset", + "total": len(endpoints), + "entry": entries, + } + + @staticmethod + def _extract_param_value(param: str, system: str) -> str | None: + """ + Extract the value from a FHIR-style parameter like 'system|value'. + + :param param: Parameter string in format 'system|value'. + :param system: Expected system URL. + :return: The value part, or None if not found. + """ + if not param or "|" not in param: + return None + + parts = param.split("|", 1) + if len(parts) != 2: + return None + + param_system, param_value = parts + if param_system == system: + return param_value.strip() + + return None + + def _error_response( + self, status_code: int, headers: dict[str, str], message: str + ) -> Response: + """ + Build an error response. + + :param status_code: HTTP status code. + :param headers: Response headers. + :param message: Error message. + :return: A :class:`requests.Response` with error details. + """ + body = { + "resourceType": "OperationOutcome", + "issue": [ + { + "severity": "error", + "code": "invalid", + "diagnostics": message, + } + ], + } + return self._create_response( + status_code=status_code, headers=dict(headers), json_data=body + ) diff --git a/gateway-api/tests/conftest.py b/gateway-api/tests/conftest.py index 7fef2c54..4f4bf8c7 100644 --- a/gateway-api/tests/conftest.py +++ b/gateway-api/tests/conftest.py @@ -30,7 +30,7 @@ def send_to_get_structured_record_endpoint( url = f"{self.base_url}/patient/$gpc.getstructuredrecord" default_headers = { "Content-Type": "application/fhir+json", - "Ods-from": "test-ods-code", + "Ods-from": "CONSUMER", "Ssp-TraceID": "test-trace-id", } if headers: diff --git a/gateway-api/tests/integration/test_sds_search.py b/gateway-api/tests/integration/test_sds_search.py new file mode 100644 index 00000000..c12623d2 --- /dev/null +++ b/gateway-api/tests/integration/test_sds_search.py @@ -0,0 +1,99 @@ +"""Integration tests for SDS (Spine Directory Service) search functionality.""" + +from __future__ import annotations + +import pytest +from gateway_api.sds_search import SdsClient, SdsSearchResults +from stubs.stub_sds import SdsFhirApiStub + + +@pytest.fixture +def sds_stub() -> SdsFhirApiStub: + """ + Create and return an SDS stub instance with default seeded data. + + :return: SdsFhirApiStub instance with PROVIDER and CONSUMER organizations. + """ + return SdsFhirApiStub() + + +@pytest.fixture +def sds_client(sds_stub: SdsFhirApiStub) -> SdsClient: + """ + Create an SdsClient configured to use the stub. + + :param sds_stub: SDS stub fixture. + :return: SdsClient configured with test stub. + """ + client = SdsClient(api_key="test-integration-key", base_url="http://stub") + # Override the get_method to use the stub + client.get_method = sds_stub.get + return client + + +class TestSdsIntegration: + """Integration tests for SDS search operations.""" + + def test_get_device_by_ods_code_returns_valid_asid( + self, sds_client: SdsClient + ) -> None: + """ + Test that querying by ODS code returns a valid ASID. + + :param sds_client: SDS client fixture configured with stub. + """ + result = sds_client.get_org_details(ods_code="PROVIDER") + + assert result is not None + assert isinstance(result, SdsSearchResults) + assert result.asid is not None + assert result.asid == "asid_PROV" + assert len(result.asid) > 0 + + def test_get_device_with_party_key_returns_endpoint( + self, sds_client: SdsClient + ) -> None: + """ + Test that a device with party key returns both ASID and endpoint. + + :param sds_client: SDS client fixture configured with stub. + """ + result = sds_client.get_org_details(ods_code="PROVIDER") + + assert result is not None + assert result.asid == "asid_PROV" + assert result.endpoint is not None + assert result.endpoint == "https://provider.example.com/fhir" + # Verify endpoint is a valid URL format + assert result.endpoint.startswith("https://") + assert "fhir" in result.endpoint + + def test_consumer_organization_lookup(self, sds_client: SdsClient) -> None: + """ + Test that CONSUMER organization can be looked up successfully. + + :param sds_client: SDS client fixture configured with stub. + """ + result = sds_client.get_org_details(ods_code="CONSUMER") + + assert result is not None + assert result.asid == "asid_CONS" + assert result.endpoint is not None + assert result.endpoint == "https://consumer.example.com/fhir" + + def test_result_contains_both_asid_and_endpoint_when_available( + self, sds_client: SdsClient + ) -> None: + """ + Test that results contain both ASID and endpoint when both are available. + + :param sds_client: SDS client fixture configured with stub. + """ + result = sds_client.get_org_details(ods_code="PROVIDER") + + assert result is not None + # Verify both fields are present and not None + assert hasattr(result, "asid") + assert hasattr(result, "endpoint") + assert result.asid is not None + assert result.endpoint is not None