170 lines
4.9 KiB
Python
170 lines
4.9 KiB
Python
from functools import partial
|
|
|
|
import pytest
|
|
|
|
from sanic import Sanic
|
|
from sanic.middleware import Middleware, MiddlewareLocation
|
|
from sanic.response import json
|
|
|
|
PRIORITY_TEST_CASES = (
|
|
([0, 1, 2], [1, 1, 1]),
|
|
([0, 1, 2], [1, 1, None]),
|
|
([0, 1, 2], [1, None, None]),
|
|
([0, 1, 2], [2, 1, None]),
|
|
([0, 1, 2], [2, 2, None]),
|
|
([0, 1, 2], [3, 2, 1]),
|
|
([0, 1, 2], [None, None, None]),
|
|
([0, 2, 1], [1, None, 1]),
|
|
([0, 2, 1], [2, None, 1]),
|
|
([0, 2, 1], [2, None, 2]),
|
|
([0, 2, 1], [3, 1, 2]),
|
|
([1, 0, 2], [1, 2, None]),
|
|
([1, 0, 2], [2, 3, 1]),
|
|
([1, 0, 2], [None, 1, None]),
|
|
([1, 2, 0], [1, 3, 2]),
|
|
([1, 2, 0], [None, 1, 1]),
|
|
([1, 2, 0], [None, 2, 1]),
|
|
([1, 2, 0], [None, 2, 2]),
|
|
([2, 0, 1], [1, None, 2]),
|
|
([2, 0, 1], [2, 1, 3]),
|
|
([2, 0, 1], [None, None, 1]),
|
|
([2, 1, 0], [1, 2, 3]),
|
|
([2, 1, 0], [None, 1, 2]),
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_middleware():
|
|
yield
|
|
Middleware.reset_count()
|
|
|
|
|
|
def test_add_register_priority(app: Sanic):
|
|
def foo(*_):
|
|
...
|
|
|
|
app.register_middleware(foo, priority=999)
|
|
assert len(app.request_middleware) == 1
|
|
assert len(app.response_middleware) == 0
|
|
assert app.request_middleware[0].priority == 999 # type: ignore
|
|
app.register_middleware(foo, attach_to="response", priority=999)
|
|
assert len(app.request_middleware) == 1
|
|
assert len(app.response_middleware) == 1
|
|
assert app.response_middleware[0].priority == 999 # type: ignore
|
|
|
|
|
|
def test_add_register_named_priority(app: Sanic):
|
|
def foo(*_):
|
|
...
|
|
|
|
app.register_named_middleware(foo, route_names=["foo"], priority=999)
|
|
assert len(app.named_request_middleware) == 1
|
|
assert len(app.named_response_middleware) == 0
|
|
assert app.named_request_middleware["foo"][0].priority == 999 # type: ignore
|
|
app.register_named_middleware(
|
|
foo, attach_to="response", route_names=["foo"], priority=999
|
|
)
|
|
assert len(app.named_request_middleware) == 1
|
|
assert len(app.named_response_middleware) == 1
|
|
assert app.named_response_middleware["foo"][0].priority == 999 # type: ignore
|
|
|
|
|
|
def test_add_decorator_priority(app: Sanic):
|
|
def foo(*_):
|
|
...
|
|
|
|
app.middleware(foo, priority=999)
|
|
assert len(app.request_middleware) == 1
|
|
assert len(app.response_middleware) == 0
|
|
assert app.request_middleware[0].priority == 999 # type: ignore
|
|
app.middleware(foo, attach_to="response", priority=999)
|
|
assert len(app.request_middleware) == 1
|
|
assert len(app.response_middleware) == 1
|
|
assert app.response_middleware[0].priority == 999 # type: ignore
|
|
|
|
|
|
def test_add_convenience_priority(app: Sanic):
|
|
def foo(*_):
|
|
...
|
|
|
|
app.on_request(foo, priority=999)
|
|
assert len(app.request_middleware) == 1
|
|
assert len(app.response_middleware) == 0
|
|
assert app.request_middleware[0].priority == 999 # type: ignore
|
|
app.on_response(foo, priority=999)
|
|
assert len(app.request_middleware) == 1
|
|
assert len(app.response_middleware) == 1
|
|
assert app.response_middleware[0].priority == 999 # type: ignore
|
|
|
|
|
|
def test_add_conflicting_priority(app: Sanic):
|
|
def foo(*_):
|
|
...
|
|
|
|
middleware = Middleware(foo, MiddlewareLocation.REQUEST, priority=998)
|
|
app.register_middleware(middleware=middleware, priority=999)
|
|
assert app.request_middleware[0].priority == 999 # type: ignore
|
|
middleware.priority == 998
|
|
|
|
|
|
def test_add_conflicting_priority_named(app: Sanic):
|
|
def foo(*_):
|
|
...
|
|
|
|
middleware = Middleware(foo, MiddlewareLocation.REQUEST, priority=998)
|
|
app.register_named_middleware(
|
|
middleware=middleware, route_names=["foo"], priority=999
|
|
)
|
|
assert app.named_request_middleware["foo"][0].priority == 999 # type: ignore
|
|
middleware.priority == 998
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"expected,priorities",
|
|
PRIORITY_TEST_CASES,
|
|
)
|
|
def test_request_middleware_order_priority(app: Sanic, expected, priorities):
|
|
order = []
|
|
|
|
def add_ident(request, ident):
|
|
order.append(ident)
|
|
|
|
@app.get("/")
|
|
def handler(request):
|
|
return json(None)
|
|
|
|
for ident, priority in enumerate(priorities):
|
|
kwargs = {}
|
|
if priority is not None:
|
|
kwargs["priority"] = priority
|
|
app.on_request(partial(add_ident, ident=ident), **kwargs)
|
|
|
|
app.test_client.get("/")
|
|
|
|
assert order == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"expected,priorities",
|
|
PRIORITY_TEST_CASES,
|
|
)
|
|
def test_response_middleware_order_priority(app: Sanic, expected, priorities):
|
|
order = []
|
|
|
|
def add_ident(request, response, ident):
|
|
order.append(ident)
|
|
|
|
@app.get("/")
|
|
def handler(request):
|
|
return json(None)
|
|
|
|
for ident, priority in enumerate(priorities):
|
|
kwargs = {}
|
|
if priority is not None:
|
|
kwargs["priority"] = priority
|
|
app.on_response(partial(add_ident, ident=ident), **kwargs)
|
|
|
|
app.test_client.get("/")
|
|
|
|
assert order[::-1] == expected
|