sanic/tests/test_routes.py

1271 lines
34 KiB
Python
Raw Normal View History

2021-02-07 09:38:37 +00:00
import asyncio
import re
2018-12-13 17:50:50 +00:00
2021-02-07 09:38:37 +00:00
from unittest.mock import Mock
2021-02-07 09:38:37 +00:00
import pytest
from sanic_routing.exceptions import (
InvalidUsage,
ParameterNameConflicts,
RouteExists,
)
from sanic_testing.testing import SanicTestClient
2021-02-07 09:38:37 +00:00
from sanic import Blueprint, Sanic
2017-07-13 04:18:56 +01:00
from sanic.constants import HTTP_METHODS
from sanic.exceptions import NotFound, SanicException
2021-02-07 09:38:37 +00:00
from sanic.request import Request
2021-12-21 20:56:12 +00:00
from sanic.response import empty, json, text
2021-01-19 13:54:20 +00:00
2021-02-07 09:38:37 +00:00
@pytest.mark.parametrize(
"path,headers,expected",
(
# app base
(b"/", {}, 200),
(b"/", {"host": "maybe.com"}, 200),
(b"/host", {"host": "matching.com"}, 200),
(b"/host", {"host": "wrong.com"}, 404),
# app strict_slashes default
(b"/without", {}, 200),
(b"/without/", {}, 200),
(b"/with", {}, 200),
(b"/with/", {}, 200),
# app strict_slashes off - expressly
(b"/expwithout", {}, 200),
(b"/expwithout/", {}, 200),
(b"/expwith", {}, 200),
(b"/expwith/", {}, 200),
# app strict_slashes on
(b"/without/strict", {}, 200),
(b"/without/strict/", {}, 404),
(b"/with/strict", {}, 404),
(b"/with/strict/", {}, 200),
# bp1 base
(b"/bp1", {}, 200),
(b"/bp1", {"host": "maybe.com"}, 200),
(b"/bp1/host", {"host": "matching.com"}, 200), # BROKEN ON MASTER
(b"/bp1/host", {"host": "wrong.com"}, 404),
# bp1 strict_slashes default
(b"/bp1/without", {}, 200),
(b"/bp1/without/", {}, 200),
(b"/bp1/with", {}, 200),
(b"/bp1/with/", {}, 200),
# bp1 strict_slashes off - expressly
(b"/bp1/expwithout", {}, 200),
(b"/bp1/expwithout/", {}, 200),
(b"/bp1/expwith", {}, 200),
(b"/bp1/expwith/", {}, 200),
# bp1 strict_slashes on
(b"/bp1/without/strict", {}, 200),
(b"/bp1/without/strict/", {}, 404),
(b"/bp1/with/strict", {}, 404),
(b"/bp1/with/strict/", {}, 200),
# bp2 base
(b"/bp2/", {}, 200),
(b"/bp2/", {"host": "maybe.com"}, 200),
(b"/bp2/host", {"host": "matching.com"}, 200), # BROKEN ON MASTER
(b"/bp2/host", {"host": "wrong.com"}, 404),
# bp2 strict_slashes default
(b"/bp2/without", {}, 200),
(b"/bp2/without/", {}, 404),
(b"/bp2/with", {}, 404),
(b"/bp2/with/", {}, 200),
# # bp2 strict_slashes off - expressly
(b"/bp2/expwithout", {}, 200),
(b"/bp2/expwithout/", {}, 200),
(b"/bp2/expwith", {}, 200),
(b"/bp2/expwith/", {}, 200),
# # bp2 strict_slashes on
(b"/bp2/without/strict", {}, 200),
(b"/bp2/without/strict/", {}, 404),
(b"/bp2/with/strict", {}, 404),
(b"/bp2/with/strict/", {}, 200),
# bp3 base
(b"/bp3", {}, 200),
(b"/bp3", {"host": "maybe.com"}, 200),
(b"/bp3/host", {"host": "matching.com"}, 200), # BROKEN ON MASTER
(b"/bp3/host", {"host": "wrong.com"}, 404),
# bp3 strict_slashes default
(b"/bp3/without", {}, 200),
(b"/bp3/without/", {}, 200),
(b"/bp3/with", {}, 200),
(b"/bp3/with/", {}, 200),
# bp3 strict_slashes off - expressly
(b"/bp3/expwithout", {}, 200),
(b"/bp3/expwithout/", {}, 200),
(b"/bp3/expwith", {}, 200),
(b"/bp3/expwith/", {}, 200),
# bp3 strict_slashes on
(b"/bp3/without/strict", {}, 200),
(b"/bp3/without/strict/", {}, 404),
(b"/bp3/with/strict", {}, 404),
(b"/bp3/with/strict/", {}, 200),
# bp4 base
(b"/bp4", {}, 404),
(b"/bp4", {"host": "maybe.com"}, 200),
(b"/bp4/host", {"host": "matching.com"}, 200), # BROKEN ON MASTER
(b"/bp4/host", {"host": "wrong.com"}, 404),
# bp4 strict_slashes default
(b"/bp4/without", {}, 404),
(b"/bp4/without/", {}, 404),
(b"/bp4/with", {}, 404),
(b"/bp4/with/", {}, 404),
# bp4 strict_slashes off - expressly
(b"/bp4/expwithout", {}, 404),
(b"/bp4/expwithout/", {}, 404),
(b"/bp4/expwith", {}, 404),
(b"/bp4/expwith/", {}, 404),
# bp4 strict_slashes on
(b"/bp4/without/strict", {}, 404),
(b"/bp4/without/strict/", {}, 404),
(b"/bp4/with/strict", {}, 404),
(b"/bp4/with/strict/", {}, 404),
),
)
def test_matching(path, headers, expected):
app = Sanic("dev")
bp1 = Blueprint("bp1", url_prefix="/bp1")
bp2 = Blueprint("bp2", url_prefix="/bp2", strict_slashes=True)
bp3 = Blueprint("bp3", url_prefix="/bp3", strict_slashes=False)
bp4 = Blueprint("bp4", url_prefix="/bp4", host="maybe.com")
def handler(request):
return text("Hello!")
defs = (
("/", None, None),
("/host", None, "matching.com"),
("/without", None, None),
("/with/", None, None),
("/expwithout", False, None),
("/expwith/", False, None),
("/without/strict", True, None),
("/with/strict/", True, None),
)
for uri, strict_slashes, host in defs:
params = {"uri": uri}
if strict_slashes is not None:
params["strict_slashes"] = strict_slashes
if host is not None:
params["host"] = host
app.route(**params)(handler)
bp1.route(**params)(handler)
bp2.route(**params)(handler)
bp3.route(**params)(handler)
bp4.route(**params)(handler)
app.blueprint(bp1)
app.blueprint(bp2)
app.blueprint(bp3)
app.blueprint(bp4)
app.router.finalize()
request = Request(path, headers, None, "GET", None, app)
try:
app.router.get(
request.path, request.method, request.headers.get("host")
)
2021-02-07 09:38:37 +00:00
except NotFound:
response = 404
2021-02-15 12:01:32 +00:00
except Exception:
2021-02-07 09:38:37 +00:00
response = 500
else:
response = 200
assert response == expected
2021-01-28 07:18:06 +00:00
# # ------------------------------------------------------------ #
# # UTF-8
# # ------------------------------------------------------------ #
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
@pytest.mark.parametrize("method", HTTP_METHODS)
def test_versioned_routes_get(app, method):
method = method.lower()
2017-07-13 04:18:56 +01:00
2021-02-07 09:38:37 +00:00
func = getattr(app, method)
if callable(func):
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
@func(f"/{method}", version=1)
def handler(request):
return text("OK")
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
else:
raise Exception(f"Method: {method} is not callable")
2017-07-13 04:18:56 +01:00
2021-02-07 09:38:37 +00:00
client_method = getattr(app.test_client, method)
2017-07-13 04:18:56 +01:00
2021-02-07 09:38:37 +00:00
request, response = client_method(f"/v1/{method}")
assert response.status == 200
2018-01-08 00:38:54 +00:00
2017-07-13 04:18:56 +01:00
2021-02-07 09:38:37 +00:00
def test_shorthand_routes_get(app):
@app.get("/get")
def handler(request):
return text("OK")
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get")
assert response.text == "OK"
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/get")
assert response.status == 405
2017-01-20 07:47:07 +00:00
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
def test_shorthand_routes_multiple(app):
@app.get("/get")
def get_handler(request):
return text("OK")
2021-02-07 09:38:37 +00:00
@app.options("/get")
def options_handler(request):
return text("")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get/")
assert response.status == 200
assert response.text == "OK"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.options("/get/")
assert response.status == 200
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
def test_route_strict_slash(app):
@app.get("/get", strict_slashes=True)
def handler1(request):
return text("OK")
2017-03-16 04:11:45 +00:00
2021-02-07 09:38:37 +00:00
@app.post("/post/", strict_slashes=True)
def handler2(request):
return text("OK")
2017-03-16 04:11:45 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get")
assert response.text == "OK"
2017-03-16 04:11:45 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get/")
assert response.status == 404
2017-03-16 04:11:45 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/post/")
assert response.text == "OK"
2017-03-16 04:11:45 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/post")
assert response.status == 404
2017-03-16 04:11:45 +00:00
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
def test_route_invalid_parameter_syntax(app):
with pytest.raises(ValueError):
@app.get("/get/<:str>", strict_slashes=True)
2021-02-07 09:38:37 +00:00
def handler(request):
return text("OK")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get")
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
def test_route_strict_slash_default_value():
app = Sanic("test_route_strict_slash", strict_slashes=True)
2017-08-21 07:37:22 +01:00
2021-02-07 09:38:37 +00:00
@app.get("/get")
def handler(request):
return text("OK")
2017-08-21 07:37:22 +01:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get/")
assert response.status == 404
2017-08-21 07:37:22 +01:00
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
def test_route_strict_slash_without_passing_default_value(app):
@app.get("/get")
def handler(request):
return text("OK")
2017-08-21 07:37:22 +01:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get/")
assert response.text == "OK"
2017-08-21 07:37:22 +01:00
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
def test_route_strict_slash_default_value_can_be_overwritten():
app = Sanic("test_route_strict_slash", strict_slashes=True)
2017-08-21 07:37:22 +01:00
2021-02-07 09:38:37 +00:00
@app.get("/get", strict_slashes=False)
def handler(request):
return text("OK")
2017-08-21 07:37:22 +01:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get/")
assert response.text == "OK"
2017-08-21 07:37:22 +01:00
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
def test_route_slashes_overload(app):
@app.get("/hello/")
def handler_get(request):
return text("OK")
2018-01-01 10:14:55 +00:00
2021-02-07 09:38:37 +00:00
@app.post("/hello/")
def handler_post(request):
return text("OK")
2018-01-01 10:14:55 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/hello")
assert response.text == "OK"
2018-01-01 10:14:55 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/hello/")
assert response.text == "OK"
2018-01-01 10:14:55 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/hello")
assert response.text == "OK"
2018-01-01 10:14:55 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/hello/")
assert response.text == "OK"
2018-01-01 10:14:55 +00:00
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
def test_route_optional_slash(app):
@app.get("/get")
def handler(request):
return text("OK")
2017-02-20 23:58:27 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get")
assert response.text == "OK"
2017-02-20 23:58:27 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/get/")
assert response.text == "OK"
2017-02-20 23:58:27 +00:00
2018-09-29 18:54:47 +01:00
2021-02-07 09:38:37 +00:00
def test_route_strict_slashes_set_to_false_and_host_is_a_list(app):
# Part of regression test for issue #1120
test_client = SanicTestClient(app, port=42101)
site1 = f"127.0.0.1:{test_client.port}"
2018-02-09 21:33:34 +00:00
2021-02-07 09:38:37 +00:00
# before fix, this raises a RouteExists error
@app.get("/get", host=[site1, "site2.com"], strict_slashes=False)
def get_handler(request):
return text("OK")
2018-02-09 21:33:34 +00:00
2021-02-07 09:38:37 +00:00
request, response = test_client.get("http://" + site1 + "/get")
assert response.text == "OK"
2018-02-09 21:33:34 +00:00
2021-02-07 09:38:37 +00:00
app.router.finalized = False
2018-02-09 21:33:34 +00:00
2021-02-07 09:38:37 +00:00
@app.post("/post", host=[site1, "site2.com"], strict_slashes=False)
def post_handler(request):
return text("OK")
2018-02-09 21:33:34 +00:00
2021-02-07 09:38:37 +00:00
request, response = test_client.post("http://" + site1 + "/post")
assert response.text == "OK"
2018-02-09 21:33:34 +00:00
2021-02-07 09:38:37 +00:00
app.router.finalized = False
2018-02-09 21:33:34 +00:00
2021-02-07 09:38:37 +00:00
@app.put("/put", host=[site1, "site2.com"], strict_slashes=False)
def put_handler(request):
return text("OK")
2018-02-09 21:33:34 +00:00
2021-02-07 09:38:37 +00:00
request, response = test_client.put("http://" + site1 + "/put")
assert response.text == "OK"
2018-02-09 21:33:34 +00:00
2021-02-07 09:38:37 +00:00
app.router.finalized = False
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
@app.delete("/delete", host=[site1, "site2.com"], strict_slashes=False)
def delete_handler(request):
return text("OK")
2018-10-01 14:46:18 +01:00
2021-02-07 09:38:37 +00:00
request, response = test_client.delete("http://" + site1 + "/delete")
assert response.text == "OK"
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
def test_shorthand_routes_post(app):
@app.post("/post")
def handler(request):
return text("OK")
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/post")
assert response.text == "OK"
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/post")
assert response.status == 405
2017-01-20 18:00:51 +00:00
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
def test_shorthand_routes_put(app):
@app.put("/put")
def handler(request):
return text("OK")
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.put("/put")
assert response.text == "OK"
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/put")
assert response.status == 405
2021-02-07 09:38:37 +00:00
def test_shorthand_routes_delete(app):
@app.delete("/delete")
def handler(request):
return text("OK")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.delete("/delete")
assert response.text == "OK"
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/delete")
assert response.status == 405
2017-01-20 18:00:51 +00:00
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
def test_shorthand_routes_patch(app):
@app.patch("/patch")
def handler(request):
return text("OK")
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.patch("/patch")
assert response.text == "OK"
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/patch")
assert response.status == 405
2017-01-20 18:00:51 +00:00
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
def test_shorthand_routes_head(app):
@app.head("/head")
def handler(request):
return text("OK")
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.head("/head")
assert response.status == 200
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/head")
assert response.status == 405
2017-01-20 07:47:07 +00:00
2017-01-20 18:00:51 +00:00
2021-02-07 09:38:37 +00:00
def test_shorthand_routes_options(app):
@app.options("/options")
def handler(request):
return text("OK")
2017-01-20 07:47:07 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.options("/options")
assert response.status == 200
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/options")
assert response.status == 405
2021-02-07 09:38:37 +00:00
def test_static_routes(app):
@app.route("/test")
async def handler1(request):
return text("OK1")
2021-02-07 09:38:37 +00:00
@app.route("/pizazz")
async def handler2(request):
return text("OK2")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/test")
assert response.text == "OK1"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/pizazz")
assert response.text == "OK2"
2021-02-07 09:38:37 +00:00
def test_dynamic_route(app):
results = []
2021-02-07 09:38:37 +00:00
@app.route("/folder/<name>")
async def handler(request, name):
results.append(name)
return text("OK")
2021-02-07 09:38:37 +00:00
app.router.finalize(False)
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/test123")
2021-02-07 09:38:37 +00:00
assert response.text == "OK"
assert results[0] == "test123"
2021-02-07 09:38:37 +00:00
def test_dynamic_route_string(app):
results = []
@app.route("/folder/<name:str>")
2021-02-07 09:38:37 +00:00
async def handler(request, name):
results.append(name)
return text("OK")
2016-10-16 21:41:56 +01:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/test123")
2016-10-16 21:41:56 +01:00
2021-02-07 09:38:37 +00:00
assert response.text == "OK"
assert results[0] == "test123"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/favicon.ico")
2021-02-07 09:38:37 +00:00
assert response.text == "OK"
assert results[1] == "favicon.ico"
2021-02-07 09:38:37 +00:00
def test_dynamic_route_int(app):
results = []
2021-02-07 09:38:37 +00:00
@app.route("/folder/<folder_id:int>")
async def handler(request, folder_id):
results.append(folder_id)
return text("OK")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/12345")
assert response.text == "OK"
assert type(results[0]) is int
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/asdf")
assert response.status == 404
2021-02-07 09:38:37 +00:00
def test_dynamic_route_number(app):
results = []
@app.route("/weight/<weight:float>")
2021-02-07 09:38:37 +00:00
async def handler(request, weight):
results.append(weight)
return text("OK")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/12345")
assert response.text == "OK"
assert type(results[0]) is float
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/1234.56")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/.12")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/12.")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/1234-56")
assert response.status == 404
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/12.34.56")
assert response.status == 404
2021-02-07 09:38:37 +00:00
def test_dynamic_route_regex(app):
@app.route("/folder/<folder_id:[A-Za-z0-9]{0,4}>")
async def handler(request, folder_id):
return text("OK")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/test")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/test1")
assert response.status == 404
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/test-123")
assert response.status == 404
2018-06-09 09:16:17 +01:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/")
assert response.status == 200
2018-06-09 09:16:17 +01:00
2021-02-07 09:38:37 +00:00
def test_dynamic_route_uuid(app):
import uuid
2018-06-09 09:16:17 +01:00
2021-02-07 09:38:37 +00:00
results = []
2018-06-09 09:16:17 +01:00
2021-02-07 09:38:37 +00:00
@app.route("/quirky/<unique_id:uuid>")
async def handler(request, unique_id):
results.append(unique_id)
return text("OK")
url = "/quirky/123e4567-e89b-12d3-a456-426655440000"
request, response = app.test_client.get(url)
assert response.text == "OK"
assert type(results[0]) is uuid.UUID
generated_uuid = uuid.uuid4()
request, response = app.test_client.get(f"/quirky/{generated_uuid}")
assert response.status == 200
request, response = app.test_client.get("/quirky/non-existing")
assert response.status == 404
2018-06-09 09:16:17 +01:00
2021-02-08 10:18:29 +00:00
def test_dynamic_route_path(app):
@app.route("/<path:path>/info")
async def handler(request, path):
return text("OK")
app.router.finalize()
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/path/1/info")
assert response.status == 200
2017-04-13 04:34:35 +01:00
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/info")
assert response.status == 404
2017-04-13 04:34:35 +01:00
2021-02-08 10:18:29 +00:00
app.router.reset()
2017-04-13 04:34:35 +01:00
2021-02-08 10:18:29 +00:00
@app.route("/<path:path>")
async def handler1(request, path):
return text("OK")
2017-04-13 04:34:35 +01:00
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/info")
assert response.status == 200
2017-04-13 04:34:35 +01:00
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/whatever/you/set")
assert response.status == 200
2017-04-13 04:34:35 +01:00
2021-02-08 10:18:29 +00:00
def test_dynamic_route_unhashable(app):
@app.route("/folder/<unhashable:[A-Za-z0-9/]+>/end/")
async def handler(request, unhashable):
return text("OK")
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/folder/test/asdf/end/")
assert response.status == 200
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/folder/test///////end/")
assert response.status == 200
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/folder/test/end/")
assert response.status == 200
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/folder/test/nope/")
assert response.status == 404
2021-02-07 09:38:37 +00:00
@pytest.mark.parametrize("url", ["/ws", "ws"])
def test_websocket_route(app, url):
ev = asyncio.Event()
2017-02-20 23:09:54 +00:00
2021-02-07 09:38:37 +00:00
@app.websocket(url)
async def handler(request, ws):
assert request.scheme == "ws"
assert ws.subprotocol is None
ev.set()
2017-02-20 23:09:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.websocket(url)
assert response.opened is True
assert ev.is_set()
2017-02-20 23:09:54 +00:00
2021-02-15 19:50:20 +00:00
def test_websocket_route_invalid_handler(app):
with pytest.raises(ValueError) as e:
@app.websocket("/")
async def handler():
...
assert e.match(
r"Required parameter `request` and/or `ws` missing in the "
r"handler\(\) route\?"
)
2021-02-08 10:18:29 +00:00
@pytest.mark.asyncio
@pytest.mark.parametrize("url", ["/ws", "ws"])
async def test_websocket_route_asgi(app, url):
@app.after_server_start
async def setup_ev(app, _):
app.ctx.ev = asyncio.Event()
2021-02-08 10:18:29 +00:00
@app.websocket(url)
async def handler(request, ws):
request.app.ctx.ev.set()
@app.get("/ev")
async def check(request):
return json({"set": request.app.ctx.ev.is_set()})
_, response = await app.asgi_client.websocket(url)
_, response = await app.asgi_client.get("/")
assert response.json["set"]
@pytest.mark.parametrize(
"subprotocols,expected",
(
New websockets (#2158) * First attempt at new Websockets implementation based on websockets >= 9.0, with sans-i/o features. Requires more work. * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * wip, update websockets code to new Sans/IO API * Refactored new websockets impl into own modules Incorporated other suggestions made by team * Another round of work on the new websockets impl * Added websocket_timeout support (matching previous/legacy support) * Lots more comments * Incorporated suggested changes from previous round of review * Changed RuntimeError usage to ServerError * Changed SanicException usage to ServerError * Removed some redundant asserts * Change remaining asserts to ServerErrors * Fixed some timeout handling issues * Fixed websocket.close() handling, and made it more robust * Made auto_close task smarter and more error-resilient * Made fail_connection routine smarter and more error-resilient * Further new websockets impl fixes * Update compatibility with Websockets v10 * Track server connection state in a more precise way * Try to handle the shutdown process more gracefully * Add a new end_connection() helper, to use as an alterative to close() or fail_connection() * Kill the auto-close task and keepalive-timeout task when sanic is shutdown * Deprecate WEBSOCKET_READ_LIMIT and WEBSOCKET_WRITE_LIMIT configs, they are not used in this implementation. * Change a warning message to debug level Remove default values for deprecated websocket parameters * Fix flake8 errors * Fix a couple of missed failing tests * remove websocket bench from examples * Integrate suggestions from code reviews Use Optional[T] instead of union[T,None] Fix mypy type logic errors change "is not None" to truthy checks where appropriate change "is None" to falsy checks were appropriate Add more debug logging when debug mode is on Change to using sanic.logger for debug logging rather than error_logger. * Fix long line lengths of debug messages Add some new debug messages when websocket IO is paused and unpaused for flow control Fix websocket example to use app.static() * remove unused import in websocket example app * re-run isort after Flake8 fixes Co-authored-by: Adam Hopkins <adam@amhopkins.com> Co-authored-by: Adam Hopkins <admhpkns@gmail.com>
2021-09-29 11:09:23 +01:00
(["one"], "one"),
(["three", "one"], "one"),
(["tree"], None),
(None, None),
),
)
def test_websocket_route_with_subprotocols(app, subprotocols, expected):
New websockets (#2158) * First attempt at new Websockets implementation based on websockets >= 9.0, with sans-i/o features. Requires more work. * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * wip, update websockets code to new Sans/IO API * Refactored new websockets impl into own modules Incorporated other suggestions made by team * Another round of work on the new websockets impl * Added websocket_timeout support (matching previous/legacy support) * Lots more comments * Incorporated suggested changes from previous round of review * Changed RuntimeError usage to ServerError * Changed SanicException usage to ServerError * Removed some redundant asserts * Change remaining asserts to ServerErrors * Fixed some timeout handling issues * Fixed websocket.close() handling, and made it more robust * Made auto_close task smarter and more error-resilient * Made fail_connection routine smarter and more error-resilient * Further new websockets impl fixes * Update compatibility with Websockets v10 * Track server connection state in a more precise way * Try to handle the shutdown process more gracefully * Add a new end_connection() helper, to use as an alterative to close() or fail_connection() * Kill the auto-close task and keepalive-timeout task when sanic is shutdown * Deprecate WEBSOCKET_READ_LIMIT and WEBSOCKET_WRITE_LIMIT configs, they are not used in this implementation. * Change a warning message to debug level Remove default values for deprecated websocket parameters * Fix flake8 errors * Fix a couple of missed failing tests * remove websocket bench from examples * Integrate suggestions from code reviews Use Optional[T] instead of union[T,None] Fix mypy type logic errors change "is not None" to truthy checks where appropriate change "is None" to falsy checks were appropriate Add more debug logging when debug mode is on Change to using sanic.logger for debug logging rather than error_logger. * Fix long line lengths of debug messages Add some new debug messages when websocket IO is paused and unpaused for flow control Fix websocket example to use app.static() * remove unused import in websocket example app * re-run isort after Flake8 fixes Co-authored-by: Adam Hopkins <adam@amhopkins.com> Co-authored-by: Adam Hopkins <admhpkns@gmail.com>
2021-09-29 11:09:23 +01:00
results = []
New websockets (#2158) * First attempt at new Websockets implementation based on websockets >= 9.0, with sans-i/o features. Requires more work. * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * wip, update websockets code to new Sans/IO API * Refactored new websockets impl into own modules Incorporated other suggestions made by team * Another round of work on the new websockets impl * Added websocket_timeout support (matching previous/legacy support) * Lots more comments * Incorporated suggested changes from previous round of review * Changed RuntimeError usage to ServerError * Changed SanicException usage to ServerError * Removed some redundant asserts * Change remaining asserts to ServerErrors * Fixed some timeout handling issues * Fixed websocket.close() handling, and made it more robust * Made auto_close task smarter and more error-resilient * Made fail_connection routine smarter and more error-resilient * Further new websockets impl fixes * Update compatibility with Websockets v10 * Track server connection state in a more precise way * Try to handle the shutdown process more gracefully * Add a new end_connection() helper, to use as an alterative to close() or fail_connection() * Kill the auto-close task and keepalive-timeout task when sanic is shutdown * Deprecate WEBSOCKET_READ_LIMIT and WEBSOCKET_WRITE_LIMIT configs, they are not used in this implementation. * Change a warning message to debug level Remove default values for deprecated websocket parameters * Fix flake8 errors * Fix a couple of missed failing tests * remove websocket bench from examples * Integrate suggestions from code reviews Use Optional[T] instead of union[T,None] Fix mypy type logic errors change "is not None" to truthy checks where appropriate change "is None" to falsy checks were appropriate Add more debug logging when debug mode is on Change to using sanic.logger for debug logging rather than error_logger. * Fix long line lengths of debug messages Add some new debug messages when websocket IO is paused and unpaused for flow control Fix websocket example to use app.static() * remove unused import in websocket example app * re-run isort after Flake8 fixes Co-authored-by: Adam Hopkins <adam@amhopkins.com> Co-authored-by: Adam Hopkins <admhpkns@gmail.com>
2021-09-29 11:09:23 +01:00
@app.websocket("/ws", subprotocols=["zero", "one", "two", "three"])
2021-02-07 09:38:37 +00:00
async def handler(request, ws):
nonlocal results
results = ws.subprotocol
2021-02-07 09:38:37 +00:00
assert ws.subprotocol is not None
2021-02-07 09:38:37 +00:00
_, response = SanicTestClient(app).websocket(
"/ws", subprotocols=subprotocols
2021-02-07 09:38:37 +00:00
)
assert response.opened is True
assert results == expected
2021-02-07 09:38:37 +00:00
@pytest.mark.parametrize("strict_slashes", [True, False, None])
def test_add_webscoket_route(app, strict_slashes):
ev = asyncio.Event()
2018-12-13 17:50:50 +00:00
2021-02-07 09:38:37 +00:00
async def handler(request, ws):
assert ws.subprotocol is None
ev.set()
2018-12-13 17:50:50 +00:00
2021-02-07 09:38:37 +00:00
app.add_websocket_route(handler, "/ws", strict_slashes=strict_slashes)
request, response = app.test_client.websocket("/ws")
assert response.opened is True
assert ev.is_set()
2018-12-13 17:50:50 +00:00
2021-02-07 09:38:37 +00:00
def test_add_webscoket_route_with_version(app):
ev = asyncio.Event()
2021-02-07 09:38:37 +00:00
async def handler(request, ws):
assert ws.subprotocol is None
ev.set()
2021-02-07 09:38:37 +00:00
app.add_websocket_route(handler, "/ws", version=1)
request, response = app.test_client.websocket("/v1/ws")
assert response.opened is True
assert ev.is_set()
2021-02-07 09:38:37 +00:00
def test_route_duplicate(app):
2021-02-07 09:38:37 +00:00
with pytest.raises(RouteExists):
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
@app.route("/test")
async def handler1(request):
pass
2021-02-07 09:38:37 +00:00
@app.route("/test")
async def handler2(request):
pass
2021-02-07 09:38:37 +00:00
with pytest.raises(RouteExists):
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
@app.route("/test/<dynamic>/")
async def handler3(request, dynamic):
pass
2021-02-07 09:38:37 +00:00
@app.route("/test/<dynamic>/")
async def handler4(request, dynamic):
pass
2021-02-07 09:38:37 +00:00
def test_double_stack_route(app):
@app.route("/test/1")
@app.route("/test/2")
async def handler1(request):
return text("OK")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/test/1")
assert response.status == 200
request, response = app.test_client.get("/test/2")
assert response.status == 200
2021-02-07 09:38:37 +00:00
@pytest.mark.asyncio
async def test_websocket_route_asgi(app):
ev = asyncio.Event()
2021-02-07 09:38:37 +00:00
@app.websocket("/test/1")
@app.websocket("/test/2")
async def handler(request, ws):
ev.set()
2021-02-07 09:38:37 +00:00
request, response = await app.asgi_client.websocket("/test/1")
first_set = ev.is_set()
ev.clear()
request, response = await app.asgi_client.websocket("/test/1")
second_set = ev.is_set()
assert first_set and second_set
2021-02-07 09:38:37 +00:00
def test_method_not_allowed(app):
@app.route("/test", methods=["GET"])
async def handler(request):
return text("OK")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/test")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/test")
assert response.status == 405
2021-02-07 09:38:37 +00:00
@pytest.mark.parametrize("strict_slashes", [True, False, None])
def test_static_add_route(app, strict_slashes):
async def handler1(request):
return text("OK1")
2021-02-07 09:38:37 +00:00
async def handler2(request):
return text("OK2")
2021-02-07 09:38:37 +00:00
app.add_route(handler1, "/test", strict_slashes=strict_slashes)
app.add_route(handler2, "/test2", strict_slashes=strict_slashes)
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/test")
assert response.text == "OK1"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/test2")
assert response.text == "OK2"
2021-02-07 09:38:37 +00:00
def test_dynamic_add_route(app):
2021-02-07 09:38:37 +00:00
results = []
2021-02-07 09:38:37 +00:00
async def handler(request, name):
results.append(name)
return text("OK")
2021-02-07 09:38:37 +00:00
app.add_route(handler, "/folder/<name>")
request, response = app.test_client.get("/folder/test123")
2021-02-07 09:38:37 +00:00
assert response.text == "OK"
assert results[0] == "test123"
2021-02-07 09:38:37 +00:00
def test_dynamic_add_route_string(app):
2021-02-07 09:38:37 +00:00
results = []
2021-02-07 09:38:37 +00:00
async def handler(request, name):
results.append(name)
return text("OK")
app.add_route(handler, "/folder/<name:str>")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/test123")
2021-02-07 09:38:37 +00:00
assert response.text == "OK"
assert results[0] == "test123"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/favicon.ico")
2021-02-07 09:38:37 +00:00
assert response.text == "OK"
assert results[1] == "favicon.ico"
2021-02-07 09:38:37 +00:00
def test_dynamic_add_route_int(app):
results = []
2021-02-07 09:38:37 +00:00
async def handler(request, folder_id):
results.append(folder_id)
return text("OK")
2021-02-07 09:38:37 +00:00
app.add_route(handler, "/folder/<folder_id:int>")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/12345")
assert response.text == "OK"
assert type(results[0]) is int
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/asdf")
assert response.status == 404
2021-02-07 09:38:37 +00:00
def test_dynamic_add_route_number(app):
results = []
2021-02-07 09:38:37 +00:00
async def handler(request, weight):
results.append(weight)
return text("OK")
app.add_route(handler, "/weight/<weight:float>")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/12345")
assert response.text == "OK"
assert type(results[0]) is float
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/1234.56")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/.12")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/12.")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/1234-56")
assert response.status == 404
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/weight/12.34.56")
assert response.status == 404
2021-02-07 09:38:37 +00:00
def test_dynamic_add_route_regex(app):
async def handler(request, folder_id):
return text("OK")
2021-02-07 09:38:37 +00:00
app.add_route(handler, "/folder/<folder_id:[A-Za-z0-9]{0,4}>")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/test")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/test1")
assert response.status == 404
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/test-123")
assert response.status == 404
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/folder/")
assert response.status == 200
2021-02-08 10:18:29 +00:00
def test_dynamic_add_route_unhashable(app):
async def handler(request, unhashable):
return text("OK")
2021-02-08 10:18:29 +00:00
app.add_route(handler, "/folder/<unhashable:[A-Za-z0-9/]+>/end/")
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/folder/test/asdf/end/")
assert response.status == 200
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/folder/test///////end/")
assert response.status == 200
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/folder/test/end/")
assert response.status == 200
2021-02-08 10:18:29 +00:00
request, response = app.test_client.get("/folder/test/nope/")
assert response.status == 404
2021-02-07 09:38:37 +00:00
def test_add_route_duplicate(app):
2021-02-07 09:38:37 +00:00
with pytest.raises(RouteExists):
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
async def handler1(request):
pass
2021-02-07 09:38:37 +00:00
async def handler2(request):
pass
2021-02-07 09:38:37 +00:00
app.add_route(handler1, "/test")
app.add_route(handler2, "/test")
2021-02-07 09:38:37 +00:00
with pytest.raises(RouteExists):
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
async def handler1(request, dynamic):
pass
2021-02-07 09:38:37 +00:00
async def handler2(request, dynamic):
pass
2021-02-07 09:38:37 +00:00
app.add_route(handler1, "/test/<dynamic>/")
app.add_route(handler2, "/test/<dynamic>/")
2021-02-07 09:38:37 +00:00
def test_add_route_method_not_allowed(app):
async def handler(request):
return text("OK")
2021-02-07 09:38:37 +00:00
app.add_route(handler, "/test", methods=["GET"])
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/test")
assert response.status == 200
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/test")
assert response.status == 405
2021-02-07 09:38:37 +00:00
def test_removing_slash(app):
@app.get("/rest/<resource>")
def get(_):
pass
2017-02-28 03:54:58 +00:00
2021-02-07 09:38:37 +00:00
@app.post("/rest/<resource>")
def post(_):
pass
2017-02-28 03:54:58 +00:00
2021-02-07 09:38:37 +00:00
assert len(app.router.routes_all.keys()) == 1
2017-02-28 03:54:58 +00:00
2021-02-07 09:38:37 +00:00
def test_overload_routes(app):
@app.route("/overload", methods=["GET"])
async def handler1(request):
return text("OK1")
2021-02-07 09:38:37 +00:00
@app.route("/overload", methods=["POST", "PUT"])
async def handler2(request):
return text("OK2")
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/overload")
assert response.text == "OK1"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/overload")
assert response.text == "OK2"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.put("/overload")
assert response.text == "OK2"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.delete("/overload")
assert response.status == 405
2021-02-07 09:38:37 +00:00
app.router.reset()
with pytest.raises(RouteExists):
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
@app.route("/overload", methods=["PUT", "DELETE"])
async def handler3(request):
return text("Duplicated")
2021-02-07 09:38:37 +00:00
def test_unmergeable_overload_routes(app):
@app.route("/overload_whole", methods=None)
async def handler1(request):
return text("OK1")
2021-02-07 09:38:37 +00:00
@app.route("/overload_whole", methods=["POST", "PUT"])
async def handler2(request):
return text("OK1")
2018-12-30 11:18:06 +00:00
assert len(app.router.static_routes) == 1
assert len(app.router.static_routes[("overload_whole",)].methods) == 3
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/overload_whole")
assert response.text == "OK1"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/overload_whole")
assert response.text == "OK1"
2021-02-07 09:38:37 +00:00
request, response = app.test_client.put("/overload_whole")
assert response.text == "OK1"
2021-02-07 09:38:37 +00:00
app.router.reset()
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
@app.route("/overload_part", methods=["GET"])
async def handler3(request):
return text("OK1")
2021-02-07 09:38:37 +00:00
with pytest.raises(RouteExists):
2021-02-07 09:38:37 +00:00
@app.route("/overload_part")
async def handler4(request):
return text("Duplicated")
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/overload_part")
assert response.text == "OK1"
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/overload_part")
assert response.status == 405
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
def test_unicode_routes(app):
@app.get("/你好")
def handler1(request):
return text("OK1")
2018-01-08 00:38:54 +00:00
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/你好")
assert response.text == "OK1"
2021-02-07 09:38:37 +00:00
app.router.reset()
2021-02-07 09:38:37 +00:00
@app.route("/overload/<param>", methods=["GET"], unquote=True)
async def handler2(request, param):
return text("OK2 " + param)
2021-02-07 09:38:37 +00:00
request, response = app.test_client.get("/overload/你好")
assert response.text == "OK2 你好"
2021-02-07 09:38:37 +00:00
def test_uri_with_different_method_and_different_params(app):
@app.route("/ads/<ad_id>", methods=["GET"])
async def ad_get(request, ad_id):
return json({"ad_id": ad_id})
@app.route("/ads/<action>", methods=["POST"])
async def ad_post(request, action):
return json({"action": action})
request, response = app.test_client.get("/ads/1234")
assert response.status == 200
assert response.json == {"ad_id": "1234"}
2021-02-07 09:38:37 +00:00
request, response = app.test_client.post("/ads/post")
assert response.status == 200
assert response.json == {"action": "post"}
def test_uri_with_different_method_and_same_params(app):
@app.route("/ads/<ad_id>", methods=["GET"])
async def ad_get(request, ad_id):
return json({"ad_id": ad_id})
@app.route("/ads/<ad_id>", methods=["POST"])
async def ad_post(request, ad_id):
return json({"ad_id": ad_id})
request, response = app.test_client.get("/ads/1234")
assert response.status == 200
assert response.json == {"ad_id": "1234"}
request, response = app.test_client.post("/ads/post")
assert response.status == 200
assert response.json == {"ad_id": "post"}
2021-02-07 09:38:37 +00:00
def test_route_raise_ParameterNameConflicts(app):
@app.get("/api/v1/<user>/<user>/")
def handler(request, user):
return text("OK")
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
with pytest.raises(ParameterNameConflicts):
app.router.finalize()
2021-02-07 09:38:37 +00:00
def test_route_invalid_host(app):
2021-02-07 09:38:37 +00:00
host = 321
with pytest.raises(ValueError) as excinfo:
2018-12-30 11:18:06 +00:00
2021-02-07 09:38:37 +00:00
@app.get("/test", host=host)
def handler(request):
return text("pass")
2021-02-07 09:38:37 +00:00
assert str(excinfo.value) == (
"Expected either string or Iterable of " "host strings, not {!r}"
).format(host)
def test_route_with_regex_group(app):
@app.route("/path/to/<ext:file\.(txt)>")
async def handler(request, ext):
return text(ext)
_, response = app.test_client.get("/path/to/file.txt")
assert response.text == "txt"
def test_route_with_regex_named_group(app):
@app.route(r"/path/to/<ext:file\.(?P<ext>txt)>")
async def handler(request, ext):
return text(ext)
_, response = app.test_client.get("/path/to/file.txt")
assert response.text == "txt"
def test_route_with_regex_named_group_invalid(app):
@app.route(r"/path/to/<ext:file\.(?P<wrong>txt)>")
async def handler(request, ext):
return text(ext)
with pytest.raises(InvalidUsage) as e:
app.router.finalize()
assert e.match(
re.escape("Named group (wrong) must match your named parameter (ext)")
)
def test_route_with_regex_group_ambiguous(app):
@app.route("/path/to/<ext:file(?:\.)(txt)>")
async def handler(request, ext):
return text(ext)
with pytest.raises(InvalidUsage) as e:
app.router.finalize()
assert e.match(
re.escape(
"Could not compile pattern file(?:\.)(txt). Try using a named "
"group instead: '(?P<ext>your_matching_group)'"
)
)
def test_route_with_bad_named_param(app):
@app.route("/foo/<__bar__>")
async def handler(request):
return text("...")
with pytest.raises(SanicException):
app.router.finalize()
def test_routes_with_and_without_slash_definitions(app):
bar = Blueprint("bar", url_prefix="bar")
baz = Blueprint("baz", url_prefix="/baz")
fizz = Blueprint("fizz", url_prefix="fizz/")
buzz = Blueprint("buzz", url_prefix="/buzz/")
instances = (
(app, "foo"),
(bar, "bar"),
(baz, "baz"),
(fizz, "fizz"),
(buzz, "buzz"),
)
for instance, term in instances:
route = f"/{term}" if isinstance(instance, Sanic) else ""
@instance.get(route, strict_slashes=True)
def get_without(request):
return text(f"{term}_without")
@instance.get(f"{route}/", strict_slashes=True)
def get_with(request):
return text(f"{term}_with")
@instance.post(route, strict_slashes=True)
def post_without(request):
return text(f"{term}_without")
@instance.post(f"{route}/", strict_slashes=True)
def post_with(request):
return text(f"{term}_with")
app.blueprint(bar)
app.blueprint(baz)
app.blueprint(fizz)
app.blueprint(buzz)
for _, term in instances:
_, response = app.test_client.get(f"/{term}")
assert response.status == 200
assert response.text == f"{term}_without"
_, response = app.test_client.get(f"/{term}/")
assert response.status == 200
assert response.text == f"{term}_with"
_, response = app.test_client.post(f"/{term}")
assert response.status == 200
assert response.text == f"{term}_without"
_, response = app.test_client.post(f"/{term}/")
assert response.status == 200
assert response.text == f"{term}_with"
2021-12-21 20:56:12 +00:00
def test_added_route_ctx_kwargs(app):
@app.route("/", ctx_foo="foo", ctx_bar=99)
async def handler(request: Request):
return empty()
request, _ = app.test_client.get("/")
assert request.route.ctx.foo == "foo"
assert request.route.ctx.bar == 99
def test_added_bad_route_kwargs(app):
message = "Unexpected keyword arguments: foo, bar"
with pytest.raises(TypeError, match=message):
@app.route("/", foo="foo", bar=99)
async def handler(request: Request):
...
@pytest.mark.asyncio
async def test_added_callable_route_ctx_kwargs(app):
def foo(*args, **kwargs):
return "foo"
async def bar(*args, **kwargs):
return 99
@app.route("/", ctx_foo=foo, ctx_bar=bar)
async def handler(request: Request):
return empty()
request, _ = await app.asgi_client.get("/")
assert request.route.ctx.foo() == "foo"
assert await request.route.ctx.bar() == 99