From 7e596af53c5dca338e62325e4bd30fd2f6ba7496 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebasti=C3=A1n=20Ram=C3=ADrez?= Date: Fri, 6 Feb 2026 16:10:55 +0100 Subject: [PATCH] =?UTF-8?q?=E2=9C=85=20Add=20tests=20for=20coverage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_router_events.py | 76 +++++++++++++++++++++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/tests/test_router_events.py b/tests/test_router_events.py index 9df299cdaa..65f2f521c1 100644 --- a/tests/test_router_events.py +++ b/tests/test_router_events.py @@ -241,3 +241,79 @@ def test_merged_mixed_state_lifespans() -> None: with TestClient(app) as client: assert client.app_state == {"router": True} + + +@pytest.mark.filterwarnings( + r"ignore:\s*on_event is deprecated, use lifespan event handlers instead.*:DeprecationWarning" +) +def test_router_async_shutdown_handler(state: State) -> None: + """Test that async on_shutdown event handlers are called correctly, for coverage.""" + app = FastAPI() + + @app.get("/") + def main() -> dict[str, str]: + return {"message": "Hello World"} + + @app.on_event("shutdown") + async def app_shutdown() -> None: + state.app_shutdown = True + + assert state.app_shutdown is False + with TestClient(app) as client: + assert state.app_shutdown is False + response = client.get("/") + assert response.status_code == 200, response.text + assert state.app_shutdown is True + + +def test_router_sync_generator_lifespan(state: State) -> None: + """Test that a sync generator lifespan works via _wrap_gen_lifespan_context.""" + from collections.abc import Generator + + def lifespan(app: FastAPI) -> Generator[None, None, None]: + state.app_startup = True + yield + state.app_shutdown = True + + app = FastAPI(lifespan=lifespan) # type: ignore[arg-type] + + @app.get("/") + def main() -> dict[str, str]: + return {"message": "Hello World"} + + assert state.app_startup is False + assert state.app_shutdown is False + with TestClient(app) as client: + assert state.app_startup is True + assert state.app_shutdown is False + response = client.get("/") + assert response.status_code == 200, response.text + assert response.json() == {"message": "Hello World"} + assert state.app_startup is True + assert state.app_shutdown is True + + +def test_router_async_generator_lifespan(state: State) -> None: + """Test that an async generator lifespan (not wrapped) works.""" + + async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: + state.app_startup = True + yield + state.app_shutdown = True + + app = FastAPI(lifespan=lifespan) # type: ignore[arg-type] + + @app.get("/") + def main() -> dict[str, str]: + return {"message": "Hello World"} + + assert state.app_startup is False + assert state.app_shutdown is False + with TestClient(app) as client: + assert state.app_startup is True + assert state.app_shutdown is False + response = client.get("/") + assert response.status_code == 200, response.text + assert response.json() == {"message": "Hello World"} + assert state.app_startup is True + assert state.app_shutdown is True