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

Merge branch 'main' into request-contextvars

Browse Source
This commit is contained in:
Adam Hopkins 2022-06-16 16:25:48 +03:00 committed by GitHub
commit f64e917746
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 425 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.black.toml
View File

@ -1,2 +0,0 @@
[tool.black]
line-length = 79

6
Makefile
View File

@ -66,15 +66,15 @@ ifdef include_tests
isort -rc sanic tests
else
$(info Sorting Imports)
isort -rc sanic tests --profile=black
isort -rc sanic tests
endif
endif
black:
black --config ./.black.toml sanic tests
black sanic tests
isort:
isort sanic tests --profile=black
isort sanic tests
pretty: black isort

2
README.rst
View File

@ -114,7 +114,7 @@ Hello World Example
from sanic import Sanic
from sanic.response import json
app = Sanic("My Hello, world app")
app = Sanic("my-hello-world-app")
@app.route('/')
async def test(request):

15
pyproject.toml
View File

@ -1,3 +1,18 @@
[build-system]
requires = ["setuptools<60.0", "wheel"]
build-backend = "setuptools.build_meta"
[tool.black]
line-length = 79
[tool.isort]
atomic = true
default_section = "THIRDPARTY"
include_trailing_comma = true
known_first_party = "sanic"
known_third_party = "pytest"
line_length = 79
lines_after_imports = 2
lines_between_types = 1
multi_line_output = 3
profile = "black"

11
sanic/app.py
View File

@ -97,7 +97,7 @@ if TYPE_CHECKING: # no cov
from sanic_ext import Extend # type: ignore
from sanic_ext.extensions.base import Extension # type: ignore
except ImportError:
Extend = TypeVar("Extend") # type: ignore
Extend = TypeVar("Extend", Type) # type: ignore
if OS_IS_WINDOWS: # no cov
@ -992,10 +992,10 @@ class Sanic(BaseSanic, RunnerMixin, metaclass=TouchUpMeta):
cancelled = False
try:
await fut
except Exception as e:
self.error_handler.log(request, e)
except (CancelledError, ConnectionClosed):
cancelled = True
except Exception as e:
self.error_handler.log(request, e)
finally:
self.websocket_tasks.remove(fut)
if cancelled:
@ -1573,8 +1573,9 @@ class Sanic(BaseSanic, RunnerMixin, metaclass=TouchUpMeta):
"shutdown",
):
raise SanicException(f"Invalid server event: {event}")
if self.state.verbosity >= 1:
logger.debug(f"Triggering server events: {event}")
logger.debug(
f"Triggering server events: {event}", extra={"verbosity": 1}
)
reverse = concern == "shutdown"
if loop is None:
loop = self.loop

5
sanic/application/state.py
View File

@ -9,7 +9,7 @@ from socket import socket
from ssl import SSLContext
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union
from sanic.log import logger
from sanic.log import VerbosityFilter, logger
from sanic.server.async_server import AsyncioServer
@ -91,6 +91,9 @@ class ApplicationState:
if getattr(self.app, "configure_logging", False) and self.app.debug:
logger.setLevel(logging.DEBUG)
def set_verbosity(self, value: int):
VerbosityFilter.verbosity = value
@property
def is_debug(self):
return self.mode is Mode.DEBUG

50
sanic/asgi.py
View File

@ -25,27 +25,28 @@ class Lifespan:
def __init__(self, asgi_app: ASGIApp) -> None:
self.asgi_app = asgi_app
if self.asgi_app.sanic_app.state.verbosity > 0:
if (
"server.init.before"
in self.asgi_app.sanic_app.signal_router.name_index
):
logger.debug(
'You have set a listener for "before_server_start" '
"in ASGI mode. "
"It will be executed as early as possible, but not before "
"the ASGI server is started."
)
if (
"server.shutdown.after"
in self.asgi_app.sanic_app.signal_router.name_index
):
logger.debug(
'You have set a listener for "after_server_stop" '
"in ASGI mode. "
"It will be executed as late as possible, but not after "
"the ASGI server is stopped."
)
if (
"server.init.before"
in self.asgi_app.sanic_app.signal_router.name_index
):
logger.debug(
'You have set a listener for "before_server_start" '
"in ASGI mode. "
"It will be executed as early as possible, but not before "
"the ASGI server is started.",
extra={"verbosity": 1},
)
if (
"server.shutdown.after"
in self.asgi_app.sanic_app.signal_router.name_index
):
logger.debug(
'You have set a listener for "after_server_stop" '
"in ASGI mode. "
"It will be executed as late as possible, but not after "
"the ASGI server is stopped.",
extra={"verbosity": 1},
)
async def startup(self) -> None:
"""
@ -163,6 +164,13 @@ class ASGIApp:
instance.request_body = True
instance.request.conn_info = ConnInfo(instance.transport)
await sanic_app.dispatch(
"http.lifecycle.request",
inline=True,
context={"request": instance.request},
fail_not_found=False,
)
return instance
async def read(self) -> Optional[bytes]:

17
sanic/log.py
View File

@ -61,21 +61,36 @@ class Colors(str, Enum): # no cov
END = "\033[0m"
BLUE = "\033[01;34m"
GREEN = "\033[01;32m"
YELLOW = "\033[01;33m"
PURPLE = "\033[01;35m"
RED = "\033[01;31m"
SANIC = "\033[38;2;255;13;104m"
YELLOW = "\033[01;33m"
class VerbosityFilter(logging.Filter):
verbosity: int = 0
def filter(self, record: logging.LogRecord) -> bool:
verbosity = getattr(record, "verbosity", 0)
return verbosity <= self.verbosity
_verbosity_filter = VerbosityFilter()
logger = logging.getLogger("sanic.root") # no cov
logger.addFilter(_verbosity_filter)
"""
General Sanic logger
"""
error_logger = logging.getLogger("sanic.error") # no cov
error_logger.addFilter(_verbosity_filter)
"""
Logger used by Sanic for error logging
"""
access_logger = logging.getLogger("sanic.access") # no cov
access_logger.addFilter(_verbosity_filter)
"""
Logger used by Sanic for access logging
"""

201
sanic/mixins/routes.py
View File

@ -8,7 +8,17 @@ from pathlib import PurePath
from re import sub
from textwrap import dedent
from time import gmtime, strftime
from typing import Any, Callable, Iterable, List, Optional, Set, Tuple, Union
from typing import (
Any,
Callable,
Iterable,
List,
Optional,
Set,
Tuple,
Union,
cast,
)
from urllib.parse import unquote
from sanic_routing.route import Route # type: ignore
@ -283,7 +293,7 @@ class RouteMixin(metaclass=SanicMeta):
version_prefix: str = "/v",
error_format: Optional[str] = None,
**ctx_kwargs,
) -> RouteWrapper:
) -> RouteHandler:
"""
Add an API URL under the **GET** *HTTP* method
@ -299,17 +309,20 @@ class RouteMixin(metaclass=SanicMeta):
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
return self.route(
uri,
methods=frozenset({"GET"}),
host=host,
strict_slashes=strict_slashes,
version=version,
name=name,
ignore_body=ignore_body,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
return cast(
RouteHandler,
self.route(
uri,
methods=frozenset({"GET"}),
host=host,
strict_slashes=strict_slashes,
version=version,
name=name,
ignore_body=ignore_body,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
),
)
def post(
@ -323,7 +336,7 @@ class RouteMixin(metaclass=SanicMeta):
version_prefix: str = "/v",
error_format: Optional[str] = None,
**ctx_kwargs,
) -> RouteWrapper:
) -> RouteHandler:
"""
Add an API URL under the **POST** *HTTP* method
@ -339,17 +352,20 @@ class RouteMixin(metaclass=SanicMeta):
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
return self.route(
uri,
methods=frozenset({"POST"}),
host=host,
strict_slashes=strict_slashes,
stream=stream,
version=version,
name=name,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
return cast(
RouteHandler,
self.route(
uri,
methods=frozenset({"POST"}),
host=host,
strict_slashes=strict_slashes,
stream=stream,
version=version,
name=name,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
),
)
def put(
@ -363,7 +379,7 @@ class RouteMixin(metaclass=SanicMeta):
version_prefix: str = "/v",
error_format: Optional[str] = None,
**ctx_kwargs,
) -> RouteWrapper:
) -> RouteHandler:
"""
Add an API URL under the **PUT** *HTTP* method
@ -379,17 +395,20 @@ class RouteMixin(metaclass=SanicMeta):
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
return self.route(
uri,
methods=frozenset({"PUT"}),
host=host,
strict_slashes=strict_slashes,
stream=stream,
version=version,
name=name,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
return cast(
RouteHandler,
self.route(
uri,
methods=frozenset({"PUT"}),
host=host,
strict_slashes=strict_slashes,
stream=stream,
version=version,
name=name,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
),
)
def head(
@ -403,7 +422,7 @@ class RouteMixin(metaclass=SanicMeta):
version_prefix: str = "/v",
error_format: Optional[str] = None,
**ctx_kwargs,
) -> RouteWrapper:
) -> RouteHandler:
"""
Add an API URL under the **HEAD** *HTTP* method
@ -427,17 +446,20 @@ class RouteMixin(metaclass=SanicMeta):
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
return self.route(
uri,
methods=frozenset({"HEAD"}),
host=host,
strict_slashes=strict_slashes,
version=version,
name=name,
ignore_body=ignore_body,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
return cast(
RouteHandler,
self.route(
uri,
methods=frozenset({"HEAD"}),
host=host,
strict_slashes=strict_slashes,
version=version,
name=name,
ignore_body=ignore_body,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
),
)
def options(
@ -451,7 +473,7 @@ class RouteMixin(metaclass=SanicMeta):
version_prefix: str = "/v",
error_format: Optional[str] = None,
**ctx_kwargs,
) -> RouteWrapper:
) -> RouteHandler:
"""
Add an API URL under the **OPTIONS** *HTTP* method
@ -475,17 +497,20 @@ class RouteMixin(metaclass=SanicMeta):
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
return self.route(
uri,
methods=frozenset({"OPTIONS"}),
host=host,
strict_slashes=strict_slashes,
version=version,
name=name,
ignore_body=ignore_body,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
return cast(
RouteHandler,
self.route(
uri,
methods=frozenset({"OPTIONS"}),
host=host,
strict_slashes=strict_slashes,
version=version,
name=name,
ignore_body=ignore_body,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
),
)
def patch(
@ -499,7 +524,7 @@ class RouteMixin(metaclass=SanicMeta):
version_prefix: str = "/v",
error_format: Optional[str] = None,
**ctx_kwargs,
) -> RouteWrapper:
) -> RouteHandler:
"""
Add an API URL under the **PATCH** *HTTP* method
@ -525,17 +550,20 @@ class RouteMixin(metaclass=SanicMeta):
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
return self.route(
uri,
methods=frozenset({"PATCH"}),
host=host,
strict_slashes=strict_slashes,
stream=stream,
version=version,
name=name,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
return cast(
RouteHandler,
self.route(
uri,
methods=frozenset({"PATCH"}),
host=host,
strict_slashes=strict_slashes,
stream=stream,
version=version,
name=name,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
),
)
def delete(
@ -549,7 +577,7 @@ class RouteMixin(metaclass=SanicMeta):
version_prefix: str = "/v",
error_format: Optional[str] = None,
**ctx_kwargs,
) -> RouteWrapper:
) -> RouteHandler:
"""
Add an API URL under the **DELETE** *HTTP* method
@ -565,17 +593,20 @@ class RouteMixin(metaclass=SanicMeta):
will be appended to the route context (``route.ctx``)
:return: Object decorated with :func:`route` method
"""
return self.route(
uri,
methods=frozenset({"DELETE"}),
host=host,
strict_slashes=strict_slashes,
version=version,
name=name,
ignore_body=ignore_body,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
return cast(
RouteHandler,
self.route(
uri,
methods=frozenset({"DELETE"}),
host=host,
strict_slashes=strict_slashes,
version=version,
name=name,
ignore_body=ignore_body,
version_prefix=version_prefix,
error_format=error_format,
**ctx_kwargs,
),
)
def websocket(

45
sanic/response.py
View File

@ -1,9 +1,12 @@
from __future__ import annotations
from datetime import datetime
from email.utils import formatdate
from functools import partial
from mimetypes import guess_type
from os import path
from pathlib import PurePath
from pathlib import Path, PurePath
from time import time
from typing import (
TYPE_CHECKING,
Any,
@ -23,7 +26,12 @@ from sanic.compat import Header, open_async
from sanic.constants import DEFAULT_HTTP_CONTENT_TYPE
from sanic.cookies import CookieJar
from sanic.exceptions import SanicException, ServerError
from sanic.helpers import has_message_body, remove_entity_headers
from sanic.helpers import (
Default,
_default,
has_message_body,
remove_entity_headers,
)
from sanic.http import Http
from sanic.models.protocol_types import HTMLProtocol, Range
@ -309,6 +317,9 @@ async def file(
mime_type: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
filename: Optional[str] = None,
last_modified: Optional[Union[datetime, float, int, Default]] = _default,
max_age: Optional[Union[float, int]] = None,
no_store: Optional[bool] = None,
_range: Optional[Range] = None,
) -> HTTPResponse:
"""Return a response object with file data.
@ -317,6 +328,9 @@ async def file(
:param mime_type: Specific mime_type.
:param headers: Custom Headers.
:param filename: Override filename.
:param last_modified: The last modified date and time of the file.
:param max_age: Max age for cache control.
:param no_store: Any cache should not store this response.
:param _range:
"""
headers = headers or {}
@ -324,6 +338,33 @@ async def file(
headers.setdefault(
"Content-Disposition", f'attachment; filename="{filename}"'
)
if isinstance(last_modified, datetime):
last_modified = last_modified.timestamp()
elif isinstance(last_modified, Default):
last_modified = Path(location).stat().st_mtime
if last_modified:
headers.setdefault(
"last-modified", formatdate(last_modified, usegmt=True)
)
if no_store:
cache_control = "no-store"
elif max_age:
cache_control = f"public, max-age={max_age}"
headers.setdefault(
"expires",
formatdate(
time() + max_age,
usegmt=True,
),
)
else:
cache_control = "no-cache"
headers.setdefault("cache-control", cache_control)
filename = filename or path.split(location)[-1]
async with await open_async(location, mode="rb") as f:

7
sanic/server/protocols/http_protocol.py
View File

@ -8,6 +8,8 @@ from sanic.touchup.meta import TouchUpMeta
if TYPE_CHECKING: # no cov
from sanic.app import Sanic
import sys
from asyncio import CancelledError
from time import monotonic as current_time
@ -169,7 +171,10 @@ class HttpProtocol(SanicProtocol, metaclass=TouchUpMeta):
)
self.loop.call_later(max(0.1, interval), self.check_timeouts)
return
self._task.cancel()
cancel_msg_args = ()
if sys.version_info >= (3, 9):
cancel_msg_args = ("Cancel connection task with a timeout",)
self._task.cancel(*cancel_msg_args)
except Exception:
error_logger.exception("protocol.check_timeouts")

13
sanic/touchup/schemes/ode.py
View File

@ -24,9 +24,7 @@ class OptionalDispatchEvent(BaseScheme):
raw_source = getsource(method)
src = dedent(raw_source)
tree = parse(src)
node = RemoveDispatch(
self._registered_events, self.app.state.verbosity
).visit(tree)
node = RemoveDispatch(self._registered_events).visit(tree)
compiled_src = compile(node, method.__name__, "exec")
exec_locals: Dict[str, Any] = {}
exec(compiled_src, module_globals, exec_locals) # nosec
@ -64,9 +62,8 @@ class OptionalDispatchEvent(BaseScheme):
class RemoveDispatch(NodeTransformer):
def __init__(self, registered_events, verbosity: int = 0) -> None:
def __init__(self, registered_events) -> None:
self._registered_events = registered_events
self._verbosity = verbosity
def visit_Expr(self, node: Expr) -> Any:
call = node.value
@ -83,8 +80,10 @@ class RemoveDispatch(NodeTransformer):
if hasattr(event, "s"):
event_name = getattr(event, "value", event.s)
if self._not_registered(event_name):
if self._verbosity >= 2:
logger.debug(f"Disabling event: {event_name}")
logger.debug(
f"Disabling event: {event_name}",
extra={"verbosity": 2},
)
return None
return node

11
setup.cfg
View File

@ -1,13 +1,2 @@
[flake8]
ignore = E203, W503
[isort]
atomic = true
default_section = THIRDPARTY
include_trailing_comma = true
known_first_party = sanic
known_third_party = pytest
line_length = 79
lines_after_imports = 2
lines_between_types = 1
multi_line_output = 3

35
tests/test_asgi.py
View File

@ -9,10 +9,11 @@ import uvicorn
from sanic import Sanic
from sanic.application.state import Mode
from sanic.asgi import MockTransport
from sanic.exceptions import Forbidden, BadRequest, ServiceUnavailable
from sanic.exceptions import BadRequest, Forbidden, ServiceUnavailable
from sanic.request import Request
from sanic.response import json, text
from sanic.server.websockets.connection import WebSocketConnection
from sanic.signals import RESERVED_NAMESPACES
@pytest.fixture
@ -221,6 +222,7 @@ def test_listeners_triggered_async(app, caplog):
assert after_server_stop
app.state.mode = Mode.DEBUG
app.state.verbosity = 0
with caplog.at_level(logging.DEBUG):
server.run()
@ -513,3 +515,34 @@ async def test_request_exception_suppressed_by_middleware(app):
_, response = await app.asgi_client.get("/error-prone")
assert response.status_code == 403
@pytest.mark.asyncio
async def test_signals_triggered(app):
@app.get("/test_signals_triggered")
async def _request(request):
return text("test_signals_triggered")
signals_triggered = []
signals_expected = [
# "http.lifecycle.begin",
# "http.lifecycle.read_head",
"http.lifecycle.request",
"http.lifecycle.handle",
"http.routing.before",
"http.routing.after",
"http.lifecycle.response",
# "http.lifecycle.send",
# "http.lifecycle.complete",
]
def signal_handler(signal):
return lambda *a, **kw: signals_triggered.append(signal)
for signal in RESERVED_NAMESPACES["http"]:
app.signal(signal)(signal_handler(signal))
_, response = await app.asgi_client.get("/test_signals_triggered")
assert response.status_code == 200
assert response.text == "test_signals_triggered"
assert signals_triggered == signals_expected

7
tests/test_blueprint_group.py
View File

@ -3,12 +3,7 @@ from pytest import raises
from sanic.app import Sanic
from sanic.blueprint_group import BlueprintGroup
from sanic.blueprints import Blueprint
from sanic.exceptions import (
Forbidden,
BadRequest,
SanicException,
ServerError,
)
from sanic.exceptions import BadRequest, Forbidden, SanicException, ServerError
from sanic.request import Request
from sanic.response import HTTPResponse, text

7
tests/test_blueprints.py
View File

@ -7,12 +7,7 @@ import pytest
from sanic.app import Sanic
from sanic.blueprints import Blueprint
from sanic.constants import HTTP_METHODS
from sanic.exceptions import (
BadRequest,
NotFound,
SanicException,
ServerError,
)
from sanic.exceptions import BadRequest, NotFound, SanicException, ServerError
from sanic.request import Request
from sanic.response import json, text

2
tests/test_exceptions_handler.py
View File

@ -10,7 +10,7 @@ from bs4 import BeautifulSoup
from pytest import LogCaptureFixture, MonkeyPatch
from sanic import Sanic, handlers
from sanic.exceptions import Forbidden, BadRequest, NotFound, ServerError
from sanic.exceptions import BadRequest, Forbidden, NotFound, ServerError
from sanic.handlers import ErrorHandler
from sanic.request import Request
from sanic.response import stream, text

39
tests/test_logging.py
View File

@ -209,3 +209,42 @@ def test_access_log_client_ip_reqip(monkeypatch):
"request": f"GET {request.scheme}://{request.host}/",
},
)
@pytest.mark.parametrize(
"app_verbosity,log_verbosity,exists",
(
(0, 0, True),
(0, 1, False),
(0, 2, False),
(1, 0, True),
(1, 1, True),
(1, 2, False),
(2, 0, True),
(2, 1, True),
(2, 2, True),
),
)
def test_verbosity(app, caplog, app_verbosity, log_verbosity, exists):
rand_string = str(uuid.uuid4())
@app.get("/")
def log_info(request):
logger.info("DEFAULT")
logger.info(rand_string, extra={"verbosity": log_verbosity})
return text("hello")
with caplog.at_level(logging.INFO):
_ = app.test_client.get(
"/", server_kwargs={"verbosity": app_verbosity}
)
record = ("sanic.root", logging.INFO, rand_string)
if exists:
assert record in caplog.record_tuples
else:
assert record not in caplog.record_tuples
if app_verbosity == 0:
assert ("sanic.root", logging.INFO, "DEFAULT") in caplog.record_tuples

105
tests/test_response.py
View File

@ -3,10 +3,13 @@ import inspect
import os
from collections import namedtuple
from datetime import datetime
from email.utils import formatdate
from logging import ERROR, LogRecord
from mimetypes import guess_type
from pathlib import Path
from random import choice
from typing import Callable, List
from typing import Callable, List, Union
from urllib.parse import unquote
import pytest
@ -328,12 +331,27 @@ def static_file_directory():
return static_directory
def get_file_content(static_file_directory, file_name):
def path_str_to_path_obj(static_file_directory: Union[Path, str]):
if isinstance(static_file_directory, str):
static_file_directory = Path(static_file_directory)
return static_file_directory
def get_file_content(static_file_directory: Union[Path, str], file_name: str):
"""The content of the static file to check"""
with open(os.path.join(static_file_directory, file_name), "rb") as file:
static_file_directory = path_str_to_path_obj(static_file_directory)
with open(static_file_directory / file_name, "rb") as file:
return file.read()
def get_file_last_modified_timestamp(
static_file_directory: Union[Path, str], file_name: str
):
"""The content of the static file to check"""
static_file_directory = path_str_to_path_obj(static_file_directory)
return (static_file_directory / file_name).stat().st_mtime
@pytest.mark.parametrize(
"file_name", ["test.file", "decode me.txt", "python.png"]
)
@ -711,3 +729,84 @@ def send_response_after_eof_should_fail(
assert "foo, " in response.text
assert message_in_records(caplog.records, error_msg1)
assert message_in_records(caplog.records, error_msg2)
@pytest.mark.parametrize(
"file_name", ["test.file", "decode me.txt", "python.png"]
)
def test_file_response_headers(
app: Sanic, file_name: str, static_file_directory: str
):
test_last_modified = datetime.now()
test_max_age = 10
test_expires = test_last_modified.timestamp() + test_max_age
@app.route("/files/cached/<filename>", methods=["GET"])
def file_route_cache(request, filename):
file_path = (Path(static_file_directory) / file_name).absolute()
return file(
file_path, max_age=test_max_age, last_modified=test_last_modified
)
@app.route(
"/files/cached_default_last_modified/<filename>", methods=["GET"]
)
def file_route_cache_default_last_modified(request, filename):
file_path = (Path(static_file_directory) / file_name).absolute()
return file(file_path, max_age=test_max_age)
@app.route("/files/no_cache/<filename>", methods=["GET"])
def file_route_no_cache(request, filename):
file_path = (Path(static_file_directory) / file_name).absolute()
return file(file_path)
@app.route("/files/no_store/<filename>", methods=["GET"])
def file_route_no_store(request, filename):
file_path = (Path(static_file_directory) / file_name).absolute()
return file(file_path, no_store=True)
_, response = app.test_client.get(f"/files/cached/{file_name}")
assert response.body == get_file_content(static_file_directory, file_name)
headers = response.headers
assert (
"cache-control" in headers
and f"max-age={test_max_age}" in headers.get("cache-control")
and f"public" in headers.get("cache-control")
)
assert (
"expires" in headers
and headers.get("expires")[:-6]
== formatdate(test_expires, usegmt=True)[:-6]
# [:-6] to allow at most 1 min difference
# It's minimal for cases like:
# Thu, 26 May 2022 05:36:49 GMT
# AND
# Thu, 26 May 2022 05:36:50 GMT
)
assert "last-modified" in headers and headers.get(
"last-modified"
) == formatdate(test_last_modified.timestamp(), usegmt=True)
_, response = app.test_client.get(
f"/files/cached_default_last_modified/{file_name}"
)
file_last_modified = get_file_last_modified_timestamp(
static_file_directory, file_name
)
headers = response.headers
assert "last-modified" in headers and headers.get(
"last-modified"
) == formatdate(file_last_modified, usegmt=True)
_, response = app.test_client.get(f"/files/no_cache/{file_name}")
headers = response.headers
assert "cache-control" in headers and f"no-cache" == headers.get(
"cache-control"
)
_, response = app.test_client.get(f"/files/no_store/{file_name}")
headers = response.headers
assert "cache-control" in headers and f"no-store" == headers.get(
"cache-control"
)

4
tox.ini
View File

@ -19,8 +19,8 @@ commands =
[testenv:lint]
commands =
flake8 sanic
black --config ./.black.toml --check --verbose sanic/
isort --check-only sanic --profile=black
black --check --verbose sanic/
isort --check-only sanic
slotscheck --verbose -m sanic
[testenv:type-checking]