Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/requests_mock/mocker.py: 66%

1# Licensed under the Apache License, Version 2.0 (the "License"); you may 

2# not use this file except in compliance with the License. You may obtain 

3# a copy of the License at 

4# 

5# https://www.apache.org/licenses/LICENSE-2.0 

6# 

7# Unless required by applicable law or agreed to in writing, software 

8# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

9# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

10# License for the specific language governing permissions and limitations 

11# under the License. 

12 

13import contextlib 

14import functools 

15import sys 

16import threading 

17import types 

18 

19import requests 

20 

21from requests_mock import adapter 

22from requests_mock import exceptions 

23 

24DELETE = 'DELETE' 

25GET = 'GET' 

26HEAD = 'HEAD' 

27OPTIONS = 'OPTIONS' 

28PATCH = 'PATCH' 

29POST = 'POST' 

30PUT = 'PUT' 

31 

32_original_send = requests.Session.send 

33 

34# NOTE(phodge): we need to use an RLock (reentrant lock) here because 

35# requests.Session.send() is reentrant. See further comments where we 

36# monkeypatch get_adapter() 

37_send_lock = threading.RLock() 

38 

39 

40@contextlib.contextmanager 

41def threading_rlock(timeout): 

42 kwargs = {} 

43 if sys.version_info.major >= 3: 

44 # python2 doesn't support the timeout argument 

45 kwargs['timeout'] = timeout 

46 

47 if not _send_lock.acquire(**kwargs): 

48 m = "Could not acquire threading lock - possible deadlock scenario" 

49 raise Exception(m) 

50 

51 try: 

52 yield 

53 finally: 

54 _send_lock.release() 

55 

56 

57def _is_bound_method(method): 

58 """ 

59 bound_method 's self is a obj 

60 unbound_method 's self is None 

61 """ 

62 if isinstance(method, types.MethodType) and hasattr(method, '__self__'): 

63 return True 

64 

65 return False 

66 

67 

68def _set_method(target, name, method): 

69 """ Set a mocked method onto the target. 

70 

71 Target may be either an instance of a Session object of the 

72 requests.Session class. First we Bind the method if it's an instance. 

73 

74 If method is a bound_method, can direct setattr 

75 """ 

76 if not isinstance(target, type) and not _is_bound_method(method): 

77 method = types.MethodType(method, target) 

78 

79 setattr(target, name, method) 

80 

81 

82class MockerCore(object): 

83 """A wrapper around common mocking functions. 

84 

85 Automate the process of mocking the requests library. This will keep the 

86 same general options available and prevent repeating code. 

87 """ 

88 

89 _PROXY_FUNCS = { 

90 'last_request', 

91 'add_matcher', 

92 'request_history', 

93 'called', 

94 'called_once', 

95 'call_count', 

96 'reset', 

97 } 

98 

99 case_sensitive = False 

100 """case_sensitive handles a backwards incompatible bug. The URL used to 

101 match against our matches and that is saved in request_history is always 

102 lowercased. This is incorrect as it reports incorrect history to the user 

103 and doesn't allow case sensitive path matching. 

104 

105 Unfortunately fixing this change is backwards incompatible in the 1.X 

106 series as people may rely on this behaviour. To work around this you can 

107 globally set: 

108 

109 requests_mock.mock.case_sensitive = True 

110 

111 or for pytest set in your configuration: 

112 

113 [pytest] 

114 requests_mock_case_sensitive = True 

115 

116 which will prevent the lowercase being executed and return case sensitive 

117 url and query information. 

118 

119 This will become the default in a 2.X release. See bug: #1584008. 

120 """ 

121 

122 def __init__(self, session=None, **kwargs): 

123 if session and not isinstance(session, requests.Session): 

124 raise TypeError("Only a requests.Session object can be mocked") 

125 

126 self._mock_target = session or requests.Session 

127 self.case_sensitive = kwargs.pop('case_sensitive', self.case_sensitive) 

128 self._adapter = ( 

129 kwargs.pop('adapter', None) or 

130 adapter.Adapter(case_sensitive=self.case_sensitive) 

131 ) 

132 

133 self._json_encoder = kwargs.pop('json_encoder', None) 

134 self.real_http = kwargs.pop('real_http', False) 

135 self._last_send = None 

136 

137 if kwargs: 

138 raise TypeError('Unexpected Arguments: %s' % ', '.join(kwargs)) 

139 

140 def start(self): 

141 """Start mocking requests. 

142 

143 Install the adapter and the wrappers required to intercept requests. 

144 """ 

145 if self._last_send: 

146 raise RuntimeError('Mocker has already been started') 

147 

148 # backup last `send` for restoration on `self.stop` 

149 self._last_send = self._mock_target.send 

150 self._last_get_adapter = self._mock_target.get_adapter 

151 

152 def _fake_get_adapter(session, url): 

153 return self._adapter 

154 

155 def _fake_send(session, request, **kwargs): 

156 # NOTE(phodge): we need to use a threading lock here in case there 

157 # are multiple threads running - one thread could restore the 

158 # original get_adapter() just as a second thread is about to 

159 # execute _original_send() below 

160 with threading_rlock(timeout=10): 

161 # mock get_adapter 

162 # 

163 # NOTE(phodge): requests.Session.send() is actually 

164 # reentrant due to how it resolves redirects with nested 

165 # calls to send(), however the reentry occurs _after_ the 

166 # call to self.get_adapter(), so it doesn't matter that we 

167 # will restore _last_get_adapter before a nested send() has 

168 # completed as long as we monkeypatch get_adapter() each 

169 # time immediately before calling original send() like we 

170 # are doing here. 

171 _set_method(session, "get_adapter", _fake_get_adapter) 

172 

173 # NOTE(jamielennox): self._last_send vs _original_send. Whilst 

174 # it seems like here we would use _last_send there is the 

175 # possibility that the user has messed up and is somehow 

176 # nesting their mockers. If we call last_send at this point 

177 # then we end up calling this function again and the outer 

178 # level adapter ends up winning. All we really care about here 

179 # is that our adapter is in place before calling send so we 

180 # always jump directly to the real function so that our most 

181 # recently patched send call ends up putting in the most recent 

182 # adapter. It feels funny, but it works. 

183 

184 try: 

185 return _original_send(session, request, **kwargs) 

186 except exceptions.NoMockAddress: 

187 if not self.real_http: 

188 raise 

189 except adapter._RunRealHTTP: 

190 # this mocker wants you to run the request through the real 

191 # requests library rather than the mocking. Let it. 

192 pass 

193 finally: 

194 # restore get_adapter 

195 _set_method(session, "get_adapter", self._last_get_adapter) 

196 

197 # if we are here it means we must run the real http request 

198 # Or, with nested mocks, to the parent mock, that is why we use 

199 # _last_send here instead of _original_send 

200 if isinstance(self._mock_target, type): 

201 return self._last_send(session, request, **kwargs) 

202 else: 

203 return self._last_send(request, **kwargs) 

204 

205 _set_method(self._mock_target, "send", _fake_send) 

206 

207 def stop(self): 

208 """Stop mocking requests. 

209 

210 This should have no impact if mocking has not been started. 

211 When nesting mockers, make sure to stop the innermost first. 

212 """ 

213 if self._last_send: 

214 self._mock_target.send = self._last_send 

215 self._last_send = None 

216 

217 # for familiarity with MagicMock 

218 def reset_mock(self): 

219 self.reset() 

220 

221 def __getattr__(self, name): 

222 if name in self._PROXY_FUNCS: 

223 try: 

224 return getattr(self._adapter, name) 

225 except AttributeError: 

226 pass 

227 

228 raise AttributeError(name) 

229 

230 def register_uri(self, *args, **kwargs): 

231 # you can pass real_http here, but it's private to pass direct to the 

232 # adapter, because if you pass direct to the adapter you'll see the exc 

233 kwargs['_real_http'] = kwargs.pop('real_http', False) 

234 kwargs.setdefault('json_encoder', self._json_encoder) 

235 return self._adapter.register_uri(*args, **kwargs) 

236 

237 def request(self, *args, **kwargs): 

238 return self.register_uri(*args, **kwargs) 

239 

240 def get(self, *args, **kwargs): 

241 return self.request(GET, *args, **kwargs) 

242 

243 def options(self, *args, **kwargs): 

244 return self.request(OPTIONS, *args, **kwargs) 

245 

246 def head(self, *args, **kwargs): 

247 return self.request(HEAD, *args, **kwargs) 

248 

249 def post(self, *args, **kwargs): 

250 return self.request(POST, *args, **kwargs) 

251 

252 def put(self, *args, **kwargs): 

253 return self.request(PUT, *args, **kwargs) 

254 

255 def patch(self, *args, **kwargs): 

256 return self.request(PATCH, *args, **kwargs) 

257 

258 def delete(self, *args, **kwargs): 

259 return self.request(DELETE, *args, **kwargs) 

260 

261 

262class Mocker(MockerCore): 

263 """The standard entry point for mock Adapter loading. 

264 """ 

265 

266 #: Defines with what should method name begin to be patched 

267 TEST_PREFIX = 'test' 

268 

269 def __init__(self, **kwargs): 

270 """Create a new mocker adapter. 

271 

272 :param str kw: Pass the mock object through to the decorated function 

273 as this named keyword argument, rather than a positional argument. 

274 :param bool real_http: True to send the request to the real requested 

275 uri if there is not a mock installed for it. Defaults to False. 

276 """ 

277 self._kw = kwargs.pop('kw', None) 

278 super(Mocker, self).__init__(**kwargs) 

279 

280 def __enter__(self): 

281 self.start() 

282 return self 

283 

284 def __exit__(self, type, value, traceback): 

285 self.stop() 

286 

287 def __call__(self, obj): 

288 if isinstance(obj, type): 

289 return self.decorate_class(obj) 

290 

291 return self.decorate_callable(obj) 

292 

293 def copy(self): 

294 """Returns an exact copy of current mock 

295 """ 

296 m = type(self)( 

297 kw=self._kw, 

298 real_http=self.real_http, 

299 case_sensitive=self.case_sensitive 

300 ) 

301 return m 

302 

303 def decorate_callable(self, func): 

304 """Decorates a callable 

305 

306 :param callable func: callable to decorate 

307 """ 

308 @functools.wraps(func) 

309 def inner(*args, **kwargs): 

310 with self.copy() as m: 

311 if self._kw: 

312 kwargs[self._kw] = m 

313 else: 

314 args = list(args) 

315 args.append(m) 

316 

317 return func(*args, **kwargs) 

318 

319 return inner 

320 

321 def decorate_class(self, klass): 

322 """Decorates methods in a class with request_mock 

323 

324 Method will be decorated only if it name begins with `TEST_PREFIX` 

325 

326 :param object klass: class which methods will be decorated 

327 """ 

328 for attr_name in dir(klass): 

329 if not attr_name.startswith(self.TEST_PREFIX): 

330 continue 

331 

332 attr = getattr(klass, attr_name) 

333 if not hasattr(attr, '__call__'): 

334 continue 

335 

336 m = self.copy() 

337 setattr(klass, attr_name, m(attr)) 

338 

339 return klass 

340 

341 

342mock = Mocker