Coverage for tests/test_dependency_contextmanager.py: 100%

271 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2025-12-04 08:29 +0000

1import json 1abcdefg

2from typing import Dict 1abcdefg

3 

4import pytest 1abcdefg

5from fastapi import BackgroundTasks, Depends, FastAPI 1abcdefg

6from fastapi.responses import StreamingResponse 1abcdefg

7from fastapi.testclient import TestClient 1abcdefg

8 

9app = FastAPI() 1abcdefg

10state = { 1abcdefg

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

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

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

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

15 "context_a": "not started a", 

16 "context_b": "not started b", 

17 "bg": "not set", 

18 "sync_bg": "not set", 

19} 

20 

21errors = [] 1abcdefg

22 

23 

24async def get_state(): 1abcdefg

25 return state 2qb. J Ebh v K Fb/ L 6bo w M rb: N GbHb; O 7bsbmb= Ibi x P Jbnb? 8bp y Q tbob@ KbLbpb[ 9bub] R Mbj z S Nb^ T !bq A U vb_ V ObPb` W #bwb{ X Qbk B Y Rb| Z $br C 0 xb} 1 SbTb~ 2 %bybab3 Ubl D 4 Vbbb5 'bs E 6 zbcb7 WbXbdb8 (bAbeb9 Ybm F ! Zbfb# )bt G $ Bbgb% 0b1bhb' *bCbib( 2bn H ) 3bjb* +bu I + Dbkb, 4b5blb- ,b

26 

27 

28class AsyncDependencyError(Exception): 1abcdefg

29 pass 1abcdefg

30 

31 

32class SyncDependencyError(Exception): 1abcdefg

33 pass 1abcdefg

34 

35 

36class OtherDependencyError(Exception): 1abcdefg

37 pass 1abcdefg

38 

39 

40async def asyncgen_state(state: Dict[str, str] = Depends(get_state)): 1abcdefg

41 state["/async"] = "asyncgen started" 2Eb6bIb8bMb!bQb$bUb'bYb)b2b+b

42 yield state["/async"] 2Eb6bIb8bMb!bQb$bUb'bYb)b2b+b

43 state["/async"] = "asyncgen completed" 2Eb6bIb8bMb!bQb$bUb'bYb)b2b+b

44 

45 

46def generator_state(state: Dict[str, str] = Depends(get_state)): 1abcdefg

47 state["/sync"] = "generator started" 2Gb7bKb9bOb#bSb%bWb(b0b*b4b,b

48 yield state["/sync"] 2Gb7bKb9bOb#bSb%bWb(b0b*b4b,b

49 state["/sync"] = "generator completed" 2Gb7bKb9bOb#bSb%bWb(b0b*b4b,b

50 

51 

52async def asyncgen_state_try(state: Dict[str, str] = Depends(get_state)): 1abcdefg

53 state["/async_raise"] = "asyncgen raise started" 2qb. J Fb/ L sbmb= Jbnb? ub] R Nb^ T wb{ X Rb| Z ybab3 Vbbb5 Abeb9 Zbfb# Cbib( 3bjb*

54 try: 2qb. J Fb/ L sbmb= Jbnb? ub] R Nb^ T wb{ X Rb| Z ybab3 Vbbb5 Abeb9 Zbfb# Cbib( 3bjb*

55 yield state["/async_raise"] 2qb. J Fb/ L sbmb= Jbnb? ub] R Nb^ T wb{ X Rb| Z ybab3 Vbbb5 Abeb9 Zbfb# Cbib( 3bjb*

56 except AsyncDependencyError: 2qb. J Fb/ L sbmb= Jbnb? ub] R Nb^ T wb{ X Rb| Z ybab3 Vbbb5 Abeb9 Zbfb# Cbib( 3bjb*

57 errors.append("/async_raise") 2. J / L mb= nb? ] R ^ T { X | Z ab3 bb5 eb9 fb# ib( jb*

58 raise 2. J / L mb= nb? ] R ^ T { X | Z ab3 bb5 eb9 fb# ib( jb*

59 finally: 

60 state["/async_raise"] = "asyncgen raise finalized" 2qb. J Fb/ L sbJbub] R Nb^ T wb{ X Rb| Z ybab3 Vbbb5 Abeb9 Zbfb# Cbib( 3bjb*

61 

62 

63def generator_state_try(state: Dict[str, str] = Depends(get_state)): 1abcdefg

64 state["/sync_raise"] = "generator raise started" 2rb: N Hb; O tbob@ Lbpb[ vb_ V Pb` W xb} 1 Tb~ 2 zbcb7 Xbdb8 Bbgb% 1bhb' Dbkb, 5blb-

65 try: 2rb: N Hb; O tbob@ Lbpb[ vb_ V Pb` W xb} 1 Tb~ 2 zbcb7 Xbdb8 Bbgb% 1bhb' Dbkb, 5blb-

66 yield state["/sync_raise"] 2rb: N Hb; O tbob@ Lbpb[ vb_ V Pb` W xb} 1 Tb~ 2 zbcb7 Xbdb8 Bbgb% 1bhb' Dbkb, 5blb-

67 except SyncDependencyError: 2rb: N Hb; O tbob@ Lbpb[ vb_ V Pb` W xb} 1 Tb~ 2 zbcb7 Xbdb8 Bbgb% 1bhb' Dbkb, 5blb-

68 errors.append("/sync_raise") 2: N ; O ob@ pb[ _ V ` W } 1 ~ 2 cb7 db8 gb% hb' kb, lb-

69 raise 2: N ; O ob@ pb[ _ V ` W } 1 ~ 2 cb7 db8 gb% hb' kb, lb-

70 finally: 

71 state["/sync_raise"] = "generator raise finalized" 2rb: N Hb; O tbLbvb_ V Pb` W xb} 1 Tb~ 2 zbcb7 Xbdb8 Bbgb% 1bhb' Dbkb, 5blb-

72 

73 

74async def context_a(state: dict = Depends(get_state)): 1abcdefg

75 state["context_a"] = "started a" 1hvKowMixPpyQjzSqAUkBYrC0lD4sE6mF!tG$nH)uI+

76 try: 1hvKowMixPpyQjzSqAUkBYrC0lD4sE6mF!tG$nH)uI+

77 yield state 1hvKowMixPpyQjzSqAUkBYrC0lD4sE6mF!tG$nH)uI+

78 finally: 

79 state["context_a"] = "finished a" 1hvKowMixPpyQjzSqAUkBYrC0lD4sE6mF!tG$nH)uI+

80 

81 

82async def context_b(state: dict = Depends(context_a)): 1abcdefg

83 state["context_b"] = "started b" 1hvKowMixPpyQjzSqAUkBYrC0lD4sE6mF!tG$nH)uI+

84 try: 1hvKowMixPpyQjzSqAUkBYrC0lD4sE6mF!tG$nH)uI+

85 yield state 1hvKowMixPpyQjzSqAUkBYrC0lD4sE6mF!tG$nH)uI+

86 finally: 

87 state["context_b"] = f"finished b with a: {state['context_a']}" 1hvKowMixPpyQjzSqAUkBYrC0lD4sE6mF!tG$nH)uI+

88 

89 

90@app.get("/async") 1abcdefg

91async def get_async(state: str = Depends(asyncgen_state)): 1abcdefg

92 return state 2EbIbMbQbUbYb2b

93 

94 

95@app.get("/sync") 1abcdefg

96async def get_sync(state: str = Depends(generator_state)): 1abcdefg

97 return state 2GbKbObSbWb0b4b

98 

99 

100@app.get("/async_raise") 1abcdefg

101async def get_async_raise(state: str = Depends(asyncgen_state_try)): 1abcdefg

102 assert state == "asyncgen raise started" 2. J mb= ] R { X ab3 eb9 ib(

103 raise AsyncDependencyError() 2. J mb= ] R { X ab3 eb9 ib(

104 

105 

106@app.get("/sync_raise") 1abcdefg

107async def get_sync_raise(state: str = Depends(generator_state_try)): 1abcdefg

108 assert state == "generator raise started" 2: N ob@ _ V } 1 cb7 gb% kb,

109 raise SyncDependencyError() 2: N ob@ _ V } 1 cb7 gb% kb,

110 

111 

112@app.get("/async_raise_other") 1abcdefg

113async def get_async_raise_other(state: str = Depends(asyncgen_state_try)): 1abcdefg

114 assert state == "asyncgen raise started" 2qbsbubwbybAbCb

115 raise OtherDependencyError() 2qbsbubwbybAbCb

116 

117 

118@app.get("/sync_raise_other") 1abcdefg

119async def get_sync_raise_other(state: str = Depends(generator_state_try)): 1abcdefg

120 assert state == "generator raise started" 2rbtbvbxbzbBbDb

121 raise OtherDependencyError() 2rbtbvbxbzbBbDb

122 

123 

124@app.get("/context_b") 1abcdefg

125async def get_context_b(state: dict = Depends(context_b)): 1abcdefg

126 return state 1vxzBDFH

127 

128 

129@app.get("/context_b_raise") 1abcdefg

130async def get_context_b_raise(state: dict = Depends(context_b)): 1abcdefg

131 assert state["context_b"] == "started b" 1KPSY4!)

132 assert state["context_a"] == "started a" 1KPSY4!)

133 raise OtherDependencyError() 1KPSY4!)

134 

135 

136@app.get("/context_b_bg") 1abcdefg

137async def get_context_b_bg(tasks: BackgroundTasks, state: dict = Depends(context_b)): 1abcdefg

138 async def bg(state: dict): 1hijklmn

139 state["bg"] = f"bg set - b: {state['context_b']} - a: {state['context_a']}" 1hijklmn

140 

141 tasks.add_task(bg, state) 1hijklmn

142 return state 1hijklmn

143 

144 

145# Sync versions 

146 

147 

148@app.get("/sync_async") 1abcdefg

149def get_sync_async(state: str = Depends(asyncgen_state)): 1abcdefg

150 return state 26b8b!b$b'b)b+b

151 

152 

153@app.get("/sync_sync") 1abcdefg

154def get_sync_sync(state: str = Depends(generator_state)): 1abcdefg

155 return state 27b9b#b%b(b*b,b

156 

157 

158@app.get("/sync_async_raise") 1abcdefg

159def get_sync_async_raise(state: str = Depends(asyncgen_state_try)): 1abcdefg

160 assert state == "asyncgen raise started" 2/ L nb? ^ T | Z bb5 fb# jb*

161 raise AsyncDependencyError() 2/ L nb? ^ T | Z bb5 fb# jb*

162 

163 

164@app.get("/sync_sync_raise") 1abcdefg

165def get_sync_sync_raise(state: str = Depends(generator_state_try)): 1abcdefg

166 assert state == "generator raise started" 2; O pb[ ` W ~ 2 db8 hb' lb-

167 raise SyncDependencyError() 2; O pb[ ` W ~ 2 db8 hb' lb-

168 

169 

170@app.get("/sync_async_raise_other") 1abcdefg

171def get_sync_async_raise_other(state: str = Depends(asyncgen_state_try)): 1abcdefg

172 assert state == "asyncgen raise started" 2FbJbNbRbVbZb3b

173 raise OtherDependencyError() 2FbJbNbRbVbZb3b

174 

175 

176@app.get("/sync_sync_raise_other") 1abcdefg

177def get_sync_sync_raise_other(state: str = Depends(generator_state_try)): 1abcdefg

178 assert state == "generator raise started" 2HbLbPbTbXb1b5b

179 raise OtherDependencyError() 2HbLbPbTbXb1b5b

180 

181 

182@app.get("/sync_context_b") 1abcdefg

183def get_sync_context_b(state: dict = Depends(context_b)): 1abcdefg

184 return state 1wyACEGI

185 

186 

187@app.get("/sync_context_b_raise") 1abcdefg

188def get_sync_context_b_raise(state: dict = Depends(context_b)): 1abcdefg

189 assert state["context_b"] == "started b" 1MQU06$+

190 assert state["context_a"] == "started a" 1MQU06$+

191 raise OtherDependencyError() 1MQU06$+

192 

193 

194@app.get("/sync_context_b_bg") 1abcdefg

195async def get_sync_context_b_bg( 1abcdefg

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

197): 

198 async def bg(state: dict): 1opqrstu

199 state["sync_bg"] = ( 1opqrstu

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

201 ) 

202 

203 tasks.add_task(bg, state) 1opqrstu

204 return state 1opqrstu

205 

206 

207@app.middleware("http") 1abcdefg

208async def middleware(request, call_next): 1abcdefg

209 response: StreamingResponse = await call_next(request) 2qb. J Ebh v K Fb/ L 6bo w M rb: N GbHb; O 7bsbmb= Ibi x P Jbnb? 8bp y Q tbob@ KbLbpb[ 9bub] R Mbj z S Nb^ T !bq A U vb_ V ObPb` W #bwb{ X Qbk B Y Rb| Z $br C 0 xb} 1 SbTb~ 2 %bybab3 Ubl D 4 Vbbb5 'bs E 6 zbcb7 WbXbdb8 (bAbeb9 Ybm F ! Zbfb# )bt G $ Bbgb% 0b1bhb' *bCbib( 2bn H ) 3bjb* +bu I + Dbkb, 4b5blb- ,b

210 response.headers["x-state"] = json.dumps(state.copy()) 2Ebh v 6bo w Gb7bIbi x 8bp y Kb9bMbj z !bq A Ob#bQbk B $br C Sb%bUbl D 'bs E Wb(bYbm F )bt G 0b*b2bn H +bu I 4b,b

211 return response 2Ebh v 6bo w Gb7bIbi x 8bp y Kb9bMbj z !bq A Ob#bQbk B $br C Sb%bUbl D 'bs E Wb(bYbm F )bt G 0b*b2bn H +bu I 4b,b

212 

213 

214client = TestClient(app) 1abcdefg

215 

216 

217def test_async_state(): 1abcdefg

218 assert state["/async"] == "asyncgen not started" 2EbIbMbQbUbYb2b

219 response = client.get("/async") 2EbIbMbQbUbYb2b

220 assert response.status_code == 200, response.text 2EbIbMbQbUbYb2b

221 assert response.json() == "asyncgen started" 2EbIbMbQbUbYb2b

222 assert state["/async"] == "asyncgen completed" 2EbIbMbQbUbYb2b

223 

224 

225def test_sync_state(): 1abcdefg

226 assert state["/sync"] == "generator not started" 2GbKbObSbWb0b4b

227 response = client.get("/sync") 2GbKbObSbWb0b4b

228 assert response.status_code == 200, response.text 2GbKbObSbWb0b4b

229 assert response.json() == "generator started" 2GbKbObSbWb0b4b

230 assert state["/sync"] == "generator completed" 2GbKbObSbWb0b4b

231 

232 

233def test_async_raise_other(): 1abcdefg

234 assert state["/async_raise"] == "asyncgen raise not started" 2qbsbubwbybAbCb

235 with pytest.raises(OtherDependencyError): 2qbsbubwbybAbCb

236 client.get("/async_raise_other") 2qbsbubwbybAbCb

237 assert state["/async_raise"] == "asyncgen raise finalized" 2qbsbubwbybAbCb

238 assert "/async_raise" not in errors 2qbsbubwbybAbCb

239 

240 

241def test_sync_raise_other(): 1abcdefg

242 assert state["/sync_raise"] == "generator raise not started" 2rbtbvbxbzbBbDb

243 with pytest.raises(OtherDependencyError): 2rbtbvbxbzbBbDb

244 client.get("/sync_raise_other") 2rbtbvbxbzbBbDb

245 assert state["/sync_raise"] == "generator raise finalized" 2rbtbvbxbzbBbDb

246 assert "/sync_raise" not in errors 2rbtbvbxbzbBbDb

247 

248 

249def test_async_raise_raises(): 1abcdefg

250 with pytest.raises(AsyncDependencyError): 2. mb] { abebib

251 client.get("/async_raise") 2. mb] { abebib

252 assert state["/async_raise"] == "asyncgen raise finalized" 2. mb] { abebib

253 assert "/async_raise" in errors 2. mb] { abebib

254 errors.clear() 2. mb] { abebib

255 

256 

257def test_async_raise_server_error(): 1abcdefg

258 client = TestClient(app, raise_server_exceptions=False) 1J=RX39(

259 response = client.get("/async_raise") 1J=RX39(

260 assert response.status_code == 500, response.text 1J=RX39(

261 assert state["/async_raise"] == "asyncgen raise finalized" 1J=RX39(

262 assert "/async_raise" in errors 1J=RX39(

263 errors.clear() 1J=RX39(

264 

265 

266def test_context_b(): 1abcdefg

267 response = client.get("/context_b") 1vxzBDFH

268 data = response.json() 1vxzBDFH

269 assert data["context_b"] == "started b" 1vxzBDFH

270 assert data["context_a"] == "started a" 1vxzBDFH

271 assert state["context_b"] == "finished b with a: started a" 1vxzBDFH

272 assert state["context_a"] == "finished a" 1vxzBDFH

273 

274 

275def test_context_b_raise(): 1abcdefg

276 with pytest.raises(OtherDependencyError): 1KPSY4!)

277 client.get("/context_b_raise") 1KPSY4!)

278 assert state["context_b"] == "finished b with a: started a" 1KPSY4!)

279 assert state["context_a"] == "finished a" 1KPSY4!)

280 

281 

282def test_background_tasks(): 1abcdefg

283 response = client.get("/context_b_bg") 1hijklmn

284 data = response.json() 1hijklmn

285 assert data["context_b"] == "started b" 1hijklmn

286 assert data["context_a"] == "started a" 1hijklmn

287 assert data["bg"] == "not set" 1hijklmn

288 middleware_state = json.loads(response.headers["x-state"]) 1hijklmn

289 assert middleware_state["context_b"] == "started b" 1hijklmn

290 assert middleware_state["context_a"] == "started a" 1hijklmn

291 assert middleware_state["bg"] == "not set" 1hijklmn

292 assert state["context_b"] == "finished b with a: started a" 1hijklmn

293 assert state["context_a"] == "finished a" 1hijklmn

294 assert state["bg"] == "bg set - b: started b - a: started a" 1hijklmn

295 

296 

297def test_sync_raise_raises(): 1abcdefg

298 with pytest.raises(SyncDependencyError): 2: ob_ } cbgbkb

299 client.get("/sync_raise") 2: ob_ } cbgbkb

300 assert state["/sync_raise"] == "generator raise finalized" 2: ob_ } cbgbkb

301 assert "/sync_raise" in errors 2: ob_ } cbgbkb

302 errors.clear() 2: ob_ } cbgbkb

303 

304 

305def test_sync_raise_server_error(): 1abcdefg

306 client = TestClient(app, raise_server_exceptions=False) 1N@V17%,

307 response = client.get("/sync_raise") 1N@V17%,

308 assert response.status_code == 500, response.text 1N@V17%,

309 assert state["/sync_raise"] == "generator raise finalized" 1N@V17%,

310 assert "/sync_raise" in errors 1N@V17%,

311 errors.clear() 1N@V17%,

312 

313 

314def test_sync_async_state(): 1abcdefg

315 response = client.get("/sync_async") 26b8b!b$b'b)b+b

316 assert response.status_code == 200, response.text 26b8b!b$b'b)b+b

317 assert response.json() == "asyncgen started" 26b8b!b$b'b)b+b

318 assert state["/async"] == "asyncgen completed" 26b8b!b$b'b)b+b

319 

320 

321def test_sync_sync_state(): 1abcdefg

322 response = client.get("/sync_sync") 27b9b#b%b(b*b,b

323 assert response.status_code == 200, response.text 27b9b#b%b(b*b,b

324 assert response.json() == "generator started" 27b9b#b%b(b*b,b

325 assert state["/sync"] == "generator completed" 27b9b#b%b(b*b,b

326 

327 

328def test_sync_async_raise_other(): 1abcdefg

329 with pytest.raises(OtherDependencyError): 2FbJbNbRbVbZb3b

330 client.get("/sync_async_raise_other") 2FbJbNbRbVbZb3b

331 assert state["/async_raise"] == "asyncgen raise finalized" 2FbJbNbRbVbZb3b

332 assert "/async_raise" not in errors 2FbJbNbRbVbZb3b

333 

334 

335def test_sync_sync_raise_other(): 1abcdefg

336 with pytest.raises(OtherDependencyError): 2HbLbPbTbXb1b5b

337 client.get("/sync_sync_raise_other") 2HbLbPbTbXb1b5b

338 assert state["/sync_raise"] == "generator raise finalized" 2HbLbPbTbXb1b5b

339 assert "/sync_raise" not in errors 2HbLbPbTbXb1b5b

340 

341 

342def test_sync_async_raise_raises(): 1abcdefg

343 with pytest.raises(AsyncDependencyError): 2/ nb^ | bbfbjb

344 client.get("/sync_async_raise") 2/ nb^ | bbfbjb

345 assert state["/async_raise"] == "asyncgen raise finalized" 2/ nb^ | bbfbjb

346 assert "/async_raise" in errors 2/ nb^ | bbfbjb

347 errors.clear() 2/ nb^ | bbfbjb

348 

349 

350def test_sync_async_raise_server_error(): 1abcdefg

351 client = TestClient(app, raise_server_exceptions=False) 1L?TZ5#*

352 response = client.get("/sync_async_raise") 1L?TZ5#*

353 assert response.status_code == 500, response.text 1L?TZ5#*

354 assert state["/async_raise"] == "asyncgen raise finalized" 1L?TZ5#*

355 assert "/async_raise" in errors 1L?TZ5#*

356 errors.clear() 1L?TZ5#*

357 

358 

359def test_sync_sync_raise_raises(): 1abcdefg

360 with pytest.raises(SyncDependencyError): 2; pb` ~ dbhblb

361 client.get("/sync_sync_raise") 2; pb` ~ dbhblb

362 assert state["/sync_raise"] == "generator raise finalized" 2; pb` ~ dbhblb

363 assert "/sync_raise" in errors 2; pb` ~ dbhblb

364 errors.clear() 2; pb` ~ dbhblb

365 

366 

367def test_sync_sync_raise_server_error(): 1abcdefg

368 client = TestClient(app, raise_server_exceptions=False) 1O[W28'-

369 response = client.get("/sync_sync_raise") 1O[W28'-

370 assert response.status_code == 500, response.text 1O[W28'-

371 assert state["/sync_raise"] == "generator raise finalized" 1O[W28'-

372 assert "/sync_raise" in errors 1O[W28'-

373 errors.clear() 1O[W28'-

374 

375 

376def test_sync_context_b(): 1abcdefg

377 response = client.get("/sync_context_b") 1wyACEGI

378 data = response.json() 1wyACEGI

379 assert data["context_b"] == "started b" 1wyACEGI

380 assert data["context_a"] == "started a" 1wyACEGI

381 assert state["context_b"] == "finished b with a: started a" 1wyACEGI

382 assert state["context_a"] == "finished a" 1wyACEGI

383 

384 

385def test_sync_context_b_raise(): 1abcdefg

386 with pytest.raises(OtherDependencyError): 1MQU06$+

387 client.get("/sync_context_b_raise") 1MQU06$+

388 assert state["context_b"] == "finished b with a: started a" 1MQU06$+

389 assert state["context_a"] == "finished a" 1MQU06$+

390 

391 

392def test_sync_background_tasks(): 1abcdefg

393 response = client.get("/sync_context_b_bg") 1opqrstu

394 data = response.json() 1opqrstu

395 assert data["context_b"] == "started b" 1opqrstu

396 assert data["context_a"] == "started a" 1opqrstu

397 assert data["sync_bg"] == "not set" 1opqrstu

398 assert state["context_b"] == "finished b with a: started a" 1opqrstu

399 assert state["context_a"] == "finished a" 1opqrstu

400 assert state["sync_bg"] == "sync_bg set - b: started b - a: started a" 1opqrstu