tests/core/test_lifespan.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 """Lifespan protocol tests (SPECIFICATION.md §4): startup in order, shutdown16 in reverse, one app's error does not block the others.17 18 The ASGI lifespan protocol is driven directly through ``BaseServer.__call__``19 (no uvicorn needed): a canned ``receive()`` queue delivers ``startup`` then20 ``shutdown``, and the recording apps append to a shared event list from their21 hooks so ordering and error isolation can be asserted.22 """23 24 from __future__ import annotations25 26 from genro_asgi import BaseApplication, BaseServer27 from genro_asgi.lifespan import QUITTING, STOPPING, FatalBootError, Lifespan28 29 30 class SyncRecordingApp(BaseApplication):31 """Test app recording sync ``on_startup``/``on_shutdown`` to a shared list.32 33 Constructor kwargs peeled here: ``name`` — identifies this app in the34 recorded events; ``events`` — the shared list; ``raise_on`` — an optional35 iterable of hook names on which this app raises instead of recording.36 """37 38 def __init__(self, **kwargs: object) -> None:39 self.name: str = kwargs.pop("name")40 self.events: list[str] = kwargs.pop("events")41 self.raise_on: frozenset[str] = frozenset(kwargs.pop("raise_on", ()))42 super().__init__(**kwargs)43 44 def on_startup(self) -> None:45 self._record_or_raise("on_startup")46 47 def on_shutdown(self) -> None:48 self._record_or_raise("on_shutdown")49 50 def _record_or_raise(self, hook: str) -> None:51 if hook in self.raise_on:52 raise RuntimeError(f"{self.name}.{hook} failed")53 self.events.append(f"{self.name}.{hook}")54 55 56 class AsyncRecordingApp(SyncRecordingApp):57 """Same recording behaviour, as async hooks."""58 59 async def on_startup(self) -> None:60 self._record_or_raise("on_startup")61 62 async def on_shutdown(self) -> None:63 self._record_or_raise("on_shutdown")64 65 66 async def drive_lifespan(67 server: BaseServer, messages: list[dict[str, object]]68 ) -> list[dict[str, object]]:69 """Feed ``messages`` to ``server``'s lifespan scope; return what it sent."""70 queue = list(messages)71 sent: list[dict[str, object]] = []72 73 async def receive() -> dict[str, object]:74 return queue.pop(0)75 76 async def send(message: dict[str, object]) -> None:77 sent.append(message)78 79 await server({"type": "lifespan"}, receive, send)80 return sent81 82 83 def startup_then_shutdown() -> list[dict[str, object]]:84 """A canned message queue: one full startup/shutdown round-trip."""85 return [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]86 87 88 class TestHandlerWiring:89 def test_the_handler_holds_the_server_that_built_it(self) -> None:90 server = BaseServer(applications=[BaseApplication(mount="")])91 assert Lifespan(server).server is server92 assert server.lifespan.server is server93 94 95 class TestOrdering:96 async def test_startup_runs_the_applications_in_registration_order(self) -> None:97 events: list[str] = []98 server = BaseServer(99 applications=[100 SyncRecordingApp(mount="", name="root", events=events),101 SyncRecordingApp(name="api", code="api", events=events),102 SyncRecordingApp(name="admin", code="admin", events=events),103 ]104 )105 106 sent = await drive_lifespan(server, startup_then_shutdown())107 108 startup_events = [e for e in events if e.endswith("on_startup")]109 assert startup_events == ["root.on_startup", "api.on_startup", "admin.on_startup"]110 assert {"type": "lifespan.startup.complete"} in sent111 112 async def test_shutdown_runs_in_reverse_order(self) -> None:113 events: list[str] = []114 server = BaseServer(115 applications=[116 SyncRecordingApp(mount="", name="root", events=events),117 SyncRecordingApp(name="api", code="api", events=events),118 SyncRecordingApp(name="admin", code="admin", events=events),119 ]120 )121 122 sent = await drive_lifespan(server, startup_then_shutdown())123 124 shutdown_events = [e for e in events if e.endswith("on_shutdown")]125 assert shutdown_events == ["admin.on_shutdown", "api.on_shutdown", "root.on_shutdown"]126 assert {"type": "lifespan.shutdown.complete"} in sent127 128 129 class TestErrorIsolation:130 async def test_raising_sync_startup_hook_does_not_block_others(self) -> None:131 events: list[str] = []132 server = BaseServer(133 applications=[134 SyncRecordingApp(mount="", name="root", events=events),135 SyncRecordingApp(name="api", code="api", events=events, raise_on={"on_startup"}),136 SyncRecordingApp(name="admin", code="admin", events=events),137 ]138 )139 140 sent = await drive_lifespan(server, startup_then_shutdown())141 142 startup_events = [e for e in events if e.endswith("on_startup")]143 assert startup_events == ["root.on_startup", "admin.on_startup"]144 assert {"type": "lifespan.startup.complete"} in sent145 assert {"type": "lifespan.shutdown.complete"} in sent146 147 async def test_raising_async_shutdown_hook_does_not_block_others(self) -> None:148 events: list[str] = []149 server = BaseServer(150 applications=[151 AsyncRecordingApp(mount="", name="root", events=events),152 AsyncRecordingApp(name="api", code="api", events=events, raise_on={"on_shutdown"}),153 AsyncRecordingApp(name="admin", code="admin", events=events),154 ]155 )156 157 sent = await drive_lifespan(server, startup_then_shutdown())158 159 shutdown_events = [e for e in events if e.endswith("on_shutdown")]160 assert shutdown_events == ["admin.on_shutdown", "root.on_shutdown"]161 assert {"type": "lifespan.startup.complete"} in sent162 assert {"type": "lifespan.shutdown.complete"} in sent163 164 165 class FatalStartupApp(SyncRecordingApp):166 """Its startup failure is declared fatal: the server must not start."""167 168 def on_startup(self) -> None:169 raise FatalBootError(f"{self.name}: the server must not start")170 171 172 class TestFatalBoot:173 async def test_a_fatal_startup_failure_stops_the_server(self) -> None:174 events: list[str] = []175 server = BaseServer(176 applications=[177 SyncRecordingApp(mount="", name="root", events=events),178 FatalStartupApp(name="api", code="api", events=events),179 SyncRecordingApp(name="admin", code="admin", events=events),180 ]181 )182 183 sent = await drive_lifespan(server, [{"type": "lifespan.startup"}])184 185 assert sent == [186 {"type": "lifespan.startup.failed", "message": "api: the server must not start"}187 ]188 assert [e for e in events if e.endswith("on_startup")] == ["root.on_startup"]189 190 async def test_a_fatal_error_on_shutdown_keeps_the_ordinary_isolation(self) -> None:191 events: list[str] = []192 193 class FatalOnShutdown(SyncRecordingApp):194 def on_shutdown(self) -> None:195 raise FatalBootError(f"{self.name}.on_shutdown failed")196 197 server = BaseServer(198 applications=[199 SyncRecordingApp(mount="", name="root", events=events),200 FatalOnShutdown(name="api", code="api", events=events),201 ]202 )203 204 sent = await drive_lifespan(server, startup_then_shutdown())205 206 shutdown_events = [e for e in events if e.endswith("on_shutdown")]207 assert shutdown_events == ["root.on_shutdown"]208 assert {"type": "lifespan.shutdown.complete"} in sent209 210 211 class TestShutdownState:212 async def test_the_shutdown_stops_accepting_before_any_hook_runs(self) -> None:213 seen: list[str] = []214 215 class Watcher(BaseApplication):216 def on_shutdown(self) -> None:217 seen.append(self.server.state)218 219 server = BaseServer(applications=[Watcher(mount="")])220 await Lifespan(server).shutdown()221 222 assert seen == [STOPPING]223 224 async def test_the_reload_trigger_makes_the_shutdown_a_quit(self) -> None:225 server = BaseServer(applications=[BaseApplication(mount="")])226 server.shutdown_mode = QUITTING227 await Lifespan(server).shutdown()228 assert server.state == QUITTING229 230 async def test_a_state_somebody_already_chose_is_respected(self) -> None:231 server = BaseServer(applications=[BaseApplication(mount="")])232 server.state = QUITTING233 await Lifespan(server).shutdown()234 assert server.state == QUITTING235 236 async def test_the_shutdown_drains_what_is_in_flight_before_the_hooks(self) -> None:237 import asyncio238 239 gate = asyncio.Event()240 in_flight_at_hook: list[int] = []241 242 class Held(BaseApplication):243 async def __call__(self, scope, receive, send) -> None:244 await gate.wait()245 await send({"type": "http.response.start", "status": 200, "headers": []})246 await send({"type": "http.response.body", "body": b"ok"})247 248 def on_shutdown(self) -> None:249 in_flight_at_hook.append(self.server.requests.in_flight)250 251 server = BaseServer(applications=[Held(mount="")])252 253 async def receive():254 return {"type": "http.request"}255 256 async def send(message):257 pass258 259 request = asyncio.ensure_future(server({"type": "http", "path": "/"}, receive, send))260 await asyncio.sleep(0)261 closing = asyncio.ensure_future(Lifespan(server).shutdown())262 await asyncio.sleep(0)263 gate.set()264 await closing265 await request266 267 assert in_flight_at_hook == [0]