import json
import pytest
from webob import exc as webob_exc
from webob.dec import wsgify
from webob.request import Request
@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:
def __str__(self):
return "42"
duo = DummyUnicodeObject()
assert webob_exc.no_escape(duo) == "42"
def test_noescape_bytes():
assert webob_exc.no_escape(b"test") == "test"
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()
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()
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()
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():
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:
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
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