Merge pull request #194 from NHSDigital/release/2024-05-15

Release/2024-05-15

Author: Rohoolio

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 2024-05-15
+- [PI-336] Changelog deletes
+- Dependabot (pydantic)
+
 ## 2024-05-02
 - [PI-341] Prod permissions
 - [PI-268] Search for a device

VERSION

@@ -1 +1 @@
-2024.05.02
+2024.05.15

changelog/2024-05-15.md

@@ -0,0 +1,2 @@
+- [PI-336] Changelog deletes
+- Dependabot (pydantic)

infrastructure/terraform/per_workspace/modules/etl/sds/main.tf

@@ -116,7 +116,7 @@ module "worker_transform" {
                     "dynamodb:Query"
                 ],
                 "Effect": "Allow",
-                "Resource": ["${var.table_arn}"]
+                "Resource": ["${var.table_arn}", "${var.table_arn}/*"]
             },
             {
                 "Action": [

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "connecting-party-manager"
-version = "2024.05.02"
+version = "2024.05.15"
 description = "Repository for the Connecting Party Manager API and related services"
 authors = ["NHS England"]
 license = "LICENSE.md"

src/etl/sds/tests/test_sds_etl_components.py

@@ -4,7 +4,7 @@
 
 import boto3
 import pytest
-from domain.core.device import DeviceType
+from domain.core.device import DeviceStatus, DeviceType
 from domain.core.device_key import DeviceKeyType
 from etl.clear_state_inputs import EMPTY_JSON_DATA, EMPTY_LDIF_DATA
 from etl_utils.constants import CHANGELOG_NUMBER, WorkerKey
@@ -24,6 +24,21 @@
 from test_helpers.pytest_skips import long_running
 from test_helpers.terraform import read_terraform_output
 
+# Note that unique identifier "000428682512" is the identifier of 'GOOD_SDS_RECORD'
+DELETION_REQUEST_000428682512 = """
+dn: o=nhs,ou=Services,uniqueIdentifier=000428682512
+changetype: delete
+objectclass: delete
+uniqueidentifier: 000428682512
+"""
+
+DELETION_REQUEST_000842065542 = """
+dn: o=nhs,ou=Services,uniqueIdentifier=000842065542
+changetype: delete
+objectclass: delete
+uniqueidentifier: 000842065542
+"""
+
 
 @pytest.fixture
 def state_machine_input(request: pytest.FixtureRequest):
@@ -74,7 +89,7 @@ def execute_state_machine(
             error_message = cause["errorMessage"]
             stack_trace = cause["stackTrace"]
         except Exception:
-            error_message = response["cause"]
+            error_message = response.get("cause", "no error message")
             stack_trace = []
 
         print(  # noqa: T201
@@ -83,7 +98,7 @@ def execute_state_machine(
             "\n",
             *stack_trace,
         )
-        raise RuntimeError(response["error"])
+        raise RuntimeError(response.get("error", "no error message"))
     return response
 
 
@@ -102,6 +117,12 @@ def get_object(key: WorkerKey) -> str:
     return response["Body"].read()
 
 
+def put_object(key: WorkerKey, body: bytes) -> str:
+    client = boto3.client("s3")
+    etl_bucket = read_terraform_output("sds_etl.value.bucket")
+    return client.put_object(Bucket=etl_bucket, Key=key, Body=body)
+
+
 @pytest.mark.integration
 @pytest.mark.parametrize(
     "worker_data",
@@ -206,3 +227,88 @@ def test_end_to_end_bulk_trigger(repository: MockDeviceRepository):
 
     assert product_count == accredited_system_count == 5670
     assert endpoint_count == message_handling_system_count == 154506
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    "worker_data",
+    [
+        {
+            WorkerKey.EXTRACT: "\n".join([GOOD_SDS_RECORD, ANOTHER_GOOD_SDS_RECORD]),
+            WorkerKey.TRANSFORM: pkl_dumps_lz4(deque()),
+            WorkerKey.LOAD: pkl_dumps_lz4(deque()),
+        }
+    ],
+    indirect=True,
+)
+@pytest.mark.parametrize(
+    "state_machine_input",
+    [
+        StateMachineInput.bulk(changelog_number=123),
+    ],
+    indirect=True,
+)
+def test_end_to_end_changelog_delete(
+    repository: MockDeviceRepository, worker_data, state_machine_input
+):
+    """Note that the start of this test is the same as test_end_to_end, and then makes changes"""
+    extract_data = get_object(key=WorkerKey.EXTRACT)
+    transform_data = pkl_loads_lz4(get_object(key=WorkerKey.TRANSFORM))
+    load_data = pkl_loads_lz4(get_object(key=WorkerKey.LOAD))
+
+    assert len(extract_data) == 0
+    assert len(transform_data) == 0
+    assert len(load_data) == 0
+    assert len(list(repository.all_devices())) == len(
+        worker_data[WorkerKey.EXTRACT].split("\n\n")
+    )
+
+    # Now execute a changelog initial state in the ETL
+    put_object(key=WorkerKey.EXTRACT, body=DELETION_REQUEST_000428682512)
+    response = execute_state_machine(
+        state_machine_input=StateMachineInput.update(
+            changelog_number_start=124, changelog_number_end=125
+        )
+    )
+    assert response["status"] == "SUCCEEDED"
+
+    # Verify that the device with unique id 000428682512 is now "inactive"
+    (device,) = repository.read_by_index(
+        questionnaire_id="spine_device/1",
+        question_name="unique_identifier",
+        value="000428682512",
+    )
+    assert device.status == DeviceStatus.INACTIVE
+
+    # Verify that the other device is still "active"
+    (device,) = repository.read_by_index(
+        questionnaire_id="spine_device/1",
+        question_name="unique_identifier",
+        value="000842065542",
+    )
+    assert device.status == DeviceStatus.ACTIVE
+
+    # Execute another changelog initial state in the ETL
+    put_object(key=WorkerKey.EXTRACT, body=DELETION_REQUEST_000842065542)
+    response = execute_state_machine(
+        state_machine_input=StateMachineInput.update(
+            changelog_number_start=124, changelog_number_end=125
+        )
+    )
+    assert response["status"] == "SUCCEEDED"
+
+    # Verify that the device with unique id 000428682512 is still "inactive"
+    (device,) = repository.read_by_index(
+        questionnaire_id="spine_device/1",
+        question_name="unique_identifier",
+        value="000428682512",
+    )
+    assert device.status == DeviceStatus.INACTIVE
+
+    # Verify that the other device is now "inactive"
+    (device,) = repository.read_by_index(
+        questionnaire_id="spine_device/1",
+        question_name="unique_identifier",
+        value="000842065542",
+    )
+    assert device.status == DeviceStatus.INACTIVE

src/etl/sds/trigger/update/operations.py

@@ -97,11 +97,9 @@ def get_latest_changelog_number_from_ldap(
         ],
     )
 
-    _, (unpack_record) = record
-
-    return int(
-        unpack_record[ChangelogAttributes.LAST_CHANGELOG_NUMBER][0].decode("utf-8")
-    )
+    _, (_record) = record
+    (last_changelog_number_str,) = _record[ChangelogAttributes.LAST_CHANGELOG_NUMBER]
+    return int(last_changelog_number_str)
 
 
 def get_changelog_entries_from_ldap(

src/etl/sds/trigger/update/tests/test_update_trigger.py

@@ -3,6 +3,7 @@
 from unittest import mock
 
 import boto3
+import pytest
 from etl_utils.constants import CHANGELOG_NUMBER
 from moto import mock_aws
 
@@ -16,39 +17,96 @@
     "CPM_FQDN": "cpm-fqdn",
     "LDAP_HOST": "ldap-host",
     "ETL_BUCKET": "etl-bucket",
-    "ETL_EXTRACT_INPUT_KEY": "etl-input",
     "LDAP_CHANGELOG_USER": "user",
     "LDAP_CHANGELOG_PASSWORD": "eggs",  # pragma: allowlist secret
 }
 
 ALLOWED_EXCEPTIONS = (JSONDecodeError,)
-CHANGELOG_NUMBER_VALUE = "538684"
+LATEST_CHANGELOG_NUMBER = b"540382"
+CURRENT_CHANGELOG_NUMBER = str(int(LATEST_CHANGELOG_NUMBER) - 1).encode()
 
+CHANGELOG_NUMBER_RESULT = [
+    101,
+    [
+        [
+            "cn=changelog,o=nhs",
+            {
+                "firstchangenumber": [b"46425"],
+                "lastchangenumber": [LATEST_CHANGELOG_NUMBER],
+            },
+        ]
+    ],
+]
+
+CHANGE_RESULT = (
+    101,
+    [
+        [
+            "changenumber=540246,cn=changelog,o=nhs",
+            {
+                "objectClass": [
+                    b"top",
+                    b"changeLogEntry",
+                    b"nhsExternalChangelogEntry",
+                ],
+                "changeNumber": [b"540246"],
+                "changes": [
+                    b"\\nobjectClass: nhsmhsservice\\nobjectClass: top\\nnhsIDCode: F2R5Q\\nnhsMHSPartyKey: F2R5Q-823886\\nnhsMHSServiceName: urn:nhs:names:services:pdsquery\\nuniqueIdentifier: 4d554a907e83a4067695"
+                ],
+                "changeTime": [b"20240502100040Z"],
+                "changeType": [b"add"],
+                "targetDN": [
+                    b"uniqueIdentifier=4d554a907e83a4067695,ou=Services,o=nhs"
+                ],
+            },
+        ]
+    ],
+)
+
+CHANGE_RESULT_WITHOUT_UNIQUE_IDENTIFIER = (
+    101,
+    [
+        [
+            "changenumber=540246,cn=changelog,o=nhs",
+            {
+                "objectClass": [
+                    b"top",
+                    b"changeLogEntry",
+                    b"nhsExternalChangelogEntry",
+                ],
+                "changeNumber": [b"540246"],
+                "changes": [
+                    b"\\nobjectClass: nhsmhsservice\\nobjectClass: top\\nnhsIDCode: F2R5Q\\nnhsMHSPartyKey: F2R5Q-823886\\nnhsMHSServiceName: urn:nhs:names:services:pdsquery\\n"
+                ],
+                "changeTime": [b"20240502100040Z"],
+                "changeType": [b"add"],
+                "targetDN": [
+                    b"uniqueIdentifier=4d554a907e83a4067695,ou=Services,o=nhs"
+                ],
+            },
+        ]
+    ],
+)
 
-def test_update():
+
+CHANGE_AS_LDIF = """dn: o=nhs,ou=services,uniqueidentifier=4d554a907e83a4067695
+changetype: add
+objectClass: nhsmhsservice
+objectClass: top
+nhsIDCode: F2R5Q
+nhsMHSPartyKey: F2R5Q-823886
+nhsMHSServiceName: urn:nhs:names:services:pdsquery
+uniqueIdentifier: 4d554a907e83a4067695""".encode()
+
+
+@pytest.mark.parametrize(
+    "change_result", [CHANGE_RESULT, CHANGE_RESULT_WITHOUT_UNIQUE_IDENTIFIER]
+)
+def test_update(change_result):
     mocked_ldap = mock.Mock()
     mocked_ldap_client = mock.Mock()
     mocked_ldap.initialize.return_value = mocked_ldap_client
-    mocked_ldap_client.result.return_value = (
-        101,
-        [
-            (
-                "changenumber=75852519,cn=changelog,o=nhs",
-                {
-                    "objectclass": {
-                        "top",
-                        "changeLogEntry",
-                        "nhsExternalChangelogEntry",
-                    },
-                    "changenumber": "75852519",
-                    "changes": "foo",
-                    "changetime": "20240116173441Z",
-                    "changetype": "add",
-                    "targetdn": "uniqueIdentifier=200000042019,ou=Services,o=nhs",
-                },
-            ),
-        ],
-    )
+    mocked_ldap_client.result.side_effect = (CHANGELOG_NUMBER_RESULT, change_result)
 
     with mock_aws(), mock.patch.dict(
         os.environ, MOCKED_UPDATE_TRIGGER_ENVIRONMENT, clear=True
@@ -75,7 +133,7 @@ def test_update():
         s3_client.put_object(
             Bucket=MOCKED_UPDATE_TRIGGER_ENVIRONMENT["ETL_BUCKET"],
             Key=CHANGELOG_NUMBER,
-            Body="0",
+            Body=CURRENT_CHANGELOG_NUMBER,
         )
 
         from etl.sds.trigger.update import update
@@ -86,12 +144,36 @@ def test_update():
         update.CACHE["ldap_client"] = mocked_ldap_client
 
         # Remove start execution, since it's meaningless
-        idx = update.steps.index(_start_execution)
-        update.steps.pop(idx)
+        if _start_execution in update.steps:
+            idx = update.steps.index(_start_execution)
+            update.steps.pop(idx)
 
         # Don't execute the notify lambda
-        update.notify = mock.Mock(return_value="abc")
+        update.notify = (
+            lambda lambda_client, function_name, result, trigger_type: result
+        )
 
+        # Execute the test
         response = update.handler()
 
-    assert response == "abc"
+        # Verify the changelog number is NOT updated (as it should be updated in the ETL, not the trigger)
+        changelog_number_response = s3_client.get_object(
+            Bucket=MOCKED_UPDATE_TRIGGER_ENVIRONMENT["ETL_BUCKET"], Key=CHANGELOG_NUMBER
+        )
+        assert changelog_number_response["Body"].read() == CURRENT_CHANGELOG_NUMBER
+
+        # Verify the history file was created
+        etl_history_response = s3_client.get_object(
+            Bucket=MOCKED_UPDATE_TRIGGER_ENVIRONMENT["ETL_BUCKET"],
+            Key=f"history/changelog/{int(LATEST_CHANGELOG_NUMBER)}/input--extract/unprocessed",
+        )
+        assert etl_history_response["Body"].read().lower() == CHANGE_AS_LDIF.lower()
+
+        # Verify the ETL input file was created
+        etl_input_response = s3_client.get_object(
+            Bucket=MOCKED_UPDATE_TRIGGER_ENVIRONMENT["ETL_BUCKET"],
+            Key="input--extract/unprocessed",
+        )
+        assert etl_input_response["Body"].read().lower() == CHANGE_AS_LDIF.lower()
+
+    assert not isinstance(response, Exception), response