Coverage for tests / test_router_events.py: 100%

1from collections.abc import AsyncGenerator 4 ctx1jkml

2from contextlib import asynccontextmanager 4 ctx1jkml

3 

4import pytest 4 ctx1jkml

5from fastapi import APIRouter, FastAPI, Request 4 ctx1jkml

6from fastapi.testclient import TestClient 4 ctx1jkml

7from pydantic import BaseModel 4 ctx1jkml

8 

9 

10class State(BaseModel): 4 ctx1jkml

11 app_startup: bool = False 4 ctx1jkml

12 app_shutdown: bool = False 4 ctx1jkml

13 router_startup: bool = False 4 ctx1jkml

14 router_shutdown: bool = False 4 ctx1jkml

15 sub_router_startup: bool = False 4 ctx1jkml

16 sub_router_shutdown: bool = False 4 ctx1jkml

17 

18 

19@pytest.fixture 4 ctx1jkml

20def state() -> State: 4 ctx1jkml

21 return State() 3 ctx1jkl

22 

23 

24@pytest.mark.filterwarnings( 4 ctx1jkml

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

26) 

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

28 app = FastAPI() 3 ctx1abc

29 

30 @app.get("/") 3 ctx1abc

31 def main() -> dict[str, str]: 3 ctx1abc

32 return {"message": "Hello World"} 3 ctx1abc

33 

34 @app.on_event("startup") 3 ctx1abc

35 def app_startup() -> None: 3 ctx1abc

36 state.app_startup = True 3 ctx1abc

37 

38 @app.on_event("shutdown") 3 ctx1abc

39 def app_shutdown() -> None: 3 ctx1abc

40 state.app_shutdown = True 3 ctx1abc

41 

42 router = APIRouter() 3 ctx1abc

43 

44 @router.on_event("startup") 3 ctx1abc

45 def router_startup() -> None: 3 ctx1abc

46 state.router_startup = True 3 ctx1abc

47 

48 @router.on_event("shutdown") 3 ctx1abc

49 def router_shutdown() -> None: 3 ctx1abc

50 state.router_shutdown = True 3 ctx1abc

51 

52 sub_router = APIRouter() 3 ctx1abc

53 

54 @sub_router.on_event("startup") 3 ctx1abc

55 def sub_router_startup() -> None: 3 ctx1abc

56 state.sub_router_startup = True 3 ctx1abc

57 

58 @sub_router.on_event("shutdown") 3 ctx1abc

59 def sub_router_shutdown() -> None: 3 ctx1abc

60 state.sub_router_shutdown = True 3 ctx1abc

61 

62 router.include_router(sub_router) 3 ctx1abc

63 app.include_router(router) 3 ctx1abc

64 

65 assert state.app_startup is False 3 ctx1abc

66 assert state.router_startup is False 3 ctx1abc

67 assert state.sub_router_startup is False 3 ctx1abc

68 assert state.app_shutdown is False 3 ctx1abc

69 assert state.router_shutdown is False 3 ctx1abc

70 assert state.sub_router_shutdown is False 3 ctx1abc

71 with TestClient(app) as client: 3 ctx1abc

72 assert state.app_startup is True 3 ctx1abc

73 assert state.router_startup is True 3 ctx1abc

74 assert state.sub_router_startup is True 3 ctx1abc

75 assert state.app_shutdown is False 3 ctx1abc

76 assert state.router_shutdown is False 3 ctx1abc

77 assert state.sub_router_shutdown is False 3 ctx1abc

78 response = client.get("/") 3 ctx1abc

79 assert response.status_code == 200, response.text 3 ctx1abc

80 assert response.json() == {"message": "Hello World"} 3 ctx1abc

81 assert state.app_startup is True 3 ctx1abc

82 assert state.router_startup is True 3 ctx1abc

83 assert state.sub_router_startup is True 3 ctx1abc

84 assert state.app_shutdown is True 3 ctx1abc

85 assert state.router_shutdown is True 3 ctx1abc

86 assert state.sub_router_shutdown is True 3 ctx1abc

87 

88 

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

90 @asynccontextmanager 3 ctx1nop

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

92 state.app_startup = True 3 ctx1nop

93 yield 3 ctx1nop

94 state.app_shutdown = True 3 ctx1nop

95 

96 app = FastAPI(lifespan=lifespan) 3 ctx1nop

97 

98 @app.get("/") 3 ctx1nop

99 def main() -> dict[str, str]: 3 ctx1nop

100 return {"message": "Hello World"} 3 ctx1nop

101 

102 assert state.app_startup is False 3 ctx1nop

103 assert state.app_shutdown is False 3 ctx1nop

104 with TestClient(app) as client: 3 ctx1nop

105 assert state.app_startup is True 3 ctx1nop

106 assert state.app_shutdown is False 3 ctx1nop

107 response = client.get("/") 3 ctx1nop

108 assert response.status_code == 200, response.text 3 ctx1nop

109 assert response.json() == {"message": "Hello World"} 3 ctx1nop

110 assert state.app_startup is True 3 ctx1nop

111 assert state.app_shutdown is True 3 ctx1nop

112 

113 

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

115 @asynccontextmanager 3 ctx1def

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

117 state.app_startup = True 3 ctx1def

118 yield {"app": True} 3 ctx1def

119 state.app_shutdown = True 3 ctx1def

120 

121 @asynccontextmanager 3 ctx1def

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

123 state.router_startup = True 3 ctx1def

124 yield {"router": True} 3 ctx1def

125 state.router_shutdown = True 3 ctx1def

126 

127 @asynccontextmanager 3 ctx1def

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

129 state.sub_router_startup = True 3 ctx1def

130 yield {"sub_router": True} 3 ctx1def

131 state.sub_router_shutdown = True 3 ctx1def

132 

133 sub_router = APIRouter(lifespan=subrouter_lifespan) 3 ctx1def

134 

135 router = APIRouter(lifespan=router_lifespan) 3 ctx1def

136 router.include_router(sub_router) 3 ctx1def

137 

138 app = FastAPI(lifespan=lifespan) 3 ctx1def

139 app.include_router(router) 3 ctx1def

140 

141 @app.get("/") 3 ctx1def

142 def main(request: Request) -> dict[str, str]: 3 ctx1def

143 assert request.state.app 3 ctx1def

144 assert request.state.router 3 ctx1def

145 assert request.state.sub_router 3 ctx1def

146 return {"message": "Hello World"} 3 ctx1def

147 

148 assert state.app_startup is False 3 ctx1def

149 assert state.router_startup is False 3 ctx1def

150 assert state.sub_router_startup is False 3 ctx1def

151 assert state.app_shutdown is False 3 ctx1def

152 assert state.router_shutdown is False 3 ctx1def

153 assert state.sub_router_shutdown is False 3 ctx1def

154 

155 with TestClient(app) as client: 3 ctx1def

156 assert state.app_startup is True 3 ctx1def

157 assert state.router_startup is True 3 ctx1def

158 assert state.sub_router_startup is True 3 ctx1def

159 assert state.app_shutdown is False 3 ctx1def

160 assert state.router_shutdown is False 3 ctx1def

161 assert state.sub_router_shutdown is False 3 ctx1def

162 response = client.get("/") 3 ctx1def

163 assert response.status_code == 200, response.text 3 ctx1def

164 assert response.json() == {"message": "Hello World"} 3 ctx1def

165 

166 assert state.app_startup is True 3 ctx1def

167 assert state.router_startup is True 3 ctx1def

168 assert state.sub_router_startup is True 3 ctx1def

169 assert state.app_shutdown is True 3 ctx1def

170 assert state.router_shutdown is True 3 ctx1def

171 assert state.sub_router_shutdown is True 3 ctx1def

172 

173 

174def test_router_nested_lifespan_state_overriding_by_parent() -> None: 4 ctx1jkml

175 @asynccontextmanager 3 ctx1CDE

176 async def lifespan( 3 ctx1CDE

177 app: FastAPI, 

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

179 yield { 3 ctx1CDE

180 "app_specific": True, 

181 "overridden": "app", 

182 } 

183 

184 @asynccontextmanager 3 ctx1CDE

185 async def router_lifespan( 3 ctx1CDE

186 app: FastAPI, 

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

188 yield { 3 ctx1CDE

189 "router_specific": True, 

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

191 } 

192 

193 router = APIRouter(lifespan=router_lifespan) 3 ctx1CDE

194 app = FastAPI(lifespan=lifespan) 3 ctx1CDE

195 app.include_router(router) 3 ctx1CDE

196 

197 with TestClient(app) as client: 3 ctx1CDE

198 assert client.app_state == { 3 ctx1CDE

199 "app_specific": True, 

200 "router_specific": True, 

201 "overridden": "app", 

202 } 

203 

204 

205def test_merged_no_return_lifespans_return_none() -> None: 4 ctx1jkml

206 @asynccontextmanager 3 ctx1FGH

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

208 yield 3 ctx1FGH

209 

210 @asynccontextmanager 3 ctx1FGH

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

212 yield 3 ctx1FGH

213 

214 router = APIRouter(lifespan=router_lifespan) 3 ctx1FGH

215 app = FastAPI(lifespan=lifespan) 3 ctx1FGH

216 app.include_router(router) 3 ctx1FGH

217 

218 with TestClient(app) as client: 3 ctx1FGH

219 assert not client.app_state 3 ctx1FGH

220 

221 

222def test_merged_mixed_state_lifespans() -> None: 4 ctx1jkml

223 @asynccontextmanager 3 ctx1wxy

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

225 yield 3 ctx1wxy

226 

227 @asynccontextmanager 3 ctx1wxy

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

229 yield {"router": True} 3 ctx1wxy

230 

231 @asynccontextmanager 3 ctx1wxy

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

233 yield 3 ctx1wxy

234 

235 sub_router = APIRouter(lifespan=sub_router_lifespan) 3 ctx1wxy

236 router = APIRouter(lifespan=router_lifespan) 3 ctx1wxy

237 app = FastAPI(lifespan=lifespan) 3 ctx1wxy

238 router.include_router(sub_router) 3 ctx1wxy

239 app.include_router(router) 3 ctx1wxy

240 

241 with TestClient(app) as client: 3 ctx1wxy

242 assert client.app_state == {"router": True} 3 ctx1wxy

243 

244 

245@pytest.mark.filterwarnings( 4 ctx1jkml

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 ctx1jkml

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

250 app = FastAPI() 3 ctx1zAB

251 

252 @app.get("/") 3 ctx1zAB

253 def main() -> dict[str, str]: 3 ctx1zAB

254 return {"message": "Hello World"} 3 ctx1zAB

255 

256 @app.on_event("shutdown") 3 ctx1zAB

257 async def app_shutdown() -> None: 3 ctx1zAB

258 state.app_shutdown = True 3 ctx1zAB

259 

260 assert state.app_shutdown is False 3 ctx1zAB

261 with TestClient(app) as client: 3 ctx1zAB

262 assert state.app_shutdown is False 3 ctx1zAB

263 response = client.get("/") 3 ctx1zAB

264 assert response.status_code == 200, response.text 3 ctx1zAB

265 assert state.app_shutdown is True 3 ctx1zAB

266 

267 

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

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

270 from collections.abc import Generator 3 ctx1qrs

271 

272 def lifespan(app: FastAPI) -> Generator[None, None, None]: 3 ctx1qrs

273 state.app_startup = True 3 ctx1qrs

274 yield 3 ctx1qrs

275 state.app_shutdown = True 3 ctx1qrs

276 

277 app = FastAPI(lifespan=lifespan) # type: ignore[arg-type] 3 ctx1qrs

278 

279 @app.get("/") 3 ctx1qrs

280 def main() -> dict[str, str]: 3 ctx1qrs

281 return {"message": "Hello World"} 3 ctx1qrs

282 

283 assert state.app_startup is False 3 ctx1qrs

284 assert state.app_shutdown is False 3 ctx1qrs

285 with TestClient(app) as client: 3 ctx1qrs

286 assert state.app_startup is True 3 ctx1qrs

287 assert state.app_shutdown is False 3 ctx1qrs

288 response = client.get("/") 3 ctx1qrs

289 assert response.status_code == 200, response.text 3 ctx1qrs

290 assert response.json() == {"message": "Hello World"} 3 ctx1qrs

291 assert state.app_startup is True 3 ctx1qrs

292 assert state.app_shutdown is True 3 ctx1qrs

293 

294 

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

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

297 

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

299 state.app_startup = True 3 ctx1tuv

300 yield 3 ctx1tuv

301 state.app_shutdown = True 3 ctx1tuv

302 

303 app = FastAPI(lifespan=lifespan) # type: ignore[arg-type] 3 ctx1tuv

304 

305 @app.get("/") 3 ctx1tuv

306 def main() -> dict[str, str]: 3 ctx1tuv

307 return {"message": "Hello World"} 3 ctx1tuv

308 

309 assert state.app_startup is False 3 ctx1tuv

310 assert state.app_shutdown is False 3 ctx1tuv

311 with TestClient(app) as client: 3 ctx1tuv

312 assert state.app_startup is True 3 ctx1tuv

313 assert state.app_shutdown is False 3 ctx1tuv

314 response = client.get("/") 3 ctx1tuv

315 assert response.status_code == 200, response.text 3 ctx1tuv

316 assert response.json() == {"message": "Hello World"} 3 ctx1tuv

317 assert state.app_startup is True 3 ctx1tuv

318 assert state.app_shutdown is True 3 ctx1tuv

319 

320 

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

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

323 

324 def app_startup() -> None: 3 ctx1ghi

325 state.app_startup = True 3 ctx1ghi

326 

327 def app_shutdown() -> None: 3 ctx1ghi

328 state.app_shutdown = True 3 ctx1ghi

329 

330 app = FastAPI(on_startup=[app_startup], on_shutdown=[app_shutdown]) 3 ctx1ghi

331 

332 @app.get("/") 3 ctx1ghi

333 def main() -> dict[str, str]: 3 ctx1ghi

334 return {"message": "Hello World"} 3 ctx1ghi

335 

336 def router_startup() -> None: 3 ctx1ghi

337 state.router_startup = True 3 ctx1ghi

338 

339 def router_shutdown() -> None: 3 ctx1ghi

340 state.router_shutdown = True 3 ctx1ghi

341 

342 router = APIRouter(on_startup=[router_startup], on_shutdown=[router_shutdown]) 3 ctx1ghi

343 

344 def sub_router_startup() -> None: 3 ctx1ghi

345 state.sub_router_startup = True 3 ctx1ghi

346 

347 def sub_router_shutdown() -> None: 3 ctx1ghi

348 state.sub_router_shutdown = True 3 ctx1ghi

349 

350 sub_router = APIRouter( 3 ctx1ghi

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

352 ) 

353 

354 router.include_router(sub_router) 3 ctx1ghi

355 app.include_router(router) 3 ctx1ghi

356 

357 assert state.app_startup is False 3 ctx1ghi

358 assert state.router_startup is False 3 ctx1ghi

359 assert state.sub_router_startup is False 3 ctx1ghi

360 assert state.app_shutdown is False 3 ctx1ghi

361 assert state.router_shutdown is False 3 ctx1ghi

362 assert state.sub_router_shutdown is False 3 ctx1ghi

363 with TestClient(app) as client: 3 ctx1ghi

364 assert state.app_startup is True 3 ctx1ghi

365 assert state.router_startup is True 3 ctx1ghi

366 assert state.sub_router_startup is True 3 ctx1ghi

367 assert state.app_shutdown is False 3 ctx1ghi

368 assert state.router_shutdown is False 3 ctx1ghi

369 assert state.sub_router_shutdown is False 3 ctx1ghi

370 response = client.get("/") 3 ctx1ghi

371 assert response.status_code == 200, response.text 3 ctx1ghi

372 assert response.json() == {"message": "Hello World"} 3 ctx1ghi

373 assert state.app_startup is True 3 ctx1ghi

374 assert state.router_startup is True 3 ctx1ghi

375 assert state.sub_router_startup is True 3 ctx1ghi

376 assert state.app_shutdown is True 3 ctx1ghi

377 assert state.router_shutdown is True 3 ctx1ghi

378 assert state.sub_router_shutdown is True 3 ctx1ghi