diff --git a/tests/test_admission.py b/tests/test_admission.py index 7c1f5a547..52ed68278 100644 --- a/tests/test_admission.py +++ b/tests/test_admission.py @@ -6,7 +6,7 @@ from __future__ import annotations import logging import threading import time -from collections.abc import Callable, Iterator +from collections.abc import Iterator import pytest @@ -328,6 +328,7 @@ def test_two_waiters_keep_throttle_state_until_both_end( starts: list[str] = [] ends: list[str] = [] events_lock = threading.Lock() + started = threading.Semaphore(0) monkeypatch.setattr(admission, "get_config", lambda: _memory_config(5 * 1024)) monkeypatch.setattr(admission, "_POLL_INTERVAL_SECONDS", 0.01) @@ -343,6 +344,7 @@ def test_two_waiters_keep_throttle_state_until_both_end( def on_start(**fields: object) -> None: with events_lock: starts.append(str(fields["stage"])) + started.release() def on_end(**fields: object) -> None: with events_lock: @@ -373,7 +375,8 @@ def test_two_waiters_keep_throttle_state_until_both_end( for thread in threads: thread.start() - _wait_until(lambda: admission.throttle_state().count == 2) + assert started.acquire(timeout=2.0) + assert started.acquire(timeout=2.0) state = admission.throttle_state() assert state.throttled is True @@ -382,7 +385,8 @@ def test_two_waiters_keep_throttle_state_until_both_end( assert state.available_mib == 2 * 1024 release_first.set() - _wait_until(lambda: not threads[0].is_alive()) + threads[0].join(timeout=2) + assert not threads[0].is_alive() state = admission.throttle_state() assert state.throttled is True @@ -401,12 +405,3 @@ def test_two_waiters_keep_throttle_state_until_both_end( ) assert sorted(starts) == ["describe", "transcribe"] assert sorted(ends) == ["describe", "transcribe"] - - -def _wait_until(predicate: Callable[[], bool], timeout: float = 2.0) -> None: - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - if predicate(): - return - time.sleep(0.01) - raise AssertionError("condition was not met before timeout") diff --git a/tests/test_callosum.py b/tests/test_callosum.py index eb7a9838a..8ec92cc65 100644 --- a/tests/test_callosum.py +++ b/tests/test_callosum.py @@ -231,30 +231,46 @@ def test_client_receives_during_outbound_flood(short_callosum_server): client.stop() -def test_client_disconnected_idle_blocks_and_drops_emit(tmp_path, monkeypatch, caplog): +def test_client_disconnected_idle_blocks_and_drops_emit(tmp_path, monkeypatch): client = CallosumConnection(socket_path=tmp_path / "missing.sock") original_wait = client.stop_event.wait wait_calls = 0 + block_reached = threading.Event() def counted_wait(timeout=None): nonlocal wait_calls wait_calls += 1 + block_reached.set() return original_wait(timeout) monkeypatch.setattr(client.stop_event, "wait", counted_wait) - with caplog.at_level(logging.INFO, logger="solstone.think.callosum"): + drop_logged = threading.Event() + drop_message = "Dropping message (not connected): test/event" + + class _DropWatcher(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + if drop_message in record.getMessage(): + drop_logged.set() + + logger = logging.getLogger("solstone.think.callosum") + previous_level = logger.level + logger.setLevel(logging.INFO) + handler = _DropWatcher() + logger.addHandler(handler) + try: client.start() try: - time.sleep(0.25) + # Wait until the reconnect loop has entered its disconnected idle block + # (first stop_event.wait), then emit while still disconnected. + assert block_reached.wait(timeout=2.0) assert client.emit("test", "event") - deadline = time.monotonic() + 1.0 - while "Dropping message (not connected): test/event" not in caplog.text: - if time.monotonic() > deadline: - pytest.fail("disconnected emit was not dropped in time") - time.sleep(0.01) + assert drop_logged.wait(timeout=2.0) finally: client.stop() + finally: + logger.removeHandler(handler) + logger.setLevel(previous_level) assert 1 <= wait_calls <= 20 diff --git a/tests/test_cortex.py b/tests/test_cortex.py index 3512d4d1d..20d3d5afb 100644 --- a/tests/test_cortex.py +++ b/tests/test_cortex.py @@ -67,15 +67,6 @@ def _completed_path(journal_path: Path, use_id: str, name: str = "chat") -> Path return journal_path / "talents" / name.replace(":", "--") / f"{use_id}.jsonl" -def _wait_until(predicate, timeout: float = 2.0) -> bool: - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - if predicate(): - return True - time.sleep(0.01) - return predicate() - - @pytest.fixture def mock_journal(tmp_path, monkeypatch): """Set up a temporary journal directory.""" @@ -156,12 +147,7 @@ def test_handle_request_dedups_existing_active_file( daemon=True, ) cortex_service._spawn_worker.start() - assert _wait_until( - lambda: ( - cortex_service.spawn_queue.qsize() == 0 - and cortex_service._pending_spawns == 0 - ) - ) + cortex_service.spawn_queue.join() cortex_service.stop_event.set() cortex_service._spawn_worker.join(timeout=1) @@ -586,12 +572,7 @@ def test_spawn_worker_processes_fifo(cortex_service): ) cortex_service._spawn_worker.start() - assert _wait_until( - lambda: ( - cortex_service.spawn_queue.qsize() == 0 - and cortex_service._pending_spawns == 0 - ) - ) + cortex_service.spawn_queue.join() cortex_service.stop_event.set() cortex_service._spawn_worker.join(timeout=1) @@ -616,12 +597,7 @@ def test_duplicate_claim_is_not_enqueued_twice(cortex_service, mock_journal): ) cortex_service._spawn_worker.start() - assert _wait_until( - lambda: ( - cortex_service.spawn_queue.qsize() == 0 - and cortex_service._pending_spawns == 0 - ) - ) + cortex_service.spawn_queue.join() cortex_service.stop_event.set() cortex_service._spawn_worker.join(timeout=1) @@ -647,12 +623,7 @@ def test_spawn_worker_isolates_per_item_failures(cortex_service, mock_journal): cortex_service._handle_callosum_message(_cortex_request("bad")) cortex_service._handle_callosum_message(_cortex_request("good")) - assert _wait_until( - lambda: ( - cortex_service.spawn_queue.qsize() == 0 - and cortex_service._pending_spawns == 0 - ) - ) + cortex_service.spawn_queue.join() completed = _completed_path(mock_journal, "bad") assert completed.exists() diff --git a/tests/test_voice_brain.py b/tests/test_voice_brain.py index 697d82801..daf33e008 100644 --- a/tests/test_voice_brain.py +++ b/tests/test_voice_brain.py @@ -4,7 +4,7 @@ from __future__ import annotations import asyncio -import time +import threading from flask import Flask @@ -102,21 +102,21 @@ def test_schedule_refresh_updates_instruction(monkeypatch, journal_copy): return "New voice", "session-3" monkeypatch.setattr(brain, "_run_claude", fake_run_claude) + refresh_applied = threading.Event() + original_complete = brain._complete_future + + def complete_and_signal(app_arg, attr_name, future): + original_complete(app_arg, attr_name, future) + if attr_name == "refresh_future": + refresh_applied.set() + + monkeypatch.setattr(brain, "_complete_future", complete_and_signal) start_voice_runtime(app) try: future = brain.schedule_refresh(app, force=True) assert future.result(timeout=1.0) == ("session-3", "New voice") - # The app-state update runs in the future's done-callback. A - # concurrent.futures.Future notifies result() waiters *before* it - # invokes done-callbacks, so the callback may not have applied the - # instruction yet when result() returns -- poll for the side effect - # rather than racing it (this was an xdist-only flake under load). - deadline = time.monotonic() + 1.0 - while ( - app.voice_brain_instruction != "New voice" and time.monotonic() < deadline - ): - time.sleep(0.01) + assert refresh_applied.wait(timeout=2.0) assert app.voice_brain_instruction == "New voice" finally: stop_voice_runtime(app)