1246046ffe
There is a difference between how urlparse handles if it is a protocol it recognises vs a general URI. This only seems to exist on pypy and py2.6. This is not something that should be managed by us. Let it be handled by the standard urlparse mechanism.
370 lines
13 KiB
Python
370 lines
13 KiB
Python
# 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 re
|
|
|
|
import requests
|
|
import six
|
|
from six.moves.urllib import parse as urlparse
|
|
|
|
import requests_mock
|
|
from requests_mock.tests import base
|
|
|
|
|
|
class SessionAdapterTests(base.TestCase):
|
|
|
|
PREFIX = "mock"
|
|
|
|
def setUp(self):
|
|
super(SessionAdapterTests, self).setUp()
|
|
|
|
self.adapter = requests_mock.Adapter()
|
|
self.session = requests.Session()
|
|
self.session.mount(self.PREFIX, self.adapter)
|
|
|
|
self.url = '%s://example.com/test' % self.PREFIX
|
|
self.headers = {'header_a': 'A', 'header_b': 'B'}
|
|
|
|
def assertHeaders(self, resp):
|
|
for k, v in six.iteritems(self.headers):
|
|
self.assertEqual(v, resp.headers[k])
|
|
|
|
def assertLastRequest(self, method='GET', body=None):
|
|
self.assertEqual(self.url, self.adapter.last_request.url)
|
|
self.assertEqual(method, self.adapter.last_request.method)
|
|
self.assertEqual(body, self.adapter.last_request.body)
|
|
|
|
url_parts = urlparse.urlparse(self.url)
|
|
qs = urlparse.parse_qs(url_parts.query)
|
|
self.assertEqual(url_parts.scheme, self.adapter.last_request.scheme)
|
|
self.assertEqual(url_parts.netloc, self.adapter.last_request.netloc)
|
|
self.assertEqual(url_parts.path, self.adapter.last_request.path)
|
|
self.assertEqual(url_parts.query, self.adapter.last_request.query)
|
|
self.assertEqual(url_parts.query, self.adapter.last_request.query)
|
|
self.assertEqual(qs, self.adapter.last_request.qs)
|
|
|
|
def test_content(self):
|
|
data = six.b('testdata')
|
|
|
|
self.adapter.register_uri('GET',
|
|
self.url,
|
|
content=data,
|
|
headers=self.headers)
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual(data, resp.content)
|
|
self.assertHeaders(resp)
|
|
self.assertLastRequest()
|
|
|
|
def test_content_callback(self):
|
|
status_code = 401
|
|
data = six.b('testdata')
|
|
|
|
def _content_cb(request, context):
|
|
context.status_code = status_code
|
|
context.headers.update(self.headers)
|
|
return data
|
|
|
|
self.adapter.register_uri('GET',
|
|
self.url,
|
|
content=_content_cb)
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual(status_code, resp.status_code)
|
|
self.assertEqual(data, resp.content)
|
|
self.assertHeaders(resp)
|
|
self.assertLastRequest()
|
|
|
|
def test_text(self):
|
|
data = 'testdata'
|
|
|
|
self.adapter.register_uri('GET',
|
|
self.url,
|
|
text=data,
|
|
headers=self.headers)
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual(six.b(data), resp.content)
|
|
self.assertEqual(six.u(data), resp.text)
|
|
self.assertEqual('utf-8', resp.encoding)
|
|
self.assertHeaders(resp)
|
|
self.assertLastRequest()
|
|
|
|
def test_text_callback(self):
|
|
status_code = 401
|
|
data = 'testdata'
|
|
|
|
def _text_cb(request, context):
|
|
context.status_code = status_code
|
|
context.headers.update(self.headers)
|
|
return six.u(data)
|
|
|
|
self.adapter.register_uri('GET', self.url, text=_text_cb)
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual(status_code, resp.status_code)
|
|
self.assertEqual(six.u(data), resp.text)
|
|
self.assertEqual(six.b(data), resp.content)
|
|
self.assertEqual('utf-8', resp.encoding)
|
|
self.assertHeaders(resp)
|
|
self.assertLastRequest()
|
|
|
|
def test_json(self):
|
|
json_data = {'hello': 'world'}
|
|
self.adapter.register_uri('GET',
|
|
self.url,
|
|
json=json_data,
|
|
headers=self.headers)
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual(six.b('{"hello": "world"}'), resp.content)
|
|
self.assertEqual(six.u('{"hello": "world"}'), resp.text)
|
|
self.assertEqual(json_data, resp.json())
|
|
self.assertEqual('utf-8', resp.encoding)
|
|
self.assertHeaders(resp)
|
|
self.assertLastRequest()
|
|
|
|
def test_json_callback(self):
|
|
status_code = 401
|
|
json_data = {'hello': 'world'}
|
|
data = '{"hello": "world"}'
|
|
|
|
def _json_cb(request, context):
|
|
context.status_code = status_code
|
|
context.headers.update(self.headers)
|
|
return json_data
|
|
|
|
self.adapter.register_uri('GET', self.url, json=_json_cb)
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual(status_code, resp.status_code)
|
|
self.assertEqual(json_data, resp.json())
|
|
self.assertEqual(six.u(data), resp.text)
|
|
self.assertEqual(six.b(data), resp.content)
|
|
self.assertEqual('utf-8', resp.encoding)
|
|
self.assertHeaders(resp)
|
|
self.assertLastRequest()
|
|
|
|
def test_no_body(self):
|
|
self.adapter.register_uri('GET', self.url)
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual(six.b(''), resp.content)
|
|
self.assertEqual(200, resp.status_code)
|
|
|
|
def test_multiple_body_elements(self):
|
|
self.assertRaises(RuntimeError,
|
|
self.adapter.register_uri,
|
|
self.url,
|
|
'GET',
|
|
content=six.b('b'),
|
|
text=six.u('u'))
|
|
|
|
def test_multiple_responses(self):
|
|
inp = [{'status_code': 400, 'text': 'abcd'},
|
|
{'status_code': 300, 'text': 'defg'},
|
|
{'status_code': 200, 'text': 'hijk'}]
|
|
|
|
self.adapter.register_uri('GET', self.url, inp)
|
|
out = [self.session.get(self.url) for i in range(0, len(inp))]
|
|
|
|
for i, o in zip(inp, out):
|
|
for k, v in six.iteritems(i):
|
|
self.assertEqual(v, getattr(o, k))
|
|
|
|
last = self.session.get(self.url)
|
|
for k, v in six.iteritems(inp[-1]):
|
|
self.assertEqual(v, getattr(last, k))
|
|
|
|
def test_callback_optional_status(self):
|
|
headers = {'a': 'b'}
|
|
|
|
def _test_cb(request, context):
|
|
context.headers.update(headers)
|
|
return ''
|
|
|
|
self.adapter.register_uri('GET',
|
|
self.url,
|
|
text=_test_cb,
|
|
status_code=300)
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual(300, resp.status_code)
|
|
|
|
for k, v in six.iteritems(headers):
|
|
self.assertEqual(v, resp.headers[k])
|
|
|
|
def test_callback_optional_headers(self):
|
|
headers = {'a': 'b'}
|
|
|
|
def _test_cb(request, context):
|
|
context.status_code = 300
|
|
return ''
|
|
|
|
self.adapter.register_uri('GET',
|
|
self.url,
|
|
text=_test_cb,
|
|
headers=headers)
|
|
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual(300, resp.status_code)
|
|
|
|
for k, v in six.iteritems(headers):
|
|
self.assertEqual(v, resp.headers[k])
|
|
|
|
def test_latest_register_overrides(self):
|
|
self.adapter.register_uri('GET', self.url, text='abc')
|
|
self.adapter.register_uri('GET', self.url, text='def')
|
|
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual('def', resp.text)
|
|
|
|
def test_no_last_request(self):
|
|
self.assertIsNone(self.adapter.last_request)
|
|
self.assertEqual(0, len(self.adapter.request_history))
|
|
|
|
def test_dont_pass_list_and_kwargs(self):
|
|
self.assertRaises(RuntimeError,
|
|
self.adapter.register_uri,
|
|
'GET',
|
|
self.url,
|
|
[{'text': 'a'}],
|
|
headers={'a': 'b'})
|
|
|
|
def test_empty_string_return(self):
|
|
# '' evaluates as False, so make sure an empty string is not ignored.
|
|
self.adapter.register_uri('GET', self.url, text='')
|
|
resp = self.session.get(self.url)
|
|
self.assertEqual('', resp.text)
|
|
|
|
def test_dont_pass_multiple_bodies(self):
|
|
self.assertRaises(RuntimeError,
|
|
self.adapter.register_uri,
|
|
'GET',
|
|
self.url,
|
|
json={'abc': 'def'},
|
|
text='ghi')
|
|
|
|
def test_dont_pass_unexpected_kwargs(self):
|
|
self.assertRaises(TypeError,
|
|
self.adapter.register_uri,
|
|
'GET',
|
|
self.url,
|
|
unknown='argument')
|
|
|
|
def test_dont_pass_unicode_as_content(self):
|
|
self.assertRaises(TypeError,
|
|
self.adapter.register_uri,
|
|
'GET',
|
|
self.url,
|
|
content=six.u('unicode'))
|
|
|
|
def test_dont_pass_bytes_as_text(self):
|
|
if six.PY2:
|
|
self.skipTest('Cannot enforce byte behaviour in PY2')
|
|
|
|
self.assertRaises(TypeError,
|
|
self.adapter.register_uri,
|
|
'GET',
|
|
self.url,
|
|
text=six.b('bytes'))
|
|
|
|
def test_dont_pass_non_str_as_content(self):
|
|
self.assertRaises(TypeError,
|
|
self.adapter.register_uri,
|
|
'GET',
|
|
self.url,
|
|
content=5)
|
|
|
|
def test_dont_pass_non_str_as_text(self):
|
|
self.assertRaises(TypeError,
|
|
self.adapter.register_uri,
|
|
'GET',
|
|
self.url,
|
|
text=5)
|
|
|
|
def test_with_any_method(self):
|
|
self.adapter.register_uri(requests_mock.ANY, self.url, text='resp')
|
|
|
|
for m in ('GET', 'HEAD', 'POST', 'UNKNOWN'):
|
|
resp = self.session.request(m, self.url)
|
|
self.assertEqual('resp', resp.text)
|
|
|
|
def test_with_any_url(self):
|
|
self.adapter.register_uri('GET', requests_mock.ANY, text='resp')
|
|
|
|
for u in ('mock://a', 'mock://b', 'mock://c'):
|
|
resp = self.session.get(u)
|
|
self.assertEqual('resp', resp.text)
|
|
|
|
def test_with_regexp(self):
|
|
self.adapter.register_uri('GET', re.compile('tester.com'), text='resp')
|
|
|
|
for u in ('mocK://www.tester.com/a', 'mock://abc.tester.com'):
|
|
resp = self.session.get(u)
|
|
self.assertEqual('resp', resp.text)
|
|
|
|
def test_requests_in_history_on_no_match(self):
|
|
self.assertRaises(requests_mock.NoMockAddress,
|
|
self.session.get,
|
|
self.url)
|
|
|
|
self.assertEqual(self.url, self.adapter.last_request.url)
|
|
|
|
def test_requests_in_history_on_exception(self):
|
|
|
|
class MyExc(Exception):
|
|
pass
|
|
|
|
def _test_cb(request, ctx):
|
|
raise MyExc()
|
|
|
|
self.adapter.register_uri('GET', self.url, text=_test_cb)
|
|
|
|
self.assertRaises(MyExc,
|
|
self.session.get,
|
|
self.url)
|
|
|
|
self.assertEqual(self.url, self.adapter.last_request.url)
|
|
|
|
def test_not_called_and_called_count(self):
|
|
m = self.adapter.register_uri('GET', self.url, text='resp')
|
|
self.assertEqual(0, m.call_count)
|
|
self.assertFalse(m.called)
|
|
|
|
self.assertEqual(0, self.adapter.call_count)
|
|
self.assertFalse(self.adapter.called)
|
|
|
|
def test_called_and_called_count(self):
|
|
m = self.adapter.register_uri('GET', self.url, text='resp')
|
|
|
|
resps = [self.session.get(self.url) for i in range(0, 3)]
|
|
|
|
for r in resps:
|
|
self.assertEqual('resp', r.text)
|
|
self.assertEqual(200, r.status_code)
|
|
|
|
self.assertEqual(len(resps), m.call_count)
|
|
self.assertTrue(m.called)
|
|
|
|
self.assertEqual(len(resps), self.adapter.call_count)
|
|
self.assertTrue(self.adapter.called)
|
|
|
|
def test_adapter_picks_correct_adatper(self):
|
|
good = '%s://test3.url/' % self.PREFIX
|
|
self.adapter.register_uri('GET',
|
|
'%s://test1.url' % self.PREFIX,
|
|
text='bad')
|
|
self.adapter.register_uri('GET',
|
|
'%s://test2.url' % self.PREFIX,
|
|
text='bad')
|
|
self.adapter.register_uri('GET', good, text='good')
|
|
self.adapter.register_uri('GET',
|
|
'%s://test4.url' % self.PREFIX,
|
|
text='bad')
|
|
|
|
resp = self.session.get(good)
|
|
|
|
self.assertEqual('good', resp.text)
|