Coverage for tests / test_dependency_yield_scope.py: 100%

1import json 4 ctx1abcd

2from typing import Annotated, Any 4 ctx1abcd

3 

4import pytest 4 ctx1abcd

5from fastapi import APIRouter, Depends, FastAPI, HTTPException 4 ctx1abcd

6from fastapi.exceptions import FastAPIError 4 ctx1abcd

7from fastapi.responses import StreamingResponse 4 ctx1abcd

8from fastapi.testclient import TestClient 4 ctx1abcd

9 

10 

11class Session: 4 ctx1abcd

12 def __init__(self) -> None: 4 ctx1abcd

13 self.open = True 18 ctx1qhkrensiltfoujmvgp

14 

15 

16def dep_session() -> Any: 4 ctx1abcd

17 s = Session() 18 ctx1qhkrensiltfoujmvgp

18 yield s 18 ctx1qhkrensiltfoujmvgp

19 s.open = False 18 ctx1qhkrensiltfoujmvgp

20 

21 

22def raise_after_yield() -> Any: 4 ctx1abcd

23 yield 12 ctx1wxFCyzGDABHE

24 raise HTTPException(status_code=503, detail="Exception after yield") 12 ctx1wxFCyzGDABHE

25 

26 

27SessionFuncDep = Annotated[Session, Depends(dep_session, scope="function")] 4 ctx1abcd

28SessionRequestDep = Annotated[Session, Depends(dep_session, scope="request")] 4 ctx1abcd

29SessionDefaultDep = Annotated[Session, Depends(dep_session)] 4 ctx1abcd

30 

31 

32class NamedSession: 4 ctx1abcd

33 def __init__(self, name: str = "default") -> None: 4 ctx1abcd

34 self.name = name 9 ctx1hkeilfjmg

35 self.open = True 9 ctx1hkeilfjmg

36 

37 

38def get_named_session(session: SessionRequestDep, session_b: SessionDefaultDep) -> Any: 4 ctx1abcd

39 assert session is session_b 3 ctx1efg

40 named_session = NamedSession(name="named") 3 ctx1efg

41 yield named_session, session_b 3 ctx1efg

42 named_session.open = False 3 ctx1efg

43 

44 

45NamedSessionsDep = Annotated[tuple[NamedSession, Session], Depends(get_named_session)] 4 ctx1abcd

46 

47 

48def get_named_func_session(session: SessionFuncDep) -> Any: 4 ctx1abcd

49 named_session = NamedSession(name="named") 3 ctx1hij

50 yield named_session, session 3 ctx1hij

51 named_session.open = False 3 ctx1hij

52 

53 

54def get_named_regular_func_session(session: SessionFuncDep) -> Any: 4 ctx1abcd

55 named_session = NamedSession(name="named") 3 ctx1klm

56 return named_session, session 3 ctx1klm

57 

58 

59BrokenSessionsDep = Annotated[ 4 ctx1abcd

60 tuple[NamedSession, Session], Depends(get_named_func_session) 

61] 

62NamedSessionsFuncDep = Annotated[ 4 ctx1abcd

63 tuple[NamedSession, Session], Depends(get_named_func_session, scope="function") 

64] 

65 

66RegularSessionsDep = Annotated[ 4 ctx1abcd

67 tuple[NamedSession, Session], Depends(get_named_regular_func_session) 

68] 

69 

70app = FastAPI() 4 ctx1abcd

71router = APIRouter() 4 ctx1abcd

72 

73 

74@router.get("/") 4 ctx1abcd

75def get_index(): 4 ctx1abcd

76 return {"status": "ok"} 6 ctx1FCGDHE

77 

78 

79@app.get("/function-scope") 4 ctx1abcd

80def function_scope(session: SessionFuncDep) -> Any: 4 ctx1abcd

81 def iter_data(): 3 ctx1qsu

82 yield json.dumps({"is_open": session.open}) 3 ctx1qsu

83 

84 return StreamingResponse(iter_data()) 3 ctx1qsu

85 

86 

87@app.get("/request-scope") 4 ctx1abcd

88def request_scope(session: SessionRequestDep) -> Any: 4 ctx1abcd

89 def iter_data(): 3 ctx1rtv

90 yield json.dumps({"is_open": session.open}) 3 ctx1rtv

91 

92 return StreamingResponse(iter_data()) 3 ctx1rtv

93 

94 

95@app.get("/two-scopes") 4 ctx1abcd

96def get_stream_session( 4 ctx1abcd

97 function_session: SessionFuncDep, request_session: SessionRequestDep 

98) -> Any: 

99 def iter_data(): 3 ctx1nop

100 yield json.dumps( 3 ctx1nop

101 {"func_is_open": function_session.open, "req_is_open": request_session.open} 

102 ) 

103 

104 return StreamingResponse(iter_data()) 3 ctx1nop

105 

106 

107@app.get("/sub") 4 ctx1abcd

108def get_sub(sessions: NamedSessionsDep) -> Any: 4 ctx1abcd

109 def iter_data(): 3 ctx1efg

110 yield json.dumps( 3 ctx1efg

111 {"named_session_open": sessions[0].open, "session_open": sessions[1].open} 

112 ) 

113 

114 return StreamingResponse(iter_data()) 3 ctx1efg

115 

116 

117@app.get("/named-function-scope") 4 ctx1abcd

118def get_named_function_scope(sessions: NamedSessionsFuncDep) -> Any: 4 ctx1abcd

119 def iter_data(): 3 ctx1hij

120 yield json.dumps( 3 ctx1hij

121 {"named_session_open": sessions[0].open, "session_open": sessions[1].open} 

122 ) 

123 

124 return StreamingResponse(iter_data()) 3 ctx1hij

125 

126 

127@app.get("/regular-function-scope") 4 ctx1abcd

128def get_regular_function_scope(sessions: RegularSessionsDep) -> Any: 4 ctx1abcd

129 def iter_data(): 3 ctx1klm

130 yield json.dumps( 3 ctx1klm

131 {"named_session_open": sessions[0].open, "session_open": sessions[1].open} 

132 ) 

133 

134 return StreamingResponse(iter_data()) 3 ctx1klm

135 

136 

137app.include_router( 4 ctx1abcd

138 prefix="/router-scope-function", 

139 router=router, 

140 dependencies=[Depends(raise_after_yield, scope="function")], 

141) 

142 

143app.include_router( 4 ctx1abcd

144 prefix="/router-scope-request", 

145 router=router, 

146 dependencies=[Depends(raise_after_yield, scope="request")], 

147) 

148 

149client = TestClient(app) 4 ctx1abcd

150 

151 

152def test_function_scope() -> None: 4 ctx1abcd

153 response = client.get("/function-scope") 3 ctx1qsu

154 assert response.status_code == 200 3 ctx1qsu

155 data = response.json() 3 ctx1qsu

156 assert data["is_open"] is False 3 ctx1qsu

157 

158 

159def test_request_scope() -> None: 4 ctx1abcd

160 response = client.get("/request-scope") 3 ctx1rtv

161 assert response.status_code == 200 3 ctx1rtv

162 data = response.json() 3 ctx1rtv

163 assert data["is_open"] is True 3 ctx1rtv

164 

165 

166def test_two_scopes() -> None: 4 ctx1abcd

167 response = client.get("/two-scopes") 3 ctx1nop

168 assert response.status_code == 200 3 ctx1nop

169 data = response.json() 3 ctx1nop

170 assert data["func_is_open"] is False 3 ctx1nop

171 assert data["req_is_open"] is True 3 ctx1nop

172 

173 

174def test_sub() -> None: 4 ctx1abcd

175 response = client.get("/sub") 3 ctx1efg

176 assert response.status_code == 200 3 ctx1efg

177 data = response.json() 3 ctx1efg

178 assert data["named_session_open"] is True 3 ctx1efg

179 assert data["session_open"] is True 3 ctx1efg

180 

181 

182def test_broken_scope() -> None: 4 ctx1abcd

183 with pytest.raises( 3 ctx1IJK

184 FastAPIError, 

185 match='The dependency "get_named_func_session" has a scope of "request", it cannot depend on dependencies with scope "function"', 

186 ): 

187 

188 @app.get("/broken-scope") 3 ctx1IJK

189 def get_broken(sessions: BrokenSessionsDep) -> Any: # pragma: no cover 3 ctx1IJK

190 pass 

191 

192 

193def test_named_function_scope() -> None: 4 ctx1abcd

194 response = client.get("/named-function-scope") 3 ctx1hij

195 assert response.status_code == 200 3 ctx1hij

196 data = response.json() 3 ctx1hij

197 assert data["named_session_open"] is False 3 ctx1hij

198 assert data["session_open"] is False 3 ctx1hij

199 

200 

201def test_regular_function_scope() -> None: 4 ctx1abcd

202 response = client.get("/regular-function-scope") 3 ctx1klm

203 assert response.status_code == 200 3 ctx1klm

204 data = response.json() 3 ctx1klm

205 assert data["named_session_open"] is True 3 ctx1klm

206 assert data["session_open"] is False 3 ctx1klm

207 

208 

209def test_router_level_dep_scope_function() -> None: 4 ctx1abcd

210 response = client.get("/router-scope-function/") 3 ctx1FGH

211 assert response.status_code == 503 3 ctx1FGH

212 assert response.json() == {"detail": "Exception after yield"} 3 ctx1FGH

213 

214 

215def test_router_level_dep_scope_request() -> None: 4 ctx1abcd

216 with TestClient(app, raise_server_exceptions=False) as client: 3 ctx1CDE

217 response = client.get("/router-scope-request/") 3 ctx1CDE

218 assert response.status_code == 200 3 ctx1CDE

219 assert response.json() == {"status": "ok"} 3 ctx1CDE

220 

221 

222def test_app_level_dep_scope_function() -> None: 4 ctx1abcd

223 app = FastAPI(dependencies=[Depends(raise_after_yield, scope="function")]) 3 ctx1wyA

224 

225 @app.get("/app-scope-function") 3 ctx1wyA

226 def get_app_scope_function(): 3 ctx1wyA

227 return {"status": "ok"} 3 ctx1wyA

228 

229 with TestClient(app) as client: 3 ctx1wyA

230 response = client.get("/app-scope-function") 3 ctx1wyA

231 assert response.status_code == 503 3 ctx1wyA

232 assert response.json() == {"detail": "Exception after yield"} 3 ctx1wyA

233 

234 

235def test_app_level_dep_scope_request() -> None: 4 ctx1abcd

236 app = FastAPI(dependencies=[Depends(raise_after_yield, scope="request")]) 3 ctx1xzB

237 

238 @app.get("/app-scope-request") 3 ctx1xzB

239 def get_app_scope_request(): 3 ctx1xzB

240 return {"status": "ok"} 3 ctx1xzB

241 

242 with TestClient(app, raise_server_exceptions=False) as client: 3 ctx1xzB

243 response = client.get("/app-scope-request") 3 ctx1xzB

244 assert response.status_code == 200 3 ctx1xzB

245 assert response.json() == {"status": "ok"} 3 ctx1xzB