Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_make_response_response(app: Quart) -> None:
response = await app.make_response(Response("Result"))
assert response.status_code == 200
assert (await response.get_data()) == b"Result" # type: ignore
response = await app.make_response((Response("Result"), {"name": "value"}))
assert response.status_code == 200
assert (await response.get_data()) == b"Result" # type: ignore
assert response.headers["name"] == "value"
response = await app.make_response((Response("Result"), 404, {"name": "value"}))
assert response.status_code == 404
assert (await response.get_data()) == b"Result" # type: ignore
assert response.headers["name"] == "value"
async def test_make_response_response(app: Quart) -> None:
response = await app.make_response(Response("Result"))
assert response.status_code == 200
assert (await response.get_data()) == b"Result" # type: ignore
response = await app.make_response((Response("Result"), {"name": "value"}))
assert response.status_code == 200
assert (await response.get_data()) == b"Result" # type: ignore
assert response.headers["name"] == "value"
response = await app.make_response((Response("Result"), 404, {"name": "value"}))
assert response.status_code == 404
assert (await response.get_data()) == b"Result" # type: ignore
assert response.headers["name"] == "value"
async def ldifDump( dn: str) -> quart.Response:
'Dump an entry as LDIF'
out = io.StringIO()
writer = ldif.LDIFWriter( out)
async for dn, attrs in result( request.ldap.search( dn, ldap.SCOPE_SUBTREE)):
writer.unparse( dn, attrs)
resp = quart.Response( out.getvalue(), content_type='text/plain')
resp.headers['Content-Disposition'] = \
'attachment; filename="%s.ldif"' % dn.split(',')[0].split('=')[1]
return resp
async def messages():
# Main bot message handler.
if "application/json" in request.headers["Content-Type"]:
body = await request.json
else:
return Response("", status=415)
activity = Activity().deserialize(body)
auth_header = (
request.headers["Authorization"] if "Authorization" in request.headers else ""
)
try:
await ADAPTER.process_activity(activity, auth_header, BOT.on_turn)
return Response("", status=201)
except Exception as exception:
raise exception
def not_found(_):
return quart.Response("The page was not found.", status=404)
def empty_response(status, headers=None):
response = Response("", status=status, headers=headers)
del response.headers['Content-Type'] # text/html by default in Flask
return response
async def messages():
# Main bot message handler.
if "application/json" in request.headers["Content-Type"]:
body = await request.json
else:
return Response("", status=415)
activity = Activity().deserialize(body)
auth_header = (
request.headers["Authorization"] if "Authorization" in request.headers else ""
)
try:
await ADAPTER.process_activity(activity, auth_header, BOT.on_turn)
return Response("", status=201)
except Exception as exception:
raise exception
async def wrapped_view( **values) -> quart.Response:
data = await authenticated( view)( **values)
if type( data) is quart.Response: return data
return quart.jsonify( data)
return wrapped_view
except ValueError:
raise ldap.INVALID_CREDENTIALS( {
'desc': 'Invalid user',
'info': "User '%s' unknown" % request.authorization.username})
# Try authenticating
await empty( request.ldap.simple_bind(
dn, request.authorization.password))
# On success, call the view function and release connection
data = await view( **values)
request.ldap.unbind_s()
return data
except ldap.INVALID_CREDENTIALS:
return quart.Response(
'Please log in', 401, UNAUTHORIZED)
except ldap.LDAPError as err:
args = err.args[0]
raise HTTPException( 500, args.get( 'info', ''),
args.get( 'desc', ''))