tests/core/test_pool.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Thread-pool tests (SPECIFICATION.md §4): a sync handler runs on the pool16 (thread identity asserted), an async handler stays on the loop, the pool is17 provisioned lazily on the first sync dispatch and torn down at shutdown.18 19 ``run_sync`` is exercised directly (no uvicorn) and the throwaway app's sync20 route is driven through ``BaseServer.__call__`` at the ASGI level.21 """22 23 from __future__ import annotations24 25 import threading26 27 from genro_asgi import BaseServer28 29 from ..throwaway_app import ThrowawayApp30 31 32 def make_server() -> BaseServer:33 """A server with a throwaway primary — enough to exercise the pool."""34 return BaseServer(applications=[ThrowawayApp(mount="", name="primary")])35 36 37 class TestDispatch:38 async def test_sync_handler_runs_on_a_pool_thread(self) -> None:39 server = make_server()40 name = await server.run_sync(lambda: threading.current_thread().name)41 assert name.startswith("genro-pool")42 43 async def test_async_stays_on_loop_sync_goes_off_loop(self) -> None:44 server = make_server()45 loop_ident = threading.get_ident()46 47 async def async_handler() -> int:48 return threading.get_ident()49 50 assert await async_handler() == loop_ident51 assert await server.run_sync(threading.get_ident) != loop_ident52 53 54 class TestMaxThreads:55 async def test_max_threads_reaches_the_executor(self) -> None:56 server = BaseServer(applications=[ThrowawayApp(mount="", name="primary")], max_threads=2)57 await server.run_sync(lambda: None)58 assert server.pool.executor._max_workers == 259 60 61 class TestProvisioning:62 async def test_pool_is_not_provisioned_before_first_dispatch(self) -> None:63 server = make_server()64 assert server.pool.provisioned is False65 await server.run_sync(lambda: None)66 assert server.pool.provisioned is True67 68 async def test_shutdown_resets_the_pool_for_reprovisioning(self) -> None:69 server = make_server()70 await server.run_sync(lambda: None)71 assert server.pool.provisioned is True72 73 server.pool.shutdown(wait=True)74 assert server.pool.provisioned is False75 76 # a later dispatch lazily re-provisions: server reuse via repeated serve()77 name = await server.run_sync(lambda: threading.current_thread().name)78 assert name.startswith("genro-pool")79 assert server.pool.provisioned is True80 81 82 class TestThrowawayRoute:83 async def test_sync_route_dispatches_through_the_pool(self) -> None:84 server = make_server()85 sent: list[dict[str, object]] = []86 87 async def receive() -> dict[str, object]:88 return {"type": "http.request"}89 90 async def send(message: dict[str, object]) -> None:91 sent.append(message)92 93 await server({"type": "http", "path": "/sync"}, receive, send)94 95 assert server.pool.provisioned is True96 body = next(m["body"] for m in sent if m["type"] == "http.response.body")97 assert body == b"sync:primary"98 99 100 class TestContextPropagation:101 async def test_sync_handler_sees_its_own_current_request(self) -> None:102 server = make_server()103 sent: list[dict[str, object]] = []104 105 async def receive() -> dict[str, object]:106 return {"type": "http.request"}107 108 async def send(message: dict[str, object]) -> None:109 sent.append(message)110 111 await server({"type": "http", "path": "/sync-current"}, receive, send)112 113 # the pool copies the caller's context: the worker thread reads the114 # registry's ContextVar and sees the request being served115 body = next(m["body"] for m in sent if m["type"] == "http.response.body")116 assert body == b"current:/sync-current"117 118 119 class TestMetrics:120 async def test_metrics_are_zeros_until_the_pool_is_provisioned(self) -> None:121 """Zeros, honestly: a pool that does not exist reports no pressure."""122 server = make_server()123 assert server.pool.metrics == {"total": 0, "busy": 0}124 125 async def test_busy_returns_to_zero_after_a_run(self) -> None:126 server = make_server()127 await server.run_sync(lambda: None)128 assert server.pool.metrics["busy"] == 0129 130 async def test_total_mirrors_max_threads(self) -> None:131 """The resolution is ours, not read off a private executor attribute."""132 server = BaseServer(applications=[ThrowawayApp(mount="", name="primary")], max_threads=3)133 await server.run_sync(lambda: None)134 assert server.pool.metrics["total"] == 3135 136 async def test_total_mirrors_the_executor_own_resolution(self) -> None:137 """The mirror against the reality, on ANY interpreter: the frozen total138 equals the thread count the stdlib default actually decided (3.13 moved139 it from cpu_count to process_cpu_count — the mirror must move with it).140 Reading the private attribute is legitimate HERE: this is the one test141 that verifies the mirror, so the source never has to."""142 server = make_server()143 await server.run_sync(lambda: None)144 assert server.pool.metrics["total"] == server.pool.executor._max_workers145 146 async def test_busy_counts_the_call_in_flight_while_it_runs(self) -> None:147 """``busy`` is demand — every run() entered and not yet exited, queued148 included — not slots held; past saturation it exceeds ``total``."""149 server = make_server()150 seen: list[int] = []151 152 def probe() -> None:153 seen.append(server.pool.metrics["busy"])154 155 await server.run_sync(probe)156 assert seen == [1]