PRMP-1122

lambdas/enums/nrl_sqs_upload.py

@@ -0,0 +1,6 @@
+from enum import StrEnum, auto
+
+
+class NrlActionTypes(StrEnum):
+    CREATE = auto()
+    DELETE = auto()

lambdas/enums/snomed_codes.py

@@ -0,0 +1,9 @@
+from enum import StrEnum
+
+
+class SnomedCodesType(StrEnum):
+    LLOYD_GEORGE = "16521000000101"
+
+
+class SnomedCodesCategory(StrEnum):
+    CARE_PLAN = "734163000"

lambdas/models/nrl_fhir_document_reference.py

@@ -13,7 +13,7 @@ class FhirDocumentReference(BaseModel):
     snomed_code_doc_type: str = "None"
     snomed_code_category: str = "None"
     snomed_code_category_display: str = "Care plan"
-    attachment: Optional[NrlAttachment] = {}
+    attachment: Optional[NrlAttachment] = NrlAttachment()
 
     def build_fhir_dict(self):
         snomed_url = "http://snomed.info/sct"

lambdas/models/nrl_sqs_message.py

@@ -1,5 +1,6 @@
 from typing import Optional
 
+from enums.snomed_codes import SnomedCodesCategory, SnomedCodesType
 from pydantic import AliasGenerator, BaseModel, ConfigDict
 from pydantic.alias_generators import to_camel
 
@@ -20,8 +21,8 @@ class NrlSqsMessage(BaseModel):
     )
 
     nhs_number: str
-    snomed_code_doc_type: str
-    snomed_code_category: str
+    snomed_code_doc_type: str = SnomedCodesType.LLOYD_GEORGE
+    snomed_code_category: str = SnomedCodesCategory.CARE_PLAN
     description: str = ""
     attachment: Optional[NrlAttachment] = None
     action: str

lambdas/repositories/bulk_upload/bulk_upload_sqs_repository.py

@@ -1,6 +1,7 @@
 import os
 import uuid
 
+from models.nrl_sqs_message import NrlSqsMessage
 from models.staging_metadata import StagingMetadata
 from services.base.sqs_service import SQSService
 from utils.audit_logging_setup import LoggingService
@@ -38,4 +39,14 @@ def put_sqs_message_back_to_queue(self, sqs_message: dict):
             queue_url=self.metadata_queue_url,
             message_body=sqs_message["body"],
             nhs_number=nhs_number,
+            group_id=f"back_to_queue_bulk_upload_{uuid.uuid4()}",
+        )
+
+    def send_message_to_nrl_fifo(
+        self, queue_url: str, message: NrlSqsMessage, group_id: str
+    ):
+        self.sqs_repository.send_message_fifo(
+            queue_url=queue_url,
+            message_body=message.model_dump_json(),
+            group_id=group_id,
         )

lambdas/services/base/sqs_service.py

@@ -8,6 +8,11 @@ def __init__(self, *args, **kwargs):
         self.client = boto3.client("sqs", config=config)
         super().__init__(*args, **kwargs)
 
+    def send_message_fifo(self, queue_url: str, message_body: str, group_id: str):
+        self.client.send_message(
+            QueueUrl=queue_url, MessageBody=message_body, MessageGroupId=group_id
+        )
+
     def send_message_standard(self, queue_url: str, message_body: str):
         self.client.send_message(QueueUrl=queue_url, MessageBody=message_body)
 

lambdas/services/bulk_upload_service.py

@@ -4,10 +4,12 @@
 
 import pydantic
 from botocore.exceptions import ClientError
+from enums.nrl_sqs_upload import NrlActionTypes
 from enums.patient_ods_inactive_status import PatientOdsInactiveStatus
 from enums.upload_status import UploadStatus
 from enums.virus_scan_result import VirusScanResult
 from models.nhs_document_reference import NHSDocumentReference
+from models.nrl_sqs_message import NrlSqsMessage
 from models.staging_metadata import MetadataFile, StagingMetadata
 from repositories.bulk_upload.bulk_upload_dynamo_repository import (
     BulkUploadDynamoRepository,
@@ -52,6 +54,7 @@ def __init__(self):
         self.pdf_content_type = "application/pdf"
         self.unhandled_messages = []
         self.file_path_cache = {}
+        self.nrl_queue_url = os.environ["NRL_SQS_URL"]
 
     def process_message_queue(self, records: list):
         for index, message in enumerate(records, start=1):
@@ -269,6 +272,15 @@ def handle_sqs_message(self, message: dict):
             patient_ods_code,
         )
 
+        nrl_sqs_message = NrlSqsMessage(
+            nhs_number=staging_metadata.nhs_number, action=NrlActionTypes.CREATE
+        )
+        self.sqs_repository.send_message_to_nrl_fifo(
+            queue_url=self.nrl_queue_url,
+            message=nrl_sqs_message,
+            group_id=f"nrl_sqs_{uuid.uuid4()}",
+        )
+
     def resolve_source_file_path(self, staging_metadata: StagingMetadata):
         sample_file_path = staging_metadata.files[0].file_path
 

lambdas/tests/unit/conftest.py

@@ -27,6 +27,7 @@
 
 MOCK_ZIP_OUTPUT_BUCKET_ENV_NAME = "ZIPPED_STORE_BUCKET_NAME"
 MOCK_ZIP_TRACE_TABLE_ENV_NAME = "ZIPPED_STORE_DYNAMODB_NAME"
+MOCK_METADATA_NRL_SQS_URL_ENV_NAME = "NRL_SQS_URL"
 
 MOCK_LG_STAGING_STORE_BUCKET_ENV_NAME = "STAGING_STORE_BUCKET_NAME"
 MOCK_LG_METADATA_SQS_QUEUE_ENV_NAME = "METADATA_SQS_QUEUE_URL"
@@ -76,6 +77,7 @@
 TEST_CURRENT_GP_ODS = "Y12345"
 
 AUTH_STATE_TABLE_NAME = "test_state_table"
+NRL_SQS_URL = "https://test-queue.com"
 AUTH_SESSION_TABLE_NAME = "test_session_table"
 FAKE_URL = "https://fake-url.com"
 OIDC_CALLBACK_URL = FAKE_URL
@@ -124,6 +126,7 @@ def set_env(monkeypatch):
     monkeypatch.setenv(MOCK_LG_METADATA_SQS_QUEUE_ENV_NAME, MOCK_LG_METADATA_SQS_QUEUE)
     monkeypatch.setenv(MOCK_LG_INVALID_SQS_QUEUE_ENV_NAME, MOCK_LG_INVALID_SQS_QUEUE)
     monkeypatch.setenv(MOCK_AUTH_STATE_TABLE_NAME_ENV_NAME, AUTH_STATE_TABLE_NAME)
+    monkeypatch.setenv(MOCK_METADATA_NRL_SQS_URL_ENV_NAME, NRL_SQS_URL)
     monkeypatch.setenv(MOCK_AUTH_SESSION_TABLE_NAME_ENV_NAME, AUTH_SESSION_TABLE_NAME)
     monkeypatch.setenv(MOCK_OIDC_CALLBACK_URL_ENV_NAME, OIDC_CALLBACK_URL)
     monkeypatch.setenv(MOCK_OIDC_CLIENT_ID_ENV_NAME, OIDC_CLIENT_ID)

lambdas/tests/unit/helpers/data/bulk_upload/test_data.py

@@ -3,9 +3,12 @@
 from enums.virus_scan_result import VirusScanResult
 from freezegun import freeze_time
 from models.nhs_document_reference import NHSDocumentReference
+from models.nrl_sqs_message import NrlSqsMessage
 from models.staging_metadata import MetadataFile, StagingMetadata
 from tests.unit.conftest import MOCK_LG_BUCKET, TEST_CURRENT_GP_ODS, TEST_UUID
 
+from lambdas.enums.nrl_sqs_upload import NrlActionTypes
+
 sample_metadata_model = MetadataFile(
     file_path="/1234567890/1of2_Lloyd_George_Record_[Joe Bloggs]_[1234567890]_[25-12-2019].pdf",
     page_count="",
@@ -143,6 +146,15 @@ def build_test_sqs_message_from_nhs_number(nhs_number: str) -> dict:
     return build_test_sqs_message(staging_metadata)
 
 
+def build_test_nrl_sqs_fifo_message(nhs_number: str, action: str) -> NrlSqsMessage:
+    message_body = {
+        "nhs_number": nhs_number,
+        "action": action,
+    }
+    nrl_sqs_message = NrlSqsMessage(**message_body)
+    return nrl_sqs_message
+
+
 @freeze_time("2024-01-01 12:00:00")
 def build_test_document_reference(file_name: str, nhs_number: str = "9000000009"):
     doc_ref = NHSDocumentReference(
@@ -161,7 +173,10 @@ def build_test_document_reference(file_name: str, nhs_number: str = "9000000009"
 TEST_STAGING_METADATA = build_test_staging_metadata(make_valid_lg_file_names(3))
 TEST_SQS_MESSAGE = build_test_sqs_message(TEST_STAGING_METADATA)
 TEST_FILE_METADATA = TEST_STAGING_METADATA.files[0]
-
+TEST_GROUP_ID = "123"
+TEST_NRL_SQS_MESSAGE = build_test_nrl_sqs_fifo_message(
+    TEST_NHS_NUMBER_FOR_BULK_UPLOAD, NrlActionTypes.CREATE
+)
 TEST_STAGING_METADATA_WITH_INVALID_FILENAME = build_test_staging_metadata(
     [*make_valid_lg_file_names(2), "invalid_file_name.txt"]
 )

lambdas/tests/unit/repositories/bulk_upload/test_bulk_upload_sqs_repository.py

@@ -2,9 +2,11 @@
 
 import pytest
 from repositories.bulk_upload.bulk_upload_sqs_repository import BulkUploadSqsRepository
-from tests.unit.conftest import MOCK_LG_METADATA_SQS_QUEUE
+from tests.unit.conftest import MOCK_LG_METADATA_SQS_QUEUE, NRL_SQS_URL
 from tests.unit.helpers.data.bulk_upload.test_data import (
+    TEST_GROUP_ID,
     TEST_NHS_NUMBER_FOR_BULK_UPLOAD,
+    TEST_NRL_SQS_MESSAGE,
     TEST_SQS_MESSAGE,
     TEST_STAGING_METADATA,
 )
@@ -34,11 +36,26 @@ def test_put_staging_metadata_back_to_queue_and_increases_retries(
     )
 
 
-def test_put_sqs_message_back_to_queue(set_env, repo_under_test):
+def test_put_sqs_message_back_to_queue(set_env, repo_under_test, mock_uuid):
     repo_under_test.put_sqs_message_back_to_queue(TEST_SQS_MESSAGE)
 
     repo_under_test.sqs_repository.send_message_with_nhs_number_attr_fifo.assert_called_with(
         queue_url=MOCK_LG_METADATA_SQS_QUEUE,
         message_body=TEST_SQS_MESSAGE["body"],
         nhs_number=TEST_NHS_NUMBER_FOR_BULK_UPLOAD,
+        group_id=f"back_to_queue_bulk_upload_{mock_uuid}",
+    )
+
+
+def test_send_message_to_nrl_sqs_fifo(set_env, repo_under_test):
+    repo_under_test.send_message_to_nrl_fifo(
+        NRL_SQS_URL,
+        TEST_NRL_SQS_MESSAGE,
+        TEST_GROUP_ID,
+    )
+    message_body = TEST_NRL_SQS_MESSAGE
+    repo_under_test.sqs_repository.send_message_fifo.assert_called_with(
+        queue_url="https://test-queue.com",
+        message_body=message_body.model_dump_json(),
+        group_id="123",
     )

lambdas/tests/unit/services/base/test_sqs_service.py

@@ -1,10 +1,12 @@
 import json
 
+import pytest
 from services.base.sqs_service import SQSService
 from tests.unit.conftest import MOCK_LG_METADATA_SQS_QUEUE, TEST_NHS_NUMBER
 
 
-def test_send_message_with_nhs_number_attr(set_env, mocker):
+@pytest.fixture()
+def mocked_sqs_client(mocker):
     mocked_sqs_client = mocker.MagicMock()
 
     def return_mock(service_name, **_kwargs):
@@ -13,7 +15,16 @@ def return_mock(service_name, **_kwargs):
 
     mocker.patch("boto3.client", side_effect=return_mock)
 
+    return mocked_sqs_client
+
+
+@pytest.fixture()
+def service(mocker, mocked_sqs_client):
     service = SQSService()
+    return service
+
+
+def test_send_message_with_nhs_number_attr(set_env, mocked_sqs_client, service):
 
     test_message_body = json.dumps(
         {"NHS-NO": "1234567890", "files": ["file1.pdf", "file2.pdf"]}
@@ -34,3 +45,21 @@ def return_mock(service_name, **_kwargs):
         MessageBody=test_message_body,
         MessageGroupId="test_group_id",
     )
+
+
+def test_send_message_fifo(set_env, mocked_sqs_client, service):
+    test_message_body = json.dumps(
+        {"NHS-NO": "1234567890", "files": ["file1.pdf", "file2.pdf"]}
+    )
+
+    service.send_message_fifo(
+        group_id="test_group_id",
+        queue_url=MOCK_LG_METADATA_SQS_QUEUE,
+        message_body=test_message_body,
+    )
+
+    mocked_sqs_client.send_message.assert_called_with(
+        QueueUrl=MOCK_LG_METADATA_SQS_QUEUE,
+        MessageBody=test_message_body,
+        MessageGroupId="test_group_id",
+    )

lambdas/tests/unit/services/test_bulk_upload_service.py

@@ -3,10 +3,12 @@
 
 import pytest
 from botocore.exceptions import ClientError
+from enums.nrl_sqs_upload import NrlActionTypes
 from enums.patient_ods_inactive_status import PatientOdsInactiveStatus
 from enums.upload_status import UploadStatus
 from enums.virus_scan_result import SCAN_RESULT_TAG_KEY, VirusScanResult
 from freezegun import freeze_time
+from models.nrl_sqs_message import NrlSqsMessage
 from models.pds_models import Patient
 from repositories.bulk_upload.bulk_upload_s3_repository import BulkUploadS3Repository
 from repositories.bulk_upload.bulk_upload_sqs_repository import BulkUploadSqsRepository
@@ -201,6 +203,9 @@ def test_handle_sqs_message_happy_path(
     mock_ods_validation,
 ):
     TEST_STAGING_METADATA.retries = 0
+    mock_nrl_message = NrlSqsMessage(
+        nhs_number=TEST_STAGING_METADATA.nhs_number, action=NrlActionTypes.CREATE
+    )
     mock_create_lg_records_and_copy_files = mocker.patch.object(
         BulkUploadService, "create_lg_records_and_copy_files"
     )
@@ -218,6 +223,11 @@ def test_handle_sqs_message_happy_path(
     )
     mock_report_upload_complete.assert_called()
     mock_remove_ingested_file_from_source_bucket.assert_called()
+    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_called_with(
+        queue_url="https://test-queue.com",
+        message=mock_nrl_message,
+        group_id=f"nrl_sqs_{mock_uuid}",
+    )
 
 
 def set_up_mocks_for_non_ascii_files(
@@ -330,6 +340,7 @@ def test_handle_sqs_message_calls_report_upload_failure_when_patient_record_alre
     mock_report_upload_failure.assert_called_with(
         TEST_STAGING_METADATA, UploadStatus.FAILED, str(mocked_error), ""
     )
+    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
 
 
 def test_handle_sqs_message_calls_report_upload_failure_when_lg_file_name_invalid(
@@ -366,6 +377,7 @@ def test_handle_sqs_message_calls_report_upload_failure_when_lg_file_name_invali
         str(mocked_error),
         "",
     )
+    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
 
 
 def test_handle_sqs_message_report_failure_when_document_is_infected(
@@ -403,6 +415,7 @@ def test_handle_sqs_message_report_failure_when_document_is_infected(
     )
     mock_create_lg_records_and_copy_files.assert_not_called()
     mock_remove_ingested_file_from_source_bucket.assert_not_called()
+    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
 
 
 def test_handle_sqs_message_report_failure_when_document_not_exist(
@@ -432,6 +445,7 @@ def test_handle_sqs_message_report_failure_when_document_not_exist(
         "One or more of the files is not accessible from staging bucket",
         "Y12345",
     )
+    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
 
 
 def test_handle_sqs_message_calls_report_upload_successful_when_patient_is_formally_deceased(
@@ -469,6 +483,7 @@ def test_handle_sqs_message_calls_report_upload_successful_when_patient_is_forma
         "Patient is deceased - FORMAL",
         PatientOdsInactiveStatus.DECEASED,
     )
+    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_called()
 
 
 def test_handle_sqs_message_calls_report_upload_successful_when_patient_is_informally_deceased_and_historical(
@@ -506,6 +521,7 @@ def test_handle_sqs_message_calls_report_upload_successful_when_patient_is_infor
         "Patient matched on historical name, Patient is deceased - INFORMAL",
         "Y12345",
     )
+    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_called()
 
 
 def test_handle_sqs_message_calls_report_upload_successful_when_patient_has_historical_name_and_rest(
@@ -617,6 +633,7 @@ def test_handle_sqs_message_put_staging_metadata_back_to_queue_when_virus_scan_r
     mock_report_upload_failure.assert_not_called()
     mock_create_lg_records_and_copy_files.assert_not_called()
     mock_remove_ingested_file_from_source_bucket.assert_not_called()
+    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
 
 
 def test_handle_sqs_message_rollback_transaction_when_validation_pass_but_file_transfer_failed_halfway(