Changes to checking state and if jobs are free in tests

changedetectionio/tests/util.py

@@ -164,11 +164,12 @@ def wait_for_all_checks(client=None):
         if q_length == 0 and not any_workers_busy:
             if empty_since is None:
                 empty_since = time.time()
-            elif time.time() - empty_since >= 0.15:  # Shorter wait
+            # Longer stabilization period to ensure async workers have finished all updates
+            elif time.time() - empty_since >= 0.5:  # Increased from 0.15 to 0.5
                 break
         else:
             empty_since = None
-        
+
         attempt += 1
         time.sleep(0.3)
 

changedetectionio/worker_handler.py

@@ -18,6 +18,9 @@ worker_threads = []  # List of WorkerThread objects
 # Track currently processing UUIDs for async workers - maps {uuid: worker_id}
 currently_processing_uuids = {}
 
+# Track worker states - maps {worker_id: state} where state is 'waiting', 'processing', etc.
+worker_states = {}
+
 # Configuration - async workers only
 USE_ASYNC_WORKERS = True
 
@@ -104,10 +107,11 @@ class WorkerThread:
 
 def start_async_workers(n_workers, update_q, notification_q, app, datastore):
     """Start async workers, each with its own thread and event loop for isolation"""
-    global worker_threads, currently_processing_uuids
+    global worker_threads, currently_processing_uuids, worker_states
 
-    # Clear any stale UUID tracking state
+    # Clear any stale state
     currently_processing_uuids.clear()
+    worker_states.clear()
 
     # Start each worker in its own thread with its own event loop
     logger.info(f"Starting {n_workers} async workers (isolated threads)")
@@ -209,6 +213,25 @@ def set_uuid_processing(uuid, worker_id=None, processing=True):
         logger.debug(f"Worker {worker_id} finished processing UUID: {uuid}")
 
 
+def set_worker_state(worker_id, state):
+    """Set the current state of a worker ('waiting', 'processing', etc.)"""
+    global worker_states
+    try:
+        worker_states[worker_id] = state
+        logger.trace(f"Worker {worker_id} state: {state}")
+    except Exception as e:
+        # Fallback in case logger.trace isn't available
+        logger.debug(f"Worker {worker_id} state: {state}")
+
+
+def are_all_workers_idle():
+    """Check if all workers are in 'waiting' state (idle and waiting for queue items)"""
+    # Must have states for all workers, otherwise they haven't started yet
+    if len(worker_states) != len(worker_threads):
+        return False
+    return all(state == 'waiting' for state in worker_states.values())
+
+
 def is_watch_running(watch_uuid):
     """Check if a specific watch is currently being processed by any worker"""
     return watch_uuid in currently_processing_uuids