Coverage for tests/test_dependency_yield_scope_websockets.py: 100%

120 statements  

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

1from contextvars import ContextVar 1abcdefg

2from typing import Any, Dict, Tuple 1abcdefg

3 

4import pytest 1abcdefg

5from fastapi import Depends, FastAPI, WebSocket 1abcdefg

6from fastapi.exceptions import FastAPIError 1abcdefg

7from fastapi.testclient import TestClient 1abcdefg

8from typing_extensions import Annotated 1abcdefg

9 

10global_context: ContextVar[Dict[str, Any]] = ContextVar("global_context", default={}) # noqa: B039 1abcdefg

11 

12 

13class Session: 1abcdefg

14 def __init__(self) -> None: 1abcdefg

15 self.open = True 1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

16 

17 

18async def dep_session() -> Any: 1abcdefg

19 s = Session() 1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

20 yield s 1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

21 s.open = False 1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

22 global_state = global_context.get() 1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

23 global_state["session_closed"] = True 1JovKhCLpwMiDNqxOjEPryQkFRszSlGTtAUmHVuBWnI

24 

25 

26SessionFuncDep = Annotated[Session, Depends(dep_session, scope="function")] 1abcdefg

27SessionRequestDep = Annotated[Session, Depends(dep_session, scope="request")] 1abcdefg

28SessionDefaultDep = Annotated[Session, Depends(dep_session)] 1abcdefg

29 

30 

31class NamedSession: 1abcdefg

32 def __init__(self, name: str = "default") -> None: 1abcdefg

33 self.name = name 1ovhpwiqxjrykszltAmuBn

34 self.open = True 1ovhpwiqxjrykszltAmuBn

35 

36 

37def get_named_session(session: SessionRequestDep, session_b: SessionDefaultDep) -> Any: 1abcdefg

38 assert session is session_b 1hijklmn

39 named_session = NamedSession(name="named") 1hijklmn

40 yield named_session, session_b 1hijklmn

41 named_session.open = False 1hijklmn

42 global_state = global_context.get() 1hijklmn

43 global_state["named_session_closed"] = True 1hijklmn

44 

45 

46NamedSessionsDep = Annotated[Tuple[NamedSession, Session], Depends(get_named_session)] 1abcdefg

47 

48 

49def get_named_func_session(session: SessionFuncDep) -> Any: 1abcdefg

50 named_session = NamedSession(name="named") 1opqrstu

51 yield named_session, session 1opqrstu

52 named_session.open = False 1opqrstu

53 global_state = global_context.get() 1opqrstu

54 global_state["named_func_session_closed"] = True 1opqrstu

55 

56 

57def get_named_regular_func_session(session: SessionFuncDep) -> Any: 1abcdefg

58 named_session = NamedSession(name="named") 1vwxyzAB

59 return named_session, session 1vwxyzAB

60 

61 

62BrokenSessionsDep = Annotated[ 1abcdefg

63 Tuple[NamedSession, Session], Depends(get_named_func_session) 

64] 

65NamedSessionsFuncDep = Annotated[ 1abcdefg

66 Tuple[NamedSession, Session], Depends(get_named_func_session, scope="function") 

67] 

68 

69RegularSessionsDep = Annotated[ 1abcdefg

70 Tuple[NamedSession, Session], Depends(get_named_regular_func_session) 

71] 

72 

73app = FastAPI() 1abcdefg

74 

75 

76@app.websocket("/function-scope") 1abcdefg

77async def function_scope(websocket: WebSocket, session: SessionFuncDep) -> Any: 1abcdefg

78 await websocket.accept() 1JLNPRTV

79 await websocket.send_json({"is_open": session.open}) 1JLNPRTV

80 

81 

82@app.websocket("/request-scope") 1abcdefg

83async def request_scope(websocket: WebSocket, session: SessionRequestDep) -> Any: 1abcdefg

84 await websocket.accept() 1KMOQSUW

85 await websocket.send_json({"is_open": session.open}) 1KMOQSUW

86 

87 

88@app.websocket("/two-scopes") 1abcdefg

89async def get_stream_session( 1abcdefg

90 websocket: WebSocket, 

91 function_session: SessionFuncDep, 

92 request_session: SessionRequestDep, 

93) -> Any: 

94 await websocket.accept() 1CDEFGHI

95 await websocket.send_json( 1CDEFGHI

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

97 ) 

98 

99 

100@app.websocket("/sub") 1abcdefg

101async def get_sub(websocket: WebSocket, sessions: NamedSessionsDep) -> Any: 1abcdefg

102 await websocket.accept() 1hijklmn

103 await websocket.send_json( 1hijklmn

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

105 ) 

106 

107 

108@app.websocket("/named-function-scope") 1abcdefg

109async def get_named_function_scope( 1abcdefg

110 websocket: WebSocket, sessions: NamedSessionsFuncDep 

111) -> Any: 

112 await websocket.accept() 1opqrstu

113 await websocket.send_json( 1opqrstu

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

115 ) 

116 

117 

118@app.websocket("/regular-function-scope") 1abcdefg

119async def get_regular_function_scope( 1abcdefg

120 websocket: WebSocket, sessions: RegularSessionsDep 

121) -> Any: 

122 await websocket.accept() 1vwxyzAB

123 await websocket.send_json( 1vwxyzAB

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

125 ) 

126 

127 

128client = TestClient(app) 1abcdefg

129 

130 

131def test_function_scope() -> None: 1abcdefg

132 global_context.set({}) 1JLNPRTV

133 global_state = global_context.get() 1JLNPRTV

134 with client.websocket_connect("/function-scope") as websocket: 1JLNPRTV

135 data = websocket.receive_json() 1JLNPRTV

136 assert data["is_open"] is True 1JLNPRTV

137 assert global_state["session_closed"] is True 1JLNPRTV

138 

139 

140def test_request_scope() -> None: 1abcdefg

141 global_context.set({}) 1KMOQSUW

142 global_state = global_context.get() 1KMOQSUW

143 with client.websocket_connect("/request-scope") as websocket: 1KMOQSUW

144 data = websocket.receive_json() 1KMOQSUW

145 assert data["is_open"] is True 1KMOQSUW

146 assert global_state["session_closed"] is True 1KMOQSUW

147 

148 

149def test_two_scopes() -> None: 1abcdefg

150 global_context.set({}) 1CDEFGHI

151 global_state = global_context.get() 1CDEFGHI

152 with client.websocket_connect("/two-scopes") as websocket: 1CDEFGHI

153 data = websocket.receive_json() 1CDEFGHI

154 assert data["func_is_open"] is True 1CDEFGHI

155 assert data["req_is_open"] is True 1CDEFGHI

156 assert global_state["session_closed"] is True 1CDEFGHI

157 

158 

159def test_sub() -> None: 1abcdefg

160 global_context.set({}) 1hijklmn

161 global_state = global_context.get() 1hijklmn

162 with client.websocket_connect("/sub") as websocket: 1hijklmn

163 data = websocket.receive_json() 1hijklmn

164 assert data["named_session_open"] is True 1hijklmn

165 assert data["session_open"] is True 1hijklmn

166 assert global_state["session_closed"] is True 1hijklmn

167 assert global_state["named_session_closed"] is True 1hijklmn

168 

169 

170def test_broken_scope() -> None: 1abcdefg

171 with pytest.raises( 1XYZ0123

172 FastAPIError, 

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

174 ): 

175 

176 @app.websocket("/broken-scope") 1XYZ0123

177 async def get_broken( 1XYZ0123

178 websocket: WebSocket, sessions: BrokenSessionsDep 1XYZ0123

179 ) -> Any: # pragma: no cover 1XYZ0123

180 pass 

181 

182 

183def test_named_function_scope() -> None: 1abcdefg

184 global_context.set({}) 1opqrstu

185 global_state = global_context.get() 1opqrstu

186 with client.websocket_connect("/named-function-scope") as websocket: 1opqrstu

187 data = websocket.receive_json() 1opqrstu

188 assert data["named_session_open"] is True 1opqrstu

189 assert data["session_open"] is True 1opqrstu

190 assert global_state["session_closed"] is True 1opqrstu

191 assert global_state["named_func_session_closed"] is True 1opqrstu

192 

193 

194def test_regular_function_scope() -> None: 1abcdefg

195 global_context.set({}) 1vwxyzAB

196 global_state = global_context.get() 1vwxyzAB

197 with client.websocket_connect("/regular-function-scope") as websocket: 1vwxyzAB

198 data = websocket.receive_json() 1vwxyzAB

199 assert data["named_session_open"] is True 1vwxyzAB

200 assert data["session_open"] is True 1vwxyzAB

201 assert global_state["session_closed"] is True 1vwxyzAB