forked from fastapi/fastapi
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_router_events.py
More file actions
108 lines (85 loc) · 3.09 KB
/
test_router_events.py
File metadata and controls
108 lines (85 loc) · 3.09 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
from contextlib import asynccontextmanager
from typing import AsyncGenerator, Dict
import pytest
from fastapi import APIRouter, FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel
class State(BaseModel):
app_startup: bool = False
app_shutdown: bool = False
router_startup: bool = False
router_shutdown: bool = False
sub_router_startup: bool = False
sub_router_shutdown: bool = False
@pytest.fixture
def state() -> State:
return State()
def test_router_events(state: State) -> None:
app = FastAPI()
@app.get("/")
def main() -> Dict[str, str]:
return {"message": "Hello World"}
@app.on_event("startup")
def app_startup() -> None:
state.app_startup = True
@app.on_event("shutdown")
def app_shutdown() -> None:
state.app_shutdown = True
router = APIRouter()
@router.on_event("startup")
def router_startup() -> None:
state.router_startup = True
@router.on_event("shutdown")
def router_shutdown() -> None:
state.router_shutdown = True
sub_router = APIRouter()
@sub_router.on_event("startup")
def sub_router_startup() -> None:
state.sub_router_startup = True
@sub_router.on_event("shutdown")
def sub_router_shutdown() -> None:
state.sub_router_shutdown = True
router.include_router(sub_router)
app.include_router(router)
assert state.app_startup is False
assert state.router_startup is False
assert state.sub_router_startup is False
assert state.app_shutdown is False
assert state.router_shutdown is False
assert state.sub_router_shutdown is False
with TestClient(app) as client:
assert state.app_startup is True
assert state.router_startup is True
assert state.sub_router_startup is True
assert state.app_shutdown is False
assert state.router_shutdown is False
assert state.sub_router_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.router_startup is True
assert state.sub_router_startup is True
assert state.app_shutdown is True
assert state.router_shutdown is True
assert state.sub_router_shutdown is True
def test_app_lifespan_state(state: State) -> None:
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
state.app_startup = True
yield
state.app_shutdown = True
app = FastAPI(lifespan=lifespan)
@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