Browse Source

Add basic tests for using context managers as dependencies

pull/10353/head
Alexander Pushkov 2 years ago
committed by Alexander Pushkov
parent
commit
affb3ca017
  1. 43
      tests/test_dependency_contextmanager.py

43
tests/test_dependency_contextmanager.py

@ -1,4 +1,5 @@
from typing import Dict from typing import Dict
from contextlib import asynccontextmanager, contextmanager
import pytest import pytest
from fastapi import BackgroundTasks, Depends, FastAPI from fastapi import BackgroundTasks, Depends, FastAPI
@ -8,6 +9,8 @@ app = FastAPI()
state = { state = {
"/async": "asyncgen not started", "/async": "asyncgen not started",
"/sync": "generator not started", "/sync": "generator not started",
"/async_ctxmgr": "asyncgen_ctxmgr not started",
"/sync_ctxmgr": "generator_ctxmgr not started",
"/async_raise": "asyncgen raise not started", "/async_raise": "asyncgen raise not started",
"/sync_raise": "generator raise not started", "/sync_raise": "generator raise not started",
"context_a": "not started a", "context_a": "not started a",
@ -47,6 +50,20 @@ def generator_state(state: Dict[str, str] = Depends(get_state)):
state["/sync"] = "generator completed" state["/sync"] = "generator completed"
@asynccontextmanager
async def asyncgen_state_ctxmgr(state: Dict[str, str] = Depends(get_state)):
state["/async_ctxmgr"] = "asyncgen_ctxmgr started"
yield state["/async_ctxmgr"]
state["/async_ctxmgr"] = "asyncgen_ctxmgr completed"
@contextmanager
def generator_state_ctxmgr(state: Dict[str, str] = Depends(get_state)):
state["/sync_ctxmgr"] = "generator_ctxmgr started"
yield state["/sync_ctxmgr"]
state["/sync_ctxmgr"] = "generator_ctxmgr completed"
async def asyncgen_state_try(state: Dict[str, str] = Depends(get_state)): async def asyncgen_state_try(state: Dict[str, str] = Depends(get_state)):
state["/async_raise"] = "asyncgen raise started" state["/async_raise"] = "asyncgen raise started"
try: try:
@ -93,6 +110,16 @@ async def get_sync(state: str = Depends(generator_state)):
return state return state
@app.get("/async_ctxmgr")
async def get_async(state: str = Depends(asyncgen_state_ctxmgr)):
return state
@app.get("/sync_ctxmgr")
async def get_sync(state: str = Depends(generator_state_ctxmgr)):
return state
@app.get("/async_raise") @app.get("/async_raise")
async def get_async_raise(state: str = Depends(asyncgen_state_try)): async def get_async_raise(state: str = Depends(asyncgen_state_try)):
assert state == "asyncgen raise started" assert state == "asyncgen raise started"
@ -219,6 +246,22 @@ def test_sync_state():
assert state["/sync"] == "generator completed" assert state["/sync"] == "generator completed"
def test_async_ctxmgr_state():
assert state["/async_ctxmgr"] == "asyncgen_ctxmgr not started"
response = client.get("/async_ctxmgr")
assert response.status_code == 200, response.text
assert response.json() == "asyncgen_ctxmgr started"
assert state["/async_ctxmgr"] == "asyncgen_ctxmgr completed"
def test_sync_ctxmgr_state():
assert state["/sync_ctxmgr"] == "generator_ctxmgr not started"
response = client.get("/sync_ctxmgr")
assert response.status_code == 200, response.text
assert response.json() == "generator_ctxmgr started"
assert state["/sync_ctxmgr"] == "generator_ctxmgr completed"
def test_async_raise_other(): def test_async_raise_other():
assert state["/async_raise"] == "asyncgen raise not started" assert state["/async_raise"] == "asyncgen raise not started"
with pytest.raises(OtherDependencyError): with pytest.raises(OtherDependencyError):

Loading…
Cancel
Save