Vasanko/sanic
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge in latest from sanic-routing branch

Browse Source
This commit is contained in:
Adam Hopkins 2021-02-15 17:20:07 +02:00
commit 973c315790
44 changed files with 1299 additions and 609 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
.github/workflows/codeql-analysis.yml vendored Normal file
View File

@ -0,0 +1,67 @@
# For most projects, this workflow file will not need changing; you simply need
# to commit it to your repository.
#
# You may wish to alter this file to override the set of languages analyzed,
# or to provide custom queries or build logic.
#
# ******** NOTE ********
# We have attempted to detect the languages in your repository. Please check
# the `language` matrix defined below to confirm you have the correct set of
# supported CodeQL languages.
#
name: "CodeQL"
on:
push:
branches: [ master ]
pull_request:
# The branches below must be a subset of the branches above
branches: [ master ]
schedule:
- cron: '25 16 * * 0'
jobs:
analyze:
name: Analyze
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
language: [ 'python' ]
# CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python' ]
# Learn more:
# https://docs.github.com/en/free-pro-team@latest/github/finding-security-vulnerabilities-and-errors-in-your-code/configuring-code-scanning#changing-the-languages-that-are-analyzed
steps:
- name: Checkout repository
uses: actions/checkout@v2
# Initializes the CodeQL tools for scanning.
- name: Initialize CodeQL
uses: github/codeql-action/init@v1
with:
languages: ${{ matrix.language }}
# If you wish to specify custom queries, you can do so here or in a config file.
# By default, queries listed here will override any specified in a config file.
# Prefix the list here with "+" to use these queries and those in the config file.
# queries: ./path/to/local/query, your-org/your-repo/queries@main
# Autobuild attempts to build any compiled languages (C/C++, C#, or Java).
# If this step fails, then you should remove it and run the build manually (see below)
- name: Autobuild
uses: github/codeql-action/autobuild@v1
# ℹ️ Command-line programs to run using the OS shell.
# 📚 https://git.io/JvXDl
# ✏️ If the Autobuild fails above, remove it and uncomment the following three lines
# and modify them (or add more) to build your code if your project
# uses a compiled language
#- run: |
# make bootstrap
# make release
- name: Perform CodeQL Analysis
uses: github/codeql-action/analyze@v1

7
Makefile
View File

@ -63,17 +63,14 @@ ifdef include_tests
isort -rc sanic tests
else
$(info Sorting Imports)
isort -rc sanic tests
isort -rc sanic tests --profile=black
endif
endif
black:
black --config ./.black.toml sanic tests
isort:
isort sanic tests
pretty: black isort
pretty: beautify
docs-clean:
cd docs && make clean

5
SECURITY.md
View File

@ -6,14 +6,15 @@ Sanic releases long term support release once a year in December. LTS releases r
| Version | LTS | Supported |
| ------- | ------------- | ------------------ |
| 20.9 | | :heavy_check_mark: |
| 20.12 | until 2022-12 | :heavy_check_mark: |
| 20.9 | | :x: |
| 20.6 | | :x: |
| 20.3 | | :x: |
| 19.12 | until 2021-12 | :white_check_mark: |
| 19.9 | | :x: |
| 19.6 | | :x: |
| 19.3 | | :x: |
| 18.12 | until 2020-12 | :white_check_mark: |
| 18.12 | | :x: |
| 0.8.3 | | :x: |
| 0.7.0 | | :x: |
| 0.6.0 | | :x: |

321
sanic/app.py
View File

@ -25,27 +25,24 @@ from typing import (
)
from urllib.parse import urlencode, urlunparse
from sanic_routing.route import Route
from sanic_routing.exceptions import FinalizationError # type: ignore
from sanic_routing.route import Route # type: ignore
from sanic import reloader_helpers
from sanic.asgi import ASGIApp
from sanic.base import BaseSanic
from sanic.blueprint_group import BlueprintGroup
from sanic.blueprints import Blueprint
from sanic.config import BASE_LOGO, Config
from sanic.exceptions import (
InvalidUsage,
NotFound,
SanicException,
ServerError,
URLBuildError,
)
from sanic.handlers import ErrorHandler, ListenerType, MiddlewareType
from sanic.log import LOGGING_CONFIG_DEFAULTS, error_logger, logger
from sanic.mixins.base import BaseMixin
from sanic.mixins.exceptions import ExceptionMixin
from sanic.mixins.listeners import ListenerEvent, ListenerMixin
from sanic.mixins.middleware import MiddlewareMixin
from sanic.mixins.routes import RouteMixin
from sanic.mixins.listeners import ListenerEvent
from sanic.models.futures import (
FutureException,
FutureListener,
@ -68,9 +65,7 @@ from sanic.static import register as static_register
from sanic.websocket import ConnectionClosed, WebSocketProtocol
class Sanic(
BaseMixin, RouteMixin, MiddlewareMixin, ListenerMixin, ExceptionMixin
):
class Sanic(BaseSanic):
"""
The main application instance
"""
@ -103,9 +98,7 @@ class Sanic(
self.name = name
self.asgi = False
self.router = router or Router(
exception=NotFound, method_handler_exception=NotFound
)
self.router = router or Router()
self.request_class = request_class
self.error_handler = error_handler or ErrorHandler()
self.config = Config(load_env=load_env)
@ -124,6 +117,9 @@ class Sanic(
self.websocket_tasks: Set[Future] = set()
self.named_request_middleware: Dict[str, Deque[MiddlewareType]] = {}
self.named_response_middleware: Dict[str, Deque[MiddlewareType]] = {}
# self.named_request_middleware: Dict[str, MiddlewareType] = {}
# self.named_response_middleware: Dict[str, MiddlewareType] = {}
self._test_manager = None
self._test_client = None
self._asgi_client = None
# Register alternative method names
@ -135,6 +131,8 @@ class Sanic(
if self.config.REGISTER:
self.__class__.register_app(self)
self.router.ctx.app = self
@property
def loop(self):
"""
@ -175,10 +173,6 @@ class Sanic(
partial(self._loop_add_task, task)
)
# Decorator
def _apply_listener(self, listener: FutureListener):
return self.register_listener(listener.listener, listener.event)
def register_listener(self, listener: Callable, event: str) -> Any:
"""
Register the listener for a given event.
@ -197,42 +191,6 @@ class Sanic(
self.listeners[_event].append(listener)
return listener
def _apply_route(self, route: FutureRoute) -> Route:
return self.router.add(**route._asdict())
def _apply_static(self, static: FutureStatic) -> Route:
return static_register(self, static)
def enable_websocket(self, enable: bool = True):
"""
Enable or disable the support for websocket.
Websocket is enabled automatically if websocket routes are
added to the application.
"""
if not self.websocket_enabled:
# if the server is stopped, we want to cancel any ongoing
# websocket tasks, to allow the server to exit promptly
self.listener("before_server_stop")(self._cancel_websocket_tasks)
self.websocket_enabled = enable
# Decorator
def _apply_exception_handler(self, handler: FutureException):
"""Decorate a function to be registered as a handler for exceptions
:param exceptions: exceptions
:return: decorated function
"""
for exception in handler.exceptions:
if isinstance(exception, (tuple, list)):
for e in exception:
self.error_handler.add(e, handler.handler)
else:
self.error_handler.add(exception, handler.handler)
return handler
def register_middleware(self, middleware, attach_to: str = "request"):
"""
Register an application level middleware that will be attached
@ -288,13 +246,49 @@ class Sanic(
if middleware not in self.named_response_middleware[_rn]:
self.named_response_middleware[_rn].appendleft(middleware)
# Decorator
def _apply_exception_handler(self, handler: FutureException):
"""Decorate a function to be registered as a handler for exceptions
:param exceptions: exceptions
:return: decorated function
"""
for exception in handler.exceptions:
if isinstance(exception, (tuple, list)):
for e in exception:
self.error_handler.add(e, handler.handler)
else:
self.error_handler.add(exception, handler.handler)
return handler
def _apply_listener(self, listener: FutureListener):
return self.register_listener(listener.listener, listener.event)
def _apply_route(self, route: FutureRoute) -> Route:
params = route._asdict()
websocket = params.pop("websocket", False)
subprotocols = params.pop("subprotocols", None)
if websocket:
self.enable_websocket()
websocket_handler = partial(
self._websocket_handler,
route.handler,
subprotocols=subprotocols,
)
websocket_handler.__name__ = route.handler.__name__ # type: ignore
websocket_handler.is_websocket = True # type: ignore
params["handler"] = websocket_handler
return self.router.add(**params)
def _apply_static(self, static: FutureStatic) -> Route:
return static_register(self, static)
def _apply_middleware(
self,
middleware: FutureMiddleware,
route_names: Optional[List[str]] = None,
):
print(f"{middleware=}")
if route_names:
return self.register_named_middleware(
middleware.middleware, route_names, middleware.attach_to
@ -304,6 +298,19 @@ class Sanic(
middleware.middleware, middleware.attach_to
)
def enable_websocket(self, enable=True):
"""Enable or disable the support for websocket.
Websocket is enabled automatically if websocket routes are
added to the application.
"""
if not self.websocket_enabled:
# if the server is stopped, we want to cancel any ongoing
# websocket tasks, to allow the server to exit promptly
self.listener("before_server_stop")(self._cancel_websocket_tasks)
self.websocket_enabled = enable
def blueprint(self, blueprint, **options):
"""Register a blueprint on the application.
@ -323,6 +330,12 @@ class Sanic(
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):
@ -351,30 +364,28 @@ class Sanic(
# 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]
@ -382,22 +393,36 @@ class Sanic(
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")
@ -415,44 +440,44 @@ class Sanic(
if "://" in netloc[:8]:
netloc = netloc.split("://", 1)[-1]
for match in matched_params:
name, _type, pattern = self.router.parse_parameter_string(match)
# find all the parameters we will need to build in the URL
# matched_params = re.findall(self.router.parameter_pattern, uri)
route.finalize()
for param_info 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
if name in kwargs:
supplied_param = kwargs.get(name)
del kwargs[name]
else:
try:
supplied_param = str(kwargs.pop(param_info.name))
except KeyError:
raise URLBuildError(
f"Required parameter `{name}` was not passed to url_for"
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 _type != str:
type_name = _type.__name__
if param_info.cast != str:
msg = (
f'Value "{supplied_param}" '
f"for parameter `{name}` does not "
f"match pattern for type `{type_name}`: {pattern}"
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 `{name}` '
f"does not satisfy pattern {pattern}"
f'Value "{supplied_param}" for parameter '
f"`{param_info.name}` does not satisfy "
f"pattern {param_info.pattern.pattern}"
)
raise URLBuildError(msg)
# replace the parameter in the URL with the supplied value
replacement_regex = f"(<{name}.*?>)"
replacement_regex = f"(<{param_info.name}.*?>)"
out = re.sub(replacement_regex, supplied_param, out)
# parse the remainder of the keyword arguments into a querystring
@ -545,14 +570,13 @@ class Sanic(
# Fetch handler from router
(
handler,
args,
kwargs,
uri,
name,
endpoint,
ignore_body,
) = self.router.get(request)
request.name = name
request._match_info = kwargs
if (
request.stream
@ -578,7 +602,7 @@ class Sanic(
# Execute Handler
# -------------------------------------------- #
request.uri_template = uri
request.uri_template = f"/{uri}"
if handler is None:
raise ServerError(
(
@ -587,10 +611,10 @@ class Sanic(
)
)
request.endpoint = endpoint
request.endpoint = request.name
# Run response handler
response = handler(request, *args, **kwargs)
response = handler(request, **kwargs)
if isawaitable(response):
response = await response
if response:
@ -615,26 +639,60 @@ class Sanic(
except CancelledError:
raise
except Exception as e:
# -------------------------------------------- #
# Response Generation Failed
# -------------------------------------------- #
await self.handle_exception(request, e)
async def _websocket_handler(
self, handler, request, *args, subprotocols=None, **kwargs
):
request.app = self
if not getattr(handler, "__blueprintname__", False):
request.endpoint = handler.__name__
else:
request.endpoint = (
getattr(handler, "__blueprintname__", "") + handler.__name__
)
pass
if self.asgi:
ws = request.transport.get_websocket_connection()
else:
protocol = request.transport.get_protocol()
protocol.app = self
ws = await protocol.websocket_handshake(request, subprotocols)
# schedule the application handler
# its future is kept in self.websocket_tasks in case it
# needs to be cancelled due to the server being stopped
fut = ensure_future(handler(request, ws, *args, **kwargs))
self.websocket_tasks.add(fut)
try:
await fut
except (CancelledError, ConnectionClosed):
pass
finally:
self.websocket_tasks.remove(fut)
await ws.close()
# -------------------------------------------------------------------- #
# Testing
# -------------------------------------------------------------------- #
@property
def test_client(self):
def test_client(self): # noqa
if self._test_client:
return self._test_client
elif self._test_manager:
return self._test_manager.test_client
from sanic_testing.testing import SanicTestClient # type: ignore
self._test_client = SanicTestClient(self)
return self._test_client
@property
def asgi_client(self):
def asgi_client(self): # noqa
"""
A testing client that uses ASGI to reach into the application to
execute hanlers.
@ -644,6 +702,8 @@ class Sanic(
"""
if self._asgi_client:
return self._asgi_client
elif self._test_manager:
return self._test_manager.asgi_client
from sanic_testing.testing import SanicASGITestClient # type: ignore
self._asgi_client = SanicASGITestClient(self)
@ -915,7 +975,11 @@ class Sanic(
):
"""Helper function used by `run` and `create_server`."""
try:
self.router.finalize()
except FinalizationError as e:
if not Sanic.test_mode:
raise e
if isinstance(ssl, dict):
# try common aliaseses
@ -950,9 +1014,7 @@ class Sanic(
"backlog": backlog,
}
# -------------------------------------------- #
# Register start/stop events
# -------------------------------------------- #
for event_name, settings_name, reverse in (
("before_server_start", "before_start", False),
@ -1014,40 +1076,6 @@ class Sanic(
for task in app.websocket_tasks:
task.cancel()
async def _websocket_handler(
self, handler, request, *args, subprotocols=None, **kwargs
):
request.app = self
if not getattr(handler, "__blueprintname__", False):
request.endpoint = handler.__name__
else:
request.endpoint = (
getattr(handler, "__blueprintname__", "") + handler.__name__
)
pass
if self.asgi:
ws = request.transport.get_websocket_connection()
else:
protocol = request.transport.get_protocol()
protocol.app = self
ws = await protocol.websocket_handshake(request, subprotocols)
# schedule the application handler
# its future is kept in self.websocket_tasks in case it
# needs to be cancelled due to the server being stopped
fut = ensure_future(handler(request, ws, *args, **kwargs))
self.websocket_tasks.add(fut)
try:
await fut
except (CancelledError, ConnectionClosed):
pass
finally:
self.websocket_tasks.remove(fut)
await ws.close()
# -------------------------------------------------------------------- #
# ASGI
# -------------------------------------------------------------------- #
@ -1057,9 +1085,10 @@ class Sanic(
To be ASGI compliant, our instance must be a callable that accepts
three arguments: scope, receive, send. See the ASGI reference for more
details: https://asgi.readthedocs.io/en/latest
/"""
"""
self.asgi = True
asgi_app = await ASGIApp.create(self, scope, receive, send)
self._asgi_app = await ASGIApp.create(self, scope, receive, send)
asgi_app = self._asgi_app
await asgi_app()
_asgi_single_callable = True # We conform to ASGI 3.0 single-callable

1
sanic/asgi.py
View File

@ -131,6 +131,7 @@ class Lifespan:
in sequence since the ASGI lifespan protocol only supports a single
startup event.
"""
self.asgi_app.sanic_app.router.finalize()
listeners = self.asgi_app.sanic_app.listeners.get(
"before_server_start", []
) + self.asgi_app.sanic_app.listeners.get("after_server_start", [])

36
sanic/base.py Normal file
View File

@ -0,0 +1,36 @@
from sanic.mixins.exceptions import ExceptionMixin
from sanic.mixins.listeners import ListenerMixin
from sanic.mixins.middleware import MiddlewareMixin
from sanic.mixins.routes import RouteMixin
class Base(type):
def __new__(cls, name, bases, attrs):
init = attrs.get("__init__")
def __init__(self, *args, **kwargs):
nonlocal init
nonlocal name
bases = [
b for base in type(self).__bases__ for b in base.__bases__
]
for base in bases:
base.__init__(self, *args, **kwargs)
if init:
init(self, *args, **kwargs)
attrs["__init__"] = __init__
return type.__new__(cls, name, bases, attrs)
class BaseSanic(
RouteMixin,
MiddlewareMixin,
ListenerMixin,
ExceptionMixin,
metaclass=Base,
):
...

39
sanic/blueprints.py
View File

@ -1,18 +1,12 @@
from collections import defaultdict, namedtuple
from typing import Iterable, Optional
from collections import defaultdict
from typing import Optional
from sanic.base import BaseSanic
from sanic.blueprint_group import BlueprintGroup
from sanic.mixins.base import BaseMixin
from sanic.mixins.exceptions import ExceptionMixin
from sanic.mixins.listeners import ListenerMixin
from sanic.mixins.middleware import MiddlewareMixin
from sanic.mixins.routes import RouteMixin
from sanic.models.futures import FutureRoute, FutureStatic
class Blueprint(
BaseMixin, RouteMixin, MiddlewareMixin, ListenerMixin, ExceptionMixin
):
class Blueprint(BaseSanic):
"""
In *Sanic* terminology, a **Blueprint** is a logical collection of
URLs that perform a specific set of tasks which can be identified by
@ -122,20 +116,35 @@ class Blueprint(
# 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
)
name = app._generate_name(future.name)
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,
name,
future.ignore_body,
future.websocket,
future.subprotocols,
future.unquote,
future.static,
)
route = app._apply_route(apply_route)
routes.append(route)
operation = (
routes.extend if isinstance(route, list) else routes.append
)
operation(route)
# Static Files
for future in self._future_statics:
@ -148,6 +157,7 @@ class Blueprint(
route_names = [route.name for route in routes if route]
# Middleware
if route_names:
for future in self._future_middleware:
app._apply_middleware(future, route_names)
@ -158,6 +168,3 @@ class Blueprint(
# 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__}"

19
sanic/mixins/base.py
View File

@ -1,19 +0,0 @@
class Base(type):
def __new__(cls, name, bases, attrs):
init = attrs.get("__init__")
def __init__(self, *args, **kwargs):
nonlocal init
for base in type(self).__bases__:
if base.__name__ != "BaseMixin":
base.__init__(self, *args, **kwargs)
if init:
init(self, *args, **kwargs)
attrs["__init__"] = __init__
return type.__new__(cls, name, bases, attrs)
class BaseMixin(metaclass=Base):
...

5
sanic/mixins/exceptions.py
View File

@ -1,5 +1,3 @@
from enum import Enum, auto
from functools import partial
from typing import Set
from sanic.models.futures import FutureException
@ -29,6 +27,9 @@ class ExceptionMixin:
nonlocal apply
nonlocal exceptions
if isinstance(exceptions[0], list):
exceptions = tuple(*exceptions)
future_exception = FutureException(handler, exceptions)
self._future_exceptions.add(future_exception)
if apply:

6
sanic/mixins/listeners.py
View File

@ -1,6 +1,6 @@
from enum import Enum, auto
from functools import partial
from typing import Any, Callable, Coroutine, Optional, Set, Union
from typing import Any, Callable, Coroutine, List, Optional, Set, Union
from sanic.models.futures import FutureListener
@ -17,7 +17,7 @@ class ListenerEvent(str, Enum):
class ListenerMixin:
def __init__(self, *args, **kwargs) -> None:
self._future_listeners: Set[FutureListener] = set()
self._future_listeners: List[FutureListener] = list()
def _apply_listener(self, listener: FutureListener):
raise NotImplementedError
@ -51,7 +51,7 @@ class ListenerMixin:
nonlocal apply
future_listener = FutureListener(listener, event)
self._future_listeners.add(future_listener)
self._future_listeners.append(future_listener)
if apply:
self._apply_listener(future_listener)
return listener

6
sanic/mixins/middleware.py
View File

@ -1,12 +1,12 @@
from functools import partial
from typing import Set
from typing import List
from sanic.models.futures import FutureMiddleware
class MiddlewareMixin:
def __init__(self, *args, **kwargs) -> None:
self._future_middleware: Set[FutureMiddleware] = set()
self._future_middleware: List[FutureMiddleware] = list()
def _apply_middleware(self, middleware: FutureMiddleware):
raise NotImplementedError
@ -30,7 +30,7 @@ class MiddlewareMixin:
nonlocal apply
future_middleware = FutureMiddleware(middleware, attach_to)
self._future_middleware.add(future_middleware)
self._future_middleware.append(future_middleware)
if apply:
self._apply_middleware(future_middleware)
return middleware

84
sanic/mixins/routes.py
View File

@ -1,9 +1,8 @@
from functools import partial
from inspect import signature
from pathlib import PurePath
from typing import Iterable, List, Optional, Set, Union
from sanic_routing.route import Route
from sanic_routing.route import Route # type: ignore
from sanic.constants import HTTP_METHODS
from sanic.models.futures import FutureRoute, FutureStatic
@ -36,6 +35,8 @@ class RouteMixin:
apply: bool = True,
subprotocols: Optional[List[str]] = None,
websocket: bool = False,
unquote: bool = False,
static: bool = False,
):
"""
Decorate a function to be registered as a route
@ -52,9 +53,6 @@ class RouteMixin:
:return: tuple of routes, decorated function
"""
if websocket:
self.enable_websocket()
# Fix case where the user did not prefix the URL with a /
# and will probably get confused as to why it's not working
if not uri.startswith("/"):
@ -63,6 +61,9 @@ class RouteMixin:
if strict_slashes is None:
strict_slashes = self.strict_slashes
if not methods and not websocket:
methods = frozenset({"GET"})
def decorator(handler):
nonlocal uri
nonlocal methods
@ -74,39 +75,43 @@ 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
if websocket:
websocket_handler = partial(
self._websocket_handler,
handler,
subprotocols=subprotocols,
)
websocket_handler.__name__ = (
"websocket_handler_" + handler.__name__
)
websocket_handler.is_websocket = True
handler = websocket_handler
name = self._generate_name(name, handler)
# TODO:
# - THink this thru.... do we want all routes namespaced?
# -
name = self._generate_name(handler, name)
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,
uri,
frozenset(methods),
None if websocket else frozenset([x.upper() for x in methods]),
host,
strict_slashes,
stream,
version,
name,
ignore_body,
websocket,
subprotocols,
unquote,
static,
)
self._future_routes.add(route)
@ -441,6 +446,7 @@ class RouteMixin:
subprotocols: Optional[List[str]] = None,
version: Optional[int] = None,
name: Optional[str] = None,
apply: bool = True,
):
"""
Decorate a function to be registered as a websocket route
@ -543,12 +549,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,
@ -566,5 +576,29 @@ 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:
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

4
sanic/models/futures.py
View File

@ -13,6 +13,10 @@ FutureRoute = namedtuple(
"version",
"name",
"ignore_body",
"websocket",
"subprotocols",
"unquote",
"static",
],
)
FutureListener = namedtuple("FutureListener", ["listener", "event"])

4
sanic/request.py
View File

@ -87,6 +87,7 @@ class Request:
"_port",
"_remote_addr",
"_socket",
"_match_info",
"app",
"body",
"conn_info",
@ -147,6 +148,7 @@ class Request:
self.uri_template: Optional[str] = None
self.request_middleware_started = False
self._cookies: Dict[str, str] = {}
self._match_info = {}
self.stream: Optional[Http] = None
self.endpoint: Optional[str] = None
@ -455,7 +457,7 @@ class Request:
"""
:return: matched info after resolving route
"""
return self.app.router.get(self)[2]
return self._match_info
# Transport properties (obtained from local interface only)

171
sanic/router.py
View File

@ -1,14 +1,22 @@
from functools import lru_cache
from typing import Any, Dict, Iterable, Optional, Tuple, Union
from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
from sanic_routing import BaseRouter
from sanic_routing.route import Route
from sanic_routing import BaseRouter # type: ignore
from sanic_routing.exceptions import NoMethod # type: ignore
from sanic_routing.exceptions import (
NotFound as RoutingNotFound, # type: ignore
)
from sanic_routing.route import Route # type: ignore
from sanic.constants import HTTP_METHODS
from sanic.exceptions import MethodNotSupported, NotFound
from sanic.handlers import RouteHandler
from sanic.request import Request
ROUTER_CACHE_SIZE = 1024
class Router(BaseRouter):
"""
The router implementation responsible for routing a :class:`Request` object
@ -18,18 +26,38 @@ class Router(BaseRouter):
DEFAULT_METHOD = "GET"
ALLOWED_METHODS = HTTP_METHODS
@lru_cache
def get(
self, request: Request
) -> Tuple[
RouteHandler,
Tuple[Any, ...],
Dict[str, Any],
str,
str,
Optional[str],
bool,
]:
# Putting the lru_cache on Router.get() performs better for the benchmarsk
# at tests/benchmark/test_route_resolution_benchmark.py
# However, overall application performance is significantly improved
# with the lru_cache on this method.
@lru_cache(maxsize=ROUTER_CACHE_SIZE)
def _get(
self, path, method, host
) -> Tuple[RouteHandler, Dict[str, Any], str, str, bool,]:
try:
route, handler, params = self.resolve(
path=path,
method=method,
extra={"host": host},
)
except RoutingNotFound as e:
raise NotFound("Requested URL {} not found".format(e.path))
except NoMethod as e:
raise MethodNotSupported(
"Method {} not allowed for URL {}".format(method, path),
method=method,
allowed_methods=e.allowed_methods,
)
return (
handler,
params,
route.path,
route.name,
route.ctx.ignore_body,
)
def get(self, request: Request):
"""
Retrieve a `Route` object containg the details about how to handle
a response for a given request
@ -41,23 +69,8 @@ class Router(BaseRouter):
:rtype: Tuple[ RouteHandler, Tuple[Any, ...], Dict[str, Any], str, str,
Optional[str], bool, ]
"""
route, handler, params = self.resolve(
path=request.path,
method=request.method,
)
# TODO: Implement response
# - args,
# - endpoint,
return (
handler,
(),
params,
route.path,
route.name,
None,
route.ctx.ignore_body,
return self._get(
request.path, request.method, request.headers.get("host")
)
def add(
@ -65,13 +78,15 @@ class Router(BaseRouter):
uri: str,
methods: Iterable[str],
handler: RouteHandler,
host: Optional[str] = None,
host: Optional[Union[str, Iterable[str]]] = None,
strict_slashes: bool = False,
stream: bool = False,
ignore_body: bool = False,
version: Union[str, float, int] = None,
name: Optional[str] = None,
) -> Route:
unquote: bool = False,
static: bool = False,
) -> Union[Route, List[Route]]:
"""
Add a handler to the router
@ -99,19 +114,93 @@ class Router(BaseRouter):
:return: the route object
:rtype: Route
"""
# TODO: Implement
# - host
# - strict_slashes
# - ignore_body
# - stream
if version is not None:
version = str(version).strip("/").lstrip("v")
uri = "/".join([f"/v{version}", uri.lstrip("/")])
route = super().add(
path=uri, handler=handler, methods=methods, name=name
params = dict(
path=uri,
handler=handler,
methods=methods,
name=name,
strict=strict_slashes,
unquote=unquote,
)
if isinstance(host, str):
hosts = [host]
else:
hosts = host or [None] # type: ignore
routes = []
for host in hosts:
if host:
params.update({"requirements": {"host": host}})
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)
if len(routes) == 1:
return routes[0]
return routes
def is_stream_handler(self, request) -> bool:
"""
Handler for request is stream or not.
:param request: Request object
:return: bool
"""
try:
handler = self.get(request)[0]
except (NotFound, MethodNotSupported):
return False
if hasattr(handler, "view_class") and hasattr(
handler.view_class, request.method.lower()
):
handler = getattr(handler.view_class, request.method.lower())
return hasattr(handler, "is_stream")
@lru_cache(maxsize=ROUTER_CACHE_SIZE)
def find_route_by_view_name(self, view_name, name=None):
"""
Find a route in the router based on the specified view name.
:param view_name: string of view name to search by
:param kwargs: additional params, usually for static files
:return: tuple containing (uri, Route)
"""
if not view_name:
return None
route = self.name_index.get(view_name)
if not route:
full_name = self.ctx.app._generate_name(view_name)
route = self.name_index.get(full_name)
if not route:
return None
return route
@property
def routes_all(self):
return self.routes
@property
def routes_static(self):
return self.static_routes
@property
def routes_dynamic(self):
return self.dynamic_routes
@property
def routes_regex(self):
return self.regex_routes

12
sanic/static.py
View File

@ -157,11 +157,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>"
# 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 +174,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

1
setup.py
View File

@ -83,6 +83,7 @@ ujson = "ujson>=1.35" + env_dependency
uvloop = "uvloop>=0.5.3" + env_dependency
requirements = [
"sanic-routing",
"httptools>=0.0.10",
uvloop,
ujson,

28
tests/benchmark/test_route_resolution_benchmark.py
View File

@ -4,6 +4,8 @@ from pytest import mark
import sanic.router
from sanic.request import Request
seed("Pack my box with five dozen liquor jugs.")
@ -23,8 +25,17 @@ class TestSanicRouteResolution:
route_to_call = choice(simple_routes)
result = benchmark.pedantic(
router._get,
("/{}".format(route_to_call[-1]), route_to_call[0], "localhost"),
router.get,
(
Request(
"/{}".format(route_to_call[-1]).encode(),
{"host": "localhost"},
"v1",
route_to_call[0],
None,
None,
),
),
iterations=1000,
rounds=1000,
)
@ -47,8 +58,17 @@ class TestSanicRouteResolution:
print("{} -> {}".format(route_to_call[-1], url))
result = benchmark.pedantic(
router._get,
("/{}".format(url), route_to_call[0], "localhost"),
router.get,
(
Request(
"/{}".format(url).encode(),
{"host": "localhost"},
"v1",
route_to_call[0],
None,
None,
),
),
iterations=1000,
rounds=1000,
)

20
tests/conftest.py
View File

@ -4,12 +4,16 @@ import string
import sys
import uuid
from typing import Tuple
import pytest
from sanic_routing.exceptions import RouteExists
from sanic_testing import TestManager
from sanic import Sanic
from sanic.router import RouteExists, Router
from sanic.constants import HTTP_METHODS
from sanic.router import Router
random.seed("Pack my box with five dozen liquor jugs.")
@ -38,12 +42,12 @@ async def _handler(request):
TYPE_TO_GENERATOR_MAP = {
"string": lambda: "".join(
[random.choice(string.ascii_letters + string.digits) for _ in range(4)]
[random.choice(string.ascii_lowercase) for _ in range(4)]
),
"int": lambda: random.choice(range(1000000)),
"number": lambda: random.random(),
"alpha": lambda: "".join(
[random.choice(string.ascii_letters) for _ in range(4)]
[random.choice(string.ascii_lowercase) for _ in range(4)]
),
"uuid": lambda: str(uuid.uuid1()),
}
@ -52,7 +56,7 @@ TYPE_TO_GENERATOR_MAP = {
class RouteStringGenerator:
ROUTE_COUNT_PER_DEPTH = 100
HTTP_METHODS = ["GET", "PUT", "POST", "PATCH", "DELETE", "OPTION"]
HTTP_METHODS = HTTP_METHODS
ROUTE_PARAM_TYPES = ["string", "int", "number", "alpha", "uuid"]
def generate_random_direct_route(self, max_route_depth=4):
@ -105,12 +109,12 @@ class RouteStringGenerator:
@pytest.fixture(scope="function")
def sanic_router(app):
# noinspection PyProtectedMember
def _setup(route_details: tuple) -> (Router, tuple):
router = Router(app)
def _setup(route_details: tuple) -> Tuple[Router, tuple]:
router = Router()
added_router = []
for method, route in route_details:
try:
router._add(
router.add(
uri=f"/{route}",
methods=frozenset({method}),
host="localhost",
@ -119,6 +123,7 @@ def sanic_router(app):
added_router.append((method, route))
except RouteExists:
pass
router.finalize()
return router, added_router
return _setup
@ -137,5 +142,4 @@ def url_param_generator():
@pytest.fixture(scope="function")
def app(request):
app = Sanic(request.node.name)
# TestManager(app)
return app

2
tests/test_app.py
View File

@ -118,7 +118,7 @@ def test_app_route_raise_value_error(app):
def test_app_handle_request_handler_is_none(app, monkeypatch):
def mockreturn(*args, **kwargs):
return None, [], {}, "", "", None, False
return None, {}, "", "", False
# Not sure how to make app.router.get() return None, so use mock here.
monkeypatch.setattr(app.router, "get", mockreturn)

13
tests/test_asgi.py
View File

@ -45,7 +45,8 @@ def protocol(transport):
return transport.get_protocol()
def test_listeners_triggered(app):
def test_listeners_triggered():
app = Sanic("app")
before_server_start = False
after_server_start = False
before_server_stop = False
@ -71,6 +72,10 @@ def test_listeners_triggered(app):
nonlocal after_server_stop
after_server_stop = True
@app.route("/")
def handler(request):
return text("...")
class CustomServer(uvicorn.Server):
def install_signal_handlers(self):
pass
@ -121,6 +126,10 @@ def test_listeners_triggered_async(app):
nonlocal after_server_stop
after_server_stop = True
@app.route("/")
def handler(request):
return text("...")
class CustomServer(uvicorn.Server):
def install_signal_handlers(self):
pass
@ -325,7 +334,7 @@ async def test_cookie_customization(app):
@pytest.mark.asyncio
async def test_json_content_type(app):
async def test_content_type(app):
@app.get("/json")
def send_json(request):
return json({"foo": "bar"})

2
tests/test_bad_request.py
View File

@ -4,6 +4,8 @@ import asyncio
def test_bad_request_response(app):
lines = []
app.get("/")(lambda x: ...)
@app.listener("after_server_start")
async def _request(sanic, loop):
connect = asyncio.open_connection("127.0.0.1", 42101)

89
tests/test_blueprints.py
View File

@ -209,18 +209,28 @@ def test_bp_with_host(app):
app.blueprint(bp)
headers = {"Host": "example.com"}
request, response = app.test_client.get("/test1/", headers=headers)
assert response.text == "Hello"
headers = {"Host": "sub.example.com"}
request, response = app.test_client.get("/test1/", headers=headers)
assert response.text == "Hello subdomain!"
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):
@ -240,6 +250,7 @@ def test_several_bp_with_host(app):
assert bp.host == "example.com"
headers = {"Host": "example.com"}
request, response = app.test_client.get("/test/", headers=headers)
assert response.text == "Hello"
assert bp2.host == "sub.example.com"
@ -352,6 +363,29 @@ def test_bp_middleware(app):
assert response.text == "FAIL"
def test_bp_middleware_with_route(app):
blueprint = Blueprint("test_bp_middleware")
@blueprint.middleware("response")
async def process_response(request, response):
return text("OK")
@app.route("/")
async def handler(request):
return text("FAIL")
@blueprint.route("/bp")
async def bp_handler(request):
return text("FAIL")
app.blueprint(blueprint)
request, response = app.test_client.get("/bp")
assert response.status == 200
assert response.text == "OK"
def test_bp_middleware_order(app):
blueprint = Blueprint("test_bp_middleware_order")
order = list()
@ -425,6 +459,7 @@ def test_bp_exception_handler(app):
def test_bp_listeners(app):
app.route("/")(lambda x: x)
blueprint = Blueprint("test_middleware")
order = []
@ -537,19 +572,19 @@ def test_bp_shorthand(app):
app.blueprint(blueprint)
request, response = app.test_client.get("/get")
assert response.text == "OK"
assert response.body == b"OK"
request, response = app.test_client.post("/get")
assert response.status == 405
request, response = app.test_client.put("/put")
assert response.text == "OK"
assert response.body == b"OK"
request, response = app.test_client.get("/post")
assert response.status == 405
request, response = app.test_client.post("/post")
assert response.text == "OK"
assert response.body == b"OK"
request, response = app.test_client.get("/post")
assert response.status == 405
@ -561,19 +596,19 @@ def test_bp_shorthand(app):
assert response.status == 405
request, response = app.test_client.options("/options")
assert response.text == "OK"
assert response.body == b"OK"
request, response = app.test_client.get("/options")
assert response.status == 405
request, response = app.test_client.patch("/patch")
assert response.text == "OK"
assert response.body == b"OK"
request, response = app.test_client.get("/patch")
assert response.status == 405
request, response = app.test_client.delete("/delete")
assert response.text == "OK"
assert response.body == b"OK"
request, response = app.test_client.get("/delete")
assert response.status == 405
@ -699,7 +734,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()
@ -814,17 +850,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
bp = Blueprint("bp")
@app.get("/f1", strict_slashes=False)
def f1(request):
return text("f1")
@bp.get("/one", strict_slashes=False)
def one(request):
@ -834,7 +872,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
@ -842,19 +888,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

2
tests/test_cookies.py
View File

@ -43,7 +43,7 @@ async def test_cookies_asgi(app):
response_cookies = SimpleCookie()
response_cookies.load(response.headers.get("set-cookie", {}))
assert response.text == "Cookies are: working!"
assert response.body == b"Cookies are: working!"
assert response_cookies["right_back"].value == "at you"

68
tests/test_dynamic_routes.py
View File

@ -1,44 +1,44 @@
import pytest
# import pytest
from sanic.response import text
from sanic.router import RouteExists
# from sanic.response import text
# from sanic.router import RouteExists
@pytest.mark.parametrize(
"method,attr, expected",
[
("get", "text", "OK1 test"),
("post", "text", "OK2 test"),
("put", "text", "OK2 test"),
("delete", "status", 405),
],
)
def test_overload_dynamic_routes(app, method, attr, expected):
@app.route("/overload/<param>", methods=["GET"])
async def handler1(request, param):
return text("OK1 " + param)
# @pytest.mark.parametrize(
# "method,attr, expected",
# [
# ("get", "text", "OK1 test"),
# ("post", "text", "OK2 test"),
# ("put", "text", "OK2 test"),
# ("delete", "status", 405),
# ],
# )
# def test_overload_dynamic_routes(app, method, attr, expected):
# @app.route("/overload/<param>", methods=["GET"])
# async def handler1(request, param):
# return text("OK1 " + param)
@app.route("/overload/<param>", methods=["POST", "PUT"])
async def handler2(request, param):
return text("OK2 " + param)
# @app.route("/overload/<param>", methods=["POST", "PUT"])
# async def handler2(request, param):
# return text("OK2 " + param)
request, response = getattr(app.test_client, method)("/overload/test")
assert getattr(response, attr) == expected
# request, response = getattr(app.test_client, method)("/overload/test")
# assert getattr(response, attr) == expected
def test_overload_dynamic_routes_exist(app):
@app.route("/overload/<param>", methods=["GET"])
async def handler1(request, param):
return text("OK1 " + param)
# def test_overload_dynamic_routes_exist(app):
# @app.route("/overload/<param>", methods=["GET"])
# async def handler1(request, param):
# return text("OK1 " + param)
@app.route("/overload/<param>", methods=["POST", "PUT"])
async def handler2(request, param):
return text("OK2 " + param)
# @app.route("/overload/<param>", methods=["POST", "PUT"])
# async def handler2(request, param):
# return text("OK2 " + param)
# if this doesn't raise an error, than at least the below should happen:
# assert response.text == 'Duplicated'
with pytest.raises(RouteExists):
# # if this doesn't raise an error, than at least the below should happen:
# # assert response.text == 'Duplicated'
# with pytest.raises(RouteExists):
@app.route("/overload/<param>", methods=["PUT", "DELETE"])
async def handler3(request, param):
return text("Duplicated")
# @app.route("/overload/<param>", methods=["PUT", "DELETE"])
# async def handler3(request, param):
# return text("Duplicated")

5
tests/test_exceptions_handler.py
View File

@ -126,8 +126,9 @@ def test_html_traceback_output_in_debug_mode():
assert response.status == 500
soup = BeautifulSoup(response.body, "html.parser")
html = str(soup)
print(html)
assert "response = handler(request, *args, **kwargs)" in html
assert "response = handler(request, **kwargs)" in html
assert "handler_4" in html
assert "foo = bar" in html
@ -151,7 +152,7 @@ def test_chained_exception_handler():
soup = BeautifulSoup(response.body, "html.parser")
html = str(soup)
assert "response = handler(request, *args, **kwargs)" in html
assert "response = handler(request, **kwargs)" in html
assert "handler_6" in html
assert "foo = 1 / arg" in html
assert "ValueError" in html

10
tests/test_logging.py
View File

@ -103,7 +103,13 @@ def test_logging_pass_customer_logconfig():
assert fmt._fmt == modified_config["formatters"]["access"]["format"]
@pytest.mark.parametrize("debug", (True, False))
@pytest.mark.parametrize(
"debug",
(
True,
False,
),
)
def test_log_connection_lost(app, debug, monkeypatch):
""" Should not log Connection lost exception on non debug """
stream = StringIO()
@ -117,7 +123,7 @@ def test_log_connection_lost(app, debug, monkeypatch):
request.transport.close()
return response
req, res = app.test_client.get("/conn_lost", debug=debug)
req, res = app.test_client.get("/conn_lost", debug=debug, allow_none=True)
assert res is None
log = stream.getvalue()

3
tests/test_middleware.py
View File

@ -102,6 +102,7 @@ def test_middleware_response_raise_exception(app, caplog):
async def process_response(request, response):
raise Exception("Exception at response middleware")
app.route("/")(lambda x: x)
with caplog.at_level(logging.ERROR):
reqrequest, response = app.test_client.get("/fail")
@ -129,7 +130,7 @@ def test_middleware_override_request(app):
async def handler(request):
return text("FAIL")
response = app.test_client.get("/", gather_request=False)
_, response = app.test_client.get("/", gather_request=False)
assert response.status == 200
assert response.text == "OK"

9
tests/test_multiprocessing.py
View File

@ -68,9 +68,12 @@ def handler(request):
@pytest.mark.parametrize("protocol", [3, 4])
def test_pickle_app(app, protocol):
app.route("/")(handler)
app.router.finalize()
app.router.reset()
p_app = pickle.dumps(app, protocol=protocol)
del app
up_p_app = pickle.loads(p_app)
up_p_app.router.finalize()
assert up_p_app
request, response = up_p_app.test_client.get("/")
assert response.text == "Hello"
@ -81,9 +84,12 @@ def test_pickle_app_with_bp(app, protocol):
bp = Blueprint("test_text")
bp.route("/")(handler)
app.blueprint(bp)
app.router.finalize()
app.router.reset()
p_app = pickle.dumps(app, protocol=protocol)
del app
up_p_app = pickle.loads(p_app)
up_p_app.router.finalize()
assert up_p_app
request, response = up_p_app.test_client.get("/")
assert response.text == "Hello"
@ -93,9 +99,12 @@ def test_pickle_app_with_bp(app, protocol):
def test_pickle_app_with_static(app, protocol):
app.route("/")(handler)
app.static("/static", "/tmp/static")
app.router.finalize()
app.router.reset()
p_app = pickle.dumps(app, protocol=protocol)
del app
up_p_app = pickle.loads(p_app)
up_p_app.router.finalize()
assert up_p_app
request, response = up_p_app.test_client.get("/static/missing.txt")
assert response.status == 404

198
tests/test_named_routes.py
View File

@ -5,6 +5,7 @@ import asyncio
import pytest
from sanic import Sanic
from sanic.blueprints import Blueprint
from sanic.constants import HTTP_METHODS
from sanic.exceptions import URLBuildError
@ -17,7 +18,9 @@ from sanic.response import text
@pytest.mark.parametrize("method", HTTP_METHODS)
def test_versioned_named_routes_get(app, method):
def test_versioned_named_routes_get(method):
app = Sanic("app")
bp = Blueprint("test_bp", url_prefix="/bp")
method = method.lower()
@ -32,7 +35,6 @@ def test_versioned_named_routes_get(app, method):
return text("OK")
else:
print(func)
raise
func = getattr(bp, method)
@ -43,15 +45,28 @@ def test_versioned_named_routes_get(app, method):
return text("OK")
else:
print(func)
raise
app.blueprint(bp)
assert app.router.routes_all[f"/v1/{method}"].name == route_name
assert (
app.router.routes_all[
(
"v1",
method,
)
].name
== f"app.{route_name}"
)
route = app.router.routes_all[f"/v1/bp/{method}"]
assert route.name == f"test_bp.{route_name2}"
route = app.router.routes_all[
(
"v1",
"bp",
method,
)
]
assert route.name == f"app.test_bp.{route_name2}"
assert app.url_for(route_name) == f"/v1/{method}"
url = app.url_for(f"test_bp.{route_name2}")
@ -60,16 +75,19 @@ def test_versioned_named_routes_get(app, method):
app.url_for("handler")
def test_shorthand_default_routes_get(app):
def test_shorthand_default_routes_get():
app = Sanic("app")
@app.get("/get")
def handler(request):
return text("OK")
assert app.router.routes_all["/get"].name == "handler"
assert app.router.routes_all[("get",)].name == "app.handler"
assert app.url_for("handler") == "/get"
def test_shorthand_named_routes_get(app):
def test_shorthand_named_routes_get():
app = Sanic("app")
bp = Blueprint("test_bp", url_prefix="/bp")
@app.get("/get", name="route_get")
@ -82,84 +100,106 @@ def test_shorthand_named_routes_get(app):
app.blueprint(bp)
assert app.router.routes_all["/get"].name == "route_get"
assert app.router.routes_all[("get",)].name == "app.route_get"
assert app.url_for("route_get") == "/get"
with pytest.raises(URLBuildError):
app.url_for("handler")
assert app.router.routes_all["/bp/get"].name == "test_bp.route_bp"
assert (
app.router.routes_all[
(
"bp",
"get",
)
].name
== "app.test_bp.route_bp"
)
assert app.url_for("test_bp.route_bp") == "/bp/get"
with pytest.raises(URLBuildError):
app.url_for("test_bp.handler2")
def test_shorthand_named_routes_post(app):
def test_shorthand_named_routes_post():
app = Sanic("app")
@app.post("/post", name="route_name")
def handler(request):
return text("OK")
assert app.router.routes_all["/post"].name == "route_name"
assert app.router.routes_all[("post",)].name == "app.route_name"
assert app.url_for("route_name") == "/post"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_shorthand_named_routes_put(app):
def test_shorthand_named_routes_put():
app = Sanic("app")
@app.put("/put", name="route_put")
def handler(request):
return text("OK")
assert app.router.routes_all["/put"].name == "route_put"
assert app.router.routes_all[("put",)].name == "app.route_put"
assert app.url_for("route_put") == "/put"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_shorthand_named_routes_delete(app):
def test_shorthand_named_routes_delete():
app = Sanic("app")
@app.delete("/delete", name="route_delete")
def handler(request):
return text("OK")
assert app.router.routes_all["/delete"].name == "route_delete"
assert app.router.routes_all[("delete",)].name == "app.route_delete"
assert app.url_for("route_delete") == "/delete"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_shorthand_named_routes_patch(app):
def test_shorthand_named_routes_patch():
app = Sanic("app")
@app.patch("/patch", name="route_patch")
def handler(request):
return text("OK")
assert app.router.routes_all["/patch"].name == "route_patch"
assert app.router.routes_all[("patch",)].name == "app.route_patch"
assert app.url_for("route_patch") == "/patch"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_shorthand_named_routes_head(app):
def test_shorthand_named_routes_head():
app = Sanic("app")
@app.head("/head", name="route_head")
def handler(request):
return text("OK")
assert app.router.routes_all["/head"].name == "route_head"
assert app.router.routes_all[("head",)].name == "app.route_head"
assert app.url_for("route_head") == "/head"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_shorthand_named_routes_options(app):
def test_shorthand_named_routes_options():
app = Sanic("app")
@app.options("/options", name="route_options")
def handler(request):
return text("OK")
assert app.router.routes_all["/options"].name == "route_options"
assert app.router.routes_all[("options",)].name == "app.route_options"
assert app.url_for("route_options") == "/options"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_named_static_routes(app):
def test_named_static_routes():
app = Sanic("app")
@app.route("/test", name="route_test")
async def handler1(request):
return text("OK1")
@ -168,20 +208,21 @@ def test_named_static_routes(app):
async def handler2(request):
return text("OK2")
assert app.router.routes_all["/test"].name == "route_test"
assert app.router.routes_static["/test"].name == "route_test"
assert app.router.routes_all[("test",)].name == "app.route_test"
assert app.router.routes_static[("test",)].name == "app.route_test"
assert app.url_for("route_test") == "/test"
with pytest.raises(URLBuildError):
app.url_for("handler1")
assert app.router.routes_all["/pizazz"].name == "route_pizazz"
assert app.router.routes_static["/pizazz"].name == "route_pizazz"
assert app.router.routes_all[("pizazz",)].name == "app.route_pizazz"
assert app.router.routes_static[("pizazz",)].name == "app.route_pizazz"
assert app.url_for("route_pizazz") == "/pizazz"
with pytest.raises(URLBuildError):
app.url_for("handler2")
def test_named_dynamic_route(app):
def test_named_dynamic_route():
app = Sanic("app")
results = []
@app.route("/folder/<name>", name="route_dynamic")
@ -189,52 +230,83 @@ def test_named_dynamic_route(app):
results.append(name)
return text("OK")
assert app.router.routes_all["/folder/<name>"].name == "route_dynamic"
assert (
app.router.routes_all[
(
"folder",
"<name>",
)
].name
== "app.route_dynamic"
)
assert app.url_for("route_dynamic", name="test") == "/folder/test"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_dynamic_named_route_regex(app):
def test_dynamic_named_route_regex():
app = Sanic("app")
@app.route("/folder/<folder_id:[A-Za-z0-9]{0,4}>", name="route_re")
async def handler(request, folder_id):
return text("OK")
route = app.router.routes_all["/folder/<folder_id:[A-Za-z0-9]{0,4}>"]
assert route.name == "route_re"
route = app.router.routes_all[
(
"folder",
"<folder_id:[A-Za-z0-9]{0,4}>",
)
]
assert route.name == "app.route_re"
assert app.url_for("route_re", folder_id="test") == "/folder/test"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_dynamic_named_route_path(app):
def test_dynamic_named_route_path():
app = Sanic("app")
@app.route("/<path:path>/info", name="route_dynamic_path")
async def handler(request, path):
return text("OK")
route = app.router.routes_all["/<path:path>/info"]
assert route.name == "route_dynamic_path"
route = app.router.routes_all[
(
"<path:path>",
"info",
)
]
assert route.name == "app.route_dynamic_path"
assert app.url_for("route_dynamic_path", path="path/1") == "/path/1/info"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_dynamic_named_route_unhashable(app):
def test_dynamic_named_route_unhashable():
app = Sanic("app")
@app.route(
"/folder/<unhashable:[A-Za-z0-9/]+>/end/", name="route_unhashable"
)
async def handler(request, unhashable):
return text("OK")
route = app.router.routes_all["/folder/<unhashable:[A-Za-z0-9/]+>/end/"]
assert route.name == "route_unhashable"
route = app.router.routes_all[
(
"folder",
"<unhashable:[A-Za-z0-9/]+>",
"end",
)
]
assert route.name == "app.route_unhashable"
url = app.url_for("route_unhashable", unhashable="test/asdf")
assert url == "/folder/test/asdf/end"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_websocket_named_route(app):
def test_websocket_named_route():
app = Sanic("app")
ev = asyncio.Event()
@app.websocket("/ws", name="route_ws")
@ -242,26 +314,29 @@ def test_websocket_named_route(app):
assert ws.subprotocol is None
ev.set()
assert app.router.routes_all["/ws"].name == "route_ws"
assert app.router.routes_all[("ws",)].name == "app.route_ws"
assert app.url_for("route_ws") == "/ws"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_websocket_named_route_with_subprotocols(app):
def test_websocket_named_route_with_subprotocols():
app = Sanic("app")
results = []
@app.websocket("/ws", subprotocols=["foo", "bar"], name="route_ws")
async def handler(request, ws):
results.append(ws.subprotocol)
assert app.router.routes_all["/ws"].name == "route_ws"
assert app.router.routes_all[("ws",)].name == "app.route_ws"
assert app.url_for("route_ws") == "/ws"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_static_add_named_route(app):
def test_static_add_named_route():
app = Sanic("app")
async def handler1(request):
return text("OK1")
@ -271,20 +346,21 @@ def test_static_add_named_route(app):
app.add_route(handler1, "/test", name="route_test")
app.add_route(handler2, "/test2", name="route_test2")
assert app.router.routes_all["/test"].name == "route_test"
assert app.router.routes_static["/test"].name == "route_test"
assert app.router.routes_all[("test",)].name == "app.route_test"
assert app.router.routes_static[("test",)].name == "app.route_test"
assert app.url_for("route_test") == "/test"
with pytest.raises(URLBuildError):
app.url_for("handler1")
assert app.router.routes_all["/test2"].name == "route_test2"
assert app.router.routes_static["/test2"].name == "route_test2"
assert app.router.routes_all[("test2",)].name == "app.route_test2"
assert app.router.routes_static[("test2",)].name == "app.route_test2"
assert app.url_for("route_test2") == "/test2"
with pytest.raises(URLBuildError):
app.url_for("handler2")
def test_dynamic_add_named_route(app):
def test_dynamic_add_named_route():
app = Sanic("app")
results = []
async def handler(request, name):
@ -292,13 +368,17 @@ def test_dynamic_add_named_route(app):
return text("OK")
app.add_route(handler, "/folder/<name>", name="route_dynamic")
assert app.router.routes_all["/folder/<name>"].name == "route_dynamic"
assert (
app.router.routes_all[("folder", "<name>")].name == "app.route_dynamic"
)
assert app.url_for("route_dynamic", name="test") == "/folder/test"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_dynamic_add_named_route_unhashable(app):
def test_dynamic_add_named_route_unhashable():
app = Sanic("app")
async def handler(request, unhashable):
return text("OK")
@ -307,15 +387,23 @@ def test_dynamic_add_named_route_unhashable(app):
"/folder/<unhashable:[A-Za-z0-9/]+>/end/",
name="route_unhashable",
)
route = app.router.routes_all["/folder/<unhashable:[A-Za-z0-9/]+>/end/"]
assert route.name == "route_unhashable"
route = app.router.routes_all[
(
"folder",
"<unhashable:[A-Za-z0-9/]+>",
"end",
)
]
assert route.name == "app.route_unhashable"
url = app.url_for("route_unhashable", unhashable="folder1")
assert url == "/folder/folder1/end"
with pytest.raises(URLBuildError):
app.url_for("handler")
def test_overload_routes(app):
def test_overload_routes():
app = Sanic("app")
@app.route("/overload", methods=["GET"], name="route_first")
async def handler1(request):
return text("OK1")
@ -342,7 +430,7 @@ def test_overload_routes(app):
request, response = app.test_client.put(app.url_for("route_second"))
assert response.text == "OK2"
assert app.router.routes_all["/overload"].name == "route_first"
assert app.router.routes_all[("overload",)].name == "app.route_first"
with pytest.raises(URLBuildError):
app.url_for("handler1")

6
tests/test_payload_too_large.py
View File

@ -13,7 +13,7 @@ def test_payload_too_large_from_error_handler(app):
def handler_exception(request, exception):
return text("Payload Too Large from error_handler.", 413)
response = app.test_client.get("/1", gather_request=False)
_, response = app.test_client.get("/1", gather_request=False)
assert response.status == 413
assert response.text == "Payload Too Large from error_handler."
@ -25,7 +25,7 @@ def test_payload_too_large_at_data_received_default(app):
async def handler2(request):
return text("OK")
response = app.test_client.get("/1", gather_request=False)
_, response = app.test_client.get("/1", gather_request=False)
assert response.status == 413
assert "Request header" in response.text
@ -38,6 +38,6 @@ def test_payload_too_large_at_on_header_default(app):
return text("OK")
data = "a" * 1000
response = app.test_client.post("/1", gather_request=False, data=data)
_, response = app.test_client.post("/1", gather_request=False, data=data)
assert response.status == 413
assert "Request body" in response.text

15
tests/test_redirect.py
View File

@ -1,4 +1,4 @@
from urllib.parse import quote
from urllib.parse import quote, unquote
import pytest
@ -109,7 +109,14 @@ def test_redirect_with_header_injection(redirect_app):
assert not response.text.startswith("test-body")
@pytest.mark.parametrize("test_str", ["sanic-test", "sanictest", "sanic test"])
@pytest.mark.parametrize(
"test_str",
[
"sanic-test",
"sanictest",
"sanic test",
],
)
def test_redirect_with_params(app, test_str):
use_in_uri = quote(test_str)
@ -117,7 +124,7 @@ def test_redirect_with_params(app, test_str):
async def init_handler(request, test):
return redirect(f"/api/v2/test/{use_in_uri}/")
@app.route("/api/v2/test/<test>/")
@app.route("/api/v2/test/<test>/", unquote=True)
async def target_handler(request, test):
assert test == test_str
return text("OK")
@ -125,4 +132,4 @@ def test_redirect_with_params(app, test_str):
_, response = app.test_client.get(f"/api/v1/test/{use_in_uri}/")
assert response.status == 200
assert response.content == b"OK"
assert response.body == b"OK"

2
tests/test_reloader.py
View File

@ -42,6 +42,8 @@ def write_app(filename, **runargs):
app = Sanic(__name__)
app.route("/")(lambda x: x)
@app.listener("after_server_start")
def complete(*args):
print("complete", os.getpid(), {text!r})

131
tests/test_requests.py
View File

@ -10,7 +10,6 @@ import pytest
from sanic_testing.testing import (
ASGI_BASE_URL,
ASGI_HOST,
ASGI_PORT,
HOST,
PORT,
@ -19,7 +18,7 @@ from sanic_testing.testing import (
from sanic import Blueprint, Sanic
from sanic.exceptions import ServerError
from sanic.request import DEFAULT_HTTP_CONTENT_TYPE, Request, RequestParameters
from sanic.request import DEFAULT_HTTP_CONTENT_TYPE, RequestParameters
from sanic.response import html, json, text
@ -35,7 +34,7 @@ def test_sync(app):
request, response = app.test_client.get("/")
assert response.text == "Hello"
assert response.body == b"Hello"
@pytest.mark.asyncio
@ -46,7 +45,7 @@ async def test_sync_asgi(app):
request, response = await app.asgi_client.get("/")
assert response.text == "Hello"
assert response.body == b"Hello"
def test_ip(app):
@ -56,7 +55,7 @@ def test_ip(app):
request, response = app.test_client.get("/")
assert response.text == "127.0.0.1"
assert response.body == b"127.0.0.1"
@pytest.mark.asyncio
@ -67,10 +66,12 @@ async def test_url_asgi(app):
request, response = await app.asgi_client.get("/")
if response.text.endswith("/") and not ASGI_BASE_URL.endswith("/"):
response.text[:-1] == ASGI_BASE_URL
if response.body.decode().endswith("/") and not ASGI_BASE_URL.endswith(
"/"
):
response.body[:-1] == ASGI_BASE_URL.encode()
else:
assert response.text == ASGI_BASE_URL
assert response.body == ASGI_BASE_URL.encode()
def test_text(app):
@ -80,7 +81,7 @@ def test_text(app):
request, response = app.test_client.get("/")
assert response.text == "Hello"
assert response.body == b"Hello"
def test_html(app):
@ -109,13 +110,13 @@ def test_html(app):
request, response = app.test_client.get("/")
assert response.content_type == "text/html; charset=utf-8"
assert response.text == "<h1>Hello</h1>"
assert response.body == b"<h1>Hello</h1>"
request, response = app.test_client.get("/foo")
assert response.text == "<h1>Foo</h1>"
assert response.body == b"<h1>Foo</h1>"
request, response = app.test_client.get("/bar")
assert response.text == "<h1>Bar object repr</h1>"
assert response.body == b"<h1>Bar object repr</h1>"
@pytest.mark.asyncio
@ -126,7 +127,7 @@ async def test_text_asgi(app):
request, response = await app.asgi_client.get("/")
assert response.text == "Hello"
assert response.body == b"Hello"
def test_headers(app):
@ -186,7 +187,7 @@ def test_invalid_response(app):
request, response = app.test_client.get("/")
assert response.status == 500
assert response.text == "Internal Server Error."
assert response.body == b"Internal Server Error."
@pytest.mark.asyncio
@ -201,7 +202,7 @@ async def test_invalid_response_asgi(app):
request, response = await app.asgi_client.get("/")
assert response.status == 500
assert response.text == "Internal Server Error."
assert response.body == b"Internal Server Error."
def test_json(app):
@ -224,7 +225,7 @@ async def test_json_asgi(app):
request, response = await app.asgi_client.get("/")
results = json_loads(response.text)
results = json_loads(response.body)
assert results.get("test") is True
@ -237,7 +238,7 @@ def test_empty_json(app):
request, response = app.test_client.get("/")
assert response.status == 200
assert response.text == "null"
assert response.body == b"null"
@pytest.mark.asyncio
@ -249,7 +250,7 @@ async def test_empty_json_asgi(app):
request, response = await app.asgi_client.get("/")
assert response.status == 200
assert response.text == "null"
assert response.body == b"null"
def test_invalid_json(app):
@ -423,12 +424,12 @@ def test_content_type(app):
request, response = app.test_client.get("/")
assert request.content_type == DEFAULT_HTTP_CONTENT_TYPE
assert response.text == DEFAULT_HTTP_CONTENT_TYPE
assert response.body.decode() == DEFAULT_HTTP_CONTENT_TYPE
headers = {"content-type": "application/json"}
request, response = app.test_client.get("/", headers=headers)
assert request.content_type == "application/json"
assert response.text == "application/json"
assert response.body == b"application/json"
@pytest.mark.asyncio
@ -439,12 +440,12 @@ async def test_content_type_asgi(app):
request, response = await app.asgi_client.get("/")
assert request.content_type == DEFAULT_HTTP_CONTENT_TYPE
assert response.text == DEFAULT_HTTP_CONTENT_TYPE
assert response.body.decode() == DEFAULT_HTTP_CONTENT_TYPE
headers = {"content-type": "application/json"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.content_type == "application/json"
assert response.text == "application/json"
assert response.body == b"application/json"
def test_standard_forwarded(app):
@ -581,14 +582,15 @@ async def test_standard_forwarded_asgi(app):
"X-Scheme": "ws",
}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"for": "127.0.0.2", "proto": "ws"}
assert response.json == {"for": "127.0.0.2", "proto": "ws"}
assert request.remote_addr == "127.0.0.2"
assert request.scheme == "ws"
assert request.server_port == ASGI_PORT
app.config.FORWARDED_SECRET = "mySecret"
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {
assert response.json == {
"for": "[::2]",
"proto": "https",
"host": "me.tld",
@ -603,13 +605,13 @@ async def test_standard_forwarded_asgi(app):
# Empty Forwarded header -> use X-headers
headers["Forwarded"] = ""
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"for": "127.0.0.2", "proto": "ws"}
assert response.json == {"for": "127.0.0.2", "proto": "ws"}
# Header present but not matching anything
request, response = await app.asgi_client.get(
"/", headers={"Forwarded": "."}
)
assert response.json() == {}
assert response.json == {}
# Forwarded header present but no matching secret -> use X-headers
headers = {
@ -617,13 +619,13 @@ async def test_standard_forwarded_asgi(app):
"X-Real-IP": "127.0.0.2",
}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"for": "127.0.0.2"}
assert response.json == {"for": "127.0.0.2"}
assert request.remote_addr == "127.0.0.2"
# Different formatting and hitting both ends of the header
headers = {"Forwarded": 'Secret="mySecret";For=127.0.0.4;Port=1234'}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {
assert response.json == {
"for": "127.0.0.4",
"port": 1234,
"secret": "mySecret",
@ -632,7 +634,7 @@ async def test_standard_forwarded_asgi(app):
# Test escapes (modify this if you see anyone implementing quoted-pairs)
headers = {"Forwarded": 'for=test;quoted="\\,x=x;y=\\";secret=mySecret'}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {
assert response.json == {
"for": "test",
"quoted": "\\,x=x;y=\\",
"secret": "mySecret",
@ -641,17 +643,17 @@ async def test_standard_forwarded_asgi(app):
# Secret insulated by malformed field #1
headers = {"Forwarded": "for=test;secret=mySecret;b0rked;proto=wss;"}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"for": "test", "secret": "mySecret"}
assert response.json == {"for": "test", "secret": "mySecret"}
# Secret insulated by malformed field #2
headers = {"Forwarded": "for=test;b0rked;secret=mySecret;proto=wss"}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"proto": "wss", "secret": "mySecret"}
assert response.json == {"proto": "wss", "secret": "mySecret"}
# Unexpected termination should not lose existing acceptable values
headers = {"Forwarded": "b0rked;secret=mySecret;proto=wss"}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"proto": "wss", "secret": "mySecret"}
assert response.json == {"proto": "wss", "secret": "mySecret"}
# Field normalization
headers = {
@ -659,7 +661,7 @@ async def test_standard_forwarded_asgi(app):
'PATH="/With%20Spaces%22Quoted%22/sanicApp?key=val";SECRET=mySecret'
}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {
assert response.json == {
"proto": "wss",
"by": "[cafe::8000]",
"host": "a:2",
@ -671,7 +673,10 @@ async def test_standard_forwarded_asgi(app):
app.config.FORWARDED_SECRET = "_proxySecret"
headers = {"Forwarded": "for=1.2.3.4; by=_proxySecret"}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"for": "1.2.3.4", "by": "_proxySecret"}
assert response.json == {
"for": "1.2.3.4",
"by": "_proxySecret",
}
def test_remote_addr_with_two_proxies(app):
@ -685,33 +690,33 @@ def test_remote_addr_with_two_proxies(app):
headers = {"X-Real-IP": "127.0.0.2", "X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.2"
assert response.text == "127.0.0.2"
assert response.body == b"127.0.0.2"
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"X-Forwarded-For": "127.0.0.1, 127.0.1.2"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
assert response.body == b"127.0.0.1"
request, response = app.test_client.get("/")
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"X-Forwarded-For": "127.0.0.1, , ,,127.0.1.2"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
assert response.body == b"127.0.0.1"
headers = {
"X-Forwarded-For": ", 127.0.2.2, , ,127.0.0.1, , ,,127.0.1.2"
}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
assert response.body == b"127.0.0.1"
@pytest.mark.asyncio
@ -726,33 +731,33 @@ async def test_remote_addr_with_two_proxies_asgi(app):
headers = {"X-Real-IP": "127.0.0.2", "X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.2"
assert response.text == "127.0.0.2"
assert response.body == b"127.0.0.2"
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"X-Forwarded-For": "127.0.0.1, 127.0.1.2"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
assert response.body == b"127.0.0.1"
request, response = await app.asgi_client.get("/")
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"X-Forwarded-For": "127.0.0.1, , ,,127.0.1.2"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
assert response.body == b"127.0.0.1"
headers = {
"X-Forwarded-For": ", 127.0.2.2, , ,127.0.0.1, , ,,127.0.1.2"
}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
assert response.body == b"127.0.0.1"
def test_remote_addr_without_proxy(app):
@ -765,17 +770,17 @@ def test_remote_addr_without_proxy(app):
headers = {"X-Real-IP": "127.0.0.2", "X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"X-Forwarded-For": "127.0.0.1, 127.0.1.2"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
@pytest.mark.asyncio
@ -789,17 +794,17 @@ async def test_remote_addr_without_proxy_asgi(app):
headers = {"X-Real-IP": "127.0.0.2", "X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"X-Forwarded-For": "127.0.0.1, 127.0.1.2"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
def test_remote_addr_custom_headers(app):
@ -814,17 +819,17 @@ def test_remote_addr_custom_headers(app):
headers = {"X-Real-IP": "127.0.0.2", "Forwarded": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.1.1"
assert response.text == "127.0.1.1"
assert response.body == b"127.0.1.1"
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"Client-IP": "127.0.0.2", "Forwarded": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.2"
assert response.text == "127.0.0.2"
assert response.body == b"127.0.0.2"
@pytest.mark.asyncio
@ -840,17 +845,17 @@ async def test_remote_addr_custom_headers_asgi(app):
headers = {"X-Real-IP": "127.0.0.2", "Forwarded": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.1.1"
assert response.text == "127.0.1.1"
assert response.body == b"127.0.1.1"
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
assert response.body == b""
headers = {"Client-IP": "127.0.0.2", "Forwarded": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.2"
assert response.text == "127.0.0.2"
assert response.body == b"127.0.0.2"
def test_forwarded_scheme(app):
@ -894,7 +899,7 @@ async def test_match_info_asgi(app):
request, response = await app.asgi_client.get("/api/v1/user/sanic_user/")
assert request.match_info == {"user_id": "sanic_user"}
assert json_loads(response.text) == {"user_id": "sanic_user"}
assert json_loads(response.body) == {"user_id": "sanic_user"}
# ------------------------------------------------------------ #
@ -916,7 +921,7 @@ def test_post_json(app):
assert request.json.get("test") == "OK"
assert request.json.get("test") == "OK" # for request.parsed_json
assert response.text == "OK"
assert response.body == b"OK"
@pytest.mark.asyncio
@ -934,7 +939,7 @@ async def test_post_json_asgi(app):
assert request.json.get("test") == "OK"
assert request.json.get("test") == "OK" # for request.parsed_json
assert response.text == "OK"
assert response.body == b"OK"
def test_post_form_urlencoded(app):
@ -2136,7 +2141,7 @@ def test_safe_method_with_body_ignored(app):
assert request.body == b""
assert request.json == None
assert response.text == "OK"
assert response.body == b"OK"
def test_safe_method_with_body(app):
@ -2153,4 +2158,4 @@ def test_safe_method_with_body(app):
assert request.body == data.encode("utf-8")
assert request.json.get("test") == "OK"
assert response.text == "OK"
assert response.body == b"OK"

27
tests/test_response.py
View File

@ -14,6 +14,7 @@ import pytest
from aiofiles import os as async_os
from sanic_testing.testing import HOST, PORT
from sanic import Sanic
from sanic.response import (
HTTPResponse,
StreamingHTTPResponse,
@ -51,16 +52,22 @@ async def sample_streaming_fn(response):
await response.write("bar")
def test_method_not_allowed(app):
def test_method_not_allowed():
app = Sanic("app")
@app.get("/")
async def test_get(request):
return response.json({"hello": "world"})
request, response = app.test_client.head("/")
assert response.headers["Allow"] == "GET"
assert set(response.headers["Allow"].split(", ")) == {
"GET",
}
request, response = app.test_client.post("/")
assert response.headers["Allow"] == "GET"
assert set(response.headers["Allow"].split(", ")) == {"GET", "HEAD"}
app.router.reset()
@app.post("/")
async def test_post(request):
@ -68,12 +75,20 @@ def test_method_not_allowed(app):
request, response = app.test_client.head("/")
assert response.status == 405
assert set(response.headers["Allow"].split(", ")) == {"GET", "POST"}
assert set(response.headers["Allow"].split(", ")) == {
"GET",
"POST",
"HEAD",
}
assert response.headers["Content-Length"] == "0"
request, response = app.test_client.patch("/")
assert response.status == 405
assert set(response.headers["Allow"].split(", ")) == {"GET", "POST"}
assert set(response.headers["Allow"].split(", ")) == {
"GET",
"POST",
"HEAD",
}
assert response.headers["Content-Length"] == "0"
@ -237,7 +252,7 @@ def test_chunked_streaming_returns_correct_content(streaming_app):
@pytest.mark.asyncio
async def test_chunked_streaming_returns_correct_content_asgi(streaming_app):
request, response = await streaming_app.asgi_client.get("/")
assert response.text == "foo,bar"
assert response.body == b"foo,bar"
def test_non_chunked_streaming_adds_correct_headers(non_chunked_streaming_app):

237
tests/test_routes.py
View File

@ -1,18 +1,180 @@
import asyncio
from unittest.mock import Mock
import pytest
from sanic_routing.exceptions import ParameterNameConflicts, RouteExists
from sanic_testing.testing import SanicTestClient
from sanic import Sanic
from sanic import Blueprint, Sanic
from sanic.constants import HTTP_METHODS
from sanic.exceptions import NotFound
from sanic.request import Request
from sanic.response import json, text
from sanic.router import ParameterNameConflicts, RouteDoesNotExist, RouteExists
# ------------------------------------------------------------ #
# UTF-8
# ------------------------------------------------------------ #
@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=request)
except NotFound:
response = 404
except Exception:
response = 500
else:
response = 200
assert response == expected
# # ------------------------------------------------------------ #
# # UTF-8
# # ------------------------------------------------------------ #
@pytest.mark.parametrize("method", HTTP_METHODS)
@ -164,7 +326,6 @@ def test_route_optional_slash(app):
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}"
@ -176,6 +337,8 @@ def test_route_strict_slashes_set_to_false_and_host_is_a_list(app):
request, response = test_client.get("http://" + site1 + "/get")
assert response.text == "OK"
app.router.finalized = False
@app.post("/post", host=[site1, "site2.com"], strict_slashes=False)
def post_handler(request):
return text("OK")
@ -183,6 +346,8 @@ def test_route_strict_slashes_set_to_false_and_host_is_a_list(app):
request, response = test_client.post("http://" + site1 + "/post")
assert response.text == "OK"
app.router.finalized = False
@app.put("/put", host=[site1, "site2.com"], strict_slashes=False)
def put_handler(request):
return text("OK")
@ -190,6 +355,8 @@ def test_route_strict_slashes_set_to_false_and_host_is_a_list(app):
request, response = test_client.put("http://" + site1 + "/put")
assert response.text == "OK"
app.router.finalized = False
@app.delete("/delete", host=[site1, "site2.com"], strict_slashes=False)
def delete_handler(request):
return text("OK")
@ -294,6 +461,8 @@ def test_dynamic_route(app):
results.append(name)
return text("OK")
app.router.finalize(False)
request, response = app.test_client.get("/folder/test123")
assert response.text == "OK"
@ -368,6 +537,9 @@ def test_dynamic_route_regex(app):
async def handler(request, folder_id):
return text("OK")
app.router.finalize()
print(app.router.find_route_src)
request, response = app.test_client.get("/folder/test")
assert response.status == 200
@ -415,6 +587,8 @@ def test_dynamic_route_path(app):
request, response = app.test_client.get("/info")
assert response.status == 404
app.router.reset()
@app.route("/<path:path>")
async def handler1(request, path):
return text("OK")
@ -774,7 +948,7 @@ def test_removing_slash(app):
def post(_):
pass
assert len(app.router.routes_all.keys()) == 2
assert len(app.router.routes_all.keys()) == 1
def test_overload_routes(app):
@ -798,6 +972,7 @@ def test_overload_routes(app):
request, response = app.test_client.delete("/overload")
assert response.status == 405
app.router.reset()
with pytest.raises(RouteExists):
@app.route("/overload", methods=["PUT", "DELETE"])
@ -810,11 +985,18 @@ def test_unmergeable_overload_routes(app):
async def handler1(request):
return text("OK1")
with pytest.raises(RouteExists):
@app.route("/overload_whole", methods=["POST", "PUT"])
async def handler2(request):
return text("Duplicated")
return text("OK1")
assert (
len(
dict(list(app.router.static_routes.values())[0].handlers)[
"overload_whole"
]
)
== 3
)
request, response = app.test_client.get("/overload_whole")
assert response.text == "OK1"
@ -822,6 +1004,11 @@ def test_unmergeable_overload_routes(app):
request, response = app.test_client.post("/overload_whole")
assert response.text == "OK1"
request, response = app.test_client.put("/overload_whole")
assert response.text == "OK1"
app.router.reset()
@app.route("/overload_part", methods=["GET"])
async def handler3(request):
return text("OK1")
@ -847,7 +1034,9 @@ def test_unicode_routes(app):
request, response = app.test_client.get("/你好")
assert response.text == "OK1"
@app.route("/overload/<param>", methods=["GET"])
app.router.reset()
@app.route("/overload/<param>", methods=["GET"], unquote=True)
async def handler2(request, param):
return text("OK2 " + param)
@ -865,21 +1054,39 @@ def test_uri_with_different_method_and_different_params(app):
return json({"action": action})
request, response = app.test_client.get("/ads/1234")
assert response.status == 200
assert response.json == {"ad_id": "1234"}
assert response.status == 405
request, response = app.test_client.post("/ads/post")
assert response.status == 200
assert response.json == {"action": "post"}
def test_route_raise_ParameterNameConflicts(app):
with pytest.raises(ParameterNameConflicts):
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"}
def test_route_raise_ParameterNameConflicts(app):
@app.get("/api/v1/<user>/<user>/")
def handler(request, user):
return text("OK")
with pytest.raises(ParameterNameConflicts):
app.router.finalize()
def test_route_invalid_host(app):

1
tests/test_static.py
View File

@ -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")

45
tests/test_url_building.py
View File

@ -7,6 +7,7 @@ import pytest as pytest
from sanic_testing.testing import HOST as test_host
from sanic_testing.testing import PORT as test_port
from sanic import Sanic
from sanic.blueprints import Blueprint
from sanic.exceptions import URLBuildError
from sanic.response import text
@ -98,36 +99,36 @@ def test_url_for_with_server_name(app):
assert response.text == "this should pass"
def test_fails_if_endpoint_not_found(app):
def test_fails_if_endpoint_not_found():
app = Sanic("app")
@app.route("/fail")
def fail(request):
return text("this should fail")
with pytest.raises(URLBuildError) as e:
app.url_for("passes")
assert str(e.value) == "Endpoint with name `passes` was not found"
e.match("Endpoint with name `app.passes` was not found")
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 +138,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 +168,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 +199,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 +215,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):
@ -259,7 +253,8 @@ def test_adds_other_supplied_values_as_query_string(app):
@pytest.fixture
def blueprint_app(app):
def blueprint_app():
app = Sanic("app")
first_print = Blueprint("first", url_prefix="/first")
second_print = Blueprint("second", url_prefix="/second")
@ -273,11 +268,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 +285,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 +293,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"

29
tests/test_url_for.py
View File

@ -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

79
tests/test_url_for_static.py
View File

@ -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

14
tests/test_vhosts.py
View File

@ -1,7 +1,14 @@
import pytest
from sanic_routing.exceptions import RouteExists
from sanic import Sanic
from sanic.response import text
def test_vhosts(app):
def test_vhosts():
app = Sanic("app")
@app.route("/", host="example.com")
async def handler1(request):
return text("You're at example.com!")
@ -38,6 +45,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 +54,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"

19
tests/test_views.py
View File

@ -45,9 +45,9 @@ def test_unexisting_methods(app):
app.add_route(DummyView.as_view(), "/")
request, response = app.test_client.get("/")
assert response.text == "I am get method"
assert response.body == b"I am get method"
request, response = app.test_client.post("/")
assert "Method POST not allowed for URL /" in response.text
assert b"Method POST not allowed for URL /" in response.body
def test_argument_methods(app):
@ -215,17 +215,18 @@ def test_composition_view_runs_methods_as_expected(app, method):
if method in ["GET", "POST", "PUT"]:
request, response = getattr(app.test_client, method.lower())("/")
assert response.status == 200
assert response.text == "first method"
response = view(request)
assert response.body.decode() == "first method"
# response = view(request)
# assert response.body.decode() == "first method"
if method in ["DELETE", "PATCH"]:
request, response = getattr(app.test_client, method.lower())("/")
assert response.text == "second method"
# if method in ["DELETE", "PATCH"]:
# request, response = getattr(app.test_client, method.lower())("/")
# assert response.text == "second method"
response = view(request)
assert response.body.decode() == "second method"
# response = view(request)
# assert response.body.decode() == "second method"
@pytest.mark.parametrize("method", HTTP_METHODS)

14
tests/test_worker.py
View File

@ -9,6 +9,8 @@ from unittest import mock
import pytest
from sanic_testing.testing import ASGI_PORT as PORT
from sanic.app import Sanic
from sanic.worker import GunicornWorker
@ -17,7 +19,7 @@ from sanic.worker import GunicornWorker
def gunicorn_worker():
command = (
"gunicorn "
"--bind 127.0.0.1:1337 "
f"--bind 127.0.0.1:{PORT} "
"--worker-class sanic.worker.GunicornWorker "
"examples.simple_server:app"
)
@ -31,7 +33,7 @@ def gunicorn_worker():
def gunicorn_worker_with_access_logs():
command = (
"gunicorn "
"--bind 127.0.0.1:1338 "
f"--bind 127.0.0.1:{PORT + 1} "
"--worker-class sanic.worker.GunicornWorker "
"examples.simple_server:app"
)
@ -45,7 +47,7 @@ def gunicorn_worker_with_env_var():
command = (
'env SANIC_ACCESS_LOG="False" '
"gunicorn "
"--bind 127.0.0.1:1339 "
f"--bind 127.0.0.1:{PORT + 2} "
"--worker-class sanic.worker.GunicornWorker "
"--log-level info "
"examples.simple_server:app"
@ -56,7 +58,7 @@ def gunicorn_worker_with_env_var():
def test_gunicorn_worker(gunicorn_worker):
with urllib.request.urlopen("http://localhost:1337/") as f:
with urllib.request.urlopen(f"http://localhost:{PORT}/") as f:
res = json.loads(f.read(100).decode())
assert res["test"]
@ -65,7 +67,7 @@ def test_gunicorn_worker_no_logs(gunicorn_worker_with_env_var):
"""
if SANIC_ACCESS_LOG was set to False do not show access logs
"""
with urllib.request.urlopen("http://localhost:1339/") as _:
with urllib.request.urlopen(f"http://localhost:{PORT + 2}/") as _:
gunicorn_worker_with_env_var.kill()
assert not gunicorn_worker_with_env_var.stdout.read()
@ -74,7 +76,7 @@ def test_gunicorn_worker_with_logs(gunicorn_worker_with_access_logs):
"""
default - show access logs
"""
with urllib.request.urlopen("http://localhost:1338/") as _:
with urllib.request.urlopen(f"http://localhost:{PORT + 1}/") as _:
gunicorn_worker_with_access_logs.kill()
assert (
b"(sanic.access)[INFO][127.0.0.1"

4
tox.ini
View File

@ -7,7 +7,7 @@ setenv =
{py36,py37,py38,py39,pyNightly}-no-ext: SANIC_NO_UJSON=1
{py36,py37,py38,py39,pyNightly}-no-ext: SANIC_NO_UVLOOP=1
deps =
sanic-testing==0.1.2
sanic-testing
coverage==5.3
pytest==5.2.1
pytest-cov
@ -35,7 +35,7 @@ deps =
commands =
flake8 sanic
black --config ./.black.toml --check --verbose sanic/
isort --check-only sanic
isort --check-only sanic --profile=black
[testenv:type-checking]
deps =