diff --git a/requests_mock/adapter.py b/requests_mock/adapter.py index 15fa9ee..5ae6e87 100644 --- a/requests_mock/adapter.py +++ b/requests_mock/adapter.py @@ -10,131 +10,19 @@ # License for the specific language governing permissions and limitations # under the License. -import copy -import json import weakref -import requests 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 +from requests_mock.request import _RequestObjectProxy +from requests_mock.response import _MatcherResponse ANY = object() -class _RequestObjectProxy(object): - """A wrapper around a requests.Request that gives some extra information. - - This will be important both for matching and so that when it's save into - the request_history users will be able to access these properties. - """ - - def __init__(self, request, **kwargs): - self._request = request - self._matcher = None - self._url_parts_ = None - self._qs = None - - # All of these params should always exist but we use a default - # to make the test setup easier. - self._timeout = kwargs.pop('timeout', None) - self._allow_redirects = kwargs.pop('allow_redirects', None) - self._verify = kwargs.pop('verify', None) - self._cert = kwargs.pop('cert', None) - self._proxies = copy.deepcopy(kwargs.pop('proxies', {})) - - # FIXME(jamielennox): This is part of bug #1584008 and should default - # to True (or simply removed) in a major version bump. - self._case_sensitive = kwargs.pop('case_sensitive', False) - - def __getattr__(self, name): - return getattr(self._request, name) - - @property - def _url_parts(self): - if self._url_parts_ is None: - url = self._request.url - - if not self._case_sensitive: - url = url.lower() - - self._url_parts_ = urlparse.urlparse(url) - - return self._url_parts_ - - @property - def scheme(self): - return self._url_parts.scheme - - @property - def netloc(self): - return self._url_parts.netloc - - @property - def path(self): - return self._url_parts.path - - @property - def query(self): - return self._url_parts.query - - @property - def qs(self): - if self._qs is None: - self._qs = urlparse.parse_qs(self.query) - - return self._qs - - @property - def timeout(self): - return self._timeout - - @property - def allow_redirects(self): - return self._allow_redirects - - @property - def verify(self): - return self._verify - - @property - def cert(self): - return self._cert - - @property - def proxies(self): - return self._proxies - - @classmethod - def _create(cls, *args, **kwargs): - return cls(requests.Request(*args, **kwargs).prepare()) - - @property - def text(self): - body = self.body - - if isinstance(body, six.binary_type): - body = body.decode('utf-8') - - return body - - def json(self, **kwargs): - return json.loads(self.text, **kwargs) - - @property - def matcher(self): - """The matcher that this request was handled by. - - The matcher object is handled by a weakref. It will return the matcher - object if it is still available - so if the mock is still in place. If - the matcher is not available it will return None. - """ - return self._matcher() - - class _RequestHistoryTracker(object): def __init__(self): @@ -361,7 +249,7 @@ class Adapter(BaseAdapter, _RequestHistoryTracker): # Ideally case_sensitive would be a value passed to match() however # this would change the contract of matchers so we pass ito to the # proxy and the matcher seperately. - responses = [response._MatcherResponse(**k) for k in response_list] + responses = [_MatcherResponse(**k) for k in response_list] matcher = _Matcher(method, url, responses, diff --git a/requests_mock/request.py b/requests_mock/request.py new file mode 100644 index 0000000..b15d15a --- /dev/null +++ b/requests_mock/request.py @@ -0,0 +1,128 @@ +# 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 copy +import json + +import requests +import six +from six.moves.urllib import parse as urlparse + + +class _RequestObjectProxy(object): + """A wrapper around a requests.Request that gives some extra information. + + This will be important both for matching and so that when it's save into + the request_history users will be able to access these properties. + """ + + def __init__(self, request, **kwargs): + self._request = request + self._matcher = None + self._url_parts_ = None + self._qs = None + + # All of these params should always exist but we use a default + # to make the test setup easier. + self._timeout = kwargs.pop('timeout', None) + self._allow_redirects = kwargs.pop('allow_redirects', None) + self._verify = kwargs.pop('verify', None) + self._cert = kwargs.pop('cert', None) + self._proxies = copy.deepcopy(kwargs.pop('proxies', {})) + + # FIXME(jamielennox): This is part of bug #1584008 and should default + # to True (or simply removed) in a major version bump. + self._case_sensitive = kwargs.pop('case_sensitive', False) + + def __getattr__(self, name): + return getattr(self._request, name) + + @property + def _url_parts(self): + if self._url_parts_ is None: + url = self._request.url + + if not self._case_sensitive: + url = url.lower() + + self._url_parts_ = urlparse.urlparse(url) + + return self._url_parts_ + + @property + def scheme(self): + return self._url_parts.scheme + + @property + def netloc(self): + return self._url_parts.netloc + + @property + def path(self): + return self._url_parts.path + + @property + def query(self): + return self._url_parts.query + + @property + def qs(self): + if self._qs is None: + self._qs = urlparse.parse_qs(self.query) + + return self._qs + + @property + def timeout(self): + return self._timeout + + @property + def allow_redirects(self): + return self._allow_redirects + + @property + def verify(self): + return self._verify + + @property + def cert(self): + return self._cert + + @property + def proxies(self): + return self._proxies + + @classmethod + def _create(cls, *args, **kwargs): + return cls(requests.Request(*args, **kwargs).prepare()) + + @property + def text(self): + body = self.body + + if isinstance(body, six.binary_type): + body = body.decode('utf-8') + + return body + + def json(self, **kwargs): + return json.loads(self.text, **kwargs) + + @property + def matcher(self): + """The matcher that this request was handled by. + + The matcher object is handled by a weakref. It will return the matcher + object if it is still available - so if the mock is still in place. If + the matcher is not available it will return None. + """ + return self._matcher() diff --git a/requests_mock/tests/test_adapter.py b/requests_mock/tests/test_adapter.py index 8ad86ca..3d370d2 100644 --- a/requests_mock/tests/test_adapter.py +++ b/requests_mock/tests/test_adapter.py @@ -584,94 +584,6 @@ class SessionAdapterTests(base.TestCase): self.assertEqual(set(['/foo', '/bar']), set(resp.cookies.list_paths())) self.assertEqual(['.example.com'], resp.cookies.list_domains()) - def test_base_params(self): - data = 'testdata' - self.adapter.register_uri('GET', self.url, text=data) - resp = self.session.get(self.url) - - self.assertEqual('GET', self.adapter.last_request.method) - self.assertEqual(200, resp.status_code) - self.assertEqual(data, resp.text) - - self.assertIs(None, self.adapter.last_request.allow_redirects) - self.assertIs(None, self.adapter.last_request.timeout) - self.assertIs(True, self.adapter.last_request.verify) - self.assertIs(None, self.adapter.last_request.cert) - - # actually it's an OrderedDict, but equality works fine - self.assertEqual({}, self.adapter.last_request.proxies) - - def test_allow_redirects(self): - data = 'testdata' - self.adapter.register_uri('GET', self.url, text=data, status_code=300) - resp = self.session.get(self.url, allow_redirects=False) - - self.assertEqual('GET', self.adapter.last_request.method) - self.assertEqual(300, resp.status_code) - self.assertEqual(data, resp.text) - self.assertFalse(self.adapter.last_request.allow_redirects) - - def test_timeout(self): - data = 'testdata' - timeout = 300 - - self.adapter.register_uri('GET', self.url, text=data) - resp = self.session.get(self.url, timeout=timeout) - - self.assertEqual('GET', self.adapter.last_request.method) - self.assertEqual(200, resp.status_code) - self.assertEqual(data, resp.text) - self.assertEqual(timeout, self.adapter.last_request.timeout) - - def test_verify_false(self): - data = 'testdata' - verify = False - - self.adapter.register_uri('GET', self.url, text=data) - resp = self.session.get(self.url, verify=verify) - self.assertEqual('GET', self.adapter.last_request.method) - self.assertEqual(200, resp.status_code) - self.assertEqual(data, resp.text) - self.assertIs(verify, self.adapter.last_request.verify) - - def test_verify_path(self): - data = 'testdata' - verify = '/path/to/cacerts.pem' - - self.adapter.register_uri('GET', self.url, text=data) - resp = self.session.get(self.url, verify=verify) - self.assertEqual('GET', self.adapter.last_request.method) - self.assertEqual(200, resp.status_code) - self.assertEqual(data, resp.text) - self.assertEqual(verify, self.adapter.last_request.verify) - - def test_certs(self): - data = 'testdata' - cert = ('/path/to/cert.pem', 'path/to/key.pem') - - self.adapter.register_uri('GET', self.url, text=data) - resp = self.session.get(self.url, cert=cert) - - self.assertEqual('GET', self.adapter.last_request.method) - self.assertEqual(200, resp.status_code) - self.assertEqual(data, resp.text) - self.assertEqual(cert, self.adapter.last_request.cert) - self.assertTrue(self.adapter.last_request.verify) - - def test_proxies(self): - data = 'testdata' - proxies = {'http': 'foo.bar:3128', - 'http://host.name': 'foo.bar:4012'} - - self.adapter.register_uri('GET', self.url, text=data) - resp = self.session.get(self.url, proxies=proxies) - - self.assertEqual('GET', self.adapter.last_request.method) - self.assertEqual(200, resp.status_code) - self.assertEqual(data, resp.text) - self.assertEqual(proxies, self.adapter.last_request.proxies) - self.assertIsNot(proxies, self.adapter.last_request.proxies) - def test_reading_closed_fp(self): self.adapter.register_uri('GET', self.url, text='abc') resp = self.session.get(self.url) diff --git a/requests_mock/tests/test_request.py b/requests_mock/tests/test_request.py new file mode 100644 index 0000000..d0af499 --- /dev/null +++ b/requests_mock/tests/test_request.py @@ -0,0 +1,91 @@ +# 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 uuid + +import requests +import requests_mock +from requests_mock.tests import base + + +class RequestTests(base.TestCase): + + def setUp(self): + super(RequestTests, self).setUp() + + self.mocker = requests_mock.Mocker() + self.addCleanup(self.mocker.stop) + self.mocker.start() + + def do_request(self, **kwargs): + method = kwargs.pop('method', 'GET') + url = kwargs.pop('url', 'http://test.example.com/path') + status_code = kwargs.pop('status_code', 200) + data = uuid.uuid4().hex + + m = self.mocker.register_uri(method, + url, + text=data, + status_code=status_code) + + resp = requests.request(method, url, **kwargs) + + self.assertEqual(status_code, resp.status_code) + self.assertEqual(data, resp.text) + + self.assertTrue(m.called_once) + return m.last_request + + def test_base_params(self): + req = self.do_request(method='GET', status_code=200) + + self.assertIs(None, req.allow_redirects) + self.assertIs(None, req.timeout) + self.assertIs(True, req.verify) + self.assertIs(None, req.cert) + + # actually it's an OrderedDict, but equality works fine + self.assertEqual({}, req.proxies) + + def test_allow_redirects(self): + req = self.do_request(allow_redirects=False, status_code=300) + self.assertFalse(req.allow_redirects) + + def test_timeout(self): + timeout = 300 + req = self.do_request(timeout=timeout) + self.assertEqual(timeout, req.timeout) + + def test_verify_false(self): + verify = False + req = self.do_request(verify=verify) + self.assertIs(verify, req.verify) + + def test_verify_path(self): + verify = '/path/to/cacerts.pem' + req = self.do_request(verify=verify) + self.assertEqual(verify, req.verify) + + def test_certs(self): + cert = ('/path/to/cert.pem', 'path/to/key.pem') + req = self.do_request(cert=cert) + self.assertEqual(cert, req.cert) + self.assertTrue(req.verify) + + def test_proxies(self): + proxies = {'http': 'foo.bar:3128', + 'http://host.name': 'foo.bar:4012'} + + req = self.do_request(proxies=proxies) + + self.assertEqual(proxies, req.proxies) + self.assertIsNot(proxies, req.proxies)