Coverage for tests / test_router_events.py: 100%

1from collections.abc import AsyncGenerator 4 ctx1mnop

2from contextlib import asynccontextmanager 4 ctx1mnop

3 

4import pytest 4 ctx1mnop

5from fastapi import APIRouter, FastAPI, Request 4 ctx1mnop

6from fastapi.testclient import TestClient 4 ctx1mnop

7from pydantic import BaseModel 4 ctx1mnop

8 

9 

10class State(BaseModel): 4 ctx1mnop

11 app_startup: bool = False 4 ctx1mnop

12 app_shutdown: bool = False 4 ctx1mnop

13 router_startup: bool = False 4 ctx1mnop

14 router_shutdown: bool = False 4 ctx1mnop

15 sub_router_startup: bool = False 4 ctx1mnop

16 sub_router_shutdown: bool = False 4 ctx1mnop

17 

18 

19@pytest.fixture 4 ctx1mnop

20def state() -> State: 4 ctx1mnop

21 return State() 23 ctx1STUVWXYZ0123456789!#$%'

22 

23 

24@pytest.mark.filterwarnings( 4 ctx1mnop

25 r"ignore:\s*on_event is deprecated, use lifespan event handlers instead.*:DeprecationWarning" 

26) 

27def test_router_events(state: State) -> None: 4 ctx1mnop

28 app = FastAPI() 4 ctx1abcd

29 

30 @app.get("/") 4 ctx1abcd

31 def main() -> dict[str, str]: 4 ctx1abcd

32 return {"message": "Hello World"} 4 ctx1abcd

33 

34 @app.on_event("startup") 4 ctx1abcd

35 def app_startup() -> None: 4 ctx1abcd

36 state.app_startup = True 4 ctx1abcd

37 

38 @app.on_event("shutdown") 4 ctx1abcd

39 def app_shutdown() -> None: 4 ctx1abcd

40 state.app_shutdown = True 4 ctx1abcd

41 

42 router = APIRouter() 4 ctx1abcd

43 

44 @router.on_event("startup") 4 ctx1abcd

45 def router_startup() -> None: 4 ctx1abcd

46 state.router_startup = True 4 ctx1abcd

47 

48 @router.on_event("shutdown") 4 ctx1abcd

49 def router_shutdown() -> None: 4 ctx1abcd

50 state.router_shutdown = True 4 ctx1abcd

51 

52 sub_router = APIRouter() 4 ctx1abcd

53 

54 @sub_router.on_event("startup") 4 ctx1abcd

55 def sub_router_startup() -> None: 4 ctx1abcd

56 state.sub_router_startup = True 4 ctx1abcd

57 

58 @sub_router.on_event("shutdown") 4 ctx1abcd

59 def sub_router_shutdown() -> None: 4 ctx1abcd

60 state.sub_router_shutdown = True 4 ctx1abcd

61 

62 router.include_router(sub_router) 4 ctx1abcd

63 app.include_router(router) 4 ctx1abcd

64 

65 assert state.app_startup is False 4 ctx1abcd

66 assert state.router_startup is False 4 ctx1abcd

67 assert state.sub_router_startup is False 4 ctx1abcd

68 assert state.app_shutdown is False 4 ctx1abcd

69 assert state.router_shutdown is False 4 ctx1abcd

70 assert state.sub_router_shutdown is False 4 ctx1abcd

71 with TestClient(app) as client: 4 ctx1abcd

72 assert state.app_startup is True 4 ctx1abcd

73 assert state.router_startup is True 4 ctx1abcd

74 assert state.sub_router_startup is True 4 ctx1abcd

75 assert state.app_shutdown is False 4 ctx1abcd

76 assert state.router_shutdown is False 4 ctx1abcd

77 assert state.sub_router_shutdown is False 4 ctx1abcd

78 response = client.get("/") 4 ctx1abcd

79 assert response.status_code == 200, response.text 4 ctx1abcd

80 assert response.json() == {"message": "Hello World"} 4 ctx1abcd

81 assert state.app_startup is True 4 ctx1abcd

82 assert state.router_startup is True 4 ctx1abcd

83 assert state.sub_router_startup is True 4 ctx1abcd

84 assert state.app_shutdown is True 4 ctx1abcd

85 assert state.router_shutdown is True 4 ctx1abcd

86 assert state.sub_router_shutdown is True 4 ctx1abcd

87 

88 

89def test_app_lifespan_state(state: State) -> None: 4 ctx1mnop

90 @asynccontextmanager 4 ctx1qrst

91 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 4 ctx1qrst

92 state.app_startup = True 4 ctx1qrst

93 yield 4 ctx1qrst

94 state.app_shutdown = True 4 ctx1qrst

95 

96 app = FastAPI(lifespan=lifespan) 4 ctx1qrst

97 

98 @app.get("/") 4 ctx1qrst

99 def main() -> dict[str, str]: 4 ctx1qrst

100 return {"message": "Hello World"} 4 ctx1qrst

101 

102 assert state.app_startup is False 4 ctx1qrst

103 assert state.app_shutdown is False 4 ctx1qrst

104 with TestClient(app) as client: 4 ctx1qrst

105 assert state.app_startup is True 4 ctx1qrst

106 assert state.app_shutdown is False 4 ctx1qrst

107 response = client.get("/") 4 ctx1qrst

108 assert response.status_code == 200, response.text 4 ctx1qrst

109 assert response.json() == {"message": "Hello World"} 4 ctx1qrst

110 assert state.app_startup is True 4 ctx1qrst

111 assert state.app_shutdown is True 4 ctx1qrst

112 

113 

114def test_router_nested_lifespan_state(state: State) -> None: 4 ctx1mnop

115 @asynccontextmanager 4 ctx1efgh

116 async def lifespan(app: FastAPI) -> AsyncGenerator[dict[str, bool], None]: 4 ctx1efgh

117 state.app_startup = True 4 ctx1efgh

118 yield {"app": True} 4 ctx1efgh

119 state.app_shutdown = True 4 ctx1efgh

120 

121 @asynccontextmanager 4 ctx1efgh

122 async def router_lifespan(app: FastAPI) -> AsyncGenerator[dict[str, bool], None]: 4 ctx1efgh

123 state.router_startup = True 4 ctx1efgh

124 yield {"router": True} 4 ctx1efgh

125 state.router_shutdown = True 4 ctx1efgh

126 

127 @asynccontextmanager 4 ctx1efgh

128 async def subrouter_lifespan(app: FastAPI) -> AsyncGenerator[dict[str, bool], None]: 4 ctx1efgh

129 state.sub_router_startup = True 4 ctx1efgh

130 yield {"sub_router": True} 4 ctx1efgh

131 state.sub_router_shutdown = True 4 ctx1efgh

132 

133 sub_router = APIRouter(lifespan=subrouter_lifespan) 4 ctx1efgh

134 

135 router = APIRouter(lifespan=router_lifespan) 4 ctx1efgh

136 router.include_router(sub_router) 4 ctx1efgh

137 

138 app = FastAPI(lifespan=lifespan) 4 ctx1efgh

139 app.include_router(router) 4 ctx1efgh

140 

141 @app.get("/") 4 ctx1efgh

142 def main(request: Request) -> dict[str, str]: 4 ctx1efgh

143 assert request.state.app 4 ctx1efgh

144 assert request.state.router 4 ctx1efgh

145 assert request.state.sub_router 4 ctx1efgh

146 return {"message": "Hello World"} 4 ctx1efgh

147 

148 assert state.app_startup is False 4 ctx1efgh

149 assert state.router_startup is False 4 ctx1efgh

150 assert state.sub_router_startup is False 4 ctx1efgh

151 assert state.app_shutdown is False 4 ctx1efgh

152 assert state.router_shutdown is False 4 ctx1efgh

153 assert state.sub_router_shutdown is False 4 ctx1efgh

154 

155 with TestClient(app) as client: 4 ctx1efgh

156 assert state.app_startup is True 4 ctx1efgh

157 assert state.router_startup is True 4 ctx1efgh

158 assert state.sub_router_startup is True 4 ctx1efgh

159 assert state.app_shutdown is False 4 ctx1efgh

160 assert state.router_shutdown is False 4 ctx1efgh

161 assert state.sub_router_shutdown is False 4 ctx1efgh

162 response = client.get("/") 4 ctx1efgh

163 assert response.status_code == 200, response.text 4 ctx1efgh

164 assert response.json() == {"message": "Hello World"} 4 ctx1efgh

165 

166 assert state.app_startup is True 4 ctx1efgh

167 assert state.router_startup is True 4 ctx1efgh

168 assert state.sub_router_startup is True 4 ctx1efgh

169 assert state.app_shutdown is True 4 ctx1efgh

170 assert state.router_shutdown is True 4 ctx1efgh

171 assert state.sub_router_shutdown is True 4 ctx1efgh

172 

173 

174def test_router_nested_lifespan_state_overriding_by_parent() -> None: 4 ctx1mnop

175 @asynccontextmanager 4 ctx1KLMN

176 async def lifespan( 4 ctx1KLMN

177 app: FastAPI, 

178 ) -> AsyncGenerator[dict[str, str | bool], None]: 

179 yield { 4 ctx1KLMN

180 "app_specific": True, 

181 "overridden": "app", 

182 } 

183 

184 @asynccontextmanager 4 ctx1KLMN

185 async def router_lifespan( 4 ctx1KLMN

186 app: FastAPI, 

187 ) -> AsyncGenerator[dict[str, str | bool], None]: 

188 yield { 4 ctx1KLMN

189 "router_specific": True, 

190 "overridden": "router", # should override parent 

191 } 

192 

193 router = APIRouter(lifespan=router_lifespan) 4 ctx1KLMN

194 app = FastAPI(lifespan=lifespan) 4 ctx1KLMN

195 app.include_router(router) 4 ctx1KLMN

196 

197 with TestClient(app) as client: 4 ctx1KLMN

198 assert client.app_state == { 4 ctx1KLMN

199 "app_specific": True, 

200 "router_specific": True, 

201 "overridden": "app", 

202 } 

203 

204 

205def test_merged_no_return_lifespans_return_none() -> None: 4 ctx1mnop

206 @asynccontextmanager 4 ctx1OPQR

207 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 4 ctx1OPQR

208 yield 4 ctx1OPQR

209 

210 @asynccontextmanager 4 ctx1OPQR

211 async def router_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 4 ctx1OPQR

212 yield 4 ctx1OPQR

213 

214 router = APIRouter(lifespan=router_lifespan) 4 ctx1OPQR

215 app = FastAPI(lifespan=lifespan) 4 ctx1OPQR

216 app.include_router(router) 4 ctx1OPQR

217 

218 with TestClient(app) as client: 4 ctx1OPQR

219 assert not client.app_state 4 ctx1OPQR

220 

221 

222def test_merged_mixed_state_lifespans() -> None: 4 ctx1mnop

223 @asynccontextmanager 4 ctx1CDEF

224 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 4 ctx1CDEF

225 yield 4 ctx1CDEF

226 

227 @asynccontextmanager 4 ctx1CDEF

228 async def router_lifespan(app: FastAPI) -> AsyncGenerator[dict[str, bool], None]: 4 ctx1CDEF

229 yield {"router": True} 4 ctx1CDEF

230 

231 @asynccontextmanager 4 ctx1CDEF

232 async def sub_router_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 4 ctx1CDEF

233 yield 4 ctx1CDEF

234 

235 sub_router = APIRouter(lifespan=sub_router_lifespan) 4 ctx1CDEF

236 router = APIRouter(lifespan=router_lifespan) 4 ctx1CDEF

237 app = FastAPI(lifespan=lifespan) 4 ctx1CDEF

238 router.include_router(sub_router) 4 ctx1CDEF

239 app.include_router(router) 4 ctx1CDEF

240 

241 with TestClient(app) as client: 4 ctx1CDEF

242 assert client.app_state == {"router": True} 4 ctx1CDEF

243 

244 

245@pytest.mark.filterwarnings( 4 ctx1mnop

246 r"ignore:\s*on_event is deprecated, use lifespan event handlers instead.*:DeprecationWarning" 

247) 

248def test_router_async_shutdown_handler(state: State) -> None: 4 ctx1mnop

249 """Test that async on_shutdown event handlers are called correctly, for coverage.""" 

250 app = FastAPI() 4 ctx1GHIJ

251 

252 @app.get("/") 4 ctx1GHIJ

253 def main() -> dict[str, str]: 4 ctx1GHIJ

254 return {"message": "Hello World"} 4 ctx1GHIJ

255 

256 @app.on_event("shutdown") 4 ctx1GHIJ

257 async def app_shutdown() -> None: 4 ctx1GHIJ

258 state.app_shutdown = True 4 ctx1GHIJ

259 

260 assert state.app_shutdown is False 4 ctx1GHIJ

261 with TestClient(app) as client: 4 ctx1GHIJ

262 assert state.app_shutdown is False 4 ctx1GHIJ

263 response = client.get("/") 4 ctx1GHIJ

264 assert response.status_code == 200, response.text 4 ctx1GHIJ

265 assert state.app_shutdown is True 4 ctx1GHIJ

266 

267 

268def test_router_sync_generator_lifespan(state: State) -> None: 4 ctx1mnop

269 """Test that a sync generator lifespan works via _wrap_gen_lifespan_context.""" 

270 from collections.abc import Generator 4 ctx1uvwx

271 

272 def lifespan(app: FastAPI) -> Generator[None, None, None]: 4 ctx1uvwx

273 state.app_startup = True 4 ctx1uvwx

274 yield 4 ctx1uvwx

275 state.app_shutdown = True 4 ctx1uvwx

276 

277 app = FastAPI(lifespan=lifespan) # type: ignore[arg-type] 4 ctx1uvwx

278 

279 @app.get("/") 4 ctx1uvwx

280 def main() -> dict[str, str]: 4 ctx1uvwx

281 return {"message": "Hello World"} 4 ctx1uvwx

282 

283 assert state.app_startup is False 4 ctx1uvwx

284 assert state.app_shutdown is False 4 ctx1uvwx

285 with TestClient(app) as client: 4 ctx1uvwx

286 assert state.app_startup is True 4 ctx1uvwx

287 assert state.app_shutdown is False 4 ctx1uvwx

288 response = client.get("/") 4 ctx1uvwx

289 assert response.status_code == 200, response.text 4 ctx1uvwx

290 assert response.json() == {"message": "Hello World"} 4 ctx1uvwx

291 assert state.app_startup is True 4 ctx1uvwx

292 assert state.app_shutdown is True 4 ctx1uvwx

293 

294 

295def test_router_async_generator_lifespan(state: State) -> None: 4 ctx1mnop

296 """Test that an async generator lifespan (not wrapped) works.""" 

297 

298 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 4 ctx1yzAB

299 state.app_startup = True 4 ctx1yzAB

300 yield 4 ctx1yzAB

301 state.app_shutdown = True 4 ctx1yzAB

302 

303 app = FastAPI(lifespan=lifespan) # type: ignore[arg-type] 4 ctx1yzAB

304 

305 @app.get("/") 4 ctx1yzAB

306 def main() -> dict[str, str]: 4 ctx1yzAB

307 return {"message": "Hello World"} 4 ctx1yzAB

308 

309 assert state.app_startup is False 4 ctx1yzAB

310 assert state.app_shutdown is False 4 ctx1yzAB

311 with TestClient(app) as client: 4 ctx1yzAB

312 assert state.app_startup is True 4 ctx1yzAB

313 assert state.app_shutdown is False 4 ctx1yzAB

314 response = client.get("/") 4 ctx1yzAB

315 assert response.status_code == 200, response.text 4 ctx1yzAB

316 assert response.json() == {"message": "Hello World"} 4 ctx1yzAB

317 assert state.app_startup is True 4 ctx1yzAB

318 assert state.app_shutdown is True 4 ctx1yzAB

319 

320 

321def test_startup_shutdown_handlers_as_parameters(state: State) -> None: 4 ctx1mnop

322 """Test that startup/shutdown handlers passed as parameters to FastAPI are called correctly.""" 

323 

324 def app_startup() -> None: 4 ctx1ijkl

325 state.app_startup = True 4 ctx1ijkl

326 

327 def app_shutdown() -> None: 4 ctx1ijkl

328 state.app_shutdown = True 4 ctx1ijkl

329 

330 app = FastAPI(on_startup=[app_startup], on_shutdown=[app_shutdown]) 4 ctx1ijkl

331 

332 @app.get("/") 4 ctx1ijkl

333 def main() -> dict[str, str]: 4 ctx1ijkl

334 return {"message": "Hello World"} 4 ctx1ijkl

335 

336 def router_startup() -> None: 4 ctx1ijkl

337 state.router_startup = True 4 ctx1ijkl

338 

339 def router_shutdown() -> None: 4 ctx1ijkl

340 state.router_shutdown = True 4 ctx1ijkl

341 

342 router = APIRouter(on_startup=[router_startup], on_shutdown=[router_shutdown]) 4 ctx1ijkl

343 

344 def sub_router_startup() -> None: 4 ctx1ijkl

345 state.sub_router_startup = True 4 ctx1ijkl

346 

347 def sub_router_shutdown() -> None: 4 ctx1ijkl

348 state.sub_router_shutdown = True 4 ctx1ijkl

349 

350 sub_router = APIRouter( 4 ctx1ijkl

351 on_startup=[sub_router_startup], on_shutdown=[sub_router_shutdown] 

352 ) 

353 

354 router.include_router(sub_router) 4 ctx1ijkl

355 app.include_router(router) 4 ctx1ijkl

356 

357 assert state.app_startup is False 4 ctx1ijkl

358 assert state.router_startup is False 4 ctx1ijkl

359 assert state.sub_router_startup is False 4 ctx1ijkl

360 assert state.app_shutdown is False 4 ctx1ijkl

361 assert state.router_shutdown is False 4 ctx1ijkl

362 assert state.sub_router_shutdown is False 4 ctx1ijkl

363 with TestClient(app) as client: 4 ctx1ijkl

364 assert state.app_startup is True 4 ctx1ijkl

365 assert state.router_startup is True 4 ctx1ijkl

366 assert state.sub_router_startup is True 4 ctx1ijkl

367 assert state.app_shutdown is False 4 ctx1ijkl

368 assert state.router_shutdown is False 4 ctx1ijkl

369 assert state.sub_router_shutdown is False 4 ctx1ijkl

370 response = client.get("/") 4 ctx1ijkl

371 assert response.status_code == 200, response.text 4 ctx1ijkl

372 assert response.json() == {"message": "Hello World"} 4 ctx1ijkl

373 assert state.app_startup is True 4 ctx1ijkl

374 assert state.router_startup is True 4 ctx1ijkl

375 assert state.sub_router_startup is True 4 ctx1ijkl

376 assert state.app_shutdown is True 4 ctx1ijkl

377 assert state.router_shutdown is True 4 ctx1ijkl

378 assert state.sub_router_shutdown is True 4 ctx1ijkl