added test and rebased on master changes

src/blueapi/worker/reworker.py

@@ -2,7 +2,7 @@
 import uuid
 from dataclasses import dataclass
 from functools import partial
-from queue import Queue
+from queue import Full, Queue
 from threading import Event, RLock
 from typing import Any, Dict, Iterable, List, Mapping, Optional, Union
 
@@ -15,6 +15,7 @@
     EventStream,
     WatchableStatus,
 )
+from blueapi.worker.worker_busy_error import WorkerBusyError
 
 from .event import (
     ProgressEvent,
@@ -85,11 +86,11 @@ def __init__(
     def submit_task(self, name: str, task: Task) -> None:
         active_task = ActiveTask(name, task)
         LOGGER.info(f"Submitting: {active_task}")
-        if self._task_queue.qsize() != 0:
+        try:
+            self._task_queue.put_nowait(active_task)
+        except Full:
             LOGGER.error("Cannot submit task while another is running")
-            raise Exception("Cannot submit task while another is running")
-
-        self._task_queue.put(active_task)
+            raise WorkerBusyError("Cannot submit task while another is running")
 
     def start(self) -> None:
         if self._started.is_set():

src/blueapi/worker/worker_busy_error.py

@@ -0,0 +1,3 @@
+class WorkerBusyError(Exception):
+    def __init__(self, message):
+        super().__init__(message)

tests/worker/test_reworker.py

@@ -14,6 +14,7 @@
     WorkerEvent,
     WorkerState,
 )
+from blueapi.worker.worker_busy_error import WorkerBusyError
 
 
 @pytest.fixture
@@ -125,3 +126,12 @@ def on_event(event: E, event_id: Optional[str]) -> None:
     sub = stream.subscribe(on_event)
     future.add_done_callback(lambda _: stream.unsubscribe(sub))
     return future
+
+
+def test_worker_only_accepts_one_task_on_queue(worker: Worker, timeout: float = 5.0):
+    worker.start()
+    task: Task = RunPlan(name="sleep", params={"time": 1.0})
+
+    worker.submit_task("first_task", task)
+    with pytest.raises(WorkerBusyError):
+        worker.submit_task("second_task", task)