Coverage for tests / test_dependency_contextmanager.py: 100%

270 statements  

« prev     ^ index     » next       coverage.py v7.13.3, created at 2026-04-06 01:24 +0000

1import json 1abcd

2 

3import pytest 1abcd

4from fastapi import BackgroundTasks, Depends, FastAPI 1abcd

5from fastapi.responses import StreamingResponse 1abcd

6from fastapi.testclient import TestClient 1abcd

7 

8app = FastAPI() 1abcd

9state = { 1abcd

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 = [] 1abcd

21 

22 

23async def get_state(): 1abcd

24 return state 1XJr4esl5Kt+humYLv67Mw,89ZNx!fyn#Oz-iAo0PB$%QC.1RD'gEp(SF/jGq2TH)*UI:

25 

26 

27class AsyncDependencyError(Exception): 1abcd

28 pass 1abcd

29 

30 

31class SyncDependencyError(Exception): 1abcd

32 pass 1abcd

33 

34 

35class OtherDependencyError(Exception): 1abcd

36 pass 1abcd

37 

38 

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

40 state["/async"] = "asyncgen started" 14+8!-'/

41 yield state["/async"] 14+8!-'/

42 state["/async"] = "asyncgen completed" 14+8!-'/

43 

44 

45def generator_state(state: dict[str, str] = Depends(get_state)): 1abcd

46 state["/sync"] = "generator started" 16,9$.):

47 yield state["/sync"] 16,9$.):

48 state["/sync"] = "generator completed" 16,9$.):

49 

50 

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

52 state["/async_raise"] = "asyncgen raise started" 1XJr5Kt;VZNx#Oz1RD(SF

53 try: 1XJr5Kt;VZNx#Oz1RD(SF

54 yield state["/async_raise"] 1XJr5Kt;VZNx#Oz1RD(SF

55 except AsyncDependencyError: 1XJr5Kt;VZNx#Oz1RD(SF

56 errors.append("/async_raise") 1JrKt[VNxOzRDSF

57 raise 1JrKt[VNxOzRDSF

58 finally: 

59 state["/async_raise"] = "asyncgen raise finalized" 1XJr5Kt;VZNx#Oz1RD(SF

60 

61 

62def generator_state_try(state: dict[str, str] = Depends(get_state)): 1abcd

63 state["/sync_raise"] = "generator raise started" 1YLv7Mw=30PB%QC2TH*UI

64 try: 1YLv7Mw=30PB%QC2TH*UI

65 yield state["/sync_raise"] 1YLv7Mw=30PB%QC2TH*UI

66 except SyncDependencyError: 1YLv7Mw=30PB%QC2TH*UI

67 errors.append("/sync_raise") 1LvMw3@PBQCTHUI

68 raise 1LvMw3@PBQCTHUI

69 finally: 

70 state["/sync_raise"] = "generator raise finalized" 1YLv7Mw=30PB%QC2TH*UI

71 

72 

73async def context_a(state: dict = Depends(get_state)): 1abcd

74 state["context_a"] = "started a" 1eslhumkWfyniAogEpjGq

75 try: 1eslhumkWfyniAogEpjGq

76 yield state 1eslhumkWfyniAogEpjGq

77 finally: 

78 state["context_a"] = "finished a" 1eslhumk]W^fyniAogEpjGq

79 

80 

81async def context_b(state: dict = Depends(context_a)): 1abcd

82 state["context_b"] = "started b" 1eslhumkWfyniAogEpjGq

83 try: 1eslhumkWfyniAogEpjGq

84 yield state 1eslhumkWfyniAogEpjGq

85 finally: 

86 state["context_b"] = f"finished b with a: {state['context_a']}" 1eslhumk]W^fyniAogEpjGq

87 

88 

89@app.get("/async") 1abcd

90async def get_async(state: str = Depends(asyncgen_state)): 1abcd

91 return state 148!'

92 

93 

94@app.get("/sync") 1abcd

95async def get_sync(state: str = Depends(generator_state)): 1abcd

96 return state 169$)

97 

98 

99@app.get("/async_raise") 1abcd

100async def get_async_raise(state: str = Depends(asyncgen_state_try)): 1abcd

101 assert state == "asyncgen raise started" 1Jr[VNxRD

102 raise AsyncDependencyError() 1Jr[VNxRD

103 

104 

105@app.get("/sync_raise") 1abcd

106async def get_sync_raise(state: str = Depends(generator_state_try)): 1abcd

107 assert state == "generator raise started" 1Lv3@PBTH

108 raise SyncDependencyError() 1Lv3@PBTH

109 

110 

111@app.get("/async_raise_other") 1abcd

112async def get_async_raise_other(state: str = Depends(asyncgen_state_try)): 1abcd

113 assert state == "asyncgen raise started" 1X;Z1

114 raise OtherDependencyError() 1X;Z1

115 

116 

117@app.get("/sync_raise_other") 1abcd

118async def get_sync_raise_other(state: str = Depends(generator_state_try)): 1abcd

119 assert state == "generator raise started" 1Y=02

120 raise OtherDependencyError() 1Y=02

121 

122 

123@app.get("/context_b") 1abcd

124async def get_context_b(state: dict = Depends(context_b)): 1abcd

125 return state 1lWnp

126 

127 

128@app.get("/context_b_raise") 1abcd

129async def get_context_b_raise(state: dict = Depends(context_b)): 1abcd

130 assert state["context_b"] == "started b" 1s]yE

131 assert state["context_a"] == "started a" 1s]yE

132 raise OtherDependencyError() 1s]yE

133 

134 

135@app.get("/context_b_bg") 1abcd

136async def get_context_b_bg(tasks: BackgroundTasks, state: dict = Depends(context_b)): 1abcd

137 async def bg(state: dict): 1ekfg

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

139 

140 tasks.add_task(bg, state) 1ekfg

141 return state 1ekfg

142 

143 

144# Sync versions 

145 

146 

147@app.get("/sync_async") 1abcd

148def get_sync_async(state: str = Depends(asyncgen_state)): 1abcd

149 return state 2+ bb- /

150 

151 

152@app.get("/sync_sync") 1abcd

153def get_sync_sync(state: str = Depends(generator_state)): 1abcd

154 return state 2, cb. :

155 

156 

157@app.get("/sync_async_raise") 1abcd

158def get_sync_async_raise(state: str = Depends(asyncgen_state_try)): 1abcd

159 assert state == "asyncgen raise started" 1Kt{_OzSF

160 raise AsyncDependencyError() 1Kt{_OzSF

161 

162 

163@app.get("/sync_sync_raise") 1abcd

164def get_sync_sync_raise(state: str = Depends(generator_state_try)): 1abcd

165 assert state == "generator raise started" 1Mw|`QCUI

166 raise SyncDependencyError() 1Mw|`QCUI

167 

168 

169@app.get("/sync_async_raise_other") 1abcd

170def get_sync_async_raise_other(state: str = Depends(asyncgen_state_try)): 1abcd

171 assert state == "asyncgen raise started" 15~#(

172 raise OtherDependencyError() 15~#(

173 

174 

175@app.get("/sync_sync_raise_other") 1abcd

176def get_sync_sync_raise_other(state: str = Depends(generator_state_try)): 1abcd

177 assert state == "generator raise started" 27 ab% *

178 raise OtherDependencyError() 27 ab% *

179 

180 

181@app.get("/sync_context_b") 1abcd

182def get_sync_context_b(state: dict = Depends(context_b)): 1abcd

183 return state 1m}oq

184 

185 

186@app.get("/sync_context_b_raise") 1abcd

187def get_sync_context_b_raise(state: dict = Depends(context_b)): 1abcd

188 assert state["context_b"] == "started b" 1u^AG

189 assert state["context_a"] == "started a" 1u^AG

190 raise OtherDependencyError() 1u^AG

191 

192 

193@app.get("/sync_context_b_bg") 1abcd

194async def get_sync_context_b_bg( 1abcd

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

196): 

197 async def bg(state: dict): 1h?ij

198 state["sync_bg"] = ( 1h?ij

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

200 ) 

201 

202 tasks.add_task(bg, state) 1h?ij

203 return state 1h?ij

204 

205 

206@app.middleware("http") 1abcd

207async def middleware(request, call_next): 1abcd

208 response: StreamingResponse = await call_next(request) 1XJr4esl5Kt+humYLv67Mw,89ZNx!fyn#Oz-iAo0PB$%QC.1RD'gEp(SF/jGq2TH)*UI:

209 response.headers["x-state"] = json.dumps(state.copy()) 14el+hm6,89!fn-io$.'gp/jq):

210 return response 14el+hm6,89!fn-io$.'gp/jq):

211 

212 

213client = TestClient(app) 1abcd

214 

215 

216def test_async_state(): 1abcd

217 assert state["/async"] == "asyncgen not started" 148!'

218 response = client.get("/async") 148!'

219 assert response.status_code == 200, response.text 148!'

220 assert response.json() == "asyncgen started" 148!'

221 assert state["/async"] == "asyncgen completed" 148!'

222 

223 

224def test_sync_state(): 1abcd

225 assert state["/sync"] == "generator not started" 169$)

226 response = client.get("/sync") 169$)

227 assert response.status_code == 200, response.text 169$)

228 assert response.json() == "generator started" 169$)

229 assert state["/sync"] == "generator completed" 169$)

230 

231 

232def test_async_raise_other(): 1abcd

233 assert state["/async_raise"] == "asyncgen raise not started" 1X;Z1

234 with pytest.raises(OtherDependencyError): 1X;Z1

235 client.get("/async_raise_other") 1X;Z1

236 assert state["/async_raise"] == "asyncgen raise finalized" 1X;Z1

237 assert "/async_raise" not in errors 1X;Z1

238 

239 

240def test_sync_raise_other(): 1abcd

241 assert state["/sync_raise"] == "generator raise not started" 1Y=02

242 with pytest.raises(OtherDependencyError): 1Y=02

243 client.get("/sync_raise_other") 1Y=02

244 assert state["/sync_raise"] == "generator raise finalized" 1Y=02

245 assert "/sync_raise" not in errors 1Y=02

246 

247 

248def test_async_raise_raises(): 1abcd

249 with pytest.raises(AsyncDependencyError): 1J[NR

250 client.get("/async_raise") 1J[NR

251 assert state["/async_raise"] == "asyncgen raise finalized" 1J[NR

252 assert "/async_raise" in errors 1J[NR

253 errors.clear() 1J[NR

254 

255 

256def test_async_raise_server_error(): 1abcd

257 client = TestClient(app, raise_server_exceptions=False) 1rVxD

258 response = client.get("/async_raise") 1rVxD

259 assert response.status_code == 500, response.text 1rVxD

260 assert state["/async_raise"] == "asyncgen raise finalized" 1rVxD

261 assert "/async_raise" in errors 1rVxD

262 errors.clear() 1rVxD

263 

264 

265def test_context_b(): 1abcd

266 response = client.get("/context_b") 1lWnp

267 data = response.json() 1lWnp

268 assert data["context_b"] == "started b" 1lWnp

269 assert data["context_a"] == "started a" 1lWnp

270 assert state["context_b"] == "finished b with a: started a" 1lWnp

271 assert state["context_a"] == "finished a" 1lWnp

272 

273 

274def test_context_b_raise(): 1abcd

275 with pytest.raises(OtherDependencyError): 1s]yE

276 client.get("/context_b_raise") 1s]yE

277 assert state["context_b"] == "finished b with a: started a" 1s]yE

278 assert state["context_a"] == "finished a" 1s]yE

279 

280 

281def test_background_tasks(): 1abcd

282 response = client.get("/context_b_bg") 1ekfg

283 data = response.json() 1ekfg

284 assert data["context_b"] == "started b" 1ekfg

285 assert data["context_a"] == "started a" 1ekfg

286 assert data["bg"] == "not set" 1ekfg

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

288 assert middleware_state["context_b"] == "started b" 1ekfg

289 assert middleware_state["context_a"] == "started a" 1ekfg

290 assert middleware_state["bg"] == "not set" 1ekfg

291 assert state["context_b"] == "finished b with a: started a" 1ekfg

292 assert state["context_a"] == "finished a" 1ekfg

293 assert state["bg"] == "bg set - b: started b - a: started a" 1ekfg

294 

295 

296def test_sync_raise_raises(): 1abcd

297 with pytest.raises(SyncDependencyError): 1L3PT

298 client.get("/sync_raise") 1L3PT

299 assert state["/sync_raise"] == "generator raise finalized" 1L3PT

300 assert "/sync_raise" in errors 1L3PT

301 errors.clear() 1L3PT

302 

303 

304def test_sync_raise_server_error(): 1abcd

305 client = TestClient(app, raise_server_exceptions=False) 1v@BH

306 response = client.get("/sync_raise") 1v@BH

307 assert response.status_code == 500, response.text 1v@BH

308 assert state["/sync_raise"] == "generator raise finalized" 1v@BH

309 assert "/sync_raise" in errors 1v@BH

310 errors.clear() 1v@BH

311 

312 

313def test_sync_async_state(): 1abcd

314 response = client.get("/sync_async") 2+ bb- /

315 assert response.status_code == 200, response.text 2+ bb- /

316 assert response.json() == "asyncgen started" 2+ bb- /

317 assert state["/async"] == "asyncgen completed" 2+ bb- /

318 

319 

320def test_sync_sync_state(): 1abcd

321 response = client.get("/sync_sync") 2, cb. :

322 assert response.status_code == 200, response.text 2, cb. :

323 assert response.json() == "generator started" 2, cb. :

324 assert state["/sync"] == "generator completed" 2, cb. :

325 

326 

327def test_sync_async_raise_other(): 1abcd

328 with pytest.raises(OtherDependencyError): 15~#(

329 client.get("/sync_async_raise_other") 15~#(

330 assert state["/async_raise"] == "asyncgen raise finalized" 15~#(

331 assert "/async_raise" not in errors 15~#(

332 

333 

334def test_sync_sync_raise_other(): 1abcd

335 with pytest.raises(OtherDependencyError): 27 ab% *

336 client.get("/sync_sync_raise_other") 27 ab% *

337 assert state["/sync_raise"] == "generator raise finalized" 27 ab% *

338 assert "/sync_raise" not in errors 27 ab% *

339 

340 

341def test_sync_async_raise_raises(): 1abcd

342 with pytest.raises(AsyncDependencyError): 1K{OS

343 client.get("/sync_async_raise") 1K{OS

344 assert state["/async_raise"] == "asyncgen raise finalized" 1K{OS

345 assert "/async_raise" in errors 1K{OS

346 errors.clear() 1K{OS

347 

348 

349def test_sync_async_raise_server_error(): 1abcd

350 client = TestClient(app, raise_server_exceptions=False) 1t_zF

351 response = client.get("/sync_async_raise") 1t_zF

352 assert response.status_code == 500, response.text 1t_zF

353 assert state["/async_raise"] == "asyncgen raise finalized" 1t_zF

354 assert "/async_raise" in errors 1t_zF

355 errors.clear() 1t_zF

356 

357 

358def test_sync_sync_raise_raises(): 1abcd

359 with pytest.raises(SyncDependencyError): 1M|QU

360 client.get("/sync_sync_raise") 1M|QU

361 assert state["/sync_raise"] == "generator raise finalized" 1M|QU

362 assert "/sync_raise" in errors 1M|QU

363 errors.clear() 1M|QU

364 

365 

366def test_sync_sync_raise_server_error(): 1abcd

367 client = TestClient(app, raise_server_exceptions=False) 1w`CI

368 response = client.get("/sync_sync_raise") 1w`CI

369 assert response.status_code == 500, response.text 1w`CI

370 assert state["/sync_raise"] == "generator raise finalized" 1w`CI

371 assert "/sync_raise" in errors 1w`CI

372 errors.clear() 1w`CI

373 

374 

375def test_sync_context_b(): 1abcd

376 response = client.get("/sync_context_b") 1m}oq

377 data = response.json() 1m}oq

378 assert data["context_b"] == "started b" 1m}oq

379 assert data["context_a"] == "started a" 1m}oq

380 assert state["context_b"] == "finished b with a: started a" 1m}oq

381 assert state["context_a"] == "finished a" 1m}oq

382 

383 

384def test_sync_context_b_raise(): 1abcd

385 with pytest.raises(OtherDependencyError): 1u^AG

386 client.get("/sync_context_b_raise") 1u^AG

387 assert state["context_b"] == "finished b with a: started a" 1u^AG

388 assert state["context_a"] == "finished a" 1u^AG

389 

390 

391def test_sync_background_tasks(): 1abcd

392 response = client.get("/sync_context_b_bg") 1h?ij

393 data = response.json() 1h?ij

394 assert data["context_b"] == "started b" 1h?ij

395 assert data["context_a"] == "started a" 1h?ij

396 assert data["sync_bg"] == "not set" 1h?ij

397 assert state["context_b"] == "finished b with a: started a" 1h?ij

398 assert state["context_a"] == "finished a" 1h?ij

399 assert state["sync_bg"] == "sync_bg set - b: started b - a: started a" 1h?ij