|
|
@ -241,3 +241,79 @@ def test_merged_mixed_state_lifespans() -> None: |
|
|
|
|
|
|
|
|
with TestClient(app) as client: |
|
|
with TestClient(app) as client: |
|
|
assert client.app_state == {"router": True} |
|
|
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 |
|
|
|