import pytest
import json
from webob.request import Request
from webob.dec import wsgify
from webob import exc as webob_exc
@wsgify
def method_not_allowed_app(req):
if req.method != 'GET':
raise webob_exc.HTTPMethodNotAllowed()
return 'hello!'
def test_noescape_null():
assert webob_exc.no_escape(None) == ''
def test_noescape_not_basestring():
assert webob_exc.no_escape(42) == '42'
def test_noescape_unicode():
class DummyUnicodeObject(object):
def __unicode__(self):
return '42'
duo = DummyUnicodeObject()
assert webob_exc.no_escape(duo) == '42'
def test_strip_tags_empty():
assert webob_exc.strip_tags('') == ''
def test_strip_tags_newline_to_space():
assert webob_exc.strip_tags('a\nb') == 'a b'
def test_strip_tags_zaps_carriage_return():
assert webob_exc.strip_tags('a\rb') == 'ab'
def test_strip_tags_br_to_newline():
assert webob_exc.strip_tags('a
b') == 'a\nb'
def test_strip_tags_zaps_comments():
assert webob_exc.strip_tags('a') == 'ab'
def test_strip_tags_zaps_tags():
assert webob_exc.strip_tags('foobaz') == 'foobaz'
def test_HTTPException():
_called = []
_result = object()
def _response(environ, start_response):
_called.append((environ, start_response))
return _result
environ = {}
start_response = object()
exc = webob_exc.HTTPException('testing', _response)
assert exc.wsgi_response is _response
result = exc(environ, start_response)
assert result is result
assert _called == [(environ, start_response)]
def test_exception_with_unicode_data():
req = Request.blank('/', method='POST')
res = req.get_response(method_not_allowed_app)
assert res.status_code == 405
def test_WSGIHTTPException_headers():
exc = webob_exc.WSGIHTTPException(headers=[('Set-Cookie', 'a=1'),
('Set-Cookie', 'a=2')])
mixed = exc.headers.mixed()
assert mixed['set-cookie'] == ['a=1', 'a=2']
def test_WSGIHTTPException_w_body_template():
from string import Template
TEMPLATE = '$foo: $bar'
exc = webob_exc.WSGIHTTPException(body_template=TEMPLATE)
assert exc.body_template == TEMPLATE
assert isinstance(exc.body_template_obj, Template)
assert exc.body_template_obj.substitute({'foo': 'FOO', 'bar': 'BAR'}) == 'FOO: BAR'
def test_WSGIHTTPException_w_empty_body():
class EmptyOnly(webob_exc.WSGIHTTPException):
empty_body = True
exc = EmptyOnly(content_type='text/plain', content_length=234)
assert 'content_type' not in exc.__dict__
assert 'content_length' not in exc.__dict__
def test_WSGIHTTPException___str__():
exc1 = webob_exc.WSGIHTTPException(detail='Detail')
assert str(exc1) == 'Detail'
class Explain(webob_exc.WSGIHTTPException):
explanation = 'Explanation'
assert str(Explain()) == 'Explanation'
def test_WSGIHTTPException_plain_body_no_comment():
class Explain(webob_exc.WSGIHTTPException):
code = '999'
title = 'Testing'
explanation = 'Explanation'
exc = Explain(detail='Detail')
assert exc.plain_body({}) == '999 Testing\n\nExplanation\n\n Detail '
def test_WSGIHTTPException_html_body_w_comment():
class Explain(webob_exc.WSGIHTTPException):
code = '999'
title = 'Testing'
explanation = 'Explanation'
exc = Explain(detail='Detail', comment='Comment')
assert exc.html_body({}) == (
'\n'
' \n'
' 999 Testing\n'
' \n'
' \n'
' 999 Testing
\n'
' Explanation
\n'
'Detail\n'
'\n\n'
' \n'
''
)
def test_WSGIHTTPException_json_body_no_comment():
class ValidationError(webob_exc.WSGIHTTPException):
code = '422'
title = 'Validation Failed'
explanation = 'Validation of an attribute failed.'
exc = ValidationError(detail='Attribute "xyz" is invalid.')
body = exc.json_body({})
assert json.loads(body) == {
"code": "422 Validation Failed",
"title": "Validation Failed",
"message": "Validation of an attribute failed.
\nAttribute"
' "xyz" is invalid.\n\n',
}
def test_WSGIHTTPException_respects_application_json():
class ValidationError(webob_exc.WSGIHTTPException):
code = '422'
title = 'Validation Failed'
explanation = 'Validation of an attribute failed.'
def start_response(status, headers, exc_info=None):
# check that json doesn't contain a charset
assert ('Content-Type', 'application/json') in headers
pass
exc = ValidationError(detail='Attribute "xyz" is invalid.')
resp = exc.generate_response(environ={
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'PUT',
'HTTP_ACCEPT': 'application/json',
}, start_response=start_response)
assert json.loads(resp[0].decode('utf-8')) == {
"code": "422 Validation Failed",
"title": "Validation Failed",
"message": "Validation of an attribute failed.
\nAttribute"
' "xyz" is invalid.\n\n',
}
def test_WSGIHTTPException_respects_accept_text_html():
def start_response(status, headers, exc_info=None):
for header in headers:
if header[0] == 'Content-Type':
assert header[1].startswith('text/html')
exc = webob_exc.WSGIHTTPException()
resp = exc.generate_response(environ={
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'GET',
'HTTP_ACCEPT': 'text/html',
}, start_response=start_response)
def test_WSGIHTTPException_respects_accept_text_plain():
def start_response(status, headers, exc_info=None):
for header in headers:
if header[0] == 'Content-Type':
assert header[1].startswith('text/plain')
exc = webob_exc.WSGIHTTPException()
resp = exc.generate_response(environ={
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'GET',
'HTTP_ACCEPT': 'text/plain',
}, start_response=start_response)
def test_WSGIHTTPException_respects_accept_star_star():
def start_response(status, headers, exc_info=None):
for header in headers:
if header[0] == 'Content-Type':
assert header[1].startswith('text/html')
exc = webob_exc.WSGIHTTPException()
resp = exc.generate_response(environ={
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'GET',
'HTTP_ACCEPT': '*/*',
}, start_response=start_response)
def test_WSGIHTTPException_allows_custom_json_formatter():
def json_formatter(body, status, title, environ):
return {"fake": True}
class ValidationError(webob_exc.WSGIHTTPException):
code = '422'
title = 'Validation Failed'
explanation = 'Validation of an attribute failed.'
exc = ValidationError(detail='Attribute "xyz" is invalid.',
json_formatter=json_formatter)
body = exc.json_body({})
assert json.loads(body) == {"fake": True}
def test_WSGIHTTPException_generate_response():
def start_response(status, headers, exc_info=None):
assert ('Content-Type', 'text/html; charset=UTF-8') in headers
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'PUT',
'HTTP_ACCEPT': 'text/html'
}
excep = webob_exc.WSGIHTTPException()
assert excep(environ, start_response) == [
b'\n'
b' \n'
b' 500 Internal Server Error\n'
b' \n'
b' \n'
b' 500 Internal Server Error
\n'
b'
\n'
b'\n'
b'\n\n'
b' \n'
b'']
def test_WSGIHTTPException_call_w_body():
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'PUT'
}
excep = webob_exc.WSGIHTTPException()
excep.body = b'test'
assert excep(environ,start_response) == [b'test']
def test_WSGIHTTPException_wsgi_response():
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'HEAD'
}
excep = webob_exc.WSGIHTTPException()
assert excep.wsgi_response(environ,start_response) == []
def test_WSGIHTTPException_exception_newstyle():
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'HEAD'
}
excep = webob_exc.WSGIHTTPException()
webob_exc.newstyle_exceptions = True
assert excep(environ,start_response) == []
def test_WSGIHTTPException_exception_no_newstyle():
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'HEAD'
}
excep = webob_exc.WSGIHTTPException()
webob_exc.newstyle_exceptions = False
assert excep(environ,start_response) == []
def test_HTTPOk_head_of_proxied_head():
# first set up a response to a HEAD request
HELLO_WORLD = "Hi!\n"
CONTENT_TYPE = "application/hello"
def head_app(environ, start_response):
"""An application object that understands HEAD"""
status = '200 OK'
response_headers = [('Content-Type', CONTENT_TYPE),
('Content-Length', len(HELLO_WORLD))]
start_response(status, response_headers)
if environ['REQUEST_METHOD'] == 'HEAD':
return []
else:
return [HELLO_WORLD]
def verify_response(resp, description):
assert resp.content_type == CONTENT_TYPE, description
assert resp.content_length == len(HELLO_WORLD), description
assert resp.body == b'', description
req = Request.blank('/', method='HEAD')
resp1 = req.get_response(head_app)
verify_response(resp1, "first response")
# Copy the response like a proxy server would.
# Copying an empty body has set content_length
# so copy the headers only afterwards.
resp2 = webob_exc.status_map[resp1.status_int](request=req)
resp2.body = resp1.body
resp2.headerlist = resp1.headerlist
verify_response(resp2, "copied response")
# evaluate it again
resp3 = req.get_response(resp2)
verify_response(resp3, "evaluated copy")
def test_HTTPMove():
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'HEAD',
'PATH_INFO': '/',
}
m = webob_exc._HTTPMove()
assert m( environ, start_response ) == []
def test_HTTPMove_location_not_none():
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'HEAD',
'PATH_INFO': '/',
}
m = webob_exc._HTTPMove(location='http://example.com')
assert m( environ, start_response ) == []
def test_HTTPMove_location_newlines():
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'HEAD',
'PATH_INFO': '/',
}
with pytest.raises(ValueError):
webob_exc._HTTPMove(location='http://example.com\r\nX-Test: false')
def test_HTTPMove_add_slash_and_location():
def start_response(status, headers, exc_info=None):
pass
with pytest.raises(TypeError):
webob_exc._HTTPMove(location='http://example.com', add_slash=True)
def test_HTTPMove_call_add_slash():
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'HEAD',
'PATH_INFO': '/',
}
m = webob_exc._HTTPMove()
m.add_slash = True
assert m( environ, start_response ) == []
def test_HTTPMove_call_query_string():
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'HEAD'
}
m = webob_exc._HTTPMove()
m.add_slash = True
environ[ 'QUERY_STRING' ] = 'querystring'
environ['PATH_INFO'] = '/'
assert m( environ, start_response ) == []
def test_HTTPFound_unused_environ_variable():
class Crashy(object):
def __str__(self):
raise Exception('I crashed!')
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'GET',
'PATH_INFO': '/',
'HTTP_ACCEPT': 'text/html',
'crashy': Crashy()
}
m = webob_exc._HTTPMove(location='http://www.example.com')
assert m(environ, start_response) == [
b'\n'
b' \n'
b' 500 Internal Server Error\n'
b' \n'
b' \n'
b' 500 Internal Server Error
\n'
b' The resource has been moved to '
b''
b'http://www.example.com;\n'
b'you should be redirected automatically.\n'
b'\n\n'
b' \n'
b'']
def test_HTTPExceptionMiddleware_ok():
def app(environ, start_response):
return '123'
application = app
m = webob_exc.HTTPExceptionMiddleware(application)
environ = {}
start_response = None
res = m(environ, start_response)
assert res == '123'
def test_HTTPExceptionMiddleware_exception():
def wsgi_response(environ, start_response):
return '123'
def app(environ, start_response):
raise webob_exc.HTTPException(None, wsgi_response)
application = app
m = webob_exc.HTTPExceptionMiddleware(application)
environ = {}
start_response = None
res = m(environ, start_response)
assert res == '123'
def test_HTTPExceptionMiddleware_exception_exc_info_none():
class DummySys:
def exc_info(self):
return None
def wsgi_response(environ, start_response):
return start_response('200 OK', [], exc_info=None)
def app(environ, start_response):
raise webob_exc.HTTPException(None, wsgi_response)
application = app
m = webob_exc.HTTPExceptionMiddleware(application)
environ = {}
def start_response(status, headers, exc_info):
pass
try:
old_sys = webob_exc.sys
sys = DummySys()
res = m(environ, start_response)
assert res is None
finally:
webob_exc.sys = old_sys
def test_status_map_is_deterministic():
for code, cls in (
(200, webob_exc.HTTPOk),
(201, webob_exc.HTTPCreated),
(202, webob_exc.HTTPAccepted),
(203, webob_exc.HTTPNonAuthoritativeInformation),
(204, webob_exc.HTTPNoContent),
(205, webob_exc.HTTPResetContent),
(206, webob_exc.HTTPPartialContent),
(300, webob_exc.HTTPMultipleChoices),
(301, webob_exc.HTTPMovedPermanently),
(302, webob_exc.HTTPFound),
(303, webob_exc.HTTPSeeOther),
(304, webob_exc.HTTPNotModified),
(305, webob_exc.HTTPUseProxy),
(307, webob_exc.HTTPTemporaryRedirect),
(308, webob_exc.HTTPPermanentRedirect),
(400, webob_exc.HTTPBadRequest),
(401, webob_exc.HTTPUnauthorized),
(402, webob_exc.HTTPPaymentRequired),
(403, webob_exc.HTTPForbidden),
(404, webob_exc.HTTPNotFound),
(405, webob_exc.HTTPMethodNotAllowed),
(406, webob_exc.HTTPNotAcceptable),
(407, webob_exc.HTTPProxyAuthenticationRequired),
(408, webob_exc.HTTPRequestTimeout),
(409, webob_exc.HTTPConflict),
(410, webob_exc.HTTPGone),
(411, webob_exc.HTTPLengthRequired),
(412, webob_exc.HTTPPreconditionFailed),
(413, webob_exc.HTTPRequestEntityTooLarge),
(414, webob_exc.HTTPRequestURITooLong),
(415, webob_exc.HTTPUnsupportedMediaType),
(416, webob_exc.HTTPRequestRangeNotSatisfiable),
(417, webob_exc.HTTPExpectationFailed),
(422, webob_exc.HTTPUnprocessableEntity),
(423, webob_exc.HTTPLocked),
(424, webob_exc.HTTPFailedDependency),
(428, webob_exc.HTTPPreconditionRequired),
(429, webob_exc.HTTPTooManyRequests),
(431, webob_exc.HTTPRequestHeaderFieldsTooLarge),
(451, webob_exc.HTTPUnavailableForLegalReasons),
(500, webob_exc.HTTPInternalServerError),
(501, webob_exc.HTTPNotImplemented),
(502, webob_exc.HTTPBadGateway),
(503, webob_exc.HTTPServiceUnavailable),
(504, webob_exc.HTTPGatewayTimeout),
(505, webob_exc.HTTPVersionNotSupported),
(511, webob_exc.HTTPNetworkAuthenticationRequired),
):
assert webob_exc.status_map[code] == cls