API: corriger flush() et isolation test compteur (#78)

This commit is contained in:
Colin Maudry
2026-05-13 14:45:26 +02:00
parent c683034e2b
commit 2f0cbada34
2 changed files with 4 additions and 11 deletions
+3 -11
View File
@@ -104,15 +104,7 @@ def enqueue_counter_update(token_id: int) -> None:
def flush(timeout: float = 2.0) -> None:
"""Attend que la queue soit drainée. Utile en test."""
if _queue is None:
q = _queue
if q is None:
return
# Use a sentinel approach to properly honour the timeout
done_event = threading.Event()
def _wait():
_queue.join()
done_event.set()
t = threading.Thread(target=_wait, daemon=True)
t.start()
done_event.wait(timeout=timeout)
q.join()
+1
View File
@@ -4,6 +4,7 @@ from src.api import tokens_db, tracking
def test_counter_worker_increments_count(temp_db):
_, token_id = tokens_db.create_token(temp_db, "x")
tracking.stop_worker() # reset any worker left by earlier tests
tracking.start_worker(str(temp_db))
try:
tracking.enqueue_counter_update(token_id)