# -*- coding: utf-8 import datetime try: import ujson as json except ImportError: import json import xml.etree.ElementTree as et import pytest import yaml import falcon import falcon.testing as testing @pytest.fixture def client(): app = falcon.API() resource = FaultyResource() app.add_route('/fail', resource) return testing.TestClient(app) class FaultyResource: def on_get(self, req, resp): status = req.get_header('X-Error-Status') title = req.get_header('X-Error-Title') description = req.get_header('X-Error-Description') code = 10042 raise falcon.HTTPError(status, title, description, code=code) def on_post(self, req, resp): raise falcon.HTTPForbidden( 'Request denied', 'You do not have write permissions for this queue.', href='http://example.com/api/rbac') def on_put(self, req, resp): raise falcon.HTTPError( falcon.HTTP_792, 'Internet crashed', 'Catastrophic weather event due to climate change.', href='http://example.com/api/climate', href_text='Drill baby drill!', code=8733224) def on_patch(self, req, resp): raise falcon.HTTPError(falcon.HTTP_400) class UnicodeFaultyResource(object): def __init__(self): self.called = False def on_get(self, req, resp): self.called = True raise falcon.HTTPError( falcon.HTTP_792, u'Internet \xe7rashed!', u'\xc7atastrophic weather event', href=u'http://example.com/api/\xe7limate', href_text=u'Drill b\xe1by drill!') class MiscErrorsResource: def __init__(self, exception, needs_title): self.needs_title = needs_title self._exception = exception def on_get(self, req, resp): if self.needs_title: raise self._exception('Excuse Us', 'Something went boink!') else: raise self._exception('Something went boink!') class UnauthorizedResource: def on_get(self, req, resp): raise falcon.HTTPUnauthorized('Authentication Required', 'Missing or invalid authorization.', ['Basic realm="simple"']) def on_post(self, req, resp): raise falcon.HTTPUnauthorized('Authentication Required', 'Missing or invalid authorization.', ['Newauth realm="apps"', 'Basic realm="simple"']) def on_put(self, req, resp): raise falcon.HTTPUnauthorized('Authentication Required', 'Missing or invalid authorization.', []) class NotFoundResource: def on_get(self, req, resp): raise falcon.HTTPNotFound() class NotFoundResourceWithBody: def on_get(self, req, resp): raise falcon.HTTPNotFound(description='Not Found') class GoneResource: def on_get(self, req, resp): raise falcon.HTTPGone() class GoneResourceWithBody: def on_get(self, req, resp): raise falcon.HTTPGone(description='Gone with the wind') class MethodNotAllowedResource: def on_get(self, req, resp): raise falcon.HTTPMethodNotAllowed(['PUT']) class MethodNotAllowedResourceWithHeaders: def on_get(self, req, resp): raise falcon.HTTPMethodNotAllowed(['PUT'], headers={ 'x-ping': 'pong'}) class MethodNotAllowedResourceWithHeadersWithAccept: def on_get(self, req, resp): raise falcon.HTTPMethodNotAllowed(['PUT'], headers={ 'x-ping': 'pong', 'accept': 'GET,PUT'}) class MethodNotAllowedResourceWithBody: def on_get(self, req, resp): raise falcon.HTTPMethodNotAllowed(['PUT'], description='Not Allowed') class LengthRequiredResource: def on_get(self, req, resp): raise falcon.HTTPLengthRequired('title', 'description') class RequestEntityTooLongResource: def on_get(self, req, resp): raise falcon.HTTPRequestEntityTooLarge('Request Rejected', 'Request Body Too Large') class TemporaryRequestEntityTooLongResource: def __init__(self, retry_after): self.retry_after = retry_after def on_get(self, req, resp): raise falcon.HTTPRequestEntityTooLarge('Request Rejected', 'Request Body Too Large', retry_after=self.retry_after) class UriTooLongResource: def __init__(self, title=None, description=None, code=None): self.title = title self.description = description self.code = code def on_get(self, req, resp): raise falcon.HTTPUriTooLong(self.title, self.description, code=self.code) class RangeNotSatisfiableResource: def on_get(self, req, resp): raise falcon.HTTPRangeNotSatisfiable(123456) class TooManyRequestsResource: def __init__(self, retry_after=None): self.retry_after = retry_after def on_get(self, req, resp): raise falcon.HTTPTooManyRequests('Too many requests', '1 per minute', retry_after=self.retry_after) class ServiceUnavailableResource: def __init__(self, retry_after): self.retry_after = retry_after def on_get(self, req, resp): raise falcon.HTTPServiceUnavailable('Oops', 'Stand by...', retry_after=self.retry_after) class InvalidHeaderResource: def on_get(self, req, resp): raise falcon.HTTPInvalidHeader( 'Please provide a valid token.', 'X-Auth-Token', code='A1001') class MissingHeaderResource: def on_get(self, req, resp): raise falcon.HTTPMissingHeader('X-Auth-Token') class InvalidParamResource: def on_get(self, req, resp): raise falcon.HTTPInvalidParam( 'The value must be a hex-encoded UUID.', 'id', code='P1002') class MissingParamResource: def on_get(self, req, resp): raise falcon.HTTPMissingParam('id', code='P1003') class TestHTTPError(object): def _misc_test(self, client, exception, status, needs_title=True): client.app.add_route('/misc', MiscErrorsResource(exception, needs_title)) response = client.simulate_request(path='/misc') assert response.status == status def test_base_class(self, client): headers = { 'X-Error-Title': 'Storage service down', 'X-Error-Description': ('The configured storage service is not ' 'responding to requests. Please contact ' 'your service provider.'), 'X-Error-Status': falcon.HTTP_503 } expected_body = { 'title': 'Storage service down', 'description': ('The configured storage service is not ' 'responding to requests. Please contact ' 'your service provider.'), 'code': 10042, } # Try it with Accept: */* headers['Accept'] = '*/*' response = client.simulate_request(path='/fail', headers=headers) assert response.status == headers['X-Error-Status'] assert response.headers['vary'] == 'Accept' assert expected_body == response.json # Now try it with application/json headers['Accept'] = 'application/json' response = client.simulate_request(path='/fail', headers=headers) assert response.status == headers['X-Error-Status'] assert response.json == expected_body def test_no_description_json(self, client): response = client.simulate_patch('/fail') assert response.status == falcon.HTTP_400 assert response.json == {'title': '400 Bad Request'} def test_no_description_xml(self, client): response = client.simulate_patch( path='/fail', headers={'Accept': 'application/xml'} ) assert response.status == falcon.HTTP_400 expected_xml = (b'' b'400 Bad Request') assert response.content == expected_xml def test_client_does_not_accept_json_or_xml(self, client): headers = { 'Accept': 'application/x-yaml', 'X-Error-Title': 'Storage service down', 'X-Error-Description': ('The configured storage service is not ' 'responding to requests. Please contact ' 'your service provider'), 'X-Error-Status': falcon.HTTP_503 } response = client.simulate_request(path='/fail', headers=headers) assert response.status == headers['X-Error-Status'] assert response.headers['Vary'] == 'Accept' assert not response.content def test_custom_old_error_serializer(self, client): headers = { 'X-Error-Title': 'Storage service down', 'X-Error-Description': ('The configured storage service is not ' 'responding to requests. Please contact ' 'your service provider'), 'X-Error-Status': falcon.HTTP_503 } expected_doc = { 'code': 10042, 'description': ('The configured storage service is not ' 'responding to requests. Please contact ' 'your service provider'), 'title': 'Storage service down' } def _my_serializer(req, exception): representation = None preferred = req.client_prefers(('application/x-yaml', 'application/json')) if preferred is not None: if preferred == 'application/json': representation = exception.to_json() else: representation = yaml.dump(exception.to_dict(), encoding=None) return (preferred, representation) def _check(media_type, deserializer): headers['Accept'] = media_type client.app.set_error_serializer(_my_serializer) response = client.simulate_request(path='/fail', headers=headers) assert response.status == headers['X-Error-Status'] actual_doc = deserializer(response.content.decode('utf-8')) assert expected_doc == actual_doc _check('application/x-yaml', yaml.load) _check('application/json', json.loads) def test_custom_old_error_serializer_no_body(self, client): headers = { 'X-Error-Status': falcon.HTTP_503 } def _my_serializer(req, exception): return None, None client.app.set_error_serializer(_my_serializer) client.simulate_request(path='/fail', headers=headers) def test_custom_new_error_serializer(self, client): headers = { 'X-Error-Title': 'Storage service down', 'X-Error-Description': ('The configured storage service is not ' 'responding to requests. Please contact ' 'your service provider'), 'X-Error-Status': falcon.HTTP_503 } expected_doc = { 'code': 10042, 'description': ('The configured storage service is not ' 'responding to requests. Please contact ' 'your service provider'), 'title': 'Storage service down' } def _my_serializer(req, resp, exception): representation = None preferred = req.client_prefers(('application/x-yaml', 'application/json')) if preferred is not None: if preferred == 'application/json': representation = exception.to_json() else: representation = yaml.dump(exception.to_dict(), encoding=None) resp.body = representation resp.content_type = preferred def _check(media_type, deserializer): headers['Accept'] = media_type client.app.set_error_serializer(_my_serializer) response = client.simulate_request(path='/fail', headers=headers) assert response.status == headers['X-Error-Status'] actual_doc = deserializer(response.content.decode('utf-8')) assert expected_doc == actual_doc _check('application/x-yaml', yaml.load) _check('application/json', json.loads) def test_client_does_not_accept_anything(self, client): headers = { 'Accept': '45087gigo;;;;', 'X-Error-Title': 'Storage service down', 'X-Error-Description': ('The configured storage service is not ' 'responding to requests. Please contact ' 'your service provider'), 'X-Error-Status': falcon.HTTP_503 } response = client.simulate_request(path='/fail', headers=headers) assert response.status == headers['X-Error-Status'] assert not response.content @pytest.mark.parametrize('media_type', [ 'application/json', 'application/vnd.company.system.project.resource+json;v=1.1', 'application/json-patch+json', ]) def test_forbidden(self, client, media_type): headers = {'Accept': media_type} expected_body = { 'title': 'Request denied', 'description': ('You do not have write permissions for this ' 'queue.'), 'link': { 'text': 'Documentation related to this error', 'href': 'http://example.com/api/rbac', 'rel': 'help', }, } response = client.simulate_post(path='/fail', headers=headers) assert response.status == falcon.HTTP_403 assert response.json == expected_body def test_epic_fail_json(self, client): headers = {'Accept': 'application/json'} expected_body = { 'title': 'Internet crashed', 'description': 'Catastrophic weather event due to climate change.', 'code': 8733224, 'link': { 'text': 'Drill baby drill!', 'href': 'http://example.com/api/climate', 'rel': 'help', }, } response = client.simulate_put('/fail', headers=headers) assert response.status == falcon.HTTP_792 assert response.json == expected_body @pytest.mark.parametrize('media_type', [ 'text/xml', 'application/xml', 'application/vnd.company.system.project.resource+xml;v=1.1', 'application/atom+xml', ]) def test_epic_fail_xml(self, client, media_type): headers = {'Accept': media_type} expected_body = ('' + '' + 'Internet crashed' + '' + 'Catastrophic weather event due to climate change.' + '' + '8733224' + '' + 'Drill baby drill!' + 'http://example.com/api/climate' + 'help' + '' + '') response = client.simulate_put(path='/fail', headers=headers) assert response.status == falcon.HTTP_792 try: et.fromstring(response.content.decode('utf-8')) except ValueError: pytest.fail() assert response.text == expected_body def test_unicode_json(self, client): unicode_resource = UnicodeFaultyResource() expected_body = { 'title': u'Internet \xe7rashed!', 'description': u'\xc7atastrophic weather event', 'link': { 'text': u'Drill b\xe1by drill!', 'href': 'http://example.com/api/%C3%A7limate', 'rel': 'help', }, } client.app.add_route('/unicode', unicode_resource) response = client.simulate_request(path='/unicode') assert unicode_resource.called assert response.status == falcon.HTTP_792 assert expected_body == response.json def test_unicode_xml(self, client): unicode_resource = UnicodeFaultyResource() expected_body = (u'' + u'' + u'Internet çrashed!' + u'' + u'Çatastrophic weather event' + u'' + u'' + u'Drill báby drill!' + u'http://example.com/api/%C3%A7limate' + u'help' + u'' + u'') client.app.add_route('/unicode', unicode_resource) response = client.simulate_request( path='/unicode', headers={'accept': 'application/xml'} ) assert unicode_resource.called assert response.status == falcon.HTTP_792 assert expected_body == response.text def test_401(self, client): client.app.add_route('/401', UnauthorizedResource()) response = client.simulate_request(path='/401') assert response.status == falcon.HTTP_401 assert response.headers['www-authenticate'] == 'Basic realm="simple"' response = client.simulate_post('/401') assert response.status == falcon.HTTP_401 assert response.headers['www-authenticate'] == 'Newauth realm="apps", Basic realm="simple"' response = client.simulate_put('/401') assert response.status == falcon.HTTP_401 assert 'www-authenticate' not in response.headers def test_404_without_body(self, client): client.app.add_route('/404', NotFoundResource()) response = client.simulate_request(path='/404') assert response.status == falcon.HTTP_404 assert not response.content def test_404_with_body(self, client): client.app.add_route('/404', NotFoundResourceWithBody()) response = client.simulate_request(path='/404') assert response.status == falcon.HTTP_404 assert response.content expected_body = { u'title': u'404 Not Found', u'description': u'Not Found' } assert response.json == expected_body def test_405_without_body(self, client): client.app.add_route('/405', MethodNotAllowedResource()) response = client.simulate_request(path='/405') assert response.status == falcon.HTTP_405 assert not response.content assert response.headers['allow'] == 'PUT' def test_405_without_body_with_extra_headers(self, client): client.app.add_route('/405', MethodNotAllowedResourceWithHeaders()) response = client.simulate_request(path='/405') assert response.status == falcon.HTTP_405 assert not response.content assert response.headers['allow'] == 'PUT' assert response.headers['x-ping'] == 'pong' def test_405_without_body_with_extra_headers_double_check(self, client): client.app.add_route( '/405/', MethodNotAllowedResourceWithHeadersWithAccept() ) response = client.simulate_request(path='/405') assert response.status == falcon.HTTP_405 assert not response.content assert response.headers['allow'] == 'PUT' assert response.headers['allow'] != 'GET,PUT' assert response.headers['allow'] != 'GET' assert response.headers['x-ping'] == 'pong' def test_405_with_body(self, client): client.app.add_route('/405', MethodNotAllowedResourceWithBody()) response = client.simulate_request(path='/405') assert response.status == falcon.HTTP_405 assert response.content expected_body = { u'title': u'405 Method Not Allowed', u'description': u'Not Allowed' } assert response.json == expected_body assert response.headers['allow'] == 'PUT' def test_410_without_body(self, client): client.app.add_route('/410', GoneResource()) response = client.simulate_request(path='/410') assert response.status == falcon.HTTP_410 assert not response.content def test_410_with_body(self, client): client.app.add_route('/410', GoneResourceWithBody()) response = client.simulate_request(path='/410') assert response.status == falcon.HTTP_410 assert response.content expected_body = { u'title': u'410 Gone', u'description': u'Gone with the wind' } assert response.json == expected_body def test_411(self, client): client.app.add_route('/411', LengthRequiredResource()) response = client.simulate_request(path='/411') assert response.status == falcon.HTTP_411 parsed_body = response.json assert parsed_body['title'] == 'title' assert parsed_body['description'] == 'description' def test_413(self, client): client.app.add_route('/413', RequestEntityTooLongResource()) response = client.simulate_request(path='/413') assert response.status == falcon.HTTP_413 parsed_body = response.json assert parsed_body['title'] == 'Request Rejected' assert parsed_body['description'] == 'Request Body Too Large' assert 'retry-after' not in response.headers def test_temporary_413_integer_retry_after(self, client): client.app.add_route('/413', TemporaryRequestEntityTooLongResource('6')) response = client.simulate_request(path='/413') assert response.status == falcon.HTTP_413 parsed_body = response.json assert parsed_body['title'] == 'Request Rejected' assert parsed_body['description'] == 'Request Body Too Large' assert response.headers['retry-after'] == '6' def test_temporary_413_datetime_retry_after(self, client): date = datetime.datetime.now() + datetime.timedelta(minutes=5) client.app.add_route( '/413', TemporaryRequestEntityTooLongResource(date) ) response = client.simulate_request(path='/413') assert response.status == falcon.HTTP_413 parsed_body = response.json assert parsed_body['title'] == 'Request Rejected' assert parsed_body['description'] == 'Request Body Too Large' assert response.headers['retry-after'] == falcon.util.dt_to_http(date) def test_414(self, client): client.app.add_route('/414', UriTooLongResource()) response = client.simulate_request(path='/414') assert response.status == falcon.HTTP_414 def test_414_with_title(self, client): title = 'Argh! Error!' client.app.add_route('/414', UriTooLongResource(title=title)) response = client.simulate_request(path='/414', headers={}) parsed_body = json.loads(response.content.decode()) assert parsed_body['title'] == title def test_414_with_description(self, client): description = 'Be short please.' client.app.add_route('/414', UriTooLongResource(description=description)) response = client.simulate_request(path='/414', headers={}) parsed_body = json.loads(response.content.decode()) assert parsed_body['description'] == description def test_414_with_custom_kwargs(self, client): code = 'someid' client.app.add_route('/414', UriTooLongResource(code=code)) response = client.simulate_request(path='/414', headers={}) parsed_body = json.loads(response.content.decode()) assert parsed_body['code'] == code def test_416(self, client): client.app = falcon.API() client.app.add_route('/416', RangeNotSatisfiableResource()) response = client.simulate_request(path='/416', headers={'accept': 'text/xml'}) assert response.status == falcon.HTTP_416 assert not response.content assert response.headers['content-range'] == 'bytes */123456' assert response.headers['content-length'] == '0' def test_429_no_retry_after(self, client): client.app.add_route('/429', TooManyRequestsResource()) response = client.simulate_request(path='/429') parsed_body = response.json assert response.status == falcon.HTTP_429 assert parsed_body['title'] == 'Too many requests' assert parsed_body['description'] == '1 per minute' assert 'retry-after' not in response.headers def test_429(self, client): client.app.add_route('/429', TooManyRequestsResource(60)) response = client.simulate_request(path='/429') parsed_body = response.json assert response.status == falcon.HTTP_429 assert parsed_body['title'] == 'Too many requests' assert parsed_body['description'] == '1 per minute' assert response.headers['retry-after'] == '60' def test_429_datetime(self, client): date = datetime.datetime.now() + datetime.timedelta(minutes=1) client.app.add_route('/429', TooManyRequestsResource(date)) response = client.simulate_request(path='/429') parsed_body = response.json assert response.status == falcon.HTTP_429 assert parsed_body['title'] == 'Too many requests' assert parsed_body['description'] == '1 per minute' assert response.headers['retry-after'] == falcon.util.dt_to_http(date) def test_503_integer_retry_after(self, client): client.app.add_route('/503', ServiceUnavailableResource(60)) response = client.simulate_request(path='/503') expected_body = { u'title': u'Oops', u'description': u'Stand by...', } assert response.status == falcon.HTTP_503 assert response.json == expected_body assert response.headers['retry-after'] == '60' def test_503_datetime_retry_after(self, client): date = datetime.datetime.now() + datetime.timedelta(minutes=5) client.app.add_route('/503', ServiceUnavailableResource(date)) response = client.simulate_request(path='/503') expected_body = { u'title': u'Oops', u'description': u'Stand by...', } assert response.status == falcon.HTTP_503 assert response.json == expected_body assert response.headers['retry-after'] == falcon.util.dt_to_http(date) def test_invalid_header(self, client): client.app.add_route('/400', InvalidHeaderResource()) response = client.simulate_request(path='/400') expected_desc = (u'The value provided for the X-Auth-Token ' u'header is invalid. Please provide a valid token.') expected_body = { u'title': u'Invalid header value', u'description': expected_desc, u'code': u'A1001', } assert response.status == falcon.HTTP_400 assert response.json == expected_body def test_missing_header(self, client): client.app.add_route('/400', MissingHeaderResource()) response = client.simulate_request(path='/400') expected_body = { u'title': u'Missing header value', u'description': u'The X-Auth-Token header is required.', } assert response.status == falcon.HTTP_400 assert response.json == expected_body def test_invalid_param(self, client): client.app.add_route('/400', InvalidParamResource()) response = client.simulate_request(path='/400') expected_desc = (u'The "id" parameter is invalid. The ' u'value must be a hex-encoded UUID.') expected_body = { u'title': u'Invalid parameter', u'description': expected_desc, u'code': u'P1002', } assert response.status == falcon.HTTP_400 assert response.json == expected_body def test_missing_param(self, client): client.app.add_route('/400', MissingParamResource()) response = client.simulate_request(path='/400') expected_body = { u'title': u'Missing parameter', u'description': u'The "id" parameter is required.', u'code': u'P1003', } assert response.status == falcon.HTTP_400 assert response.json == expected_body def test_misc(self, client): self._misc_test(client, falcon.HTTPBadRequest, falcon.HTTP_400) self._misc_test(client, falcon.HTTPNotAcceptable, falcon.HTTP_406, needs_title=False) self._misc_test(client, falcon.HTTPConflict, falcon.HTTP_409) self._misc_test(client, falcon.HTTPPreconditionFailed, falcon.HTTP_412) self._misc_test(client, falcon.HTTPUnsupportedMediaType, falcon.HTTP_415, needs_title=False) self._misc_test(client, falcon.HTTPUnprocessableEntity, falcon.HTTP_422) self._misc_test(client, falcon.HTTPUnavailableForLegalReasons, falcon.HTTP_451, needs_title=False) self._misc_test(client, falcon.HTTPInternalServerError, falcon.HTTP_500) self._misc_test(client, falcon.HTTPBadGateway, falcon.HTTP_502) def test_title_default_message_if_none(self, client): headers = { 'X-Error-Status': falcon.HTTP_503 } response = client.simulate_request(path='/fail', headers=headers) assert response.status == headers['X-Error-Status'] assert response.json['title'] == headers['X-Error-Status']