HCW-345 add sandbox-only lambda handler

infrastructure/modules/hcw-api/lambda.tf

@@ -19,7 +19,7 @@ resource "aws_lambda_function" "hcw-app" {
 
   s3_bucket   = data.aws_s3_bucket.app_deployment.id
   s3_key      = var.s3_filename
-  handler     = "main.lambda_handler"
+  handler     = var.sandbox_mode ? "sandbox_main.lambda_handler" : "main.lambda_handler"
   timeout     = 30
   memory_size = 512
 
@@ -41,7 +41,6 @@ resource "aws_lambda_function" "hcw-app" {
     variables = {
       LDAP_CREDENTIALS_SECRET_ID = data.aws_secretsmanager_secret.ldap_credentials.arn
       LDAP_GATEWAY_URL           = var.ldap_gateway_url
-      SANDBOX_MODE               = var.sandbox_mode
       BASE_URL                   = "https://${var.apim_environment}.api.service.nhs.uk/healthcare-worker"
 
       # Extension configuration (optional - using defaults)

scripts/build-app.sh

@@ -10,4 +10,6 @@ cd dist || exit
 tar -xvf hcw_api-*.tar.gz
 
 cd hcw_api-*/src || exit
+mkdir -p specification/components/examples
+cp ../../../specification/components/examples/*.json specification/components/examples/
 zip -r ../../../hcw-api.zip . -x "*.pyc" -x "*__pycache__*"

src/ldap/connection_fetch.py

@@ -1,9 +1,7 @@
-import os
 from datetime import datetime, timedelta
 from typing import Optional
 
 from ldap.connection import HcwLdapConnection
-from ldap.mock_connection import MockHcwLdapConnection
 from ldap.real_connection import RealHcwLdapConnection
 from logs.log import Log
 
@@ -14,28 +12,13 @@
 ldap_connection: Optional[HcwLdapConnection] = None
 
 
-def is_sandbox_mode() -> bool:
-    """
-    Check if we're running in sandbox mode.
-    """
-    return os.environ.get("SANDBOX_MODE", "false").lower() == "true"
-
-
 def get_connection():
     global ldap_connection
 
-    if is_sandbox_mode():
-        # In sandbox mode, just create mock connection if we don't have one
-        # No need for sophisticated connection pooling with mocks
-        if not ldap_connection:
-            logger.info("Creating new mock connection for sandbox", "LDAP_CONN_MOCK", "null")
-            ldap_connection = MockHcwLdapConnection()
-    else:
-        # In real mode, do full connection pooling logic
-        if (not ldap_connection or ldap_connection.connection.closed
-                or ldap_connection.bind_time < datetime.now() - timedelta(minutes=5)):
-            logger.info("Creating new real LDAP connection", "LDAP_CONN_REAL", "null")
-            ldap_connection = RealHcwLdapConnection()
+    if (not ldap_connection or ldap_connection.connection.closed
+            or ldap_connection.bind_time < datetime.now() - timedelta(minutes=5)):
+        logger.info("Creating new real LDAP connection", "LDAP_CONN_REAL", "null")
+        ldap_connection = RealHcwLdapConnection()
 
     return ldap_connection
 

src/ldap/connection_fetch_test.py

@@ -5,7 +5,6 @@
 
 import ldap.connection_fetch
 from ldap.connection_fetch import get_connection
-from ldap.mock_connection import MockHcwLdapConnection
 
 
 @pytest.fixture(autouse=True)
@@ -15,23 +14,23 @@ def tidy_up():
     ldap.connection_fetch.ldap_connection = None
 
 
-@patch.dict(os.environ, {"SANDBOX_MODE": "true"})
-def test_sandbox_connection():
+@patch("ldap.connection_fetch.RealHcwLdapConnection")
+def test_real_connection(connection_mock):
     connection = get_connection()
-    assert isinstance(connection, MockHcwLdapConnection)
+    assert connection == connection_mock.return_value
 
 
-@patch.dict(os.environ, {"SANDBOX_MODE": "false"})
 @patch("ldap.connection_fetch.RealHcwLdapConnection")
-def test_real_connection(connection_mock):
-    connection = get_connection()
+def test_real_connection_even_if_sandbox_mode_is_true(connection_mock):
+    with patch.dict(os.environ, {"SANDBOX_MODE": "true"}):
+        connection = get_connection()
     assert connection == connection_mock.return_value
 
 
-@patch.dict(os.environ, {"SANDBOX_MODE": "non_boolean_value"})
 @patch("ldap.connection_fetch.RealHcwLdapConnection")
 def test_real_connection_with_unknown_value(connection_mock):
-    connection = get_connection()
+    with patch.dict(os.environ, {"SANDBOX_MODE": "non_boolean_value"}):
+        connection = get_connection()
     assert connection == connection_mock.return_value
 
 

src/request_handlers/handlers.py

@@ -1,4 +1,3 @@
-import os
 from typing import Dict, Type
 
 from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent

src/request_handlers/sandbox_responses_test.py

@@ -0,0 +1,142 @@
+import json
+from pathlib import Path
+
+from sandbox_main import lambda_handler
+
+EXAMPLES_DIR = Path(__file__).resolve().parents[2] / "specification" / "components" / "examples"
+BASE_URL = "https://int.api.service.nhs.uk/healthcare-worker"
+SANDBOX_UUID = "123456789012"
+
+
+class LambdaContext:
+    pass
+
+
+def load_example(file_name: str) -> dict:
+    with open(EXAMPLES_DIR / file_name, encoding="utf-8") as example_file:
+        return json.load(example_file)
+
+
+def sort_bundle_entries(bundle: dict) -> dict:
+    bundle_copy = dict(bundle)
+    bundle_copy["entry"] = sorted(
+        bundle_copy.get("entry", []),
+        key=lambda entry: (
+            entry["search"]["mode"],
+            entry["resource"]["resourceType"],
+            entry["resource"].get("id", ""),
+            entry["fullUrl"],
+        ),
+    )
+    return bundle_copy
+
+
+def sandbox_event(resource: str, query_string_parameters: dict, multi_value_query_string_parameters: dict | None = None) -> dict:
+    return {
+        "resource": resource,
+        "queryStringParameters": query_string_parameters,
+        "multiValueQueryStringParameters": multi_value_query_string_parameters or {
+            key: [value] if not isinstance(value, list) else value
+            for key, value in query_string_parameters.items()
+        },
+    }
+
+
+def test_sandbox_practitioner_basic_matches_spec_example():
+    response = lambda_handler(
+        sandbox_event("/Practitioner", {"identifier": SANDBOX_UUID}),
+        LambdaContext(),
+    )
+
+    assert response["statusCode"] == 200
+    assert sort_bundle_entries(json.loads(response["body"])) == sort_bundle_entries(
+        load_example("GetHealthcareWorkerDetailsResponseSuccessBasic.json")
+    )
+
+
+def test_sandbox_practitioner_with_revinclude_matches_spec_example():
+    response = lambda_handler(
+        sandbox_event(
+            "/Practitioner",
+            {"identifier": SANDBOX_UUID, "_revinclude": "PractitionerRole:practitioner"},
+        ),
+        LambdaContext(),
+    )
+
+    assert response["statusCode"] == 200
+    assert sort_bundle_entries(json.loads(response["body"])) == sort_bundle_entries(
+        load_example("GetHealthcareWorkerDetailsResponseSuccessWithIncludes.json")
+    )
+
+
+def test_sandbox_practitioner_role_basic_matches_spec_example():
+    response = lambda_handler(
+        sandbox_event("/PractitionerRole", {"practitioner.identifier": SANDBOX_UUID}),
+        LambdaContext(),
+    )
+
+    assert response["statusCode"] == 200
+    assert sort_bundle_entries(json.loads(response["body"])) == sort_bundle_entries(
+        load_example("GetHealthcareWorkerRoleDetailsResponseSuccessBasic.json")
+    )
+
+
+def test_sandbox_practitioner_role_with_include_matches_spec_example():
+    response = lambda_handler(
+        sandbox_event(
+            "/PractitionerRole",
+            {"practitioner.identifier": SANDBOX_UUID, "_include": "PractitionerRole:practitioner"},
+        ),
+        LambdaContext(),
+    )
+
+    assert response["statusCode"] == 200
+    assert sort_bundle_entries(json.loads(response["body"])) == sort_bundle_entries(
+        load_example("GetHealthcareWorkerRoleDetailsResponseSuccessWithIncludes.json")
+    )
+
+
+def test_sandbox_practitioner_unknown_uuid_returns_not_found():
+    response = lambda_handler(
+        sandbox_event("/Practitioner", {"identifier": "999999999999"}),
+        LambdaContext(),
+    )
+
+    assert response["statusCode"] == 404
+    assert json.loads(response["body"]) == {
+        "resourceType": "OperationOutcome",
+        "issue": [{
+            "severity": "error",
+            "code": "unknown",
+            "details": {
+                "coding": [{
+                    "system": "https://fhir.nhs.uk/STU3/ValueSet/Spine-ErrorOrWarningCode-1",
+                    "code": "404",
+                    "display": "User with id 999999999999 not found",
+                }]
+            },
+        }],
+    }
+
+
+def test_sandbox_practitioner_role_unknown_uuid_returns_not_found():
+    response = lambda_handler(
+        sandbox_event("/PractitionerRole", {"practitioner.identifier": "999999999999"}),
+        LambdaContext(),
+    )
+
+    assert response["statusCode"] == 404
+    assert json.loads(response["body"]) == {
+        "resourceType": "OperationOutcome",
+        "issue": [{
+            "severity": "error",
+            "code": "unknown",
+            "details": {
+                "coding": [{
+                    "system": "https://fhir.nhs.uk/STU3/ValueSet/Spine-ErrorOrWarningCode-1",
+                    "code": "404",
+                    "display": "User with id 999999999999 not found",
+                }]
+            },
+        }],
+    }

src/request_handlers/sandbox_static_responses.py

@@ -0,0 +1,77 @@
+import json
+from pathlib import Path
+
+from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
+
+SANDBOX_UUID = "123456789012"
+EXAMPLES_SUBPATH = Path("specification") / "components" / "examples"
+
+
+def get_sandbox_response(event: APIGatewayProxyEvent) -> tuple[int, str] | None:
+    query_parameters = event.query_string_parameters or {}
+
+    if event.resource == "/Practitioner":
+        return _get_practitioner_response(query_parameters)
+
+    if event.resource == "/PractitionerRole":
+        return _get_practitioner_role_response(query_parameters)
+
+    return None
+
+
+def _get_practitioner_response(query_parameters: dict[str, str]) -> tuple[int, str]:
+    practitioner_id = query_parameters.get("identifier")
+    if practitioner_id != SANDBOX_UUID:
+        return _not_found_response(practitioner_id)
+
+    if query_parameters.get("_revinclude") == "PractitionerRole:practitioner":
+        return 200, _load_example("GetHealthcareWorkerDetailsResponseSuccessWithIncludes.json")
+
+    return 200, _load_example("GetHealthcareWorkerDetailsResponseSuccessBasic.json")
+
+
+def _get_practitioner_role_response(query_parameters: dict[str, str]) -> tuple[int, str]:
+    practitioner_id = query_parameters.get("practitioner.identifier")
+    if practitioner_id != SANDBOX_UUID:
+        return _not_found_response(practitioner_id)
+
+    if query_parameters.get("_include") == "PractitionerRole:practitioner":
+        return 200, _load_example("GetHealthcareWorkerRoleDetailsResponseSuccessWithIncludes.json")
+
+    return 200, _load_example("GetHealthcareWorkerRoleDetailsResponseSuccessBasic.json")
+
+
+def _not_found_response(practitioner_id: str | None) -> tuple[int, str]:
+    response = {
+        "resourceType": "OperationOutcome",
+        "issue": [{
+            "severity": "error",
+            "code": "unknown",
+            "details": {
+                "coding": [{
+                    "system": "https://fhir.nhs.uk/STU3/ValueSet/Spine-ErrorOrWarningCode-1",
+                    "code": "404",
+                    "display": f"User with id {practitioner_id} not found",
+                }]
+            },
+        }],
+    }
+
+    return 404, json.dumps(response)
+
+
+def _load_example(file_name: str) -> str:
+    with open(_get_examples_directory() / file_name, encoding="utf-8") as example_file:
+        return example_file.read()
+
+
+def _get_examples_directory() -> Path:
+    current_file = Path(__file__).resolve()
+    candidate_roots = [current_file.parents[2], current_file.parents[1]]
+
+    for root in candidate_roots:
+        examples_directory = root / EXAMPLES_SUBPATH
+        if examples_directory.exists():
+            return examples_directory
+
+    raise FileNotFoundError(f"Could not locate sandbox example files under {EXAMPLES_SUBPATH}")