Support updating worklist item status via relay connection

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}")
 

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}'")

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)}

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)}

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,25 +285,13 @@ 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."""
 
     pass
 
 
 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()
 

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"

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"]

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'"

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)

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)