Jamie Lennox 86e33d1eff Move request code and tests into their own files
The Request proxy object is growing bigger and already has its own
category of tests. Move each into their own file with some simplified
helpers for testing.

Change-Id: I0dcc5d8d09feaf3febfcc8a4d114973096279c51
2016-10-13 15:03:21 +11:00

636 lines
22 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 json
import re
import requests
import six
from six.moves.urllib import parse as urlparse
import requests_mock
from requests_mock.tests import base
class MyExc(Exception):
pass
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):
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.assertFalse(m.called_once)
self.assertEqual(0, self.adapter.call_count)
self.assertFalse(self.adapter.called)
self.assertFalse(m.called_once)
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.assertFalse(m.called_once)
self.assertEqual(len(resps), self.adapter.call_count)
self.assertTrue(self.adapter.called)
self.assertFalse(m.called_once)
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)
def test_adapter_is_connection(self):
url = '%s://test.url' % self.PREFIX
text = 'text'
self.adapter.register_uri('GET', url, text=text)
resp = self.session.get(url)
self.assertEqual(text, resp.text)
self.assertIs(self.adapter, resp.connection)
def test_send_to_connection(self):
url1 = '%s://test1.url/' % self.PREFIX
url2 = '%s://test2.url/' % self.PREFIX
text1 = 'text1'
text2 = 'text2'
self.adapter.register_uri('GET', url1, text=text1)
self.adapter.register_uri('GET', url2, text=text2)
req = requests.Request(method='GET', url=url2).prepare()
resp1 = self.session.get(url1)
self.assertEqual(text1, resp1.text)
resp2 = resp1.connection.send(req)
self.assertEqual(text2, resp2.text)
def test_request_json_with_str_data(self):
dict_req = {'hello': 'world'}
dict_resp = {'goodbye': 'world'}
m = self.adapter.register_uri('POST', self.url, json=dict_resp)
data = json.dumps(dict_req)
resp = self.session.post(self.url, data=data)
self.assertIs(data, m.last_request.body)
self.assertEqual(dict_resp, resp.json())
self.assertEqual(dict_req, m.last_request.json())
def test_request_json_with_bytes_data(self):
dict_req = {'hello': 'world'}
dict_resp = {'goodbye': 'world'}
m = self.adapter.register_uri('POST', self.url, json=dict_resp)
data = json.dumps(dict_req).encode('utf-8')
resp = self.session.post(self.url, data=data)
self.assertIs(data, m.last_request.body)
self.assertEqual(dict_resp, resp.json())
self.assertEqual(dict_req, m.last_request.json())
def test_request_json_with_cb(self):
dict_req = {'hello': 'world'}
dict_resp = {'goodbye': 'world'}
data = json.dumps(dict_req)
def _cb(req, context):
self.assertEqual(dict_req, req.json())
return dict_resp
m = self.adapter.register_uri('POST', self.url, json=_cb)
resp = self.session.post(self.url, data=data)
self.assertEqual(1, m.call_count)
self.assertTrue(m.called_once)
self.assertEqual(dict_resp, resp.json())
def test_raises_exception(self):
self.adapter.register_uri('GET', self.url, exc=MyExc)
self.assertRaises(MyExc,
self.session.get,
self.url)
self.assertTrue(self.adapter.called_once)
self.assertEqual(self.url, self.adapter.last_request.url)
def test_raises_exception_with_body_args_fails(self):
self.assertRaises(TypeError,
self.adapter.register_uri,
'GET',
self.url,
exc=MyExc,
text='fail')
def test_sets_request_matcher_in_history(self):
url1 = '%s://test1.url/' % self.PREFIX
url2 = '%s://test2.url/' % self.PREFIX
text1 = 'text1'
text2 = 'text2'
m1 = self.adapter.register_uri('GET', url1, text=text1)
m2 = self.adapter.register_uri('GET', url2, text=text2)
resp1 = self.session.get(url1)
resp2 = self.session.get(url2)
self.assertEqual(text1, resp1.text)
self.assertEqual(text2, resp2.text)
self.assertEqual(2, self.adapter.call_count)
self.assertFalse(self.adapter.called_once)
self.assertEqual(url1, self.adapter.request_history[0].url)
self.assertEqual(url2, self.adapter.request_history[1].url)
self.assertIs(m1, self.adapter.request_history[0].matcher)
self.assertIs(m2, self.adapter.request_history[1].matcher)
def test_sets_request_matcher_on_exception(self):
m = self.adapter.register_uri('GET', self.url, exc=MyExc)
self.assertRaises(MyExc,
self.session.get,
self.url)
self.assertEqual(self.url, self.adapter.last_request.url)
self.assertIs(m, self.adapter.last_request.matcher)
def test_cookies_from_header(self):
headers = {'Set-Cookie': 'fig=newton; Path=/test; domain=.example.com'}
self.adapter.register_uri('GET',
self.url,
text='text',
headers=headers)
resp = self.session.get(self.url)
self.assertEqual('newton', resp.cookies['fig'])
self.assertEqual(['/test'], resp.cookies.list_paths())
self.assertEqual(['.example.com'], resp.cookies.list_domains())
def test_cookies_from_dict(self):
# This is a syntax we get from requests. I'm not sure i like it.
self.adapter.register_uri('GET',
self.url,
text='text',
cookies={'fig': 'newton', 'sugar': 'apple'})
resp = self.session.get(self.url)
self.assertEqual('newton', resp.cookies['fig'])
self.assertEqual('apple', resp.cookies['sugar'])
def test_cookies_with_jar(self):
jar = requests_mock.CookieJar()
jar.set('fig', 'newton', path='/foo', domain='.example.com')
jar.set('sugar', 'apple', path='/bar', domain='.example.com')
self.adapter.register_uri('GET', self.url, text='text', cookies=jar)
resp = self.session.get(self.url)
self.assertEqual('newton', resp.cookies['fig'])
self.assertEqual('apple', resp.cookies['sugar'])
self.assertEqual(set(['/foo', '/bar']), set(resp.cookies.list_paths()))
self.assertEqual(['.example.com'], resp.cookies.list_domains())
def test_cookies_header_with_cb(self):
def _cb(request, context):
val = 'fig=newton; Path=/test; domain=.example.com'
context.headers['Set-Cookie'] = val
return 'text'
self.adapter.register_uri('GET', self.url, text=_cb)
resp = self.session.get(self.url)
self.assertEqual('newton', resp.cookies['fig'])
self.assertEqual(['/test'], resp.cookies.list_paths())
self.assertEqual(['.example.com'], resp.cookies.list_domains())
def test_cookies_from_dict_with_cb(self):
def _cb(request, context):
# converted into a jar by now
context.cookies.set('sugar', 'apple', path='/test')
return 'text'
self.adapter.register_uri('GET',
self.url,
text=_cb,
cookies={'fig': 'newton'})
resp = self.session.get(self.url)
self.assertEqual('newton', resp.cookies['fig'])
self.assertEqual('apple', resp.cookies['sugar'])
self.assertEqual(['/', '/test'], resp.cookies.list_paths())
def test_cookies_with_jar_cb(self):
def _cb(request, context):
context.cookies.set('sugar',
'apple',
path='/bar',
domain='.example.com')
return 'text'
jar = requests_mock.CookieJar()
jar.set('fig', 'newton', path='/foo', domain='.example.com')
self.adapter.register_uri('GET', self.url, text=_cb, cookies=jar)
resp = self.session.get(self.url)
self.assertEqual('newton', resp.cookies['fig'])
self.assertEqual('apple', resp.cookies['sugar'])
self.assertEqual(set(['/foo', '/bar']), set(resp.cookies.list_paths()))
self.assertEqual(['.example.com'], resp.cookies.list_domains())
def test_reading_closed_fp(self):
self.adapter.register_uri('GET', self.url, text='abc')
resp = self.session.get(self.url)
# raw will have been closed during the request reading
self.assertTrue(resp.raw.closed)
data = resp.raw.read()
self.assertIsInstance(data, six.binary_type)
self.assertEqual(0, len(data))
def test_case_sensitive_headers(self):
data = 'testdata'
headers = {'aBcDe': 'FgHiJ'}
self.adapter.register_uri('GET', self.url, text=data)
resp = self.session.get(self.url, headers=headers)
self.assertEqual('GET', self.adapter.last_request.method)
self.assertEqual(200, resp.status_code)
self.assertEqual(data, resp.text)
for k, v in headers.items():
self.assertEqual(v, self.adapter.last_request.headers[k])
def test_case_sensitive_history(self):
self.adapter._case_sensitive = True
data = 'testdata'
netloc = 'examPlE.CoM'
path = '/TesTER'
query = 'aBC=deF'
mock_url = '%s://%s%s' % (self.PREFIX, netloc.lower(), path)
request_url = '%s://%s%s?%s' % (self.PREFIX, netloc, path, query)
# test that the netloc is ignored when actually making the request
self.adapter.register_uri('GET', mock_url, text=data)
resp = self.session.get(request_url)
self.assertEqual('GET', self.adapter.last_request.method)
self.assertEqual(200, resp.status_code)
self.assertEqual(data, resp.text)
# but even still the mixed case parameters come out in history
self.assertEqual(netloc, self.adapter.last_request.netloc)
self.assertEqual(path, self.adapter.last_request.path)
self.assertEqual(query, self.adapter.last_request.query)