Browse Source

refactor: simplified tests for top level dependencies. Moved tests into test_dependency_yield_scope

pull/14301/head
luzzodev 9 months ago
parent
commit
a7878821e0
  1. 64
      tests/test_dependency_yield_scope.py
  2. 78
      tests/test_dependency_yield_scope_api_router.py

64
tests/test_dependency_yield_scope.py

@ -2,7 +2,7 @@ import json
from typing import Any, Tuple
import pytest
from fastapi import Depends, FastAPI
from fastapi import APIRouter, Depends, FastAPI, HTTPException
from fastapi.exceptions import FastAPIError
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient
@ -20,6 +20,11 @@ def dep_session() -> Any:
s.open = False
def raise_after_yield() -> Any:
yield
raise HTTPException(status_code=503, detail="Exception after yield")
SessionFuncDep = Annotated[Session, Depends(dep_session, scope="function")]
SessionRequestDep = Annotated[Session, Depends(dep_session, scope="request")]
SessionDefaultDep = Annotated[Session, Depends(dep_session)]
@ -64,6 +69,12 @@ RegularSessionsDep = Annotated[
]
app = FastAPI()
router = APIRouter()
@router.get("/")
def get_index():
return {"status": "ok"}
@app.get("/function-scope")
@ -124,6 +135,18 @@ def get_regular_function_scope(sessions: RegularSessionsDep) -> Any:
return StreamingResponse(iter_data())
app.include_router(
prefix="/router-scope-function",
router=router,
dependencies=[Depends(raise_after_yield, scope="function")],
)
app.include_router(
prefix="/router-scope-request",
router=router,
dependencies=[Depends(raise_after_yield, scope="request")],
)
client = TestClient(app)
@ -182,3 +205,42 @@ def test_regular_function_scope() -> None:
data = response.json()
assert data["named_session_open"] is True
assert data["session_open"] is False
def test_router_level_dep_scope_function() -> None:
response = client.get("/router-scope-function/")
assert response.status_code == 503
assert response.json() == {"detail": "Exception after yield"}
def test_router_level_dep_scope_request() -> None:
with TestClient(app, raise_server_exceptions=False) as client:
response = client.get("/router-scope-request/")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_app_level_dep_scope_function() -> None:
app = FastAPI(dependencies=[Depends(raise_after_yield, scope="function")])
@app.get("/app-scope-function")
def get_app_scope_function():
return {"status": "ok"}
with TestClient(app) as client:
response = client.get("/app-scope-function")
assert response.status_code == 503
assert response.json() == {"detail": "Exception after yield"}
def test_app_level_dep_scope_request() -> None:
app = FastAPI(dependencies=[Depends(raise_after_yield, scope="request")])
@app.get("/app-scope-request")
def get_app_scope_request():
return {"status": "ok"}
with TestClient(app, raise_server_exceptions=False) as client:
response = client.get("/app-scope-request")
assert response.status_code == 200
assert response.json() == {"status": "ok"}

78
tests/test_dependency_yield_scope_api_router.py

@ -1,78 +0,0 @@
from typing import Any
from fastapi import APIRouter, Depends, FastAPI, HTTPException
from fastapi.testclient import TestClient
def broken_dependency() -> Any:
yield "s"
raise RuntimeError("Caught exception after yield")
def caught_dependency() -> Any:
yield "s"
raise HTTPException(status_code=503, detail="Something went wrong")
app = FastAPI()
r_caught_function = APIRouter(
prefix="/scope/function/caught",
dependencies=[Depends(caught_dependency, scope="function")],
)
r_broken_function = APIRouter(
prefix="/scope/function/broken",
dependencies=[Depends(broken_dependency, scope="function")],
)
r_caught_request = APIRouter(
prefix="/scope/request/caught",
dependencies=[Depends(caught_dependency, scope="request")],
)
r_broken_request = APIRouter(
prefix="/scope/request/broken",
dependencies=[Depends(broken_dependency, scope="request")],
)
@r_caught_function.get("/orders")
@r_broken_function.get("/orders")
@r_caught_request.get("/orders")
@r_broken_request.get("/orders")
def get_orders() -> Any:
return {"message": "Maybe not here!"}
app.include_router(r_caught_function)
app.include_router(r_broken_function)
app.include_router(r_caught_request)
app.include_router(r_broken_request)
def test_caught_dependency_scope_function():
with TestClient(app, raise_server_exceptions=False) as client:
response = client.get("/scope/function/caught/orders")
assert response.status_code == 503
assert response.json() == {"detail": "Something went wrong"}
def test_broken_dependency_scope_function():
with TestClient(app, raise_server_exceptions=False) as client:
response = client.get("/scope/function/broken/orders")
assert response.status_code == 500
assert response.content == b"Internal Server Error"
def test_caught_dependency_scope_request():
with TestClient(app, raise_server_exceptions=False) as client:
response = client.get("/scope/request/caught/orders")
assert response.status_code == 200
assert response.json() == {"message": "Maybe not here!"}
def test_broken_dependency_scope_request():
with TestClient(app, raise_server_exceptions=False) as client:
response = client.get("/scope/request/broken/orders")
assert response.status_code == 200
assert response.json() == {"message": "Maybe not here!"}
Loading…
Cancel
Save