Codebase list responses / 7b42398
Import upstream version 0.13.2+git20210427.1.bdf4c3b Debian Janitor 2 years ago
11 changed file(s) with 2471 addition(s) and 2145 deletion(s). Raw diff Collapse all Expand all
0 0.13.3
1 ------
2
3 * Switch from Travis to GHA for deployment.
4
5 0.13.2
6 ------
7
8 * Fixed incorrect type stubs for `add_callback`
9
10 0.13.1
11 ------
12
13 * Fixed packages not containing type stubs.
14
15 0.13.0
16 ------
17
18 * `responses.upsert()` was added. This method will `add()` a response if one
19 has not already been registered for a URL, or `replace()` an existing
20 response.
21 * `responses.registered()` was added. The method allows you to get a list of
22 the currently registered responses. This formalizes the previously private
23 `responses.mock._matches` method.
24 * A more useful `__repr__` has been added to `Response`.
25 * Error messages have been improved.
26
027 0.12.1
128 ------
229
00 Metadata-Version: 2.1
11 Name: responses
2 Version: 0.12.1
2 Version: 0.13.3
33 Summary: A utility library for mocking out the `requests` Python library.
44 Home-page: https://github.com/getsentry/responses
55 Author: David Cramer
154154 responses.add(
155155 responses.POST,
156156 url='http://calc.com/sum',
157 body=4,
157 body="4",
158158 match=[
159 responses.urlencoded_params_matcher({"left": 1, "right": 3})
159 responses.urlencoded_params_matcher({"left": "1", "right": "3"})
160160 ]
161161 )
162162 requests.post("http://calc.com/sum", data={"left": 1, "right": 3})
337337 resp = requests.get('http://twitter.com/api/1/foobar')
338338 assert resp.status_code == 200
339339
340 Responses inside a unittest setUp()
341 -----------------------------------
342
343 When run with unittest tests, this can be used to set up some
344 generic class-level responses, that may be complemented by each test
345
346 .. code-block:: python
347
348 def setUp():
349 self.responses = responses.RequestsMock()
350 self.responses.start()
351
352 # self.responses.add(...)
353
354 self.addCleanup(self.responses.stop)
355 self.addCleanup(self.responses.reset)
356
357 def test_api(self):
358 self.responses.add(
359 responses.GET, 'http://twitter.com/api/1/foobar',
360 body='{}', status=200,
361 content_type='application/json')
362 resp = requests.get('http://twitter.com/api/1/foobar')
363 assert resp.status_code == 200
364
340365 Assertions on declared responses
341366 --------------------------------
342367
456481 Viewing/Modifying registered responses
457482 --------------------------------------
458483
459 Registered responses are available as a private attribute of the RequestMock
484 Registered responses are available as a public method of the RequestMock
460485 instance. It is sometimes useful for debugging purposes to view the stack of
461 registered responses which can be accessed via ``responses.mock._matches``.
486 registered responses which can be accessed via ``responses.registered()``.
462487
463488 The ``replace`` function allows a previously registered ``response`` to be
464489 changed. The method signature is identical to ``add``. ``response`` s are
480505
481506 assert resp.json() == {'data': 2}
482507
508
509 The ``upsert`` function allows a previously registered ``response`` to be
510 changed like ``replace``. If the response is registered, the ``upsert`` function
511 will registered it like ``add``.
483512
484513 ``remove`` takes a ``method`` and ``url`` argument and will remove **all**
485514 matched responses from the registered list.
147147 responses.add(
148148 responses.POST,
149149 url='http://calc.com/sum',
150 body=4,
150 body="4",
151151 match=[
152 responses.urlencoded_params_matcher({"left": 1, "right": 3})
152 responses.urlencoded_params_matcher({"left": "1", "right": "3"})
153153 ]
154154 )
155155 requests.post("http://calc.com/sum", data={"left": 1, "right": 3})
330330 resp = requests.get('http://twitter.com/api/1/foobar')
331331 assert resp.status_code == 200
332332
333 Responses inside a unittest setUp()
334 -----------------------------------
335
336 When run with unittest tests, this can be used to set up some
337 generic class-level responses, that may be complemented by each test
338
339 .. code-block:: python
340
341 def setUp():
342 self.responses = responses.RequestsMock()
343 self.responses.start()
344
345 # self.responses.add(...)
346
347 self.addCleanup(self.responses.stop)
348 self.addCleanup(self.responses.reset)
349
350 def test_api(self):
351 self.responses.add(
352 responses.GET, 'http://twitter.com/api/1/foobar',
353 body='{}', status=200,
354 content_type='application/json')
355 resp = requests.get('http://twitter.com/api/1/foobar')
356 assert resp.status_code == 200
357
333358 Assertions on declared responses
334359 --------------------------------
335360
449474 Viewing/Modifying registered responses
450475 --------------------------------------
451476
452 Registered responses are available as a private attribute of the RequestMock
477 Registered responses are available as a public method of the RequestMock
453478 instance. It is sometimes useful for debugging purposes to view the stack of
454 registered responses which can be accessed via ``responses.mock._matches``.
479 registered responses which can be accessed via ``responses.registered()``.
455480
456481 The ``replace`` function allows a previously registered ``response`` to be
457482 changed. The method signature is identical to ``add``. ``response`` s are
474499 assert resp.json() == {'data': 2}
475500
476501
502 The ``upsert`` function allows a previously registered ``response`` to be
503 changed like ``replace``. If the response is registered, the ``upsert`` function
504 will registered it like ``add``.
505
477506 ``remove`` takes a ``method`` and ``url`` argument and will remove **all**
478507 matched responses from the registered list.
479508
0 from __future__ import absolute_import, print_function, division, unicode_literals
1
2 import _io
3 import inspect
4 import json as json_module
5 import logging
6 import re
7 from itertools import groupby
8
9 import six
10
11 from collections import namedtuple
12 from functools import update_wrapper
13 from requests.adapters import HTTPAdapter
14 from requests.exceptions import ConnectionError
15 from requests.sessions import REDIRECT_STATI
16 from requests.utils import cookiejar_from_dict
17
18 try:
19 from collections.abc import Sequence, Sized
20 except ImportError:
21 from collections import Sequence, Sized
22
23 try:
24 from requests.packages.urllib3.response import HTTPResponse
25 except ImportError:
26 from urllib3.response import HTTPResponse
27 try:
28 from requests.packages.urllib3.connection import HTTPHeaderDict
29 except ImportError:
30 from urllib3.response import HTTPHeaderDict
31
32 if six.PY2:
33 from urlparse import urlparse, parse_qsl, urlsplit, urlunsplit
34 from urllib import quote
35 else:
36 from urllib.parse import urlparse, parse_qsl, urlsplit, urlunsplit, quote
37
38 if six.PY2:
39 try:
40 from six import cStringIO as BufferIO
41 except ImportError:
42 from six import StringIO as BufferIO
43 else:
44 from io import BytesIO as BufferIO
45
46 try:
47 from unittest import mock as std_mock
48 except ImportError:
49 import mock as std_mock
50
51 try:
52 Pattern = re._pattern_type
53 except AttributeError:
54 # Python 3.7
55 Pattern = re.Pattern
56
57 try:
58 from json.decoder import JSONDecodeError
59 except ImportError:
60 JSONDecodeError = ValueError
61
62 UNSET = object()
63
64 Call = namedtuple("Call", ["request", "response"])
65
66 _real_send = HTTPAdapter.send
67
68 logger = logging.getLogger("responses")
69
70
71 def _is_string(s):
72 return isinstance(s, six.string_types)
73
74
75 def _has_unicode(s):
76 return any(ord(char) > 128 for char in s)
77
78
79 def _clean_unicode(url):
80 # Clean up domain names, which use punycode to handle unicode chars
81 urllist = list(urlsplit(url))
82 netloc = urllist[1]
83 if _has_unicode(netloc):
84 domains = netloc.split(".")
85 for i, d in enumerate(domains):
86 if _has_unicode(d):
87 d = "xn--" + d.encode("punycode").decode("ascii")
88 domains[i] = d
89 urllist[1] = ".".join(domains)
90 url = urlunsplit(urllist)
91
92 # Clean up path/query/params, which use url-encoding to handle unicode chars
93 if isinstance(url.encode("utf8"), six.string_types):
94 url = url.encode("utf8")
95 chars = list(url)
96 for i, x in enumerate(chars):
97 if ord(x) > 128:
98 chars[i] = quote(x)
99
100 return "".join(chars)
101
102
103 def _is_redirect(response):
104 try:
105 # 2.0.0 <= requests <= 2.2
106 return response.is_redirect
107
108 except AttributeError:
109 # requests > 2.2
110 return (
111 # use request.sessions conditional
112 response.status_code in REDIRECT_STATI
113 and "location" in response.headers
114 )
115
116
117 def _ensure_str(s):
118 if six.PY2:
119 s = s.encode("utf-8") if isinstance(s, six.text_type) else s
120 return s
121
122
123 def _cookies_from_headers(headers):
124 try:
125 import http.cookies as cookies
126
127 resp_cookie = cookies.SimpleCookie()
128 resp_cookie.load(headers["set-cookie"])
129
130 cookies_dict = {name: v.value for name, v in resp_cookie.items()}
131 except ImportError:
132 from cookies import Cookies
133
134 resp_cookies = Cookies.from_request(_ensure_str(headers["set-cookie"]))
135 cookies_dict = {
136 v.name: quote(_ensure_str(v.value)) for _, v in resp_cookies.items()
137 }
138 return cookiejar_from_dict(cookies_dict)
139
140
141 _wrapper_template = """\
142 def wrapper%(wrapper_args)s:
143 with responses:
144 return func%(func_args)s
145 """
146
147
148 def get_wrapped(func, responses):
149 if six.PY2:
150 args, a, kw, defaults = inspect.getargspec(func)
151 wrapper_args = inspect.formatargspec(args, a, kw, defaults)
152
153 # Preserve the argspec for the wrapped function so that testing
154 # tools such as pytest can continue to use their fixture injection.
155 if hasattr(func, "__self__"):
156 args = args[1:] # Omit 'self'
157 func_args = inspect.formatargspec(args, a, kw, None)
158 else:
159 signature = inspect.signature(func)
160 signature = signature.replace(return_annotation=inspect.Signature.empty)
161 # If the function is wrapped, switch to *args, **kwargs for the parameters
162 # as we can't rely on the signature to give us the arguments the function will
163 # be called with. For example unittest.mock.patch uses required args that are
164 # not actually passed to the function when invoked.
165 if hasattr(func, "__wrapped__"):
166 wrapper_params = [
167 inspect.Parameter("args", inspect.Parameter.VAR_POSITIONAL),
168 inspect.Parameter("kwargs", inspect.Parameter.VAR_KEYWORD),
169 ]
170 else:
171 wrapper_params = [
172 param.replace(annotation=inspect.Parameter.empty)
173 for param in signature.parameters.values()
174 ]
175 signature = signature.replace(parameters=wrapper_params)
176
177 wrapper_args = str(signature)
178 params_without_defaults = [
179 param.replace(
180 annotation=inspect.Parameter.empty, default=inspect.Parameter.empty
181 )
182 for param in signature.parameters.values()
183 ]
184 signature = signature.replace(parameters=params_without_defaults)
185 func_args = str(signature)
186
187 evaldict = {"func": func, "responses": responses}
188 six.exec_(
189 _wrapper_template % {"wrapper_args": wrapper_args, "func_args": func_args},
190 evaldict,
191 )
192 wrapper = evaldict["wrapper"]
193 update_wrapper(wrapper, func)
194 return wrapper
195
196
197 class CallList(Sequence, Sized):
198 def __init__(self):
199 self._calls = []
200
201 def __iter__(self):
202 return iter(self._calls)
203
204 def __len__(self):
205 return len(self._calls)
206
207 def __getitem__(self, idx):
208 return self._calls[idx]
209
210 def add(self, request, response):
211 self._calls.append(Call(request, response))
212
213 def reset(self):
214 self._calls = []
215
216
217 def _ensure_url_default_path(url):
218 if _is_string(url):
219 url_parts = list(urlsplit(url))
220 if url_parts[2] == "":
221 url_parts[2] = "/"
222 url = urlunsplit(url_parts)
223 return url
224
225
226 def _handle_body(body):
227 if isinstance(body, six.text_type):
228 body = body.encode("utf-8")
229 if isinstance(body, _io.BufferedReader):
230 return body
231
232 return BufferIO(body)
233
234
235 _unspecified = object()
236
237
238 def urlencoded_params_matcher(params):
239 def match(request_body):
240 return (
241 params is None
242 if request_body is None
243 else sorted(params.items()) == sorted(parse_qsl(request_body))
244 )
245
246 return match
247
248
249 def json_params_matcher(params):
250 def match(request_body):
251 try:
252 if isinstance(request_body, bytes):
253 request_body = request_body.decode("utf-8")
254 return (
255 params is None
256 if request_body is None
257 else params == json_module.loads(request_body)
258 )
259 except JSONDecodeError:
260 return False
261
262 return match
263
264
265 class BaseResponse(object):
266 content_type = None
267 headers = None
268
269 stream = False
270
271 def __init__(self, method, url, match_querystring=_unspecified, match=[]):
272 self.method = method
273 # ensure the url has a default path set if the url is a string
274 self.url = _ensure_url_default_path(url)
275 self.match_querystring = self._should_match_querystring(match_querystring)
276 self.match = match
277 self.call_count = 0
278
279 def __eq__(self, other):
280 if not isinstance(other, BaseResponse):
281 return False
282
283 if self.method != other.method:
284 return False
285
286 # Can't simply do an equality check on the objects directly here since __eq__ isn't
287 # implemented for regex. It might seem to work as regex is using a cache to return
288 # the same regex instances, but it doesn't in all cases.
289 self_url = self.url.pattern if isinstance(self.url, Pattern) else self.url
290 other_url = other.url.pattern if isinstance(other.url, Pattern) else other.url
291
292 return self_url == other_url
293
294 def __ne__(self, other):
295 return not self.__eq__(other)
296
297 def _url_matches_strict(self, url, other):
298 url_parsed = urlparse(url)
299 other_parsed = urlparse(other)
300
301 if url_parsed[:3] != other_parsed[:3]:
302 return False
303
304 url_qsl = sorted(parse_qsl(url_parsed.query))
305 other_qsl = sorted(parse_qsl(other_parsed.query))
306
307 return url_qsl == other_qsl
308
309 def _should_match_querystring(self, match_querystring_argument):
310 if match_querystring_argument is not _unspecified:
311 return match_querystring_argument
312
313 if isinstance(self.url, Pattern):
314 # the old default from <= 0.9.0
315 return False
316
317 return bool(urlparse(self.url).query)
318
319 def _url_matches(self, url, other, match_querystring=False):
320 if _is_string(url):
321 if _has_unicode(url):
322 url = _clean_unicode(url)
323 if not isinstance(other, six.text_type):
324 other = other.encode("ascii").decode("utf8")
325 if match_querystring:
326 return self._url_matches_strict(url, other)
327
328 else:
329 url_without_qs = url.split("?", 1)[0]
330 other_without_qs = other.split("?", 1)[0]
331 return url_without_qs == other_without_qs
332
333 elif isinstance(url, Pattern) and url.match(other):
334 return True
335
336 else:
337 return False
338
339 def _body_matches(self, match, request_body):
340 for matcher in match:
341 if not matcher(request_body):
342 return False
343
344 return True
345
346 def get_headers(self):
347 headers = HTTPHeaderDict() # Duplicate headers are legal
348 if self.content_type is not None:
349 headers["Content-Type"] = self.content_type
350 if self.headers:
351 headers.extend(self.headers)
352 return headers
353
354 def get_response(self, request):
355 raise NotImplementedError
356
357 def matches(self, request):
358 if request.method != self.method:
359 return False, "Method does not match"
360
361 if not self._url_matches(self.url, request.url, self.match_querystring):
362 return False, "URL does not match"
363
364 if not self._body_matches(self.match, request.body):
365 return False, "Parameters do not match"
366
367 return True, ""
368
369
370 class Response(BaseResponse):
371 def __init__(
372 self,
373 method,
374 url,
375 body="",
376 json=None,
377 status=200,
378 headers=None,
379 stream=False,
380 content_type=UNSET,
381 **kwargs
382 ):
383 # if we were passed a `json` argument,
384 # override the body and content_type
385 if json is not None:
386 assert not body
387 body = json_module.dumps(json)
388 if content_type is UNSET:
389 content_type = "application/json"
390
391 if content_type is UNSET:
392 if isinstance(body, six.text_type) and _has_unicode(body):
393 content_type = "text/plain; charset=utf-8"
394 else:
395 content_type = "text/plain"
396
397 self.body = body
398 self.status = status
399 self.headers = headers
400 self.stream = stream
401 self.content_type = content_type
402 super(Response, self).__init__(method, url, **kwargs)
403
404 def get_response(self, request):
405 if self.body and isinstance(self.body, Exception):
406 raise self.body
407
408 headers = self.get_headers()
409 status = self.status
410 body = _handle_body(self.body)
411 return HTTPResponse(
412 status=status,
413 reason=six.moves.http_client.responses.get(status),
414 body=body,
415 headers=headers,
416 original_response=OriginalResponseShim(headers),
417 preload_content=False,
418 )
419
420 def __repr__(self):
421 return (
422 "<Response(url='{url}' status={status} "
423 "content_type='{content_type}' headers='{headers}')>".format(
424 url=self.url,
425 status=self.status,
426 content_type=self.content_type,
427 headers=json_module.dumps(self.headers),
428 )
429 )
430
431
432 class CallbackResponse(BaseResponse):
433 def __init__(
434 self, method, url, callback, stream=False, content_type="text/plain", **kwargs
435 ):
436 self.callback = callback
437 self.stream = stream
438 self.content_type = content_type
439 super(CallbackResponse, self).__init__(method, url, **kwargs)
440
441 def get_response(self, request):
442 headers = self.get_headers()
443
444 result = self.callback(request)
445 if isinstance(result, Exception):
446 raise result
447
448 status, r_headers, body = result
449 if isinstance(body, Exception):
450 raise body
451
452 # If the callback set a content-type remove the one
453 # set in add_callback() so that we don't have multiple
454 # content type values.
455 has_content_type = False
456 if isinstance(r_headers, dict) and "Content-Type" in r_headers:
457 has_content_type = True
458 elif isinstance(r_headers, list):
459 has_content_type = any(
460 [h for h in r_headers if h and h[0].lower() == "content-type"]
461 )
462 if has_content_type:
463 headers.pop("Content-Type", None)
464
465 body = _handle_body(body)
466 headers.extend(r_headers)
467
468 return HTTPResponse(
469 status=status,
470 reason=six.moves.http_client.responses.get(status),
471 body=body,
472 headers=headers,
473 original_response=OriginalResponseShim(headers),
474 preload_content=False,
475 )
476
477
478 class OriginalResponseShim(object):
479 """
480 Shim for compatibility with older versions of urllib3
481
482 requests cookie handling depends on responses having a property chain of
483 `response._original_response.msg` which contains the response headers [1]
484
485 Using HTTPResponse() for this purpose causes compatibility errors with
486 urllib3<1.23.0. To avoid adding more dependencies we can use this shim.
487
488 [1]: https://github.com/psf/requests/blob/75bdc998e2d/requests/cookies.py#L125
489 """
490
491 def __init__(self, headers):
492 self.msg = headers
493
494 def isclosed(self):
495 return True
496
497
498 class RequestsMock(object):
499 DELETE = "DELETE"
500 GET = "GET"
501 HEAD = "HEAD"
502 OPTIONS = "OPTIONS"
503 PATCH = "PATCH"
504 POST = "POST"
505 PUT = "PUT"
506 response_callback = None
507
508 def __init__(
509 self,
510 assert_all_requests_are_fired=True,
511 response_callback=None,
512 passthru_prefixes=(),
513 target="requests.adapters.HTTPAdapter.send",
514 ):
515 self._calls = CallList()
516 self.reset()
517 self.assert_all_requests_are_fired = assert_all_requests_are_fired
518 self.response_callback = response_callback
519 self.passthru_prefixes = tuple(passthru_prefixes)
520 self.target = target
521
522 def reset(self):
523 self._matches = []
524 self._calls.reset()
525
526 def add(
527 self,
528 method=None, # method or ``Response``
529 url=None,
530 body="",
531 adding_headers=None,
532 *args,
533 **kwargs
534 ):
535 """
536 A basic request:
537
538 >>> responses.add(responses.GET, 'http://example.com')
539
540 You can also directly pass an object which implements the
541 ``BaseResponse`` interface:
542
543 >>> responses.add(Response(...))
544
545 A JSON payload:
546
547 >>> responses.add(
548 >>> method='GET',
549 >>> url='http://example.com',
550 >>> json={'foo': 'bar'},
551 >>> )
552
553 Custom headers:
554
555 >>> responses.add(
556 >>> method='GET',
557 >>> url='http://example.com',
558 >>> headers={'X-Header': 'foo'},
559 >>> )
560
561
562 Strict query string matching:
563
564 >>> responses.add(
565 >>> method='GET',
566 >>> url='http://example.com?foo=bar',
567 >>> match_querystring=True
568 >>> )
569 """
570 if isinstance(method, BaseResponse):
571 self._matches.append(method)
572 return
573
574 if adding_headers is not None:
575 kwargs.setdefault("headers", adding_headers)
576
577 self._matches.append(Response(method=method, url=url, body=body, **kwargs))
578
579 def add_passthru(self, prefix):
580 """
581 Register a URL prefix or regex to passthru any non-matching mock requests to.
582
583 For example, to allow any request to 'https://example.com', but require
584 mocks for the remainder, you would add the prefix as so:
585
586 >>> responses.add_passthru('https://example.com')
587
588 Regex can be used like:
589
590 >>> responses.add_passthru(re.compile('https://example.com/\\w+'))
591 """
592 if not isinstance(prefix, Pattern) and _has_unicode(prefix):
593 prefix = _clean_unicode(prefix)
594 self.passthru_prefixes += (prefix,)
595
596 def remove(self, method_or_response=None, url=None):
597 """
598 Removes a response previously added using ``add()``, identified
599 either by a response object inheriting ``BaseResponse`` or
600 ``method`` and ``url``. Removes all matching responses.
601
602 >>> response.add(responses.GET, 'http://example.org')
603 >>> response.remove(responses.GET, 'http://example.org')
604 """
605 if isinstance(method_or_response, BaseResponse):
606 response = method_or_response
607 else:
608 response = BaseResponse(method=method_or_response, url=url)
609
610 while response in self._matches:
611 self._matches.remove(response)
612
613 def replace(self, method_or_response=None, url=None, body="", *args, **kwargs):
614 """
615 Replaces a response previously added using ``add()``. The signature
616 is identical to ``add()``. The response is identified using ``method``
617 and ``url``, and the first matching response is replaced.
618
619 >>> responses.add(responses.GET, 'http://example.org', json={'data': 1})
620 >>> responses.replace(responses.GET, 'http://example.org', json={'data': 2})
621 """
622 if isinstance(method_or_response, BaseResponse):
623 url = method_or_response.url
624 response = method_or_response
625 else:
626 response = Response(method=method_or_response, url=url, body=body, **kwargs)
627
628 try:
629 index = self._matches.index(response)
630 except ValueError:
631 raise ValueError("Response is not registered for URL %s" % url)
632 self._matches[index] = response
633
634 def upsert(self, method_or_response=None, url=None, body="", *args, **kwargs):
635 """
636 Replaces a response previously added using ``add()``, or adds the response
637 if no response exists. Responses are matched using ``method``and ``url``.
638 The first matching response is replaced.
639
640 >>> responses.add(responses.GET, 'http://example.org', json={'data': 1})
641 >>> responses.upsert(responses.GET, 'http://example.org', json={'data': 2})
642 """
643 try:
644 self.replace(method_or_response, url, body, *args, **kwargs)
645 except ValueError:
646 self.add(method_or_response, url, body, *args, **kwargs)
647
648 def add_callback(
649 self, method, url, callback, match_querystring=False, content_type="text/plain"
650 ):
651 # ensure the url has a default path set if the url is a string
652 # url = _ensure_url_default_path(url, match_querystring)
653
654 self._matches.append(
655 CallbackResponse(
656 url=url,
657 method=method,
658 callback=callback,
659 content_type=content_type,
660 match_querystring=match_querystring,
661 )
662 )
663
664 def registered(self):
665 return self._matches
666
667 @property
668 def calls(self):
669 return self._calls
670
671 def __enter__(self):
672 self.start()
673 return self
674
675 def __exit__(self, type, value, traceback):
676 success = type is None
677 self.stop(allow_assert=success)
678 self.reset()
679 return success
680
681 def activate(self, func):
682 return get_wrapped(func, self)
683
684 def _find_match(self, request):
685 found = None
686 found_match = None
687 match_failed_reasons = []
688 for i, match in enumerate(self._matches):
689 match_result, reason = match.matches(request)
690 if match_result:
691 if found is None:
692 found = i
693 found_match = match
694 else:
695 # Multiple matches found. Remove & return the first match.
696 return self._matches.pop(found), match_failed_reasons
697 else:
698 match_failed_reasons.append(reason)
699 return found_match, match_failed_reasons
700
701 def _parse_request_params(self, url):
702 params = {}
703 for key, val in groupby(parse_qsl(urlparse(url).query), lambda kv: kv[0]):
704 values = list(map(lambda x: x[1], val))
705 if len(values) == 1:
706 values = values[0]
707 params[key] = values
708 return params
709
710 def _on_request(self, adapter, request, **kwargs):
711 match, match_failed_reasons = self._find_match(request)
712 resp_callback = self.response_callback
713 request.params = self._parse_request_params(request.path_url)
714
715 if match is None:
716 if any(
717 [
718 p.match(request.url)
719 if isinstance(p, Pattern)
720 else request.url.startswith(p)
721 for p in self.passthru_prefixes
722 ]
723 ):
724 logger.info("request.allowed-passthru", extra={"url": request.url})
725 return _real_send(adapter, request, **kwargs)
726
727 error_msg = (
728 "Connection refused by Responses - the call doesn't "
729 "match any registered mock.\n\n"
730 "Request: \n"
731 "- %s %s\n\n"
732 "Available matches:\n" % (request.method, request.url)
733 )
734 for i, m in enumerate(self._matches):
735 error_msg += "- {} {} {}\n".format(
736 m.method, m.url, match_failed_reasons[i]
737 )
738
739 response = ConnectionError(error_msg)
740 response.request = request
741
742 self._calls.add(request, response)
743 response = resp_callback(response) if resp_callback else response
744 raise response
745
746 try:
747 response = adapter.build_response(request, match.get_response(request))
748 except BaseException as response:
749 match.call_count += 1
750 self._calls.add(request, response)
751 response = resp_callback(response) if resp_callback else response
752 raise
753
754 if not match.stream:
755 response.content # NOQA
756
757 response = resp_callback(response) if resp_callback else response
758 match.call_count += 1
759 self._calls.add(request, response)
760 return response
761
762 def start(self):
763 def unbound_on_send(adapter, request, *a, **kwargs):
764 return self._on_request(adapter, request, *a, **kwargs)
765
766 self._patcher = std_mock.patch(target=self.target, new=unbound_on_send)
767 self._patcher.start()
768
769 def stop(self, allow_assert=True):
770 self._patcher.stop()
771 if not self.assert_all_requests_are_fired:
772 return
773
774 if not allow_assert:
775 return
776
777 not_called = [m for m in self._matches if m.call_count == 0]
778 if not_called:
779 raise AssertionError(
780 "Not all requests have been executed {0!r}".format(
781 [(match.method, match.url) for match in not_called]
782 )
783 )
784
785 def assert_call_count(self, url, count):
786 call_count = len(
787 [
788 1
789 for call in self.calls
790 if call.request.url == _ensure_url_default_path(url)
791 ]
792 )
793 if call_count == count:
794 return True
795 else:
796 raise AssertionError(
797 "Expected URL '{0}' to be called {1} times. Called {2} times.".format(
798 url, count, call_count
799 )
800 )
801
802
803 # expose default mock namespace
804 mock = _default_mock = RequestsMock(assert_all_requests_are_fired=False)
805 __all__ = [
806 "CallbackResponse",
807 "Response",
808 "RequestsMock",
809 # Exposed by the RequestsMock class:
810 "activate",
811 "add",
812 "add_callback",
813 "add_passthru",
814 "assert_all_requests_are_fired",
815 "assert_call_count",
816 "calls",
817 "DELETE",
818 "GET",
819 "HEAD",
820 "OPTIONS",
821 "passthru_prefixes",
822 "PATCH",
823 "POST",
824 "PUT",
825 "registered",
826 "remove",
827 "replace",
828 "reset",
829 "response_callback",
830 "start",
831 "stop",
832 "target",
833 "upsert",
834 ]
835
836 activate = _default_mock.activate
837 add = _default_mock.add
838 add_callback = _default_mock.add_callback
839 add_passthru = _default_mock.add_passthru
840 assert_all_requests_are_fired = _default_mock.assert_all_requests_are_fired
841 assert_call_count = _default_mock.assert_call_count
842 calls = _default_mock.calls
843 DELETE = _default_mock.DELETE
844 GET = _default_mock.GET
845 HEAD = _default_mock.HEAD
846 OPTIONS = _default_mock.OPTIONS
847 passthru_prefixes = _default_mock.passthru_prefixes
848 PATCH = _default_mock.PATCH
849 POST = _default_mock.POST
850 PUT = _default_mock.PUT
851 registered = _default_mock.registered
852 remove = _default_mock.remove
853 replace = _default_mock.replace
854 reset = _default_mock.reset
855 response_callback = _default_mock.response_callback
856 start = _default_mock.start
857 stop = _default_mock.stop
858 target = _default_mock.target
859 upsert = _default_mock.upsert
0 # coding: utf-8
1
2 from __future__ import absolute_import, print_function, division, unicode_literals
3
4 import inspect
5 import re
6 import six
7 from io import BufferedReader, BytesIO
8
9 import pytest
10 import requests
11 import responses
12 from requests.exceptions import ConnectionError, HTTPError
13 from responses import BaseResponse, Response
14
15 try:
16 from mock import patch, Mock
17 except ImportError:
18 from unittest.mock import patch, Mock # type: ignore
19
20
21 def assert_reset():
22 assert len(responses._default_mock._matches) == 0
23 assert len(responses.calls) == 0
24
25
26 def assert_response(resp, body=None, content_type="text/plain"):
27 assert resp.status_code == 200
28 assert resp.reason == "OK"
29 if content_type is not None:
30 assert resp.headers["Content-Type"] == content_type
31 else:
32 assert "Content-Type" not in resp.headers
33 assert resp.text == body
34
35
36 def assert_params(resp, expected):
37 assert hasattr(resp, "request"), "Missing request"
38 assert hasattr(
39 resp.request, "params"
40 ), "Missing params on request that responses should add"
41 assert getattr(resp.request, "params") == expected, "Incorrect parameters"
42
43
44 def test_response():
45 @responses.activate
46 def run():
47 responses.add(responses.GET, "http://example.com", body=b"test")
48 resp = requests.get("http://example.com")
49 assert_response(resp, "test")
50 assert len(responses.calls) == 1
51 assert responses.calls[0].request.url == "http://example.com/"
52 assert responses.calls[0].response.content == b"test"
53
54 resp = requests.get("http://example.com?foo=bar")
55 assert_response(resp, "test")
56 assert len(responses.calls) == 2
57 assert responses.calls[1].request.url == "http://example.com/?foo=bar"
58 assert responses.calls[1].response.content == b"test"
59
60 run()
61 assert_reset()
62
63
64 def test_response_encoded():
65 @responses.activate
66 def run():
67 # Path contains urlencoded =/()[]
68 url = "http://example.org/foo.bar%3D%2F%28%29%5B%5D"
69 responses.add(responses.GET, url, body="it works", status=200)
70 resp = requests.get(url)
71 assert_response(resp, "it works")
72
73 run()
74 assert_reset()
75
76
77 def test_response_with_instance():
78 @responses.activate
79 def run():
80 responses.add(
81 responses.Response(method=responses.GET, url="http://example.com")
82 )
83 resp = requests.get("http://example.com")
84 assert_response(resp, "")
85 assert len(responses.calls) == 1
86 assert responses.calls[0].request.url == "http://example.com/"
87
88 resp = requests.get("http://example.com?foo=bar")
89 assert_response(resp, "")
90 assert len(responses.calls) == 2
91 assert responses.calls[1].request.url == "http://example.com/?foo=bar"
92
93
94 @pytest.mark.parametrize(
95 "original,replacement",
96 [
97 ("http://example.com/two", "http://example.com/two"),
98 (
99 Response(method=responses.GET, url="http://example.com/two"),
100 Response(
101 method=responses.GET, url="http://example.com/two", body="testtwo"
102 ),
103 ),
104 (
105 re.compile(r"http://example\.com/two"),
106 re.compile(r"http://example\.com/two"),
107 ),
108 ],
109 )
110 def test_replace(original, replacement):
111 @responses.activate
112 def run():
113 responses.add(responses.GET, "http://example.com/one", body="test1")
114
115 if isinstance(original, BaseResponse):
116 responses.add(original)
117 else:
118 responses.add(responses.GET, original, body="test2")
119
120 responses.add(responses.GET, "http://example.com/three", body="test3")
121 responses.add(
122 responses.GET, re.compile(r"http://example\.com/four"), body="test3"
123 )
124
125 if isinstance(replacement, BaseResponse):
126 responses.replace(replacement)
127 else:
128 responses.replace(responses.GET, replacement, body="testtwo")
129
130 resp = requests.get("http://example.com/two")
131 assert_response(resp, "testtwo")
132
133 run()
134 assert_reset()
135
136
137 @pytest.mark.parametrize(
138 "original,replacement",
139 [
140 ("http://example.com/one", re.compile(r"http://example\.com/one")),
141 (re.compile(r"http://example\.com/one"), "http://example.com/one"),
142 ],
143 )
144 def test_replace_error(original, replacement):
145 @responses.activate
146 def run():
147 responses.add(responses.GET, original)
148 with pytest.raises(ValueError) as excinfo:
149 responses.replace(responses.GET, replacement)
150 assert "Response is not registered for URL %s" % replacement in str(
151 excinfo.value
152 )
153
154 run()
155 assert_reset()
156
157
158 def test_replace_response_object_error():
159 @responses.activate
160 def run():
161 responses.add(Response(method=responses.GET, url="http://example.com/one"))
162 with pytest.raises(ValueError) as excinfo:
163 responses.replace(
164 Response(method=responses.GET, url="http://example.com/two")
165 )
166 assert "Response is not registered for URL http://example.com/two" in str(
167 excinfo.value
168 )
169
170 run()
171 assert_reset()
172
173
174 @pytest.mark.parametrize(
175 "original,replacement",
176 [
177 ("http://example.com/two", "http://example.com/two"),
178 (
179 Response(method=responses.GET, url="http://example.com/two"),
180 Response(
181 method=responses.GET, url="http://example.com/two", body="testtwo"
182 ),
183 ),
184 (
185 re.compile(r"http://example\.com/two"),
186 re.compile(r"http://example\.com/two"),
187 ),
188 ],
189 )
190 def test_upsert_replace(original, replacement):
191 @responses.activate
192 def run():
193 responses.add(responses.GET, "http://example.com/one", body="test1")
194
195 if isinstance(original, BaseResponse):
196 responses.add(original)
197 else:
198 responses.add(responses.GET, original, body="test2")
199
200 if isinstance(replacement, BaseResponse):
201 responses.upsert(replacement)
202 else:
203 responses.upsert(responses.GET, replacement, body="testtwo")
204
205 resp = requests.get("http://example.com/two")
206 assert_response(resp, "testtwo")
207
208 run()
209 assert_reset()
210
211
212 @pytest.mark.parametrize(
213 "original,replacement",
214 [
215 ("http://example.com/two", "http://example.com/two"),
216 (
217 Response(method=responses.GET, url="http://example.com/two"),
218 Response(
219 method=responses.GET, url="http://example.com/two", body="testtwo"
220 ),
221 ),
222 (
223 re.compile(r"http://example\.com/two"),
224 re.compile(r"http://example\.com/two"),
225 ),
226 ],
227 )
228 def test_upsert_add(original, replacement):
229 @responses.activate
230 def run():
231 responses.add(responses.GET, "http://example.com/one", body="test1")
232
233 if isinstance(replacement, BaseResponse):
234 responses.upsert(replacement)
235 else:
236 responses.upsert(responses.GET, replacement, body="testtwo")
237
238 resp = requests.get("http://example.com/two")
239 assert_response(resp, "testtwo")
240
241 run()
242 assert_reset()
243
244
245 def test_remove():
246 @responses.activate
247 def run():
248 responses.add(responses.GET, "http://example.com/zero")
249 responses.add(responses.GET, "http://example.com/one")
250 responses.add(responses.GET, "http://example.com/two")
251 responses.add(responses.GET, re.compile(r"http://example\.com/three"))
252 responses.add(responses.GET, re.compile(r"http://example\.com/four"))
253 re.purge()
254 responses.remove(responses.GET, "http://example.com/two")
255 responses.remove(Response(method=responses.GET, url="http://example.com/zero"))
256 responses.remove(responses.GET, re.compile(r"http://example\.com/four"))
257
258 with pytest.raises(ConnectionError):
259 requests.get("http://example.com/zero")
260 requests.get("http://example.com/one")
261 with pytest.raises(ConnectionError):
262 requests.get("http://example.com/two")
263 requests.get("http://example.com/three")
264 with pytest.raises(ConnectionError):
265 requests.get("http://example.com/four")
266
267 run()
268 assert_reset()
269
270
271 @pytest.mark.parametrize(
272 "args1,kwargs1,args2,kwargs2,expected",
273 [
274 ((responses.GET, "a"), {}, (responses.GET, "a"), {}, True),
275 ((responses.GET, "a"), {}, (responses.GET, "b"), {}, False),
276 ((responses.GET, "a"), {}, (responses.POST, "a"), {}, False),
277 (
278 (responses.GET, "a"),
279 {"match_querystring": True},
280 (responses.GET, "a"),
281 {},
282 True,
283 ),
284 ],
285 )
286 def test_response_equality(args1, kwargs1, args2, kwargs2, expected):
287 o1 = BaseResponse(*args1, **kwargs1)
288 o2 = BaseResponse(*args2, **kwargs2)
289 assert (o1 == o2) is expected
290 assert (o1 != o2) is not expected
291
292
293 def test_response_equality_different_objects():
294 o1 = BaseResponse(method=responses.GET, url="a")
295 o2 = "str"
296 assert (o1 == o2) is False
297 assert (o1 != o2) is True
298
299
300 def test_connection_error():
301 @responses.activate
302 def run():
303 responses.add(responses.GET, "http://example.com")
304
305 with pytest.raises(ConnectionError):
306 requests.get("http://example.com/foo")
307
308 assert len(responses.calls) == 1
309 assert responses.calls[0].request.url == "http://example.com/foo"
310 assert type(responses.calls[0].response) is ConnectionError
311 assert responses.calls[0].response.request
312
313 run()
314 assert_reset()
315
316
317 def test_match_querystring():
318 @responses.activate
319 def run():
320 url = "http://example.com?test=1&foo=bar"
321 responses.add(responses.GET, url, match_querystring=True, body=b"test")
322 resp = requests.get("http://example.com?test=1&foo=bar")
323 assert_response(resp, "test")
324 resp = requests.get("http://example.com?foo=bar&test=1")
325 assert_response(resp, "test")
326 resp = requests.get("http://example.com/?foo=bar&test=1")
327 assert_response(resp, "test")
328
329 run()
330 assert_reset()
331
332
333 def test_match_empty_querystring():
334 @responses.activate
335 def run():
336 responses.add(
337 responses.GET, "http://example.com", body=b"test", match_querystring=True
338 )
339 resp = requests.get("http://example.com")
340 assert_response(resp, "test")
341 resp = requests.get("http://example.com/")
342 assert_response(resp, "test")
343 with pytest.raises(ConnectionError):
344 requests.get("http://example.com?query=foo")
345
346 run()
347 assert_reset()
348
349
350 def test_match_querystring_error():
351 @responses.activate
352 def run():
353 responses.add(
354 responses.GET, "http://example.com/?test=1", match_querystring=True
355 )
356
357 with pytest.raises(ConnectionError):
358 requests.get("http://example.com/foo/?test=2")
359
360 run()
361 assert_reset()
362
363
364 def test_match_querystring_regex():
365 @responses.activate
366 def run():
367 """Note that `match_querystring` value shouldn't matter when passing a
368 regular expression"""
369
370 responses.add(
371 responses.GET,
372 re.compile(r"http://example\.com/foo/\?test=1"),
373 body="test1",
374 match_querystring=True,
375 )
376
377 resp = requests.get("http://example.com/foo/?test=1")
378 assert_response(resp, "test1")
379
380 responses.add(
381 responses.GET,
382 re.compile(r"http://example\.com/foo/\?test=2"),
383 body="test2",
384 match_querystring=False,
385 )
386
387 resp = requests.get("http://example.com/foo/?test=2")
388 assert_response(resp, "test2")
389
390 run()
391 assert_reset()
392
393
394 def test_match_querystring_error_regex():
395 @responses.activate
396 def run():
397 """Note that `match_querystring` value shouldn't matter when passing a
398 regular expression"""
399
400 responses.add(
401 responses.GET,
402 re.compile(r"http://example\.com/foo/\?test=1"),
403 match_querystring=True,
404 )
405
406 with pytest.raises(ConnectionError):
407 requests.get("http://example.com/foo/?test=3")
408
409 responses.add(
410 responses.GET,
411 re.compile(r"http://example\.com/foo/\?test=2"),
412 match_querystring=False,
413 )
414
415 with pytest.raises(ConnectionError):
416 requests.get("http://example.com/foo/?test=4")
417
418 run()
419 assert_reset()
420
421
422 def test_match_querystring_auto_activates():
423 @responses.activate
424 def run():
425 responses.add(responses.GET, "http://example.com?test=1", body=b"test")
426 resp = requests.get("http://example.com?test=1")
427 assert_response(resp, "test")
428 with pytest.raises(ConnectionError):
429 requests.get("http://example.com/?test=2")
430
431 run()
432 assert_reset()
433
434
435 def test_match_querystring_missing_key():
436 @responses.activate
437 def run():
438 responses.add(responses.GET, "http://example.com?foo=1&bar=2", body=b"test")
439 with pytest.raises(ConnectionError):
440 requests.get("http://example.com/?foo=1&baz=2")
441
442 with pytest.raises(ConnectionError):
443 requests.get("http://example.com/?bar=2&fez=1")
444
445 run()
446 assert_reset()
447
448
449 def test_accept_string_body():
450 @responses.activate
451 def run():
452 url = "http://example.com/"
453 responses.add(responses.GET, url, body="test")
454 resp = requests.get(url)
455 assert_response(resp, "test")
456
457 run()
458 assert_reset()
459
460
461 def test_accept_json_body():
462 @responses.activate
463 def run():
464 content_type = "application/json"
465
466 url = "http://example.com/"
467 responses.add(responses.GET, url, json={"message": "success"})
468 resp = requests.get(url)
469 assert_response(resp, '{"message": "success"}', content_type)
470
471 url = "http://example.com/1/"
472 responses.add(responses.GET, url, json=[])
473 resp = requests.get(url)
474 assert_response(resp, "[]", content_type)
475
476 run()
477 assert_reset()
478
479
480 def test_no_content_type():
481 @responses.activate
482 def run():
483 url = "http://example.com/"
484 responses.add(responses.GET, url, body="test", content_type=None)
485 resp = requests.get(url)
486 assert_response(resp, "test", content_type=None)
487
488 run()
489 assert_reset()
490
491
492 def test_arbitrary_status_code():
493 @responses.activate
494 def run():
495 url = "http://example.com/"
496 responses.add(responses.GET, url, body="test", status=419)
497 resp = requests.get(url)
498 assert resp.status_code == 419
499 assert resp.reason is None
500
501 run()
502 assert_reset()
503
504
505 def test_throw_connection_error_explicit():
506 @responses.activate
507 def run():
508 url = "http://example.com"
509 exception = HTTPError("HTTP Error")
510 responses.add(responses.GET, url, exception)
511
512 with pytest.raises(HTTPError) as HE:
513 requests.get(url)
514
515 assert str(HE.value) == "HTTP Error"
516
517 run()
518 assert_reset()
519
520
521 def test_callback():
522 body = b"test callback"
523 status = 400
524 reason = "Bad Request"
525 headers = {
526 "foo": "bar",
527 "Content-Type": "application/json",
528 "Content-Length": "13",
529 }
530 url = "http://example.com/"
531
532 def request_callback(request):
533 return (status, headers, body)
534
535 @responses.activate
536 def run():
537 responses.add_callback(responses.GET, url, request_callback)
538 resp = requests.get(url)
539 assert resp.text == "test callback"
540 assert resp.status_code == status
541 assert resp.reason == reason
542 assert "bar" == resp.headers.get("foo")
543 assert "application/json" == resp.headers.get("Content-Type")
544 assert "13" == resp.headers.get("Content-Length")
545
546 run()
547 assert_reset()
548
549
550 def test_callback_exception_result():
551 result = Exception()
552 url = "http://example.com/"
553
554 def request_callback(request):
555 return result
556
557 @responses.activate
558 def run():
559 responses.add_callback(responses.GET, url, request_callback)
560
561 with pytest.raises(Exception) as e:
562 requests.get(url)
563
564 assert e.value is result
565
566 run()
567 assert_reset()
568
569
570 def test_callback_exception_body():
571 body = Exception()
572 url = "http://example.com/"
573
574 def request_callback(request):
575 return (200, {}, body)
576
577 @responses.activate
578 def run():
579 responses.add_callback(responses.GET, url, request_callback)
580
581 with pytest.raises(Exception) as e:
582 requests.get(url)
583
584 assert e.value is body
585
586 run()
587 assert_reset()
588
589
590 def test_callback_no_content_type():
591 body = b"test callback"
592 status = 400
593 reason = "Bad Request"
594 headers = {"foo": "bar"}
595 url = "http://example.com/"
596
597 def request_callback(request):
598 return (status, headers, body)
599
600 @responses.activate
601 def run():
602 responses.add_callback(responses.GET, url, request_callback, content_type=None)
603 resp = requests.get(url)
604 assert resp.text == "test callback"
605 assert resp.status_code == status
606 assert resp.reason == reason
607 assert "foo" in resp.headers
608 assert "Content-Type" not in resp.headers
609
610 run()
611 assert_reset()
612
613
614 def test_callback_content_type_dict():
615 def request_callback(request):
616 return (
617 200,
618 {"Content-Type": "application/json"},
619 b"foo",
620 )
621
622 @responses.activate
623 def run():
624 responses.add_callback("GET", "http://mockhost/.foo", callback=request_callback)
625 resp = requests.get("http://mockhost/.foo")
626 assert resp.text == "foo"
627 assert resp.headers["content-type"] == "application/json"
628
629 run()
630 assert_reset()
631
632
633 def test_callback_content_type_tuple():
634 def request_callback(request):
635 return (
636 200,
637 [("Content-Type", "application/json")],
638 b"foo",
639 )
640
641 @responses.activate
642 def run():
643 responses.add_callback("GET", "http://mockhost/.foo", callback=request_callback)
644 resp = requests.get("http://mockhost/.foo")
645 assert resp.text == "foo"
646 assert resp.headers["content-type"] == "application/json"
647
648 run()
649 assert_reset()
650
651
652 def test_regular_expression_url():
653 @responses.activate
654 def run():
655 url = re.compile(r"https?://(.*\.)?example.com")
656 responses.add(responses.GET, url, body=b"test")
657
658 resp = requests.get("http://example.com")
659 assert_response(resp, "test")
660
661 resp = requests.get("https://example.com")
662 assert_response(resp, "test")
663
664 resp = requests.get("https://uk.example.com")
665 assert_response(resp, "test")
666
667 with pytest.raises(ConnectionError):
668 requests.get("https://uk.exaaample.com")
669
670 run()
671 assert_reset()
672
673
674 def test_custom_adapter():
675 @responses.activate
676 def run():
677 url = "http://example.com"
678 responses.add(responses.GET, url, body=b"test")
679
680 calls = [0]
681
682 class DummyAdapter(requests.adapters.HTTPAdapter):
683 def send(self, *a, **k):
684 calls[0] += 1
685 return super(DummyAdapter, self).send(*a, **k)
686
687 # Test that the adapter is actually used
688 session = requests.Session()
689 session.mount("http://", DummyAdapter())
690
691 resp = session.get(url, allow_redirects=False)
692 assert calls[0] == 1
693
694 # Test that the response is still correctly emulated
695 session = requests.Session()
696 session.mount("http://", DummyAdapter())
697
698 resp = session.get(url)
699 assert_response(resp, "test")
700
701 run()
702
703
704 def test_responses_as_context_manager():
705 def run():
706 with responses.mock:
707 responses.add(responses.GET, "http://example.com", body=b"test")
708 resp = requests.get("http://example.com")
709 assert_response(resp, "test")
710 assert len(responses.calls) == 1
711 assert responses.calls[0].request.url == "http://example.com/"
712 assert responses.calls[0].response.content == b"test"
713
714 resp = requests.get("http://example.com?foo=bar")
715 assert_response(resp, "test")
716 assert len(responses.calls) == 2
717 assert responses.calls[1].request.url == "http://example.com/?foo=bar"
718 assert responses.calls[1].response.content == b"test"
719
720 run()
721 assert_reset()
722
723
724 def test_activate_doesnt_change_signature():
725 def test_function(a, b=None):
726 return (a, b)
727
728 decorated_test_function = responses.activate(test_function)
729 if hasattr(inspect, "signature"):
730 assert inspect.signature(test_function) == inspect.signature(
731 decorated_test_function
732 )
733 else:
734 assert inspect.getargspec(test_function) == inspect.getargspec(
735 decorated_test_function
736 )
737 assert decorated_test_function(1, 2) == test_function(1, 2)
738 assert decorated_test_function(3) == test_function(3)
739
740
741 def test_activate_mock_interaction():
742 @patch("sys.stdout")
743 def test_function(mock_stdout):
744 return mock_stdout
745
746 decorated_test_function = responses.activate(test_function)
747 if hasattr(inspect, "signature"):
748 assert inspect.signature(test_function) == inspect.signature(
749 decorated_test_function
750 )
751 else:
752 assert inspect.getargspec(test_function) == inspect.getargspec(
753 decorated_test_function
754 )
755
756 value = test_function()
757 assert isinstance(value, Mock)
758
759 value = decorated_test_function()
760 assert isinstance(value, Mock)
761
762
763 @pytest.mark.skipif(six.PY2, reason="Cannot run in python2")
764 def test_activate_doesnt_change_signature_with_return_type():
765 def test_function(a, b=None):
766 return (a, b)
767
768 # Add type annotations as they are syntax errors in py2.
769 # Use a class to test for import errors in evaled code.
770 test_function.__annotations__["return"] = Mock
771 test_function.__annotations__["a"] = Mock
772
773 decorated_test_function = responses.activate(test_function)
774 if hasattr(inspect, "signature"):
775 assert inspect.signature(test_function) == inspect.signature(
776 decorated_test_function
777 )
778 else:
779 assert inspect.getargspec(test_function) == inspect.getargspec(
780 decorated_test_function
781 )
782 assert decorated_test_function(1, 2) == test_function(1, 2)
783 assert decorated_test_function(3) == test_function(3)
784
785
786 def test_activate_doesnt_change_signature_for_method():
787 class TestCase(object):
788 def test_function(self, a, b=None):
789 return (self, a, b)
790
791 decorated_test_function = responses.activate(test_function)
792
793 test_case = TestCase()
794 assert test_case.decorated_test_function(1, 2) == test_case.test_function(1, 2)
795 assert test_case.decorated_test_function(3) == test_case.test_function(3)
796
797
798 def test_response_cookies():
799 body = b"test callback"
800 status = 200
801 headers = {"set-cookie": "session_id=12345; a=b; c=d"}
802 url = "http://example.com/"
803
804 def request_callback(request):
805 return (status, headers, body)
806
807 @responses.activate
808 def run():
809 responses.add_callback(responses.GET, url, request_callback)
810 resp = requests.get(url)
811 assert resp.text == "test callback"
812 assert resp.status_code == status
813 assert "session_id" in resp.cookies
814 assert resp.cookies["session_id"] == "12345"
815 assert set(resp.cookies.keys()) == set(["session_id"])
816
817 run()
818 assert_reset()
819
820
821 def test_response_secure_cookies():
822 body = b"test callback"
823 status = 200
824 headers = {"set-cookie": "session_id=12345; a=b; c=d; secure"}
825 url = "http://example.com/"
826
827 def request_callback(request):
828 return (status, headers, body)
829
830 @responses.activate
831 def run():
832 responses.add_callback(responses.GET, url, request_callback)
833 resp = requests.get(url)
834 assert resp.text == "test callback"
835 assert resp.status_code == status
836 assert "session_id" in resp.cookies
837 assert resp.cookies["session_id"] == "12345"
838 assert set(resp.cookies.keys()) == set(["session_id"])
839
840 run()
841 assert_reset()
842
843
844 def test_response_cookies_multiple():
845 body = b"test callback"
846 status = 200
847 headers = [
848 ("set-cookie", "1P_JAR=2019-12-31-23; path=/; domain=.example.com; HttpOnly"),
849 ("set-cookie", "NID=some=value; path=/; domain=.example.com; secure"),
850 ]
851 url = "http://example.com/"
852
853 def request_callback(request):
854 return (status, headers, body)
855
856 @responses.activate
857 def run():
858 responses.add_callback(responses.GET, url, request_callback)
859 resp = requests.get(url)
860 assert resp.text == "test callback"
861 assert resp.status_code == status
862 assert set(resp.cookies.keys()) == set(["1P_JAR", "NID"])
863 assert resp.cookies["1P_JAR"] == "2019-12-31-23"
864 assert resp.cookies["NID"] == "some=value"
865
866 run()
867 assert_reset()
868
869
870 def test_response_callback():
871 """adds a callback to decorate the response, then checks it"""
872
873 def run():
874 def response_callback(resp):
875 resp._is_mocked = True
876 return resp
877
878 with responses.RequestsMock(response_callback=response_callback) as m:
879 m.add(responses.GET, "http://example.com", body=b"test")
880 resp = requests.get("http://example.com")
881 assert resp.text == "test"
882 assert hasattr(resp, "_is_mocked")
883 assert getattr(resp, "_is_mocked") is True
884
885 run()
886 assert_reset()
887
888
889 def test_response_filebody():
890 """ Adds the possibility to use actual (binary) files as responses """
891
892 def run():
893 with responses.RequestsMock() as m:
894 with open("README.rst", "r") as out:
895 m.add(responses.GET, "http://example.com", body=out.read(), stream=True)
896 resp = requests.get("http://example.com")
897 with open("README.rst", "r") as out:
898 assert resp.text == out.read()
899
900
901 def test_assert_all_requests_are_fired():
902 def request_callback(request):
903 raise BaseException()
904
905 def run():
906 with pytest.raises(AssertionError) as excinfo:
907 with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
908 m.add(responses.GET, "http://example.com", body=b"test")
909 assert "http://example.com" in str(excinfo.value)
910 assert responses.GET in str(excinfo.value)
911
912 # check that assert_all_requests_are_fired default to True
913 with pytest.raises(AssertionError):
914 with responses.RequestsMock() as m:
915 m.add(responses.GET, "http://example.com", body=b"test")
916
917 # check that assert_all_requests_are_fired doesn't swallow exceptions
918 with pytest.raises(ValueError):
919 with responses.RequestsMock() as m:
920 m.add(responses.GET, "http://example.com", body=b"test")
921 raise ValueError()
922
923 # check that assert_all_requests_are_fired=True doesn't remove urls
924 with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
925 m.add(responses.GET, "http://example.com", body=b"test")
926 assert len(m._matches) == 1
927 requests.get("http://example.com")
928 assert len(m._matches) == 1
929
930 # check that assert_all_requests_are_fired=True counts mocked errors
931 with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
932 m.add(responses.GET, "http://example.com", body=Exception())
933 assert len(m._matches) == 1
934 with pytest.raises(Exception):
935 requests.get("http://example.com")
936 assert len(m._matches) == 1
937
938 with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
939 m.add_callback(responses.GET, "http://example.com", request_callback)
940 assert len(m._matches) == 1
941 with pytest.raises(BaseException):
942 requests.get("http://example.com")
943 assert len(m._matches) == 1
944
945 run()
946 assert_reset()
947
948
949 def test_allow_redirects_samehost():
950 redirecting_url = "http://example.com"
951 final_url_path = "/1"
952 final_url = "{0}{1}".format(redirecting_url, final_url_path)
953 url_re = re.compile(r"^http://example.com(/)?(\d+)?$")
954
955 def request_callback(request):
956 # endpoint of chained redirect
957 if request.url.endswith(final_url_path):
958 return 200, (), b"test"
959
960 # otherwise redirect to an integer path
961 else:
962 if request.url.endswith("/0"):
963 n = 1
964 else:
965 n = 0
966 redirect_headers = {"location": "/{0!s}".format(n)}
967 return 301, redirect_headers, None
968
969 def run():
970 # setup redirect
971 with responses.mock:
972 responses.add_callback(responses.GET, url_re, request_callback)
973 resp_no_redirects = requests.get(redirecting_url, allow_redirects=False)
974 assert resp_no_redirects.status_code == 301
975 assert len(responses.calls) == 1 # 1x300
976 assert responses.calls[0][1].status_code == 301
977 assert_reset()
978
979 with responses.mock:
980 responses.add_callback(responses.GET, url_re, request_callback)
981 resp_yes_redirects = requests.get(redirecting_url, allow_redirects=True)
982 assert len(responses.calls) == 3 # 2x300 + 1x200
983 assert len(resp_yes_redirects.history) == 2
984 assert resp_yes_redirects.status_code == 200
985 assert final_url == resp_yes_redirects.url
986 status_codes = [call[1].status_code for call in responses.calls]
987 assert status_codes == [301, 301, 200]
988 assert_reset()
989
990 run()
991 assert_reset()
992
993
994 def test_handles_unicode_querystring():
995 url = "http://example.com/test?type=2&ie=utf8&query=汉字"
996
997 @responses.activate
998 def run():
999 responses.add(responses.GET, url, body="test", match_querystring=True)
1000
1001 resp = requests.get(url)
1002
1003 assert_response(resp, "test")
1004
1005 run()
1006 assert_reset()
1007
1008
1009 def test_handles_unicode_url():
1010 url = "http://www.संजाल.भारत/hi/वेबसाइट-डिजाइन"
1011
1012 @responses.activate
1013 def run():
1014 responses.add(responses.GET, url, body="test")
1015
1016 resp = requests.get(url)
1017
1018 assert_response(resp, "test")
1019
1020 run()
1021 assert_reset()
1022
1023
1024 def test_handles_unicode_body():
1025 url = "http://example.com/test"
1026
1027 @responses.activate
1028 def run():
1029 responses.add(responses.GET, url, body="михољско лето")
1030
1031 resp = requests.get(url)
1032
1033 assert_response(resp, "михољско лето", content_type="text/plain; charset=utf-8")
1034
1035 run()
1036 assert_reset()
1037
1038
1039 def test_handles_buffered_reader_body():
1040 url = "http://example.com/test"
1041
1042 @responses.activate
1043 def run():
1044 responses.add(responses.GET, url, body=BufferedReader(BytesIO(b"test"))) # type: ignore
1045
1046 resp = requests.get(url)
1047
1048 assert_response(resp, "test")
1049
1050 run()
1051 assert_reset()
1052
1053
1054 def test_headers():
1055 @responses.activate
1056 def run():
1057 responses.add(
1058 responses.GET, "http://example.com", body="", headers={"X-Test": "foo"}
1059 )
1060 resp = requests.get("http://example.com")
1061 assert resp.headers["X-Test"] == "foo"
1062
1063 run()
1064 assert_reset()
1065
1066
1067 def test_legacy_adding_headers():
1068 @responses.activate
1069 def run():
1070 responses.add(
1071 responses.GET,
1072 "http://example.com",
1073 body="",
1074 adding_headers={"X-Test": "foo"},
1075 )
1076 resp = requests.get("http://example.com")
1077 assert resp.headers["X-Test"] == "foo"
1078
1079 run()
1080 assert_reset()
1081
1082
1083 def test_multiple_responses():
1084 @responses.activate
1085 def run():
1086 responses.add(responses.GET, "http://example.com", body="test")
1087 responses.add(responses.GET, "http://example.com", body="rest")
1088
1089 resp = requests.get("http://example.com")
1090 assert_response(resp, "test")
1091 resp = requests.get("http://example.com")
1092 assert_response(resp, "rest")
1093 # After all responses are used, last response should be repeated
1094 resp = requests.get("http://example.com")
1095 assert_response(resp, "rest")
1096
1097 run()
1098 assert_reset()
1099
1100
1101 def test_multiple_urls():
1102 @responses.activate
1103 def run():
1104 responses.add(responses.GET, "http://example.com/one", body="one")
1105 responses.add(responses.GET, "http://example.com/two", body="two")
1106
1107 resp = requests.get("http://example.com/two")
1108 assert_response(resp, "two")
1109 resp = requests.get("http://example.com/one")
1110 assert_response(resp, "one")
1111
1112 run()
1113 assert_reset()
1114
1115
1116 def test_multiple_methods():
1117 @responses.activate
1118 def run():
1119 responses.add(responses.GET, "http://example.com/one", body="gotcha")
1120 responses.add(responses.POST, "http://example.com/one", body="posted")
1121
1122 resp = requests.get("http://example.com/one")
1123 assert_response(resp, "gotcha")
1124 resp = requests.post("http://example.com/one")
1125 assert_response(resp, "posted")
1126
1127 run()
1128 assert_reset()
1129
1130
1131 def test_passthru(httpserver):
1132 httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
1133
1134 @responses.activate
1135 def run():
1136 responses.add_passthru(httpserver.url)
1137 responses.add(responses.GET, "{}/one".format(httpserver.url), body="one")
1138 responses.add(responses.GET, "http://example.com/two", body="two")
1139
1140 resp = requests.get("http://example.com/two")
1141 assert_response(resp, "two")
1142 resp = requests.get("{}/one".format(httpserver.url))
1143 assert_response(resp, "one")
1144 resp = requests.get(httpserver.url)
1145 assert_response(resp, "OK")
1146
1147 run()
1148 assert_reset()
1149
1150
1151 def test_passthru_regex(httpserver):
1152 httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
1153
1154 @responses.activate
1155 def run():
1156 responses.add_passthru(re.compile("{}/\\w+".format(httpserver.url)))
1157 responses.add(responses.GET, "{}/one".format(httpserver.url), body="one")
1158 responses.add(responses.GET, "http://example.com/two", body="two")
1159
1160 resp = requests.get("http://example.com/two")
1161 assert_response(resp, "two")
1162 resp = requests.get("{}/one".format(httpserver.url))
1163 assert_response(resp, "one")
1164 resp = requests.get("{}/two".format(httpserver.url))
1165 assert_response(resp, "OK")
1166 resp = requests.get("{}/three".format(httpserver.url))
1167 assert_response(resp, "OK")
1168
1169 run()
1170 assert_reset()
1171
1172
1173 def test_method_named_param():
1174 @responses.activate
1175 def run():
1176 responses.add(method=responses.GET, url="http://example.com", body="OK")
1177 resp = requests.get("http://example.com")
1178 assert_response(resp, "OK")
1179
1180 run()
1181 assert_reset()
1182
1183
1184 def test_passthru_unicode():
1185 @responses.activate
1186 def run():
1187 with responses.RequestsMock() as m:
1188 url = "http://موقع.وزارة-الاتصالات.مصر/"
1189 clean_url = "http://xn--4gbrim.xn----ymcbaaajlc6dj7bxne2c.xn--wgbh1c/"
1190 m.add_passthru(url)
1191 assert m.passthru_prefixes[0] == clean_url
1192
1193 run()
1194 assert_reset()
1195
1196
1197 def test_custom_target(monkeypatch):
1198 requests_mock = responses.RequestsMock(target="something.else")
1199 std_mock_mock = responses.std_mock.MagicMock()
1200 patch_mock = std_mock_mock.patch
1201 monkeypatch.setattr(responses, "std_mock", std_mock_mock)
1202 requests_mock.start()
1203 assert len(patch_mock.call_args_list) == 1
1204 assert patch_mock.call_args[1]["target"] == "something.else"
1205
1206
1207 def _quote(s):
1208 return responses.quote(responses._ensure_str(s))
1209
1210
1211 def test_cookies_from_headers():
1212 text = "こんにちは/世界"
1213 quoted_text = _quote(text)
1214 expected = {"x": "a", "y": quoted_text}
1215 headers = {"set-cookie": "; ".join(k + "=" + v for k, v in expected.items())}
1216 cookiejar = responses._cookies_from_headers(headers)
1217 for k, v in cookiejar.items():
1218 assert isinstance(v, str)
1219 assert v == expected[k]
1220
1221
1222 @pytest.mark.parametrize(
1223 "url",
1224 (
1225 "http://example.com",
1226 "http://example.com/some/path",
1227 "http://example.com/other/path/",
1228 ),
1229 )
1230 def test_request_param(url):
1231 @responses.activate
1232 def run():
1233 params = {"hello": "world", "example": "params"}
1234 responses.add(
1235 method=responses.GET,
1236 url="{0}?hello=world".format(url),
1237 body="test",
1238 match_querystring=False,
1239 )
1240 resp = requests.get(url, params=params)
1241 assert_response(resp, "test")
1242 assert_params(resp, params)
1243
1244 resp = requests.get(url)
1245 assert_response(resp, "test")
1246 assert_params(resp, {})
1247
1248 run()
1249 assert_reset()
1250
1251
1252 def test_request_param_with_multiple_values_for_the_same_key():
1253 @responses.activate
1254 def run():
1255 url = "http://example.com"
1256 params = {"key1": ["one", "two"], "key2": "three"}
1257 responses.add(
1258 method=responses.GET,
1259 url=url,
1260 body="test",
1261 )
1262 resp = requests.get(url, params=params)
1263 assert_response(resp, "test")
1264 assert_params(resp, params)
1265
1266 run()
1267 assert_reset()
1268
1269
1270 @pytest.mark.parametrize(
1271 "url", ("http://example.com", "http://example.com?hello=world")
1272 )
1273 def test_assert_call_count(url):
1274 @responses.activate
1275 def run():
1276 responses.add(responses.GET, url)
1277 responses.add(responses.GET, "http://example1.com")
1278
1279 assert responses.assert_call_count(url, 0) is True
1280
1281 with pytest.raises(AssertionError) as excinfo:
1282 responses.assert_call_count(url, 2)
1283 assert "Expected URL '{0}' to be called 2 times. Called 0 times.".format(
1284 url
1285 ) in str(excinfo.value)
1286
1287 requests.get(url)
1288 assert responses.assert_call_count(url, 1) is True
1289
1290 requests.get("http://example1.com")
1291 assert responses.assert_call_count(url, 1) is True
1292
1293 requests.get(url)
1294 with pytest.raises(AssertionError) as excinfo:
1295 responses.assert_call_count(url, 3)
1296 assert "Expected URL '{0}' to be called 3 times. Called 2 times.".format(
1297 url
1298 ) in str(excinfo.value)
1299
1300 run()
1301 assert_reset()
1302
1303
1304 def test_request_matches_post_params():
1305 @responses.activate
1306 def run():
1307 responses.add(
1308 method=responses.POST,
1309 url="http://example.com/",
1310 body="one",
1311 match=[
1312 responses.json_params_matcher(
1313 {"page": {"name": "first", "type": "json"}}
1314 )
1315 ],
1316 )
1317 responses.add(
1318 method=responses.POST,
1319 url="http://example.com/",
1320 body="two",
1321 match=[
1322 responses.urlencoded_params_matcher(
1323 {"page": "second", "type": "urlencoded"}
1324 )
1325 ],
1326 )
1327
1328 resp = requests.request(
1329 "POST",
1330 "http://example.com/",
1331 headers={"Content-Type": "x-www-form-urlencoded"},
1332 data={"page": "second", "type": "urlencoded"},
1333 )
1334 assert_response(resp, "two")
1335
1336 resp = requests.request(
1337 "POST",
1338 "http://example.com/",
1339 headers={"Content-Type": "application/json"},
1340 json={"page": {"name": "first", "type": "json"}},
1341 )
1342 assert_response(resp, "one")
1343
1344 run()
1345 assert_reset()
1346
1347
1348 def test_request_matches_empty_body():
1349 @responses.activate
1350 def run():
1351 responses.add(
1352 method=responses.POST,
1353 url="http://example.com/",
1354 body="one",
1355 match=[responses.json_params_matcher(None)],
1356 )
1357
1358 responses.add(
1359 method=responses.POST,
1360 url="http://example.com/",
1361 body="two",
1362 match=[responses.urlencoded_params_matcher(None)],
1363 )
1364
1365 resp = requests.request("POST", "http://example.com/")
1366 assert_response(resp, "one")
1367
1368 resp = requests.request(
1369 "POST",
1370 "http://example.com/",
1371 headers={"Content-Type": "x-www-form-urlencoded"},
1372 )
1373 assert_response(resp, "two")
1374
1375 run()
1376 assert_reset()
1377
1378
1379 def test_fail_request_error():
1380 @responses.activate
1381 def run():
1382 responses.add("POST", "http://example1.com")
1383 responses.add("GET", "http://example.com")
1384 responses.add(
1385 "POST",
1386 "http://example.com",
1387 match=[responses.urlencoded_params_matcher({"foo": "bar"})],
1388 )
1389
1390 with pytest.raises(ConnectionError) as excinfo:
1391 requests.post("http://example.com", data={"id": "bad"})
1392 msg = str(excinfo.value)
1393 assert "- POST http://example1.com/ URL does not match" in msg
1394 assert "- GET http://example.com/ Method does not match" in msg
1395 assert "- POST http://example.com/ Parameters do not match" in msg
1396
1397 run()
1398 assert_reset()
1399
1400
1401 @pytest.mark.parametrize(
1402 "response_params, expected_representation",
1403 [
1404 (
1405 {"method": responses.GET, "url": "http://example.com/"},
1406 (
1407 "<Response(url='http://example.com/' status=200 "
1408 "content_type='text/plain' headers='null')>"
1409 ),
1410 ),
1411 (
1412 {
1413 "method": responses.POST,
1414 "url": "http://another-domain.com/",
1415 "content_type": "application/json",
1416 "status": 404,
1417 },
1418 (
1419 "<Response(url='http://another-domain.com/' status=404 "
1420 "content_type='application/json' headers='null')>"
1421 ),
1422 ),
1423 (
1424 {
1425 "method": responses.PUT,
1426 "url": "http://abcd.com/",
1427 "content_type": "text/html",
1428 "status": 500,
1429 "headers": {"X-Test": "foo"},
1430 "body": {"it_wont_be": "considered"},
1431 },
1432 (
1433 "<Response(url='http://abcd.com/' status=500 "
1434 "content_type='text/html' headers='{\"X-Test\": \"foo\"}')>"
1435 ),
1436 ),
1437 ],
1438 )
1439 def test_response_representations(response_params, expected_representation):
1440 response = Response(**response_params)
1441
1442 assert str(response) == expected_representation
1443 assert repr(response) == expected_representation
1444
1445
1446 def test_mocked_responses_list_registered():
1447 @responses.activate
1448 def run():
1449 first_response = Response(
1450 responses.GET,
1451 "http://example.com/",
1452 body="",
1453 headers={"X-Test": "foo"},
1454 status=404,
1455 )
1456 second_response = Response(
1457 responses.GET, "http://example.com/", body="", headers={"X-Test": "foo"}
1458 )
1459 third_response = Response(
1460 responses.POST,
1461 "http://anotherdomain.com/",
1462 )
1463 responses.add(first_response)
1464 responses.add(second_response)
1465 responses.add(third_response)
1466
1467 mocks_list = responses.registered()
1468
1469 assert mocks_list == responses.mock._matches
1470 assert mocks_list == [first_response, second_response, third_response]
1471
1472 run()
1473 assert_reset()
00 Metadata-Version: 2.1
11 Name: responses
2 Version: 0.12.1
2 Version: 0.13.3
33 Summary: A utility library for mocking out the `requests` Python library.
44 Home-page: https://github.com/getsentry/responses
55 Author: David Cramer
154154 responses.add(
155155 responses.POST,
156156 url='http://calc.com/sum',
157 body=4,
157 body="4",
158158 match=[
159 responses.urlencoded_params_matcher({"left": 1, "right": 3})
159 responses.urlencoded_params_matcher({"left": "1", "right": "3"})
160160 ]
161161 )
162162 requests.post("http://calc.com/sum", data={"left": 1, "right": 3})
337337 resp = requests.get('http://twitter.com/api/1/foobar')
338338 assert resp.status_code == 200
339339
340 Responses inside a unittest setUp()
341 -----------------------------------
342
343 When run with unittest tests, this can be used to set up some
344 generic class-level responses, that may be complemented by each test
345
346 .. code-block:: python
347
348 def setUp():
349 self.responses = responses.RequestsMock()
350 self.responses.start()
351
352 # self.responses.add(...)
353
354 self.addCleanup(self.responses.stop)
355 self.addCleanup(self.responses.reset)
356
357 def test_api(self):
358 self.responses.add(
359 responses.GET, 'http://twitter.com/api/1/foobar',
360 body='{}', status=200,
361 content_type='application/json')
362 resp = requests.get('http://twitter.com/api/1/foobar')
363 assert resp.status_code == 200
364
340365 Assertions on declared responses
341366 --------------------------------
342367
456481 Viewing/Modifying registered responses
457482 --------------------------------------
458483
459 Registered responses are available as a private attribute of the RequestMock
484 Registered responses are available as a public method of the RequestMock
460485 instance. It is sometimes useful for debugging purposes to view the stack of
461 registered responses which can be accessed via ``responses.mock._matches``.
486 registered responses which can be accessed via ``responses.registered()``.
462487
463488 The ``replace`` function allows a previously registered ``response`` to be
464489 changed. The method signature is identical to ``add``. ``response`` s are
480505
481506 assert resp.json() == {'data': 2}
482507
508
509 The ``upsert`` function allows a previously registered ``response`` to be
510 changed like ``replace``. If the response is registered, the ``upsert`` function
511 will registered it like ``add``.
483512
484513 ``remove`` takes a ``method`` and ``url`` argument and will remove **all**
485514 matched responses from the registered list.
11 LICENSE
22 MANIFEST.in
33 README.rst
4 responses.py
54 setup.cfg
65 setup.py
7 test_responses.py
86 tox.ini
7 responses/__init__.py
8 responses/test_responses.py
99 responses.egg-info/PKG-INFO
1010 responses.egg-info/SOURCES.txt
1111 responses.egg-info/dependency_links.txt
00 requests>=2.0
1 six
12 urllib3>=1.25.10
2 six
33
44 [:python_version < "3.3"]
55 mock
99
1010 [tests]
1111 coverage<6.0.0,>=3.7.1
12 flake8
1213 pytest-cov
1314 pytest-localserver
14 flake8
1515
1616 [tests:python_version < "3.5"]
1717 pytest<5.0,>=4.6
1818
1919 [tests:python_version >= "3.5"]
20 mypy
2021 pytest>=4.6
+0
-824
responses.py less more
0 from __future__ import absolute_import, print_function, division, unicode_literals
1
2 import _io
3 import inspect
4 import json as json_module
5 import logging
6 import re
7 from itertools import groupby
8
9 import six
10
11 from collections import namedtuple
12 from functools import update_wrapper
13 from requests.adapters import HTTPAdapter
14 from requests.exceptions import ConnectionError
15 from requests.sessions import REDIRECT_STATI
16 from requests.utils import cookiejar_from_dict
17
18 try:
19 from collections.abc import Sequence, Sized
20 except ImportError:
21 from collections import Sequence, Sized
22
23 try:
24 from requests.packages.urllib3.response import HTTPResponse
25 except ImportError:
26 from urllib3.response import HTTPResponse
27 try:
28 from requests.packages.urllib3.connection import HTTPHeaderDict
29 except ImportError:
30 from urllib3.response import HTTPHeaderDict
31
32 if six.PY2:
33 from urlparse import urlparse, parse_qsl, urlsplit, urlunsplit
34 from urllib import quote
35 else:
36 from urllib.parse import urlparse, parse_qsl, urlsplit, urlunsplit, quote
37
38 if six.PY2:
39 try:
40 from six import cStringIO as BufferIO
41 except ImportError:
42 from six import StringIO as BufferIO
43 else:
44 from io import BytesIO as BufferIO
45
46 try:
47 from unittest import mock as std_mock
48 except ImportError:
49 import mock as std_mock
50
51 try:
52 Pattern = re._pattern_type
53 except AttributeError:
54 # Python 3.7
55 Pattern = re.Pattern
56
57 try:
58 from json.decoder import JSONDecodeError
59 except ImportError:
60 JSONDecodeError = ValueError
61
62 UNSET = object()
63
64 Call = namedtuple("Call", ["request", "response"])
65
66 _real_send = HTTPAdapter.send
67
68 logger = logging.getLogger("responses")
69
70
71 def _is_string(s):
72 return isinstance(s, six.string_types)
73
74
75 def _has_unicode(s):
76 return any(ord(char) > 128 for char in s)
77
78
79 def _clean_unicode(url):
80 # Clean up domain names, which use punycode to handle unicode chars
81 urllist = list(urlsplit(url))
82 netloc = urllist[1]
83 if _has_unicode(netloc):
84 domains = netloc.split(".")
85 for i, d in enumerate(domains):
86 if _has_unicode(d):
87 d = "xn--" + d.encode("punycode").decode("ascii")
88 domains[i] = d
89 urllist[1] = ".".join(domains)
90 url = urlunsplit(urllist)
91
92 # Clean up path/query/params, which use url-encoding to handle unicode chars
93 if isinstance(url.encode("utf8"), six.string_types):
94 url = url.encode("utf8")
95 chars = list(url)
96 for i, x in enumerate(chars):
97 if ord(x) > 128:
98 chars[i] = quote(x)
99
100 return "".join(chars)
101
102
103 def _is_redirect(response):
104 try:
105 # 2.0.0 <= requests <= 2.2
106 return response.is_redirect
107
108 except AttributeError:
109 # requests > 2.2
110 return (
111 # use request.sessions conditional
112 response.status_code in REDIRECT_STATI
113 and "location" in response.headers
114 )
115
116
117 def _ensure_str(s):
118 if six.PY2:
119 s = s.encode("utf-8") if isinstance(s, six.text_type) else s
120 return s
121
122
123 def _cookies_from_headers(headers):
124 try:
125 import http.cookies as cookies
126
127 resp_cookie = cookies.SimpleCookie()
128 resp_cookie.load(headers["set-cookie"])
129
130 cookies_dict = {name: v.value for name, v in resp_cookie.items()}
131 except ImportError:
132 from cookies import Cookies
133
134 resp_cookies = Cookies.from_request(_ensure_str(headers["set-cookie"]))
135 cookies_dict = {
136 v.name: quote(_ensure_str(v.value)) for _, v in resp_cookies.items()
137 }
138 return cookiejar_from_dict(cookies_dict)
139
140
141 _wrapper_template = """\
142 def wrapper%(wrapper_args)s:
143 with responses:
144 return func%(func_args)s
145 """
146
147
148 def get_wrapped(func, responses):
149 if six.PY2:
150 args, a, kw, defaults = inspect.getargspec(func)
151 wrapper_args = inspect.formatargspec(args, a, kw, defaults)
152
153 # Preserve the argspec for the wrapped function so that testing
154 # tools such as pytest can continue to use their fixture injection.
155 if hasattr(func, "__self__"):
156 args = args[1:] # Omit 'self'
157 func_args = inspect.formatargspec(args, a, kw, None)
158 else:
159 signature = inspect.signature(func)
160 signature = signature.replace(return_annotation=inspect.Signature.empty)
161 # If the function is wrapped, switch to *args, **kwargs for the parameters
162 # as we can't rely on the signature to give us the arguments the function will
163 # be called with. For example unittest.mock.patch uses required args that are
164 # not actually passed to the function when invoked.
165 if hasattr(func, "__wrapped__"):
166 wrapper_params = [
167 inspect.Parameter("args", inspect.Parameter.VAR_POSITIONAL),
168 inspect.Parameter("kwargs", inspect.Parameter.VAR_KEYWORD),
169 ]
170 else:
171 wrapper_params = [
172 param.replace(annotation=inspect.Parameter.empty)
173 for param in signature.parameters.values()
174 ]
175 signature = signature.replace(parameters=wrapper_params)
176
177 wrapper_args = str(signature)
178 params_without_defaults = [
179 param.replace(
180 annotation=inspect.Parameter.empty, default=inspect.Parameter.empty
181 )
182 for param in signature.parameters.values()
183 ]
184 signature = signature.replace(parameters=params_without_defaults)
185 func_args = str(signature)
186
187 evaldict = {"func": func, "responses": responses}
188 six.exec_(
189 _wrapper_template % {"wrapper_args": wrapper_args, "func_args": func_args},
190 evaldict,
191 )
192 wrapper = evaldict["wrapper"]
193 update_wrapper(wrapper, func)
194 return wrapper
195
196
197 class CallList(Sequence, Sized):
198 def __init__(self):
199 self._calls = []
200
201 def __iter__(self):
202 return iter(self._calls)
203
204 def __len__(self):
205 return len(self._calls)
206
207 def __getitem__(self, idx):
208 return self._calls[idx]
209
210 def add(self, request, response):
211 self._calls.append(Call(request, response))
212
213 def reset(self):
214 self._calls = []
215
216
217 def _ensure_url_default_path(url):
218 if _is_string(url):
219 url_parts = list(urlsplit(url))
220 if url_parts[2] == "":
221 url_parts[2] = "/"
222 url = urlunsplit(url_parts)
223 return url
224
225
226 def _handle_body(body):
227 if isinstance(body, six.text_type):
228 body = body.encode("utf-8")
229 if isinstance(body, _io.BufferedReader):
230 return body
231
232 return BufferIO(body)
233
234
235 _unspecified = object()
236
237
238 def urlencoded_params_matcher(params):
239 def match(request_body):
240 return (
241 params is None
242 if request_body is None
243 else sorted(params.items()) == sorted(parse_qsl(request_body))
244 )
245
246 return match
247
248
249 def json_params_matcher(params):
250 def match(request_body):
251 try:
252 if isinstance(request_body, bytes):
253 request_body = request_body.decode("utf-8")
254 return (
255 params is None
256 if request_body is None
257 else params == json_module.loads(request_body)
258 )
259 except JSONDecodeError:
260 return False
261
262 return match
263
264
265 class BaseResponse(object):
266 content_type = None
267 headers = None
268
269 stream = False
270
271 def __init__(self, method, url, match_querystring=_unspecified, match=[]):
272 self.method = method
273 # ensure the url has a default path set if the url is a string
274 self.url = _ensure_url_default_path(url)
275 self.match_querystring = self._should_match_querystring(match_querystring)
276 self.match = match
277 self.call_count = 0
278
279 def __eq__(self, other):
280 if not isinstance(other, BaseResponse):
281 return False
282
283 if self.method != other.method:
284 return False
285
286 # Can't simply do an equality check on the objects directly here since __eq__ isn't
287 # implemented for regex. It might seem to work as regex is using a cache to return
288 # the same regex instances, but it doesn't in all cases.
289 self_url = self.url.pattern if isinstance(self.url, Pattern) else self.url
290 other_url = other.url.pattern if isinstance(other.url, Pattern) else other.url
291
292 return self_url == other_url
293
294 def __ne__(self, other):
295 return not self.__eq__(other)
296
297 def _url_matches_strict(self, url, other):
298 url_parsed = urlparse(url)
299 other_parsed = urlparse(other)
300
301 if url_parsed[:3] != other_parsed[:3]:
302 return False
303
304 url_qsl = sorted(parse_qsl(url_parsed.query))
305 other_qsl = sorted(parse_qsl(other_parsed.query))
306
307 return url_qsl == other_qsl
308
309 def _should_match_querystring(self, match_querystring_argument):
310 if match_querystring_argument is not _unspecified:
311 return match_querystring_argument
312
313 if isinstance(self.url, Pattern):
314 # the old default from <= 0.9.0
315 return False
316
317 return bool(urlparse(self.url).query)
318
319 def _url_matches(self, url, other, match_querystring=False):
320 if _is_string(url):
321 if _has_unicode(url):
322 url = _clean_unicode(url)
323 if not isinstance(other, six.text_type):
324 other = other.encode("ascii").decode("utf8")
325 if match_querystring:
326 return self._url_matches_strict(url, other)
327
328 else:
329 url_without_qs = url.split("?", 1)[0]
330 other_without_qs = other.split("?", 1)[0]
331 return url_without_qs == other_without_qs
332
333 elif isinstance(url, Pattern) and url.match(other):
334 return True
335
336 else:
337 return False
338
339 def _body_matches(self, match, request_body):
340 for matcher in match:
341 if not matcher(request_body):
342 return False
343
344 return True
345
346 def get_headers(self):
347 headers = HTTPHeaderDict() # Duplicate headers are legal
348 if self.content_type is not None:
349 headers["Content-Type"] = self.content_type
350 if self.headers:
351 headers.extend(self.headers)
352 return headers
353
354 def get_response(self, request):
355 raise NotImplementedError
356
357 def matches(self, request):
358 if request.method != self.method:
359 return False, "Method does not match"
360
361 if not self._url_matches(self.url, request.url, self.match_querystring):
362 return False, "URL does not match"
363
364 if not self._body_matches(self.match, request.body):
365 return False, "Parameters do not match"
366
367 return True, ""
368
369
370 class Response(BaseResponse):
371 def __init__(
372 self,
373 method,
374 url,
375 body="",
376 json=None,
377 status=200,
378 headers=None,
379 stream=False,
380 content_type=UNSET,
381 **kwargs
382 ):
383 # if we were passed a `json` argument,
384 # override the body and content_type
385 if json is not None:
386 assert not body
387 body = json_module.dumps(json)
388 if content_type is UNSET:
389 content_type = "application/json"
390
391 if content_type is UNSET:
392 if isinstance(body, six.text_type) and _has_unicode(body):
393 content_type = "text/plain; charset=utf-8"
394 else:
395 content_type = "text/plain"
396
397 self.body = body
398 self.status = status
399 self.headers = headers
400 self.stream = stream
401 self.content_type = content_type
402 super(Response, self).__init__(method, url, **kwargs)
403
404 def get_response(self, request):
405 if self.body and isinstance(self.body, Exception):
406 raise self.body
407
408 headers = self.get_headers()
409 status = self.status
410 body = _handle_body(self.body)
411 return HTTPResponse(
412 status=status,
413 reason=six.moves.http_client.responses.get(status),
414 body=body,
415 headers=headers,
416 original_response=OriginalResponseShim(headers),
417 preload_content=False,
418 )
419
420
421 class CallbackResponse(BaseResponse):
422 def __init__(
423 self, method, url, callback, stream=False, content_type="text/plain", **kwargs
424 ):
425 self.callback = callback
426 self.stream = stream
427 self.content_type = content_type
428 super(CallbackResponse, self).__init__(method, url, **kwargs)
429
430 def get_response(self, request):
431 headers = self.get_headers()
432
433 result = self.callback(request)
434 if isinstance(result, Exception):
435 raise result
436
437 status, r_headers, body = result
438 if isinstance(body, Exception):
439 raise body
440
441 # If the callback set a content-type remove the one
442 # set in add_callback() so that we don't have multiple
443 # content type values.
444 has_content_type = False
445 if isinstance(r_headers, dict) and "Content-Type" in r_headers:
446 has_content_type = True
447 elif isinstance(r_headers, list):
448 has_content_type = any(
449 [h for h in r_headers if h and h[0].lower() == "content-type"]
450 )
451 if has_content_type:
452 headers.pop("Content-Type", None)
453
454 body = _handle_body(body)
455 headers.extend(r_headers)
456
457 return HTTPResponse(
458 status=status,
459 reason=six.moves.http_client.responses.get(status),
460 body=body,
461 headers=headers,
462 original_response=OriginalResponseShim(headers),
463 preload_content=False,
464 )
465
466
467 class OriginalResponseShim(object):
468 """
469 Shim for compatibility with older versions of urllib3
470
471 requests cookie handling depends on responses having a property chain of
472 `response._original_response.msg` which contains the response headers [1]
473
474 Using HTTPResponse() for this purpose causes compatibility errors with
475 urllib3<1.23.0. To avoid adding more dependencies we can use this shim.
476
477 [1]: https://github.com/psf/requests/blob/75bdc998e2d/requests/cookies.py#L125
478 """
479
480 def __init__(self, headers):
481 self.msg = headers
482
483 def isclosed(self):
484 return True
485
486
487 class RequestsMock(object):
488 DELETE = "DELETE"
489 GET = "GET"
490 HEAD = "HEAD"
491 OPTIONS = "OPTIONS"
492 PATCH = "PATCH"
493 POST = "POST"
494 PUT = "PUT"
495 response_callback = None
496
497 def __init__(
498 self,
499 assert_all_requests_are_fired=True,
500 response_callback=None,
501 passthru_prefixes=(),
502 target="requests.adapters.HTTPAdapter.send",
503 ):
504 self._calls = CallList()
505 self.reset()
506 self.assert_all_requests_are_fired = assert_all_requests_are_fired
507 self.response_callback = response_callback
508 self.passthru_prefixes = tuple(passthru_prefixes)
509 self.target = target
510
511 def reset(self):
512 self._matches = []
513 self._calls.reset()
514
515 def add(
516 self,
517 method=None, # method or ``Response``
518 url=None,
519 body="",
520 adding_headers=None,
521 *args,
522 **kwargs
523 ):
524 """
525 A basic request:
526
527 >>> responses.add(responses.GET, 'http://example.com')
528
529 You can also directly pass an object which implements the
530 ``BaseResponse`` interface:
531
532 >>> responses.add(Response(...))
533
534 A JSON payload:
535
536 >>> responses.add(
537 >>> method='GET',
538 >>> url='http://example.com',
539 >>> json={'foo': 'bar'},
540 >>> )
541
542 Custom headers:
543
544 >>> responses.add(
545 >>> method='GET',
546 >>> url='http://example.com',
547 >>> headers={'X-Header': 'foo'},
548 >>> )
549
550
551 Strict query string matching:
552
553 >>> responses.add(
554 >>> method='GET',
555 >>> url='http://example.com?foo=bar',
556 >>> match_querystring=True
557 >>> )
558 """
559 if isinstance(method, BaseResponse):
560 self._matches.append(method)
561 return
562
563 if adding_headers is not None:
564 kwargs.setdefault("headers", adding_headers)
565
566 self._matches.append(Response(method=method, url=url, body=body, **kwargs))
567
568 def add_passthru(self, prefix):
569 """
570 Register a URL prefix or regex to passthru any non-matching mock requests to.
571
572 For example, to allow any request to 'https://example.com', but require
573 mocks for the remainder, you would add the prefix as so:
574
575 >>> responses.add_passthru('https://example.com')
576
577 Regex can be used like:
578
579 >>> responses.add_passthru(re.compile('https://example.com/\\w+'))
580 """
581 if not isinstance(prefix, Pattern) and _has_unicode(prefix):
582 prefix = _clean_unicode(prefix)
583 self.passthru_prefixes += (prefix,)
584
585 def remove(self, method_or_response=None, url=None):
586 """
587 Removes a response previously added using ``add()``, identified
588 either by a response object inheriting ``BaseResponse`` or
589 ``method`` and ``url``. Removes all matching responses.
590
591 >>> response.add(responses.GET, 'http://example.org')
592 >>> response.remove(responses.GET, 'http://example.org')
593 """
594 if isinstance(method_or_response, BaseResponse):
595 response = method_or_response
596 else:
597 response = BaseResponse(method=method_or_response, url=url)
598
599 while response in self._matches:
600 self._matches.remove(response)
601
602 def replace(self, method_or_response=None, url=None, body="", *args, **kwargs):
603 """
604 Replaces a response previously added using ``add()``. The signature
605 is identical to ``add()``. The response is identified using ``method``
606 and ``url``, and the first matching response is replaced.
607
608 >>> responses.add(responses.GET, 'http://example.org', json={'data': 1})
609 >>> responses.replace(responses.GET, 'http://example.org', json={'data': 2})
610 """
611 if isinstance(method_or_response, BaseResponse):
612 response = method_or_response
613 else:
614 response = Response(method=method_or_response, url=url, body=body, **kwargs)
615
616 index = self._matches.index(response)
617 self._matches[index] = response
618
619 def add_callback(
620 self, method, url, callback, match_querystring=False, content_type="text/plain"
621 ):
622 # ensure the url has a default path set if the url is a string
623 # url = _ensure_url_default_path(url, match_querystring)
624
625 self._matches.append(
626 CallbackResponse(
627 url=url,
628 method=method,
629 callback=callback,
630 content_type=content_type,
631 match_querystring=match_querystring,
632 )
633 )
634
635 @property
636 def calls(self):
637 return self._calls
638
639 def __enter__(self):
640 self.start()
641 return self
642
643 def __exit__(self, type, value, traceback):
644 success = type is None
645 self.stop(allow_assert=success)
646 self.reset()
647 return success
648
649 def activate(self, func):
650 return get_wrapped(func, self)
651
652 def _find_match(self, request):
653 found = None
654 found_match = None
655 match_failed_reasons = []
656 for i, match in enumerate(self._matches):
657 match_result, reason = match.matches(request)
658 if match_result:
659 if found is None:
660 found = i
661 found_match = match
662 else:
663 # Multiple matches found. Remove & return the first match.
664 return self._matches.pop(found), match_failed_reasons
665 else:
666 match_failed_reasons.append(reason)
667 return found_match, match_failed_reasons
668
669 def _parse_request_params(self, url):
670 params = {}
671 for key, val in groupby(parse_qsl(urlparse(url).query), lambda kv: kv[0]):
672 values = list(map(lambda x: x[1], val))
673 if len(values) == 1:
674 values = values[0]
675 params[key] = values
676 return params
677
678 def _on_request(self, adapter, request, **kwargs):
679 match, match_failed_reasons = self._find_match(request)
680 resp_callback = self.response_callback
681 request.params = self._parse_request_params(request.path_url)
682
683 if match is None:
684 if any(
685 [
686 p.match(request.url)
687 if isinstance(p, Pattern)
688 else request.url.startswith(p)
689 for p in self.passthru_prefixes
690 ]
691 ):
692 logger.info("request.allowed-passthru", extra={"url": request.url})
693 return _real_send(adapter, request, **kwargs)
694
695 error_msg = (
696 "Connection refused by Responses - the call doesn't "
697 "match any registered mock.\n\n"
698 "Request: \n"
699 "- %s %s\n\n"
700 "Available matches:\n" % (request.method, request.url)
701 )
702 for i, m in enumerate(self._matches):
703 error_msg += "- {} {} {}\n".format(
704 m.method, m.url, match_failed_reasons[i]
705 )
706
707 response = ConnectionError(error_msg)
708 response.request = request
709
710 self._calls.add(request, response)
711 response = resp_callback(response) if resp_callback else response
712 raise response
713
714 try:
715 response = adapter.build_response(request, match.get_response(request))
716 except BaseException as response:
717 match.call_count += 1
718 self._calls.add(request, response)
719 response = resp_callback(response) if resp_callback else response
720 raise
721
722 if not match.stream:
723 response.content # NOQA
724
725 response = resp_callback(response) if resp_callback else response
726 match.call_count += 1
727 self._calls.add(request, response)
728 return response
729
730 def start(self):
731 def unbound_on_send(adapter, request, *a, **kwargs):
732 return self._on_request(adapter, request, *a, **kwargs)
733
734 self._patcher = std_mock.patch(target=self.target, new=unbound_on_send)
735 self._patcher.start()
736
737 def stop(self, allow_assert=True):
738 self._patcher.stop()
739 if not self.assert_all_requests_are_fired:
740 return
741
742 if not allow_assert:
743 return
744
745 not_called = [m for m in self._matches if m.call_count == 0]
746 if not_called:
747 raise AssertionError(
748 "Not all requests have been executed {0!r}".format(
749 [(match.method, match.url) for match in not_called]
750 )
751 )
752
753 def assert_call_count(self, url, count):
754 call_count = len(
755 [
756 1
757 for call in self.calls
758 if call.request.url == _ensure_url_default_path(url)
759 ]
760 )
761 if call_count == count:
762 return True
763 else:
764 raise AssertionError(
765 "Expected URL '{0}' to be called {1} times. Called {2} times.".format(
766 url, count, call_count
767 )
768 )
769
770
771 # expose default mock namespace
772 mock = _default_mock = RequestsMock(assert_all_requests_are_fired=False)
773 __all__ = [
774 "CallbackResponse",
775 "Response",
776 "RequestsMock",
777 # Exposed by the RequestsMock class:
778 "activate",
779 "add",
780 "add_callback",
781 "add_passthru",
782 "assert_all_requests_are_fired",
783 "assert_call_count",
784 "calls",
785 "DELETE",
786 "GET",
787 "HEAD",
788 "OPTIONS",
789 "passthru_prefixes",
790 "PATCH",
791 "POST",
792 "PUT",
793 "remove",
794 "replace",
795 "reset",
796 "response_callback",
797 "start",
798 "stop",
799 "target",
800 ]
801
802 activate = _default_mock.activate
803 add = _default_mock.add
804 add_callback = _default_mock.add_callback
805 add_passthru = _default_mock.add_passthru
806 assert_all_requests_are_fired = _default_mock.assert_all_requests_are_fired
807 assert_call_count = _default_mock.assert_call_count
808 calls = _default_mock.calls
809 DELETE = _default_mock.DELETE
810 GET = _default_mock.GET
811 HEAD = _default_mock.HEAD
812 OPTIONS = _default_mock.OPTIONS
813 passthru_prefixes = _default_mock.passthru_prefixes
814 PATCH = _default_mock.PATCH
815 POST = _default_mock.POST
816 PUT = _default_mock.PUT
817 remove = _default_mock.remove
818 replace = _default_mock.replace
819 reset = _default_mock.reset
820 response_callback = _default_mock.response_callback
821 start = _default_mock.start
822 stop = _default_mock.stop
823 target = _default_mock.target
3333 "pytest-cov",
3434 "pytest-localserver",
3535 "flake8",
36 "mypy; python_version >= '3.5'",
3637 ]
3738
3839 extras_require = {"tests": tests_require}
5455
5556 setup(
5657 name="responses",
57 version="0.12.1",
58 version="0.13.3",
5859 author="David Cramer",
5960 description=("A utility library for mocking out the `requests` Python library."),
6061 url="https://github.com/getsentry/responses",
6162 license="Apache 2.0",
6263 long_description=open("README.rst").read(),
6364 long_description_content_type="text/x-rst",
64 py_modules=["responses"],
65 packages=["responses"],
6566 zip_safe=False,
6667 python_requires=">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*",
6768 install_requires=install_requires,
6970 tests_require=tests_require,
7071 setup_requires=setup_requires,
7172 cmdclass={"test": PyTest},
73 package_data={"responses": ["py.typed", "__init__.pyi"]},
7274 include_package_data=True,
7375 classifiers=[
7476 "Intended Audience :: Developers",
+0
-1301
test_responses.py less more
0 # coding: utf-8
1
2 from __future__ import absolute_import, print_function, division, unicode_literals
3
4 import inspect
5 import re
6 import six
7 from io import BufferedReader, BytesIO
8
9 import pytest
10 import requests
11 import responses
12 from requests.exceptions import ConnectionError, HTTPError
13 from responses import BaseResponse, Response
14
15 try:
16 from mock import patch, Mock
17 except ImportError:
18 from unittest.mock import patch, Mock
19
20
21 def assert_reset():
22 assert len(responses._default_mock._matches) == 0
23 assert len(responses.calls) == 0
24
25
26 def assert_response(resp, body=None, content_type="text/plain"):
27 assert resp.status_code == 200
28 assert resp.reason == "OK"
29 if content_type is not None:
30 assert resp.headers["Content-Type"] == content_type
31 else:
32 assert "Content-Type" not in resp.headers
33 assert resp.text == body
34
35
36 def test_response():
37 @responses.activate
38 def run():
39 responses.add(responses.GET, "http://example.com", body=b"test")
40 resp = requests.get("http://example.com")
41 assert_response(resp, "test")
42 assert len(responses.calls) == 1
43 assert responses.calls[0].request.url == "http://example.com/"
44 assert responses.calls[0].response.content == b"test"
45
46 resp = requests.get("http://example.com?foo=bar")
47 assert_response(resp, "test")
48 assert len(responses.calls) == 2
49 assert responses.calls[1].request.url == "http://example.com/?foo=bar"
50 assert responses.calls[1].response.content == b"test"
51
52 run()
53 assert_reset()
54
55
56 def test_response_encoded():
57 @responses.activate
58 def run():
59 # Path contains urlencoded =/()[]
60 url = "http://example.org/foo.bar%3D%2F%28%29%5B%5D"
61 responses.add(responses.GET, url, body="it works", status=200)
62 resp = requests.get(url)
63 assert_response(resp, "it works")
64
65 run()
66 assert_reset()
67
68
69 def test_response_with_instance():
70 @responses.activate
71 def run():
72 responses.add(
73 responses.Response(method=responses.GET, url="http://example.com")
74 )
75 resp = requests.get("http://example.com")
76 assert_response(resp, "")
77 assert len(responses.calls) == 1
78 assert responses.calls[0].request.url == "http://example.com/"
79
80 resp = requests.get("http://example.com?foo=bar")
81 assert_response(resp, "")
82 assert len(responses.calls) == 2
83 assert responses.calls[1].request.url == "http://example.com/?foo=bar"
84
85
86 @pytest.mark.parametrize(
87 "original,replacement",
88 [
89 ("http://example.com/two", "http://example.com/two"),
90 (
91 Response(method=responses.GET, url="http://example.com/two"),
92 Response(
93 method=responses.GET, url="http://example.com/two", body="testtwo"
94 ),
95 ),
96 (
97 re.compile(r"http://example\.com/two"),
98 re.compile(r"http://example\.com/two"),
99 ),
100 ],
101 )
102 def test_replace(original, replacement):
103 @responses.activate
104 def run():
105 responses.add(responses.GET, "http://example.com/one", body="test1")
106
107 if isinstance(original, BaseResponse):
108 responses.add(original)
109 else:
110 responses.add(responses.GET, original, body="test2")
111
112 responses.add(responses.GET, "http://example.com/three", body="test3")
113 responses.add(
114 responses.GET, re.compile(r"http://example\.com/four"), body="test3"
115 )
116
117 if isinstance(replacement, BaseResponse):
118 responses.replace(replacement)
119 else:
120 responses.replace(responses.GET, replacement, body="testtwo")
121
122 resp = requests.get("http://example.com/two")
123 assert_response(resp, "testtwo")
124
125 run()
126 assert_reset()
127
128
129 @pytest.mark.parametrize(
130 "original,replacement",
131 [
132 ("http://example.com/one", re.compile(r"http://example\.com/one")),
133 (re.compile(r"http://example\.com/one"), "http://example.com/one"),
134 ],
135 )
136 def test_replace_error(original, replacement):
137 @responses.activate
138 def run():
139 responses.add(responses.GET, original)
140 with pytest.raises(ValueError):
141 responses.replace(responses.GET, replacement)
142
143 run()
144 assert_reset()
145
146
147 def test_remove():
148 @responses.activate
149 def run():
150 responses.add(responses.GET, "http://example.com/zero")
151 responses.add(responses.GET, "http://example.com/one")
152 responses.add(responses.GET, "http://example.com/two")
153 responses.add(responses.GET, re.compile(r"http://example\.com/three"))
154 responses.add(responses.GET, re.compile(r"http://example\.com/four"))
155 re.purge()
156 responses.remove(responses.GET, "http://example.com/two")
157 responses.remove(Response(method=responses.GET, url="http://example.com/zero"))
158 responses.remove(responses.GET, re.compile(r"http://example\.com/four"))
159
160 with pytest.raises(ConnectionError):
161 requests.get("http://example.com/zero")
162 requests.get("http://example.com/one")
163 with pytest.raises(ConnectionError):
164 requests.get("http://example.com/two")
165 requests.get("http://example.com/three")
166 with pytest.raises(ConnectionError):
167 requests.get("http://example.com/four")
168
169 run()
170 assert_reset()
171
172
173 @pytest.mark.parametrize(
174 "args1,kwargs1,args2,kwargs2,expected",
175 [
176 ((responses.GET, "a"), {}, (responses.GET, "a"), {}, True),
177 ((responses.GET, "a"), {}, (responses.GET, "b"), {}, False),
178 ((responses.GET, "a"), {}, (responses.POST, "a"), {}, False),
179 (
180 (responses.GET, "a"),
181 {"match_querystring": True},
182 (responses.GET, "a"),
183 {},
184 True,
185 ),
186 ],
187 )
188 def test_response_equality(args1, kwargs1, args2, kwargs2, expected):
189 o1 = BaseResponse(*args1, **kwargs1)
190 o2 = BaseResponse(*args2, **kwargs2)
191 assert (o1 == o2) is expected
192 assert (o1 != o2) is not expected
193
194
195 def test_response_equality_different_objects():
196 o1 = BaseResponse(method=responses.GET, url="a")
197 o2 = "str"
198 assert (o1 == o2) is False
199 assert (o1 != o2) is True
200
201
202 def test_connection_error():
203 @responses.activate
204 def run():
205 responses.add(responses.GET, "http://example.com")
206
207 with pytest.raises(ConnectionError):
208 requests.get("http://example.com/foo")
209
210 assert len(responses.calls) == 1
211 assert responses.calls[0].request.url == "http://example.com/foo"
212 assert type(responses.calls[0].response) is ConnectionError
213 assert responses.calls[0].response.request
214
215 run()
216 assert_reset()
217
218
219 def test_match_querystring():
220 @responses.activate
221 def run():
222 url = "http://example.com?test=1&foo=bar"
223 responses.add(responses.GET, url, match_querystring=True, body=b"test")
224 resp = requests.get("http://example.com?test=1&foo=bar")
225 assert_response(resp, "test")
226 resp = requests.get("http://example.com?foo=bar&test=1")
227 assert_response(resp, "test")
228 resp = requests.get("http://example.com/?foo=bar&test=1")
229 assert_response(resp, "test")
230
231 run()
232 assert_reset()
233
234
235 def test_match_empty_querystring():
236 @responses.activate
237 def run():
238 responses.add(
239 responses.GET, "http://example.com", body=b"test", match_querystring=True
240 )
241 resp = requests.get("http://example.com")
242 assert_response(resp, "test")
243 resp = requests.get("http://example.com/")
244 assert_response(resp, "test")
245 with pytest.raises(ConnectionError):
246 requests.get("http://example.com?query=foo")
247
248 run()
249 assert_reset()
250
251
252 def test_match_querystring_error():
253 @responses.activate
254 def run():
255 responses.add(
256 responses.GET, "http://example.com/?test=1", match_querystring=True
257 )
258
259 with pytest.raises(ConnectionError):
260 requests.get("http://example.com/foo/?test=2")
261
262 run()
263 assert_reset()
264
265
266 def test_match_querystring_regex():
267 @responses.activate
268 def run():
269 """Note that `match_querystring` value shouldn't matter when passing a
270 regular expression"""
271
272 responses.add(
273 responses.GET,
274 re.compile(r"http://example\.com/foo/\?test=1"),
275 body="test1",
276 match_querystring=True,
277 )
278
279 resp = requests.get("http://example.com/foo/?test=1")
280 assert_response(resp, "test1")
281
282 responses.add(
283 responses.GET,
284 re.compile(r"http://example\.com/foo/\?test=2"),
285 body="test2",
286 match_querystring=False,
287 )
288
289 resp = requests.get("http://example.com/foo/?test=2")
290 assert_response(resp, "test2")
291
292 run()
293 assert_reset()
294
295
296 def test_match_querystring_error_regex():
297 @responses.activate
298 def run():
299 """Note that `match_querystring` value shouldn't matter when passing a
300 regular expression"""
301
302 responses.add(
303 responses.GET,
304 re.compile(r"http://example\.com/foo/\?test=1"),
305 match_querystring=True,
306 )
307
308 with pytest.raises(ConnectionError):
309 requests.get("http://example.com/foo/?test=3")
310
311 responses.add(
312 responses.GET,
313 re.compile(r"http://example\.com/foo/\?test=2"),
314 match_querystring=False,
315 )
316
317 with pytest.raises(ConnectionError):
318 requests.get("http://example.com/foo/?test=4")
319
320 run()
321 assert_reset()
322
323
324 def test_match_querystring_auto_activates():
325 @responses.activate
326 def run():
327 responses.add(responses.GET, "http://example.com?test=1", body=b"test")
328 resp = requests.get("http://example.com?test=1")
329 assert_response(resp, "test")
330 with pytest.raises(ConnectionError):
331 requests.get("http://example.com/?test=2")
332
333 run()
334 assert_reset()
335
336
337 def test_match_querystring_missing_key():
338 @responses.activate
339 def run():
340 responses.add(responses.GET, "http://example.com?foo=1&bar=2", body=b"test")
341 with pytest.raises(ConnectionError):
342 requests.get("http://example.com/?foo=1&baz=2")
343
344 with pytest.raises(ConnectionError):
345 requests.get("http://example.com/?bar=2&fez=1")
346
347 run()
348 assert_reset()
349
350
351 def test_accept_string_body():
352 @responses.activate
353 def run():
354 url = "http://example.com/"
355 responses.add(responses.GET, url, body="test")
356 resp = requests.get(url)
357 assert_response(resp, "test")
358
359 run()
360 assert_reset()
361
362
363 def test_accept_json_body():
364 @responses.activate
365 def run():
366 content_type = "application/json"
367
368 url = "http://example.com/"
369 responses.add(responses.GET, url, json={"message": "success"})
370 resp = requests.get(url)
371 assert_response(resp, '{"message": "success"}', content_type)
372
373 url = "http://example.com/1/"
374 responses.add(responses.GET, url, json=[])
375 resp = requests.get(url)
376 assert_response(resp, "[]", content_type)
377
378 run()
379 assert_reset()
380
381
382 def test_no_content_type():
383 @responses.activate
384 def run():
385 url = "http://example.com/"
386 responses.add(responses.GET, url, body="test", content_type=None)
387 resp = requests.get(url)
388 assert_response(resp, "test", content_type=None)
389
390 run()
391 assert_reset()
392
393
394 def test_arbitrary_status_code():
395 @responses.activate
396 def run():
397 url = "http://example.com/"
398 responses.add(responses.GET, url, body="test", status=419)
399 resp = requests.get(url)
400 assert resp.status_code == 419
401 assert resp.reason is None
402
403 run()
404 assert_reset()
405
406
407 def test_throw_connection_error_explicit():
408 @responses.activate
409 def run():
410 url = "http://example.com"
411 exception = HTTPError("HTTP Error")
412 responses.add(responses.GET, url, exception)
413
414 with pytest.raises(HTTPError) as HE:
415 requests.get(url)
416
417 assert str(HE.value) == "HTTP Error"
418
419 run()
420 assert_reset()
421
422
423 def test_callback():
424 body = b"test callback"
425 status = 400
426 reason = "Bad Request"
427 headers = {
428 "foo": "bar",
429 "Content-Type": "application/json",
430 "Content-Length": "13",
431 }
432 url = "http://example.com/"
433
434 def request_callback(request):
435 return (status, headers, body)
436
437 @responses.activate
438 def run():
439 responses.add_callback(responses.GET, url, request_callback)
440 resp = requests.get(url)
441 assert resp.text == "test callback"
442 assert resp.status_code == status
443 assert resp.reason == reason
444 assert "bar" == resp.headers.get("foo")
445 assert "application/json" == resp.headers.get("Content-Type")
446 assert "13" == resp.headers.get("Content-Length")
447
448 run()
449 assert_reset()
450
451
452 def test_callback_exception_result():
453 result = Exception()
454 url = "http://example.com/"
455
456 def request_callback(request):
457 return result
458
459 @responses.activate
460 def run():
461 responses.add_callback(responses.GET, url, request_callback)
462
463 with pytest.raises(Exception) as e:
464 requests.get(url)
465
466 assert e.value is result
467
468 run()
469 assert_reset()
470
471
472 def test_callback_exception_body():
473 body = Exception()
474 url = "http://example.com/"
475
476 def request_callback(request):
477 return (200, {}, body)
478
479 @responses.activate
480 def run():
481 responses.add_callback(responses.GET, url, request_callback)
482
483 with pytest.raises(Exception) as e:
484 requests.get(url)
485
486 assert e.value is body
487
488 run()
489 assert_reset()
490
491
492 def test_callback_no_content_type():
493 body = b"test callback"
494 status = 400
495 reason = "Bad Request"
496 headers = {"foo": "bar"}
497 url = "http://example.com/"
498
499 def request_callback(request):
500 return (status, headers, body)
501
502 @responses.activate
503 def run():
504 responses.add_callback(responses.GET, url, request_callback, content_type=None)
505 resp = requests.get(url)
506 assert resp.text == "test callback"
507 assert resp.status_code == status
508 assert resp.reason == reason
509 assert "foo" in resp.headers
510 assert "Content-Type" not in resp.headers
511
512 run()
513 assert_reset()
514
515
516 def test_callback_content_type_dict():
517 def request_callback(request):
518 return (
519 200,
520 {"Content-Type": "application/json"},
521 b"foo",
522 )
523
524 @responses.activate
525 def run():
526 responses.add_callback("GET", "http://mockhost/.foo", callback=request_callback)
527 resp = requests.get("http://mockhost/.foo")
528 assert resp.text == "foo"
529 assert resp.headers["content-type"] == "application/json"
530
531 run()
532 assert_reset()
533
534
535 def test_callback_content_type_tuple():
536 def request_callback(request):
537 return (
538 200,
539 [("Content-Type", "application/json")],
540 b"foo",
541 )
542
543 @responses.activate
544 def run():
545 responses.add_callback("GET", "http://mockhost/.foo", callback=request_callback)
546 resp = requests.get("http://mockhost/.foo")
547 assert resp.text == "foo"
548 assert resp.headers["content-type"] == "application/json"
549
550 run()
551 assert_reset()
552
553
554 def test_regular_expression_url():
555 @responses.activate
556 def run():
557 url = re.compile(r"https?://(.*\.)?example.com")
558 responses.add(responses.GET, url, body=b"test")
559
560 resp = requests.get("http://example.com")
561 assert_response(resp, "test")
562
563 resp = requests.get("https://example.com")
564 assert_response(resp, "test")
565
566 resp = requests.get("https://uk.example.com")
567 assert_response(resp, "test")
568
569 with pytest.raises(ConnectionError):
570 requests.get("https://uk.exaaample.com")
571
572 run()
573 assert_reset()
574
575
576 def test_custom_adapter():
577 @responses.activate
578 def run():
579 url = "http://example.com"
580 responses.add(responses.GET, url, body=b"test")
581
582 calls = [0]
583
584 class DummyAdapter(requests.adapters.HTTPAdapter):
585 def send(self, *a, **k):
586 calls[0] += 1
587 return super(DummyAdapter, self).send(*a, **k)
588
589 # Test that the adapter is actually used
590 session = requests.Session()
591 session.mount("http://", DummyAdapter())
592
593 resp = session.get(url, allow_redirects=False)
594 assert calls[0] == 1
595
596 # Test that the response is still correctly emulated
597 session = requests.Session()
598 session.mount("http://", DummyAdapter())
599
600 resp = session.get(url)
601 assert_response(resp, "test")
602
603 run()
604
605
606 def test_responses_as_context_manager():
607 def run():
608 with responses.mock:
609 responses.add(responses.GET, "http://example.com", body=b"test")
610 resp = requests.get("http://example.com")
611 assert_response(resp, "test")
612 assert len(responses.calls) == 1
613 assert responses.calls[0].request.url == "http://example.com/"
614 assert responses.calls[0].response.content == b"test"
615
616 resp = requests.get("http://example.com?foo=bar")
617 assert_response(resp, "test")
618 assert len(responses.calls) == 2
619 assert responses.calls[1].request.url == "http://example.com/?foo=bar"
620 assert responses.calls[1].response.content == b"test"
621
622 run()
623 assert_reset()
624
625
626 def test_activate_doesnt_change_signature():
627 def test_function(a, b=None):
628 return (a, b)
629
630 decorated_test_function = responses.activate(test_function)
631 if hasattr(inspect, "signature"):
632 assert inspect.signature(test_function) == inspect.signature(
633 decorated_test_function
634 )
635 else:
636 assert inspect.getargspec(test_function) == inspect.getargspec(
637 decorated_test_function
638 )
639 assert decorated_test_function(1, 2) == test_function(1, 2)
640 assert decorated_test_function(3) == test_function(3)
641
642
643 def test_activate_mock_interaction():
644 @patch("sys.stdout")
645 def test_function(mock_stdout):
646 return mock_stdout
647
648 decorated_test_function = responses.activate(test_function)
649 if hasattr(inspect, "signature"):
650 assert inspect.signature(test_function) == inspect.signature(
651 decorated_test_function
652 )
653 else:
654 assert inspect.getargspec(test_function) == inspect.getargspec(
655 decorated_test_function
656 )
657
658 value = test_function()
659 assert isinstance(value, Mock)
660
661 value = decorated_test_function()
662 assert isinstance(value, Mock)
663
664
665 @pytest.mark.skipif(six.PY2, reason="Cannot run in python2")
666 def test_activate_doesnt_change_signature_with_return_type():
667 def test_function(a, b=None):
668 return (a, b)
669
670 # Add type annotations as they are syntax errors in py2.
671 # Use a class to test for import errors in evaled code.
672 test_function.__annotations__["return"] = Mock
673 test_function.__annotations__["a"] = Mock
674
675 decorated_test_function = responses.activate(test_function)
676 if hasattr(inspect, "signature"):
677 assert inspect.signature(test_function) == inspect.signature(
678 decorated_test_function
679 )
680 else:
681 assert inspect.getargspec(test_function) == inspect.getargspec(
682 decorated_test_function
683 )
684 assert decorated_test_function(1, 2) == test_function(1, 2)
685 assert decorated_test_function(3) == test_function(3)
686
687
688 def test_activate_doesnt_change_signature_for_method():
689 class TestCase(object):
690 def test_function(self, a, b=None):
691 return (self, a, b)
692
693 decorated_test_function = responses.activate(test_function)
694
695 test_case = TestCase()
696 assert test_case.decorated_test_function(1, 2) == test_case.test_function(1, 2)
697 assert test_case.decorated_test_function(3) == test_case.test_function(3)
698
699
700 def test_response_cookies():
701 body = b"test callback"
702 status = 200
703 headers = {"set-cookie": "session_id=12345; a=b; c=d"}
704 url = "http://example.com/"
705
706 def request_callback(request):
707 return (status, headers, body)
708
709 @responses.activate
710 def run():
711 responses.add_callback(responses.GET, url, request_callback)
712 resp = requests.get(url)
713 assert resp.text == "test callback"
714 assert resp.status_code == status
715 assert "session_id" in resp.cookies
716 assert resp.cookies["session_id"] == "12345"
717 assert set(resp.cookies.keys()) == set(["session_id"])
718
719 run()
720 assert_reset()
721
722
723 def test_response_secure_cookies():
724 body = b"test callback"
725 status = 200
726 headers = {"set-cookie": "session_id=12345; a=b; c=d; secure"}
727 url = "http://example.com/"
728
729 def request_callback(request):
730 return (status, headers, body)
731
732 @responses.activate
733 def run():
734 responses.add_callback(responses.GET, url, request_callback)
735 resp = requests.get(url)
736 assert resp.text == "test callback"
737 assert resp.status_code == status
738 assert "session_id" in resp.cookies
739 assert resp.cookies["session_id"] == "12345"
740 assert set(resp.cookies.keys()) == set(["session_id"])
741
742 run()
743 assert_reset()
744
745
746 def test_response_cookies_multiple():
747 body = b"test callback"
748 status = 200
749 headers = [
750 ("set-cookie", "1P_JAR=2019-12-31-23; path=/; domain=.example.com; HttpOnly"),
751 ("set-cookie", "NID=some=value; path=/; domain=.example.com; secure"),
752 ]
753 url = "http://example.com/"
754
755 def request_callback(request):
756 return (status, headers, body)
757
758 @responses.activate
759 def run():
760 responses.add_callback(responses.GET, url, request_callback)
761 resp = requests.get(url)
762 assert resp.text == "test callback"
763 assert resp.status_code == status
764 assert set(resp.cookies.keys()) == set(["1P_JAR", "NID"])
765 assert resp.cookies["1P_JAR"] == "2019-12-31-23"
766 assert resp.cookies["NID"] == "some=value"
767
768 run()
769 assert_reset()
770
771
772 def test_response_callback():
773 """adds a callback to decorate the response, then checks it"""
774
775 def run():
776 def response_callback(resp):
777 resp._is_mocked = True
778 return resp
779
780 with responses.RequestsMock(response_callback=response_callback) as m:
781 m.add(responses.GET, "http://example.com", body=b"test")
782 resp = requests.get("http://example.com")
783 assert resp.text == "test"
784 assert hasattr(resp, "_is_mocked")
785 assert resp._is_mocked is True
786
787 run()
788 assert_reset()
789
790
791 def test_response_filebody():
792 """ Adds the possibility to use actual (binary) files as responses """
793
794 def run():
795 with responses.RequestsMock() as m:
796 with open("README.rst", "rb") as out:
797 m.add(responses.GET, "http://example.com", body=out, stream=True)
798 resp = requests.get("http://example.com")
799 with open("README.rst", "r") as out:
800 assert resp.text == out.read()
801
802
803 def test_assert_all_requests_are_fired():
804 def request_callback(request):
805 raise BaseException()
806
807 def run():
808 with pytest.raises(AssertionError) as excinfo:
809 with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
810 m.add(responses.GET, "http://example.com", body=b"test")
811 assert "http://example.com" in str(excinfo.value)
812 assert responses.GET in str(excinfo.value)
813
814 # check that assert_all_requests_are_fired default to True
815 with pytest.raises(AssertionError):
816 with responses.RequestsMock() as m:
817 m.add(responses.GET, "http://example.com", body=b"test")
818
819 # check that assert_all_requests_are_fired doesn't swallow exceptions
820 with pytest.raises(ValueError):
821 with responses.RequestsMock() as m:
822 m.add(responses.GET, "http://example.com", body=b"test")
823 raise ValueError()
824
825 # check that assert_all_requests_are_fired=True doesn't remove urls
826 with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
827 m.add(responses.GET, "http://example.com", body=b"test")
828 assert len(m._matches) == 1
829 requests.get("http://example.com")
830 assert len(m._matches) == 1
831
832 # check that assert_all_requests_are_fired=True counts mocked errors
833 with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
834 m.add(responses.GET, "http://example.com", body=Exception())
835 assert len(m._matches) == 1
836 with pytest.raises(Exception):
837 requests.get("http://example.com")
838 assert len(m._matches) == 1
839
840 with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
841 m.add_callback(responses.GET, "http://example.com", request_callback)
842 assert len(m._matches) == 1
843 with pytest.raises(BaseException):
844 requests.get("http://example.com")
845 assert len(m._matches) == 1
846
847 run()
848 assert_reset()
849
850
851 def test_allow_redirects_samehost():
852 redirecting_url = "http://example.com"
853 final_url_path = "/1"
854 final_url = "{0}{1}".format(redirecting_url, final_url_path)
855 url_re = re.compile(r"^http://example.com(/)?(\d+)?$")
856
857 def request_callback(request):
858 # endpoint of chained redirect
859 if request.url.endswith(final_url_path):
860 return 200, (), b"test"
861
862 # otherwise redirect to an integer path
863 else:
864 if request.url.endswith("/0"):
865 n = 1
866 else:
867 n = 0
868 redirect_headers = {"location": "/{0!s}".format(n)}
869 return 301, redirect_headers, None
870
871 def run():
872 # setup redirect
873 with responses.mock:
874 responses.add_callback(responses.GET, url_re, request_callback)
875 resp_no_redirects = requests.get(redirecting_url, allow_redirects=False)
876 assert resp_no_redirects.status_code == 301
877 assert len(responses.calls) == 1 # 1x300
878 assert responses.calls[0][1].status_code == 301
879 assert_reset()
880
881 with responses.mock:
882 responses.add_callback(responses.GET, url_re, request_callback)
883 resp_yes_redirects = requests.get(redirecting_url, allow_redirects=True)
884 assert len(responses.calls) == 3 # 2x300 + 1x200
885 assert len(resp_yes_redirects.history) == 2
886 assert resp_yes_redirects.status_code == 200
887 assert final_url == resp_yes_redirects.url
888 status_codes = [call[1].status_code for call in responses.calls]
889 assert status_codes == [301, 301, 200]
890 assert_reset()
891
892 run()
893 assert_reset()
894
895
896 def test_handles_unicode_querystring():
897 url = "http://example.com/test?type=2&ie=utf8&query=汉字"
898
899 @responses.activate
900 def run():
901 responses.add(responses.GET, url, body="test", match_querystring=True)
902
903 resp = requests.get(url)
904
905 assert_response(resp, "test")
906
907 run()
908 assert_reset()
909
910
911 def test_handles_unicode_url():
912 url = "http://www.संजाल.भारत/hi/वेबसाइट-डिजाइन"
913
914 @responses.activate
915 def run():
916 responses.add(responses.GET, url, body="test")
917
918 resp = requests.get(url)
919
920 assert_response(resp, "test")
921
922 run()
923 assert_reset()
924
925
926 def test_handles_unicode_body():
927 url = "http://example.com/test"
928
929 @responses.activate
930 def run():
931 responses.add(responses.GET, url, body="михољско лето")
932
933 resp = requests.get(url)
934
935 assert_response(resp, "михољско лето", content_type="text/plain; charset=utf-8")
936
937 run()
938 assert_reset()
939
940
941 def test_handles_buffered_reader_body():
942 url = "http://example.com/test"
943
944 @responses.activate
945 def run():
946 responses.add(responses.GET, url, body=BufferedReader(BytesIO(b"test")))
947
948 resp = requests.get(url)
949
950 assert_response(resp, "test")
951
952 run()
953 assert_reset()
954
955
956 def test_headers():
957 @responses.activate
958 def run():
959 responses.add(
960 responses.GET, "http://example.com", body="", headers={"X-Test": "foo"}
961 )
962 resp = requests.get("http://example.com")
963 assert resp.headers["X-Test"] == "foo"
964
965 run()
966 assert_reset()
967
968
969 def test_legacy_adding_headers():
970 @responses.activate
971 def run():
972 responses.add(
973 responses.GET,
974 "http://example.com",
975 body="",
976 adding_headers={"X-Test": "foo"},
977 )
978 resp = requests.get("http://example.com")
979 assert resp.headers["X-Test"] == "foo"
980
981 run()
982 assert_reset()
983
984
985 def test_multiple_responses():
986 @responses.activate
987 def run():
988 responses.add(responses.GET, "http://example.com", body="test")
989 responses.add(responses.GET, "http://example.com", body="rest")
990
991 resp = requests.get("http://example.com")
992 assert_response(resp, "test")
993 resp = requests.get("http://example.com")
994 assert_response(resp, "rest")
995 # After all responses are used, last response should be repeated
996 resp = requests.get("http://example.com")
997 assert_response(resp, "rest")
998
999 run()
1000 assert_reset()
1001
1002
1003 def test_multiple_urls():
1004 @responses.activate
1005 def run():
1006 responses.add(responses.GET, "http://example.com/one", body="one")
1007 responses.add(responses.GET, "http://example.com/two", body="two")
1008
1009 resp = requests.get("http://example.com/two")
1010 assert_response(resp, "two")
1011 resp = requests.get("http://example.com/one")
1012 assert_response(resp, "one")
1013
1014 run()
1015 assert_reset()
1016
1017
1018 def test_multiple_methods():
1019 @responses.activate
1020 def run():
1021 responses.add(responses.GET, "http://example.com/one", body="gotcha")
1022 responses.add(responses.POST, "http://example.com/one", body="posted")
1023
1024 resp = requests.get("http://example.com/one")
1025 assert_response(resp, "gotcha")
1026 resp = requests.post("http://example.com/one")
1027 assert_response(resp, "posted")
1028
1029 run()
1030 assert_reset()
1031
1032
1033 def test_passthru(httpserver):
1034 httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
1035
1036 @responses.activate
1037 def run():
1038 responses.add_passthru(httpserver.url)
1039 responses.add(responses.GET, "{}/one".format(httpserver.url), body="one")
1040 responses.add(responses.GET, "http://example.com/two", body="two")
1041
1042 resp = requests.get("http://example.com/two")
1043 assert_response(resp, "two")
1044 resp = requests.get("{}/one".format(httpserver.url))
1045 assert_response(resp, "one")
1046 resp = requests.get(httpserver.url)
1047 assert_response(resp, "OK")
1048
1049 run()
1050 assert_reset()
1051
1052
1053 def test_passthru_regex(httpserver):
1054 httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
1055
1056 @responses.activate
1057 def run():
1058 responses.add_passthru(re.compile("{}/\\w+".format(httpserver.url)))
1059 responses.add(responses.GET, "{}/one".format(httpserver.url), body="one")
1060 responses.add(responses.GET, "http://example.com/two", body="two")
1061
1062 resp = requests.get("http://example.com/two")
1063 assert_response(resp, "two")
1064 resp = requests.get("{}/one".format(httpserver.url))
1065 assert_response(resp, "one")
1066 resp = requests.get("{}/two".format(httpserver.url))
1067 assert_response(resp, "OK")
1068 resp = requests.get("{}/three".format(httpserver.url))
1069 assert_response(resp, "OK")
1070
1071 run()
1072 assert_reset()
1073
1074
1075 def test_method_named_param():
1076 @responses.activate
1077 def run():
1078 responses.add(method=responses.GET, url="http://example.com", body="OK")
1079 resp = requests.get("http://example.com")
1080 assert_response(resp, "OK")
1081
1082 run()
1083 assert_reset()
1084
1085
1086 def test_passthru_unicode():
1087 @responses.activate
1088 def run():
1089 with responses.RequestsMock() as m:
1090 url = "http://موقع.وزارة-الاتصالات.مصر/"
1091 clean_url = "http://xn--4gbrim.xn----ymcbaaajlc6dj7bxne2c.xn--wgbh1c/"
1092 m.add_passthru(url)
1093 assert m.passthru_prefixes[0] == clean_url
1094
1095 run()
1096 assert_reset()
1097
1098
1099 def test_custom_target(monkeypatch):
1100 requests_mock = responses.RequestsMock(target="something.else")
1101 std_mock_mock = responses.std_mock.MagicMock()
1102 patch_mock = std_mock_mock.patch
1103 monkeypatch.setattr(responses, "std_mock", std_mock_mock)
1104 requests_mock.start()
1105 assert len(patch_mock.call_args_list) == 1
1106 assert patch_mock.call_args[1]["target"] == "something.else"
1107
1108
1109 def _quote(s):
1110 return responses.quote(responses._ensure_str(s))
1111
1112
1113 def test_cookies_from_headers():
1114 text = "こんにちは/世界"
1115 quoted_text = _quote(text)
1116 expected = {"x": "a", "y": quoted_text}
1117 headers = {"set-cookie": "; ".join(k + "=" + v for k, v in expected.items())}
1118 cookiejar = responses._cookies_from_headers(headers)
1119 for k, v in cookiejar.items():
1120 assert isinstance(v, str)
1121 assert v == expected[k]
1122
1123
1124 @pytest.mark.parametrize(
1125 "url",
1126 (
1127 "http://example.com",
1128 "http://example.com/some/path",
1129 "http://example.com/other/path/",
1130 ),
1131 )
1132 def test_request_param(url):
1133 @responses.activate
1134 def run():
1135 params = {"hello": "world", "example": "params"}
1136 responses.add(
1137 method=responses.GET,
1138 url="{0}?hello=world".format(url),
1139 body="test",
1140 match_querystring=False,
1141 )
1142 resp = requests.get(url, params=params)
1143 assert_response(resp, "test")
1144 assert resp.request.params == params
1145
1146 resp = requests.get(url)
1147 assert_response(resp, "test")
1148 assert resp.request.params == {}
1149
1150 run()
1151 assert_reset()
1152
1153
1154 def test_request_param_with_multiple_values_for_the_same_key():
1155 @responses.activate
1156 def run():
1157 url = "http://example.com"
1158 params = {"key1": ["one", "two"], "key2": "three"}
1159 responses.add(
1160 method=responses.GET,
1161 url=url,
1162 body="test",
1163 )
1164 resp = requests.get(url, params=params)
1165 assert_response(resp, "test")
1166 assert resp.request.params == params
1167
1168 run()
1169 assert_reset()
1170
1171
1172 @pytest.mark.parametrize(
1173 "url", ("http://example.com", "http://example.com?hello=world")
1174 )
1175 def test_assert_call_count(url):
1176 @responses.activate
1177 def run():
1178 responses.add(responses.GET, url)
1179 responses.add(responses.GET, "http://example1.com")
1180
1181 assert responses.assert_call_count(url, 0) is True
1182
1183 with pytest.raises(AssertionError) as excinfo:
1184 responses.assert_call_count(url, 2)
1185 assert "Expected URL '{0}' to be called 2 times. Called 0 times.".format(
1186 url
1187 ) in str(excinfo.value)
1188
1189 requests.get(url)
1190 assert responses.assert_call_count(url, 1) is True
1191
1192 requests.get("http://example1.com")
1193 assert responses.assert_call_count(url, 1) is True
1194
1195 requests.get(url)
1196 with pytest.raises(AssertionError) as excinfo:
1197 responses.assert_call_count(url, 3)
1198 assert "Expected URL '{0}' to be called 3 times. Called 2 times.".format(
1199 url
1200 ) in str(excinfo.value)
1201
1202 run()
1203 assert_reset()
1204
1205
1206 def test_request_matches_post_params():
1207 @responses.activate
1208 def run():
1209 responses.add(
1210 method=responses.POST,
1211 url="http://example.com/",
1212 body="one",
1213 match=[
1214 responses.json_params_matcher(
1215 {"page": {"name": "first", "type": "json"}}
1216 )
1217 ],
1218 )
1219 responses.add(
1220 method=responses.POST,
1221 url="http://example.com/",
1222 body="two",
1223 match=[
1224 responses.urlencoded_params_matcher(
1225 {"page": "second", "type": "urlencoded"}
1226 )
1227 ],
1228 )
1229
1230 resp = requests.request(
1231 "POST",
1232 "http://example.com/",
1233 headers={"Content-Type": "x-www-form-urlencoded"},
1234 data={"page": "second", "type": "urlencoded"},
1235 )
1236 assert_response(resp, "two")
1237
1238 resp = requests.request(
1239 "POST",
1240 "http://example.com/",
1241 headers={"Content-Type": "application/json"},
1242 json={"page": {"name": "first", "type": "json"}},
1243 )
1244 assert_response(resp, "one")
1245
1246 run()
1247 assert_reset()
1248
1249
1250 def test_request_matches_empty_body():
1251 @responses.activate
1252 def run():
1253 responses.add(
1254 method=responses.POST,
1255 url="http://example.com/",
1256 body="one",
1257 match=[responses.json_params_matcher(None)],
1258 )
1259
1260 responses.add(
1261 method=responses.POST,
1262 url="http://example.com/",
1263 body="two",
1264 match=[responses.urlencoded_params_matcher(None)],
1265 )
1266
1267 resp = requests.request("POST", "http://example.com/")
1268 assert_response(resp, "one")
1269
1270 resp = requests.request(
1271 "POST",
1272 "http://example.com/",
1273 headers={"Content-Type": "x-www-form-urlencoded"},
1274 )
1275 assert_response(resp, "two")
1276
1277 run()
1278 assert_reset()
1279
1280
1281 def test_fail_request_error():
1282 @responses.activate
1283 def run():
1284 responses.add("POST", "http://example1.com")
1285 responses.add("GET", "http://example.com")
1286 responses.add(
1287 "POST",
1288 "http://example.com",
1289 match=[responses.urlencoded_params_matcher({"foo": "bar"})],
1290 )
1291
1292 with pytest.raises(ConnectionError) as excinfo:
1293 requests.post("http://example.com", data={"id": "bad"})
1294 msg = str(excinfo.value)
1295 assert "- POST http://example1.com/ URL does not match" in msg
1296 assert "- GET http://example.com/ Method does not match" in msg
1297 assert "- POST http://example.com/ Parameters do not match" in msg
1298
1299 run()
1300 assert_reset()