NHSDigital
diff --git a/‎.github/dependabot.yml‎
Lines changed: 1 addition & 0 deletions b/‎.github/dependabot.yml‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎.gitignore‎
Lines changed: 1 addition & 0 deletions b/‎.gitignore‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎Makefile‎
Lines changed: 15 additions & 22 deletions b/‎Makefile‎
Lines changed: 15 additions & 22 deletions
diff --git a/‎application/change_event_dlq_handler/change_event_dlq_handler.py‎
Lines changed: 32 additions & 14 deletions b/‎application/change_event_dlq_handler/change_event_dlq_handler.py‎
Lines changed: 32 additions & 14 deletions
diff --git a/‎application/change_event_dlq_handler/requirements.txt‎
Lines changed: 1 addition & 1 deletion b/‎application/change_event_dlq_handler/requirements.txt‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎application/change_event_dlq_handler/tests/test_change_event_dlq_handler.py‎
Lines changed: 80 additions & 17 deletions b/‎application/change_event_dlq_handler/tests/test_change_event_dlq_handler.py‎
Lines changed: 80 additions & 17 deletions
diff --git a/‎application/common/dos_db_connection.py‎
Lines changed: 8 additions & 2 deletions b/‎application/common/dos_db_connection.py‎
Lines changed: 8 additions & 2 deletions
@@ -13,6 +13,7 @@ updates:
     schedule:
       interval: "daily"
     target-branch: "develop"
+    versioning-strategy: increase-if-necessary
 
   # Docker Dependencies
   - package-ecosystem: "docker"
 
@@ -12,3 +12,4 @@
 *.xml
 jenkins.cookies
 html
+__pycache__/
@@ -79,7 +79,7 @@ unit-test:
 	FOLDER_PATH=$$(make -s get-unit-test-path)
 	make -s docker-run-tools \
 	IMAGE=$$(make _docker-get-reg)/tester:latest \
-	CMD="python -m pytest $$FOLDER_PATH --junitxml=./testresults.xml --cov-report term-missing  --cov-report xml:coverage.xml --cov=. -vv" \
+	CMD="python -m pytest $$FOLDER_PATH --junitxml=./testresults.xml --cov-report term-missing  --cov-report xml:coverage.xml --cov=application -vv" \
 	ARGS=$(UNIT_TEST_ARGS)
 
 coverage-report: # Runs whole project coverage unit tests
@@ -103,9 +103,9 @@ UNIT_TEST_ARGS=" \
 		-e POWERTOOLS_LOG_DEDUPLICATION_DISABLED="1" \
 		--volume $(APPLICATION_DIR)/common:/tmp/.packages/common \
 		--volume $(APPLICATION_DIR)/change_event_dlq_handler:/tmp/.packages/change_event_dlq_handler \
-		--volume $(APPLICATION_DIR)/comparison_reporting:/tmp/.packages/comparison_reporting \
 		--volume $(APPLICATION_DIR)/dos_db_update_dlq_handler:/tmp/.packages/dos_db_update_dlq_handler \
 		--volume $(APPLICATION_DIR)/event_replay:/tmp/.packages/event_replay \
+		--volume $(APPLICATION_DIR)/ingest_change_event:/tmp/.packages/ingest_change_event \
 		--volume $(APPLICATION_DIR)/orchestrator:/tmp/.packages/orchestrator \
 		--volume $(APPLICATION_DIR)/send_email:/tmp/.packages/send_email \
 		--volume $(APPLICATION_DIR)/service_matcher:/tmp/.packages/service_matcher \
@@ -115,7 +115,7 @@ UNIT_TEST_ARGS=" \
 
 integration-test-autoflags-no-logs: #End to end test DI project - mandatory: PROFILE; optional: ENVIRONMENT, PARALLEL_TEST_COUNT
 	aws appconfig get-configuration --application uec-dos-int-$(ENVIRONMENT)-lambda-app-config --environment $(ENVIRONMENT) \
-	--configuration service-matcher --client-id test-id test_tmp.txt
+	--configuration ingest-change-event --client-id test-id test_tmp.txt
 	VALUE=$$(jq ".accepted_org_types.rules.org_type_in_list.conditions[0].value" test_tmp.txt)
 	if [[ $$VALUE =~ .*"PHA".* ]]; then
 		echo "PHA"
@@ -129,7 +129,7 @@ integration-test-autoflags-no-logs: #End to end test DI project - mandatory: PRO
 
 integration-test-autoflags-cloudwatch-logs: #End to end test DI project - mandatory: PROFILE; optional: ENVIRONMENT, PARALLEL_TEST_COUNT
 	aws appconfig get-configuration --application uec-dos-int-$(ENVIRONMENT)-lambda-app-config --environment $(ENVIRONMENT) \
-	--configuration service-matcher --client-id test-id test_tmp.txt
+	--configuration ingest-change-event --client-id test-id test_tmp.txt
 	VALUE=$$(jq ".accepted_org_types.rules.org_type_in_list.conditions[0].value" test_tmp.txt)
 	if [[ $$VALUE =~ .*"PHA".* ]]; then
 		echo "PHA"
@@ -146,7 +146,7 @@ integration-test: #End to end test DI project - mandatory: PROFILE, TAGS=[comple
 	echo RUN_ID=$$RUN_ID
 	make -s docker-run-tools \
 	IMAGE=$$(make _docker-get-reg)/tester:latest \
-	CMD="pytest steps -k $(TAGS) -vvvv --gherkin-terminal-reporter -p no:sugar -n $(PARALLEL_TEST_COUNT) --cucumberjson=./testresults.json --reruns 2 --reruns-delay 60" \
+	CMD="pytest steps -k $(TAGS) -vvvv --gherkin-terminal-reporter -p no:sugar -n $(PARALLEL_TEST_COUNT) --cucumberjson=./testresults.json --reruns 2 --reruns-delay 10" \
 	DIR=./test/integration \
 	ARGS=" \
 		-e API_KEY_SECRET=$(TF_VAR_api_gateway_api_key_name) \
@@ -166,22 +166,6 @@ integration-test: #End to end test DI project - mandatory: PROFILE, TAGS=[comple
 		-e CR_FIFO_DLQ=$(TF_VAR_dos_db_update_dlq_handler_lambda_name) \
 		"
 
-create-dentist-reports: # Must use a PROFILE argument with appropriate DB details, or manually pass in details as arguments themselves
-	make -s docker-run-tools \
-	IMAGE=$$(make _docker-get-reg)/tester:latest \
-	CMD="python application/comparison_reporting/run_dentist_reports.py" \
-	ARGS=" \
-		-e DB_SERVER=$$(make -s aws-rds-describe-instance-value DB_INSTANCE=$(DB_SERVER_NAME) KEY_DOT_PATH=Endpoint.Address) \
-		-e DB_PORT=$(DB_PORT) \
-		-e DB_NAME=$(DB_NAME) \
-		-e DB_USER_NAME=$$(make -s secret-get-existing-value NAME=$(DB_USER_NAME_SECRET_NAME) KEY=$(DB_USER_NAME_SECRET_KEY)) \
-		-e DB_SECRET_NAME=$(DB_SECRET_NAME) \
-		-e DB_SECRET_KEY=$(DB_SECRET_KEY) \
-		-e DB_SCHEMA=$(DB_SCHEMA) \
-		--volume $(APPLICATION_DIR)/common:/tmp/.packages/common \
-		--volume $(APPLICATION_DIR)/comparison_reporting:/tmp/.packages/comparison_reporting \
-	"
-
 clean: # Runs whole project clean
 	make \
 		docker-clean \
@@ -246,6 +230,12 @@ orchestrator-build-and-deploy: ### Build and deploy orchestrator lambda docker i
 send-email-build-and-deploy: ### Build and deploy send email lambda docker image - mandatory: PROFILE, ENVIRONMENT, FUNCTION_NAME
 	make build-and-deploy-single-function FUNCTION_NAME=send-email
 
+# ==============================================================================
+# Ingest Change Event
+
+ingest-change-event-build-and-deploy: ### Build and deploy ingest change event lambda docker image - mandatory: PROFILE, ENVIRONMENT, FUNCTION_NAME
+	make build-and-deploy-single-function FUNCTION_NAME=ingest-change-event
+
 # ==============================================================================
 # Deployments
 
@@ -263,6 +253,7 @@ build-and-deploy-single-function: # Build and deploy single lambda only (meant t
 	make build-lambda GENERIC_IMAGE_NAME=lambda VERSION=$(BUILD_TAG) NAME=$(FUNCTION_NAME)
 	make docker-push NAME=$(FUNCTION_NAME) VERSION=$(BUILD_TAG)
 	eval "$$(make -s populate-deployment-variables)"
+	eval "$$(make -s populate-serverless-variables)"
 	make serverless-deploy-single-function FUNCTION_NAME=$(FUNCTION_NAME) VERSION=$(BUILD_TAG)
 
 push-images: # Use VERSION=[] to push a perticular version otherwise with default to latest
@@ -536,7 +527,7 @@ trigger-dos-deployment-pipeline:
 	--user $$JENKINS_USERNAME:$$JENKINS_PASSWORD \
 	-H "Jenkins-Crumb: $$JENKINS_CRUMB" \
 	-F "TARGET=\"regressiondi\"" \
-	-F "IMAGE_TAG=\"7.12.1_d393439\"" \
+	-F "IMAGE_TAG=\"7.12.1_8659599\"" \
 	-F "REFRESH=\"true\""
 	echo Jenkins Job has started
 	echo Sleeping for 3 minutes
@@ -569,13 +560,15 @@ python-format:
 python-check-imports:
 	make -s docker-run-python \
 		IMAGE=$$(make _docker-get-reg)/tester:latest \
+		DIR=$(APPLICATION_DIR) \
 		CMD="python -m isort . -l=120 --check-only --profile=black \
 			--force-alphabetical-sort-within-sections --known-local-folder=common \
 			"
 
 python-fix-imports:
 	make -s docker-run-python \
 		IMAGE=$$(make _docker-get-reg)/tester:latest \
+		DIR=$(APPLICATION_DIR) \
 		CMD="python -m isort . -l=120 --profile=black --force-alphabetical-sort-within-sections \
 			--known-local-folder=common \
 			"
 
@@ -17,32 +17,51 @@
 @unhandled_exception_logging()
 @tracer.capture_lambda_handler()
 @event_source(data_class=SQSEvent)
-@logger.inject_lambda_context()
+@logger.inject_lambda_context(clear_state=True)
 @metric_scope
 def lambda_handler(event: SQSEvent, context: LambdaContext, metrics) -> None:
-    """Entrypoint handler for the lambda
+    """Entrypoint handler for the change event dlq handler lambda
+
+    Messages are sent to the change event dlq handler lambda when a message
+    fails in either the change event queue or holding queue
 
     Args:
         event (SQSEvent): Lambda function invocation event (list of 1 SQS Message)
         context (LambdaContext): Lambda function context object
     """
     record = next(event.records)
-    attributes = handle_sqs_msg_attributes(record.message_attributes)
-    correlation_id = get_sqs_msg_attribute(record.message_attributes, "correlation-id")
-    logger.set_correlation_id(correlation_id)
-    logger.append_keys(dynamo_record_id=get_sqs_msg_attribute(record.message_attributes, "dynamo_record_id"))
-    logger.append_keys(message_received=get_sqs_msg_attribute(record.message_attributes, "message_received"))
-    logger.append_keys(ods_code=get_sqs_msg_attribute(record.message_attributes, "ods_code"))
-    message = record.body
-    body = extract_body(message)
+    handle_sqs_msg_attributes(record.message_attributes)
+    body = extract_body(record.body)
+    if "dynamo_record_id" not in record.body:
+        # This is when a message comes from the change event queue
+        attributes = handle_sqs_msg_attributes(record.message_attributes)
+        correlation_id = get_sqs_msg_attribute(record.message_attributes, "correlation-id")
+        logger.set_correlation_id(correlation_id)
+        logger.append_keys(dynamo_record_id=get_sqs_msg_attribute(record.message_attributes, "dynamo_record_id"))
+        logger.append_keys(message_received=get_sqs_msg_attribute(record.message_attributes, "message_received"))
+        logger.append_keys(ods_code=get_sqs_msg_attribute(record.message_attributes, "ods_code"))
+        change_event = body
+        sequence_number = get_sequence_number(record)
+    else:
+        # This is when a message comes from the holding queue
+        attributes = handle_sqs_msg_attributes(record.message_attributes)
+        logger.info("Message received from holding queue", extra={"body": record.body})
+        change_event = body["change_event"]
+        correlation_id = body.get("correlation_id")
+        logger.set_correlation_id(correlation_id)
+        logger.append_keys(dynamo_record_id=body.get("dynamo_record_id"))
+        logger.append_keys(message_received=body.get("message_received"))
+        logger.append_keys(ods_code=change_event.get("ODSCode"))
+        sequence_number = body.get("sequence_number")
+
     error_msg = attributes["error_msg"]
     logger.warning(
-        "FIFO Dead Letter Queue Handler received event",
+        "Change Event Dead Letter Queue Handler received event",
         extra={
             "report_key": FIFO_DLQ_HANDLER_REPORT_ID,
             "error_msg": f"Message Abandoned: {error_msg}",
             "error_msg_http_code": attributes["error_msg_http_code"],
-            "payload": body,
+            "payload": change_event,
         },
     )
     metrics.set_namespace("AWS/SQS")
@@ -52,5 +71,4 @@ def lambda_handler(event: SQSEvent, context: LambdaContext, metrics) -> None:
     metrics.put_metric("NumberOfMessagesReceived", 1, "Count")
 
     sqs_timestamp = int(record.attributes["SentTimestamp"])
-    sequence_number = get_sequence_number(record)
-    add_change_event_to_dynamodb(body, sequence_number, sqs_timestamp)
+    add_change_event_to_dynamodb(change_event, sequence_number, sqs_timestamp)
@@ -1,2 +1,2 @@
 aws-embedded-metrics
-aws-lambda-powertools == 1.31.1
+aws-lambda-powertools[tracer] ~= 2.0.0
@@ -1,22 +1,35 @@
 from dataclasses import dataclass
-from unittest.mock import patch
+from json import dumps
+from typing import Any, Dict
+from unittest.mock import MagicMock, patch
 
 from aws_embedded_metrics.logger.metrics_logger import MetricsLogger
 from pytest import fixture
 
 from ..change_event_dlq_handler import lambda_handler
+from common.tests.conftest import PHARMACY_STANDARD_EVENT
 
 FILE_PATH = "application.change_event_dlq_handler.change_event_dlq_handler"
 
 
+CHANGE_EVENT_FROM_CHANGE_EVENT_QUEUE = PHARMACY_STANDARD_EVENT.copy()
+CHANGE_EVENT_FROM_HOLDING_QUEUE = {
+    "change_event": CHANGE_EVENT_FROM_CHANGE_EVENT_QUEUE,
+    "dynamo_record_id": "123456789",
+    "message_received": "2021-01-01T00:00:00.000000Z",
+    "sequence_number": "123456789",
+    "correlation_id": "123456789",
+}
+
+
 @fixture
-def dead_letter_change_event():
+def dead_letter_change_event_from_change_event_queue():
     yield {
         "Records": [
             {
                 "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
                 "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
-                "body": "Test message.",
+                "body": dumps(CHANGE_EVENT_FROM_CHANGE_EVENT_QUEUE),
                 "attributes": {
                     "ApproximateReceiveCount": "1",
                     "SentTimestamp": "1545082649183",
@@ -29,7 +42,13 @@ def dead_letter_change_event():
                         "stringListValues": [],
                         "binaryListValues": [],
                         "dataType": "String",
-                    }
+                    },
+                    "sequence-number": {
+                        "stringValue": "123456789",
+                        "stringListValues": [],
+                        "binaryListValues": [],
+                        "dataType": "String",
+                    },
                 },
                 "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
                 "eventSource": "aws:sqs",
@@ -40,6 +59,30 @@ def dead_letter_change_event():
     }
 
 
+@fixture
+def dead_letter_change_event_from_holding_queue():
+    yield {
+        "Records": [
+            {
+                "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
+                "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
+                "body": dumps(CHANGE_EVENT_FROM_HOLDING_QUEUE),
+                "attributes": {
+                    "ApproximateReceiveCount": "1",
+                    "SentTimestamp": "1545082649183",
+                    "SenderId": "AIDAIENQZJOLO23YVJ4VO",
+                    "ApproximateFirstReceiveTimestamp": "1545082649185",
+                },
+                "messageAttributes": {},
+                "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
+                "eventSource": "aws:sqs",
+                "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
+                "awsRegion": "us-east-2",
+            }
+        ]
+    }
+
+
 @fixture
 def lambda_context():
     @dataclass
@@ -58,20 +101,40 @@ class LambdaContext:
 @patch(f"{FILE_PATH}.add_change_event_to_dynamodb")
 @patch.object(MetricsLogger, "put_metric")
 @patch.object(MetricsLogger, "set_dimensions")
-def test_lambda_handler(
-    mock_put_metric,
-    mock_set_dimensions,
-    mock_add_change_event_to_dynamodb,
-    mock_extract_body,
-    dead_letter_change_event,
+def test_lambda_handler_event_from_change_event_queue(
+    mock_put_metric: MagicMock,
+    mock_set_dimensions: MagicMock,
+    mock_add_change_event_to_dynamodb: MagicMock,
+    mock_extract_body: MagicMock,
+    dead_letter_change_event_from_change_event_queue: Dict[str, Any],
+    lambda_context,
+):
+    # Arrang
+    mock_extract_body.return_value = extracted_body = "Test message1."
+    # Act
+    lambda_handler(dead_letter_change_event_from_change_event_queue, lambda_context)
+    # Assert
+    mock_extract_body.assert_called_once_with(dead_letter_change_event_from_change_event_queue["Records"][0]["body"])
+    expected_timestamp = int(
+        dead_letter_change_event_from_change_event_queue["Records"][0]["attributes"]["SentTimestamp"]
+    )
+    mock_add_change_event_to_dynamodb.assert_called_once_with(extracted_body, 123456789, expected_timestamp)
+
+
+@patch(f"{FILE_PATH}.add_change_event_to_dynamodb")
+@patch.object(MetricsLogger, "put_metric")
+@patch.object(MetricsLogger, "set_dimensions")
+def test_lambda_handler_event_from_holding_queue(
+    mock_put_metric: MagicMock,
+    mock_set_dimensions: MagicMock,
+    mock_add_change_event_to_dynamodb: MagicMock,
+    dead_letter_change_event_from_holding_queue: Dict[str, Any],
     lambda_context,
 ):
-    # Arrange
-    extracted_body = "Test message1."
-    mock_extract_body.return_value = extracted_body
     # Act
-    lambda_handler(dead_letter_change_event, lambda_context)
+    lambda_handler(dead_letter_change_event_from_holding_queue, lambda_context)
     # Assert
-    mock_extract_body.assert_called_once_with(dead_letter_change_event["Records"][0]["body"])
-    expected_timestamp = int(dead_letter_change_event["Records"][0]["attributes"]["SentTimestamp"])
-    mock_add_change_event_to_dynamodb.assert_called_once_with(extracted_body, None, expected_timestamp)
+    expected_timestamp = int(dead_letter_change_event_from_holding_queue["Records"][0]["attributes"]["SentTimestamp"])
+    mock_add_change_event_to_dynamodb.assert_called_once_with(
+        CHANGE_EVENT_FROM_CHANGE_EVENT_QUEUE, CHANGE_EVENT_FROM_HOLDING_QUEUE["sequence_number"], expected_timestamp
+    )
@@ -21,15 +21,21 @@ def connect_to_dos_db_replica() -> Generator[connection, None, None]:
     Yields:
         Generator[connection, None, None]: Connection to the database
     """
+    # Use AWS secret values, or failing that check env for DB password
+    if "DB_REPLICA_SECRET_NAME" in environ and "DB_REPLICA_SECRET_KEY" in environ:
+        db_secret = get_secret(environ["DB_REPLICA_SECRET_NAME"])
+        db_password = db_secret[environ["DB_REPLICA_SECRET_KEY"]]
+    else:
+        db_password = environ["DB_SECRET"]
+
     # Before the context manager is entered, the connection is created
-    db_secret = get_secret(environ["DB_REPLICA_SECRET_NAME"])
     db_connection = connection_to_db(
         server=environ["DB_REPLICA_SERVER"],
         port=environ["DB_PORT"],
         db_name=environ["DB_NAME"],
         db_schema=environ["DB_SCHEMA"],
         db_user=environ["DB_READ_ONLY_USER_NAME"],
-        db_password=db_secret[environ["DB_REPLICA_SECRET_KEY"]],
+        db_password=db_password
     )
     # Yield the connection object to the context manager
     yield db_connection
Original file line number	Diff line number	Diff line change
`@@ -1,2 +1,2 @@`
`1`	`1`	`aws-embedded-metrics`
`2`		`-aws-lambda-powertools == 1.31.1`
	`2`	`+aws-lambda-powertools[tracer] ~= 2.0.0`