essentiaMarco
4 weeks ago
committed by
GitHub
GitHub
9 changed files with 512 additions and 8 deletions
@ -0,0 +1,282 @@ |
|||
"""Tests for lifespan-scoped dependencies (Depends(..., scope="lifespan")).""" |
|||
|
|||
from contextlib import asynccontextmanager |
|||
from typing import Annotated |
|||
|
|||
import pytest |
|||
from fastapi import Depends, FastAPI |
|||
from fastapi.exceptions import DependencyScopeError |
|||
from fastapi.testclient import TestClient |
|||
from starlette.requests import Request |
|||
|
|||
|
|||
def test_lifespan_dependency_single_request() -> None: |
|||
"""Lifespan-scoped dependency is created once and reused across requests.""" |
|||
started: list[str] = [] |
|||
stopped: list[str] = [] |
|||
|
|||
def get_db() -> str: |
|||
started.append("db") |
|||
yield "db_conn" |
|||
stopped.append("db") |
|||
|
|||
app = FastAPI() |
|||
|
|||
@app.get("/") |
|||
def root(db: Annotated[str, Depends(get_db, scope="lifespan")]) -> dict[str, str]: |
|||
return {"db": db} |
|||
|
|||
assert len(started) == 0 |
|||
assert len(stopped) == 0 |
|||
|
|||
with TestClient(app) as client: |
|||
assert len(started) == 1, "lifespan dep should start once at app startup" |
|||
r1 = client.get("/") |
|||
assert r1.status_code == 200 |
|||
assert r1.json() == {"db": "db_conn"} |
|||
r2 = client.get("/") |
|||
assert r2.status_code == 200 |
|||
assert r2.json() == {"db": "db_conn"} |
|||
assert len(started) == 1, "lifespan dep should not restart per request" |
|||
|
|||
assert len(stopped) == 1, "lifespan dep should stop once at app shutdown" |
|||
|
|||
|
|||
def test_lifespan_dependency_with_custom_lifespan() -> None: |
|||
"""Lifespan-scoped dependency runs inside app lifespan and is cleaned up on shutdown.""" |
|||
started: list[str] = [] |
|||
stopped: list[str] = [] |
|||
|
|||
@asynccontextmanager |
|||
async def lifespan(app: FastAPI): |
|||
started.append("lifespan") |
|||
yield |
|||
stopped.append("lifespan") |
|||
|
|||
def get_pool() -> str: |
|||
started.append("pool") |
|||
yield "pool_conn" |
|||
stopped.append("pool") |
|||
|
|||
app = FastAPI(lifespan=lifespan) |
|||
|
|||
@app.get("/") |
|||
def root( |
|||
pool: Annotated[str, Depends(get_pool, scope="lifespan")], |
|||
) -> dict[str, str]: |
|||
return {"pool": pool} |
|||
|
|||
with TestClient(app) as client: |
|||
assert "lifespan" in started |
|||
assert "pool" in started |
|||
r = client.get("/") |
|||
assert r.status_code == 200 |
|||
assert r.json() == {"pool": "pool_conn"} |
|||
|
|||
assert "pool" in stopped |
|||
assert "lifespan" in stopped |
|||
|
|||
|
|||
def test_lifespan_dependency_same_instance_across_requests() -> None: |
|||
"""The same instance is injected for every request when scope is lifespan.""" |
|||
instances: list[object] = [] |
|||
|
|||
def get_singleton() -> object: |
|||
inst = object() |
|||
instances.append(inst) |
|||
yield inst |
|||
|
|||
app = FastAPI() |
|||
|
|||
@app.get("/") |
|||
def root( |
|||
s: Annotated[object, Depends(get_singleton, scope="lifespan")], |
|||
) -> dict[str, bool]: |
|||
return {"is_singleton": len(instances) == 1 and s is instances[0]} |
|||
|
|||
with TestClient(app) as client: |
|||
r1 = client.get("/") |
|||
r2 = client.get("/") |
|||
assert r1.status_code == 200 and r2.status_code == 200 |
|||
assert r1.json()["is_singleton"] is True |
|||
assert r2.json()["is_singleton"] is True |
|||
assert len(instances) == 1 |
|||
|
|||
|
|||
def test_lifespan_dependency_decorator_level_dependencies_runs_at_startup() -> None: |
|||
"""Decorator-level dependencies=[Depends(..., scope='lifespan')] run at startup once.""" |
|||
started: list[str] = [] |
|||
stopped: list[str] = [] |
|||
|
|||
def lifespan_dep() -> str: |
|||
started.append("lifespan_dep") |
|||
yield "ok" |
|||
stopped.append("lifespan_dep") |
|||
|
|||
app = FastAPI() |
|||
|
|||
@app.get("/", dependencies=[Depends(lifespan_dep, scope="lifespan")]) |
|||
def root() -> dict[str, str]: |
|||
return {"ok": "yes"} |
|||
|
|||
with TestClient(app) as client: |
|||
assert started == ["lifespan_dep"] |
|||
r1 = client.get("/") |
|||
r2 = client.get("/") |
|||
assert r1.status_code == 200 and r2.status_code == 200 |
|||
assert r1.json() == {"ok": "yes"} |
|||
assert r2.json() == {"ok": "yes"} |
|||
assert started == ["lifespan_dep"] |
|||
|
|||
assert stopped == ["lifespan_dep"] |
|||
|
|||
|
|||
def test_lifespan_dependency_synthetic_request_receive_send() -> None: |
|||
"""Lifespan dep that uses Request.receive covers noop_receive during startup.""" |
|||
|
|||
async def lifespan_dep(request: Request) -> str: |
|||
await request.receive() |
|||
return "ok" |
|||
|
|||
app = FastAPI() |
|||
|
|||
@app.get("/") |
|||
def root( |
|||
v: Annotated[str, Depends(lifespan_dep, scope="lifespan")], |
|||
) -> dict[str, str]: |
|||
return {"v": v} |
|||
|
|||
with TestClient(app) as client: |
|||
r = client.get("/") |
|||
assert r.status_code == 200 |
|||
assert r.json() == {"v": "ok"} |
|||
|
|||
|
|||
def test_lifespan_dependency_sync_callable() -> None: |
|||
"""Sync (non-gen, non-coroutine) lifespan dep runs via run_in_threadpool (utils 702).""" |
|||
|
|||
def sync_lifespan_dep() -> str: |
|||
return "sync_val" |
|||
|
|||
app = FastAPI() |
|||
|
|||
@app.get("/") |
|||
def root( |
|||
v: Annotated[str, Depends(sync_lifespan_dep, scope="lifespan")], |
|||
) -> dict[str, str]: |
|||
return {"v": v} |
|||
|
|||
with TestClient(app) as client: |
|||
r = client.get("/") |
|||
assert r.status_code == 200 |
|||
assert r.json() == {"v": "sync_val"} |
|||
|
|||
|
|||
def test_lifespan_dependency_nested() -> None: |
|||
"""Lifespan dep B depending on A covers dependency_cache hit path (utils.py line 685).""" |
|||
order: list[str] = [] |
|||
|
|||
def lifespan_a() -> str: |
|||
order.append("a") |
|||
yield "a" |
|||
|
|||
def lifespan_b( |
|||
a: Annotated[str, Depends(lifespan_a, scope="lifespan")], |
|||
) -> str: |
|||
order.append("b") |
|||
yield a + "-b" |
|||
|
|||
app = FastAPI() |
|||
|
|||
@app.get("/") |
|||
def root( |
|||
b: Annotated[str, Depends(lifespan_b, scope="lifespan")], |
|||
) -> dict[str, str]: |
|||
return {"b": b} |
|||
|
|||
with TestClient(app) as client: |
|||
r = client.get("/") |
|||
assert r.status_code == 200 |
|||
assert r.json() == {"b": "a-b"} |
|||
assert order == ["a", "b"] |
|||
|
|||
|
|||
def test_lifespan_dependency_shared_cache_hit() -> None: |
|||
"""Two lifespan deps B and C both depend on A; second resolution hits cache (utils 687).""" |
|||
order: list[str] = [] |
|||
|
|||
def lifespan_a() -> str: |
|||
order.append("a") |
|||
yield "a" |
|||
|
|||
def lifespan_b( |
|||
a: Annotated[str, Depends(lifespan_a, scope="lifespan")], |
|||
) -> str: |
|||
order.append("b") |
|||
yield a + "-b" |
|||
|
|||
def lifespan_c( |
|||
a: Annotated[str, Depends(lifespan_a, scope="lifespan")], |
|||
) -> str: |
|||
order.append("c") |
|||
yield a + "-c" |
|||
|
|||
app = FastAPI() |
|||
|
|||
@app.get("/") |
|||
def root( |
|||
b: Annotated[str, Depends(lifespan_b, scope="lifespan")], |
|||
c: Annotated[str, Depends(lifespan_c, scope="lifespan")], |
|||
) -> dict[str, str]: |
|||
return {"b": b, "c": c} |
|||
|
|||
with TestClient(app) as client: |
|||
r = client.get("/") |
|||
assert r.status_code == 200 |
|||
assert r.json() == {"b": "a-b", "c": "a-c"} |
|||
assert order == ["a", "b", "c"] |
|||
|
|||
|
|||
def test_lifespan_dependency_cannot_depend_on_request_scope() -> None: |
|||
"""Lifespan-scoped dependency that depends on request-scoped dep raises.""" |
|||
|
|||
def request_scoped() -> int: |
|||
return 1 # pragma: no cover - never run; raises at app.get("/")(root) |
|||
|
|||
def lifespan_dep( |
|||
x: Annotated[int, Depends(request_scoped, scope="request")], |
|||
) -> int: |
|||
return x # pragma: no cover - never run; raises at app.get("/")(root) |
|||
|
|||
def root( |
|||
y: Annotated[int, Depends(lifespan_dep, scope="lifespan")], |
|||
) -> dict[str, int]: |
|||
return {"y": y} # pragma: no cover - never run; raises at app.get("/")(root) |
|||
|
|||
app = FastAPI() |
|||
with pytest.raises(DependencyScopeError) as exc_info: |
|||
app.get("/")(root) |
|||
assert "lifespan" in str(exc_info.value) and "cannot depend" in str(exc_info.value) |
|||
|
|||
|
|||
def test_lifespan_dependency_not_initialized_raises() -> None: |
|||
"""Request that needs a lifespan dep which was not run (e.g. mounted sub-app) raises.""" |
|||
|
|||
def lifespan_dep() -> str: |
|||
yield "conn" # pragma: no cover - never run; request raises before dep runs |
|||
|
|||
sub_app = FastAPI() |
|||
|
|||
@sub_app.get("/sub") |
|||
def sub_root( |
|||
x: Annotated[str, Depends(lifespan_dep, scope="lifespan")], |
|||
) -> dict[str, str]: |
|||
return {"x": x} # pragma: no cover - never run; request raises before handler |
|||
|
|||
main_app = FastAPI() |
|||
main_app.mount("/mounted", sub_app) |
|||
|
|||
with TestClient(main_app) as client: |
|||
with pytest.raises(DependencyScopeError) as exc_info: |
|||
client.get("/mounted/sub") |
|||
assert "lifespan" in str(exc_info.value).lower() |
|||
Loading…
Reference in new issue