openstack/deb-python-falcon
Code Issues Proposed changes
Files
e1d92ebd896234df76a6d1a1327be7aad1310da2
deb-python-falcon/tests/test_req_vars.py
Kurt Griffiths e1d92ebd89 fix(Request): Improve support for content-length under wsgiref 
Under wsgiref, if no content-length header is provided, an empty
string is inserted. Update falcon.Request to handle this case,
as well as to more gracefully handle the case that the header is
set to any invalid value.
2014-10-29 14:23:37 -05:00

553 lines
21 KiB
Python
Raw Blame History

import datetime
import ddt
import falcon
from falcon.request import Request
import falcon.testing as testing
@ddt.ddt
class TestReqVars(testing.TestBase):
def before(self):
self.qs = 'marker=deadbeef&limit=10'
self.headers = {
'Content-Type': 'text/plain',
'Content-Length': '4829',
'Authorization': ''
}
self.app = '/test'
self.path = '/hello'
self.relative_uri = self.path + '?' + self.qs
self.req = Request(testing.create_environ(
app=self.app,
port=8080,
path='/hello',
query_string=self.qs,
headers=self.headers))
self.req_noqs = Request(testing.create_environ(
app=self.app,
path='/hello',
headers=self.headers))
def test_missing_qs(self):
env = testing.create_environ()
if 'QUERY_STRING' in env:
del env['QUERY_STRING']
# Should not cause an exception when Request is instantiated
Request(env)
def test_empty(self):
self.assertIs(self.req.auth, None)
def test_host(self):
self.assertEqual(self.req.host, testing.DEFAULT_HOST)
def test_subdomain(self):
req = Request(testing.create_environ(
host='com',
path='/hello',
headers=self.headers))
self.assertIs(req.subdomain, None)
req = Request(testing.create_environ(
host='example.com',
path='/hello',
headers=self.headers))
self.assertEqual(req.subdomain, 'example')
req = Request(testing.create_environ(
host='highwire.example.com',
path='/hello',
headers=self.headers))
self.assertEqual(req.subdomain, 'highwire')
req = Request(testing.create_environ(
host='lb01.dfw01.example.com',
port=8080,
path='/hello',
headers=self.headers))
self.assertEqual(req.subdomain, 'lb01')
# NOTE(kgriffs): Behavior for IP addresses is undefined,
# so just make sure it doesn't blow up.
req = Request(testing.create_environ(
host='127.0.0.1',
path='/hello',
headers=self.headers))
self.assertEqual(type(req.subdomain), str)
# NOTE(kgriffs): Test fallback to SERVER_NAME by using
# HTTP 1.0, which will cause .create_environ to not set
# HTTP_HOST.
req = Request(testing.create_environ(
protocol='HTTP/1.0',
host='example.com',
path='/hello',
headers=self.headers))
self.assertEqual(req.subdomain, 'example')
def test_reconstruct_url(self):
req = self.req
scheme = req.protocol
host = req.get_header('host')
app = req.app
path = req.path
query_string = req.query_string
expected_uri = ''.join([scheme, '://', host, app, path,
'?', query_string])
self.assertEqual(expected_uri, req.uri)
def test_uri(self):
uri = ('http://' + testing.DEFAULT_HOST + ':8080' +
self.app + self.relative_uri)
self.assertEqual(self.req.url, uri)
# NOTE(kgriffs): Call twice to check caching works
self.assertEqual(self.req.uri, uri)
self.assertEqual(self.req.uri, uri)
uri_noqs = ('http://' + testing.DEFAULT_HOST + self.app + self.path)
self.assertEqual(self.req_noqs.uri, uri_noqs)
def test_uri_http_1_0(self):
# =======================================================
# HTTP, 80
# =======================================================
req = Request(testing.create_environ(
protocol='HTTP/1.0',
app=self.app,
port=80,
path='/hello',
query_string=self.qs,
headers=self.headers))
uri = ('http://' + testing.DEFAULT_HOST +
self.app + self.relative_uri)
self.assertEqual(req.uri, uri)
# =======================================================
# HTTP, 80
# =======================================================
req = Request(testing.create_environ(
protocol='HTTP/1.0',
app=self.app,
port=8080,
path='/hello',
query_string=self.qs,
headers=self.headers))
uri = ('http://' + testing.DEFAULT_HOST + ':8080' +
self.app + self.relative_uri)
self.assertEqual(req.uri, uri)
# =======================================================
# HTTP, 80
# =======================================================
req = Request(testing.create_environ(
protocol='HTTP/1.0',
scheme='https',
app=self.app,
port=443,
path='/hello',
query_string=self.qs,
headers=self.headers))
uri = ('https://' + testing.DEFAULT_HOST +
self.app + self.relative_uri)
self.assertEqual(req.uri, uri)
# =======================================================
# HTTP, 80
# =======================================================
req = Request(testing.create_environ(
protocol='HTTP/1.0',
scheme='https',
app=self.app,
port=22,
path='/hello',
query_string=self.qs,
headers=self.headers))
uri = ('https://' + testing.DEFAULT_HOST + ':22' +
self.app + self.relative_uri)
self.assertEqual(req.uri, uri)
def test_relative_uri(self):
self.assertEqual(self.req.relative_uri, self.app + self.relative_uri)
self.assertEqual(
self.req_noqs.relative_uri, self.app + self.path)
req_noapp = Request(testing.create_environ(
path='/hello',
query_string=self.qs,
headers=self.headers))
self.assertEqual(req_noapp.relative_uri, self.relative_uri)
req_noapp = Request(testing.create_environ(
path='/hello/',
query_string=self.qs,
headers=self.headers))
# NOTE(kgriffs): Call twice to check caching works
self.assertEqual(req_noapp.relative_uri, self.relative_uri)
self.assertEqual(req_noapp.relative_uri, self.relative_uri)
def test_client_accepts(self):
headers = {'Accept': 'application/xml'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('application/xml'))
headers = {'Accept': '*/*'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('application/xml'))
self.assertTrue(req.client_accepts('application/json'))
self.assertTrue(req.client_accepts('application/x-msgpack'))
headers = {'Accept': 'application/x-msgpack'}
req = Request(testing.create_environ(headers=headers))
self.assertFalse(req.client_accepts('application/xml'))
self.assertFalse(req.client_accepts('application/json'))
self.assertTrue(req.client_accepts('application/x-msgpack'))
headers = {} # NOTE(kgriffs): Equivalent to '*/*' per RFC
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('application/xml'))
headers = {'Accept': 'application/json'}
req = Request(testing.create_environ(headers=headers))
self.assertFalse(req.client_accepts('application/xml'))
headers = {'Accept': 'application/x-msgpack'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('application/x-msgpack'))
headers = {'Accept': 'application/xm'}
req = Request(testing.create_environ(headers=headers))
self.assertFalse(req.client_accepts('application/xml'))
headers = {'Accept': 'application/*'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('application/json'))
self.assertTrue(req.client_accepts('application/xml'))
self.assertTrue(req.client_accepts('application/x-msgpack'))
headers = {'Accept': 'text/*'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('text/plain'))
self.assertTrue(req.client_accepts('text/csv'))
self.assertFalse(req.client_accepts('application/xhtml+xml'))
headers = {'Accept': 'text/*, application/xhtml+xml; q=0.0'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('text/plain'))
self.assertTrue(req.client_accepts('text/csv'))
self.assertTrue(req.client_accepts('application/xhtml+xml'))
headers = {'Accept': 'text/*; q=0.1, application/xhtml+xml; q=0.5'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('text/plain'))
headers = {'Accept': 'text/*, application/*'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('text/plain'))
self.assertTrue(req.client_accepts('application/xml'))
self.assertTrue(req.client_accepts('application/json'))
self.assertTrue(req.client_accepts('application/x-msgpack'))
headers = {'Accept': 'text/*,application/*'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts('text/plain'))
self.assertTrue(req.client_accepts('application/xml'))
self.assertTrue(req.client_accepts('application/json'))
self.assertTrue(req.client_accepts('application/x-msgpack'))
def test_client_accepts_bogus(self):
headers = {'Accept': '~'}
req = Request(testing.create_environ(headers=headers))
self.assertFalse(req.client_accepts('text/plain'))
self.assertFalse(req.client_accepts('application/json'))
def test_client_accepts_props(self):
headers = {'Accept': 'application/xml'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts_xml)
self.assertFalse(req.client_accepts_json)
self.assertFalse(req.client_accepts_msgpack)
headers = {'Accept': 'application/*'}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts_xml)
self.assertTrue(req.client_accepts_json)
self.assertTrue(req.client_accepts_msgpack)
headers = {'Accept': 'application/json'}
req = Request(testing.create_environ(headers=headers))
self.assertFalse(req.client_accepts_xml)
self.assertTrue(req.client_accepts_json)
self.assertFalse(req.client_accepts_msgpack)
headers = {'Accept': 'application/x-msgpack'}
req = Request(testing.create_environ(headers=headers))
self.assertFalse(req.client_accepts_xml)
self.assertFalse(req.client_accepts_json)
self.assertTrue(req.client_accepts_msgpack)
headers = {
'Accept': 'application/json,application/xml,application/x-msgpack'
}
req = Request(testing.create_environ(headers=headers))
self.assertTrue(req.client_accepts_xml)
self.assertTrue(req.client_accepts_json)
self.assertTrue(req.client_accepts_msgpack)
def test_client_prefers(self):
headers = {'Accept': 'application/xml'}
req = Request(testing.create_environ(headers=headers))
preferred_type = req.client_prefers(['application/xml'])
self.assertEqual(preferred_type, 'application/xml')
headers = {'Accept': '*/*'}
preferred_type = req.client_prefers(('application/xml',
'application/json'))
# NOTE(kgriffs): If client doesn't care, "preferr" the first one
self.assertEqual(preferred_type, 'application/xml')
headers = {'Accept': 'text/*; q=0.1, application/xhtml+xml; q=0.5'}
req = Request(testing.create_environ(headers=headers))
preferred_type = req.client_prefers(['application/xhtml+xml'])
self.assertEqual(preferred_type, 'application/xhtml+xml')
headers = {'Accept': '3p12845j;;;asfd;'}
req = Request(testing.create_environ(headers=headers))
preferred_type = req.client_prefers(['application/xhtml+xml'])
self.assertEqual(preferred_type, None)
def test_range(self):
headers = {'Range': '10-'}
req = Request(testing.create_environ(headers=headers))
self.assertEqual(req.range, (10, -1))
headers = {'Range': '10-20'}
req = Request(testing.create_environ(headers=headers))
self.assertEqual(req.range, (10, 20))
headers = {'Range': '-10240'}
req = Request(testing.create_environ(headers=headers))
self.assertEqual(req.range, (-10240, -1))
headers = {'Range': ''}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPInvalidHeader, lambda: req.range)
req = Request(testing.create_environ())
self.assertIs(req.range, None)
def test_range_invalid(self):
headers = {'Range': '10240'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': '-'}
expected_desc = ('The value provided for the Range header is '
'invalid. The byte offsets are missing.')
self._test_error_details(headers, 'range',
falcon.HTTPInvalidHeader,
'Invalid header value', expected_desc)
headers = {'Range': '--'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': '-3-'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': '-3-4'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': '3-3-4'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': '3-3-'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': '3-3- '}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': 'fizbit'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': 'a-'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': 'a-3'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': '-b'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': '3-b'}
req = Request(testing.create_environ(headers=headers))
self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
headers = {'Range': 'x-y'}
expected_desc = ('The value provided for the Range header is '
'invalid. It must be a byte range formatted '
'according to RFC 2616.')
self._test_error_details(headers, 'range',
falcon.HTTPInvalidHeader,
'Invalid header value', expected_desc)
headers = {'Range': 'bytes=0-0,-1'}
expected_desc = ('The value provided for the Range '
'header is invalid. The value must be a '
'continuous byte range.')
self._test_error_details(headers, 'range',
falcon.HTTPInvalidHeader,
'Invalid header value', expected_desc)
def test_missing_attribute_header(self):
req = Request(testing.create_environ())
self.assertEqual(req.range, None)
req = Request(testing.create_environ())
self.assertEqual(req.content_length, None)
def test_content_length(self):
headers = {'content-length': '5656'}
req = Request(testing.create_environ(headers=headers))
self.assertEqual(req.content_length, 5656)
headers = {'content-length': ''}
req = Request(testing.create_environ(headers=headers))
self.assertEqual(req.content_length, None)
def test_bogus_content_length_nan(self):
headers = {'content-length': 'fuzzy-bunnies'}
expected_desc = ('The value provided for the '
'Content-Length header is invalid. The value '
'of the header must be a number.')
self._test_error_details(headers, 'content_length',
falcon.HTTPInvalidHeader,
'Invalid header value', expected_desc)
def test_bogus_content_length_neg(self):
headers = {'content-length': '-1'}
expected_desc = ('The value provided for the Content-Length '
'header is invalid. The value of the header '
'must be a positive number.')
self._test_error_details(headers, 'content_length',
falcon.HTTPInvalidHeader,
'Invalid header value', expected_desc)
def test_date(self):
date = datetime.datetime(2013, 4, 4, 5, 19, 18)
headers = {'date': 'Thu, 04 Apr 2013 05:19:18 GMT'}
req = Request(testing.create_environ(headers=headers))
self.assertEqual(req.date, date)
@ddt.data('Thu, 04 Apr 2013', '')
def test_date_invalid(self, http_date):
headers = {'date': http_date}
expected_desc = ('The value provided for the Date '
'header is invalid. It must be formatted '
'according to RFC 1123.')
self._test_error_details(headers, 'date',
falcon.HTTPInvalidHeader,
'Invalid header value', expected_desc)
def test_date_missing(self):
req = Request(testing.create_environ())
self.assertIs(req.date, None)
def test_attribute_headers(self):
date = testing.httpnow()
hash = 'fa0d1a60ef6616bb28038515c8ea4cb2'
auth = 'HMAC_SHA1 c590afa9bb59191ffab30f223791e82d3fd3e3af'
agent = 'testing/1.0.1'
default_agent = 'curl/7.24.0 (x86_64-apple-darwin12.0)'
self._test_attribute_header('Accept', 'x-falcon', 'accept',
default='*/*')
self._test_attribute_header('Authorization', auth, 'auth')
self._test_attribute_header('Content-Type', 'text/plain',
'content_type')
self._test_attribute_header('Expect', '100-continue', 'expect')
self._test_attribute_header('If-Match', hash, 'if_match')
self._test_attribute_header('If-Modified-Since', date,
'if_modified_since')
self._test_attribute_header('If-None-Match', hash, 'if_none_match')
self._test_attribute_header('If-Range', hash, 'if_range')
self._test_attribute_header('If-Unmodified-Since', date,
'if_unmodified_since')
self._test_attribute_header('User-Agent', agent, 'user_agent',
default=default_agent)
def test_method(self):
self.assertEqual(self.req.method, 'GET')
self.req = Request(testing.create_environ(path='', method='HEAD'))
self.assertEqual(self.req.method, 'HEAD')
def test_empty_path(self):
self.req = Request(testing.create_environ(path=''))
self.assertEqual(self.req.path, '/')
def test_content_type_method(self):
self.assertEqual(self.req.get_header('content-type'), 'text/plain')
def test_content_length_method(self):
self.assertEqual(self.req.get_header('content-length'), '4829')
# -------------------------------------------------------------------------
# Helpers
# -------------------------------------------------------------------------
def _test_attribute_header(self, name, value, attr, default=None):
headers = {name: value}
req = Request(testing.create_environ(headers=headers))
self.assertEqual(getattr(req, attr), value)
req = Request(testing.create_environ())
self.assertEqual(getattr(req, attr), default)
def _test_error_details(self, headers, attr_name,
error_type, title, description):
req = Request(testing.create_environ(headers=headers))
try:
getattr(req, attr_name)
self.fail('{0} not raised'.format(error_type.__name__))
except error_type as ex:
self.assertEqual(ex.title, title)
self.assertEqual(ex.description, description)
View Git Blame Copy Permalink