From acc610944f40c7eeaa2fee4b0789945f4b6a31d7 Mon Sep 17 00:00:00 2001 From: Steve Laing Date: Wed, 8 Jul 2026 17:56:36 +0100 Subject: [PATCH 1/4] Accept worklist.create_test_item actions and process with Modality Emulator Processes test worklist items immediately sending emulated DICOM files with sample images via the uploader. --- src/modality_emulator.py | 12 ++++++++---- src/relay_listener.py | 18 ++++++++++++++++++ tests/test_modality_emulator.py | 30 ++++++++++++++++++++++++++++++ tests/test_relay_listener.py | 27 +++++++++++++++++++++++++++ 4 files changed, 83 insertions(+), 4 deletions(-) diff --git a/src/modality_emulator.py b/src/modality_emulator.py index 398bbd3d..753e40c1 100644 --- a/src/modality_emulator.py +++ b/src/modality_emulator.py @@ -128,7 +128,7 @@ def __init__(self, mwl_storage: MWLStorage): self.mwl_storage = mwl_storage self.processed_items = set() - def process_worklist_items(self, ae: AE): + def process_worklist_items(self, ae: AE, patient_name: str | None = None): """ Queries the MWL for items scheduled for today and sends generated DICOM files to the PACS server for each item. """ @@ -140,7 +140,8 @@ def process_worklist_items(self, ae: AE): logger.info(f"Connected to PACS server {PACS_HOST}:{PACS_PORT} ({PACS_AET})") logger.info("Querying MWL for scheduled items...") - responses = mwl_assoc.send_c_find(self.c_find_dataset, query_model=ModalityWorklistInformationFind) + c_find_dataset = self.c_find_dataset(patient_name=patient_name) + responses = mwl_assoc.send_c_find(c_find_dataset, query_model=ModalityWorklistInformationFind) for status, ds in responses: status_code = getattr(status, "Status", SUCCESS) @@ -189,8 +190,7 @@ def process_worklist_items(self, ae: AE): mwl_assoc.release() pacs_assoc.release() - @property - def c_find_dataset(self) -> Dataset: + def c_find_dataset(self, patient_name: str | None = None) -> Dataset: date_today = datetime.date.today() ds = Dataset() sps_dataset = Dataset() @@ -198,6 +198,10 @@ def c_find_dataset(self) -> Dataset: sps_dataset.ScheduledProcedureStepStartDate = date_today.strftime("%Y%m%d") sps_dataset.ScheduledProcedureStepStartTime = "000000-" ds.ScheduledProcedureStepSequence = [sps_dataset] + + if patient_name: + ds.PatientName = patient_name + return ds diff --git a/src/relay_listener.py b/src/relay_listener.py index d24ba635..b4acab6f 100644 --- a/src/relay_listener.py +++ b/src/relay_listener.py @@ -16,10 +16,16 @@ from azure.identity import DefaultAzureCredential, ManagedIdentityCredential from dotenv import load_dotenv +from pynetdicom import AE +from pynetdicom.sop_class import ( + DigitalMammographyXRayImageStorageForPresentation, # type: ignore + ModalityWorklistInformationFind, # type: ignore +) from websockets.asyncio.client import connect from websockets.exceptions import ConnectionClosedError from environment import Environment +from modality_emulator import ModalityEmulator from services.mwl.create_worklist_item import CreateWorklistItem from services.mwl.update_worklist_item_status import UpdateWorklistItemStatus from services.storage import MWLStorage @@ -94,6 +100,10 @@ def process_action(self, payload: dict): return {"status": "echo", "payload": payload} elif action_name == "worklist.create_item": return CreateWorklistItem(self.storage).call(payload) + elif action_name == "worklist.create_test_item": + result = CreateWorklistItem(self.storage).call(payload) + self.process_with_modality_emulator(patient_name=payload.get("patient_name")) + return result elif action_name == "worklist.update_status": return UpdateWorklistItemStatus(self.storage).call(payload) else: @@ -106,6 +116,14 @@ def _connect(self): compression=None, ) + def process_with_modality_emulator(self, patient_name: str | None = None): + """Process worklist items with ModalityEmulator.""" + ae = AE(ae_title="ModalityEmulator") + ae.add_requested_context(DigitalMammographyXRayImageStorageForPresentation) + ae.add_requested_context(ModalityWorklistInformationFind) + + ModalityEmulator(self.storage).process_worklist_items(ae, patient_name=patient_name) + class RelayURI: def __init__(self): diff --git a/tests/test_modality_emulator.py b/tests/test_modality_emulator.py index 9eebe081..bebfeaff 100644 --- a/tests/test_modality_emulator.py +++ b/tests/test_modality_emulator.py @@ -187,6 +187,36 @@ def test_process_worklist_items_returns_when_no_items( mwl_assoc.release.assert_called_once() pacs_assoc.release.assert_called_once() + @patch.object(ModalityEmulator, "c_find_dataset") + def test_process_worklist_items_passes_patient_name_to_c_find_dataset( + self, + mock_c_find_dataset, + success_status, + ): + """Process worklist items passes patient_name to c_find_dataset.""" + mwl_storage = MagicMock() + emulator = ModalityEmulator(mwl_storage) + + query_ds = Dataset() + mock_c_find_dataset.return_value = query_ds + + mwl_assoc = MagicMock() + mwl_assoc.is_established = True + mwl_assoc.send_c_find.return_value = [(success_status, None)] + + pacs_assoc = MagicMock() + pacs_assoc.is_established = True + + ae = MagicMock() + ae.associate.side_effect = [mwl_assoc, pacs_assoc] + + emulator.process_worklist_items(ae, patient_name="Jane Doe") + + mock_c_find_dataset.assert_called_once_with(patient_name="Jane Doe") + mwl_storage.update_status.assert_not_called() + mwl_assoc.release.assert_called_once() + pacs_assoc.release.assert_called_once() + @patch("modality_emulator.time.sleep") def test_process_worklist_items_handles_failed_association( self, diff --git a/tests/test_relay_listener.py b/tests/test_relay_listener.py index 3e4de26f..9ce39838 100644 --- a/tests/test_relay_listener.py +++ b/tests/test_relay_listener.py @@ -114,6 +114,33 @@ def test_process_update_item_status_action(self, storage_instance, listener_payl storage_instance.update_status.assert_called_once_with("ACC999999", "IN PROGRESS") + def test_process_create_test_item_action_triggers_modality_emulator(self, storage_instance, listener_payload): + """Process create test item action and trigger modality emulator.""" + subject = RelayListener(storage_instance) + payload = dict(listener_payload) + payload["action_type"] = "worklist.create_test_item" + + with patch.object(subject, "process_with_modality_emulator") as mock_emulator: + response = subject.process_action(payload) + + assert response == {"action_id": "action-12345", "status": "created"} + mock_emulator.assert_called_once_with(patient_name=payload.get("patient_name")) + + storage_instance.store_worklist_item.assert_called_once_with( + WorklistItem( + accession_number="ACC999999", + patient_id="999123456", + patient_name="SMITH^JANE", + patient_birth_date="19900202", + patient_sex="F", + scheduled_date="20240615", + scheduled_time="101500", + modality="MG", + study_description="MAMMOGRAPHY", + source_message_id="action-12345", + ) + ) + def test_process_action_missing_keys(self, storage_instance, listener_payload): """Process action missing keys.""" subject = RelayListener(storage_instance) From 1ad5b3e365b5ff95fb685954b20ecaf15f27f5de Mon Sep 17 00:00:00 2001 From: Steve Laing Date: Tue, 14 Jul 2026 16:41:58 +0100 Subject: [PATCH 2/4] Respond to unknown actions with error and message --- src/relay_listener.py | 3 ++- tests/test_relay_listener.py | 16 +++++++--------- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/src/relay_listener.py b/src/relay_listener.py index b4acab6f..9d38ca04 100644 --- a/src/relay_listener.py +++ b/src/relay_listener.py @@ -107,7 +107,8 @@ def process_action(self, payload: dict): elif action_name == "worklist.update_status": return UpdateWorklistItemStatus(self.storage).call(payload) else: - raise ValueError(f"Unsupported action: {action_name}") + logger.error("Unsupported action: %s", action_name) + return {"status": "error", "message": f"Unsupported action: {action_name}"} def _connect(self): """Connect to Azure Relay.""" diff --git a/tests/test_relay_listener.py b/tests/test_relay_listener.py index 9ce39838..b0900c91 100644 --- a/tests/test_relay_listener.py +++ b/tests/test_relay_listener.py @@ -159,15 +159,13 @@ def test_process_action_invalid_type(self, storage_instance, listener_payload): listener_payload["action_type"] = "worklist.unknown_action" - with pytest.raises(ValueError): - response = subject.process_action(listener_payload) - assert response == { - "status": "error", - "action_id": "action-12345", - "error": "Unknown action type: worklist.unknown_action", - } - - storage_instance.store_worklist_item.assert_not_called() + response = subject.process_action(listener_payload) + assert response == { + "status": "error", + "message": "Unsupported action: worklist.unknown_action", + } + + storage_instance.store_worklist_item.assert_not_called() class TestRelayURIWithDefaultAzureCredential: From 51480a0083c8187c12e846648b33e4ef3d294818 Mon Sep 17 00:00:00 2001 From: Steve Laing Date: Tue, 14 Jul 2026 16:51:52 +0100 Subject: [PATCH 3/4] Fix patient name lookup from payload Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/relay_listener.py | 17 ++++++++++++++++- tests/test_relay_listener.py | 4 +++- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/src/relay_listener.py b/src/relay_listener.py index 9d38ca04..1048fefa 100644 --- a/src/relay_listener.py +++ b/src/relay_listener.py @@ -102,7 +102,22 @@ def process_action(self, payload: dict): return CreateWorklistItem(self.storage).call(payload) elif action_name == "worklist.create_test_item": result = CreateWorklistItem(self.storage).call(payload) - self.process_with_modality_emulator(patient_name=payload.get("patient_name")) + patient_name = payload.get("parameters", {}).get("worklist_item", {}).get("participant", {}).get("name") + + def _run_emulator(): + try: + self.process_with_modality_emulator(patient_name=patient_name) + except Exception: + logger.exception("Modality emulator processing failed") + + try: + loop = asyncio.get_running_loop() + except RuntimeError: + # Called outside an event loop (e.g. unit tests) + _run_emulator() + else: + loop.create_task(asyncio.to_thread(_run_emulator)) + return result elif action_name == "worklist.update_status": return UpdateWorklistItemStatus(self.storage).call(payload) diff --git a/tests/test_relay_listener.py b/tests/test_relay_listener.py index b0900c91..0c72e316 100644 --- a/tests/test_relay_listener.py +++ b/tests/test_relay_listener.py @@ -124,7 +124,9 @@ def test_process_create_test_item_action_triggers_modality_emulator(self, storag response = subject.process_action(payload) assert response == {"action_id": "action-12345", "status": "created"} - mock_emulator.assert_called_once_with(patient_name=payload.get("patient_name")) + mock_emulator.assert_called_once_with( + patient_name=payload["parameters"]["worklist_item"]["participant"]["name"] + ) storage_instance.store_worklist_item.assert_called_once_with( WorklistItem( From cfb9072dabcf97ad814d7c3edd987e4ce4a5c250 Mon Sep 17 00:00:00 2001 From: Steve Laing Date: Mon, 20 Jul 2026 10:48:48 +0100 Subject: [PATCH 4/4] Ensure patient name is used to process test items Safeguard against accidentally processing items without a patient name in the C-FIND criteria. This ensures processing test items are correctly scoped. --- src/relay_listener.py | 33 ++++++++++++++++++++------------- tests/test_relay_listener.py | 17 +++++++++++++++++ 2 files changed, 37 insertions(+), 13 deletions(-) diff --git a/src/relay_listener.py b/src/relay_listener.py index 1048fefa..5415d8cd 100644 --- a/src/relay_listener.py +++ b/src/relay_listener.py @@ -104,19 +104,14 @@ def process_action(self, payload: dict): result = CreateWorklistItem(self.storage).call(payload) patient_name = payload.get("parameters", {}).get("worklist_item", {}).get("participant", {}).get("name") - def _run_emulator(): - try: - self.process_with_modality_emulator(patient_name=patient_name) - except Exception: - logger.exception("Modality emulator processing failed") + if not patient_name: + logger.warning("No patient name provided for ModalityEmulator test item processing") + return { + "status": "error", + "message": "No patient name provided for ModalityEmulator test item processing", + } - try: - loop = asyncio.get_running_loop() - except RuntimeError: - # Called outside an event loop (e.g. unit tests) - _run_emulator() - else: - loop.create_task(asyncio.to_thread(_run_emulator)) + self.process_with_modality_emulator(patient_name=patient_name) return result elif action_name == "worklist.update_status": @@ -138,7 +133,19 @@ def process_with_modality_emulator(self, patient_name: str | None = None): ae.add_requested_context(DigitalMammographyXRayImageStorageForPresentation) ae.add_requested_context(ModalityWorklistInformationFind) - ModalityEmulator(self.storage).process_worklist_items(ae, patient_name=patient_name) + def _run_emulator(): + try: + ModalityEmulator(self.storage).process_worklist_items(ae, patient_name=patient_name) + except Exception: + logger.exception("Modality emulator processing failed") + + try: + loop = asyncio.get_running_loop() + except RuntimeError: + # Called outside an event loop (e.g. unit tests) + _run_emulator() + else: + loop.create_task(asyncio.to_thread(_run_emulator)) class RelayURI: diff --git a/tests/test_relay_listener.py b/tests/test_relay_listener.py index 0c72e316..0e4ad64c 100644 --- a/tests/test_relay_listener.py +++ b/tests/test_relay_listener.py @@ -143,6 +143,23 @@ def test_process_create_test_item_action_triggers_modality_emulator(self, storag ) ) + def test_process_create_test_item_action_without_patient_name_returns_error( + self, storage_instance, listener_payload + ): + """Process create test item action without a patient name returns an error.""" + subject = RelayListener(storage_instance) + payload = dict(listener_payload) + payload["action_type"] = "worklist.create_test_item" + del payload["parameters"]["worklist_item"]["participant"]["name"] + + with patch.object(subject, "process_with_modality_emulator"): + response = subject.process_action(payload) + + assert response == { + "status": "error", + "message": "No patient name provided for ModalityEmulator test item processing", + } + def test_process_action_missing_keys(self, storage_instance, listener_payload): """Process action missing keys.""" subject = RelayListener(storage_instance)