This patch updates several helper properties and methods on the Request object to raise HTTPBadRequest when the value of the field is not in the expected format. If it should be an int, then it must be an int. If Content-Range is not in a supported format, the request is denied. BREAKING CHANGE: Whereas previously parsing an incorrectly-formatted header or query parameter would result in a None response, attempting to parse such values now raises HTTPBadRequest instead. Fixes #99
200 lines
5.5 KiB
Python
200 lines
5.5 KiB
Python
from functools import wraps
|
|
import io
|
|
|
|
import six
|
|
from testtools.matchers import Contains
|
|
|
|
import falcon
|
|
import falcon.testing as testing
|
|
|
|
HTTP_METHODS = (
|
|
'CONNECT',
|
|
'DELETE',
|
|
'GET',
|
|
'HEAD',
|
|
'OPTIONS',
|
|
'POST',
|
|
'PUT',
|
|
'TRACE',
|
|
'PATCH'
|
|
)
|
|
|
|
|
|
class ThingsResource(object):
|
|
def __init__(self):
|
|
self.called = False
|
|
|
|
# Test non-callable attribute
|
|
self.on_patch = {}
|
|
|
|
# Field names ordered differently than in uri template
|
|
def on_get(self, req, resp, sid, id):
|
|
self.called = True
|
|
|
|
self.req, self.resp = req, resp
|
|
resp.status = falcon.HTTP_204
|
|
|
|
# Field names ordered the same as in uri template
|
|
def on_head(self, req, resp, id, sid):
|
|
self.called = True
|
|
|
|
self.req, self.resp = req, resp
|
|
resp.status = falcon.HTTP_204
|
|
|
|
def on_post(self, req, resp):
|
|
self.called = True
|
|
|
|
self.req, self.resp = req, resp
|
|
resp.status = falcon.HTTP_201
|
|
|
|
|
|
def capture(func):
|
|
@wraps(func)
|
|
def with_capture(*args, **kwargs):
|
|
self = args[0]
|
|
self.called = True
|
|
self.req, self.resp = args[1:]
|
|
func(*args, **kwargs)
|
|
|
|
return with_capture
|
|
|
|
|
|
def selfless_decorator(func):
|
|
def faulty(req, resp, foo, bar):
|
|
pass
|
|
|
|
return faulty
|
|
|
|
|
|
class MiscResource(object):
|
|
def __init__(self):
|
|
self.called = False
|
|
|
|
@capture
|
|
def on_get(self, req, resp):
|
|
resp.status = falcon.HTTP_204
|
|
|
|
@capture
|
|
def on_head(self, req, resp):
|
|
resp.status = falcon.HTTP_204
|
|
|
|
@capture
|
|
def on_put(self, req, resp):
|
|
resp.status = falcon.HTTP_400
|
|
|
|
@capture
|
|
def on_patch(self, req, resp):
|
|
pass
|
|
|
|
|
|
class GetWithFaultyPutResource(object):
|
|
def __init__(self):
|
|
self.called = False
|
|
|
|
@capture
|
|
def on_get(self, req, resp):
|
|
resp.status = falcon.HTTP_204
|
|
|
|
def on_put(self, req, resp, param):
|
|
raise TypeError()
|
|
|
|
|
|
class FaultyDecoratedResource(object):
|
|
|
|
@selfless_decorator
|
|
def on_get(self, req, resp):
|
|
pass
|
|
|
|
|
|
class TestHttpMethodRouting(testing.TestBase):
|
|
|
|
def before(self):
|
|
self.resource_things = ThingsResource()
|
|
self.api.add_route('/things', self.resource_things)
|
|
self.api.add_route('/things/{id}/stuff/{sid}', self.resource_things)
|
|
|
|
self.resource_misc = MiscResource()
|
|
self.api.add_route('/misc', self.resource_misc)
|
|
|
|
self.resource_get_with_faulty_put = GetWithFaultyPutResource()
|
|
self.api.add_route('/get_with_param/{param}',
|
|
self.resource_get_with_faulty_put)
|
|
|
|
def test_get(self):
|
|
self.simulate_request('/things/42/stuff/57')
|
|
self.assertEquals(self.srmock.status, falcon.HTTP_204)
|
|
self.assertTrue(self.resource_things.called)
|
|
|
|
def test_get_not_allowed(self):
|
|
self.simulate_request('/things')
|
|
self.assertEquals(self.srmock.status, falcon.HTTP_405)
|
|
self.assertFalse(self.resource_things.called)
|
|
|
|
def test_post(self):
|
|
self.simulate_request('/things', method='POST')
|
|
self.assertEquals(self.srmock.status, falcon.HTTP_201)
|
|
self.assertTrue(self.resource_things.called)
|
|
|
|
def test_post_not_allowed(self):
|
|
self.simulate_request('/things/42/stuff/57', method='POST')
|
|
self.assertEquals(self.srmock.status, falcon.HTTP_405)
|
|
self.assertFalse(self.resource_things.called)
|
|
|
|
def test_misc(self):
|
|
for method in ['GET', 'HEAD', 'PUT', 'PATCH']:
|
|
self.resource_misc.called = False
|
|
self.simulate_request('/misc', method=method)
|
|
self.assertTrue(self.resource_misc.called)
|
|
self.assertEquals(self.resource_misc.req.method, method)
|
|
|
|
def test_method_not_allowed(self):
|
|
for method in HTTP_METHODS:
|
|
if method in ('GET', 'POST', 'HEAD'):
|
|
continue
|
|
|
|
self.resource_things.called = False
|
|
self.simulate_request('/things/84/stuff/65', method=method)
|
|
|
|
self.assertFalse(self.resource_things.called)
|
|
self.assertEquals(self.srmock.status, falcon.HTTP_405)
|
|
|
|
headers = self.srmock.headers
|
|
allow_header = ('Allow', 'GET, HEAD, POST')
|
|
|
|
self.assertThat(headers, Contains(allow_header))
|
|
|
|
def test_method_not_allowed_with_param(self):
|
|
for method in HTTP_METHODS:
|
|
if method == 'GET' or method == 'PUT':
|
|
continue
|
|
|
|
self.resource_get_with_faulty_put.called = False
|
|
self.simulate_request(
|
|
'/get_with_param/bogus_param', method=method)
|
|
|
|
self.assertFalse(self.resource_get_with_faulty_put.called)
|
|
self.assertEquals(self.srmock.status, falcon.HTTP_405)
|
|
|
|
headers = self.srmock.headers
|
|
allow_header = ('Allow', 'GET, PUT')
|
|
|
|
self.assertThat(headers, Contains(allow_header))
|
|
|
|
def test_unexpected_type_error(self):
|
|
# Suppress logging
|
|
stream = io.StringIO() if six.PY3 else io.BytesIO()
|
|
self.simulate_request(
|
|
'/get_with_param/bogus_param', method='PUT', wsgierrors=stream)
|
|
|
|
self.assertEquals(self.srmock.status, falcon.HTTP_500)
|
|
|
|
def test_type_error(self):
|
|
self.api.add_route('/faulty', FaultyDecoratedResource())
|
|
self.simulate_request('/faulty')
|
|
self.assertEquals(self.srmock.status, falcon.HTTP_405)
|
|
|
|
def test_bogus_method(self):
|
|
self.simulate_request('/things', method=self.getUniqueString())
|
|
self.assertFalse(self.resource_things.called)
|
|
self.assertEquals(self.srmock.status, falcon.HTTP_400)
|