Use MWLStatus enum for valid status updates with N-SET

Author: steventux

src/services/mwl/__init__.py

@@ -2,10 +2,6 @@
 
 from enum import Enum
 
-from services.mwl.c_find import CFindHandler
-
-__all__ = ["CFindHandler"]
-
 
 class MWLStatus(Enum):
     SCHEDULED = "SCHEDULED"

src/services/mwl/n_set.py

@@ -4,7 +4,8 @@
 from pynetdicom.events import Event
 from pynetdicom.sop_class import ModalityPerformedProcedureStep  # pyright: ignore[reportAttributeAccessIssue]
 
-from services.dicom import PROCESSING_FAILURE, SUCCESS, UNKNOWN_SOP_INSTANCE
+from services.dicom import INVALID_ATTRIBUTE, MISSING_ATTRIBUTE, PROCESSING_FAILURE, SUCCESS, UNKNOWN_SOP_INSTANCE
+from services.mwl import MWLStatus
 from services.storage import MWLStorage
 
 logger = logging.getLogger(__name__)
@@ -18,24 +19,30 @@ def call(self, event: Event) -> tuple[int, Dataset | None]:
         try:
             req = event.request
             requested_sop_instance_uid = getattr(req, "RequestedSOPInstanceUID", None)
-            logger.info(f"MPPS N-SET: Received request for SOP Instance UID: {requested_sop_instance_uid}")
+            logger.info("MPPS N-SET: Received request for SOP Instance UID: %s", requested_sop_instance_uid)
 
             mod_list = event.attribute_list
             status = mod_list.get("PerformedProcedureStepStatus")
             if not status:
                 logger.warning("MPPS N-SET: Missing PerformedProcedureStepStatus in request")
-                return PROCESSING_FAILURE, None
+                return MISSING_ATTRIBUTE, None
+
+            if status not in [MWLStatus.COMPLETED.value, MWLStatus.DISCONTINUED.value]:
+                logger.warning("MPPS N-SET: Invalid PerformedProcedureStepStatus: %s", status)
+                return INVALID_ATTRIBUTE, None
 
             worklist_item = self.storage.get_worklist_item_by_mpps_instance_uid(requested_sop_instance_uid)
             if not worklist_item:
-                logger.warning(f"MPPS N-SET: No worklist item found for SOP Instance UID: {requested_sop_instance_uid}")
+                logger.warning(
+                    "MPPS N-SET: No worklist item found for SOP Instance UID: %s", requested_sop_instance_uid
+                )
                 return UNKNOWN_SOP_INSTANCE, None
 
             accession_number = worklist_item.accession_number
 
             source_message_id = self.storage.update_status(accession_number, status)
             if source_message_id:
-                logger.info(f"Database updated: {accession_number} -> {status}")
+                logger.info("Database updated: %s -> %s", accession_number, status)
 
                 ds = Dataset()
                 ds.SOPClassUID = ModalityPerformedProcedureStep
@@ -48,5 +55,5 @@ def call(self, event: Event) -> tuple[int, Dataset | None]:
                 logger.warning("MPPS N-SET: Failed to update database with new status")
                 return PROCESSING_FAILURE, None
         except Exception as e:
-            logger.error(f"Error in handle_set: {str(e)}", exc_info=True)
+            logger.error("Error in handle_set: %s", str(e), exc_info=True)
             return PROCESSING_FAILURE, None

src/services/storage.py

@@ -7,6 +7,8 @@
 from pathlib import Path
 from typing import Dict, List, Optional
 
+from services.mwl import MWLStatus
+
 logger = logging.getLogger(__name__)
 
 
@@ -288,7 +290,7 @@ class WorklistItem:
     patient_name: str = field(doc="Name of the patient. Lastname^Firstname.")
     scheduled_date: str = field(doc="Date the procedure is scheduled for.")
     scheduled_time: str = field(doc="Time the procedure is scheduled for.")
-    status: str = field(doc="Status of the worklist item", default="SCHEDULED")
+    status: str = field(doc="Status of the worklist item", default=MWLStatus.SCHEDULED.value)
     source_message_id: Optional[str] = field(
         default=None, doc="Message ID from system which created this worklist item", hash=True
     )
@@ -429,7 +431,7 @@ def update_status(
 
         Args:
             accession_number: The accession number to update
-            status: New status (SCHEDULED, IN_PROGRESS, COMPLETED, DISCONTINUED)
+            status: New status (SCHEDULED, IN PROGRESS, COMPLETED, DISCONTINUED)
             mpps_instance_uid: Optional MPPS instance UID
 
         Returns:

tests/services/mwl/test_n_set.py

@@ -6,6 +6,8 @@
 from pynetdicom.sop_class import ModalityPerformedProcedureStep  # pyright: ignore[reportAttributeAccessIssue]
 
 from services.dicom import (
+    INVALID_ATTRIBUTE,
+    MISSING_ATTRIBUTE,
     PROCESSING_FAILURE,
     SUCCESS,
     UNKNOWN_SOP_INSTANCE,
@@ -32,12 +34,21 @@ def test_missing_status_returns_processing_failure(self, mock_storage, event):
         # No PerformedProcedureStepStatus set
         status, ds = NSet(mock_storage).call(event)
 
-        assert status == PROCESSING_FAILURE
+        assert status == MISSING_ATTRIBUTE
+        assert ds is None
+
+    def test_invalid_status_returns_invalid_attribute(self, mock_storage, event, requested_sop_instance_uid):
+        event.request.RequestedSOPInstanceUID = requested_sop_instance_uid
+        event.attribute_list.PerformedProcedureStepStatus = "INVALID_STATUS"
+
+        status, ds = NSet(mock_storage).call(event)
+
+        assert status == INVALID_ATTRIBUTE
         assert ds is None
 
     def test_unknown_sop_instance_returns_unknown(self, mock_storage, event, requested_sop_instance_uid):
         event.request.RequestedSOPInstanceUID = requested_sop_instance_uid
-        event.attribute_list.PerformedProcedureStepStatus = "IN PROGRESS"
+        event.attribute_list.PerformedProcedureStepStatus = "COMPLETED"
 
         mock_storage.get_worklist_item_by_mpps_instance_uid.return_value = None