diff --git a/src/relay_listener.py b/src/relay_listener.py index bef3f9dc..316a1ae9 100644 --- a/src/relay_listener.py +++ b/src/relay_listener.py @@ -21,6 +21,7 @@ from environment import Environment from services.mwl.create_worklist_item import CreateWorklistItem +from services.mwl.update_worklist_item_status import UpdateWorklistItemStatus from services.storage import MWLStorage from telemetry import configure_telemetry @@ -93,6 +94,8 @@ 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.update_item_status": + return UpdateWorklistItemStatus(self.storage).call(payload) else: raise ValueError(f"Unsupported action: {action_name}") diff --git a/src/services/mwl/__init__.py b/src/services/mwl/__init__.py index a4fec572..8a8ae7c8 100644 --- a/src/services/mwl/__init__.py +++ b/src/services/mwl/__init__.py @@ -3,8 +3,37 @@ from enum import Enum +class InvalidStatusTransitionError(Exception): + """Raised when a requested status transition is not permitted.""" + + pass + + class MWLStatus(Enum): SCHEDULED = "SCHEDULED" IN_PROGRESS = "IN PROGRESS" COMPLETED = "COMPLETED" DISCONTINUED = "DISCONTINUED" + + +class MWLStatusManager: + _TRANSITIONS = { + MWLStatus.IN_PROGRESS: MWLStatus.SCHEDULED, + MWLStatus.COMPLETED: MWLStatus.IN_PROGRESS, + MWLStatus.DISCONTINUED: MWLStatus.IN_PROGRESS, + } + + @staticmethod + def transition_for(status: str) -> tuple[MWLStatus, MWLStatus]: + """ + Get the previous and next status for a given MWL status. + + Raises: + InvalidStatusTransitionError: If the transition is not permitted + """ + try: + current_status = MWLStatus(status) + previous_status = MWLStatusManager._TRANSITIONS[current_status] + return previous_status, current_status + except KeyError, ValueError: + raise InvalidStatusTransitionError(f"Cannot transition to '{status}'") diff --git a/src/services/mwl/create_worklist_item.py b/src/services/mwl/create_worklist_item.py index 3e718a02..d798c78c 100644 --- a/src/services/mwl/create_worklist_item.py +++ b/src/services/mwl/create_worklist_item.py @@ -12,14 +12,10 @@ def __init__(self, storage: MWLStorage): def call(self, payload: dict): try: - action_id = payload.get("action_id") - if not action_id: - raise ValueError("Missing action_id in payload") - + action_id = payload["action_id"] params = payload.get("parameters", {}) - item = params.get("worklist_item", {}) - accession_number = item.get("accession_number") + accession_number = item["accession_number"] participant = item.get("participant", {}) scheduled = item.get("scheduled", {}) procedure = item.get("procedure", {}) @@ -43,6 +39,9 @@ def call(self, payload: dict): except WorklistItemExistsError: logger.info(f"Worklist item exists: accession_number={accession_number}, action_id={action_id!r}") return {"status": "exists", "action_id": action_id} + except KeyError as e: + logger.error(f"Missing key in payload: {e}") + return {"status": "error", "message": f"Missing key: {e}"} except Exception as e: logger.error(f"Failed to create worklist item: {e}") - return {"status": "error", "action_id": action_id, "error": str(e)} + return {"status": "error", "message": str(e)} diff --git a/src/services/mwl/update_worklist_item_status.py b/src/services/mwl/update_worklist_item_status.py new file mode 100644 index 00000000..2dedbfed --- /dev/null +++ b/src/services/mwl/update_worklist_item_status.py @@ -0,0 +1,24 @@ +from services.storage import MWLStorage + + +class UpdateWorklistItemStatus: + def __init__(self, storage: MWLStorage): + self.storage = storage + + def call(self, payload: dict): + """Update an existing worklist item.""" + try: + item = payload["parameters"]["worklist_item"] + accession_number = item["accession_number"] + status = item["status"].upper() + + updated_item = self.storage.update_status(accession_number, status) + + if updated_item is None: + return {"status": "error", "message": f"Worklist item '{accession_number}' not found"} + + return {"accession_number": accession_number, "status": "updated"} + except KeyError as e: + return {"status": "error", "message": f"Missing key: {e}"} + except Exception as e: + return {"status": "error", "message": str(e)} diff --git a/src/services/storage.py b/src/services/storage.py index 388d2f14..76d8f072 100644 --- a/src/services/storage.py +++ b/src/services/storage.py @@ -7,7 +7,7 @@ from typing import Dict, List, Optional from models import WorklistItem -from services.mwl import MWLStatus +from services.mwl import MWLStatusManager logger = logging.getLogger(__name__) @@ -285,12 +285,6 @@ class WorklistItemNotFoundError(Exception): pass -class InvalidStatusTransitionError(Exception): - """Raised when a requested status transition is not permitted.""" - - pass - - class WorklistItemExistsError(Exception): """Raised when a worklist item with the same accession number already exists.""" @@ -298,12 +292,6 @@ class WorklistItemExistsError(Exception): class MWLStorage(Storage): - _STATUS_TRANSITIONS: dict[MWLStatus, MWLStatus] = { - MWLStatus.IN_PROGRESS: MWLStatus.SCHEDULED, - MWLStatus.COMPLETED: MWLStatus.IN_PROGRESS, - MWLStatus.DISCONTINUED: MWLStatus.IN_PROGRESS, - } - def __init__(self, db_path: str = "/var/lib/pacs/worklist.db"): """ Initialize Worklist storage. @@ -475,20 +463,14 @@ def update_status( Transition a worklist item to a new status, enforcing valid state transitions. Args: - accession_number: The accession number to update + accession_number: The accession number of the worklist item to update status: Target status mpps_instance_uid: Optional MPPS instance UID Returns: source_message_id if item was updated, None if not found - - Raises: - InvalidStatusTransitionError: If the transition is not permitted """ - target = MWLStatus(status) - if target not in self._STATUS_TRANSITIONS: - raise InvalidStatusTransitionError(f"Cannot transition to '{status}'") - from_status = self._STATUS_TRANSITIONS[target] + from_status, to_status = MWLStatusManager.transition_for(status) with self._get_connection() as conn: cursor = conn.execute( @@ -500,7 +482,7 @@ def update_status( WHERE accession_number = ? AND status = ? """, - (status, mpps_instance_uid, accession_number, from_status.value), + (to_status.value, mpps_instance_uid, accession_number, from_status.value), ) conn.commit() diff --git a/tests/integration/test_relay_listener_processes_actions.py b/tests/integration/test_relay_listener_processes_actions.py index c4a45c7b..857dc790 100644 --- a/tests/integration/test_relay_listener_processes_actions.py +++ b/tests/integration/test_relay_listener_processes_actions.py @@ -2,11 +2,20 @@ import pytest +from models import WorklistItem from relay_listener import RelayListener from services.storage import MWLStorage class TestRelayListenerProcessesActions: + @pytest.fixture + def update_payload(self): + return { + "action_id": "action-12345", + "action_type": "worklist.update_item_status", + "parameters": {"worklist_item": {"accession_number": "ACC999999", "status": "in progress"}}, + } + @pytest.mark.asyncio async def test_relay_listener_creates_worklist_items(self, listener_payload, tmp_dir, fake_relay): storage = MWLStorage(f"{tmp_dir}/test_worklist.db") @@ -22,3 +31,34 @@ async def test_relay_listener_creates_worklist_items(self, listener_payload, tmp item = stored_items[0] assert item.accession_number == "ACC999999" assert item.patient_id == "999123456" + + @pytest.mark.asyncio + async def test_relay_listener_updates_worklist_item_status(self, update_payload, tmp_dir, fake_relay): + storage = MWLStorage(f"{tmp_dir}/test_worklist.db") + listener = RelayListener(storage) + relay_message = json.dumps({"accept": {"address": "wss://accept-url"}}) + + storage.store_worklist_item( + WorklistItem(**{ + "accession_number": "ACC999999", + "patient_id": "999123456", + "patient_name": "Test^Patient", + "patient_birth_date": "19900101", + "patient_sex": "F", + "scheduled_date": "20240101", + "scheduled_time": "090000", + "modality": "MG", + "study_description": "Mammogram", + "source_message_id": "action-12345", + }) + ) + + with fake_relay(relay_message, json.dumps(update_payload)) as ws_client: + await listener.listen() + + ws_client.send.assert_called_once_with(json.dumps({"accession_number": "ACC999999", "status": "updated"})) + stored_items = storage.find_worklist_items() + assert len(stored_items) == 1 + item = stored_items[0] + assert item.accession_number == "ACC999999" + assert item.patient_id == "999123456" diff --git a/tests/services/mwl/test_create_worklist_item.py b/tests/services/mwl/test_create_worklist_item.py index 7ff848f7..59814542 100644 --- a/tests/services/mwl/test_create_worklist_item.py +++ b/tests/services/mwl/test_create_worklist_item.py @@ -12,7 +12,7 @@ def db_file(tmp_path): return f"{tmp_path}/test.db" @pytest.fixture - def mwl_storage(db_file): + def mwl_storage(self, db_file): return MWLStorage(str(db_file)) def test_call_success(self, mwl_storage, listener_payload): @@ -28,7 +28,16 @@ def test_call_missing_action_id(self, mwl_storage, listener_payload): response = subject.call(listener_payload) assert response["status"] == "error" - assert "Missing action_id" in response["error"] + assert response["message"] == "Missing key: 'action_id'" + + def test_call_missing_accession_number(self, mwl_storage, listener_payload): + subject = CreateWorklistItem(mwl_storage) + + del listener_payload["parameters"]["worklist_item"]["accession_number"] + + response = subject.call(listener_payload) + assert response["status"] == "error" + assert response["message"] == "Missing key: 'accession_number'" def test_call_existing_worklist_item(self, mwl_storage, listener_payload): CreateWorklistItem(mwl_storage).call(listener_payload) @@ -44,4 +53,4 @@ def test_call_storage_exception(self, _, mwl_storage, listener_payload): response = subject.call(listener_payload) assert response["status"] == "error" - assert "DB error" in response["error"] + assert "DB error" in response["message"] diff --git a/tests/services/mwl/test_update_worklist_item_status.py b/tests/services/mwl/test_update_worklist_item_status.py new file mode 100644 index 00000000..8dfca41f --- /dev/null +++ b/tests/services/mwl/test_update_worklist_item_status.py @@ -0,0 +1,71 @@ +import pytest + +from models import WorklistItem +from services.mwl.update_worklist_item_status import UpdateWorklistItemStatus +from services.storage import MWLStorage + + +class TestUpdateWorklistItemStatus: + @pytest.fixture + def worklist_item_data(self): + return { + "accession_number": "ACC123", + "modality": "MG", + "patient_birth_date": "19800101", + "patient_id": "1234567890", + "patient_name": "Jane^Doe", + "scheduled_date": "20240101", + "scheduled_time": "120000", + "source_message_id": "action-12345", + } + + @pytest.fixture + def status_update_payload(self): + return { + "action_id": "action-12345", + "action_type": "worklist.update_item_status", + "parameters": {"worklist_item": {"accession_number": "ACC123", "status": "completed"}}, + } + + @pytest.fixture + def db_file(self, tmp_path) -> str: + return f"{tmp_path}/test.db" + + @pytest.fixture + def mwl_storage(self, db_file: str): + return MWLStorage(db_file) + + def test_call_success(self, mwl_storage, worklist_item_data, status_update_payload): + mwl_storage.store_worklist_item(WorklistItem(**worklist_item_data)) + mwl_storage.update_status("ACC123", "IN PROGRESS") + + subject = UpdateWorklistItemStatus(mwl_storage) + response = subject.call(status_update_payload) + assert response == {"accession_number": "ACC123", "status": "updated"} + + def test_call_missing_keys(self, mwl_storage, status_update_payload): + subject = UpdateWorklistItemStatus(mwl_storage) + + del status_update_payload["parameters"]["worklist_item"]["status"] + + response = subject.call(status_update_payload) + assert response["status"] == "error" + assert "Missing key" in response["message"] + + def test_call_nonexistent_item(self, mwl_storage, status_update_payload): + subject = UpdateWorklistItemStatus(mwl_storage) + + response = subject.call(status_update_payload) + assert response["status"] == "error" + assert response["message"] == "Worklist item 'ACC123' not found" + + def test_call_invalid_status_transition(self, mwl_storage, worklist_item_data, status_update_payload): + subject = UpdateWorklistItemStatus(mwl_storage) + + mwl_storage.store_worklist_item(WorklistItem(**worklist_item_data)) + + status_update_payload["parameters"]["worklist_item"]["status"] = "SCHEDULED" + + response = subject.call(status_update_payload) + assert response["status"] == "error" + assert response["message"] == "Cannot transition to 'SCHEDULED'" diff --git a/tests/services/test_storage.py b/tests/services/test_storage.py index b3564a46..1f22c22b 100644 --- a/tests/services/test_storage.py +++ b/tests/services/test_storage.py @@ -6,8 +6,8 @@ from pydicom.uid import generate_uid from models import WorklistItem +from services.mwl import InvalidStatusTransitionError from services.storage import ( - InvalidStatusTransitionError, MWLStorage, PACSStorage, WorklistItemExistsError, @@ -116,7 +116,7 @@ def test_store_instance_saves_to_db(self, pacs_storage): assert row["accession_number"] == metadata["accession_number"] -class TestWorkingStorage: +class TestMWLStorage: def _insert_item(self, storage, result): item = WorklistItem(**result) storage.store_worklist_item(item) diff --git a/tests/test_relay_listener.py b/tests/test_relay_listener.py index 548bfced..943fb58d 100644 --- a/tests/test_relay_listener.py +++ b/tests/test_relay_listener.py @@ -72,7 +72,7 @@ async def test_relay_listener_listen(self, storage_instance, listener_payload, f ) ) - def test_process_action(self, storage_instance, listener_payload): + def test_process_create_item_action(self, storage_instance, listener_payload): subject = RelayListener(storage_instance) response = subject.process_action(listener_payload) @@ -93,6 +93,33 @@ def test_process_action(self, storage_instance, listener_payload): ) ) + def test_process_update_item_status_action(self, storage_instance, listener_payload): + subject = RelayListener(storage_instance) + + subject.process_action(listener_payload) + + update_payload = { + "action_id": "action-12345", + "action_type": "worklist.update_item_status", + "parameters": {"worklist_item": {"accession_number": "ACC999999", "status": "IN PROGRESS"}}, + } + + response = subject.process_action(update_payload) + assert response == {"accession_number": "ACC999999", "status": "updated"} + + storage_instance.update_status.assert_called_once_with("ACC999999", "IN PROGRESS") + + def test_process_action_missing_keys(self, storage_instance, listener_payload): + subject = RelayListener(storage_instance) + + del listener_payload["parameters"]["worklist_item"]["accession_number"] + + response = subject.process_action(listener_payload) + assert response["status"] == "error" + assert "Missing key" in response["message"] + + storage_instance.store_worklist_item.assert_not_called() + def test_process_action_invalid_type(self, storage_instance, listener_payload): subject = RelayListener(storage_instance)