Feature/gpcapim 251

gateway-api/src/gateway_api/provider_request.py

@@ -0,0 +1,146 @@
+"""
+Module: gateway_api.provider_request
+
+This module contains the GpProviderClient class, which provides a
+simple client for interacting with the GPProvider FHIR GP System.
+
+The GpProviderClient class includes methods to fetch structured patient
+records from a GPProvider FHIR API endpoint.
+
+Usage:
+    Instantiate a GpProviderClient with:
+        - provider_endpoint: The FHIR API endpoint for the provider.
+        - provider_asid: The ASID for the provider.
+        - consumer_asid: The ASID for the consumer.
+
+    Use the `access_structured_record` method to fetch a structured patient record:
+        Parameters:
+            - trace_id (str): A unique identifier for the request.
+            - body (str): The request body in FHIR format.
+
+    Returns:
+        The response from the provider FHIR API.
+"""
+
+from collections.abc import Callable
+from urllib.parse import urljoin
+
+from requests import HTTPError, Response
+from stubs.stub_provider import GpProviderStub
+
+ARS_INTERACTION_ID = (
+    "urn:nhs:names:services:gpconnect:structured"
+    ":fhir:operation:gpc.getstructuredrecord-1"
+)
+ARS_FHIR_BASE = "FHIR/STU3"
+FHIR_RESOURCE = "patient"
+ARS_FHIR_OPERATION = "$gpc.getstructuredrecord"
+TIMEOUT: int | None = None  # None used for quicker dev, adjust as needed
+
+# Direct all requests to the stub provider for steel threading in dev.
+# Replace with `from requests import post` for real requests.
+PostCallable = Callable[..., Response]
+_provider_stub = GpProviderStub()
+
+
+def _stubbed_post(trace_id: str, body: str) -> Response:
+    """A stubbed requests.post function that routes to the GPProviderStub."""
+    return _provider_stub.access_record_structured(trace_id, body)
+
+
+post: PostCallable = _stubbed_post
+
+
+class ExternalServiceError(Exception):
+    """
+    Exception raised when the downstream GPProvider FHIR API request fails.
+    """
+
+
+class GpProviderClient:
+    """
+    A client for interacting with the GPProvider FHIR GP System.
+
+    This class provides methods to interact with the GPProvider FHIR API,
+    including fetching structured patient records.
+
+    Attributes:
+        provider_endpoint (str): The FHIR API endpoint for the provider.
+        provider_asid (str): The ASID for the provider.
+        consumer_asid (str): The ASID for the consumer.
+
+    Methods:
+        access_structured_record(trace_id: str, body: str) -> Response:
+            Fetch a structured patient record from the GPProvider FHIR API.
+    """
+
+    def __init__(
+        self,
+        provider_endpoint: str,
+        provider_asid: str,
+        consumer_asid: str,
+    ) -> None:
+        self.provider_endpoint = provider_endpoint
+        self.provider_asid = provider_asid
+        self.consumer_asid = consumer_asid
+
+    def _build_headers(self, trace_id: str) -> dict[str, str]:
+        """
+        Build the headers required for the GPProvider FHIR API request.
+
+        Args:
+            trace_id (str): A unique identifier for the request.
+
+        Returns:
+            dict[str, str]: A dictionary containing the headers for the request,
+            including content type, interaction ID, and ASIDs for the provider
+            and consumer.
+        """
+        return {
+            "Content-Type": "application/fhir+json",
+            "Accept": "application/fhir+json",
+            "Ssp-InteractionID": ARS_INTERACTION_ID,
+            "Ssp-To": self.provider_asid,
+            "Ssp-From": self.consumer_asid,
+            "Ssp-TraceID": trace_id,
+        }
+
+    def access_structured_record(
+        self,
+        trace_id: str,
+        body: str,
+    ) -> Response:
+        """
+        Fetch a structured patient record from the GPProvider FHIR API.
+
+        Args:
+            trace_id (str): A unique identifier for the request, passed in the headers.
+            body (str): The request body in FHIR format.
+
+        Returns:
+            Response: The response from the GPProvider FHIR API.
+
+        Raises:
+            ExternalServiceError: If the API request fails with an HTTP error.
+        """
+
+        headers = self._build_headers(trace_id)
+
+        endpoint_path = "/".join([ARS_FHIR_BASE, FHIR_RESOURCE, ARS_FHIR_OPERATION])
+        url = urljoin(self.provider_endpoint, endpoint_path)
+
+        response = post(
+            url,
+            headers=headers,
+            data=body,
+            timeout=TIMEOUT,
+        )
+
+        try:
+            response.raise_for_status()
+        except HTTPError as err:
+            raise ExternalServiceError(
+                f"GPProvider FHIR API request failed:{err.response.reason}"
+            ) from err
+
+        return response

gateway-api/src/gateway_api/test_provider_request.py

@@ -0,0 +1,240 @@
+"""
+Unit tests for :mod:`gateway_api.provider_request`.
+
+This module contains unit tests for the `GpProviderClient` class, which is responsible
+for interacting with the GPProvider FHIR API.
+
+"""
+
+from typing import Any
+
+import pytest
+from requests import Response
+from requests.structures import CaseInsensitiveDict
+from stubs.stub_provider import GpProviderStub
+
+from gateway_api import provider_request
+from gateway_api.provider_request import ExternalServiceError, GpProviderClient
+
+ars_interactionId = (
+    "urn:nhs:names:services:gpconnect:structured"
+    ":fhir:operation:gpc.getstructuredrecord-1"
+)
+
+
+@pytest.fixture
+def stub() -> GpProviderStub:
+    return GpProviderStub()
+
+
+@pytest.fixture
+def mock_request_post(
+    monkeypatch: pytest.MonkeyPatch, stub: GpProviderStub
+) -> dict[str, Any]:
+    """
+    Fixture to patch the `requests.post` method for testing.
+
+    This fixture intercepts calls to `requests.post` and routes them to the
+    stub provider. It also captures the most recent request details, such as
+    headers, body, and URL, for verification in tests.
+
+    Returns:
+        dict[str, Any]: A dictionary containing the captured request details.
+    """
+    capture: dict[str, Any] = {}
+
+    def _fake_post(
+        url: str,
+        headers: CaseInsensitiveDict[str],
+        data: str,
+        timeout: int,
+    ) -> Response:
+        """A fake requests.post implementation."""
+
+        capture["headers"] = dict(headers)
+        capture["data"] = data
+        capture["url"] = url
+
+        # Provide dummy or captured arguments as required by the stub signature
+        return stub.access_record_structured(
+            trace_id=headers.get("Ssp-TraceID", "dummy-trace-id"), body=data
+        )
+
+    monkeypatch.setattr(provider_request, "post", _fake_post)
+    return capture
+
+
+def test_valid_gpprovider_access_structured_record_makes_request_correct_url_post_200(
+    mock_request_post: dict[str, Any],
+    stub: GpProviderStub,
+) -> None:
+    """
+    Test that the `access_structured_record` method constructs the correct URL
+    for the GPProvider FHIR API request and receives a 200 OK response.
+
+    This test verifies that the URL includes the correct FHIR base path and
+    operation for accessing a structured patient record.
+    """
+    provider_asid = "200000001154"
+    consumer_asid = "200000001152"
+    provider_endpoint = "https://test.com"
+    trace_id = "some_uuid_value"
+
+    client = GpProviderClient(
+        provider_endpoint=provider_endpoint,
+        provider_asid=provider_asid,
+        consumer_asid=consumer_asid,
+    )
+
+    result = client.access_structured_record(trace_id, "body")
+
+    captured_url = mock_request_post.get("url", provider_endpoint)
+
+    assert (
+        captured_url
+        == provider_endpoint + "/FHIR/STU3/patient/$gpc.getstructuredrecord"
+    )
+    assert result.status_code == 200
+
+
+def test_valid_gpprovider_access_structured_record_with_correct_headers_post_200(
+    mock_request_post: dict[str, Any],
+    stub: GpProviderStub,
+) -> None:
+    """
+    Test that the `access_structured_record` method includes the correct headers
+    in the GPProvider FHIR API request and receives a 200 OK response.
+
+    This test verifies that the headers include:
+        - Content-Type and Accept headers for FHIR+JSON.
+        - Ssp-TraceID, Ssp-From, Ssp-To, and Ssp-InteractionID for GPConnect.
+    """
+    provider_asid = "200000001154"
+    consumer_asid = "200000001152"
+    provider_endpoint = "https://test.com"
+    trace_id = "some_uuid_value"
+
+    client = GpProviderClient(
+        provider_endpoint=provider_endpoint,
+        provider_asid=provider_asid,
+        consumer_asid=consumer_asid,
+    )
+    expected_headers = {
+        "Content-Type": "application/fhir+json",
+        "Accept": "application/fhir+json",
+        "Ssp-TraceID": str(trace_id),
+        "Ssp-From": consumer_asid,
+        "Ssp-To": provider_asid,
+        "Ssp-InteractionID": ars_interactionId,
+    }
+
+    result = client.access_structured_record(trace_id, "body")
+
+    captured_headers = mock_request_post["headers"]
+
+    assert expected_headers == captured_headers
+    assert result.status_code == 200
+
+
+def test_valid_gpprovider_access_structured_record_with_correct_body_200(
+    mock_request_post: dict[str, Any],
+    stub: GpProviderStub,
+) -> None:
+    """
+    Test that the `access_structured_record` method includes the correct body
+    in the GPProvider FHIR API request and receives a 200 OK response.
+
+    This test verifies that the request body matches the expected FHIR parameters
+    resource sent to the GPProvider API.
+    """
+    provider_asid = "200000001154"
+    consumer_asid = "200000001152"
+    provider_endpoint = "https://test.com"
+    trace_id = "some_uuid_value"
+
+    request_body = "some_FHIR_request_params"
+
+    client = GpProviderClient(
+        provider_endpoint=provider_endpoint,
+        provider_asid=provider_asid,
+        consumer_asid=consumer_asid,
+    )
+
+    result = client.access_structured_record(trace_id, request_body)
+
+    captured_body = mock_request_post["data"]
+
+    assert result.status_code == 200
+    assert captured_body == request_body
+
+
+def test_valid_gpprovider_access_structured_record_returns_stub_response_200(
+    mock_request_post: dict[str, Any],
+    stub: GpProviderStub,
+) -> None:
+    """
+    Test that the `access_structured_record` method returns the same response
+    as provided by the stub provider.
+
+    This test verifies that the response from the GPProvider FHIR API matches
+    the expected response, including the status code and content.
+    """
+    provider_asid = "200000001154"
+    consumer_asid = "200000001152"
+    provider_endpoint = "https://test.com"
+    trace_id = "some_uuid_value"
+
+    client = GpProviderClient(
+        provider_endpoint=provider_endpoint,
+        provider_asid=provider_asid,
+        consumer_asid=consumer_asid,
+    )
+
+    expected_response = stub.access_record_structured(trace_id, "body")
+
+    result = client.access_structured_record(trace_id, "body")
+
+    assert result.status_code == 200
+    assert result.content == expected_response.content
+
+
+def test_access_structured_record_raises_external_service_error(
+    mock_request_post: dict[str, Any],
+    stub: GpProviderStub,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    Test that the `access_structured_record` method raises an `ExternalServiceError`
+    when the GPProvider FHIR API request fails with an HTTP error.
+    """
+    provider_asid = "200000001154"
+    consumer_asid = "200000001152"
+    provider_endpoint = "https://test.com"
+    trace_id = "invalid for test"
+
+    client = GpProviderClient(
+        provider_endpoint=provider_endpoint,
+        provider_asid=provider_asid,
+        consumer_asid=consumer_asid,
+    )
+
+    with pytest.raises(
+        ExternalServiceError,
+        match="GPProvider FHIR API request failed:Bad Request",
+    ):
+        client.access_structured_record(trace_id, "body")
+
+
+def test_stubbed_post_function(stub: GpProviderStub) -> None:
+    """
+    Test the `_stubbed_post` function to ensure it routes to the stub provider.
+    """
+    trace_id = "test-trace-id"
+    body = "test-body"
+
+    # Call the `_stubbed_post` function
+    response = provider_request._stubbed_post(trace_id, body)  # noqa: SLF001 this is testing the private method
+
+    # Verify the response is as expected
+    assert response.status_code == 200
+    assert response.json() == stub.access_record_structured(trace_id, body).json()