[feature/PI-606-changelog_add_an_mhs] update message set updater

Author: jaklinger

src/layers/domain/core/device_reference_data/v1.py

@@ -6,6 +6,7 @@
 from domain.core.aggregate_root import AggregateRoot, event
 from domain.core.cpm_system_id import ProductId
 from domain.core.device import DuplicateQuestionnaireResponse
+from domain.core.device.v1 import QuestionnaireResponseNotFoundError
 from domain.core.enum import Status
 from domain.core.event import Event, EventDeserializer
 from domain.core.questionnaire import QuestionnaireResponse
@@ -95,11 +96,20 @@ def remove_questionnaire(
     def remove_questionnaire_response(
         self, questionnaire_id: str, questionnaire_response_id: str
     ) -> QuestionnaireResponseUpdatedEvent:
-        self.questionnaire_responses[questionnaire_id] = [
-            qr
-            for qr in self.questionnaire_responses[questionnaire_id]
-            if qr.id != questionnaire_response_id
+        qid_to_remove = str(questionnaire_response_id)
+        questionnaire_response_ids = [
+            str(qr.id) for qr in self.questionnaire_responses[questionnaire_id]
         ]
+
+        try:
+            idx_to_remove = questionnaire_response_ids.index(qid_to_remove)
+        except ValueError:
+            raise QuestionnaireResponseNotFoundError(
+                f"Could not find QuestionnaireResponse {qid_to_remove}"
+            )
+        else:
+            self.questionnaire_responses[questionnaire_id].pop(idx_to_remove)
+
         return QuestionnaireResponseUpdatedEvent(
             id=self.id,
             questionnaire_responses={

src/layers/sds/epr/tests/test_updaters.py

@@ -105,8 +105,8 @@ def test_update_message_sets(message_sets: DeviceReferenceData):
         questionnaire.validate(
             {
                 str(SdsFieldName.INTERACTION_ID): interaction_id,
-                "MHS SN": "",
-                "MHS IN": "",
+                "MHS SN": "the new value",
+                "MHS IN": "the new value",
             }
         )
         for interaction_id in ("foo", "bar")
@@ -119,17 +119,31 @@ def test_update_message_sets(message_sets: DeviceReferenceData):
     qid = f"{QuestionnaireInstance.SPINE_MHS_MESSAGE_SETS}/1"
     assert final_state.pop("updated_on") > initial_state.pop("updated_on")
     questionnaire_responses = final_state.pop("questionnaire_responses")[qid]
-    assert len(questionnaire_responses) == 2
-    assert questionnaire_responses[0]["data"] == {
-        str(SdsFieldName.INTERACTION_ID): "foo",
-        "MHS SN": "",
-        "MHS IN": "",
-    }
-    assert questionnaire_responses[1]["data"] == {
-        str(SdsFieldName.INTERACTION_ID): "bar",
-        "MHS SN": "",
-        "MHS IN": "",
-    }
+    assert len(questionnaire_responses) == 4
+
+    qr_data = [qr["data"] for qr in questionnaire_responses]
+    assert qr_data == [
+        {
+            "Interaction ID": "baz",
+            "MHS IN": "",
+            "MHS SN": "",
+        },
+        {
+            "Interaction ID": "bob",
+            "MHS IN": "",
+            "MHS SN": "",
+        },
+        {
+            "Interaction ID": "foo",  # replaced old "foo"
+            "MHS IN": "the new value",
+            "MHS SN": "the new value",
+        },
+        {
+            "Interaction ID": "bar",
+            "MHS IN": "the new value",
+            "MHS SN": "the new value",
+        },
+    ]
 
     del initial_state["questionnaire_responses"]
     assert initial_state == final_state

src/layers/sds/epr/updaters.py

@@ -21,12 +21,25 @@ def remove_erroneous_additional_interactions(
 
 
 def update_message_sets(
-    message_sets: DeviceReferenceData, message_set_data: QuestionnaireResponse
+    message_sets: DeviceReferenceData, message_set_data: list[QuestionnaireResponse]
 ) -> DeviceReferenceData:
-    if message_sets.questionnaire_responses:
-        (questionnaire_id,) = message_sets.questionnaire_responses.keys()
-        message_sets.remove_questionnaire(questionnaire_id)
+    """
+    Updates the MessageSets questionnaire responses with the provided message_set_data,
+    with any replacements based on matching by interaction id.
+    """
+    interaction_id_to_qid = {
+        msg_set.data[SdsFieldName.INTERACTION_ID]: msg_set.id
+        for msg_sets in message_sets.questionnaire_responses.values()
+        for msg_set in msg_sets
+    }
 
     for _message_set in message_set_data:
+        new_interaction_id = _message_set.data[SdsFieldName.INTERACTION_ID]
+        qid_to_remove = interaction_id_to_qid.get(new_interaction_id)
+        if qid_to_remove:
+            message_sets.remove_questionnaire_response(
+                questionnaire_id=_message_set.questionnaire_id,
+                questionnaire_response_id=qid_to_remove,
+            )
         message_sets.add_questionnaire_response(_message_set)
     return message_sets

src/layers/sds/epr/updates/change_request_routing.py

@@ -1,6 +1,9 @@
 from typing import Literal
 
-from domain.core.aggregate_root import AggregateRoot
+from domain.core.cpm_product.v1 import CpmProduct
+from domain.core.device.v1 import Device
+from domain.core.device_reference_data.v1 import DeviceReferenceData
+from domain.core.product_team.v1 import ProductTeam
 from domain.core.questionnaire import Questionnaire
 from domain.repository.cpm_product_repository import CpmProductRepository
 from domain.repository.device_reference_data_repository import (
@@ -13,6 +16,7 @@
 from sds.domain.parse import UnknownSdsModel
 from sds.domain.sds_deletion_request import SdsDeletionRequest
 from sds.domain.sds_modification_request import SdsModificationRequest
+from sds.epr.constants import BAD_UNIQUE_IDENTIFIERS
 from sds.epr.updates.change_request_processors import (
     process_request_to_add_as,
     process_request_to_add_mhs,
@@ -26,28 +30,6 @@
 )
 from sds.epr.utils import is_as_device, is_mhs_device
 
-EXCEPTIONAL_ODS_CODES = {  # TODO: use these
-    "696B001",
-    "TESTEBS1",
-    "TESTLSP0",
-    "TESTLSP1",
-    "TESTLSP3",
-    "TMSAsync1",
-    "TMSAsync2",
-    "TMSAsync3",
-    "TMSAsync4",
-    "TMSAsync5",
-    "TMSAsync6",
-    "TMSEbs2",
-}
-
-BAD_UNIQUE_IDENTIFIERS = {  # TODO: put these somewhere
-    "31af51067f47f1244d38",  # pragma: allowlist secret
-    "a83e1431f26461894465",  # pragma: allowlist secret
-    "S2202584A2577603",
-    "S100049A300185",
-}
-
 
 def process_change_request(
     record: dict[Literal["object_class", "unique_identifier"] | str, str],
@@ -63,7 +45,7 @@ def process_change_request(
     message_set_questionnaire: Questionnaire,
     message_set_field_mapping: dict,
     additional_interactions_questionnaire: Questionnaire,
-) -> list[AggregateRoot]:
+) -> list[ProductTeam | CpmProduct | Device | DeviceReferenceData]:
     unique_identifier = record["unique_identifier"]
     if unique_identifier in BAD_UNIQUE_IDENTIFIERS:
         return []