Coverage for tests / test_dependency_contextmanager.py: 100%

1import json 4 ctx1abcd

2 

3import pytest 4 ctx1abcd

4from fastapi import BackgroundTasks, Depends, FastAPI 4 ctx1abcd

5from fastapi.responses import StreamingResponse 4 ctx1abcd

6from fastapi.testclient import TestClient 4 ctx1abcd

7 

8app = FastAPI() 4 ctx1abcd

9state = { 4 ctx1abcd

10 "/async": "asyncgen not started", 

11 "/sync": "generator not started", 

12 "/async_raise": "asyncgen raise not started", 

13 "/sync_raise": "generator raise not started", 

14 "context_a": "not started a", 

15 "context_b": "not started b", 

16 "bg": "not set", 

17 "sync_bg": "not set", 

18} 

19 

20errors = [] 4 ctx1abcd

21 

22 

23async def get_state(): 4 ctx1abcd

24 return state 66 ctx1UIq0ekr1Js$hltVKu23Lv%WMw4fmx5Ny'inzXOA67PB(YQC8goD9RE)jpFZSG!#TH*

25 

26 

27class AsyncDependencyError(Exception): 4 ctx1abcd

28 pass 4 ctx1abcd

29 

30 

31class SyncDependencyError(Exception): 4 ctx1abcd

32 pass 4 ctx1abcd

33 

34 

35class OtherDependencyError(Exception): 4 ctx1abcd

36 pass 4 ctx1abcd

37 

38 

39async def asyncgen_state(state: dict[str, str] = Depends(get_state)): 4 ctx1abcd

40 state["/async"] = "asyncgen started" 6 ctx10$4'8)

41 yield state["/async"] 6 ctx10$4'8)

42 state["/async"] = "asyncgen completed" 6 ctx10$4'8)

43 

44 

45def generator_state(state: dict[str, str] = Depends(get_state)): 4 ctx1abcd

46 state["/sync"] = "generator started" 6 ctx12%6(!*

47 yield state["/sync"] 6 ctx12%6(!*

48 state["/sync"] = "generator completed" 6 ctx12%6(!*

49 

50 

51async def asyncgen_state_try(state: dict[str, str] = Depends(get_state)): 4 ctx1abcd

52 state["/async_raise"] = "asyncgen raise started" 18 ctx1UIq1JsWMw5NyYQC9RE

53 try: 18 ctx1UIq1JsWMw5NyYQC9RE

54 yield state["/async_raise"] 18 ctx1UIq1JsWMw5NyYQC9RE

55 except AsyncDependencyError: 18 ctx1UIq1JsWMw5NyYQC9RE

56 errors.append("/async_raise") 12 ctx1IqJsMwNyQCRE

57 raise 12 ctx1IqJsMwNyQCRE

58 finally: 

59 state["/async_raise"] = "asyncgen raise finalized" 18 ctx1UIq1JsWMw5NyYQC9RE

60 

61 

62def generator_state_try(state: dict[str, str] = Depends(get_state)): 4 ctx1abcd

63 state["/sync_raise"] = "generator raise started" 18 ctx1VKu3LvXOA7PBZSG#TH

64 try: 18 ctx1VKu3LvXOA7PBZSG#TH

65 yield state["/sync_raise"] 18 ctx1VKu3LvXOA7PBZSG#TH

66 except SyncDependencyError: 18 ctx1VKu3LvXOA7PBZSG#TH

67 errors.append("/sync_raise") 12 ctx1KuLvOAPBSGTH

68 raise 12 ctx1KuLvOAPBSGTH

69 finally: 

70 state["/sync_raise"] = "generator raise finalized" 18 ctx1VKu3LvXOA7PBZSG#TH

71 

72 

73async def context_a(state: dict = Depends(get_state)): 4 ctx1abcd

74 state["context_a"] = "started a" 18 ctx1ekrhltfmxinzgoDjpF

75 try: 18 ctx1ekrhltfmxinzgoDjpF

76 yield state 18 ctx1ekrhltfmxinzgoDjpF

77 finally: 

78 state["context_a"] = "finished a" 18 ctx1ekrhltfmxinzgoDjpF

79 

80 

81async def context_b(state: dict = Depends(context_a)): 4 ctx1abcd

82 state["context_b"] = "started b" 18 ctx1ekrhltfmxinzgoDjpF

83 try: 18 ctx1ekrhltfmxinzgoDjpF

84 yield state 18 ctx1ekrhltfmxinzgoDjpF

85 finally: 

86 state["context_b"] = f"finished b with a: {state['context_a']}" 18 ctx1ekrhltfmxinzgoDjpF

87 

88 

89@app.get("/async") 4 ctx1abcd

90async def get_async(state: str = Depends(asyncgen_state)): 4 ctx1abcd

91 return state 3 ctx1048

92 

93 

94@app.get("/sync") 4 ctx1abcd

95async def get_sync(state: str = Depends(generator_state)): 4 ctx1abcd

96 return state 3 ctx126!

97 

98 

99@app.get("/async_raise") 4 ctx1abcd

100async def get_async_raise(state: str = Depends(asyncgen_state_try)): 4 ctx1abcd

101 assert state == "asyncgen raise started" 6 ctx1IqMwQC

102 raise AsyncDependencyError() 6 ctx1IqMwQC

103 

104 

105@app.get("/sync_raise") 4 ctx1abcd

106async def get_sync_raise(state: str = Depends(generator_state_try)): 4 ctx1abcd

107 assert state == "generator raise started" 6 ctx1KuOASG

108 raise SyncDependencyError() 6 ctx1KuOASG

109 

110 

111@app.get("/async_raise_other") 4 ctx1abcd

112async def get_async_raise_other(state: str = Depends(asyncgen_state_try)): 4 ctx1abcd

113 assert state == "asyncgen raise started" 3 ctx1UWY

114 raise OtherDependencyError() 3 ctx1UWY

115 

116 

117@app.get("/sync_raise_other") 4 ctx1abcd

118async def get_sync_raise_other(state: str = Depends(generator_state_try)): 4 ctx1abcd

119 assert state == "generator raise started" 3 ctx1VXZ

120 raise OtherDependencyError() 3 ctx1VXZ

121 

122 

123@app.get("/context_b") 4 ctx1abcd

124async def get_context_b(state: dict = Depends(context_b)): 4 ctx1abcd

125 return state 3 ctx1kmo

126 

127 

128@app.get("/context_b_raise") 4 ctx1abcd

129async def get_context_b_raise(state: dict = Depends(context_b)): 4 ctx1abcd

130 assert state["context_b"] == "started b" 3 ctx1rxD

131 assert state["context_a"] == "started a" 3 ctx1rxD

132 raise OtherDependencyError() 3 ctx1rxD

133 

134 

135@app.get("/context_b_bg") 4 ctx1abcd

136async def get_context_b_bg(tasks: BackgroundTasks, state: dict = Depends(context_b)): 4 ctx1abcd

137 async def bg(state: dict): 3 ctx1efg

138 state["bg"] = f"bg set - b: {state['context_b']} - a: {state['context_a']}" 3 ctx1efg

139 

140 tasks.add_task(bg, state) 3 ctx1efg

141 return state 3 ctx1efg

142 

143 

144# Sync versions 

145 

146 

147@app.get("/sync_async") 4 ctx1abcd

148def get_sync_async(state: str = Depends(asyncgen_state)): 4 ctx1abcd

149 return state 3 ctx1$')

150 

151 

152@app.get("/sync_sync") 4 ctx1abcd

153def get_sync_sync(state: str = Depends(generator_state)): 4 ctx1abcd

154 return state 3 ctx1%(*

155 

156 

157@app.get("/sync_async_raise") 4 ctx1abcd

158def get_sync_async_raise(state: str = Depends(asyncgen_state_try)): 4 ctx1abcd

159 assert state == "asyncgen raise started" 6 ctx1JsNyRE

160 raise AsyncDependencyError() 6 ctx1JsNyRE

161 

162 

163@app.get("/sync_sync_raise") 4 ctx1abcd

164def get_sync_sync_raise(state: str = Depends(generator_state_try)): 4 ctx1abcd

165 assert state == "generator raise started" 6 ctx1LvPBTH

166 raise SyncDependencyError() 6 ctx1LvPBTH

167 

168 

169@app.get("/sync_async_raise_other") 4 ctx1abcd

170def get_sync_async_raise_other(state: str = Depends(asyncgen_state_try)): 4 ctx1abcd

171 assert state == "asyncgen raise started" 3 ctx1159

172 raise OtherDependencyError() 3 ctx1159

173 

174 

175@app.get("/sync_sync_raise_other") 4 ctx1abcd

176def get_sync_sync_raise_other(state: str = Depends(generator_state_try)): 4 ctx1abcd

177 assert state == "generator raise started" 3 ctx137#

178 raise OtherDependencyError() 3 ctx137#

179 

180 

181@app.get("/sync_context_b") 4 ctx1abcd

182def get_sync_context_b(state: dict = Depends(context_b)): 4 ctx1abcd

183 return state 3 ctx1lnp

184 

185 

186@app.get("/sync_context_b_raise") 4 ctx1abcd

187def get_sync_context_b_raise(state: dict = Depends(context_b)): 4 ctx1abcd

188 assert state["context_b"] == "started b" 3 ctx1tzF

189 assert state["context_a"] == "started a" 3 ctx1tzF

190 raise OtherDependencyError() 3 ctx1tzF

191 

192 

193@app.get("/sync_context_b_bg") 4 ctx1abcd

194async def get_sync_context_b_bg( 4 ctx1abcd

195 tasks: BackgroundTasks, state: dict = Depends(context_b) 

196): 

197 async def bg(state: dict): 3 ctx1hij

198 state["sync_bg"] = ( 3 ctx1hij

199 f"sync_bg set - b: {state['context_b']} - a: {state['context_a']}" 

200 ) 

201 

202 tasks.add_task(bg, state) 3 ctx1hij

203 return state 3 ctx1hij

204 

205 

206@app.middleware("http") 4 ctx1abcd

207async def middleware(request, call_next): 4 ctx1abcd

208 response: StreamingResponse = await call_next(request) 66 ctx1UIq0ekr1Js$hltVKu23Lv%WMw4fmx5Ny'inzXOA67PB(YQC8goD9RE)jpFZSG!#TH*

209 response.headers["x-state"] = json.dumps(state.copy()) 24 ctx10ek$hl2%4fm'in6(8go)jp!*

210 return response 24 ctx10ek$hl2%4fm'in6(8go)jp!*

211 

212 

213client = TestClient(app) 4 ctx1abcd

214 

215 

216def test_async_state(): 4 ctx1abcd

217 assert state["/async"] == "asyncgen not started" 3 ctx1048

218 response = client.get("/async") 3 ctx1048

219 assert response.status_code == 200, response.text 3 ctx1048

220 assert response.json() == "asyncgen started" 3 ctx1048

221 assert state["/async"] == "asyncgen completed" 3 ctx1048

222 

223 

224def test_sync_state(): 4 ctx1abcd

225 assert state["/sync"] == "generator not started" 3 ctx126!

226 response = client.get("/sync") 3 ctx126!

227 assert response.status_code == 200, response.text 3 ctx126!

228 assert response.json() == "generator started" 3 ctx126!

229 assert state["/sync"] == "generator completed" 3 ctx126!

230 

231 

232def test_async_raise_other(): 4 ctx1abcd

233 assert state["/async_raise"] == "asyncgen raise not started" 3 ctx1UWY

234 with pytest.raises(OtherDependencyError): 3 ctx1UWY

235 client.get("/async_raise_other") 3 ctx1UWY

236 assert state["/async_raise"] == "asyncgen raise finalized" 3 ctx1UWY

237 assert "/async_raise" not in errors 3 ctx1UWY

238 

239 

240def test_sync_raise_other(): 4 ctx1abcd

241 assert state["/sync_raise"] == "generator raise not started" 3 ctx1VXZ

242 with pytest.raises(OtherDependencyError): 3 ctx1VXZ

243 client.get("/sync_raise_other") 3 ctx1VXZ

244 assert state["/sync_raise"] == "generator raise finalized" 3 ctx1VXZ

245 assert "/sync_raise" not in errors 3 ctx1VXZ

246 

247 

248def test_async_raise_raises(): 4 ctx1abcd

249 with pytest.raises(AsyncDependencyError): 3 ctx1IMQ

250 client.get("/async_raise") 3 ctx1IMQ

251 assert state["/async_raise"] == "asyncgen raise finalized" 3 ctx1IMQ

252 assert "/async_raise" in errors 3 ctx1IMQ

253 errors.clear() 3 ctx1IMQ

254 

255 

256def test_async_raise_server_error(): 4 ctx1abcd

257 client = TestClient(app, raise_server_exceptions=False) 3 ctx1qwC

258 response = client.get("/async_raise") 3 ctx1qwC

259 assert response.status_code == 500, response.text 3 ctx1qwC

260 assert state["/async_raise"] == "asyncgen raise finalized" 3 ctx1qwC

261 assert "/async_raise" in errors 3 ctx1qwC

262 errors.clear() 3 ctx1qwC

263 

264 

265def test_context_b(): 4 ctx1abcd

266 response = client.get("/context_b") 3 ctx1kmo

267 data = response.json() 3 ctx1kmo

268 assert data["context_b"] == "started b" 3 ctx1kmo

269 assert data["context_a"] == "started a" 3 ctx1kmo

270 assert state["context_b"] == "finished b with a: started a" 3 ctx1kmo

271 assert state["context_a"] == "finished a" 3 ctx1kmo

272 

273 

274def test_context_b_raise(): 4 ctx1abcd

275 with pytest.raises(OtherDependencyError): 3 ctx1rxD

276 client.get("/context_b_raise") 3 ctx1rxD

277 assert state["context_b"] == "finished b with a: started a" 3 ctx1rxD

278 assert state["context_a"] == "finished a" 3 ctx1rxD

279 

280 

281def test_background_tasks(): 4 ctx1abcd

282 response = client.get("/context_b_bg") 3 ctx1efg

283 data = response.json() 3 ctx1efg

284 assert data["context_b"] == "started b" 3 ctx1efg

285 assert data["context_a"] == "started a" 3 ctx1efg

286 assert data["bg"] == "not set" 3 ctx1efg

287 middleware_state = json.loads(response.headers["x-state"]) 3 ctx1efg

288 assert middleware_state["context_b"] == "started b" 3 ctx1efg

289 assert middleware_state["context_a"] == "started a" 3 ctx1efg

290 assert middleware_state["bg"] == "not set" 3 ctx1efg

291 assert state["context_b"] == "finished b with a: started a" 3 ctx1efg

292 assert state["context_a"] == "finished a" 3 ctx1efg

293 assert state["bg"] == "bg set - b: started b - a: started a" 3 ctx1efg

294 

295 

296def test_sync_raise_raises(): 4 ctx1abcd

297 with pytest.raises(SyncDependencyError): 3 ctx1KOS

298 client.get("/sync_raise") 3 ctx1KOS

299 assert state["/sync_raise"] == "generator raise finalized" 3 ctx1KOS

300 assert "/sync_raise" in errors 3 ctx1KOS

301 errors.clear() 3 ctx1KOS

302 

303 

304def test_sync_raise_server_error(): 4 ctx1abcd

305 client = TestClient(app, raise_server_exceptions=False) 3 ctx1uAG

306 response = client.get("/sync_raise") 3 ctx1uAG

307 assert response.status_code == 500, response.text 3 ctx1uAG

308 assert state["/sync_raise"] == "generator raise finalized" 3 ctx1uAG

309 assert "/sync_raise" in errors 3 ctx1uAG

310 errors.clear() 3 ctx1uAG

311 

312 

313def test_sync_async_state(): 4 ctx1abcd

314 response = client.get("/sync_async") 3 ctx1$')

315 assert response.status_code == 200, response.text 3 ctx1$')

316 assert response.json() == "asyncgen started" 3 ctx1$')

317 assert state["/async"] == "asyncgen completed" 3 ctx1$')

318 

319 

320def test_sync_sync_state(): 4 ctx1abcd

321 response = client.get("/sync_sync") 3 ctx1%(*

322 assert response.status_code == 200, response.text 3 ctx1%(*

323 assert response.json() == "generator started" 3 ctx1%(*

324 assert state["/sync"] == "generator completed" 3 ctx1%(*

325 

326 

327def test_sync_async_raise_other(): 4 ctx1abcd

328 with pytest.raises(OtherDependencyError): 3 ctx1159

329 client.get("/sync_async_raise_other") 3 ctx1159

330 assert state["/async_raise"] == "asyncgen raise finalized" 3 ctx1159

331 assert "/async_raise" not in errors 3 ctx1159

332 

333 

334def test_sync_sync_raise_other(): 4 ctx1abcd

335 with pytest.raises(OtherDependencyError): 3 ctx137#

336 client.get("/sync_sync_raise_other") 3 ctx137#

337 assert state["/sync_raise"] == "generator raise finalized" 3 ctx137#

338 assert "/sync_raise" not in errors 3 ctx137#

339 

340 

341def test_sync_async_raise_raises(): 4 ctx1abcd

342 with pytest.raises(AsyncDependencyError): 3 ctx1JNR

343 client.get("/sync_async_raise") 3 ctx1JNR

344 assert state["/async_raise"] == "asyncgen raise finalized" 3 ctx1JNR

345 assert "/async_raise" in errors 3 ctx1JNR

346 errors.clear() 3 ctx1JNR

347 

348 

349def test_sync_async_raise_server_error(): 4 ctx1abcd

350 client = TestClient(app, raise_server_exceptions=False) 3 ctx1syE

351 response = client.get("/sync_async_raise") 3 ctx1syE

352 assert response.status_code == 500, response.text 3 ctx1syE

353 assert state["/async_raise"] == "asyncgen raise finalized" 3 ctx1syE

354 assert "/async_raise" in errors 3 ctx1syE

355 errors.clear() 3 ctx1syE

356 

357 

358def test_sync_sync_raise_raises(): 4 ctx1abcd

359 with pytest.raises(SyncDependencyError): 3 ctx1LPT

360 client.get("/sync_sync_raise") 3 ctx1LPT

361 assert state["/sync_raise"] == "generator raise finalized" 3 ctx1LPT

362 assert "/sync_raise" in errors 3 ctx1LPT

363 errors.clear() 3 ctx1LPT

364 

365 

366def test_sync_sync_raise_server_error(): 4 ctx1abcd

367 client = TestClient(app, raise_server_exceptions=False) 3 ctx1vBH

368 response = client.get("/sync_sync_raise") 3 ctx1vBH

369 assert response.status_code == 500, response.text 3 ctx1vBH

370 assert state["/sync_raise"] == "generator raise finalized" 3 ctx1vBH

371 assert "/sync_raise" in errors 3 ctx1vBH

372 errors.clear() 3 ctx1vBH

373 

374 

375def test_sync_context_b(): 4 ctx1abcd

376 response = client.get("/sync_context_b") 3 ctx1lnp

377 data = response.json() 3 ctx1lnp

378 assert data["context_b"] == "started b" 3 ctx1lnp

379 assert data["context_a"] == "started a" 3 ctx1lnp

380 assert state["context_b"] == "finished b with a: started a" 3 ctx1lnp

381 assert state["context_a"] == "finished a" 3 ctx1lnp

382 

383 

384def test_sync_context_b_raise(): 4 ctx1abcd

385 with pytest.raises(OtherDependencyError): 3 ctx1tzF

386 client.get("/sync_context_b_raise") 3 ctx1tzF

387 assert state["context_b"] == "finished b with a: started a" 3 ctx1tzF

388 assert state["context_a"] == "finished a" 3 ctx1tzF

389 

390 

391def test_sync_background_tasks(): 4 ctx1abcd

392 response = client.get("/sync_context_b_bg") 3 ctx1hij

393 data = response.json() 3 ctx1hij

394 assert data["context_b"] == "started b" 3 ctx1hij

395 assert data["context_a"] == "started a" 3 ctx1hij

396 assert data["sync_bg"] == "not set" 3 ctx1hij

397 assert state["context_b"] == "finished b with a: started a" 3 ctx1hij

398 assert state["context_a"] == "finished a" 3 ctx1hij

399 assert state["sync_bg"] == "sync_bg set - b: started b - a: started a" 3 ctx1hij