diff --git a/tests/test_router_events.py b/tests/test_router_events.py index 9df299cda..65f2f521c 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