from webob.request import Request
from webob.dec import wsgify
from webob.exc import no_escape
from webob.exc import strip_tags
from webob.exc import HTTPException
from webob.exc import WSGIHTTPException
from webob.exc import _HTTPMove
from webob.exc import HTTPMethodNotAllowed
from webob.exc import HTTPExceptionMiddleware
from webob.exc import status_map
from nose.tools import eq_, ok_, assert_equal, assert_raises
@wsgify
def method_not_allowed_app(req):
if req.method != 'GET':
raise HTTPMethodNotAllowed()
return 'hello!'
def test_noescape_null():
assert_equal(no_escape(None), '')
def test_noescape_not_basestring():
assert_equal(no_escape(42), '42')
def test_noescape_unicode():
class DummyUnicodeObject(object):
def __unicode__(self):
return '42'
duo = DummyUnicodeObject()
assert_equal(no_escape(duo), '42')
def test_strip_tags_empty():
assert_equal(strip_tags(''), '')
def test_strip_tags_newline_to_space():
assert_equal(strip_tags('a\nb'), 'a b')
def test_strip_tags_zaps_carriage_return():
assert_equal(strip_tags('a\rb'), 'ab')
def test_strip_tags_br_to_newline():
assert_equal(strip_tags('a
b'), 'a\nb')
def test_strip_tags_zaps_comments():
assert_equal(strip_tags('a'), 'ab')
def test_strip_tags_zaps_tags():
assert_equal(strip_tags('foobaz'), 'foobaz')
def test_HTTPException():
import warnings
_called = []
_result = object()
def _response(environ, start_response):
_called.append((environ, start_response))
return _result
environ = {}
start_response = object()
exc = HTTPException('testing', _response)
ok_(exc.wsgi_response is _response)
result = exc(environ, start_response)
ok_(result is result)
assert_equal(_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 = 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 = WSGIHTTPException(body_template = TEMPLATE)
assert_equal(exc.body_template, TEMPLATE)
ok_(isinstance(exc.body_template_obj, Template))
eq_(exc.body_template_obj.substitute({'foo': 'FOO', 'bar': 'BAR'}),
'FOO: BAR')
def test_WSGIHTTPException_w_empty_body():
class EmptyOnly(WSGIHTTPException):
empty_body = True
exc = EmptyOnly(content_type='text/plain', content_length=234)
ok_('content_type' not in exc.__dict__)
ok_('content_length' not in exc.__dict__)
def test_WSGIHTTPException___str__():
exc1 = WSGIHTTPException(detail='Detail')
eq_(str(exc1), 'Detail')
class Explain(WSGIHTTPException):
explanation = 'Explanation'
eq_(str(Explain()), 'Explanation')
def test_WSGIHTTPException_plain_body_no_comment():
class Explain(WSGIHTTPException):
code = '999'
title = 'Testing'
explanation = 'Explanation'
exc = Explain(detail='Detail')
eq_(exc.plain_body({}),
'999 Testing\n\nExplanation\n\n Detail ')
def test_WSGIHTTPException_html_body_w_comment():
class Explain(WSGIHTTPException):
code = '999'
title = 'Testing'
explanation = 'Explanation'
exc = Explain(detail='Detail', comment='Comment')
eq_(exc.html_body({}),
'\n'
' \n'
' 999 Testing\n'
' \n'
' \n'
' 999 Testing
\n'
' Explanation
\n'
'Detail\n'
'\n\n'
' \n'
''
)
def test_WSGIHTTPException_generate_response():
def start_response(status, headers, exc_info=None):
pass
environ = {
'wsgi.url_scheme': 'HTTP',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'REQUEST_METHOD': 'PUT',
'HTTP_ACCEPT': 'text/html'
}
excep = WSGIHTTPException()
assert_equal( excep(environ,start_response), [
b'\n'
b' \n'
b' None None\n'
b' \n'
b' \n'
b' None None
\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 = WSGIHTTPException()
excep.body = b'test'
assert_equal( 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 = WSGIHTTPException()
assert_equal( 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 = WSGIHTTPException()
from webob import exc
exc.newstyle_exceptions = True
assert_equal( 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 = WSGIHTTPException()
from webob import exc
exc.newstyle_exceptions = False
assert_equal( 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_equal(resp.content_type, CONTENT_TYPE, description)
assert_equal(resp.content_length, len(HELLO_WORLD), description)
assert_equal(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 = 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 = _HTTPMove()
assert_equal( 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 = _HTTPMove(location='http://example.com')
assert_equal( m( environ, start_response ), [] )
def test_HTTPMove_add_slash_and_location():
def start_response(status, headers, exc_info=None):
pass
assert_raises( TypeError, _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 = _HTTPMove()
m.add_slash = True
assert_equal( 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 = _HTTPMove()
m.add_slash = True
environ[ 'QUERY_STRING' ] = 'querystring'
environ['PATH_INFO'] = '/'
assert_equal( m( environ, start_response ), [] )
def test_HTTPExceptionMiddleware_ok():
def app( environ, start_response ):
return '123'
application = app
m = HTTPExceptionMiddleware(application)
environ = {}
start_response = None
res = m( environ, start_response )
assert_equal( res, '123' )
def test_HTTPExceptionMiddleware_exception():
def wsgi_response( environ, start_response):
return '123'
def app( environ, start_response ):
raise HTTPException( None, wsgi_response )
application = app
m = HTTPExceptionMiddleware(application)
environ = {}
start_response = None
res = m( environ, start_response )
assert_equal( 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 HTTPException( None, wsgi_response )
application = app
m = HTTPExceptionMiddleware(application)
environ = {}
def start_response(status, headers, exc_info):
pass
try:
from webob import exc
old_sys = exc.sys
sys = DummySys()
res = m( environ, start_response )
assert_equal( res, None )
finally:
exc.sys = old_sys