383 lines
16 KiB
Python
383 lines
16 KiB
Python
import logging
|
|
|
|
import pytest
|
|
import mohawk
|
|
import mock
|
|
from httmock import urlmatch, HTTMock, response
|
|
|
|
from django.test import override_settings
|
|
|
|
from passerelle.utils import Request, CaseInsensitiveDict
|
|
from passerelle.utils.http_authenticators import HawkAuth
|
|
import utils
|
|
from utils import FakedResponse
|
|
|
|
|
|
class MockFileField(object):
|
|
def __init__(self, path):
|
|
self.path = path
|
|
|
|
|
|
class MockResource(object):
|
|
logger = logging.getLogger('requests')
|
|
basic_auth_username = ''
|
|
basic_auth_password = ''
|
|
client_certificate = None
|
|
trusted_certificate_authorities = None
|
|
verify_cert = True
|
|
http_proxy = ''
|
|
|
|
|
|
@pytest.fixture(params=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])
|
|
def log_level(request):
|
|
return request.param
|
|
|
|
@urlmatch(netloc=r'(.*\.)?httpbin\.org$')
|
|
def httpbin_mock(url, request):
|
|
return response(200, {"message": "Are you really josh ?"},
|
|
headers={"Content-Type": "application/json"}, request=request)
|
|
|
|
@urlmatch(netloc=r'(.*\.)?httperror\.org$')
|
|
def http400_mock(url, request):
|
|
return response(400, {"foo": "bar"},
|
|
headers={"Content-Type": "application/json"}, request=request)
|
|
|
|
|
|
def test_log_level(caplog, log_level):
|
|
url = 'https://httpbin.org/post'
|
|
|
|
logger = logging.getLogger('requests')
|
|
logger.setLevel(log_level)
|
|
|
|
with HTTMock(httpbin_mock):
|
|
requests = Request(logger=logger)
|
|
response = requests.post(url, json={'name':'josh'})
|
|
|
|
records = [record for record in caplog.records if record.name == 'requests']
|
|
|
|
records_length = len(records)
|
|
|
|
if logger.level > 20:
|
|
assert records_length == 0
|
|
else:
|
|
assert records_length == 1
|
|
|
|
record = records[0]
|
|
assert record.request_url == url
|
|
assert record.response_status == response.status_code
|
|
if logger.level == 10: # DEBUG
|
|
assert record.request_payload == '\'{"name": "josh"}\''
|
|
assert record.response_content == repr(response.content)
|
|
assert record.response_headers
|
|
else:
|
|
assert not hasattr(record, 'request_payload')
|
|
assert not hasattr(record, 'response_content')
|
|
assert not hasattr(record, 'response_headers')
|
|
|
|
def test_log_error(caplog, log_level):
|
|
url = 'https://httperror.org/plop'
|
|
|
|
logger = logging.getLogger('requests')
|
|
logger.setLevel(log_level)
|
|
|
|
with HTTMock(http400_mock):
|
|
requests = Request(logger=logger)
|
|
response = requests.post(url, json={'name':'josh'})
|
|
|
|
records = [record for record in caplog.records if record.name == 'requests']
|
|
|
|
records_length = len(records)
|
|
|
|
if logger.level > 40:
|
|
assert records_length == 0
|
|
else:
|
|
assert records_length == 1
|
|
|
|
record = records[0]
|
|
assert record.request_url == url
|
|
assert record.response_status == response.status_code
|
|
if logger.level == 10: # DEBUG
|
|
assert record.request_payload == '\'{"name": "josh"}\''
|
|
assert record.response_content == repr(response.content)
|
|
assert record.response_headers
|
|
else:
|
|
assert not hasattr(record, 'request_payload')
|
|
assert not hasattr(record, 'response_content')
|
|
assert not hasattr(record, 'response_headers')
|
|
|
|
def test_log_error_request_max_size(caplog, log_level, settings):
|
|
url = 'https://httperror.org/plop'
|
|
|
|
logger = logging.getLogger('requests')
|
|
logger.setLevel(log_level)
|
|
|
|
settings.LOGGED_REQUEST_MAX_SIZE = 8
|
|
with HTTMock(http400_mock):
|
|
requests = Request(logger=logger)
|
|
response = requests.post(url, json={'name':'josh'})
|
|
|
|
print logger.level
|
|
if logger.level == 10: # DEBUG
|
|
records = [record for record in caplog.records if record.name == 'requests']
|
|
assert records[0].request_payload == '\'{"name":\''
|
|
|
|
|
|
@pytest.fixture(params=['xml', 'whatever', 'jpeg', 'pdf'])
|
|
def endpoint_response(request):
|
|
response_request = mock.Mock(
|
|
headers={'Accept': '*/*', 'Authorization': 'Basic dG9rZW46dG9rZW4='}, body=None)
|
|
xml = FakedResponse(
|
|
headers={'Content-Type': 'application/xml; charset=charset=utf-8'}, status_code=200,
|
|
content='<tests><test>xml test</test></tests>', request=response_request)
|
|
|
|
whatever = FakedResponse(
|
|
headers={'Content-Type': 'texto/csv'}, status_code=200,
|
|
content='username;age\ntoken;10\ncartman:10', request=response_request)
|
|
|
|
jpeg = FakedResponse(
|
|
headers={'Content-Type': 'image/jpeg'}, status_code=200,
|
|
content='binary content to be ignored', request=response_request)
|
|
|
|
pdf = FakedResponse(
|
|
headers={'Content-Type': 'application/pdf'}, status_code=200,
|
|
content='binary content to be ignored', request=response_request)
|
|
|
|
return locals().get(request.param)
|
|
|
|
|
|
@mock.patch('passerelle.utils.RequestSession.send')
|
|
def test_skip_content_type(mocked_get, caplog, endpoint_response):
|
|
mocked_get.return_value = endpoint_response
|
|
logger = logging.getLogger('requests')
|
|
logger.setLevel(logging.DEBUG)
|
|
requests = Request(logger=logger)
|
|
response = requests.get('http://example.net/whatever').body
|
|
records = [record for record in caplog.records if record.name == 'requests']
|
|
|
|
if 'xml' in endpoint_response.headers.get('Content-Type'):
|
|
assert len(records) == 1
|
|
assert records[0].response_content == "'<tests><test>xml test</test></tests>'"
|
|
else:
|
|
assert len(records) == 1
|
|
|
|
|
|
@mock.patch('passerelle.utils.RequestSession.request')
|
|
def test_proxies(mocked_get, caplog, endpoint_response):
|
|
mocked_get.return_value = endpoint_response
|
|
logger = logging.getLogger('requests')
|
|
Request(logger=logger).get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('proxies') is None
|
|
Request(logger=logger).get('http://example.net/whatever',
|
|
proxies={'http': 'http://proxy'})
|
|
assert mocked_get.call_args[1].get('proxies') == {'http': 'http://proxy'}
|
|
|
|
with override_settings(REQUESTS_PROXIES={'http': 'http://globalproxy'}):
|
|
Request(logger=logger).get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('proxies') == {'http': 'http://globalproxy'}
|
|
Request(logger=logger).get('http://example.net/whatever',
|
|
proxies={'http': 'http://proxy'})
|
|
assert mocked_get.call_args[1].get('proxies') == {'http': 'http://proxy'}
|
|
|
|
# with a linked resource
|
|
resource = MockResource()
|
|
request = Request(resource=resource, logger=logger)
|
|
request.get('http://example.net/whatever')
|
|
assert 'proxies' not in mocked_get.call_args[1]
|
|
|
|
resource.http_proxy = 'http://resourceproxy'
|
|
resource_proxies = {'http': 'http://resourceproxy', 'https': 'http://resourceproxy'}
|
|
|
|
request.get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('proxies') == resource_proxies
|
|
request.get('http://example.net/whatever', proxies={'http': 'http://proxy'})
|
|
assert mocked_get.call_args[1].get('proxies') == {'http': 'http://proxy'}
|
|
|
|
with override_settings(REQUESTS_PROXIES={'http': 'http://globalproxy'}):
|
|
request.get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('proxies') == resource_proxies
|
|
request.get('http://example.net/whatever', proxies={'http': 'http://proxy'})
|
|
assert mocked_get.call_args[1].get('proxies') == {'http': 'http://proxy'}
|
|
resource.http_proxy = None
|
|
request.get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('proxies') == {'http': 'http://globalproxy'}
|
|
request.get('http://example.net/whatever', proxies={'http': 'http://proxy'})
|
|
assert mocked_get.call_args[1].get('proxies') == {'http': 'http://proxy'}
|
|
|
|
|
|
@mock.patch('passerelle.utils.RequestSession.request')
|
|
def test_resource_auth(mocked_get, caplog, endpoint_response):
|
|
mocked_get.return_value = endpoint_response
|
|
logger = logging.getLogger('requests')
|
|
resource = MockResource()
|
|
request = Request(resource=resource, logger=logger)
|
|
|
|
request.get('http://example.net/whatever')
|
|
assert 'auth' not in mocked_get.call_args[1]
|
|
|
|
request.get('http://example.net/whatever', auth=('user', 'pass'))
|
|
assert mocked_get.call_args[1].get('auth') == ('user', 'pass')
|
|
|
|
resource.basic_auth_username = 'ruser'
|
|
resource.basic_auth_password = 'rpass'
|
|
request.get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('auth') == ('ruser', 'rpass')
|
|
|
|
request.get('http://example.net/whatever', auth=('user', 'pass'))
|
|
assert mocked_get.call_args[1].get('auth') == ('user', 'pass')
|
|
|
|
request.get('http://example.net/whatever', auth=None)
|
|
assert mocked_get.call_args[1].get('auth') is None
|
|
|
|
@mock.patch('passerelle.utils.RequestSession.send')
|
|
def test_resource_hawk_auth(mocked_send, caplog, endpoint_response):
|
|
mocked_send.return_value = endpoint_response
|
|
logger = logging.getLogger('requests')
|
|
resource = MockResource()
|
|
request = Request(resource=resource, logger=logger)
|
|
|
|
credentials = {'id': 'id', 'key': 'key', 'algorithm': 'sha256'}
|
|
hawk_auth = HawkAuth(**credentials)
|
|
|
|
resp = request.get('http://httpbin.org/get', auth=hawk_auth)
|
|
prepared_method = mocked_send.call_args[0][0]
|
|
assert 'Authorization' in prepared_method.headers
|
|
generated_header = prepared_method.headers['Authorization']
|
|
|
|
sender = mohawk.Sender(credentials, nonce=hawk_auth.nonce, _timestamp=hawk_auth.timestamp,
|
|
url='http://httpbin.org/get', method='GET', content_type='',
|
|
content='')
|
|
expected_header = sender.request_header
|
|
generated_parts = [tuple(e.strip().split('=', 1)) for e in generated_header[5:].split(',')]
|
|
expected_parts = [tuple(e.strip().split('=', 1)) for e in expected_header[5:].split(',')]
|
|
# compare generated header elements
|
|
assert dict(generated_parts) == dict(expected_parts)
|
|
|
|
hawk_auth = HawkAuth(ext='extra attribute', **credentials)
|
|
resp = request.post('http://httpbin.org/post', auth=hawk_auth, json={'key': 'value'})
|
|
prepared_method = mocked_send.call_args[0][0]
|
|
assert 'Authorization' in prepared_method.headers
|
|
generated_header = prepared_method.headers['Authorization']
|
|
|
|
sender = mohawk.Sender(credentials, nonce=hawk_auth.nonce, _timestamp=hawk_auth.timestamp,
|
|
url='http://httpbin.org/post', method='POST', content_type='application/json',
|
|
content='{"key": "value"}', ext="extra attribute")
|
|
expected_header = sender.request_header
|
|
|
|
generated_parts = [tuple(e.strip().split('=', 1)) for e in generated_header[5:].split(',')]
|
|
expected_parts = [tuple(e.strip().split('=', 1)) for e in expected_header[5:].split(',')]
|
|
assert dict(generated_parts) == dict(expected_parts)
|
|
|
|
|
|
@mock.patch('passerelle.utils.RequestSession.request')
|
|
def test_resource_certificates(mocked_get, caplog, endpoint_response):
|
|
mocked_get.return_value = endpoint_response
|
|
logger = logging.getLogger('requests')
|
|
resource = MockResource()
|
|
request = Request(resource=resource, logger=logger)
|
|
|
|
request.get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('verify') is True
|
|
assert 'cert' not in mocked_get.call_args[1]
|
|
|
|
resource.verify_cert = False
|
|
request.get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('verify') is False
|
|
|
|
resource.trusted_certificate_authorities = MockFileField('/ca.pem')
|
|
request.get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('verify') == '/ca.pem'
|
|
assert 'cert' not in mocked_get.call_args[1]
|
|
request.get('http://example.net/whatever', verify=False)
|
|
assert mocked_get.call_args[1].get('verify') is False
|
|
|
|
resource.client_certificate = MockFileField('/client.pem')
|
|
request.get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1].get('cert') == '/client.pem'
|
|
assert mocked_get.call_args[1].get('verify') == '/ca.pem'
|
|
request.get('http://example.net/whatever', cert='/local.pem', verify=False)
|
|
assert mocked_get.call_args[1].get('cert') == '/local.pem'
|
|
assert mocked_get.call_args[1].get('verify') is False
|
|
|
|
|
|
@mock.patch('passerelle.utils.RequestSession.request')
|
|
def test_requests_cache(mocked_get, caplog):
|
|
resource = MockResource()
|
|
logger = logging.getLogger('requests')
|
|
request = Request(resource=resource, logger=logger)
|
|
|
|
response_request = mock.Mock(headers={'Accept': '*/*'}, body=None)
|
|
mocked_get.return_value = FakedResponse(
|
|
headers={'Content-Type': 'text/plain; charset=charset=utf-8'},
|
|
request=response_request,
|
|
content='hello world', status_code=200)
|
|
|
|
# by default there is no cache
|
|
assert request.get('http://cache.example.org/').content == 'hello world'
|
|
assert request.get('http://cache.example.org/').content == 'hello world'
|
|
assert mocked_get.call_count == 2
|
|
|
|
# add some cache
|
|
mocked_get.reset_mock()
|
|
assert request.get('http://cache.example.org/', cache_duration=15).content == 'hello world'
|
|
assert mocked_get.call_count == 1
|
|
assert request.get('http://cache.example.org/', cache_duration=15).content == 'hello world'
|
|
assert mocked_get.call_count == 1 # got a cached response
|
|
|
|
# value changed
|
|
mocked_get.return_value = FakedResponse(
|
|
headers={'Content-Type': 'text/plain; charset=charset=utf-8'},
|
|
request=response_request,
|
|
content='hello second world', status_code=200)
|
|
assert request.get('http://cache.example.org/', cache_duration=15).content == 'hello world'
|
|
assert mocked_get.call_count == 1
|
|
|
|
# force cache invalidation
|
|
assert request.get('http://cache.example.org/', invalidate_cache=True).content == 'hello second world'
|
|
assert mocked_get.call_count == 2
|
|
|
|
# do not cache errors
|
|
mocked_get.return_value = FakedResponse(
|
|
headers={'Content-Type': 'text/plain; charset=charset=utf-8'},
|
|
request=response_request,
|
|
content='no such world', status_code=404)
|
|
mocked_get.reset_mock()
|
|
response = request.get('http://cache.example.org/404', cache_duration=15)
|
|
assert response.content == 'no such world'
|
|
assert response.status_code == 404
|
|
assert mocked_get.call_count == 1
|
|
response = request.get('http://cache.example.org/404', cache_duration=15)
|
|
assert mocked_get.call_count == 2
|
|
|
|
# check response headers
|
|
mocked_get.reset_mock()
|
|
mocked_get.return_value = FakedResponse(
|
|
headers=CaseInsensitiveDict({'Content-Type': 'image/png'}),
|
|
request=response_request,
|
|
content='hello world', status_code=200)
|
|
assert request.get('http://cache.example.org/img', cache_duration=15).headers.get('content-type') == 'image/png'
|
|
assert mocked_get.call_count == 1
|
|
assert request.get('http://cache.example.org/img', cache_duration=15).headers.get('content-type') == 'image/png'
|
|
assert mocked_get.call_count == 1 # got a cached response
|
|
|
|
|
|
@mock.patch('passerelle.utils.RequestSession.request')
|
|
def test_timeout(mocked_get, caplog, endpoint_response):
|
|
mocked_get.return_value = endpoint_response
|
|
logger = logging.getLogger('requests')
|
|
|
|
Request(logger=logger).get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1]['timeout'] == 25
|
|
|
|
Request(logger=logger).get('http://example.net/whatever', timeout=42)
|
|
assert mocked_get.call_args[1]['timeout'] == 42
|
|
Request(logger=logger).get('http://example.net/whatever', timeout=None)
|
|
assert mocked_get.call_args[1]['timeout'] is None
|
|
|
|
with override_settings(REQUESTS_TIMEOUT=57):
|
|
Request(logger=logger).get('http://example.net/whatever')
|
|
assert mocked_get.call_args[1]['timeout'] == 57
|
|
Request(logger=logger).get('http://example.net/whatever', timeout=42)
|
|
assert mocked_get.call_args[1]['timeout'] == 42
|
|
Request(logger=logger).get('http://example.net/whatever', timeout=None)
|
|
assert mocked_get.call_args[1]['timeout'] is None
|