Coverage for tests/test_dependency_contextmanager.py: 100%

1import json 5 ctx1abcde

2from typing import Dict 5 ctx1abcde

3 

4import pytest 5 ctx1abcde

5from fastapi import BackgroundTasks, Depends, FastAPI 5 ctx1abcde

6from fastapi.responses import StreamingResponse 5 ctx1abcde

7from fastapi.testclient import TestClient 5 ctx1abcde

8 

9app = FastAPI() 5 ctx1abcde

10state = { 5 ctx1abcde

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 = [] 5 ctx1abcde

22 

23 

24async def get_state(): 5 ctx1abcde

25 return state 5 ctx1abcde

26 

27 

28class AsyncDependencyError(Exception): 5 ctx1abcde

29 pass 5 ctx1abcde

30 

31 

32class SyncDependencyError(Exception): 5 ctx1abcde

33 pass 5 ctx1abcde

34 

35 

36class OtherDependencyError(Exception): 5 ctx1abcde

37 pass 5 ctx1abcde

38 

39 

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

41 state["/async"] = "asyncgen started" 5 ctx1abcde

42 yield state["/async"] 5 ctx1abcde

43 state["/async"] = "asyncgen completed" 5 ctx1abcde

44 

45 

46def generator_state(state: Dict[str, str] = Depends(get_state)): 5 ctx1abcde

47 state["/sync"] = "generator started" 5 ctx1abcde

48 yield state["/sync"] 5 ctx1abcde

49 state["/sync"] = "generator completed" 5 ctx1abcde

50 

51 

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

53 state["/async_raise"] = "asyncgen raise started" 5 ctx1abcde

54 try: 5 ctx1abcde

55 yield state["/async_raise"] 5 ctx1abcde

56 except AsyncDependencyError: 5 ctx1abcde

57 errors.append("/async_raise") 5 ctx1abcde

58 raise 5 ctx1abcde

59 finally: 

60 state["/async_raise"] = "asyncgen raise finalized" 5 ctx1abcde

61 

62 

63def generator_state_try(state: Dict[str, str] = Depends(get_state)): 5 ctx1abcde

64 state["/sync_raise"] = "generator raise started" 5 ctx1abcde

65 try: 5 ctx1abcde

66 yield state["/sync_raise"] 5 ctx1abcde

67 except SyncDependencyError: 5 ctx1abcde

68 errors.append("/sync_raise") 5 ctx1abcde

69 raise 5 ctx1abcde

70 finally: 

71 state["/sync_raise"] = "generator raise finalized" 5 ctx1abcde

72 

73 

74async def context_a(state: dict = Depends(get_state)): 5 ctx1abcde

75 state["context_a"] = "started a" 5 ctx1abcde

76 try: 5 ctx1abcde

77 yield state 5 ctx1abcde

78 finally: 

79 state["context_a"] = "finished a" 5 ctx1abcde

80 

81 

82async def context_b(state: dict = Depends(context_a)): 5 ctx1abcde

83 state["context_b"] = "started b" 5 ctx1abcde

84 try: 5 ctx1abcde

85 yield state 5 ctx1abcde

86 finally: 

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

88 

89 

90@app.get("/async") 5 ctx1abcde

91async def get_async(state: str = Depends(asyncgen_state)): 5 ctx1abcde

92 return state 5 ctx1abcde

93 

94 

95@app.get("/sync") 5 ctx1abcde

96async def get_sync(state: str = Depends(generator_state)): 5 ctx1abcde

97 return state 5 ctx1abcde

98 

99 

100@app.get("/async_raise") 5 ctx1abcde

101async def get_async_raise(state: str = Depends(asyncgen_state_try)): 5 ctx1abcde

102 assert state == "asyncgen raise started" 5 ctx1abcde

103 raise AsyncDependencyError() 5 ctx1abcde

104 

105 

106@app.get("/sync_raise") 5 ctx1abcde

107async def get_sync_raise(state: str = Depends(generator_state_try)): 5 ctx1abcde

108 assert state == "generator raise started" 5 ctx1abcde

109 raise SyncDependencyError() 5 ctx1abcde

110 

111 

112@app.get("/async_raise_other") 5 ctx1abcde

113async def get_async_raise_other(state: str = Depends(asyncgen_state_try)): 5 ctx1abcde

114 assert state == "asyncgen raise started" 5 ctx1abcde

115 raise OtherDependencyError() 5 ctx1abcde

116 

117 

118@app.get("/sync_raise_other") 5 ctx1abcde

119async def get_sync_raise_other(state: str = Depends(generator_state_try)): 5 ctx1abcde

120 assert state == "generator raise started" 5 ctx1abcde

121 raise OtherDependencyError() 5 ctx1abcde

122 

123 

124@app.get("/context_b") 5 ctx1abcde

125async def get_context_b(state: dict = Depends(context_b)): 5 ctx1abcde

126 return state 5 ctx1abcde

127 

128 

129@app.get("/context_b_raise") 5 ctx1abcde

130async def get_context_b_raise(state: dict = Depends(context_b)): 5 ctx1abcde

131 assert state["context_b"] == "started b" 5 ctx1abcde

132 assert state["context_a"] == "started a" 5 ctx1abcde

133 raise OtherDependencyError() 5 ctx1abcde

134 

135 

136@app.get("/context_b_bg") 5 ctx1abcde

137async def get_context_b_bg(tasks: BackgroundTasks, state: dict = Depends(context_b)): 5 ctx1abcde

138 async def bg(state: dict): 5 ctx1abcde

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

140 

141 tasks.add_task(bg, state) 5 ctx1abcde

142 return state 5 ctx1abcde

143 

144 

145# Sync versions 

146 

147 

148@app.get("/sync_async") 5 ctx1abcde

149def get_sync_async(state: str = Depends(asyncgen_state)): 5 ctx1abcde

150 return state 5 ctx1abcde

151 

152 

153@app.get("/sync_sync") 5 ctx1abcde

154def get_sync_sync(state: str = Depends(generator_state)): 5 ctx1abcde

155 return state 5 ctx1abcde

156 

157 

158@app.get("/sync_async_raise") 5 ctx1abcde

159def get_sync_async_raise(state: str = Depends(asyncgen_state_try)): 5 ctx1abcde

160 assert state == "asyncgen raise started" 5 ctx1abcde

161 raise AsyncDependencyError() 5 ctx1abcde

162 

163 

164@app.get("/sync_sync_raise") 5 ctx1abcde

165def get_sync_sync_raise(state: str = Depends(generator_state_try)): 5 ctx1abcde

166 assert state == "generator raise started" 5 ctx1abcde

167 raise SyncDependencyError() 5 ctx1abcde

168 

169 

170@app.get("/sync_async_raise_other") 5 ctx1abcde

171def get_sync_async_raise_other(state: str = Depends(asyncgen_state_try)): 5 ctx1abcde

172 assert state == "asyncgen raise started" 5 ctx1abcde

173 raise OtherDependencyError() 5 ctx1abcde

174 

175 

176@app.get("/sync_sync_raise_other") 5 ctx1abcde

177def get_sync_sync_raise_other(state: str = Depends(generator_state_try)): 5 ctx1abcde

178 assert state == "generator raise started" 5 ctx1abcde

179 raise OtherDependencyError() 5 ctx1abcde

180 

181 

182@app.get("/sync_context_b") 5 ctx1abcde

183def get_sync_context_b(state: dict = Depends(context_b)): 5 ctx1abcde

184 return state 5 ctx1abcde

185 

186 

187@app.get("/sync_context_b_raise") 5 ctx1abcde

188def get_sync_context_b_raise(state: dict = Depends(context_b)): 5 ctx1abcde

189 assert state["context_b"] == "started b" 5 ctx1abcde

190 assert state["context_a"] == "started a" 5 ctx1abcde

191 raise OtherDependencyError() 5 ctx1abcde

192 

193 

194@app.get("/sync_context_b_bg") 5 ctx1abcde

195async def get_sync_context_b_bg( 5 ctx1abcde

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

197): 

198 async def bg(state: dict): 5 ctx1abcde

199 state[ 5 ctx1abcde

200 "sync_bg" 

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

202 

203 tasks.add_task(bg, state) 5 ctx1abcde

204 return state 5 ctx1abcde

205 

206 

207@app.middleware("http") 5 ctx1abcde

208async def middleware(request, call_next): 5 ctx1abcde

209 response: StreamingResponse = await call_next(request) 5 ctx1abcde

210 response.headers["x-state"] = json.dumps(state.copy()) 5 ctx1abcde

211 return response 5 ctx1abcde

212 

213 

214client = TestClient(app) 5 ctx1abcde

215 

216 

217def test_async_state(): 5 ctx1abcde

218 assert state["/async"] == "asyncgen not started" 5 ctx1abcde

219 response = client.get("/async") 5 ctx1abcde

220 assert response.status_code == 200, response.text 5 ctx1abcde

221 assert response.json() == "asyncgen started" 5 ctx1abcde

222 assert state["/async"] == "asyncgen completed" 5 ctx1abcde

223 

224 

225def test_sync_state(): 5 ctx1abcde

226 assert state["/sync"] == "generator not started" 5 ctx1abcde

227 response = client.get("/sync") 5 ctx1abcde

228 assert response.status_code == 200, response.text 5 ctx1abcde

229 assert response.json() == "generator started" 5 ctx1abcde

230 assert state["/sync"] == "generator completed" 5 ctx1abcde

231 

232 

233def test_async_raise_other(): 5 ctx1abcde

234 assert state["/async_raise"] == "asyncgen raise not started" 5 ctx1abcde

235 with pytest.raises(OtherDependencyError): 5 ctx1abcde

236 client.get("/async_raise_other") 5 ctx1abcde

237 assert state["/async_raise"] == "asyncgen raise finalized" 5 ctx1abcde

238 assert "/async_raise" not in errors 5 ctx1abcde

239 

240 

241def test_sync_raise_other(): 5 ctx1abcde

242 assert state["/sync_raise"] == "generator raise not started" 5 ctx1abcde

243 with pytest.raises(OtherDependencyError): 5 ctx1abcde

244 client.get("/sync_raise_other") 5 ctx1abcde

245 assert state["/sync_raise"] == "generator raise finalized" 5 ctx1abcde

246 assert "/sync_raise" not in errors 5 ctx1abcde

247 

248 

249def test_async_raise_raises(): 5 ctx1abcde

250 with pytest.raises(AsyncDependencyError): 5 ctx1abcde

251 client.get("/async_raise") 5 ctx1abcde

252 assert state["/async_raise"] == "asyncgen raise finalized" 5 ctx1abcde

253 assert "/async_raise" in errors 5 ctx1abcde

254 errors.clear() 5 ctx1abcde

255 

256 

257def test_async_raise_server_error(): 5 ctx1abcde

258 client = TestClient(app, raise_server_exceptions=False) 5 ctx1abcde

259 response = client.get("/async_raise") 5 ctx1abcde

260 assert response.status_code == 500, response.text 5 ctx1abcde

261 assert state["/async_raise"] == "asyncgen raise finalized" 5 ctx1abcde

262 assert "/async_raise" in errors 5 ctx1abcde

263 errors.clear() 5 ctx1abcde

264 

265 

266def test_context_b(): 5 ctx1abcde

267 response = client.get("/context_b") 5 ctx1abcde

268 data = response.json() 5 ctx1abcde

269 assert data["context_b"] == "started b" 5 ctx1abcde

270 assert data["context_a"] == "started a" 5 ctx1abcde

271 assert state["context_b"] == "finished b with a: started a" 5 ctx1abcde

272 assert state["context_a"] == "finished a" 5 ctx1abcde

273 

274 

275def test_context_b_raise(): 5 ctx1abcde

276 with pytest.raises(OtherDependencyError): 5 ctx1abcde

277 client.get("/context_b_raise") 5 ctx1abcde

278 assert state["context_b"] == "finished b with a: started a" 5 ctx1abcde

279 assert state["context_a"] == "finished a" 5 ctx1abcde

280 

281 

282def test_background_tasks(): 5 ctx1abcde

283 response = client.get("/context_b_bg") 5 ctx1abcde

284 data = response.json() 5 ctx1abcde

285 assert data["context_b"] == "started b" 5 ctx1abcde

286 assert data["context_a"] == "started a" 5 ctx1abcde

287 assert data["bg"] == "not set" 5 ctx1abcde

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

289 assert middleware_state["context_b"] == "finished b with a: started a" 5 ctx1abcde

290 assert middleware_state["context_a"] == "finished a" 5 ctx1abcde

291 assert middleware_state["bg"] == "not set" 5 ctx1abcde

292 assert state["context_b"] == "finished b with a: started a" 5 ctx1abcde

293 assert state["context_a"] == "finished a" 5 ctx1abcde

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

295 

296 

297def test_sync_raise_raises(): 5 ctx1abcde

298 with pytest.raises(SyncDependencyError): 5 ctx1abcde

299 client.get("/sync_raise") 5 ctx1abcde

300 assert state["/sync_raise"] == "generator raise finalized" 5 ctx1abcde

301 assert "/sync_raise" in errors 5 ctx1abcde

302 errors.clear() 5 ctx1abcde

303 

304 

305def test_sync_raise_server_error(): 5 ctx1abcde

306 client = TestClient(app, raise_server_exceptions=False) 5 ctx1abcde

307 response = client.get("/sync_raise") 5 ctx1abcde

308 assert response.status_code == 500, response.text 5 ctx1abcde

309 assert state["/sync_raise"] == "generator raise finalized" 5 ctx1abcde

310 assert "/sync_raise" in errors 5 ctx1abcde

311 errors.clear() 5 ctx1abcde

312 

313 

314def test_sync_async_state(): 5 ctx1abcde

315 response = client.get("/sync_async") 5 ctx1abcde

316 assert response.status_code == 200, response.text 5 ctx1abcde

317 assert response.json() == "asyncgen started" 5 ctx1abcde

318 assert state["/async"] == "asyncgen completed" 5 ctx1abcde

319 

320 

321def test_sync_sync_state(): 5 ctx1abcde

322 response = client.get("/sync_sync") 5 ctx1abcde

323 assert response.status_code == 200, response.text 5 ctx1abcde

324 assert response.json() == "generator started" 5 ctx1abcde

325 assert state["/sync"] == "generator completed" 5 ctx1abcde

326 

327 

328def test_sync_async_raise_other(): 5 ctx1abcde

329 with pytest.raises(OtherDependencyError): 5 ctx1abcde

330 client.get("/sync_async_raise_other") 5 ctx1abcde

331 assert state["/async_raise"] == "asyncgen raise finalized" 5 ctx1abcde

332 assert "/async_raise" not in errors 5 ctx1abcde

333 

334 

335def test_sync_sync_raise_other(): 5 ctx1abcde

336 with pytest.raises(OtherDependencyError): 5 ctx1abcde

337 client.get("/sync_sync_raise_other") 5 ctx1abcde

338 assert state["/sync_raise"] == "generator raise finalized" 5 ctx1abcde

339 assert "/sync_raise" not in errors 5 ctx1abcde

340 

341 

342def test_sync_async_raise_raises(): 5 ctx1abcde

343 with pytest.raises(AsyncDependencyError): 5 ctx1abcde

344 client.get("/sync_async_raise") 5 ctx1abcde

345 assert state["/async_raise"] == "asyncgen raise finalized" 5 ctx1abcde

346 assert "/async_raise" in errors 5 ctx1abcde

347 errors.clear() 5 ctx1abcde

348 

349 

350def test_sync_async_raise_server_error(): 5 ctx1abcde

351 client = TestClient(app, raise_server_exceptions=False) 5 ctx1abcde

352 response = client.get("/sync_async_raise") 5 ctx1abcde

353 assert response.status_code == 500, response.text 5 ctx1abcde

354 assert state["/async_raise"] == "asyncgen raise finalized" 5 ctx1abcde

355 assert "/async_raise" in errors 5 ctx1abcde

356 errors.clear() 5 ctx1abcde

357 

358 

359def test_sync_sync_raise_raises(): 5 ctx1abcde

360 with pytest.raises(SyncDependencyError): 5 ctx1abcde

361 client.get("/sync_sync_raise") 5 ctx1abcde

362 assert state["/sync_raise"] == "generator raise finalized" 5 ctx1abcde

363 assert "/sync_raise" in errors 5 ctx1abcde

364 errors.clear() 5 ctx1abcde

365 

366 

367def test_sync_sync_raise_server_error(): 5 ctx1abcde

368 client = TestClient(app, raise_server_exceptions=False) 5 ctx1abcde

369 response = client.get("/sync_sync_raise") 5 ctx1abcde

370 assert response.status_code == 500, response.text 5 ctx1abcde

371 assert state["/sync_raise"] == "generator raise finalized" 5 ctx1abcde

372 assert "/sync_raise" in errors 5 ctx1abcde

373 errors.clear() 5 ctx1abcde

374 

375 

376def test_sync_context_b(): 5 ctx1abcde

377 response = client.get("/sync_context_b") 5 ctx1abcde

378 data = response.json() 5 ctx1abcde

379 assert data["context_b"] == "started b" 5 ctx1abcde

380 assert data["context_a"] == "started a" 5 ctx1abcde

381 assert state["context_b"] == "finished b with a: started a" 5 ctx1abcde

382 assert state["context_a"] == "finished a" 5 ctx1abcde

383 

384 

385def test_sync_context_b_raise(): 5 ctx1abcde

386 with pytest.raises(OtherDependencyError): 5 ctx1abcde

387 client.get("/sync_context_b_raise") 5 ctx1abcde

388 assert state["context_b"] == "finished b with a: started a" 5 ctx1abcde

389 assert state["context_a"] == "finished a" 5 ctx1abcde

390 

391 

392def test_sync_background_tasks(): 5 ctx1abcde

393 response = client.get("/sync_context_b_bg") 5 ctx1abcde

394 data = response.json() 5 ctx1abcde

395 assert data["context_b"] == "started b" 5 ctx1abcde

396 assert data["context_a"] == "started a" 5 ctx1abcde

397 assert data["sync_bg"] == "not set" 5 ctx1abcde

398 assert state["context_b"] == "finished b with a: started a" 5 ctx1abcde

399 assert state["context_a"] == "finished a" 5 ctx1abcde

400 assert ( 5 ctx1abcde

401 state["sync_bg"] 

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

403 )