650 lines
		
	
	
		
			25 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			650 lines
		
	
	
		
			25 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
import datetime
 | 
						|
import six
 | 
						|
import testtools
 | 
						|
 | 
						|
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)
 | 
						|
 | 
						|
    @testtools.skipUnless(six.PY3, 'Test only applies to Python 3')
 | 
						|
    def test_nonlatin_path(self):
 | 
						|
        cyrillic_path = u'/hello_\u043f\u0440\u0438\u0432\u0435\u0442'
 | 
						|
        cyrillic_path_decoded = cyrillic_path.encode('utf-8').decode('latin1')
 | 
						|
        req = Request(testing.create_environ(
 | 
						|
            host='com',
 | 
						|
            path=cyrillic_path_decoded,
 | 
						|
            headers=self.headers))
 | 
						|
        self.assertEqual(req.path, cyrillic_path)
 | 
						|
 | 
						|
    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_https(self):
 | 
						|
        # =======================================================
 | 
						|
        # Default port, implicit
 | 
						|
        # =======================================================
 | 
						|
        req = Request(testing.create_environ(
 | 
						|
            path='/hello', scheme='https'))
 | 
						|
        uri = ('https://' + testing.DEFAULT_HOST + '/hello')
 | 
						|
 | 
						|
        self.assertEqual(req.uri, uri)
 | 
						|
 | 
						|
        # =======================================================
 | 
						|
        # Default port, explicit
 | 
						|
        # =======================================================
 | 
						|
        req = Request(testing.create_environ(
 | 
						|
            path='/hello', scheme='https', port=443))
 | 
						|
        uri = ('https://' + testing.DEFAULT_HOST + '/hello')
 | 
						|
 | 
						|
        self.assertEqual(req.uri, uri)
 | 
						|
 | 
						|
        # =======================================================
 | 
						|
        # Non-default port
 | 
						|
        # =======================================================
 | 
						|
        req = Request(testing.create_environ(
 | 
						|
            path='/hello', scheme='https', port=22))
 | 
						|
        uri = ('https://' + testing.DEFAULT_HOST + ':22/hello')
 | 
						|
 | 
						|
        self.assertEqual(req.uri, uri)
 | 
						|
 | 
						|
    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.assertFalse(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'))
 | 
						|
        self.assertTrue(req.client_accepts('application/xhtml+xml'))
 | 
						|
 | 
						|
        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/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, "prefer" 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': 'bytes=10-'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertEqual(req.range, (10, -1))
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=10-20'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertEqual(req.range, (10, 20))
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=-10240'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertEqual(req.range, (-10240, -1))
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=0-2'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertEqual(req.range, (0, 2))
 | 
						|
 | 
						|
        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_unit(self):
 | 
						|
        headers = {'Range': 'bytes=10-'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertEqual(req.range, (10, -1))
 | 
						|
        self.assertEqual(req.range_unit, 'bytes')
 | 
						|
 | 
						|
        headers = {'Range': 'items=10-'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertEqual(req.range, (10, -1))
 | 
						|
        self.assertEqual(req.range_unit, 'items')
 | 
						|
 | 
						|
        headers = {'Range': ''}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPInvalidHeader, lambda: req.range_unit)
 | 
						|
 | 
						|
        req = Request(testing.create_environ())
 | 
						|
        self.assertIs(req.range_unit, None)
 | 
						|
 | 
						|
    def test_range_invalid(self):
 | 
						|
        headers = {'Range': 'bytes=10240'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=-'}
 | 
						|
        expected_desc = ('The value provided for the Range header is '
 | 
						|
                         'invalid. The range offsets are missing.')
 | 
						|
        self._test_error_details(headers, 'range',
 | 
						|
                                 falcon.HTTPInvalidHeader,
 | 
						|
                                 'Invalid header value', expected_desc)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=--'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=-3-'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=-3-4'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=3-3-4'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=3-3-'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=3-3- '}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=fizbit'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=a-'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=a-3'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=-b'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=3-b'}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertRaises(falcon.HTTPBadRequest, lambda: req.range)
 | 
						|
 | 
						|
        headers = {'Range': 'bytes=x-y'}
 | 
						|
        expected_desc = ('The value provided for the Range header is '
 | 
						|
                         'invalid. It must be a range formatted '
 | 
						|
                         'according to RFC 7233.')
 | 
						|
        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 range.')
 | 
						|
        self._test_error_details(headers, 'range',
 | 
						|
                                 falcon.HTTPInvalidHeader,
 | 
						|
                                 'Invalid header value', expected_desc)
 | 
						|
 | 
						|
        headers = {'Range': '10-'}
 | 
						|
        expected_desc = ('The value provided for the Range '
 | 
						|
                         'header is invalid. The value must be '
 | 
						|
                         "prefixed with a range unit, e.g. 'bytes='")
 | 
						|
        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)
 | 
						|
 | 
						|
    @ddt.data(('Date', 'date'),
 | 
						|
              ('If-Modified-since', 'if_modified_since'),
 | 
						|
              ('If-Unmodified-since', 'if_unmodified_since'),
 | 
						|
              )
 | 
						|
    @ddt.unpack
 | 
						|
    def test_date(self, header, attr):
 | 
						|
        date = datetime.datetime(2013, 4, 4, 5, 19, 18)
 | 
						|
        date_str = 'Thu, 04 Apr 2013 05:19:18 GMT'
 | 
						|
 | 
						|
        self._test_header_expected_value(header, date_str, attr, date)
 | 
						|
 | 
						|
    @ddt.data(('Date', 'date'),
 | 
						|
              ('If-Modified-Since', 'if_modified_since'),
 | 
						|
              ('If-Unmodified-Since', 'if_unmodified_since'),
 | 
						|
              )
 | 
						|
    @ddt.unpack
 | 
						|
    def test_date_invalid(self, header, attr):
 | 
						|
 | 
						|
        # Date formats don't conform to RFC 1123
 | 
						|
        headers = {header: 'Thu, 04 Apr 2013'}
 | 
						|
        expected_desc = ('The value provided for the {0} '
 | 
						|
                         'header is invalid. It must be formatted '
 | 
						|
                         'according to RFC 7231, Section 7.1.1.1')
 | 
						|
 | 
						|
        self._test_error_details(headers, attr,
 | 
						|
                                 falcon.HTTPInvalidHeader,
 | 
						|
                                 'Invalid header value',
 | 
						|
                                 expected_desc.format(header))
 | 
						|
 | 
						|
        headers = {header: ''}
 | 
						|
        self._test_error_details(headers, attr,
 | 
						|
                                 falcon.HTTPInvalidHeader,
 | 
						|
                                 'Invalid header value',
 | 
						|
                                 expected_desc.format(header))
 | 
						|
 | 
						|
    @ddt.data('date', 'if_modified_since', 'if_unmodified_since')
 | 
						|
    def test_date_missing(self, attr):
 | 
						|
        req = Request(testing.create_environ())
 | 
						|
        self.assertIs(getattr(req, attr), None)
 | 
						|
 | 
						|
    def test_attribute_headers(self):
 | 
						|
        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-None-Match', hash, 'if_none_match')
 | 
						|
        self._test_attribute_header('If-Range', hash, 'if_range')
 | 
						|
 | 
						|
        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_header_expected_value(self, name, value, attr, expected_value):
 | 
						|
        headers = {name: value}
 | 
						|
        req = Request(testing.create_environ(headers=headers))
 | 
						|
        self.assertEqual(getattr(req, attr), expected_value)
 | 
						|
 | 
						|
    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)
 |