Coverage for tests / test_router_events.py: 100%

1from collections.abc import AsyncGenerator 4 ctx1jkml

2from contextlib import asynccontextmanager 4 ctx1jkml

3from typing import Union 4 ctx1jkml

4 

5import pytest 4 ctx1jkml

6from fastapi import APIRouter, FastAPI, Request 4 ctx1jkml

7from fastapi.testclient import TestClient 4 ctx1jkml

8from pydantic import BaseModel 4 ctx1jkml

9 

10 

11class State(BaseModel): 4 ctx1jkml

12 app_startup: bool = False 4 ctx1jkml

13 app_shutdown: bool = False 4 ctx1jkml

14 router_startup: bool = False 4 ctx1jkml

15 router_shutdown: bool = False 4 ctx1jkml

16 sub_router_startup: bool = False 4 ctx1jkml

17 sub_router_shutdown: bool = False 4 ctx1jkml

18 

19 

20@pytest.fixture 4 ctx1jkml

21def state() -> State: 4 ctx1jkml

22 return State() 3 ctx1jkl

23 

24 

25@pytest.mark.filterwarnings( 4 ctx1jkml

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

27) 

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

29 app = FastAPI() 3 ctx1abc

30 

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

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

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

34 

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

36 def app_startup() -> None: 3 ctx1abc

37 state.app_startup = True 3 ctx1abc

38 

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

40 def app_shutdown() -> None: 3 ctx1abc

41 state.app_shutdown = True 3 ctx1abc

42 

43 router = APIRouter() 3 ctx1abc

44 

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

46 def router_startup() -> None: 3 ctx1abc

47 state.router_startup = True 3 ctx1abc

48 

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

50 def router_shutdown() -> None: 3 ctx1abc

51 state.router_shutdown = True 3 ctx1abc

52 

53 sub_router = APIRouter() 3 ctx1abc

54 

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

56 def sub_router_startup() -> None: 3 ctx1abc

57 state.sub_router_startup = True 3 ctx1abc

58 

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

60 def sub_router_shutdown() -> None: 3 ctx1abc

61 state.sub_router_shutdown = True 3 ctx1abc

62 

63 router.include_router(sub_router) 3 ctx1abc

64 app.include_router(router) 3 ctx1abc

65 

66 assert state.app_startup is False 3 ctx1abc

67 assert state.router_startup is False 3 ctx1abc

68 assert state.sub_router_startup is False 3 ctx1abc

69 assert state.app_shutdown is False 3 ctx1abc

70 assert state.router_shutdown is False 3 ctx1abc

71 assert state.sub_router_shutdown is False 3 ctx1abc

72 with TestClient(app) as client: 3 ctx1abc

73 assert state.app_startup is True 3 ctx1abc

74 assert state.router_startup is True 3 ctx1abc

75 assert state.sub_router_startup is True 3 ctx1abc

76 assert state.app_shutdown is False 3 ctx1abc

77 assert state.router_shutdown is False 3 ctx1abc

78 assert state.sub_router_shutdown is False 3 ctx1abc

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

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

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

82 assert state.app_startup is True 3 ctx1abc

83 assert state.router_startup is True 3 ctx1abc

84 assert state.sub_router_startup is True 3 ctx1abc

85 assert state.app_shutdown is True 3 ctx1abc

86 assert state.router_shutdown is True 3 ctx1abc

87 assert state.sub_router_shutdown is True 3 ctx1abc

88 

89 

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

91 @asynccontextmanager 3 ctx1nop

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

93 state.app_startup = True 3 ctx1nop

94 yield 3 ctx1nop

95 state.app_shutdown = True 3 ctx1nop

96 

97 app = FastAPI(lifespan=lifespan) 3 ctx1nop

98 

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

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

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

102 

103 assert state.app_startup is False 3 ctx1nop

104 assert state.app_shutdown is False 3 ctx1nop

105 with TestClient(app) as client: 3 ctx1nop

106 assert state.app_startup is True 3 ctx1nop

107 assert state.app_shutdown is False 3 ctx1nop

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

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

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

111 assert state.app_startup is True 3 ctx1nop

112 assert state.app_shutdown is True 3 ctx1nop

113 

114 

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

116 @asynccontextmanager 3 ctx1def

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

118 state.app_startup = True 3 ctx1def

119 yield {"app": True} 3 ctx1def

120 state.app_shutdown = True 3 ctx1def

121 

122 @asynccontextmanager 3 ctx1def

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

124 state.router_startup = True 3 ctx1def

125 yield {"router": True} 3 ctx1def

126 state.router_shutdown = True 3 ctx1def

127 

128 @asynccontextmanager 3 ctx1def

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

130 state.sub_router_startup = True 3 ctx1def

131 yield {"sub_router": True} 3 ctx1def

132 state.sub_router_shutdown = True 3 ctx1def

133 

134 sub_router = APIRouter(lifespan=subrouter_lifespan) 3 ctx1def

135 

136 router = APIRouter(lifespan=router_lifespan) 3 ctx1def

137 router.include_router(sub_router) 3 ctx1def

138 

139 app = FastAPI(lifespan=lifespan) 3 ctx1def

140 app.include_router(router) 3 ctx1def

141 

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

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

144 assert request.state.app 3 ctx1def

145 assert request.state.router 3 ctx1def

146 assert request.state.sub_router 3 ctx1def

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

148 

149 assert state.app_startup is False 3 ctx1def

150 assert state.router_startup is False 3 ctx1def

151 assert state.sub_router_startup is False 3 ctx1def

152 assert state.app_shutdown is False 3 ctx1def

153 assert state.router_shutdown is False 3 ctx1def

154 assert state.sub_router_shutdown is False 3 ctx1def

155 

156 with TestClient(app) as client: 3 ctx1def

157 assert state.app_startup is True 3 ctx1def

158 assert state.router_startup is True 3 ctx1def

159 assert state.sub_router_startup is True 3 ctx1def

160 assert state.app_shutdown is False 3 ctx1def

161 assert state.router_shutdown is False 3 ctx1def

162 assert state.sub_router_shutdown is False 3 ctx1def

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

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

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

166 

167 assert state.app_startup is True 3 ctx1def

168 assert state.router_startup is True 3 ctx1def

169 assert state.sub_router_startup is True 3 ctx1def

170 assert state.app_shutdown is True 3 ctx1def

171 assert state.router_shutdown is True 3 ctx1def

172 assert state.sub_router_shutdown is True 3 ctx1def

173 

174 

175def test_router_nested_lifespan_state_overriding_by_parent() -> None: 4 ctx1jkml

176 @asynccontextmanager 3 ctx1CDE

177 async def lifespan( 3 ctx1CDE

178 app: FastAPI, 

179 ) -> AsyncGenerator[dict[str, Union[str, bool]], None]: 

180 yield { 3 ctx1CDE

181 "app_specific": True, 

182 "overridden": "app", 

183 } 

184 

185 @asynccontextmanager 3 ctx1CDE

186 async def router_lifespan( 3 ctx1CDE

187 app: FastAPI, 

188 ) -> AsyncGenerator[dict[str, Union[str, bool]], None]: 

189 yield { 3 ctx1CDE

190 "router_specific": True, 

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

192 } 

193 

194 router = APIRouter(lifespan=router_lifespan) 3 ctx1CDE

195 app = FastAPI(lifespan=lifespan) 3 ctx1CDE

196 app.include_router(router) 3 ctx1CDE

197 

198 with TestClient(app) as client: 3 ctx1CDE

199 assert client.app_state == { 3 ctx1CDE

200 "app_specific": True, 

201 "router_specific": True, 

202 "overridden": "app", 

203 } 

204 

205 

206def test_merged_no_return_lifespans_return_none() -> None: 4 ctx1jkml

207 @asynccontextmanager 3 ctx1FGH

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

209 yield 3 ctx1FGH

210 

211 @asynccontextmanager 3 ctx1FGH

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

213 yield 3 ctx1FGH

214 

215 router = APIRouter(lifespan=router_lifespan) 3 ctx1FGH

216 app = FastAPI(lifespan=lifespan) 3 ctx1FGH

217 app.include_router(router) 3 ctx1FGH

218 

219 with TestClient(app) as client: 3 ctx1FGH

220 assert not client.app_state 3 ctx1FGH

221 

222 

223def test_merged_mixed_state_lifespans() -> None: 4 ctx1jkml

224 @asynccontextmanager 3 ctx1wxy

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

226 yield 3 ctx1wxy

227 

228 @asynccontextmanager 3 ctx1wxy

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

230 yield {"router": True} 3 ctx1wxy

231 

232 @asynccontextmanager 3 ctx1wxy

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

234 yield 3 ctx1wxy

235 

236 sub_router = APIRouter(lifespan=sub_router_lifespan) 3 ctx1wxy

237 router = APIRouter(lifespan=router_lifespan) 3 ctx1wxy

238 app = FastAPI(lifespan=lifespan) 3 ctx1wxy

239 router.include_router(sub_router) 3 ctx1wxy

240 app.include_router(router) 3 ctx1wxy

241 

242 with TestClient(app) as client: 3 ctx1wxy

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

244 

245 

246@pytest.mark.filterwarnings( 4 ctx1jkml

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

248) 

249def test_router_async_shutdown_handler(state: State) -> None: 4 ctx1jkml

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

251 app = FastAPI() 3 ctx1zAB

252 

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

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

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

256 

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

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

259 state.app_shutdown = True 3 ctx1zAB

260 

261 assert state.app_shutdown is False 3 ctx1zAB

262 with TestClient(app) as client: 3 ctx1zAB

263 assert state.app_shutdown is False 3 ctx1zAB

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

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

266 assert state.app_shutdown is True 3 ctx1zAB

267 

268 

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

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

271 from collections.abc import Generator 3 ctx1qrs

272 

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

274 state.app_startup = True 3 ctx1qrs

275 yield 3 ctx1qrs

276 state.app_shutdown = True 3 ctx1qrs

277 

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

279 

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

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

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

283 

284 assert state.app_startup is False 3 ctx1qrs

285 assert state.app_shutdown is False 3 ctx1qrs

286 with TestClient(app) as client: 3 ctx1qrs

287 assert state.app_startup is True 3 ctx1qrs

288 assert state.app_shutdown is False 3 ctx1qrs

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

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

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

292 assert state.app_startup is True 3 ctx1qrs

293 assert state.app_shutdown is True 3 ctx1qrs

294 

295 

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

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

298 

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

300 state.app_startup = True 3 ctx1tuv

301 yield 3 ctx1tuv

302 state.app_shutdown = True 3 ctx1tuv

303 

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

305 

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

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

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

309 

310 assert state.app_startup is False 3 ctx1tuv

311 assert state.app_shutdown is False 3 ctx1tuv

312 with TestClient(app) as client: 3 ctx1tuv

313 assert state.app_startup is True 3 ctx1tuv

314 assert state.app_shutdown is False 3 ctx1tuv

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

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

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

318 assert state.app_startup is True 3 ctx1tuv

319 assert state.app_shutdown is True 3 ctx1tuv

320 

321 

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

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

324 

325 def app_startup() -> None: 3 ctx1ghi

326 state.app_startup = True 3 ctx1ghi

327 

328 def app_shutdown() -> None: 3 ctx1ghi

329 state.app_shutdown = True 3 ctx1ghi

330 

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

332 

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

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

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

336 

337 def router_startup() -> None: 3 ctx1ghi

338 state.router_startup = True 3 ctx1ghi

339 

340 def router_shutdown() -> None: 3 ctx1ghi

341 state.router_shutdown = True 3 ctx1ghi

342 

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

344 

345 def sub_router_startup() -> None: 3 ctx1ghi

346 state.sub_router_startup = True 3 ctx1ghi

347 

348 def sub_router_shutdown() -> None: 3 ctx1ghi

349 state.sub_router_shutdown = True 3 ctx1ghi

350 

351 sub_router = APIRouter( 3 ctx1ghi

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

353 ) 

354 

355 router.include_router(sub_router) 3 ctx1ghi

356 app.include_router(router) 3 ctx1ghi

357 

358 assert state.app_startup is False 3 ctx1ghi

359 assert state.router_startup is False 3 ctx1ghi

360 assert state.sub_router_startup is False 3 ctx1ghi

361 assert state.app_shutdown is False 3 ctx1ghi

362 assert state.router_shutdown is False 3 ctx1ghi

363 assert state.sub_router_shutdown is False 3 ctx1ghi

364 with TestClient(app) as client: 3 ctx1ghi

365 assert state.app_startup is True 3 ctx1ghi

366 assert state.router_startup is True 3 ctx1ghi

367 assert state.sub_router_startup is True 3 ctx1ghi

368 assert state.app_shutdown is False 3 ctx1ghi

369 assert state.router_shutdown is False 3 ctx1ghi

370 assert state.sub_router_shutdown is False 3 ctx1ghi

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

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

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

374 assert state.app_startup is True 3 ctx1ghi

375 assert state.router_startup is True 3 ctx1ghi

376 assert state.sub_router_startup is True 3 ctx1ghi

377 assert state.app_shutdown is True 3 ctx1ghi

378 assert state.router_shutdown is True 3 ctx1ghi

379 assert state.sub_router_shutdown is True 3 ctx1ghi