diff --git a/tests/core/tools/adapters/vibe/test_mcp_init_timeout.py b/tests/core/tools/adapters/vibe/test_mcp_init_timeout.py index 78e1b84ad3..3898b6d324 100644 --- a/tests/core/tools/adapters/vibe/test_mcp_init_timeout.py +++ b/tests/core/tools/adapters/vibe/test_mcp_init_timeout.py @@ -2,7 +2,6 @@ agent setup (or pin resources) indefinitely.""" import asyncio -import time import pytest @@ -18,6 +17,7 @@ @pytest.mark.asyncio +@pytest.mark.timeout(30) async def test_stalled_server_times_out_and_other_servers_still_load(monkeypatch): """A server whose initialize/list-tools stalls is skipped at the timeout; the remaining servers still load.""" @@ -36,14 +36,12 @@ async def fake_load_direct(server_name, connection, **kwargs): monkeypatch.setattr(mcp_adapter_module, "_load_direct_mcp_tools", fake_load_direct) - started = time.monotonic() result = await load_mcp_tools_as_agent_tools( { "stalled": {"transport": "streamable_http", "url": "http://x"}, "healthy": {"transport": "streamable_http", "url": "http://y"}, } ) - elapsed = time.monotonic() - started assert result.tools == (healthy_tool,) assert result.loaded_servers == ("healthy",) @@ -51,12 +49,10 @@ async def fake_load_direct(server_name, connection, **kwargs): assert result.failures[0].server_name == "stalled" assert result.failures[0].phase is MCPFailurePhase.INITIALIZE assert result.failures[0].error_type == "TimeoutError" - # 1s timeout for the stalled server plus fast healthy load; the old - # behavior blocked forever. - assert elapsed < 5 @pytest.mark.asyncio +@pytest.mark.timeout(30) async def test_bounded_load_returns_even_when_cleanup_hangs(): """The bound must hold even if the load task ignores cancellation (e.g. a hung streamable-HTTP session blocking in __aexit__).""" @@ -80,12 +76,9 @@ async def uncancellable_load(): continue return [] # pragma: no cover - started = time.monotonic() with pytest.raises(TimeoutError): await _load_server_tools_bounded("hung", uncancellable_load(), 1) - elapsed = time.monotonic() - started - assert elapsed < 5 # The load was cancelled (cleanup began) but the caller did not wait on # it: the bounded call returned while cleanup was still blocked. await asyncio.wait_for(cleanup_entered.wait(), timeout=5) @@ -94,6 +87,7 @@ async def uncancellable_load(): @pytest.mark.asyncio +@pytest.mark.timeout(30) async def test_burst_larger_than_gate_does_not_fan_out(monkeypatch): """A burst of concurrent loads for the same hung server must not create more underlying load tasks (transports/sockets) than the per-server cap: @@ -120,13 +114,9 @@ async def one_caller(): with pytest.raises(TimeoutError): await _load_server_tools_bounded("burst-server", uncancellable_load(), 1) - began = time.monotonic() await asyncio.gather(*(one_caller() for _ in range(6))) - elapsed = time.monotonic() - began - # Every caller returned within its own bound... - assert elapsed < 5 - # ...but only cap-many loads (transports) ever started; the abandoned + # Only cap-many loads (transports) ever started; the abandoned # ones keep holding their slots so the other four callers failed fast # at the gate. assert started_loads == 2 diff --git a/tests/core/tools/core/test_document_search_collection_concurrency.py b/tests/core/tools/core/test_document_search_collection_concurrency.py index 50436679f2..7e891b757e 100644 --- a/tests/core/tools/core/test_document_search_collection_concurrency.py +++ b/tests/core/tools/core/test_document_search_collection_concurrency.py @@ -359,6 +359,7 @@ async def _list( @pytest.mark.asyncio +@pytest.mark.timeout(30) async def test_a_stuck_collection_times_out_without_holding_the_batch( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -369,19 +370,15 @@ async def test_a_stuck_collection_times_out_without_holding_the_batch( def search(*, collection: str, **_kwargs: Any) -> SearchPipelineResult: if collection == "stuck": - release.wait(10) + release.wait() return _pipeline_result(collection) monkeypatch.setattr(document_search, "run_document_search", search) - started = time.perf_counter() try: result = await document_search._search_knowledge_base_impl(_args(), user_id=1) finally: release.set() - elapsed = time.perf_counter() - started - - assert elapsed < 5 assert [entry.collection for entry in result.results] == ["fast"] assert "stuck: search timed out after 0.2s" in result.summary diff --git a/tests/web/api/test_kb_collections_listing.py b/tests/web/api/test_kb_collections_listing.py index edc3c7feba..0039fbec1d 100644 --- a/tests/web/api/test_kb_collections_listing.py +++ b/tests/web/api/test_kb_collections_listing.py @@ -5,7 +5,6 @@ """ import asyncio -import time from types import SimpleNamespace import pytest @@ -137,6 +136,7 @@ async def test_team_owner_scan_results_are_merged_per_owner(team_listing_env): @pytest.mark.asyncio +@pytest.mark.timeout(30) async def test_scan_timeout_comes_from_config(monkeypatch): """The per-scan deadline must be configurable, not a hardcoded constant.""" monkeypatch.setenv("XAGENT_KB_COLLECTIONS_TIMEOUT_SECONDS", "1") @@ -145,16 +145,22 @@ async def test_scan_timeout_comes_from_config(monkeypatch): ) async def _slow_scan(user_id: int, is_admin: bool = False): - await asyncio.sleep(5) + await asyncio.Event().wait() return _result("never") monkeypatch.setattr(kb_api, "list_collections", _slow_scan) user = SimpleNamespace(id=1, is_admin=False) - started = time.perf_counter() + configured_timeouts = [] + original_wait_for = asyncio.wait_for + + async def observed_wait_for(awaitable, timeout): + configured_timeouts.append(timeout) + return await original_wait_for(awaitable, timeout=timeout) + + monkeypatch.setattr(kb_api.asyncio, "wait_for", observed_wait_for) with pytest.raises(kb_api.HTTPException) as excinfo: await kb_api.list_collections_api(_user=user, db=None) - elapsed = time.perf_counter() - started assert excinfo.value.status_code == 503 - assert elapsed < 3, f"configured 1s timeout was not honoured ({elapsed:.3f}s)" + assert configured_timeouts == [1] diff --git a/tests/web/api/test_upload_connection_boundary.py b/tests/web/api/test_upload_connection_boundary.py index e3cefb8ec8..33c6698b88 100644 --- a/tests/web/api/test_upload_connection_boundary.py +++ b/tests/web/api/test_upload_connection_boundary.py @@ -6,7 +6,6 @@ import builtins import io import threading -import time from pathlib import Path import pytest @@ -96,14 +95,18 @@ def get_upload_path( class _SlowTrackingSource(io.BytesIO): """A synchronous source that reveals an accidental event-loop read.""" - def __init__(self, payload: bytes, delay_seconds: float) -> None: + def __init__(self, payload: bytes) -> None: super().__init__(payload) - self.delay_seconds = delay_seconds + self.started = threading.Event() + self.release = threading.Event() + self.loop_thread = threading.get_ident() self.read_threads: list[int] = [] def read(self, size: int = -1) -> bytes: self.read_threads.append(threading.get_ident()) - time.sleep(self.delay_seconds) + self.started.set() + assert threading.get_ident() != self.loop_thread + assert self.release.wait(timeout=30), "upload read was never released" return super().read(size) @@ -213,14 +216,13 @@ async def test_staging_copy_reads_off_loop_with_no_database_checkout( files_api, "get_upload_path", _stage_path_in(upload_root, user_id) ) engine = _install_one_slot_queue_pool(monkeypatch) - source = _SlowTrackingSource(b"bounded-copy", delay_seconds=0.2) + source = _SlowTrackingSource(b"bounded-copy") upload = UploadFile( filename="off-loop.txt", file=source, headers={"content-type": "text/plain"}, ) loop_thread = threading.get_ident() - started_at = asyncio.get_running_loop().time() task = asyncio.create_task( files_api.store_uploaded_files( upload_items=[upload], @@ -232,15 +234,20 @@ async def test_staging_copy_reads_off_loop_with_no_database_checkout( ) ) try: - await asyncio.sleep(0.01) - assert asyncio.get_running_loop().time() - started_at < 0.1 + assert await asyncio.to_thread(source.started.wait, 30) + assert not task.done() assert engine.pool.checkedout() == 0 - await task + source.release.set() + await asyncio.wait_for(task, timeout=30) assert source.read_threads assert all(thread_id != loop_thread for thread_id in source.read_threads) assert engine.pool.checkedout() == 0 finally: - engine.dispose() + source.release.set() + try: + await asyncio.wait_for(task, timeout=30) + finally: + engine.dispose() def test_reserve_and_copy_enforces_max_size_and_cleans_partial_file( diff --git a/tests/web/api/v1/test_auth.py b/tests/web/api/v1/test_auth.py index 8851f00cc0..7365c652be 100644 --- a/tests/web/api/v1/test_auth.py +++ b/tests/web/api/v1/test_auth.py @@ -519,15 +519,19 @@ async def test_record_key_usage_pool_wait_keeps_event_loop_responsive( async def invoke_usage() -> None: await record_key_usage(prefix) - usage_task = asyncio.create_task(invoke_usage()) - started_at = asyncio.get_running_loop().time() - ticker_task = asyncio.create_task(asyncio.sleep(0.02)) + from tests.web.pool_contention_shared import GUARD_TIMEOUT, gated_pool_checkout + try: - await ticker_task - assert asyncio.get_running_loop().time() - started_at < 0.08 + with gated_pool_checkout(engine) as gate: + usage_task = asyncio.create_task(invoke_usage()) + try: + await gate.wait_until_contending() + assert not usage_task.done() + finally: + held_connection.close() + gate.let_through() + await asyncio.wait_for(usage_task, timeout=GUARD_TIMEOUT) finally: - held_connection.close() - await usage_task engine.dispose() diff --git a/tests/web/api/v1/test_tasks.py b/tests/web/api/v1/test_tasks.py index 959087a9a1..529fb41a79 100644 --- a/tests/web/api/v1/test_tasks.py +++ b/tests/web/api/v1/test_tasks.py @@ -17,7 +17,6 @@ import asyncio import io import threading -import time from dataclasses import replace from datetime import UTC, datetime from types import SimpleNamespace @@ -618,10 +617,15 @@ async def test_upload_durable_phase_releases_pool_and_event_loop( engine = _install_one_slot_queue_pool(monkeypatch) checked_out_during_durable: list[int] = [] original_sync = ManagedFileRef.sync_to_durable + entered = threading.Event() + release = threading.Event() + loop_thread = threading.get_ident() def delayed_sync(self, *args, **kwargs): # type: ignore[no-untyped-def] checked_out_during_durable.append(engine.pool.checkedout()) - time.sleep(0.1) + entered.set() + assert threading.get_ident() != loop_thread + assert release.wait(timeout=30), "durable upload was never released" return original_sync(self, *args, **kwargs) monkeypatch.setattr(ManagedFileRef, "sync_to_durable", delayed_sync) @@ -641,17 +645,18 @@ async def upload_once() -> None: user_id=user_id, ) - started_at = asyncio.get_running_loop().time() upload_task = asyncio.create_task(upload_once()) - ticker_task = asyncio.create_task(asyncio.sleep(0.02)) try: - await ticker_task - assert asyncio.get_running_loop().time() - started_at < 0.08 - await upload_task + assert await asyncio.to_thread(entered.wait, 30) + assert not upload_task.done() assert checked_out_during_durable == [0] assert engine.pool.checkedout() == 0 finally: - engine.dispose() + release.set() + try: + await asyncio.wait_for(upload_task, timeout=30) + finally: + engine.dispose() @pytest.mark.asyncio diff --git a/tests/web/services/test_task_orchestrator.py b/tests/web/services/test_task_orchestrator.py index c0156bfdd3..5c023c39ac 100644 --- a/tests/web/services/test_task_orchestrator.py +++ b/tests/web/services/test_task_orchestrator.py @@ -20,7 +20,7 @@ from concurrent.futures import ThreadPoolExecutor from contextlib import ExitStack, contextmanager from datetime import datetime, timedelta, timezone -from threading import Barrier, get_ident +from threading import Barrier, Event, get_ident from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -1411,17 +1411,17 @@ async def test_schedule_claimed_create_turn_offloads_cache_invalidation( monkeypatch: pytest.MonkeyPatch, ) -> None: """A committed domain claim must not invalidate Redis on the event loop.""" - import time as _time - task_id = 987654 event_loop_thread = get_ident() invalidations: list[tuple[int, int]] = [] - ticker_stop = asyncio.Event() - ticks = 0 + entered = Event() + release = Event() def slow_invalidate(observed_task_id: int) -> None: invalidations.append((observed_task_id, get_ident())) - _time.sleep(0.08) + entered.set() + assert get_ident() != event_loop_thread + assert release.wait(timeout=30), "cache invalidation was never released" async def fake_schedule(**_kwargs): async def done() -> None: @@ -1429,12 +1429,6 @@ async def done() -> None: return asyncio.create_task(done()) - async def ticker() -> None: - nonlocal ticks - while not ticker_stop.is_set(): - ticks += 1 - await asyncio.sleep(0.005) - monkeypatch.setattr( task_orchestrator_module, "invalidate_task_cache", @@ -1458,24 +1452,26 @@ async def ticker() -> None: run_id="committed-run", ) - ticker_task = asyncio.create_task(ticker()) - try: - started = await TaskTurnOrchestrator.schedule_claimed_create_turn( + startup = asyncio.create_task( + TaskTurnOrchestrator.schedule_claimed_create_turn( task_id=task_id, task_owner_user_id=1, actor_user_id=1, payload=TaskTurnPayload("start"), claimed=claimed, ) - await started.background_task + ) + try: + assert await asyncio.to_thread(entered.wait, 30) + assert not startup.done() finally: - ticker_stop.set() - await ticker_task + release.set() + started = await asyncio.wait_for(startup, timeout=30) + await started.background_task assert len(invalidations) == 1 assert invalidations[0][0] == task_id assert invalidations[0][1] != event_loop_thread - assert ticks >= 3, "claim-cache invalidation blocked the asyncio event loop" @pytest.mark.asyncio