def test_clean_check_error_raising(self):
"""
Check if clean function returns errors dict when `clean_param_name` raise `ParamException`
"""
resource = CollectionResource(objects_class=None)
env = create_environ(path='/')
req = Request(env)
req.context = {
'doc': {
'id': 1,
'name': 'Opentopic'
}
}
Response()
def clean_name_test(self):
raise ParamException('Test Error Message')
resource.clean_name = clean_name_test
data, errors = resource.clean(req.context['doc'])
self.assertEqual(data, {
'id': 1,
})
self.assertEqual(errors, {'name': ['Test Error Message']})
python类Request()的实例源码
def test_on_put_success_result(self):
"""
Test if we will receive correct response
"""
resource = CollectionResource(objects_class=FakeObjectClass)
env = create_environ(path='/')
req = Request(env)
req.context = {
'doc': {
'id': 1,
'name': 'Opentopic'
}
}
resp = Response()
resource.on_put(req=req, resp=resp)
self.assertEqual(
resp.body,
{'id': 1, 'name': 'Opentopic'}
)
def test_custom_router_takes_req_positional_argument():
def responder(req, resp):
resp.body = 'OK'
class CustomRouter(object):
def find(self, uri, req):
if uri == '/test' and isinstance(req, falcon.Request):
return responder, {'GET': responder}, {}, None
router = CustomRouter()
app = falcon.API(router=router)
client = testing.TestClient(app)
response = client.simulate_request(path='/test')
assert response.content == b'OK'
def test_custom_router_takes_req_keyword_argument():
def responder(req, resp):
resp.body = 'OK'
class CustomRouter(object):
def find(self, uri, req=None):
if uri == '/test' and isinstance(req, falcon.Request):
return responder, {'GET': responder}, {}, None
router = CustomRouter()
app = falcon.API(router=router)
client = testing.TestClient(app)
response = client.simulate_request(path='/test')
assert response.content == b'OK'
def test_slots_request(self):
env = testing.create_environ()
req = Request(env)
try:
req.doesnt = 'exist'
except AttributeError:
pytest.fail('Unable to add additional variables dynamically')
def test_legacy_middleware_called_with_correct_args(self):
global context
app = falcon.API(middleware=[ExecutedFirstMiddleware()])
app.add_route(TEST_ROUTE, MiddlewareClassResource())
client = testing.TestClient(app)
client.simulate_request(path=TEST_ROUTE)
assert isinstance(context['req'], falcon.Request)
assert isinstance(context['resp'], falcon.Response)
assert isinstance(context['resource'], MiddlewareClassResource)
def test_error_composed_before_resp_middleware_called(self):
mw = CaptureResponseMiddleware()
app = falcon.API(middleware=mw)
app.add_route('/', MiddlewareClassResource())
client = testing.TestClient(app)
response = client.simulate_request(path='/', method='POST')
assert response.status == falcon.HTTP_403
assert mw.resp.status == response.status
composed_body = json.loads(mw.resp.body)
assert composed_body['title'] == response.status
assert not mw.req_succeeded
# NOTE(kgriffs): Sanity-check the other params passed to
# process_response()
assert isinstance(mw.req, falcon.Request)
assert isinstance(mw.resource, MiddlewareClassResource)
def test_request_cookie_parsing():
# testing with a github-ish set of cookies
headers = [
(
'Cookie',
"""
logged_in=no;_gh_sess=eyJzZXXzaW9uX2lkIjoiN2;
tz=Europe/Berlin; _ga=GA1.2.332347814.1422308165;
_gat=1;
_octo=GH1.1.201722077.1422308165
"""
),
]
environ = testing.create_environ(headers=headers)
req = falcon.Request(environ)
assert req.cookies['logged_in'] == 'no'
assert req.cookies['tz'] == 'Europe/Berlin'
assert req.cookies['_octo'] == 'GH1.1.201722077.1422308165'
assert 'logged_in' in req.cookies
assert '_gh_sess' in req.cookies
assert 'tz' in req.cookies
assert '_ga' in req.cookies
assert '_gat' in req.cookies
assert '_octo' in req.cookies
def test_bounded_stream_property_empty_body(self):
"""Test that we can get a bounded stream outside of wsgiref."""
environ = testing.create_environ()
req = falcon.Request(environ)
bounded_stream = req.bounded_stream
# NOTE(kgriffs): Verify that we aren't creating a new object
# each time the property is called. Also ensures branch
# coverage of the property implementation.
assert bounded_stream is req.bounded_stream
data = bounded_stream.read()
assert len(data) == 0
def test_request_repr(self):
environ = testing.create_environ()
req = falcon.Request(environ)
_repr = '<%s: %s %r>' % (req.__class__.__name__, req.method, req.url)
assert req.__repr__() == _repr
def test_content_header_missing(self, client):
environ = testing.create_environ()
req = falcon.Request(environ)
for header in ('Content-Type', 'Content-Length'):
assert req.get_header(header) is None
def test_headers_as_list(self, client):
headers = [
('Client-ID', '692ba466-74bb-11e3-bf3f-7567c531c7ca'),
('Accept', 'audio/*; q=0.2, audio/basic')
]
# Unit test
environ = testing.create_environ(headers=headers)
req = falcon.Request(environ)
for name, value in headers:
assert (name.upper(), value) in req.headers.items()
# Functional test
client.app.add_route('/', testing.SimpleTestResource(headers=headers))
result = client.simulate_get()
for name, value in headers:
assert result.headers[name] == value
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_nonlatin_path(self, test_path):
# NOTE(kgriffs): When a request comes in, web servers decode
# the path. The decoded path may contain UTF-8 characters,
# but according to the WSGI spec, no strings can contain chars
# outside ISO-8859-1. Therefore, to reconcile the URI
# encoding standard that allows UTF-8 with the WSGI spec
# that does not, WSGI servers tunnel the string via
# ISO-8859-1. falcon.testing.create_environ() mimics this
# behavior, e.g.:
#
# tunnelled_path = path.encode('utf-8').decode('iso-8859-1')
#
# falcon.Request does the following to reverse the process:
#
# path = tunnelled_path.encode('iso-8859-1').decode('utf-8', 'replace')
#
req = Request(testing.create_environ(
host='com',
path=test_path,
headers=self.headers))
assert req.path == falcon.uri.decode(test_path)
def test_uri_https(self):
# =======================================================
# Default port, implicit
# =======================================================
req = Request(testing.create_environ(
path='/hello', scheme='https'))
uri = ('https://' + testing.DEFAULT_HOST + '/hello')
assert req.uri == uri
# =======================================================
# Default port, explicit
# =======================================================
req = Request(testing.create_environ(
path='/hello', scheme='https', port=443))
uri = ('https://' + testing.DEFAULT_HOST + '/hello')
assert req.uri == uri
# =======================================================
# Non-default port
# =======================================================
req = Request(testing.create_environ(
path='/hello', scheme='https', port=22))
uri = ('https://' + testing.DEFAULT_HOST + ':22/hello')
assert req.uri == uri
def test_relative_uri(self):
assert self.req.relative_uri == self.app + self.relative_uri
assert self.req_noqs.relative_uri == self.app + self.path
req_noapp = Request(testing.create_environ(
path='/hello',
query_string=self.qs,
headers=self.headers))
assert 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
assert req_noapp.relative_uri == self.relative_uri
assert req_noapp.relative_uri == self.relative_uri
options = RequestOptions()
options.strip_url_path_trailing_slash = False
req_noapp = Request(testing.create_environ(
path='/hello/',
query_string=self.qs,
headers=self.headers),
options=options)
assert req_noapp.relative_uri == '/hello/' + '?' + self.qs
def test_client_accepts_props(self):
headers = {'Accept': 'application/xml'}
req = Request(testing.create_environ(headers=headers))
assert req.client_accepts_xml
assert not req.client_accepts_json
assert not req.client_accepts_msgpack
headers = {'Accept': 'application/*'}
req = Request(testing.create_environ(headers=headers))
assert req.client_accepts_xml
assert req.client_accepts_json
assert req.client_accepts_msgpack
headers = {'Accept': 'application/json'}
req = Request(testing.create_environ(headers=headers))
assert not req.client_accepts_xml
assert req.client_accepts_json
assert not req.client_accepts_msgpack
headers = {'Accept': 'application/x-msgpack'}
req = Request(testing.create_environ(headers=headers))
assert not req.client_accepts_xml
assert not req.client_accepts_json
assert req.client_accepts_msgpack
headers = {'Accept': 'application/msgpack'}
req = Request(testing.create_environ(headers=headers))
assert not req.client_accepts_xml
assert not req.client_accepts_json
assert req.client_accepts_msgpack
headers = {
'Accept': 'application/json,application/xml,application/x-msgpack'
}
req = Request(testing.create_environ(headers=headers))
assert req.client_accepts_xml
assert req.client_accepts_json
assert 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'])
assert 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
assert 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'])
assert preferred_type == 'application/xhtml+xml'
headers = {'Accept': '3p12845j;;;asfd;'}
req = Request(testing.create_environ(headers=headers))
preferred_type = req.client_prefers(['application/xhtml+xml'])
assert preferred_type is None
def test_range(self):
headers = {'Range': 'bytes=10-'}
req = Request(testing.create_environ(headers=headers))
assert req.range == (10, -1)
headers = {'Range': 'bytes=10-20'}
req = Request(testing.create_environ(headers=headers))
assert req.range == (10, 20)
headers = {'Range': 'bytes=-10240'}
req = Request(testing.create_environ(headers=headers))
assert req.range == (-10240, -1)
headers = {'Range': 'bytes=0-2'}
req = Request(testing.create_environ(headers=headers))
assert req.range == (0, 2)
headers = {'Range': ''}
req = Request(testing.create_environ(headers=headers))
with pytest.raises(falcon.HTTPInvalidHeader):
req.range
req = Request(testing.create_environ())
assert req.range is None
def test_missing_attribute_header(self):
req = Request(testing.create_environ())
assert req.range is None
req = Request(testing.create_environ())
assert req.content_length is None
