Make a public create_response function
We need a way for custom matchers to be able to return fully formed responses. Make a new function for this and convert the existing response matchers to use it.
This commit is contained in:
Jamie Lennox
parent
1246046ffe
commit
35951b0386
@ -14,10 +14,12 @@ from requests_mock.adapter import Adapter, ANY
|
||||
from requests_mock.exceptions import MockException, NoMockAddress
|
||||
from requests_mock.mocker import mock, Mocker, MockerCore
|
||||
from requests_mock.mocker import DELETE, GET, HEAD, OPTIONS, PATCH, POST, PUT
|
||||
from requests_mock.response import create_response
|
||||
|
||||
|
||||
__all__ = ['Adapter',
|
||||
'ANY',
|
||||
'create_response',
|
||||
'mock',
|
||||
'Mocker',
|
||||
'MockerCore',
|
||||
|
||||
@ -10,15 +10,13 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import json as jsonutils
|
||||
|
||||
import requests
|
||||
from requests.adapters import BaseAdapter, HTTPAdapter
|
||||
from requests.packages.urllib3.response import HTTPResponse
|
||||
from requests.adapters import BaseAdapter
|
||||
import six
|
||||
from six.moves.urllib import parse as urlparse
|
||||
|
||||
from requests_mock import exceptions
|
||||
from requests_mock import response
|
||||
|
||||
ANY = object()
|
||||
|
||||
@ -107,106 +105,9 @@ class _RequestHistoryTracker(object):
|
||||
return len(self.request_history)
|
||||
|
||||
|
||||
class _MatcherResponse(object):
|
||||
|
||||
_BODY_ARGS = ['raw', 'body', 'content', 'text', 'json']
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
"""
|
||||
:param int status_code: The status code to return upon a successful
|
||||
match. Defaults to 200.
|
||||
:param HTTPResponse raw: A HTTPResponse object to return upon a
|
||||
successful match.
|
||||
:param io.IOBase body: An IO object with a read() method that can
|
||||
return a body on successful match.
|
||||
:param bytes content: A byte string to return upon a successful match.
|
||||
:param unicode text: A text string to return upon a successful match.
|
||||
:param object json: A python object to be converted to a JSON string
|
||||
and returned upon a successful match.
|
||||
:param dict headers: A dictionary object containing headers that are
|
||||
returned upon a successful match.
|
||||
"""
|
||||
# mutual exclusion, only 1 body method may be provided
|
||||
provided = [x for x in self._BODY_ARGS if kwargs.get(x) is not None]
|
||||
|
||||
self.status_code = kwargs.pop('status_code', 200)
|
||||
self.raw = kwargs.pop('raw', None)
|
||||
self.body = kwargs.pop('body', None)
|
||||
self.content = kwargs.pop('content', None)
|
||||
self.text = kwargs.pop('text', None)
|
||||
self.json = kwargs.pop('json', None)
|
||||
self.reason = kwargs.pop('reason', None)
|
||||
self.headers = kwargs.pop('headers', {})
|
||||
|
||||
if kwargs:
|
||||
raise TypeError('Too many arguments provided. Unexpected '
|
||||
'arguments %s.' % ', '.join(kwargs.keys()))
|
||||
|
||||
if len(provided) == 0:
|
||||
self.body = six.BytesIO(six.b(''))
|
||||
elif len(provided) > 1:
|
||||
raise RuntimeError('You may only supply one body element. You '
|
||||
'supplied %s' % ', '.join(provided))
|
||||
|
||||
# whilst in general you shouldn't do type checking in python this
|
||||
# makes sure we don't end up with differences between the way types
|
||||
# are handled between python 2 and 3.
|
||||
if self.content and not (callable(self.content) or
|
||||
isinstance(self.content, six.binary_type)):
|
||||
raise TypeError('Content should be a callback or binary data')
|
||||
if self.text and not (callable(self.text) or
|
||||
isinstance(self.text, six.string_types)):
|
||||
raise TypeError('Text should be a callback or string data')
|
||||
|
||||
def get_response(self, request):
|
||||
encoding = None
|
||||
context = _Context(self.headers.copy(),
|
||||
self.status_code,
|
||||
self.reason)
|
||||
|
||||
# if a body element is a callback then execute it
|
||||
def _call(f, *args, **kwargs):
|
||||
return f(request, context, *args, **kwargs) if callable(f) else f
|
||||
|
||||
content = self.content
|
||||
text = self.text
|
||||
body = self.body
|
||||
raw = self.raw
|
||||
|
||||
if self.json is not None:
|
||||
data = _call(self.json)
|
||||
text = jsonutils.dumps(data)
|
||||
if text is not None:
|
||||
data = _call(text)
|
||||
encoding = 'utf-8'
|
||||
content = data.encode(encoding)
|
||||
if content is not None:
|
||||
data = _call(content)
|
||||
body = six.BytesIO(data)
|
||||
if body is not None:
|
||||
data = _call(body)
|
||||
raw = HTTPResponse(status=context.status_code,
|
||||
body=data,
|
||||
headers=context.headers,
|
||||
reason=context.reason,
|
||||
decode_content=False,
|
||||
preload_content=False)
|
||||
|
||||
return encoding, raw
|
||||
|
||||
|
||||
class _FakeConnection(object):
|
||||
"""An object that can mock the necessary parts of a socket interface."""
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
|
||||
class _Matcher(_RequestHistoryTracker):
|
||||
"""Contains all the information about a provided URL to match."""
|
||||
|
||||
_http_adapter = HTTPAdapter()
|
||||
|
||||
def __init__(self, method, url, responses, complete_qs, request_headers):
|
||||
"""
|
||||
:param bool complete_qs: Match the entire query string. By default URLs
|
||||
@ -296,12 +197,7 @@ class _Matcher(_RequestHistoryTracker):
|
||||
response_matcher = self._responses[0]
|
||||
|
||||
self._add_to_history(request)
|
||||
|
||||
encoding, response = response_matcher.get_response(request)
|
||||
req_resp = self._http_adapter.build_response(request, response)
|
||||
req_resp.connection = _FakeConnection()
|
||||
req_resp.encoding = encoding
|
||||
return req_resp
|
||||
return response_matcher.get_response(request)
|
||||
|
||||
|
||||
class Adapter(BaseAdapter, _RequestHistoryTracker):
|
||||
@ -317,9 +213,9 @@ class Adapter(BaseAdapter, _RequestHistoryTracker):
|
||||
self._add_to_history(request)
|
||||
|
||||
for matcher in reversed(self._matchers):
|
||||
response = matcher(request)
|
||||
if response is not None:
|
||||
return response
|
||||
resp = matcher(request)
|
||||
if resp is not None:
|
||||
return resp
|
||||
|
||||
raise exceptions.NoMockAddress(request)
|
||||
|
||||
@ -341,7 +237,7 @@ class Adapter(BaseAdapter, _RequestHistoryTracker):
|
||||
elif not response_list:
|
||||
response_list = [kwargs]
|
||||
|
||||
responses = [_MatcherResponse(**k) for k in response_list]
|
||||
responses = [response._MatcherResponse(**k) for k in response_list]
|
||||
matcher = _Matcher(method,
|
||||
url,
|
||||
responses,
|
||||
|
||||
144
requests_mock/response.py
Normal file
144
requests_mock/response.py
Normal file
@ -0,0 +1,144 @@
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import json as jsonutils
|
||||
|
||||
from requests.adapters import HTTPAdapter
|
||||
from requests.packages.urllib3.response import HTTPResponse
|
||||
import six
|
||||
|
||||
_BODY_ARGS = frozenset(['raw', 'body', 'content', 'text', 'json'])
|
||||
_HTTP_ARGS = frozenset(['status_code', 'reason', 'headers'])
|
||||
|
||||
_DEFAULT_STATUS = 200
|
||||
_http_adapter = HTTPAdapter()
|
||||
|
||||
|
||||
def _check_body_arguments(**kwargs):
|
||||
# mutual exclusion, only 1 body method may be provided
|
||||
provided = [x for x in _BODY_ARGS if kwargs.pop(x, None) is not None]
|
||||
|
||||
if len(provided) > 1:
|
||||
raise RuntimeError('You may only supply one body element. You '
|
||||
'supplied %s' % ', '.join(provided))
|
||||
|
||||
extra = [x for x in kwargs if x not in _HTTP_ARGS]
|
||||
|
||||
if extra:
|
||||
raise TypeError('Too many arguments provided. Unexpected '
|
||||
'arguments %s.' % ', '.join(extra))
|
||||
|
||||
|
||||
class _FakeConnection(object):
|
||||
"""An object that can mock the necessary parts of a socket interface."""
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
|
||||
def create_response(request, **kwargs):
|
||||
"""
|
||||
:param int status_code: The status code to return upon a successful
|
||||
match. Defaults to 200.
|
||||
:param HTTPResponse raw: A HTTPResponse object to return upon a
|
||||
successful match.
|
||||
:param io.IOBase body: An IO object with a read() method that can
|
||||
return a body on successful match.
|
||||
:param bytes content: A byte string to return upon a successful match.
|
||||
:param unicode text: A text string to return upon a successful match.
|
||||
:param object json: A python object to be converted to a JSON string
|
||||
and returned upon a successful match.
|
||||
:param dict headers: A dictionary object containing headers that are
|
||||
returned upon a successful match.
|
||||
"""
|
||||
_check_body_arguments(**kwargs)
|
||||
|
||||
raw = kwargs.pop('raw', None)
|
||||
body = kwargs.pop('body', None)
|
||||
content = kwargs.pop('content', None)
|
||||
text = kwargs.pop('text', None)
|
||||
json = kwargs.pop('json', None)
|
||||
encoding = None
|
||||
|
||||
if content and not isinstance(content, six.binary_type):
|
||||
raise TypeError('Content should be a callback or binary data')
|
||||
if text and not isinstance(text, six.string_types):
|
||||
raise TypeError('Text should be a callback or string data')
|
||||
|
||||
if json is not None:
|
||||
text = jsonutils.dumps(json)
|
||||
if text is not None:
|
||||
encoding = 'utf-8'
|
||||
content = text.encode(encoding)
|
||||
if content is not None:
|
||||
body = six.BytesIO(content)
|
||||
if not raw:
|
||||
raw = HTTPResponse(status=kwargs.get('status_code', _DEFAULT_STATUS),
|
||||
headers=kwargs.get('headers', {}),
|
||||
reason=kwargs.get('reason'),
|
||||
body=body or six.BytesIO(six.b('')),
|
||||
decode_content=False,
|
||||
preload_content=False)
|
||||
|
||||
response = _http_adapter.build_response(request, raw)
|
||||
response.connection = _FakeConnection()
|
||||
response.encoding = encoding
|
||||
return response
|
||||
|
||||
|
||||
class _Context(object):
|
||||
"""Stores the data being used to process a current URL match."""
|
||||
|
||||
def __init__(self, headers, status_code, reason):
|
||||
self.headers = headers
|
||||
self.status_code = status_code
|
||||
self.reason = reason
|
||||
|
||||
|
||||
class _MatcherResponse(object):
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
_check_body_arguments(**kwargs)
|
||||
self._params = kwargs
|
||||
|
||||
# whilst in general you shouldn't do type checking in python this
|
||||
# makes sure we don't end up with differences between the way types
|
||||
# are handled between python 2 and 3.
|
||||
content = self._params.get('content')
|
||||
text = self._params.get('text')
|
||||
|
||||
if content and not (callable(content) or
|
||||
isinstance(content, six.binary_type)):
|
||||
raise TypeError('Content should be a callback or binary data')
|
||||
|
||||
if text and not (callable(text) or
|
||||
isinstance(text, six.string_types)):
|
||||
raise TypeError('Text should be a callback or string data')
|
||||
|
||||
def get_response(self, request):
|
||||
context = _Context(self._params.get('headers', {}).copy(),
|
||||
self._params.get('status_code', _DEFAULT_STATUS),
|
||||
self._params.get('reason'))
|
||||
|
||||
# if a body element is a callback then execute it
|
||||
def _call(f, *args, **kwargs):
|
||||
return f(request, context, *args, **kwargs) if callable(f) else f
|
||||
|
||||
return create_response(request,
|
||||
json=_call(self._params.get('json')),
|
||||
text=_call(self._params.get('text')),
|
||||
content=_call(self._params.get('content')),
|
||||
body=_call(self._params.get('body')),
|
||||
raw=self._params.get('raw'),
|
||||
status_code=context.status_code,
|
||||
reason=context.reason,
|
||||
headers=context.headers)
|
||||
68
requests_mock/tests/test_response.py
Normal file
68
requests_mock/tests/test_response.py
Normal file
@ -0,0 +1,68 @@
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import six
|
||||
|
||||
from requests_mock import adapter
|
||||
from requests_mock import response
|
||||
from requests_mock.tests import base
|
||||
|
||||
|
||||
class ResponseTests(base.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(ResponseTests, self).setUp()
|
||||
self.method = 'GET'
|
||||
self.url = 'http://test.url/path'
|
||||
self.request = adapter._RequestObjectProxy._create(self.method,
|
||||
self.url,
|
||||
{})
|
||||
|
||||
def create_response(self, **kwargs):
|
||||
return response.create_response(self.request, **kwargs)
|
||||
|
||||
def test_create_response_body_args(self):
|
||||
self.assertRaises(RuntimeError,
|
||||
self.create_response,
|
||||
raw='abc',
|
||||
body='abc')
|
||||
|
||||
self.assertRaises(RuntimeError,
|
||||
self.create_response,
|
||||
text='abc',
|
||||
json={'a': 1})
|
||||
|
||||
def test_content_type(self):
|
||||
self.assertRaises(TypeError, self.create_response, text=55)
|
||||
self.assertRaises(TypeError, self.create_response, text={'a': 1})
|
||||
|
||||
def test_text_type(self):
|
||||
self.assertRaises(TypeError, self.create_response, content=six.u('t'))
|
||||
self.assertRaises(TypeError, self.create_response, content={'a': 1})
|
||||
|
||||
def test_json_body(self):
|
||||
data = {'a': 1}
|
||||
resp = self.create_response(json=data)
|
||||
|
||||
self.assertEqual('{"a": 1}', resp.text)
|
||||
self.assertIsInstance(resp.text, six.string_types)
|
||||
self.assertIsInstance(resp.content, six.binary_type)
|
||||
self.assertEqual(data, resp.json())
|
||||
|
||||
def test_body_body(self):
|
||||
value = 'data'
|
||||
body = six.BytesIO(six.b(value))
|
||||
resp = self.create_response(body=body)
|
||||
|
||||
self.assertEqual(value, resp.text)
|
||||
self.assertIsInstance(resp.text, six.string_types)
|
||||
self.assertIsInstance(resp.content, six.binary_type)
|
||||
Loading…
x
Reference in New Issue
Block a user