# -*- coding: utf-8 -*-
"""
Tests for webapp2 webapp2.RequestHandler
"""
import os
import StringIO
import sys
import urllib
import webapp2
import test_base
class BareHandler(object):
def __init__(self, request, response):
self.response = response
response.write('I am not a RequestHandler but I work.')
def dispatch(self):
return self.response
class HomeHandler(webapp2.RequestHandler):
def get(self, **kwargs):
self.response.out.write('home sweet home')
def post(self, **kwargs):
self.response.out.write('home sweet home - POST')
class MethodsHandler(HomeHandler):
def put(self, **kwargs):
self.response.out.write('home sweet home - PUT')
def delete(self, **kwargs):
self.response.out.write('home sweet home - DELETE')
def head(self, **kwargs):
self.response.out.write('home sweet home - HEAD')
def trace(self, **kwargs):
self.response.out.write('home sweet home - TRACE')
def options(self, **kwargs):
self.response.out.write('home sweet home - OPTIONS')
class RedirectToHandler(webapp2.RequestHandler):
def get(self, **kwargs):
return self.redirect_to('route-test', _fragment='my-anchor', year='2010',
month='07', name='test', foo='bar')
class RedirectAbortHandler(webapp2.RequestHandler):
def get(self, **kwargs):
self.redirect('/somewhere', abort=True)
class BrokenHandler(webapp2.RequestHandler):
def get(self, **kwargs):
raise ValueError('booo!')
class BrokenButFixedHandler(BrokenHandler):
def handle_exception(self, exception, debug_mode):
# Let's fix it.
self.response.set_status(200)
self.response.out.write('that was close!')
def handle_404(request, response, exception):
response.out.write('404 custom handler')
response.set_status(404)
def handle_405(request, response, exception):
response.out.write('405 custom handler')
response.set_status(405, 'Custom Error Message')
response.headers['Allow'] = 'GET'
def handle_500(request, response, exception):
response.out.write('500 custom handler')
response.set_status(500)
class PositionalHandler(webapp2.RequestHandler):
def get(self, month, day, slug=None):
self.response.out.write('%s:%s:%s' % (month, day, slug))
class HandlerWithError(webapp2.RequestHandler):
def get(self, **kwargs):
self.response.out.write('bla bla bla bla bla bla')
self.error(403)
class InitializeHandler(webapp2.RequestHandler):
def __init__(self):
pass
def get(self):
self.response.out.write('Request method: %s' % self.request.method)
class WebDavHandler(webapp2.RequestHandler):
def version_control(self):
self.response.out.write('Method: VERSION-CONTROL')
def unlock(self):
self.response.out.write('Method: UNLOCK')
def propfind(self):
self.response.out.write('Method: PROPFIND')
class AuthorizationHandler(webapp2.RequestHandler):
def get(self):
self.response.out.write('nothing here')
class HandlerWithEscapedArg(webapp2.RequestHandler):
def get(self, name):
self.response.out.write(urllib.unquote_plus(name))
def get_redirect_url(handler, **kwargs):
return handler.uri_for('methods')
app = webapp2.WSGIApplication([
('/bare', BareHandler),
webapp2.Route('/', HomeHandler, name='home'),
webapp2.Route('/methods', MethodsHandler, name='methods'),
webapp2.Route('/broken', BrokenHandler),
webapp2.Route('/broken-but-fixed', BrokenButFixedHandler),
webapp2.Route('/<year:\d{4}>/<month:\d{1,2}>/<name>', None, name='route-test'),
webapp2.Route('/<:\d\d>/<:\d{2}>/<:\w+>', PositionalHandler, name='positional'),
webapp2.Route('/redirect-me', webapp2.RedirectHandler, defaults={'_uri': '/broken'}),
webapp2.Route('/redirect-me2', webapp2.RedirectHandler, defaults={'_uri': get_redirect_url}),
webapp2.Route('/redirect-me3', webapp2.RedirectHandler, defaults={'_uri': '/broken', '_permanent': False}),
webapp2.Route('/redirect-me4', webapp2.RedirectHandler, defaults={'_uri': get_redirect_url, '_permanent': False}),
webapp2.Route('/redirect-me5', RedirectToHandler),
webapp2.Route('/redirect-me6', RedirectAbortHandler),
webapp2.Route('/lazy', 'resources.handlers.LazyHandler'),
webapp2.Route('/error', HandlerWithError),
webapp2.Route('/initialize', InitializeHandler),
webapp2.Route('/webdav', WebDavHandler),
webapp2.Route('/authorization', AuthorizationHandler),
webapp2.Route('/escape/<name:.*>', HandlerWithEscapedArg, 'escape'),
], debug=False)
DEFAULT_RESPONSE = """Status: 404 Not Found
content-type: text/html; charset=utf8
Content-Length: 52
404 Not Found
The resource could not be found.
"""
class TestHandler(test_base.BaseTestCase):
def tearDown(self):
super(TestHandler, self).tearDown()
app.error_handlers = {}
def test_200(self):
rsp = app.get_response('/')
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'home sweet home')
def test_404(self):
req = webapp2.Request.blank('/nowhere')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 404)
def test_405(self):
req = webapp2.Request.blank('/')
req.method = 'PUT'
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 405)
self.assertEqual(rsp.headers.get('Allow'), 'GET, POST')
def test_500(self):
req = webapp2.Request.blank('/broken')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 500)
def test_500_but_fixed(self):
req = webapp2.Request.blank('/broken-but-fixed')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'that was close!')
def test_501(self):
# 501 Not Implemented
req = webapp2.Request.blank('/methods')
req.method = 'FOOBAR'
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 501)
def test_lazy_handler(self):
req = webapp2.Request.blank('/lazy')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'I am a laaazy view.')
def test_handler_with_error(self):
req = webapp2.Request.blank('/error')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 403)
self.assertEqual(rsp.body, '')
def test_debug_mode(self):
app = webapp2.WSGIApplication([
webapp2.Route('/broken', BrokenHandler),
], debug=True)
req = webapp2.Request.blank('/broken')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 500)
def test_custom_error_handlers(self):
app.error_handlers = {
404: handle_404,
405: handle_405,
500: handle_500,
}
req = webapp2.Request.blank('/nowhere')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 404)
self.assertEqual(rsp.body, '404 custom handler')
req = webapp2.Request.blank('/')
req.method = 'PUT'
rsp = req.get_response(app)
self.assertEqual(rsp.status, '405 Custom Error Message')
self.assertEqual(rsp.body, '405 custom handler')
self.assertEqual(rsp.headers.get('Allow'), 'GET')
req = webapp2.Request.blank('/broken')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 500)
self.assertEqual(rsp.body, '500 custom handler')
def test_methods(self):
app.debug = True
req = webapp2.Request.blank('/methods')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'home sweet home')
req = webapp2.Request.blank('/methods')
req.method = 'POST'
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'home sweet home - POST')
req = webapp2.Request.blank('/methods')
req.method = 'PUT'
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'home sweet home - PUT')
req = webapp2.Request.blank('/methods')
req.method = 'DELETE'
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'home sweet home - DELETE')
req = webapp2.Request.blank('/methods')
req.method = 'HEAD'
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, '')
req = webapp2.Request.blank('/methods')
req.method = 'OPTIONS'
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'home sweet home - OPTIONS')
req = webapp2.Request.blank('/methods')
req.method = 'TRACE'
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'home sweet home - TRACE')
app.debug = False
def test_positional(self):
req = webapp2.Request.blank('/07/31/test')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, '07:31:test')
req = webapp2.Request.blank('/10/18/wooohooo')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, '10:18:wooohooo')
def test_redirect(self):
req = webapp2.Request.blank('/redirect-me')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 301)
self.assertEqual(rsp.body, '')
self.assertEqual(rsp.headers['Location'], 'http://localhost/broken')
def test_redirect_with_callable(self):
req = webapp2.Request.blank('/redirect-me2')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 301)
self.assertEqual(rsp.body, '')
self.assertEqual(rsp.headers['Location'], 'http://localhost/methods')
def test_redirect_not_permanent(self):
req = webapp2.Request.blank('/redirect-me3')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 302)
self.assertEqual(rsp.body, '')
self.assertEqual(rsp.headers['Location'], 'http://localhost/broken')
def test_redirect_with_callable_not_permanent(self):
req = webapp2.Request.blank('/redirect-me4')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 302)
self.assertEqual(rsp.body, '')
self.assertEqual(rsp.headers['Location'], 'http://localhost/methods')
def test_redirect_to(self):
req = webapp2.Request.blank('/redirect-me5')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 302)
self.assertEqual(rsp.body, '')
self.assertEqual(rsp.headers['Location'], 'http://localhost/2010/07/test?foo=bar#my-anchor')
def test_redirect_abort(self):
req = webapp2.Request.blank('/redirect-me6')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 302)
self.assertEqual(rsp.body, """302 Moved Temporarily
The resource was found at http://localhost/somewhere; you should be redirected automatically. """)
self.assertEqual(rsp.headers['Location'], 'http://localhost/somewhere')
def test_run(self):
os.environ['REQUEST_METHOD'] = 'GET'
app.run()
#self.assertEqual(sys.stdout.read(), DEFAULT_RESPONSE)
def test_run_bare(self):
os.environ['REQUEST_METHOD'] = 'GET'
app.run(bare=True)
#self.assertEqual(sys.stdout.read(), DEFAULT_RESPONSE)
def test_run_debug(self):
debug = app.debug
app.debug = True
os.environ['REQUEST_METHOD'] = 'GET'
os.environ['PATH_INFO'] = '/'
res = app.run(bare=True)
#self.assertEqual(sys.stdout.read(), DEFAULT_RESPONSE)
app.debug = debug
'''
def test_get_valid_methods(self):
req = webapp2.Request.blank('http://localhost:80/')
req.app = app
app.set_globals(app=app, request=req)
handler = BrokenHandler(req, None)
handler.app = app
self.assertEqual(handler.get_valid_methods().sort(), ['GET'].sort())
handler = HomeHandler(req, None)
handler.app = app
self.assertEqual(handler.get_valid_methods().sort(),
['GET', 'POST'].sort())
handler = MethodsHandler(req, None)
handler.app = app
self.assertEqual(handler.get_valid_methods().sort(),
['GET', 'POST', 'HEAD', 'OPTIONS', 'PUT', 'DELETE', 'TRACE'].sort())
'''
def test_uri_for(self):
class Handler(webapp2.RequestHandler):
def get(self, *args, **kwargs):
pass
req = webapp2.Request.blank('http://localhost:80/')
req.route = webapp2.Route('')
req.route_args = tuple()
req.route_kwargs = {}
req.app = app
app.set_globals(app=app, request=req)
handler = Handler(req, webapp2.Response())
handler.app = app
for func in (handler.uri_for,):
self.assertEqual(func('home'), '/')
self.assertEqual(func('home', foo='bar'), '/?foo=bar')
self.assertEqual(func('home', _fragment='my-anchor', foo='bar'), '/?foo=bar#my-anchor')
self.assertEqual(func('home', _fragment='my-anchor'), '/#my-anchor')
self.assertEqual(func('home', _full=True), 'http://localhost:80/')
self.assertEqual(func('home', _full=True, _fragment='my-anchor'), 'http://localhost:80/#my-anchor')
self.assertEqual(func('home', _scheme='https'), 'https://localhost:80/')
self.assertEqual(func('home', _scheme='https', _full=False), 'https://localhost:80/')
self.assertEqual(func('home', _scheme='https', _fragment='my-anchor'), 'https://localhost:80/#my-anchor')
self.assertEqual(func('methods'), '/methods')
self.assertEqual(func('methods', foo='bar'), '/methods?foo=bar')
self.assertEqual(func('methods', _fragment='my-anchor', foo='bar'), '/methods?foo=bar#my-anchor')
self.assertEqual(func('methods', _fragment='my-anchor'), '/methods#my-anchor')
self.assertEqual(func('methods', _full=True), 'http://localhost:80/methods')
self.assertEqual(func('methods', _full=True, _fragment='my-anchor'), 'http://localhost:80/methods#my-anchor')
self.assertEqual(func('methods', _scheme='https'), 'https://localhost:80/methods')
self.assertEqual(func('methods', _scheme='https', _full=False), 'https://localhost:80/methods')
self.assertEqual(func('methods', _scheme='https', _fragment='my-anchor'), 'https://localhost:80/methods#my-anchor')
self.assertEqual(func('route-test', year='2010', month='0', name='test'), '/2010/0/test')
self.assertEqual(func('route-test', year='2010', month='07', name='test'), '/2010/07/test')
self.assertEqual(func('route-test', year='2010', month='07', name='test', foo='bar'), '/2010/07/test?foo=bar')
self.assertEqual(func('route-test', _fragment='my-anchor', year='2010', month='07', name='test', foo='bar'), '/2010/07/test?foo=bar#my-anchor')
self.assertEqual(func('route-test', _fragment='my-anchor', year='2010', month='07', name='test'), '/2010/07/test#my-anchor')
self.assertEqual(func('route-test', _full=True, year='2010', month='07', name='test'), 'http://localhost:80/2010/07/test')
self.assertEqual(func('route-test', _full=True, _fragment='my-anchor', year='2010', month='07', name='test'), 'http://localhost:80/2010/07/test#my-anchor')
self.assertEqual(func('route-test', _scheme='https', year='2010', month='07', name='test'), 'https://localhost:80/2010/07/test')
self.assertEqual(func('route-test', _scheme='https', _full=False, year='2010', month='07', name='test'), 'https://localhost:80/2010/07/test')
self.assertEqual(func('route-test', _scheme='https', _fragment='my-anchor', year='2010', month='07', name='test'), 'https://localhost:80/2010/07/test#my-anchor')
def test_extra_request_methods(self):
allowed_methods_backup = app.allowed_methods
webdav_methods = ('VERSION-CONTROL', 'UNLOCK', 'PROPFIND')
for method in webdav_methods:
# It is still not possible to use WebDav methods...
req = webapp2.Request.blank('/webdav')
req.method = method
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 501)
# Let's extend ALLOWED_METHODS with some WebDav methods.
app.allowed_methods = tuple(app.allowed_methods) + webdav_methods
#self.assertEqual(sorted(webapp2.get_valid_methods(WebDavHandler)), sorted(list(webdav_methods)))
# Now we can use WebDav methods...
for method in webdav_methods:
req = webapp2.Request.blank('/webdav')
req.method = method
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'Method: %s' % method)
# Restore initial values.
app.allowed_methods = allowed_methods_backup
self.assertEqual(len(app.allowed_methods), 7)
def test_escaping(self):
def get_req(uri):
req = webapp2.Request.blank(uri)
app.set_globals(app=app, request=req)
handler = webapp2.RequestHandler(req, None)
handler.app = req.app = app
return req, handler
req, handler = get_req('http://localhost:80/')
uri = webapp2.uri_for('escape', name='with space')
req, handler = get_req(uri)
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'with space')
req, handler = get_req('http://localhost:80/')
uri = webapp2.uri_for('escape', name='with+plus')
req, handler = get_req(uri)
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'with plus')
req, handler = get_req('http://localhost:80/')
uri = webapp2.uri_for('escape', name='with/slash')
req, handler = get_req(uri)
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'with/slash')
def test_handle_exception_with_error(self):
class HomeHandler(webapp2.RequestHandler):
def get(self, **kwargs):
raise TypeError()
def handle_exception(request, response, exception):
raise ValueError()
app = webapp2.WSGIApplication([
webapp2.Route('/', HomeHandler, name='home'),
], debug=False)
app.error_handlers[500] = handle_exception
req = webapp2.Request.blank('/')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 500)
def test_handle_exception_with_error_debug(self):
class HomeHandler(webapp2.RequestHandler):
def get(self, **kwargs):
raise TypeError()
def handle_exception(request, response, exception):
raise ValueError()
app = webapp2.WSGIApplication([
webapp2.Route('/', HomeHandler, name='home'),
], debug=True)
app.error_handlers[500] = handle_exception
req = webapp2.Request.blank('/')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 500)
def test_function_handler(self):
def my_view(request, *args, **kwargs):
return webapp2.Response('Hello, function world!')
def other_view(request, *args, **kwargs):
return webapp2.Response('Hello again, function world!')
'''
def one_more_view(request, response):
self.assertEqual(request.route_args, ())
self.assertEqual(request.route_kwargs, {'foo': 'bar'})
response.write('Hello you too, deprecated arguments world!')
'''
def one_more_view(request, *args, **kwargs):
self.assertEqual(args, ())
self.assertEqual(kwargs, {'foo': 'bar'})
return webapp2.Response('Hello you too!')
app = webapp2.WSGIApplication([
webapp2.Route('/', my_view),
webapp2.Route('/other', other_view),
webapp2.Route('/one-more/<foo>', one_more_view),
#webapp2.Route('/one-more/<foo>', one_more_view),
])
req = webapp2.Request.blank('/')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'Hello, function world!')
# Twice to test factory.
req = webapp2.Request.blank('/')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'Hello, function world!')
req = webapp2.Request.blank('/other')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'Hello again, function world!')
# Twice to test factory.
req = webapp2.Request.blank('/other')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'Hello again, function world!')
req = webapp2.Request.blank('/one-more/bar')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'Hello you too!')
def test_custom_method(self):
class MyHandler(webapp2.RequestHandler):
def my_method(self):
self.response.out.write('Hello, custom method world!')
def my_other_method(self):
self.response.out.write('Hello again, custom method world!')
app = webapp2.WSGIApplication([
webapp2.Route('/', MyHandler, handler_method='my_method'),
webapp2.Route('/other', MyHandler, handler_method='my_other_method'),
])
req = webapp2.Request.blank('/')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'Hello, custom method world!')
req = webapp2.Request.blank('/other')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'Hello again, custom method world!')
def test_custom_method_with_string(self):
app = webapp2.WSGIApplication([
webapp2.Route('/', handler='resources.handlers.CustomMethodHandler:custom_method'),
webapp2.Route('/bleh', handler='resources.handlers.CustomMethodHandler:custom_method'),
])
req = webapp2.Request.blank('/')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'I am a custom method.')
req = webapp2.Request.blank('/bleh')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'I am a custom method.')
self.assertRaises(ValueError, webapp2.Route, '/', handler='resources.handlers.CustomMethodHandler:custom_method', handler_method='custom_method')
def test_factory_1(self):
app.debug = True
rsp = app.get_response('/bare')
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'I am not a RequestHandler but I work.')
app.debug = False
def test_factory_2(self):
"""Very crazy stuff. Please ignore it."""
class MyHandler(object):
def __init__(self, request, response):
self.request = request
self.response = response
def __new__(cls, *args, **kwargs):
return cls.create_instance(*args, **kwargs)()
@classmethod
def create_instance(cls, *args, **kwargs):
obj = object.__new__(cls)
if isinstance(obj, cls):
obj.__init__(*args, **kwargs)
return obj
def __call__(self):
return self
def dispatch(self):
self.response.write('hello')
app = webapp2.WSGIApplication([
webapp2.Route('/', handler=MyHandler),
])
req = webapp2.Request.blank('/')
rsp = req.get_response(app)
self.assertEqual(rsp.status_int, 200)
self.assertEqual(rsp.body, 'hello')
def test_encoding(self):
class PostHandler(webapp2.RequestHandler):
def post(self):
foo = self.request.POST['foo']
if not foo:
foo = 'empty'
self.response.write(foo)
app = webapp2.WSGIApplication([
webapp2.Route('/', PostHandler),
], debug=True)
# foo with umlauts in the vowels.
value = 'f\xc3\xb6\xc3\xb6'
rsp = app.get_response('/', POST={'foo': value},
headers=[('Content-Type', 'application/x-www-form-urlencoded; charset=utf-8')])
self.assertEqual(rsp.body, 'föö')
rsp = app.get_response('/', POST={'foo': value},
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
self.assertEqual(rsp.body, 'föö')
if __name__ == '__main__':
test_base.main()