diff --git a/tests/hermes_cli/test_web_read_coalescing.py b/tests/hermes_cli/test_web_read_coalescing.py index 467bf28f91..d1e1a0db8e 100644 --- a/tests/hermes_cli/test_web_read_coalescing.py +++ b/tests/hermes_cli/test_web_read_coalescing.py @@ -1,4 +1,4 @@ -"""Behavioral coverage for pre-threadpool HTTP read admission.""" +"""GET /api/profiles coalesces a poll burst before AnyIO threadpool admission (#105179).""" import asyncio import threading from functools import wraps @@ -19,209 +19,6 @@ def async_test(fn): return run -def decorator(): - from hermes_cli import web_read_coalescing - return web_read_coalescing.coalesced_read - - -@async_test -async def test_identical_inflight_calls_share_one_read(): - release = threading.Event() - calls = [] - - @decorator() - def read(key=1): - calls.append(key) - assert release.wait(3) - return key - - tasks = [asyncio.create_task(read()) for _ in range(12)] - try: - await wait_until(lambda: calls) - await asyncio.sleep(0.05) - assert calls == [1] - finally: - release.set() - results = await asyncio.gather(*tasks) - assert results == [1] * 12 - assert await read() == 1 - assert calls == [1, 1] - - -@async_test -async def test_differing_keys_wait_before_threadpool_admission(): - import anyio.to_thread - release = threading.Event() - calls = [] - - @decorator() - def read(key): - calls.append(key) - assert release.wait(3) - return key - - limiter = anyio.to_thread.current_default_thread_limiter() - tasks = [asyncio.create_task(read(key)) for key in range(8)] - try: - await wait_until(lambda: calls) - await asyncio.sleep(0.05) - assert len(calls) == 1 - assert limiter.borrowed_tokens == 1 - finally: - release.set() - results = await asyncio.gather(*tasks) - assert results == list(range(8)) - - -@async_test -async def test_caller_cancellation_keeps_worker_and_admission(): - release = threading.Event() - calls = [] - - @decorator() - def read(key): - calls.append(key) - assert release.wait(3) - return key - - first = asyncio.create_task(read(1)) - tasks = [] - try: - await wait_until(lambda: calls) - first.cancel() - with pytest.raises(asyncio.CancelledError): - await first - tasks = [asyncio.create_task(read(1)), asyncio.create_task(read(2))] - await asyncio.sleep(0.05) - assert calls == [1] - finally: - release.set() - results = await asyncio.gather(*tasks) - assert results == [1, 2] - assert calls == [1, 2] - - -@async_test -async def test_canonical_arguments_coalesce(): - release = threading.Event() - calls = [] - - @decorator() - def read(key=1, *, limit=2): - calls.append(key) - assert release.wait(3) - return (key, limit) - - tasks = [asyncio.create_task(read()), asyncio.create_task(read(1)), - asyncio.create_task(read(limit=2, key=1))] - try: - await wait_until(lambda: calls) - await asyncio.sleep(0.05) - finally: - release.set() - results = await asyncio.gather(*tasks) - assert results == [(1, 2)] * 3 - assert calls == [1] - - -@async_test -async def test_home_isolation_preserves_worker_context(monkeypatch): - from contextvars import ContextVar - from pathlib import Path - from hermes_cli import web_read_coalescing - home = ContextVar("test_read_home", default=Path("/first")) - monkeypatch.setattr(web_read_coalescing, "get_hermes_home", home.get, raising=False) - release = threading.Event() - calls = [] - - @decorator() - def read(): - calls.append(str(home.get())) - assert release.wait(3) - return str(home.get()) - - first = asyncio.create_task(read()) - home.set(Path("/second")) - second = asyncio.create_task(read()) - try: - await wait_until(lambda: calls) - await asyncio.sleep(0.05) - finally: - release.set() - results = await asyncio.gather(first, second) - assert results == ["/first", "/second"] - assert calls == results - - -def test_fastapi_can_resolve_postponed_annotations(): - import inspect - from pathlib import Path - from fastapi import FastAPI - namespace = {"Path": Path} - exec("from __future__ import annotations\ndef read(key: Path) -> dict:\n return {'key': str(key)}", namespace) - read = decorator()(namespace["read"]) - assert inspect.iscoroutinefunction(read) - assert inspect.signature(read).parameters["key"].annotation is Path - app = FastAPI() - app.get("/read")(read) - assert app.openapi()["paths"]["/read"]["get"]["parameters"][0]["name"] == "key" - - -@async_test -async def test_failed_shared_read_is_retried(): - release = threading.Event() - calls = [] - - @decorator() - def read(): - calls.append(1) - assert release.wait(3) - if len(calls) == 1: - raise ValueError("read failed") - return "recovered" - - tasks = [asyncio.create_task(read()) for _ in range(5)] - await wait_until(lambda: calls) - await asyncio.sleep(0.05) - release.set() - errors = await asyncio.gather(*tasks, return_exceptions=True) - assert all(isinstance(error, ValueError) for error in errors) - assert calls == [1] - assert await read() == "recovered" - assert calls == [1, 1] - - -@async_test -async def test_abandoned_failure_is_observed_and_retryable(): - import gc - release = threading.Event() - calls = [] - errors = [] - loop = asyncio.get_running_loop() - loop.set_exception_handler(lambda loop, context: errors.append(context)) - - @decorator() - def read(): - calls.append(1) - assert release.wait(3) - raise ValueError("abandoned read") - - caller = asyncio.create_task(read()) - try: - await wait_until(lambda: calls) - caller.cancel() - with pytest.raises(asyncio.CancelledError): - await caller - finally: - release.set() - await asyncio.sleep(0.05) - gc.collect() - assert errors == [] - with pytest.raises(ValueError, match="abandoned read"): - await read() - assert calls == [1, 1] - - @async_test async def test_profiles_burst_leaves_threadpool_status_responsive(monkeypatch): import anyio.to_thread @@ -269,106 +66,3 @@ async def test_profiles_burst_leaves_threadpool_status_responsive(monkeypatch): assert all(r.json() == {"profiles": [{"name": "example"}]} for r in responses) finally: limiter.total_tokens = previous - - -@async_test -async def test_profiles_retains_threadpool_monkeypatch_seam(monkeypatch): - from hermes_cli import profiles as profiles_mod - from hermes_cli.web_routers import profiles - calls = [] - - async def runner(func): - calls.append("runner") - return func() - - monkeypatch.setattr(profiles, "run_in_threadpool", runner) - monkeypatch.setattr(profiles_mod, "list_profiles", lambda: ["example"]) - monkeypatch.setattr(profiles, "_profile_to_dict", lambda p: {"name": p}) - assert await profiles.list_profiles_endpoint() == {"profiles": [{"name": "example"}]} - assert calls == ["runner"] - - -@async_test -async def test_container_arguments_are_canonical_and_type_safe(): - release = threading.Event() - calls = [] - - @decorator() - def read(options): - calls.append(options) - assert release.wait(3) - return options - - values = [{"a": [1], "b": {2, 3}}, {"b": {3, 2}, "a": [1]}, True, 1] - tasks = [asyncio.create_task(read(value)) for value in values] - try: - await asyncio.sleep(0.05) - finally: - release.set() - results = await asyncio.gather(*tasks) - assert results == values - assert len(calls) == 3 - assert type(results[2]) is bool - assert type(results[3]) is int - - -def test_decorated_function_is_isolated_between_live_event_loops(): - from concurrent.futures import ThreadPoolExecutor - rendezvous = threading.Barrier(2) - - @decorator() - def read(): - rendezvous.wait(timeout=3) - return threading.get_ident() - - with ThreadPoolExecutor(max_workers=2) as executor: - futures = [executor.submit(asyncio.run, read()) for _ in range(2)] - results = [future.result(timeout=5) for future in futures] - assert len(set(results)) == 2 - - -@async_test -async def test_profiles_failure_preserves_fallback_and_retries(monkeypatch): - from hermes_cli import profiles as profiles_mod - from hermes_cli.web_routers import profiles - from starlette.concurrency import run_in_threadpool - calls = [] - - def read(): - calls.append(1) - if len(calls) == 1: - raise OSError("unavailable") - return ["recovered"] - - monkeypatch.setattr(profiles, "run_in_threadpool", run_in_threadpool) - monkeypatch.setattr(profiles_mod, "list_profiles", read) - monkeypatch.setattr(profiles, "_profile_to_dict", lambda p: {"name": p}) - monkeypatch.setattr(profiles, "_fallback_profile_dicts", lambda mod: [{"name": "fallback"}]) - assert await profiles.list_profiles_endpoint() == {"profiles": [{"name": "fallback"}]} - assert await profiles.list_profiles_endpoint() == {"profiles": [{"name": "recovered"}]} - assert calls == [1, 1] - - -@async_test -async def test_profiles_fallback_is_coalesced_off_event_loop(monkeypatch): - from hermes_cli import profiles as profiles_mod - from hermes_cli.web_routers import profiles - from starlette.concurrency import run_in_threadpool - event_loop_thread = threading.get_ident() - fallback_threads = [] - expected = {"profiles": [{"name": "fallback"}]} - - def unavailable(): - raise OSError("unavailable") - - def fallback(_): - fallback_threads.append(threading.get_ident()) - return expected["profiles"] - - monkeypatch.setattr(profiles, "run_in_threadpool", run_in_threadpool) - monkeypatch.setattr(profiles_mod, "list_profiles", unavailable) - monkeypatch.setattr(profiles, "_fallback_profile_dicts", fallback) - results = await asyncio.gather(*(profiles.list_profiles_endpoint() for _ in range(8))) - assert all(result == expected for result in results) - assert event_loop_thread not in fallback_threads - assert len(fallback_threads) == 1 \ No newline at end of file