VED-1260 Mock PDS lambda rate limiting in ref env

.github/workflows/pr-deploy-and-test.yml

@@ -20,6 +20,10 @@ jobs:
     uses: ./.github/workflows/deploy-backend.yml
     with:
       apigee_environment: internal-dev
+      lambda_build_flags: >-
+        ${{ (github.event.action == 'opened' || github.event.action == 'reopened')
+            && '{"recordprocessor":true,"ack-backend":true}'
+            || '{}' }}
       diff_base_sha: ${{ github.event.action == 'synchronize' && github.event.before || github.event.pull_request.base.sha }}
       diff_head_sha: ${{ github.event.pull_request.head.sha }}
       run_diff_check: ${{ github.event.action == 'synchronize' }}
@@ -35,15 +39,19 @@ jobs:
         include:
           - apigee_environment_name: internal-dev
             required_test_suite: smoke
+            require_matching_commit_id: true
           - apigee_environment_name: internal-dev-sandbox
             required_test_suite: sandbox
+            require_matching_commit_id: false
     uses: ./.github/workflows/run-e2e-automation-tests.yml
     with:
       apigee_environment: ${{ matrix.apigee_environment_name }}
       environment: dev
       sub_environment: pr-${{github.event.pull_request.number}}
       service_under_test: all
       suite_to_run: ${{ matrix.required_test_suite }}
+      expected_commit_id: ${{ github.sha }}
+      require_matching_commit_id: ${{ matrix.require_matching_commit_id }}
     secrets:
       APIGEE_PASSWORD: ${{ secrets.APIGEE_PASSWORD }}
       APIGEE_BASIC_AUTH_TOKEN: ${{ secrets.APIGEE_BASIC_AUTH_TOKEN }}

.github/workflows/quality-checks.yml

@@ -9,6 +9,7 @@ on:
 env:
   SHARED_PATH: ${{ github.workspace }}/lambdas/shared
   LAMBDA_PATH: ${{ github.workspace }}/lambdas
+  POETRY_INSTALLER_ONLY_BINARY: ":all:"
 
 jobs:
   lint-specification:
@@ -156,6 +157,7 @@ jobs:
         id: id_sync
         env:
           PYTHONPATH: ${{ env.LAMBDA_PATH }}/id_sync/src:${{ env.LAMBDA_PATH }}/id_sync/tests:${{ env.SHARED_PATH }}/src
+          POETRY_INSTALLER_ONLY_BINARY: ""
         continue-on-error: true
         run: |
           poetry install
@@ -178,6 +180,7 @@ jobs:
         id: mnspublisher
         env:
           PYTHONPATH: ${{ env.LAMBDA_PATH }}/mns_publisher/src:${{ env.LAMBDA_PATH }}/mns_publisher/tests:${{ env.SHARED_PATH }}/src
+          POETRY_INSTALLER_ONLY_BINARY: ""
         continue-on-error: true
         run: |
           poetry install
@@ -234,11 +237,24 @@ jobs:
         id: shared
         env:
           PYTHONPATH: ${{ env.SHARED_PATH }}/src
+          POETRY_INSTALLER_ONLY_BINARY: ""
+          GITHUB_WORKSPACE: ${{ github.workspace }}
         continue-on-error: true
         run: |
           poetry install
           poetry run coverage run --rcfile=.coveragerc --source=src -m unittest discover -s tests -p "test_*.py" -v || echo "shared tests failed" >> ../../failed_tests.txt
           poetry run coverage xml -o ../../shared-coverage.xml
+          python - <<'PY'
+          import os
+          import re
+          from pathlib import Path
+
+          coverage_file = Path("../../shared-coverage.xml")
+          xml = coverage_file.read_text()
+          xml = re.sub(r"<source>.*?</source>", f"<source>{os.environ['GITHUB_WORKSPACE']}</source>", xml, count=1)
+          xml = xml.replace('filename="src/', 'filename="lambdas/shared/src/')
+          coverage_file.write_text(xml)
+          PY
 
       - name: Run Test Failure Summary
         id: check_failure

.github/workflows/run-e2e-automation-tests.yml

@@ -18,6 +18,14 @@ on:
       suite_to_run:
         required: true
         type: string
+      expected_commit_id:
+        required: false
+        type: string
+        default: ""
+      require_matching_commit_id:
+        required: false
+        type: boolean
+        default: true
     secrets:
       APIGEE_PASSWORD:
         required: true
@@ -72,6 +80,16 @@ on:
         description: Set to true if you want the MNS validation to be performed as part of the tests. please keep in mind it will increase execution time.
         default: false
         type: boolean
+      expected_commit_id:
+        description: Optional commit SHA expected from the deployed _status endpoint.
+        required: false
+        type: string
+        default: ""
+      require_matching_commit_id:
+        description: Whether the _status endpoint commitId must match the expected commit SHA.
+        required: false
+        type: boolean
+        default: true
 
 env:
   APIGEE_AUTH_ENV: ${{ inputs.apigee_environment == 'int' && inputs.apigee_environment || 'internal-dev' }}
@@ -81,7 +99,8 @@ env:
   SERVICE_BASE_PATH: ${{ startsWith(inputs.sub_environment, 'pr-') && format('immunisation-fhir-api/FHIR/R4-{0}', inputs.sub_environment) || 'immunisation-fhir-api/FHIR/R4' }}
   PROXY_NAME: ${{ startsWith(inputs.sub_environment, 'pr-') && format('immunisation-fhir-api-{0}', inputs.sub_environment) || format('immunisation-fhir-api-{0}', inputs.apigee_environment) }}
   STATUS_API_KEY: ${{ secrets.STATUS_API_KEY }}
-  SOURCE_COMMIT_ID: ${{ github.sha }}
+  SOURCE_COMMIT_ID: ${{ inputs.expected_commit_id || github.sha }}
+  REQUIRE_MATCHING_COMMIT_ID: ${{ inputs.require_matching_commit_id }}
   MNS_VALIDATION_REQUIRED: ${{ inputs.mns_validation_required || startsWith(inputs.sub_environment, 'pr-') || inputs.apigee_environment == 'internal-dev' }}
 
 jobs:
@@ -94,7 +113,6 @@ jobs:
       - name: Wait for API to be available
         if: github.event_name != 'workflow_dispatch'
         run: |
-          endpoint=""
           if [[ ${APIGEE_ENVIRONMENT} =~ "prod" ]]; then
             endpoint="https://api.service.nhs.uk/${SERVICE_BASE_PATH}/_status"
           else
@@ -112,17 +130,21 @@ jobs:
 
             if [[ "${response_code}" -eq 200 ]] && [[ "${response_body}" == "OK" ]] && [[ "${status}" == "pass" ]]; then
               echo "Status test successful"
+              if [[ "${REQUIRE_MATCHING_COMMIT_ID}" != "true" ]]; then
+                echo "Skipping commit hash validation for ${APIGEE_ENVIRONMENT}"
+                break
+              fi
               if [[ "${commitId}" == "${SOURCE_COMMIT_ID}" ]]; then
                 echo "Commit hash test successful"
                 break
               else
-                echo "Waiting for ${endpoint} to return the correct commit hash..."
+                echo "Waiting for ${endpoint} to return the correct commit hash... expected ${SOURCE_COMMIT_ID}, got ${commitId}"
               fi
             else
               echo "Waiting for ${endpoint} to return a 200 response with 'OK' body..."
             fi
 
-            ((counter=counter+1)) # Increment counter by 1
+            ((counter++))
             echo "Attempt ${counter}"
             sleep 30
           done

infrastructure/instance/environments/dev/ref/variables.tfvars

@@ -2,7 +2,7 @@ environment                       = "dev"
 immunisation_account_id           = "345594581768"
 dspp_core_account_id              = "603871901111"
 pds_environment                   = "ref"
-mns_environment                   = "dev"
+mns_environment                   = "int"
 error_alarm_notifications_enabled = true
 create_mesh_processor             = false
 has_sub_environment_scope         = true

infrastructure/instance/id_sync_lambda.tf

@@ -213,6 +213,7 @@ resource "aws_lambda_function" "id_sync_lambda" {
     variables = {
       IEDS_TABLE_NAME      = aws_dynamodb_table.events-dynamodb-table.name
       PDS_ENV              = var.pds_environment
+      PDS_BASE_URL         = ""
       SPLUNK_FIREHOSE_NAME = module.splunk.firehose_stream_name
     }
   }

infrastructure/instance/mns_publisher.tf

@@ -15,6 +15,7 @@ module "mns_publisher" {
   secrets_manager_policy_path        = "${local.policy_path}/secret_manager.json"
   account_id                         = data.aws_caller_identity.current.account_id
   pds_environment                    = var.pds_environment
+  pds_base_url                       = ""
   mns_environment                    = var.mns_environment
 
   private_subnet_ids = local.private_subnet_ids

infrastructure/instance/modules/mns_publisher/mns_publisher_lambda.tf

@@ -122,6 +122,7 @@ resource "aws_lambda_function" "mns_publisher_lambda" {
       IMMUNIZATION_ENV       = var.resource_scope,
       IMMUNIZATION_BASE_PATH = var.imms_base_path
       PDS_ENV                = var.pds_environment
+      PDS_BASE_URL           = var.pds_base_url
       MNS_ENV                = var.mns_environment
     }
   }

infrastructure/instance/modules/mns_publisher/variables.tf

@@ -94,6 +94,12 @@ variable "pds_environment" {
   type = string
 }
 
+variable "pds_base_url" {
+  type        = string
+  default     = ""
+  description = "Optional override for the PDS base URL, used by ref to route to the mock PDS endpoint."
+}
+
 variable "account_id" {
   type        = string
   description = "AWS account ID used for IAM policy templating (e.g., Secrets Manager ARNs)."

lambdas/mns_publisher/tests/test_lambda_handler.py

@@ -6,6 +6,7 @@
 import responses
 from moto import mock_aws
 
+import common.api_clients.get_pds_details as get_pds_details_module
 from lambda_handler import lambda_handler
 from process_records import extract_trace_ids, process_record, process_records
 from test_utils import generate_private_key_b64, load_sample_sqs_event
@@ -246,6 +247,7 @@ class TestLambdaHandlerIntegration(unittest.TestCase):
     def setUp(self):
         """Set up mocked AWS services and test data."""
         self.sample_sqs_record = load_sample_sqs_event()
+        get_pds_details_module._pds_service = None
         self.secrets_client = boto3.client("secretsmanager", region_name="eu-west-2")
         self.secrets_client.create_secret(
             Name="imms/pds/int/jwt-secrets",
@@ -254,6 +256,9 @@ def setUp(self):
             ),
         )
 
+    def tearDown(self):
+        get_pds_details_module._pds_service = None
+
     @responses.activate
     @patch("common.api_clients.authentication.AppRestrictedAuth.get_access_token")
     @patch("process_records.logger")
@@ -378,3 +383,49 @@ def test_successful_notification_creation_with_expired_gp(self, mock_logger, moc
         self.assertEqual(mns_payload["filtering"]["subjectage"], 21)
 
         mock_logger.info.assert_any_call("Successfully processed all 1 messages")
+
+    @responses.activate
+    @patch.dict("os.environ", {"PDS_BASE_URL": "https://mock-pds.example/Patient"}, clear=False)
+    @patch("process_records._get_runtime_mns_service")
+    @patch("process_records.logger")
+    def test_successful_notification_creation_with_mock_pds_base_url(self, mock_logger, mock_get_mns):
+        responses.add(
+            responses.GET,
+            "https://mock-pds.example/Patient/9481152782",
+            json={"generalPractitioner": [{"identifier": {"value": "Y12345", "period": {"start": "2024-01-01"}}}]},
+            status=200,
+        )
+
+        mock_mns_service = Mock()
+        mock_get_mns.return_value = mock_mns_service
+
+        sqs_event = {"Records": [self.sample_sqs_record]}
+        result = lambda_handler(sqs_event, Mock())
+
+        self.assertEqual(result, {"batchItemFailures": []})
+        mock_mns_service.publish_notification.assert_called_once()
+        mns_payload = mock_mns_service.publish_notification.call_args.args[0]
+        self.assertEqual(mns_payload["filtering"]["generalpractitioner"], "Y12345")
+        mock_logger.info.assert_any_call("Successfully processed all 1 messages")
+
+    @responses.activate
+    @patch.dict("os.environ", {"PDS_BASE_URL": "https://mock-pds.example/Patient"}, clear=False)
+    @patch("process_records._get_runtime_mns_service")
+    @patch("process_records.logger")
+    def test_mock_pds_rate_limit_results_in_batch_failure(self, mock_logger, mock_get_mns):
+        responses.add(
+            responses.GET,
+            "https://mock-pds.example/Patient/9481152782",
+            json={"code": 429, "message": "Mock PDS rate limit has been exceeded"},
+            status=429,
+        )
+
+        mock_mns_service = Mock()
+        mock_get_mns.return_value = mock_mns_service
+
+        sqs_event = {"Records": [self.sample_sqs_record]}
+        result = lambda_handler(sqs_event, Mock())
+
+        self.assertEqual(len(result["batchItemFailures"]), 1)
+        mock_mns_service.publish_notification.assert_not_called()
+        mock_logger.warning.assert_called_with("Batch completed with 1 failures")

lambdas/shared/src/common/api_clients/get_pds_details.py

@@ -9,28 +9,31 @@
 from common.api_clients.pds_service import PdsService
 from common.clients import get_secrets_manager_client, logger
 
-PDS_ENV = os.getenv("PDS_ENV", "int")
-
 _pds_service: PdsService | None = None
+_pds_service_config: tuple[str, str | None] | None = None
 
 
 def get_pds_service() -> PdsService:
-    global _pds_service
-    if _pds_service is None:
-        authenticator = AppRestrictedAuth(
-            secret_manager_client=get_secrets_manager_client(),
-            environment=PDS_ENV,
+    global _pds_service, _pds_service_config
+    environment = os.getenv("PDS_ENV", "int")
+    base_url = os.getenv("PDS_BASE_URL", "").strip() or None
+    config = (environment, base_url)
+
+    if _pds_service is None or _pds_service_config != config:
+        authenticator = (
+            None
+            if base_url
+            else AppRestrictedAuth(secret_manager_client=get_secrets_manager_client(), environment=environment)
         )
-        _pds_service = PdsService(authenticator, PDS_ENV)
-
+        _pds_service = PdsService(authenticator, environment, base_url=base_url)
+        _pds_service_config = config
     return _pds_service
 
 
 # Get Patient details from external service PDS using NHS number from MNS notification
 def pds_get_patient_details(nhs_number: str) -> dict:
     try:
-        patient = get_pds_service().get_patient_details(nhs_number)
-        return patient
+        return get_pds_service().get_patient_details(nhs_number)
     except Exception as e:
         msg = "Error retrieving patient details from PDS"
         logger.exception(msg)

lambdas/shared/src/common/api_clients/pds_service.py

@@ -7,31 +7,28 @@
 
 
 class PdsService:
-    def __init__(self, authenticator: AppRestrictedAuth, environment):
+    def __init__(
+        self,
+        authenticator: AppRestrictedAuth | None,
+        environment: str,
+        base_url: str | None = None,
+    ):
         logger.info(f"PdsService init: {environment}")
         self.authenticator = authenticator
-
-        self.base_url = (
-            f"https://{environment}.api.service.nhs.uk/personal-demographics/FHIR/R4/Patient"
-            if environment != "prod"
-            else "https://api.service.nhs.uk/personal-demographics/FHIR/R4/Patient"
-        )
-
+        host = "api.service.nhs.uk" if environment == "prod" else f"{environment}.api.service.nhs.uk"
+        self.base_url = base_url.rstrip("/") if base_url else f"https://{host}/personal-demographics/FHIR/R4/Patient"
         logger.info(f"PDS Service URL: {self.base_url}")
 
     def get_patient_details(self, patient_id: str) -> dict | None:
-        access_token = self.authenticator.get_access_token()
-        request_headers = {
-            "Authorization": f"Bearer {access_token}",
-            "X-Request-ID": str(uuid.uuid4()),
-            "X-Correlation-ID": str(uuid.uuid4()),
-        }
-        response = request_with_retry_backoff("GET", f"{self.base_url}/{patient_id}", headers=request_headers)
+        headers = {"X-Request-ID": str(uuid.uuid4()), "X-Correlation-ID": str(uuid.uuid4())}
+        if self.authenticator is not None:
+            headers["Authorization"] = f"Bearer {self.authenticator.get_access_token()}"
+
+        response = request_with_retry_backoff("GET", f"{self.base_url}/{patient_id}", headers=headers)
 
         if response.status_code == 200:
             return response.json()
-        elif response.status_code == 404:
+        if response.status_code == 404:
             logger.info("Patient not found")
             return None
-        else:
-            raise_error_response(response)
+        raise_error_response(response)