Resolve some more tests
This commit is contained in:
parent
a434ffa8b7
commit
c08b153cee
142
sanic/app.py
142
sanic/app.py
|
@ -16,7 +16,7 @@ from urllib.parse import urlencode, urlunparse
|
|||
|
||||
from sanic_routing.route import Route
|
||||
|
||||
from sanic import reloader_helpers, websocket
|
||||
from sanic import reloader_helpers
|
||||
from sanic.asgi import ASGIApp
|
||||
from sanic.base import BaseSanic
|
||||
from sanic.blueprint_group import BlueprintGroup
|
||||
|
@ -114,6 +114,8 @@ class Sanic(BaseSanic):
|
|||
if self.config.REGISTER:
|
||||
self.__class__.register_app(self)
|
||||
|
||||
self.router.ctx.app = self
|
||||
|
||||
@property
|
||||
def loop(self):
|
||||
"""Synonymous with asyncio.get_event_loop().
|
||||
|
@ -230,7 +232,6 @@ class Sanic(BaseSanic):
|
|||
websocket = params.pop("websocket", False)
|
||||
subprotocols = params.pop("subprotocols", None)
|
||||
|
||||
|
||||
if websocket:
|
||||
self.enable_websocket()
|
||||
websocket_handler = partial(
|
||||
|
@ -294,6 +295,12 @@ class Sanic(BaseSanic):
|
|||
else:
|
||||
self.blueprints[blueprint.name] = blueprint
|
||||
self._blueprint_order.append(blueprint)
|
||||
|
||||
if (
|
||||
self.strict_slashes is not None
|
||||
and blueprint.strict_slashes is None
|
||||
):
|
||||
blueprint.strict_slashes = self.strict_slashes
|
||||
blueprint.register(self, options)
|
||||
|
||||
def url_for(self, view_name: str, **kwargs):
|
||||
|
@ -319,30 +326,28 @@ class Sanic(BaseSanic):
|
|||
# find the route by the supplied view name
|
||||
kw: Dict[str, str] = {}
|
||||
# special static files url_for
|
||||
if view_name == "static":
|
||||
kw.update(name=kwargs.pop("name", "static"))
|
||||
elif view_name.endswith(".static"): # blueprint.static
|
||||
kwargs.pop("name", None)
|
||||
|
||||
if "." not in view_name:
|
||||
view_name = f"{self.name}.{view_name}"
|
||||
|
||||
if view_name.endswith(".static"):
|
||||
name = kwargs.pop("name", None)
|
||||
if name:
|
||||
view_name = view_name.replace("static", name)
|
||||
kw.update(name=view_name)
|
||||
|
||||
uri, route = self.router.find_route_by_view_name(view_name, **kw)
|
||||
if not (uri and route):
|
||||
route = self.router.find_route_by_view_name(view_name, **kw)
|
||||
if not route:
|
||||
raise URLBuildError(
|
||||
f"Endpoint with name `{view_name}` was not found"
|
||||
)
|
||||
|
||||
# If the route has host defined, split that off
|
||||
# TODO: Retain netloc and path separately in Route objects
|
||||
host = uri.find("/")
|
||||
if host > 0:
|
||||
host, uri = uri[:host], uri[host:]
|
||||
else:
|
||||
host = None
|
||||
uri = route.path
|
||||
|
||||
if view_name == "static" or view_name.endswith(".static"):
|
||||
filename = kwargs.pop("filename", None)
|
||||
if getattr(route.ctx, "static", None):
|
||||
filename = kwargs.pop("filename", "")
|
||||
# it's static folder
|
||||
if "<file_uri:" in uri:
|
||||
if "file_uri" in uri:
|
||||
folder_ = uri.split("<file_uri:", 1)[0]
|
||||
if folder_.endswith("/"):
|
||||
folder_ = folder_[:-1]
|
||||
|
@ -350,22 +355,36 @@ class Sanic(BaseSanic):
|
|||
if filename.startswith("/"):
|
||||
filename = filename[1:]
|
||||
|
||||
uri = f"{folder_}/{filename}"
|
||||
kwargs["file_uri"] = filename
|
||||
|
||||
if uri != "/" and uri.endswith("/"):
|
||||
uri = uri[:-1]
|
||||
|
||||
out = uri
|
||||
if not uri.startswith("/"):
|
||||
uri = f"/{uri}"
|
||||
|
||||
# find all the parameters we will need to build in the URL
|
||||
# matched_params = re.findall(self.router.parameter_pattern, uri)
|
||||
out = uri
|
||||
|
||||
# _method is only a placeholder now, don't know how to support it
|
||||
kwargs.pop("_method", None)
|
||||
anchor = kwargs.pop("_anchor", "")
|
||||
# _external need SERVER_NAME in config or pass _server arg
|
||||
external = kwargs.pop("_external", False)
|
||||
host = kwargs.pop("_host", None)
|
||||
external = kwargs.pop("_external", False) or bool(host)
|
||||
scheme = kwargs.pop("_scheme", "")
|
||||
if route.ctx.hosts and external:
|
||||
if not host and len(route.ctx.hosts) > 1:
|
||||
raise ValueError(
|
||||
f"Host is ambiguous: {', '.join(route.ctx.hosts)}"
|
||||
)
|
||||
elif host and host not in route.ctx.hosts:
|
||||
raise ValueError(
|
||||
f"Requested host ({host}) is not available for this "
|
||||
f"route: {route.ctx.hosts}"
|
||||
)
|
||||
elif not host:
|
||||
host = list(route.ctx.hosts)[0]
|
||||
|
||||
if scheme and not external:
|
||||
raise ValueError("When specifying _scheme, _external must be True")
|
||||
|
||||
|
@ -383,45 +402,49 @@ class Sanic(BaseSanic):
|
|||
if "://" in netloc[:8]:
|
||||
netloc = netloc.split("://", 1)[-1]
|
||||
|
||||
# for match in matched_params:
|
||||
# find all the parameters we will need to build in the URL
|
||||
# matched_params = re.findall(self.router.parameter_pattern, uri)
|
||||
route.finalize_params()
|
||||
for params in route.params.values():
|
||||
# name, _type, pattern = self.router.parse_parameter_string(match)
|
||||
# # we only want to match against each individual parameter
|
||||
# specific_pattern = f"^{pattern}$"
|
||||
# supplied_param = None
|
||||
# we only want to match against each individual parameter
|
||||
|
||||
# if name in kwargs:
|
||||
# supplied_param = kwargs.get(name)
|
||||
# del kwargs[name]
|
||||
# else:
|
||||
# raise URLBuildError(
|
||||
# f"Required parameter `{name}` was not passed to url_for"
|
||||
# )
|
||||
for idx, param_info in enumerate(params):
|
||||
try:
|
||||
supplied_param = str(kwargs.pop(param_info.name))
|
||||
except KeyError:
|
||||
raise URLBuildError(
|
||||
f"Required parameter `{param_info.name}` was not "
|
||||
"passed to url_for"
|
||||
)
|
||||
|
||||
# supplied_param = str(supplied_param)
|
||||
# # determine if the parameter supplied by the caller passes the test
|
||||
# # in the URL
|
||||
# passes_pattern = re.match(specific_pattern, supplied_param)
|
||||
# determine if the parameter supplied by the caller
|
||||
# passes the test in the URL
|
||||
if param_info.pattern:
|
||||
passes_pattern = param_info.pattern.match(supplied_param)
|
||||
if not passes_pattern:
|
||||
if idx + 1 == len(params):
|
||||
if param_info.cast != str:
|
||||
msg = (
|
||||
f'Value "{supplied_param}" '
|
||||
f"for parameter `{param_info.name}` does "
|
||||
"not match pattern for type "
|
||||
f"`{param_info.cast.__name__}`: "
|
||||
f"{param_info.pattern.pattern}"
|
||||
)
|
||||
else:
|
||||
msg = (
|
||||
f'Value "{supplied_param}" for parameter '
|
||||
f"`{param_info.name}` does not satisfy "
|
||||
f"pattern {param_info.pattern.pattern}"
|
||||
)
|
||||
raise URLBuildError(msg)
|
||||
else:
|
||||
continue
|
||||
|
||||
# if not passes_pattern:
|
||||
# if _type != str:
|
||||
# type_name = _type.__name__
|
||||
|
||||
# msg = (
|
||||
# f'Value "{supplied_param}" '
|
||||
# f"for parameter `{name}` does not "
|
||||
# f"match pattern for type `{type_name}`: {pattern}"
|
||||
# )
|
||||
# else:
|
||||
# msg = (
|
||||
# f'Value "{supplied_param}" for parameter `{name}` '
|
||||
# f"does not satisfy pattern {pattern}"
|
||||
# )
|
||||
# raise URLBuildError(msg)
|
||||
|
||||
# # replace the parameter in the URL with the supplied value
|
||||
# replacement_regex = f"(<{name}.*?>)"
|
||||
|
||||
# out = re.sub(replacement_regex, supplied_param, out)
|
||||
# replace the parameter in the URL with the supplied value
|
||||
replacement_regex = f"(<{param_info.name}.*?>)"
|
||||
out = re.sub(replacement_regex, supplied_param, out)
|
||||
|
||||
# parse the remainder of the keyword arguments into a querystring
|
||||
query_string = urlencode(kwargs, doseq=True) if kwargs else ""
|
||||
|
@ -845,9 +868,6 @@ class Sanic(BaseSanic):
|
|||
await result
|
||||
|
||||
async def _run_request_middleware(self, request, request_name=None):
|
||||
print(self.request_middleware)
|
||||
print(self.named_request_middleware)
|
||||
print(request_name)
|
||||
# The if improves speed. I don't know why
|
||||
named_middleware = self.named_request_middleware.get(
|
||||
request_name, deque()
|
||||
|
|
|
@ -109,22 +109,35 @@ class Blueprint(BaseSanic):
|
|||
# Prepend the blueprint URI prefix if available
|
||||
uri = url_prefix + future.uri if url_prefix else future.uri
|
||||
|
||||
strict_slashes = (
|
||||
self.strict_slashes
|
||||
if future.strict_slashes is None
|
||||
and self.strict_slashes is not None
|
||||
else future.strict_slashes
|
||||
)
|
||||
|
||||
print(uri, strict_slashes)
|
||||
|
||||
apply_route = FutureRoute(
|
||||
future.handler,
|
||||
uri[1:] if uri.startswith("//") else uri,
|
||||
future.methods,
|
||||
future.host or self.host,
|
||||
future.strict_slashes,
|
||||
strict_slashes,
|
||||
future.stream,
|
||||
future.version or self.version,
|
||||
future.name,
|
||||
future.ignore_body,
|
||||
future.websocket,
|
||||
future.subprotocols,
|
||||
future.unquote,
|
||||
future.static,
|
||||
)
|
||||
|
||||
route = app._apply_route(apply_route)
|
||||
operation = routes.extend if isinstance(route, list) else routes.append
|
||||
operation = (
|
||||
routes.extend if isinstance(route, list) else routes.append
|
||||
)
|
||||
operation(route)
|
||||
|
||||
# Static Files
|
||||
|
@ -149,6 +162,3 @@ class Blueprint(BaseSanic):
|
|||
# Event listeners
|
||||
for listener in self._future_listeners:
|
||||
app._apply_listener(listener)
|
||||
|
||||
def _generate_name(self, handler, name: str) -> str:
|
||||
return f"{self.name}.{name or handler.__name__}"
|
||||
|
|
|
@ -36,6 +36,8 @@ class RouteMixin:
|
|||
apply=True,
|
||||
subprotocols=None,
|
||||
websocket=False,
|
||||
unquote=False,
|
||||
static=False,
|
||||
):
|
||||
"""Create a blueprint route from a decorated function.
|
||||
|
||||
|
@ -74,21 +76,28 @@ class RouteMixin:
|
|||
nonlocal ignore_body
|
||||
nonlocal subprotocols
|
||||
nonlocal websocket
|
||||
nonlocal static
|
||||
|
||||
if isinstance(handler, tuple):
|
||||
# if a handler fn is already wrapped in a route, the handler
|
||||
# variable will be a tuple of (existing routes, handler fn)
|
||||
_, handler = handler
|
||||
|
||||
# TODO:
|
||||
# - THink this thru.... do we want all routes namespaced?
|
||||
# -
|
||||
name = self._generate_name(handler, name)
|
||||
name = self._generate_name(name, handler)
|
||||
|
||||
if isinstance(host, str):
|
||||
host = frozenset([host])
|
||||
elif host and not isinstance(host, frozenset):
|
||||
try:
|
||||
host = frozenset(host)
|
||||
except TypeError:
|
||||
raise ValueError(
|
||||
"Expected either string or Iterable of host strings, "
|
||||
"not %s" % host
|
||||
)
|
||||
|
||||
if isinstance(subprotocols, (list, tuple, set)):
|
||||
subprotocols = frozenset(subprotocols)
|
||||
|
||||
route = FutureRoute(
|
||||
handler,
|
||||
|
@ -102,6 +111,8 @@ class RouteMixin:
|
|||
ignore_body,
|
||||
websocket,
|
||||
subprotocols,
|
||||
unquote,
|
||||
static,
|
||||
)
|
||||
|
||||
self._future_routes.add(route)
|
||||
|
@ -499,12 +510,16 @@ class RouteMixin:
|
|||
:rtype: List[sanic.router.Route]
|
||||
"""
|
||||
|
||||
if not name.startswith(self.name + "."):
|
||||
name = f"{self.name}.{name}"
|
||||
name = self._generate_name(name)
|
||||
|
||||
if strict_slashes is None and self.strict_slashes is not None:
|
||||
strict_slashes = self.strict_slashes
|
||||
|
||||
if not isinstance(file_or_directory, (str, bytes, PurePath)):
|
||||
raise ValueError(
|
||||
f"Static route must be a valid path, not {file_or_directory}"
|
||||
)
|
||||
|
||||
static = FutureStatic(
|
||||
uri,
|
||||
file_or_directory,
|
||||
|
@ -522,5 +537,25 @@ class RouteMixin:
|
|||
if apply:
|
||||
self._apply_static(static)
|
||||
|
||||
def _generate_name(self, handler, name: str) -> str:
|
||||
return name or handler.__name__
|
||||
def _generate_name(self, *objects) -> str:
|
||||
name = None
|
||||
for obj in objects:
|
||||
if obj:
|
||||
if isinstance(obj, str):
|
||||
name = obj
|
||||
break
|
||||
|
||||
try:
|
||||
name = obj.__name__
|
||||
except AttributeError:
|
||||
continue
|
||||
else:
|
||||
break
|
||||
|
||||
if not name:
|
||||
raise Exception("...")
|
||||
|
||||
if not name.startswith(f"{self.name}."):
|
||||
name = f"{self.name}.{name}"
|
||||
|
||||
return name
|
||||
|
|
|
@ -15,6 +15,8 @@ FutureRoute = namedtuple(
|
|||
"ignore_body",
|
||||
"websocket",
|
||||
"subprotocols",
|
||||
"unquote",
|
||||
"static",
|
||||
],
|
||||
)
|
||||
FutureListener = namedtuple("FutureListener", ["listener", "event"])
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
from functools import lru_cache
|
||||
from typing import FrozenSet, Iterable, List, Optional, Union
|
||||
|
||||
from sanic_routing import BaseRouter, route
|
||||
from sanic_routing import BaseRouter
|
||||
from sanic_routing.exceptions import NoMethod
|
||||
from sanic_routing.exceptions import NotFound as RoutingNotFound
|
||||
from sanic_routing.route import Route
|
||||
|
@ -37,7 +37,7 @@ class Router(BaseRouter):
|
|||
route, handler, params = self.resolve(
|
||||
path=request.path,
|
||||
method=request.method,
|
||||
extra={"host": request.headers.get("host")}
|
||||
extra={"host": request.headers.get("host")},
|
||||
)
|
||||
except RoutingNotFound as e:
|
||||
raise NotFound("Requested URL {} not found".format(e.path))
|
||||
|
@ -75,6 +75,8 @@ class Router(BaseRouter):
|
|||
ignore_body: bool = False,
|
||||
version: Union[str, float, int] = None,
|
||||
name: Optional[str] = None,
|
||||
unquote: bool = False,
|
||||
static: bool = False,
|
||||
) -> Union[Route, List[Route]]:
|
||||
"""
|
||||
Add a handler to the router
|
||||
|
@ -118,6 +120,7 @@ class Router(BaseRouter):
|
|||
methods=methods,
|
||||
name=name,
|
||||
strict=strict_slashes,
|
||||
unquote=unquote,
|
||||
)
|
||||
|
||||
if isinstance(host, str):
|
||||
|
@ -134,6 +137,8 @@ class Router(BaseRouter):
|
|||
route = super().add(**params)
|
||||
route.ctx.ignore_body = ignore_body
|
||||
route.ctx.stream = stream
|
||||
route.ctx.hosts = hosts
|
||||
route.ctx.static = static
|
||||
|
||||
routes.append(route)
|
||||
|
||||
|
@ -168,15 +173,19 @@ class Router(BaseRouter):
|
|||
:return: tuple containing (uri, Route)
|
||||
"""
|
||||
if not view_name:
|
||||
return None, None
|
||||
return None
|
||||
|
||||
if view_name == "static" or view_name.endswith(".static"):
|
||||
looking_for = f"_static_{name}"
|
||||
route = self.name_index.get(looking_for)
|
||||
else:
|
||||
route = self.name_index.get(view_name)
|
||||
name = self.ctx.app._generate_name(view_name)
|
||||
route = self.name_index.get(name)
|
||||
|
||||
if not route:
|
||||
return None, None
|
||||
return None
|
||||
|
||||
return route.path, route
|
||||
return route
|
||||
|
||||
@property
|
||||
def routes_all(self):
|
||||
return {
|
||||
**self.static_routes,
|
||||
**self.dynamic_routes,
|
||||
}
|
||||
|
|
|
@ -6,6 +6,8 @@ from re import sub
|
|||
from time import gmtime, strftime
|
||||
from urllib.parse import unquote
|
||||
|
||||
from sanic_routing.patterns import REGEX_TYPES
|
||||
|
||||
from sanic.compat import stat_async
|
||||
from sanic.exceptions import (
|
||||
ContentRangeError,
|
||||
|
@ -157,11 +159,11 @@ def register(
|
|||
# If we're not trying to match a file directly,
|
||||
# serve from the folder
|
||||
if not path.isfile(file_or_directory):
|
||||
uri += "<file_uri:" + static.pattern + ">"
|
||||
uri += "/<file_uri:path>"
|
||||
|
||||
# special prefix for static files
|
||||
if not static.name.startswith("_static_"):
|
||||
name = f"_static_{static.name}"
|
||||
# if not static.name.startswith("_static_"):
|
||||
# name = f"_static_{static.name}"
|
||||
|
||||
_handler = wraps(_static_request_handler)(
|
||||
partial(
|
||||
|
@ -174,11 +176,13 @@ def register(
|
|||
)
|
||||
)
|
||||
|
||||
_routes, _ = app.route(
|
||||
route, _ = app.route(
|
||||
uri=uri,
|
||||
methods=["GET", "HEAD"],
|
||||
name=name,
|
||||
host=static.host,
|
||||
strict_slashes=static.strict_slashes,
|
||||
static=True,
|
||||
)(_handler)
|
||||
return _routes
|
||||
|
||||
return route
|
||||
|
|
|
@ -304,24 +304,18 @@ async def test_cookie_customization(app):
|
|||
_, response = await app.asgi_client.get("/cookie")
|
||||
|
||||
CookieDef = namedtuple("CookieDef", ("value", "httponly"))
|
||||
Cookie = namedtuple("Cookie", ("domain", "path", "value", "httponly"))
|
||||
cookie_map = {
|
||||
"test": CookieDef("Cookie1", True),
|
||||
"c2": CookieDef("Cookie2", False),
|
||||
}
|
||||
|
||||
cookies = {
|
||||
c.name: Cookie(c.domain, c.path, c.value, "HttpOnly" in c._rest.keys())
|
||||
for c in response.cookies.jar
|
||||
}
|
||||
|
||||
for name, definition in cookie_map.items():
|
||||
cookie = cookies.get(name)
|
||||
cookie = response.cookies.get(name)
|
||||
assert cookie
|
||||
assert cookie.value == definition.value
|
||||
assert cookie.domain == "mockserver.local"
|
||||
assert cookie.path == "/"
|
||||
assert cookie.httponly == definition.httponly
|
||||
assert cookie.get("domain") == "mockserver.local"
|
||||
assert cookie.get("path") == "/"
|
||||
assert cookie.get("httponly", False) == definition.httponly
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
@ -197,7 +197,12 @@ def test_several_bp_with_url_prefix(app):
|
|||
|
||||
|
||||
def test_bp_with_host(app):
|
||||
bp = Blueprint("test_bp_host", url_prefix="/test1", host="example.com")
|
||||
bp = Blueprint(
|
||||
"test_bp_host",
|
||||
url_prefix="/test1",
|
||||
host="example.com",
|
||||
strict_slashes=True,
|
||||
)
|
||||
|
||||
@bp.route("/")
|
||||
def handler1(request):
|
||||
|
@ -209,18 +214,29 @@ def test_bp_with_host(app):
|
|||
|
||||
app.blueprint(bp)
|
||||
headers = {"Host": "example.com"}
|
||||
app.router.finalize()
|
||||
|
||||
request, response = app.test_client.get("/test1/", headers=headers)
|
||||
assert response.body == b"Hello"
|
||||
|
||||
headers = {"Host": "sub.example.com"}
|
||||
request, response = app.test_client.get("/test1/", headers=headers)
|
||||
print(app.router.find_route_src)
|
||||
assert response.body == b"Hello subdomain!"
|
||||
|
||||
|
||||
def test_several_bp_with_host(app):
|
||||
bp = Blueprint("test_text", url_prefix="/test", host="example.com")
|
||||
bp2 = Blueprint("test_text2", url_prefix="/test", host="sub.example.com")
|
||||
bp = Blueprint(
|
||||
"test_text",
|
||||
url_prefix="/test",
|
||||
host="example.com",
|
||||
strict_slashes=True,
|
||||
)
|
||||
bp2 = Blueprint(
|
||||
"test_text2",
|
||||
url_prefix="/test",
|
||||
host="sub.example.com",
|
||||
strict_slashes=True,
|
||||
)
|
||||
|
||||
@bp.route("/")
|
||||
def handler(request):
|
||||
|
@ -449,6 +465,7 @@ def test_bp_exception_handler(app):
|
|||
|
||||
|
||||
def test_bp_listeners(app):
|
||||
app.route("/")(lambda x: x)
|
||||
blueprint = Blueprint("test_middleware")
|
||||
|
||||
order = []
|
||||
|
@ -723,7 +740,8 @@ def test_blueprint_middleware_with_args(app: Sanic):
|
|||
|
||||
|
||||
@pytest.mark.parametrize("file_name", ["test.file"])
|
||||
def test_static_blueprint_name(app: Sanic, static_file_directory, file_name):
|
||||
def test_static_blueprint_name(static_file_directory, file_name):
|
||||
app = Sanic("app")
|
||||
current_file = inspect.getfile(inspect.currentframe())
|
||||
with open(current_file, "rb") as file:
|
||||
file.read()
|
||||
|
@ -738,9 +756,6 @@ def test_static_blueprint_name(app: Sanic, static_file_directory, file_name):
|
|||
)
|
||||
|
||||
app.blueprint(bp)
|
||||
print(app.router.name_index)
|
||||
print(app.router.static_routes)
|
||||
print(app.router.dynamic_routes)
|
||||
|
||||
uri = app.url_for("static", name="static.testing")
|
||||
assert uri == "/static/test.file"
|
||||
|
@ -841,18 +856,19 @@ def test_duplicate_blueprint(app):
|
|||
)
|
||||
|
||||
|
||||
def test_strict_slashes_behavior_adoption(app):
|
||||
def test_strict_slashes_behavior_adoption():
|
||||
app = Sanic("app")
|
||||
app.strict_slashes = True
|
||||
bp = Blueprint("bp")
|
||||
bp2 = Blueprint("bp2", strict_slashes=False)
|
||||
|
||||
@app.get("/test")
|
||||
def handler_test(request):
|
||||
return text("Test")
|
||||
|
||||
assert app.test_client.get("/test")[1].status == 200
|
||||
assert app.test_client.get("/test/")[1].status == 404
|
||||
|
||||
app.router.finalized = False
|
||||
bp = Blueprint("bp")
|
||||
@app.get("/f1", strict_slashes=False)
|
||||
def f1(request):
|
||||
return text("f1")
|
||||
|
||||
@bp.get("/one", strict_slashes=False)
|
||||
def one(request):
|
||||
|
@ -862,7 +878,15 @@ def test_strict_slashes_behavior_adoption(app):
|
|||
def second(request):
|
||||
return text("second")
|
||||
|
||||
@bp2.get("/third")
|
||||
def third(request):
|
||||
return text("third")
|
||||
|
||||
app.blueprint(bp)
|
||||
app.blueprint(bp2)
|
||||
|
||||
assert app.test_client.get("/test")[1].status == 200
|
||||
assert app.test_client.get("/test/")[1].status == 404
|
||||
|
||||
assert app.test_client.get("/one")[1].status == 200
|
||||
assert app.test_client.get("/one/")[1].status == 200
|
||||
|
@ -870,19 +894,8 @@ def test_strict_slashes_behavior_adoption(app):
|
|||
assert app.test_client.get("/second")[1].status == 200
|
||||
assert app.test_client.get("/second/")[1].status == 404
|
||||
|
||||
bp2 = Blueprint("bp2", strict_slashes=False)
|
||||
|
||||
@bp2.get("/third")
|
||||
def third(request):
|
||||
return text("third")
|
||||
|
||||
app.blueprint(bp2)
|
||||
assert app.test_client.get("/third")[1].status == 200
|
||||
assert app.test_client.get("/third/")[1].status == 200
|
||||
|
||||
@app.get("/f1", strict_slashes=False)
|
||||
def f1(request):
|
||||
return text("f1")
|
||||
|
||||
assert app.test_client.get("/f1")[1].status == 200
|
||||
assert app.test_client.get("/f1/")[1].status == 200
|
||||
|
|
1296
tests/test_routes.py
1296
tests/test_routes.py
File diff suppressed because it is too large
Load Diff
|
@ -106,6 +106,7 @@ def test_static_file_bytes(app, static_file_directory, file_name):
|
|||
[dict(), list(), object()],
|
||||
)
|
||||
def test_static_file_invalid_path(app, static_file_directory, file_name):
|
||||
app.route("/")(lambda x: x)
|
||||
with pytest.raises(ValueError):
|
||||
app.static("/testing.file", file_name)
|
||||
request, response = app.test_client.get("/testing.file")
|
||||
|
|
|
@ -112,22 +112,21 @@ def test_fails_if_endpoint_not_found(app):
|
|||
def test_fails_url_build_if_param_not_passed(app):
|
||||
url = "/"
|
||||
|
||||
for letter in string.ascii_letters:
|
||||
for letter in string.ascii_lowercase:
|
||||
url += f"<{letter}>/"
|
||||
|
||||
@app.route(url)
|
||||
def fail(request):
|
||||
return text("this should fail")
|
||||
|
||||
fail_args = list(string.ascii_letters)
|
||||
fail_args = list(string.ascii_lowercase)
|
||||
fail_args.pop()
|
||||
|
||||
fail_kwargs = {l: l for l in fail_args}
|
||||
|
||||
with pytest.raises(URLBuildError) as e:
|
||||
app.url_for("fail", **fail_kwargs)
|
||||
|
||||
assert "Required parameter `Z` was not passed to url_for" in str(e.value)
|
||||
assert e.match("Required parameter `z` was not passed to url_for")
|
||||
|
||||
|
||||
def test_fails_url_build_if_params_not_passed(app):
|
||||
|
@ -137,8 +136,7 @@ def test_fails_url_build_if_params_not_passed(app):
|
|||
|
||||
with pytest.raises(ValueError) as e:
|
||||
app.url_for("fail", _scheme="http")
|
||||
|
||||
assert str(e.value) == "When specifying _scheme, _external must be True"
|
||||
assert e.match("When specifying _scheme, _external must be True")
|
||||
|
||||
|
||||
COMPLEX_PARAM_URL = (
|
||||
|
@ -168,7 +166,7 @@ def test_fails_with_int_message(app):
|
|||
|
||||
expected_error = (
|
||||
r'Value "not_int" for parameter `foo` '
|
||||
r"does not match pattern for type `int`: -?\d+"
|
||||
r"does not match pattern for type `int`: ^-?\d+"
|
||||
)
|
||||
assert str(e.value) == expected_error
|
||||
|
||||
|
@ -199,14 +197,11 @@ def test_fails_with_two_letter_string_message(app):
|
|||
|
||||
with pytest.raises(URLBuildError) as e:
|
||||
app.url_for("fail", **failing_kwargs)
|
||||
|
||||
expected_error = (
|
||||
e.match(
|
||||
'Value "foobar" for parameter `two_letter_string` '
|
||||
"does not satisfy pattern [A-z]{2}"
|
||||
"does not satisfy pattern ^[A-z]{2}$"
|
||||
)
|
||||
|
||||
assert str(e.value) == expected_error
|
||||
|
||||
|
||||
def test_fails_with_number_message(app):
|
||||
@app.route(COMPLEX_PARAM_URL)
|
||||
|
@ -218,14 +213,11 @@ def test_fails_with_number_message(app):
|
|||
|
||||
with pytest.raises(URLBuildError) as e:
|
||||
app.url_for("fail", **failing_kwargs)
|
||||
|
||||
expected_error = (
|
||||
e.match(
|
||||
'Value "foo" for parameter `some_number` '
|
||||
r"does not match pattern for type `float`: -?(?:\d+(?:\.\d*)?|\.\d+)"
|
||||
r"does not match pattern for type `float`: ^-?(?:\d+(?:\.\d*)?|\.\d+)$"
|
||||
)
|
||||
|
||||
assert str(e.value) == expected_error
|
||||
|
||||
|
||||
@pytest.mark.parametrize("number", [3, -3, 13.123, -13.123])
|
||||
def test_passes_with_negative_number_message(app, number):
|
||||
|
@ -273,11 +265,11 @@ def blueprint_app(app):
|
|||
return text(f"foo from first : {param}")
|
||||
|
||||
@second_print.route("/foo") # noqa
|
||||
def foo(request):
|
||||
def bar(request):
|
||||
return text("foo from second")
|
||||
|
||||
@second_print.route("/foo/<param>") # noqa
|
||||
def foo_with_param(request, param):
|
||||
def bar_with_param(request, param):
|
||||
return text(f"foo from second : {param}")
|
||||
|
||||
app.blueprint(first_print)
|
||||
|
@ -290,7 +282,7 @@ def test_blueprints_are_named_correctly(blueprint_app):
|
|||
first_url = blueprint_app.url_for("first.foo")
|
||||
assert first_url == "/first/foo"
|
||||
|
||||
second_url = blueprint_app.url_for("second.foo")
|
||||
second_url = blueprint_app.url_for("second.bar")
|
||||
assert second_url == "/second/foo"
|
||||
|
||||
|
||||
|
@ -298,7 +290,7 @@ def test_blueprints_work_with_params(blueprint_app):
|
|||
first_url = blueprint_app.url_for("first.foo_with_param", param="bar")
|
||||
assert first_url == "/first/foo/bar"
|
||||
|
||||
second_url = blueprint_app.url_for("second.foo_with_param", param="bar")
|
||||
second_url = blueprint_app.url_for("second.bar_with_param", param="bar")
|
||||
assert second_url == "/second/foo/bar"
|
||||
|
||||
|
||||
|
|
|
@ -1,18 +1,18 @@
|
|||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from sanic_testing.testing import SanicTestClient
|
||||
|
||||
from sanic.blueprints import Blueprint
|
||||
|
||||
|
||||
def test_routes_with_host(app):
|
||||
@app.route("/")
|
||||
@app.route("/", name="hostindex", host="example.com")
|
||||
@app.route("/path", name="hostpath", host="path.example.com")
|
||||
def index(request):
|
||||
pass
|
||||
|
||||
assert app.url_for("index") == "/"
|
||||
assert app.url_for("hostindex") == "/"
|
||||
assert app.url_for("hostpath") == "/path"
|
||||
assert app.url_for("hostindex", _external=True) == "http://example.com/"
|
||||
|
@ -22,6 +22,27 @@ def test_routes_with_host(app):
|
|||
)
|
||||
|
||||
|
||||
def test_routes_with_multiple_hosts(app):
|
||||
@app.route("/", name="hostindex", host=["example.com", "path.example.com"])
|
||||
def index(request):
|
||||
pass
|
||||
|
||||
assert app.url_for("hostindex") == "/"
|
||||
assert (
|
||||
app.url_for("hostindex", _host="example.com") == "http://example.com/"
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError) as e:
|
||||
assert app.url_for("hostindex", _external=True)
|
||||
assert str(e.value).startswith("Host is ambiguous")
|
||||
|
||||
with pytest.raises(ValueError) as e:
|
||||
assert app.url_for("hostindex", _host="unknown.com")
|
||||
assert str(e.value).startswith(
|
||||
"Requested host (unknown.com) is not available for this route"
|
||||
)
|
||||
|
||||
|
||||
def test_websocket_bp_route_name(app):
|
||||
"""Tests that blueprint websocket route is named."""
|
||||
event = asyncio.Event()
|
||||
|
@ -63,3 +84,7 @@ def test_websocket_bp_route_name(app):
|
|||
|
||||
uri = app.url_for("test_bp.foobar_3")
|
||||
assert uri == "/bp/route3"
|
||||
|
||||
|
||||
# TODO: add test with a route with multiple hosts
|
||||
# TODO: add test with a route with _host in url_for
|
||||
|
|
|
@ -3,6 +3,7 @@ import os
|
|||
|
||||
import pytest
|
||||
|
||||
from sanic import Sanic
|
||||
from sanic.blueprints import Blueprint
|
||||
|
||||
|
||||
|
@ -26,9 +27,15 @@ def get_file_content(static_file_directory, file_name):
|
|||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"file_name", ["test.file", "decode me.txt", "python.png"]
|
||||
"file_name",
|
||||
[
|
||||
"test.file",
|
||||
"decode me.txt",
|
||||
"python.png",
|
||||
],
|
||||
)
|
||||
def test_static_file(app, static_file_directory, file_name):
|
||||
def test_static_file(static_file_directory, file_name):
|
||||
app = Sanic("qq")
|
||||
app.static(
|
||||
"/testing.file", get_file_path(static_file_directory, file_name)
|
||||
)
|
||||
|
@ -38,6 +45,8 @@ def test_static_file(app, static_file_directory, file_name):
|
|||
name="testing_file",
|
||||
)
|
||||
|
||||
app.router.finalize()
|
||||
|
||||
uri = app.url_for("static")
|
||||
uri2 = app.url_for("static", filename="any")
|
||||
uri3 = app.url_for("static", name="static", filename="any")
|
||||
|
@ -46,10 +55,14 @@ def test_static_file(app, static_file_directory, file_name):
|
|||
assert uri == uri2
|
||||
assert uri2 == uri3
|
||||
|
||||
app.router.reset()
|
||||
|
||||
request, response = app.test_client.get(uri)
|
||||
assert response.status == 200
|
||||
assert response.body == get_file_content(static_file_directory, file_name)
|
||||
|
||||
app.router.reset()
|
||||
|
||||
bp = Blueprint("test_bp_static", url_prefix="/bp")
|
||||
|
||||
bp.static("/testing.file", get_file_path(static_file_directory, file_name))
|
||||
|
@ -61,19 +74,14 @@ def test_static_file(app, static_file_directory, file_name):
|
|||
|
||||
app.blueprint(bp)
|
||||
|
||||
uri = app.url_for("static", name="test_bp_static.static")
|
||||
uri2 = app.url_for("static", name="test_bp_static.static", filename="any")
|
||||
uri3 = app.url_for("test_bp_static.static")
|
||||
uri4 = app.url_for("test_bp_static.static", name="any")
|
||||
uri5 = app.url_for("test_bp_static.static", filename="any")
|
||||
uri6 = app.url_for("test_bp_static.static", name="any", filename="any")
|
||||
uris = [
|
||||
app.url_for("static", name="test_bp_static.static"),
|
||||
app.url_for("static", name="test_bp_static.static", filename="any"),
|
||||
app.url_for("test_bp_static.static"),
|
||||
app.url_for("test_bp_static.static", filename="any"),
|
||||
]
|
||||
|
||||
assert uri == "/bp/testing.file"
|
||||
assert uri == uri2
|
||||
assert uri2 == uri3
|
||||
assert uri3 == uri4
|
||||
assert uri4 == uri5
|
||||
assert uri5 == uri6
|
||||
assert all(uri == "/bp/testing.file" for uri in uris)
|
||||
|
||||
request, response = app.test_client.get(uri)
|
||||
assert response.status == 200
|
||||
|
@ -112,7 +120,9 @@ def test_static_file(app, static_file_directory, file_name):
|
|||
|
||||
@pytest.mark.parametrize("file_name", ["test.file", "decode me.txt"])
|
||||
@pytest.mark.parametrize("base_uri", ["/static", "", "/dir"])
|
||||
def test_static_directory(app, file_name, base_uri, static_file_directory):
|
||||
def test_static_directory(file_name, base_uri, static_file_directory):
|
||||
app = Sanic("base")
|
||||
|
||||
app.static(base_uri, static_file_directory)
|
||||
base_uri2 = base_uri + "/2"
|
||||
app.static(base_uri2, static_file_directory, name="uploads")
|
||||
|
@ -141,6 +151,8 @@ def test_static_directory(app, file_name, base_uri, static_file_directory):
|
|||
|
||||
bp.static(base_uri, static_file_directory)
|
||||
bp.static(base_uri2, static_file_directory, name="uploads")
|
||||
|
||||
app.router.reset()
|
||||
app.blueprint(bp)
|
||||
|
||||
uri = app.url_for(
|
||||
|
@ -169,7 +181,8 @@ def test_static_directory(app, file_name, base_uri, static_file_directory):
|
|||
|
||||
|
||||
@pytest.mark.parametrize("file_name", ["test.file", "decode me.txt"])
|
||||
def test_static_head_request(app, file_name, static_file_directory):
|
||||
def test_static_head_request(file_name, static_file_directory):
|
||||
app = Sanic("base")
|
||||
app.static(
|
||||
"/testing.file",
|
||||
get_file_path(static_file_directory, file_name),
|
||||
|
@ -214,7 +227,8 @@ def test_static_head_request(app, file_name, static_file_directory):
|
|||
|
||||
|
||||
@pytest.mark.parametrize("file_name", ["test.file", "decode me.txt"])
|
||||
def test_static_content_range_correct(app, file_name, static_file_directory):
|
||||
def test_static_content_range_correct(file_name, static_file_directory):
|
||||
app = Sanic("base")
|
||||
app.static(
|
||||
"/testing.file",
|
||||
get_file_path(static_file_directory, file_name),
|
||||
|
@ -252,11 +266,6 @@ def test_static_content_range_correct(app, file_name, static_file_directory):
|
|||
"static", name="test_bp_static.static", filename="any"
|
||||
)
|
||||
assert uri == app.url_for("test_bp_static.static")
|
||||
assert uri == app.url_for("test_bp_static.static", name="any")
|
||||
assert uri == app.url_for("test_bp_static.static", filename="any")
|
||||
assert uri == app.url_for(
|
||||
"test_bp_static.static", name="any", filename="any"
|
||||
)
|
||||
|
||||
request, response = app.test_client.get(uri, headers=headers)
|
||||
assert response.status == 206
|
||||
|
@ -270,7 +279,8 @@ def test_static_content_range_correct(app, file_name, static_file_directory):
|
|||
|
||||
|
||||
@pytest.mark.parametrize("file_name", ["test.file", "decode me.txt"])
|
||||
def test_static_content_range_front(app, file_name, static_file_directory):
|
||||
def test_static_content_range_front(file_name, static_file_directory):
|
||||
app = Sanic("base")
|
||||
app.static(
|
||||
"/testing.file",
|
||||
get_file_path(static_file_directory, file_name),
|
||||
|
@ -308,11 +318,7 @@ def test_static_content_range_front(app, file_name, static_file_directory):
|
|||
"static", name="test_bp_static.static", filename="any"
|
||||
)
|
||||
assert uri == app.url_for("test_bp_static.static")
|
||||
assert uri == app.url_for("test_bp_static.static", name="any")
|
||||
assert uri == app.url_for("test_bp_static.static", filename="any")
|
||||
assert uri == app.url_for(
|
||||
"test_bp_static.static", name="any", filename="any"
|
||||
)
|
||||
|
||||
request, response = app.test_client.get(uri, headers=headers)
|
||||
assert response.status == 206
|
||||
|
@ -326,7 +332,8 @@ def test_static_content_range_front(app, file_name, static_file_directory):
|
|||
|
||||
|
||||
@pytest.mark.parametrize("file_name", ["test.file", "decode me.txt"])
|
||||
def test_static_content_range_back(app, file_name, static_file_directory):
|
||||
def test_static_content_range_back(file_name, static_file_directory):
|
||||
app = Sanic("base")
|
||||
app.static(
|
||||
"/testing.file",
|
||||
get_file_path(static_file_directory, file_name),
|
||||
|
@ -364,11 +371,7 @@ def test_static_content_range_back(app, file_name, static_file_directory):
|
|||
"static", name="test_bp_static.static", filename="any"
|
||||
)
|
||||
assert uri == app.url_for("test_bp_static.static")
|
||||
assert uri == app.url_for("test_bp_static.static", name="any")
|
||||
assert uri == app.url_for("test_bp_static.static", filename="any")
|
||||
assert uri == app.url_for(
|
||||
"test_bp_static.static", name="any", filename="any"
|
||||
)
|
||||
|
||||
request, response = app.test_client.get(uri, headers=headers)
|
||||
assert response.status == 206
|
||||
|
@ -382,7 +385,8 @@ def test_static_content_range_back(app, file_name, static_file_directory):
|
|||
|
||||
|
||||
@pytest.mark.parametrize("file_name", ["test.file", "decode me.txt"])
|
||||
def test_static_content_range_empty(app, file_name, static_file_directory):
|
||||
def test_static_content_range_empty(file_name, static_file_directory):
|
||||
app = Sanic("base")
|
||||
app.static(
|
||||
"/testing.file",
|
||||
get_file_path(static_file_directory, file_name),
|
||||
|
@ -420,11 +424,7 @@ def test_static_content_range_empty(app, file_name, static_file_directory):
|
|||
"static", name="test_bp_static.static", filename="any"
|
||||
)
|
||||
assert uri == app.url_for("test_bp_static.static")
|
||||
assert uri == app.url_for("test_bp_static.static", name="any")
|
||||
assert uri == app.url_for("test_bp_static.static", filename="any")
|
||||
assert uri == app.url_for(
|
||||
"test_bp_static.static", name="any", filename="any"
|
||||
)
|
||||
|
||||
request, response = app.test_client.get(uri)
|
||||
assert response.status == 200
|
||||
|
@ -440,6 +440,7 @@ def test_static_content_range_empty(app, file_name, static_file_directory):
|
|||
|
||||
@pytest.mark.parametrize("file_name", ["test.file", "decode me.txt"])
|
||||
def test_static_content_range_error(app, file_name, static_file_directory):
|
||||
app = Sanic("base")
|
||||
app.static(
|
||||
"/testing.file",
|
||||
get_file_path(static_file_directory, file_name),
|
||||
|
@ -475,11 +476,7 @@ def test_static_content_range_error(app, file_name, static_file_directory):
|
|||
"static", name="test_bp_static.static", filename="any"
|
||||
)
|
||||
assert uri == app.url_for("test_bp_static.static")
|
||||
assert uri == app.url_for("test_bp_static.static", name="any")
|
||||
assert uri == app.url_for("test_bp_static.static", filename="any")
|
||||
assert uri == app.url_for(
|
||||
"test_bp_static.static", name="any", filename="any"
|
||||
)
|
||||
|
||||
request, response = app.test_client.get(uri, headers=headers)
|
||||
assert response.status == 416
|
||||
|
|
|
@ -1,3 +1,7 @@
|
|||
import pytest
|
||||
|
||||
from sanic_routing.exceptions import RouteExists
|
||||
|
||||
from sanic.response import text
|
||||
|
||||
|
||||
|
@ -38,6 +42,8 @@ def test_vhosts_with_defaults(app):
|
|||
async def handler1(request):
|
||||
return text("Hello, world!")
|
||||
|
||||
with pytest.raises(RouteExists):
|
||||
|
||||
@app.route("/")
|
||||
async def handler2(request):
|
||||
return text("default")
|
||||
|
@ -45,6 +51,3 @@ def test_vhosts_with_defaults(app):
|
|||
headers = {"Host": "hello.com"}
|
||||
request, response = app.test_client.get("/", headers=headers)
|
||||
assert response.text == "Hello, world!"
|
||||
|
||||
request, response = app.test_client.get("/")
|
||||
assert response.text == "default"
|
||||
|
|
Loading…
Reference in New Issue
Block a user