Coverage for tests/test_dependency_contextmanager.py: 100%

271 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2025-05-05 00:03 +0000

1import json 1abcdef

2from typing import Dict 1abcdef

3 

4import pytest 1abcdef

5from fastapi import BackgroundTasks, Depends, FastAPI 1abcdef

6from fastapi.responses import StreamingResponse 1abcdef

7from fastapi.testclient import TestClient 1abcdef

8 

9app = FastAPI() 1abcdef

10state = { 1abcdef

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

22 

23 

24async def get_state(): 1abcdef

25 return state 2bb! E nbg s F ob# G Lbm t H cb$ I pbqb% J Mbdb| ' rbh u K sb} ( Nbn v L eb~ ) tbubab* Obfb+ M vbi w N wb, O Pbo x P gb- Q xbyb. R Qbhb/ S zbj y T Ab: U Rbp z V ib; W BbCb= X Sbjb? Y Dbk A Z Eb@ 0 Tbq B 1 kb[ 2 FbGb] 3 Ublb^ 4 Hbl C 5 Ib_ 6 Vbr D 7 mb` 8 JbKb{ 9 Wb

26 

27 

28class AsyncDependencyError(Exception): 1abcdef

29 pass 1abcdef

30 

31 

32class SyncDependencyError(Exception): 1abcdef

33 pass 1abcdef

34 

35 

36class OtherDependencyError(Exception): 1abcdef

37 pass 1abcdef

38 

39 

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

41 state["/async"] = "asyncgen started" 2nbLbrbNbvbPbzbRbDbTbHbVb

42 yield state["/async"] 2nbLbrbNbvbPbzbRbDbTbHbVb

43 state["/async"] = "asyncgen completed" 2nbLbrbNbvbPbzbRbDbTbHbVb

44 

45 

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

47 state["/sync"] = "generator started" 2pbMbtbObxbQbBbSbFbUbJbWb

48 yield state["/sync"] 2pbMbtbObxbQbBbSbFbUbJbWb

49 state["/sync"] = "generator completed" 2pbMbtbObxbQbBbSbFbUbJbWb

50 

51 

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

53 state["/async_raise"] = "asyncgen raise started" 2bb! E ob# G db| ' sb} ( fb+ M wb, O hb/ S Ab: U jb? Y Eb@ 0 lb^ 4 Ib_ 6

54 try: 2bb! E ob# G db| ' sb} ( fb+ M wb, O hb/ S Ab: U jb? Y Eb@ 0 lb^ 4 Ib_ 6

55 yield state["/async_raise"] 2bb! E ob# G db| ' sb} ( fb+ M wb, O hb/ S Ab: U jb? Y Eb@ 0 lb^ 4 Ib_ 6

56 except AsyncDependencyError: 2bb! E ob# G db| ' sb} ( fb+ M wb, O hb/ S Ab: U jb? Y Eb@ 0 lb^ 4 Ib_ 6

57 errors.append("/async_raise") 1!E#G|'}(+M,O/S:U?Y@0^4_6

58 raise 1!E#G|'}(+M,O/S:U?Y@0^4_6

59 finally: 

60 state["/async_raise"] = "asyncgen raise finalized" 2bb! E ob# G dbsbfb+ M wb, O hb/ S Ab: U jb? Y Eb@ 0 lb^ 4 Ib_ 6

61 

62 

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

64 state["/sync_raise"] = "generator raise started" 2cb$ I qb% J eb~ ) ubab* gb- Q yb. R ib; W Cb= X kb[ 2 Gb] 3 mb` 8 Kb{ 9

65 try: 2cb$ I qb% J eb~ ) ubab* gb- Q yb. R ib; W Cb= X kb[ 2 Gb] 3 mb` 8 Kb{ 9

66 yield state["/sync_raise"] 2cb$ I qb% J eb~ ) ubab* gb- Q yb. R ib; W Cb= X kb[ 2 Gb] 3 mb` 8 Kb{ 9

67 except SyncDependencyError: 2cb$ I qb% J eb~ ) ubab* gb- Q yb. R ib; W Cb= X kb[ 2 Gb] 3 mb` 8 Kb{ 9

68 errors.append("/sync_raise") 2$ I % J ~ ) ab* - Q . R ; W = X [ 2 ] 3 ` 8 { 9

69 raise 2$ I % J ~ ) ab* - Q . R ; W = X [ 2 ] 3 ` 8 { 9

70 finally: 

71 state["/sync_raise"] = "generator raise finalized" 2cb$ I qb% J ebubgb- Q yb. R ib; W Cb= X kb[ 2 Gb] 3 mb` 8 Kb{ 9

72 

73 

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

75 state["context_a"] = "started a" 1gsFmtHhuKnvLiwNoxPjyTpzVkAZqB1lC5rD7

76 try: 1gsFmtHhuKnvLiwNoxPjyTpzVkAZqB1lC5rD7

77 yield state 1gsFmtHhuKnvLiwNoxPjyTpzVkAZqB1lC5rD7

78 finally: 

79 state["context_a"] = "finished a" 1gsFmtHhuKnvLiwNoxPjyTpzVkAZqB1lC5rD7

80 

81 

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

83 state["context_b"] = "started b" 1gsFmtHhuKnvLiwNoxPjyTpzVkAZqB1lC5rD7

84 try: 1gsFmtHhuKnvLiwNoxPjyTpzVkAZqB1lC5rD7

85 yield state 1gsFmtHhuKnvLiwNoxPjyTpzVkAZqB1lC5rD7

86 finally: 

87 state["context_b"] = f"finished b with a: {state['context_a']}" 1gsFmtHhuKnvLiwNoxPjyTpzVkAZqB1lC5rD7

88 

89 

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

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

92 return state 2nbrbvbzbDbHb

93 

94 

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

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

97 return state 2pbtbxbBbFbJb

98 

99 

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

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

102 assert state == "asyncgen raise started" 1!E|'+M/S?Y^4

103 raise AsyncDependencyError() 1!E|'+M/S?Y^4

104 

105 

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

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

108 assert state == "generator raise started" 1$I~)-Q;W[2`8

109 raise SyncDependencyError() 1$I~)-Q;W[2`8

110 

111 

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

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

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

115 raise OtherDependencyError() 2bbdbfbhbjblb

116 

117 

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

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

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

121 raise OtherDependencyError() 2cbebgbibkbmb

122 

123 

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

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

126 return state 1suwyAC

127 

128 

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

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

131 assert state["context_b"] == "started b" 1FKNTZ5

132 assert state["context_a"] == "started a" 1FKNTZ5

133 raise OtherDependencyError() 1FKNTZ5

134 

135 

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

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

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

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

140 

141 tasks.add_task(bg, state) 1ghijkl

142 return state 1ghijkl

143 

144 

145# Sync versions 

146 

147 

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

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

150 return state 2LbNbPbRbTbVb

151 

152 

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

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

155 return state 2MbObQbSbUbWb

156 

157 

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

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

160 assert state == "asyncgen raise started" 1#G}(,O:U@0_6

161 raise AsyncDependencyError() 1#G}(,O:U@0_6

162 

163 

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

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

166 assert state == "generator raise started" 2% J ab* . R = X ] 3 { 9

167 raise SyncDependencyError() 2% J ab* . R = X ] 3 { 9

168 

169 

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

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

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

173 raise OtherDependencyError() 2obsbwbAbEbIb

174 

175 

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

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

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

179 raise OtherDependencyError() 2qbubybCbGbKb

180 

181 

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

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

184 return state 1tvxzBD

185 

186 

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

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

189 assert state["context_b"] == "started b" 1HLPV17

190 assert state["context_a"] == "started a" 1HLPV17

191 raise OtherDependencyError() 1HLPV17

192 

193 

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

195async def get_sync_context_b_bg( 1abcdef

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

197): 

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

199 state["sync_bg"] = ( 1mnopqr

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

201 ) 

202 

203 tasks.add_task(bg, state) 1mnopqr

204 return state 1mnopqr

205 

206 

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

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

209 response: StreamingResponse = await call_next(request) 2bb! E nbg s F ob# G Lbm t H cb$ I pbqb% J Mbdb| ' rbh u K sb} ( Nbn v L eb~ ) tbubab* Obfb+ M vbi w N wb, O Pbo x P gb- Q xbyb. R Qbhb/ S zbj y T Ab: U Rbp z V ib; W BbCb= X Sbjb? Y Dbk A Z Eb@ 0 Tbq B 1 kb[ 2 FbGb] 3 Ublb^ 4 Hbl C 5 Ib_ 6 Vbr D 7 mb` 8 JbKb{ 9 Wb

210 response.headers["x-state"] = json.dumps(state.copy()) 2nbg s Lbm t pbMbrbh u Nbn v tbObvbi w Pbo x xbQbzbj y Rbp z BbSbDbk A Tbq B FbUbHbl C Vbr D JbWb

211 return response 2nbg s Lbm t pbMbrbh u Nbn v tbObvbi w Pbo x xbQbzbj y Rbp z BbSbDbk A Tbq B FbUbHbl C Vbr D JbWb

212 

213 

214client = TestClient(app) 1abcdef

215 

216 

217def test_async_state(): 1abcdef

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

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

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

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

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

223 

224 

225def test_sync_state(): 1abcdef

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

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

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

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

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

231 

232 

233def test_async_raise_other(): 1abcdef

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

235 with pytest.raises(OtherDependencyError): 2bbdbfbhbjblb

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

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

238 assert "/async_raise" not in errors 2bbdbfbhbjblb

239 

240 

241def test_sync_raise_other(): 1abcdef

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

243 with pytest.raises(OtherDependencyError): 2cbebgbibkbmb

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

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

246 assert "/sync_raise" not in errors 2cbebgbibkbmb

247 

248 

249def test_async_raise_raises(): 1abcdef

250 with pytest.raises(AsyncDependencyError): 1!|+/?^

251 client.get("/async_raise") 1!|+/?^

252 assert state["/async_raise"] == "asyncgen raise finalized" 1!|+/?^

253 assert "/async_raise" in errors 1!|+/?^

254 errors.clear() 1!|+/?^

255 

256 

257def test_async_raise_server_error(): 1abcdef

258 client = TestClient(app, raise_server_exceptions=False) 1E'MSY4

259 response = client.get("/async_raise") 1E'MSY4

260 assert response.status_code == 500, response.text 1E'MSY4

261 assert state["/async_raise"] == "asyncgen raise finalized" 1E'MSY4

262 assert "/async_raise" in errors 1E'MSY4

263 errors.clear() 1E'MSY4

264 

265 

266def test_context_b(): 1abcdef

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

268 data = response.json() 1suwyAC

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

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

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

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

273 

274 

275def test_context_b_raise(): 1abcdef

276 with pytest.raises(OtherDependencyError): 1FKNTZ5

277 client.get("/context_b_raise") 1FKNTZ5

278 assert state["context_b"] == "finished b with a: started a" 1FKNTZ5

279 assert state["context_a"] == "finished a" 1FKNTZ5

280 

281 

282def test_background_tasks(): 1abcdef

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

284 data = response.json() 1ghijkl

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

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

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

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

289 assert middleware_state["context_b"] == "finished b with a: started a" 1ghijkl

290 assert middleware_state["context_a"] == "finished a" 1ghijkl

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

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

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

294 assert state["bg"] == "bg set - b: finished b with a: started a - a: finished a" 1ghijkl

295 

296 

297def test_sync_raise_raises(): 1abcdef

298 with pytest.raises(SyncDependencyError): 1$~-;[`

299 client.get("/sync_raise") 1$~-;[`

300 assert state["/sync_raise"] == "generator raise finalized" 1$~-;[`

301 assert "/sync_raise" in errors 1$~-;[`

302 errors.clear() 1$~-;[`

303 

304 

305def test_sync_raise_server_error(): 1abcdef

306 client = TestClient(app, raise_server_exceptions=False) 1I)QW28

307 response = client.get("/sync_raise") 1I)QW28

308 assert response.status_code == 500, response.text 1I)QW28

309 assert state["/sync_raise"] == "generator raise finalized" 1I)QW28

310 assert "/sync_raise" in errors 1I)QW28

311 errors.clear() 1I)QW28

312 

313 

314def test_sync_async_state(): 1abcdef

315 response = client.get("/sync_async") 2LbNbPbRbTbVb

316 assert response.status_code == 200, response.text 2LbNbPbRbTbVb

317 assert response.json() == "asyncgen started" 2LbNbPbRbTbVb

318 assert state["/async"] == "asyncgen completed" 2LbNbPbRbTbVb

319 

320 

321def test_sync_sync_state(): 1abcdef

322 response = client.get("/sync_sync") 2MbObQbSbUbWb

323 assert response.status_code == 200, response.text 2MbObQbSbUbWb

324 assert response.json() == "generator started" 2MbObQbSbUbWb

325 assert state["/sync"] == "generator completed" 2MbObQbSbUbWb

326 

327 

328def test_sync_async_raise_other(): 1abcdef

329 with pytest.raises(OtherDependencyError): 2obsbwbAbEbIb

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

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

332 assert "/async_raise" not in errors 2obsbwbAbEbIb

333 

334 

335def test_sync_sync_raise_other(): 1abcdef

336 with pytest.raises(OtherDependencyError): 2qbubybCbGbKb

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

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

339 assert "/sync_raise" not in errors 2qbubybCbGbKb

340 

341 

342def test_sync_async_raise_raises(): 1abcdef

343 with pytest.raises(AsyncDependencyError): 1#},:@_

344 client.get("/sync_async_raise") 1#},:@_

345 assert state["/async_raise"] == "asyncgen raise finalized" 1#},:@_

346 assert "/async_raise" in errors 1#},:@_

347 errors.clear() 1#},:@_

348 

349 

350def test_sync_async_raise_server_error(): 1abcdef

351 client = TestClient(app, raise_server_exceptions=False) 1G(OU06

352 response = client.get("/sync_async_raise") 1G(OU06

353 assert response.status_code == 500, response.text 1G(OU06

354 assert state["/async_raise"] == "asyncgen raise finalized" 1G(OU06

355 assert "/async_raise" in errors 1G(OU06

356 errors.clear() 1G(OU06

357 

358 

359def test_sync_sync_raise_raises(): 1abcdef

360 with pytest.raises(SyncDependencyError): 2% ab. = ] {

361 client.get("/sync_sync_raise") 2% ab. = ] {

362 assert state["/sync_raise"] == "generator raise finalized" 2% ab. = ] {

363 assert "/sync_raise" in errors 2% ab. = ] {

364 errors.clear() 2% ab. = ] {

365 

366 

367def test_sync_sync_raise_server_error(): 1abcdef

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

369 response = client.get("/sync_sync_raise") 1J*RX39

370 assert response.status_code == 500, response.text 1J*RX39

371 assert state["/sync_raise"] == "generator raise finalized" 1J*RX39

372 assert "/sync_raise" in errors 1J*RX39

373 errors.clear() 1J*RX39

374 

375 

376def test_sync_context_b(): 1abcdef

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

378 data = response.json() 1tvxzBD

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

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

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

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

383 

384 

385def test_sync_context_b_raise(): 1abcdef

386 with pytest.raises(OtherDependencyError): 1HLPV17

387 client.get("/sync_context_b_raise") 1HLPV17

388 assert state["context_b"] == "finished b with a: started a" 1HLPV17

389 assert state["context_a"] == "finished a" 1HLPV17

390 

391 

392def test_sync_background_tasks(): 1abcdef

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

394 data = response.json() 1mnopqr

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

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

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

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

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

400 assert ( 1mnopqr

401 state["sync_bg"] 

402 == "sync_bg set - b: finished b with a: started a - a: finished a" 

403 )