Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_specific_enforced_routes_with_unenforced_route(self):
config = PxConfig({'app_id': 'PXfake_app_id',
'auth_token': '',
'module_mode': px_constants.MODULE_MODE_BLOCKING,
'enforced_specific_routes': ['/profile'],
});
headers = {'X-FORWARDED-FOR': '127.0.0.1',
'remote-addr': '127.0.0.1',
'content_length': '100'}
request_handler = PxRequestVerifier(config)
builder = EnvironBuilder(headers=headers, path='/')
env = builder.get_environ()
request = Request(env)
context = PxContext(request, request_handler.config)
context.score = 100
response = request_handler.verify_request(context, request)
self.assertEqual(response, True)
def test_wrapper_internals(self):
req = Request.from_values(data={'foo': 'bar'}, method='POST')
req._load_form_data()
assert req.form.to_dict() == {'foo': 'bar'}
# second call does not break
req._load_form_data()
assert req.form.to_dict() == {'foo': 'bar'}
# check reprs
assert repr(req) == ""
resp = Response()
assert repr(resp) == ''
resp.set_data('Hello World!')
assert repr(resp) == ''
resp.response = iter(['Test'])
assert repr(resp) == ''
def test_wrapper_support(self):
req = Request.from_values()
resp = Response()
c = SecureCookie.load_cookie(req, secret_key=b'foo')
assert c.new
c['foo'] = 42
self.assert_equal(c.secret_key, b'foo')
c.save_cookie(resp)
req = Request.from_values(headers={
'Cookie': 'session="%s"' % parse_cookie(resp.headers['set-cookie'])['session']
})
c2 = SecureCookie.load_cookie(req, secret_key=b'foo')
assert not c2.new
self.assert_equal(c2, c)
def wsgi_app(self, environ, start_response):
"""Return a response to the request described by *environ*.
Invokes the application object as a WSGI application and returns an
HTTP Response.
:param environ: The current environment variables.
:param start_response: A callable to provide the response.
"""
request = Request(environ)
urls = self.url_map.bind_to_environ(environ)
response = self.dispatch_request(urls, request)
return response(environ, start_response)
def __init__(self, corpus, src_obj):
"""
arguments:
corpus -- a manatee.Corpus instance (enriched version returned by corplib.CorpusManager)
src_obj -- object holding argument names and values (request or controller.args)
"""
self._corp = corpus
self._src_obj = src_obj
if isinstance(src_obj, Args):
self._attr_producer_fn = lambda o: dir(o)
self._access_fn = lambda o, att: getattr(o, att)
elif isinstance(src_obj, Request):
self._attr_producer_fn = lambda o: o.form.keys()
self._access_fn = lambda o, x: apply(o.form.getlist, (x,))
else:
raise ValueError('Invalid source object (must be either argmapping.Args or Request): %s' % (
src_obj.__class__.__name__,))
def _load_form_data(self):
RequestBase._load_form_data(self)
# In debug mode we're replacing the files multidict with an ad-hoc
# subclass that raises a different error for key errors.
ctx = _request_ctx_stack.top
if ctx is not None and ctx.app.debug and \
self.mimetype != 'multipart/form-data' and not self.files:
from .debughelpers import attach_enctype_error_multidict
attach_enctype_error_multidict(self)
@wrappers.Request.application
def _serve_tags(self, request):
index = self._index_impl()
return http_util.Respond(request, index, 'application/json')
@Request.application
def __call__(self, request):
# This function must return a valid response.
try:
doc = self.get_response_doc()
return xml_response(doc)
except NoApplicableCode as e:
return e
except Exception as e:
return NoApplicableCode(str(e))
def dispatch(self, environ, start_response):
"""
Performs the actual WSGI dispatching for the application.
"""
try:
httprequest = werkzeug.wrappers.Request(environ)
httprequest.app = self
httprequest.parameter_storage_class = werkzeug.datastructures.ImmutableOrderedMultiDict
current_thread = threading.current_thread()
current_thread.url = httprequest.url
current_thread.query_count = 0
current_thread.query_time = 0
current_thread.perf_t0 = time.time()
explicit_session = self.setup_session(httprequest)
self.setup_db(httprequest)
self.setup_lang(httprequest)
request = self.get_request(httprequest)
def _dispatch_nodb():
@provides(MapAdapter)
@scope(RequestScope)
@inject(Map, Request)
def provide_map_adapter(self, url_map, request):
return url_map.bind_to_environ(request.environ)