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_blueprint_url_prefix() -> None:
app = Quart(__name__)
blueprint = Blueprint("blueprint", __name__)
prefix = Blueprint("prefix", __name__, url_prefix="/prefix")
@app.route("/page/")
@blueprint.route("/page/")
@prefix.route("/page/")
async def route() -> ResponseReturnValue:
return "OK"
app.register_blueprint(blueprint, url_prefix="/blueprint")
app.register_blueprint(prefix)
async with app.test_request_context("/page/"):
assert request.blueprint is None
async with app.test_request_context("/prefix/page/"):
assert request.blueprint == "prefix"
async with app.test_request_context("/blueprint/page/"):
assert request.blueprint == "blueprint"
def _app() -> Quart:
app = Quart(__name__)
@app.route("/", methods=["GET", "POST"])
def index() -> ResponseReturnValue:
return request.method
@app.route("/gen")
def gen() -> ResponseReturnValue:
def _gen() -> Generator[bytes, None, None]:
yield b"%d" % threading.current_thread().ident
for _ in range(2):
yield b"b"
return _gen(), 200
return app
def serving_app() -> Quart:
app = Quart(__name__)
@app.route('/')
async def index() -> ResponseReturnValue:
return BASIC_DATA, 202, {'X-Test': 'Test'}
@app.route('/push')
async def push() -> ResponseReturnValue:
response = await make_response(BASIC_DATA, 202, {'X-Test': 'Test'})
response.push_promises.add('/')
return response
return app
async def test_overlapping_request_ctx() -> None:
app = Quart(__name__)
request = Request(
"GET", "http", "/", b"", CIMultiDict(), "", "1.1", send_push_promise=no_op_push
)
ctx1 = app.request_context(request)
await ctx1.__aenter__()
ctx2 = app.request_context(request)
await ctx2.__aenter__()
await ctx1.__aexit__(None, None, None)
assert has_app_context() # Ensure the app context still exists for ctx2
await ctx2.__aexit__(None, None, None)
def test_access_log_header_atoms() -> None:
request_headers = CIMultiDict({
'Random': 'Request',
'Remote-Addr': '127.0.0.1',
})
request = Request('GET', 'http', '/', b'', request_headers)
response_headers = CIMultiDict({
'Random': 'Response',
})
response = Response('Hello', 200, response_headers)
atoms = AccessLogAtoms(request, response, 'h2', 0)
assert atoms['{random}i'] == 'Request'
assert atoms['{RANDOM}i'] == 'Request'
assert atoms['{not-atom}i'] == '-'
assert atoms['{random}o'] == 'Response'
assert atoms['{RANDOM}o'] == 'Response'
async def test_secure_cookie_session_interface_open_session() -> None:
session = SecureCookieSession()
session["something"] = "else"
interface = SecureCookieSessionInterface()
app = Quart(__name__)
app.secret_key = "secret"
response = Response("")
await interface.save_session(app, session, response)
request = Request(
"GET", "http", "/", b"", CIMultiDict(), "", "1.1", send_push_promise=no_op_push
)
request.headers["Cookie"] = response.headers["Set-Cookie"]
new_session = await interface.open_session(app, request)
assert new_session == session
def test_digest_authorization() -> None:
headers = CIMultiDict()
headers["Authorization"] = (
"Digest "
'username="identity", '
'realm="realm@rea.lm", '
'nonce="abcd1234", '
'uri="/path", '
'response="abcd1235", '
'opaque="abcd1236"'
)
request = BaseRequestWebsocket("GET", "http", "/", b"", headers, "", "1.1")
auth = request.authorization
assert auth.username == "identity"
assert auth.realm == "realm@rea.lm"
assert auth.nonce == "abcd1234"
assert auth.uri == "/path"
assert auth.response == "abcd1235"
def basic_map() -> Map:
map_ = Map()
map_.add(Rule("/", {"POST"}, "index"))
map_.add(Rule("/", {"DELETE"}, "delete_index"))
map_.add(Rule("/leaf", {"GET"}, "leaf"))
map_.add(Rule("/branch/", {"GET"}, "branch"))
return map_
def test_path_converter() -> None:
map_ = Map()
map_.add(Rule("/", {"GET"}, "index"))
map_.add(Rule("/constant", {"GET"}, "constant"))
map_.add(Rule("/", {"GET"}, "integer"))
map_.add(Rule("/", {"GET"}, "page"))
map_.add(Rule("//constant", {"GET"}, "page_constant"))
map_.add(Rule("//middle/", {"GET"}, "double_page"))
map_.add(Rule("//middle//constant", {"GET"}, "double_page_constant"))
map_.add(Rule("/Colon:", {"GET"}, "colon_path"))
map_.add(Rule("/Colon:", {"GET"}, "colon_base"))
_test_match(map_, "/", "GET", (map_.endpoints["index"][0], {}))
_test_match(map_, "/constant", "GET", (map_.endpoints["constant"][0], {}))
_test_match(map_, "/20", "GET", (map_.endpoints["integer"][0], {"integer": 20}))
_test_match(map_, "/branch/leaf", "GET", (map_.endpoints["page"][0], {"page": "branch/leaf"}))
_test_match(
map_, "/branch/constant", "GET", (map_.endpoints["page_constant"][0], {"page": "branch"})
)
_test_match(
map_,
"/branch/middle/leaf",
"GET",
(map_.endpoints["double_page"][0], {"left": "branch", "right": "leaf"}),
)
_test_match(
def test_root_path_build() -> None:
map_ = Map()
map_.add(Rule("/", {"GET"}, "http"))
adapter = map_.bind_to_request(False, "", "GET", "/", b"", False, "/rooti")
assert adapter.build("http", method="GET") == "/rooti/"
# Relative root_path case
adapter = map_.bind_to_request(False, "", "GET", "/", b"", False, "rooti")
assert adapter.build("http", method="GET") == "/rooti/"