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

Merge branch 'main' into zhiwei/route-overwrite

Browse Source
This commit is contained in:
Zhiwei
2023-03-29 00:12:04 -04:00
committed by GitHub
parent 158cf16964 6eaab2a7e5
commit 4536fd0559
53 changed files with 1774 additions and 1014 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/fake/server.py
View File

@@ -49,6 +49,6 @@ def create_app_with_args(args):
try:
logger.info(f"foo={args.foo}")
except AttributeError:
logger.info(f"module={args.module}")
logger.info(f"target={args.target}")
return app

47
tests/http3/test_http_receiver.py
View File

@@ -11,7 +11,7 @@ from aioquic.quic.events import ProtocolNegotiated
from sanic import Request, Sanic
from sanic.compat import Header
from sanic.config import DEFAULT_CONFIG
from sanic.exceptions import PayloadTooLarge
from sanic.exceptions import BadRequest, PayloadTooLarge
from sanic.http.constants import Stage
from sanic.http.http3 import Http3, HTTPReceiver
from sanic.models.server_types import ConnInfo
@@ -292,3 +292,48 @@ def test_request_conn_info(app):
receiver = http3.get_receiver_by_stream_id(1)
assert isinstance(receiver.request.conn_info, ConnInfo)
def test_request_header_encoding(app):
protocol = generate_protocol(app)
http3 = Http3(protocol, protocol.transmit)
with pytest.raises(BadRequest) as exc_info:
http3.http_event_received(
HeadersReceived(
[
(b":method", b"GET"),
(b":path", b"/location"),
(b":scheme", b"https"),
(b":authority", b"localhost:8443"),
("foo\u00A0".encode(), b"bar"),
],
1,
False,
)
)
assert exc_info.value.status_code == 400
assert (
str(exc_info.value)
== "Header names may only contain US-ASCII characters."
)
def test_request_url_encoding(app):
protocol = generate_protocol(app)
http3 = Http3(protocol, protocol.transmit)
with pytest.raises(BadRequest) as exc_info:
http3.http_event_received(
HeadersReceived(
[
(b":method", b"GET"),
(b":path", b"/location\xA0"),
(b":scheme", b"https"),
(b":authority", b"localhost:8443"),
(b"foo", b"bar"),
],
1,
False,
)
)
assert exc_info.value.status_code == 400
assert str(exc_info.value) == "URL may only contain US-ASCII characters."

17
tests/test_app.py
View File

@@ -448,7 +448,7 @@ def test_custom_context():
@pytest.mark.parametrize("use", (False, True))
def test_uvloop_config(app: Sanic, monkeypatch, use):
@app.get("/test")
@app.get("/test", name="test")
def handler(request):
return text("ok")
@@ -571,21 +571,6 @@ def test_cannot_run_single_process_and_workers_or_auto_reload(
app.run(single_process=True, **extra)
def test_cannot_run_single_process_and_legacy(app: Sanic):
message = "Cannot run single process and legacy mode"
with pytest.raises(RuntimeError, match=message):
app.run(single_process=True, legacy=True)
def test_cannot_run_without_sys_signals_with_workers(app: Sanic):
message = (
"Cannot run Sanic.serve with register_sys_signals=False. "
"Use either Sanic.serve_single or Sanic.serve_legacy."
)
with pytest.raises(RuntimeError, match=message):
app.run(register_sys_signals=False, single_process=False, legacy=False)
def test_default_configure_logging():
with patch("sanic.app.logging") as mock:
Sanic("Test")

14
tests/test_asgi.py
View File

@@ -652,3 +652,17 @@ async def test_asgi_headers_decoding(app: Sanic, monkeypatch: MonkeyPatch):
_, response = await app.asgi_client.get("/", headers={"Test-Header": "😅"})
assert response.status_code == 200
@pytest.mark.asyncio
async def test_asgi_url_decoding(app):
@app.get("/dir/<name>", unquote=True)
def _request(request: Request, name):
return text(name)
# 2F should not become a path separator (unquoted later)
_, response = await app.asgi_client.get("/dir/some%2Fpath")
assert response.text == "some/path"
_, response = await app.asgi_client.get("/dir/some%F0%9F%98%80path")
assert response.text == "some😀path"

8
tests/test_blueprint_copy.py
View File

@@ -67,6 +67,14 @@ def test_bp_copy(app: Sanic):
_, response = app.test_client.get("/version6/page")
assert "Hello world!" in response.text
route_names = [route.name for route in app.router.routes]
assert "test_bp_copy.test_bp1.handle_request" in route_names
assert "test_bp_copy.test_bp2.handle_request" in route_names
assert "test_bp_copy.test_bp3.handle_request" in route_names
assert "test_bp_copy.test_bp4.handle_request" in route_names
assert "test_bp_copy.test_bp5.handle_request" in route_names
assert "test_bp_copy.test_bp6.handle_request" in route_names
def test_bp_copy_with_route_overwriting(app: Sanic):
bpv1 = Blueprint("bp_v1", version=1)

4
tests/test_blueprints.py
View File

@@ -303,6 +303,10 @@ def test_bp_with_host_list(app: Sanic):
assert response.text == "Hello subdomain!"
route_names = [r.name for r in app.router.routes]
assert "test_bp_with_host_list.test_bp_host.handler1" in route_names
assert "test_bp_with_host_list.test_bp_host.handler2" in route_names
def test_several_bp_with_host_list(app: Sanic):
bp = Blueprint(

36
tests/test_cli.py
View File

@@ -43,8 +43,10 @@ def read_app_info(lines: List[str]):
"appname,extra",
(
("fake.server.app", None),
("fake.server", None),
("fake.server:create_app", "--factory"),
("fake.server.create_app()", None),
("fake.server.create_app", None),
),
)
def test_server_run(
@@ -60,14 +62,17 @@ def test_server_run(
assert "Goin' Fast @ http://127.0.0.1:8000" in lines
def test_server_run_factory_with_args(caplog):
command = [
"fake.server.create_app_with_args",
"--factory",
]
@pytest.mark.parametrize(
"command",
(
["fake.server.create_app_with_args", "--factory"],
["fake.server.create_app_with_args"],
),
)
def test_server_run_factory_with_args(caplog, command):
lines = capture(command, caplog)
assert "module=fake.server.create_app_with_args" in lines
assert "target=fake.server.create_app_with_args" in lines
def test_server_run_factory_with_args_arbitrary(caplog):
@@ -81,25 +86,6 @@ def test_server_run_factory_with_args_arbitrary(caplog):
assert "foo=bar" in lines
def test_error_with_function_as_instance_without_factory_arg(caplog):
command = ["fake.server.create_app"]
lines = capture(command, caplog)
assert (
"Failed to run app: Module is not a Sanic app, it is a function\n "
"If this callable returns a Sanic instance try: \n"
"sanic fake.server.create_app --factory"
) in lines
def test_error_with_path_as_instance_without_simple_arg(caplog):
command = ["./fake/"]
lines = capture(command, caplog)
assert (
"Failed to run app: App not found.\n Please use --simple if you "
"are passing a directory to sanic.\n eg. sanic ./fake/ --simple"
) in lines
@pytest.mark.parametrize(
"cmd",
(

288
tests/test_cookies.py
View File

@@ -1,11 +1,16 @@
from datetime import datetime, timedelta
from http.cookies import SimpleCookie
from unittest.mock import Mock
import pytest
from sanic import Sanic
from sanic.cookies import Cookie
from sanic import Request, Sanic
from sanic.compat import Header
from sanic.cookies import Cookie, CookieJar
from sanic.cookies.request import CookieRequestParameters
from sanic.exceptions import ServerError
from sanic.response import text
from sanic.response.convenience import json
# ------------------------------------------------------------ #
@@ -111,21 +116,23 @@ def test_cookie_options(app):
def test_cookie_deletion(app):
cookie_jar = None
@app.route("/")
def handler(request):
nonlocal cookie_jar
response = text("OK")
del response.cookies["i_want_to_die"]
response.cookies["i_never_existed"] = "testing"
del response.cookies["i_never_existed"]
del response.cookies["one"]
response.cookies["two"] = "testing"
del response.cookies["two"]
cookie_jar = response.cookies
return response
request, response = app.test_client.get("/")
response_cookies = SimpleCookie()
response_cookies.load(response.headers.get("Set-Cookie", {}))
_, response = app.test_client.get("/")
assert int(response_cookies["i_want_to_die"]["max-age"]) == 0
with pytest.raises(KeyError):
response.cookies["i_never_existed"]
assert cookie_jar.get_cookie("one").max_age == 0
assert cookie_jar.get_cookie("two").max_age == 0
assert len(response.cookies) == 0
def test_cookie_reserved_cookie():
@@ -252,3 +259,262 @@ def test_cookie_expires_illegal_instance_type(expires):
with pytest.raises(expected_exception=TypeError) as e:
c["expires"] = expires
assert e.message == "Cookie 'expires' property must be a datetime"
@pytest.mark.parametrize("value", ("foo=one; foo=two", "foo=one;foo=two"))
def test_request_with_duplicate_cookie_key(value):
headers = Header({"Cookie": value})
request = Request(b"/", headers, "1.1", "GET", Mock(), Mock())
assert request.cookies["foo"] == "one"
assert request.cookies.get("foo") == "one"
assert request.cookies.getlist("foo") == ["one", "two"]
assert request.cookies.get("bar") is None
def test_cookie_jar_cookies():
headers = Header()
jar = CookieJar(headers)
jar.add_cookie("foo", "one")
jar.add_cookie("foo", "two", domain="example.com")
assert len(jar.cookies) == 2
assert len(headers) == 2
def test_cookie_jar_has_cookie():
headers = Header()
jar = CookieJar(headers)
jar.add_cookie("foo", "one")
jar.add_cookie("foo", "two", domain="example.com")
assert jar.has_cookie("foo")
assert jar.has_cookie("foo", domain="example.com")
assert not jar.has_cookie("foo", path="/unknown")
assert not jar.has_cookie("bar")
def test_cookie_jar_get_cookie():
headers = Header()
jar = CookieJar(headers)
cookie1 = jar.add_cookie("foo", "one")
cookie2 = jar.add_cookie("foo", "two", domain="example.com")
assert jar.get_cookie("foo") is cookie1
assert jar.get_cookie("foo", domain="example.com") is cookie2
assert jar.get_cookie("foo", path="/unknown") is None
assert jar.get_cookie("bar") is None
def test_cookie_jar_add_cookie_encode():
headers = Header()
jar = CookieJar(headers)
jar.add_cookie("foo", "one")
jar.add_cookie(
"foo",
"two",
domain="example.com",
path="/something",
secure=True,
max_age=999,
httponly=True,
samesite="strict",
)
jar.add_cookie("foo", "three", secure_prefix=True)
jar.add_cookie("foo", "four", host_prefix=True)
jar.add_cookie("foo", "five", host_prefix=True, partitioned=True)
encoded = [cookie.encode("ascii") for cookie in jar.cookies]
assert encoded == [
b"foo=one; Path=/; SameSite=Lax; Secure",
b"foo=two; Path=/something; Domain=example.com; Max-Age=999; SameSite=Strict; Secure; HttpOnly", # noqa
b"__Secure-foo=three; Path=/; SameSite=Lax; Secure",
b"__Host-foo=four; Path=/; SameSite=Lax; Secure",
b"__Host-foo=five; Path=/; SameSite=Lax; Secure; Partitioned",
]
def test_cookie_jar_old_school_cookie_encode():
headers = Header()
jar = CookieJar(headers)
jar["foo"] = "one"
jar["bar"] = "two"
jar["bar"]["domain"] = "example.com"
jar["bar"]["path"] = "/something"
jar["bar"]["secure"] = True
jar["bar"]["max-age"] = 999
jar["bar"]["httponly"] = True
jar["bar"]["samesite"] = "strict"
encoded = [cookie.encode("ascii") for cookie in jar.cookies]
assert encoded == [
b"foo=one; Path=/",
b"bar=two; Path=/something; Domain=example.com; Max-Age=999; SameSite=Strict; Secure; HttpOnly", # noqa
]
def test_cookie_jar_delete_cookie_encode():
headers = Header()
jar = CookieJar(headers)
jar.delete_cookie("foo")
jar.delete_cookie("foo", domain="example.com")
encoded = [cookie.encode("ascii") for cookie in jar.cookies]
assert encoded == [
b'foo=""; Path=/; Max-Age=0; Secure',
b'foo=""; Path=/; Domain=example.com; Max-Age=0; Secure',
]
def test_cookie_jar_old_school_delete_encode():
headers = Header()
jar = CookieJar(headers)
del jar["foo"]
encoded = [cookie.encode("ascii") for cookie in jar.cookies]
assert encoded == [
b'foo=""; Path=/; Max-Age=0; Secure',
]
def test_bad_cookie_prarms():
headers = Header()
jar = CookieJar(headers)
with pytest.raises(
ServerError,
match=(
"Both host_prefix and secure_prefix were requested. "
"A cookie should have only one prefix."
),
):
jar.add_cookie("foo", "bar", host_prefix=True, secure_prefix=True)
with pytest.raises(
ServerError,
match="Cannot set host_prefix on a cookie without secure=True",
):
jar.add_cookie("foo", "bar", host_prefix=True, secure=False)
with pytest.raises(
ServerError,
match="Cannot set host_prefix on a cookie unless path='/'",
):
jar.add_cookie(
"foo", "bar", host_prefix=True, secure=True, path="/foo"
)
with pytest.raises(
ServerError,
match="Cannot set host_prefix on a cookie with a defined domain",
):
jar.add_cookie(
"foo", "bar", host_prefix=True, secure=True, domain="foo.bar"
)
with pytest.raises(
ServerError,
match="Cannot set secure_prefix on a cookie without secure=True",
):
jar.add_cookie("foo", "bar", secure_prefix=True, secure=False)
with pytest.raises(
ServerError,
match=(
"Cannot create a partitioned cookie without "
"also setting host_prefix=True"
),
):
jar.add_cookie("foo", "bar", partitioned=True)
def test_cookie_accessors(app: Sanic):
@app.get("/")
async def handler(request: Request):
return json(
{
"getitem": {
"one": request.cookies["one"],
"two": request.cookies["two"],
"three": request.cookies["three"],
},
"get": {
"one": request.cookies.get("one", "fallback"),
"two": request.cookies.get("two", "fallback"),
"three": request.cookies.get("three", "fallback"),
"four": request.cookies.get("four", "fallback"),
},
"getlist": {
"one": request.cookies.getlist("one", ["fallback"]),
"two": request.cookies.getlist("two", ["fallback"]),
"three": request.cookies.getlist("three", ["fallback"]),
"four": request.cookies.getlist("four", ["fallback"]),
},
"getattr": {
"one": request.cookies.one,
"two": request.cookies.two,
"three": request.cookies.three,
"four": request.cookies.four,
},
}
)
_, response = app.test_client.get(
"/",
cookies={
"__Host-one": "1",
"__Secure-two": "2",
"three": "3",
},
)
assert response.json == {
"getitem": {
"one": "1",
"two": "2",
"three": "3",
},
"get": {
"one": "1",
"two": "2",
"three": "3",
"four": "fallback",
},
"getlist": {
"one": ["1"],
"two": ["2"],
"three": ["3"],
"four": ["fallback"],
},
"getattr": {
"one": "1",
"two": "2",
"three": "3",
"four": "",
},
}
def test_cookie_accessor_hyphens():
cookies = CookieRequestParameters({"session-token": ["abc123"]})
assert cookies.get("session-token") == cookies.session_token
def test_cookie_passthru(app):
cookie_jar = None
@app.route("/")
def handler(request):
nonlocal cookie_jar
response = text("OK")
response.add_cookie("one", "1", host_prefix=True)
response.delete_cookie("two", secure_prefix=True)
cookie_jar = response.cookies
return response
_, response = app.test_client.get("/")
assert cookie_jar.get_cookie("two", secure_prefix=True).max_age == 0
assert len(response.cookies) == 1
assert response.cookies["__Host-one"] == "1"

6
tests/test_errorpages.py
View File

@@ -248,9 +248,9 @@ def test_fallback_with_content_type_mismatch_accept(app):
app.router.reset()
@app.route("/alt1")
@app.route("/alt2", error_format="text")
@app.route("/alt3", error_format="html")
@app.route("/alt1", name="alt1")
@app.route("/alt2", error_format="text", name="alt2")
@app.route("/alt3", error_format="html", name="alt3")
def handler(_):
raise Exception("problem here")
# Yes, we know this return value is unreachable. This is on purpose.

24
tests/test_exceptions.py
View File

@@ -285,9 +285,15 @@ def test_contextual_exception_context(debug):
def fail():
raise TeapotError(context={"foo": "bar"})
app.post("/coffee/json", error_format="json")(lambda _: fail())
app.post("/coffee/html", error_format="html")(lambda _: fail())
app.post("/coffee/text", error_format="text")(lambda _: fail())
app.post("/coffee/json", error_format="json", name="json")(
lambda _: fail()
)
app.post("/coffee/html", error_format="html", name="html")(
lambda _: fail()
)
app.post("/coffee/text", error_format="text", name="text")(
lambda _: fail()
)
_, response = app.test_client.post("/coffee/json", debug=debug)
assert response.status == 418
@@ -323,9 +329,15 @@ def test_contextual_exception_extra(debug):
def fail():
raise TeapotError(extra={"foo": "bar"})
app.post("/coffee/json", error_format="json")(lambda _: fail())
app.post("/coffee/html", error_format="html")(lambda _: fail())
app.post("/coffee/text", error_format="text")(lambda _: fail())
app.post("/coffee/json", error_format="json", name="json")(
lambda _: fail()
)
app.post("/coffee/html", error_format="html", name="html")(
lambda _: fail()
)
app.post("/coffee/text", error_format="text", name="text")(
lambda _: fail()
)
_, response = app.test_client.post("/coffee/json", debug=debug)
assert response.status == 418

17
tests/test_exceptions_handler.py
View File

@@ -266,20 +266,17 @@ def test_exception_handler_response_was_sent(
assert "Error" in response.text
def test_warn_on_duplicate(
app: Sanic, caplog: LogCaptureFixture, recwarn: WarningsRecorder
):
def test_errir_on_duplicate(app: Sanic):
@app.exception(ServerError)
async def exception_handler_1(request, exception):
...
@app.exception(ServerError)
async def exception_handler_2(request, exception):
...
assert len(caplog.records) == 1
assert len(recwarn) == 1
assert caplog.records[0].message == (
message = (
"Duplicate exception handler definition on: route=__ALL_ROUTES__ and "
"exception=<class 'sanic.exceptions.ServerError'>"
)
with pytest.raises(ServerError, match=message):
@app.exception(ServerError)
async def exception_handler_2(request, exception):
...

14
tests/test_http.py
View File

@@ -98,3 +98,17 @@ def test_transfer_chunked(client):
data = stdjson.loads(body)
assert data == ["foo", "bar"]
def test_url_encoding(client):
client.send(
"""
GET /invalid\xA0url HTTP/1.1
"""
)
response = client.recv()
headers, body = response.rsplit(b"\r\n\r\n", 1)
assert b"400 Bad Request" in headers
assert b"URL may only contain US-ASCII characters." in body

90
tests/test_multiprocessing.py
View File

@@ -49,96 +49,6 @@ def test_multiprocessing(app):
assert len(process_list) == num_workers + 1
@pytest.mark.skipif(
not hasattr(signal, "SIGALRM"),
reason="SIGALRM is not implemented for this platform, we have to come "
"up with another timeout strategy to test these",
)
def test_multiprocessing_legacy(app):
"""Tests that the number of children we produce is correct"""
# Selects a number at random so we can spot check
num_workers = random.choice(range(2, multiprocessing.cpu_count() * 2 + 1))
process_list = set()
@app.after_server_start
async def shutdown(app):
await sleep(2.1)
app.stop()
def stop_on_alarm(*args):
for process in multiprocessing.active_children():
process_list.add(process.pid)
signal.signal(signal.SIGALRM, stop_on_alarm)
signal.alarm(2)
app.run(HOST, 4121, workers=num_workers, debug=True, legacy=True)
assert len(process_list) == num_workers
@pytest.mark.skipif(
not hasattr(signal, "SIGALRM"),
reason="SIGALRM is not implemented for this platform, we have to come "
"up with another timeout strategy to test these",
)
def test_multiprocessing_legacy_sock(app):
"""Tests that the number of children we produce is correct"""
# Selects a number at random so we can spot check
num_workers = random.choice(range(2, multiprocessing.cpu_count() * 2 + 1))
process_list = set()
@app.after_server_start
async def shutdown(app):
await sleep(2.1)
app.stop()
def stop_on_alarm(*args):
for process in multiprocessing.active_children():
process_list.add(process.pid)
signal.signal(signal.SIGALRM, stop_on_alarm)
signal.alarm(2)
sock = configure_socket(
{
"host": HOST,
"port": 4121,
"unix": None,
"backlog": 100,
}
)
app.run(workers=num_workers, debug=True, legacy=True, sock=sock)
sock.close()
assert len(process_list) == num_workers
@pytest.mark.skipif(
not hasattr(signal, "SIGALRM"),
reason="SIGALRM is not implemented for this platform, we have to come "
"up with another timeout strategy to test these",
)
def test_multiprocessing_legacy_unix(app):
"""Tests that the number of children we produce is correct"""
# Selects a number at random so we can spot check
num_workers = random.choice(range(2, multiprocessing.cpu_count() * 2 + 1))
process_list = set()
@app.after_server_start
async def shutdown(app):
await sleep(2.1)
app.stop()
def stop_on_alarm(*args):
for process in multiprocessing.active_children():
process_list.add(process.pid)
signal.signal(signal.SIGALRM, stop_on_alarm)
signal.alarm(2)
app.run(workers=num_workers, debug=True, legacy=True, unix="./test.sock")
assert len(process_list) == num_workers
@pytest.mark.skipif(
not hasattr(signal, "SIGALRM"),
reason="SIGALRM is not implemented for this platform",

4
tests/test_request.py
View File

@@ -1,3 +1,5 @@
import uuid
from unittest.mock import Mock
from uuid import UUID, uuid4
@@ -5,7 +7,7 @@ import pytest
from sanic import Sanic, response
from sanic.exceptions import BadURL, SanicException
from sanic.request import Request, uuid
from sanic.request import Request
from sanic.server import HttpProtocol

57
tests/test_requests.py
View File

@@ -16,8 +16,9 @@ from sanic_testing.testing import (
)
from sanic import Blueprint, Sanic
from sanic.constants import DEFAULT_HTTP_CONTENT_TYPE
from sanic.exceptions import ServerError
from sanic.request import DEFAULT_HTTP_CONTENT_TYPE, RequestParameters
from sanic.request import RequestParameters
from sanic.response import html, json, text
@@ -104,11 +105,11 @@ def test_html(app):
return html("<h1>Hello</h1>")
@app.route("/foo")
async def handler(request):
async def handler_foo(request):
return html(Foo())
@app.route("/bar")
async def handler(request):
async def handler_bar(request):
return html(Bar())
request, response = app.test_client.get("/")
@@ -1813,8 +1814,8 @@ def test_request_cookies(app):
request, response = app.test_client.get("/", cookies=cookies)
assert request.cookies == cookies
assert request.cookies == cookies # For request._cookies
assert len(request.cookies) == len(cookies)
assert request.cookies["test"] == cookies["test"]
@pytest.mark.asyncio
@@ -1827,8 +1828,8 @@ async def test_request_cookies_asgi(app):
request, response = await app.asgi_client.get("/", cookies=cookies)
assert request.cookies == cookies
assert request.cookies == cookies # For request._cookies
assert len(request.cookies) == len(cookies)
assert request.cookies["test"] == cookies["test"]
def test_request_cookies_without_cookies(app):
@@ -2198,10 +2199,25 @@ def test_safe_method_with_body(app):
assert response.body == b"OK"
def test_conflicting_body_methods_overload(app):
@pytest.mark.asyncio
async def test_conflicting_body_methods_overload_error(app: Sanic):
@app.put("/")
@app.put("/p/")
@app.put("/p/<foo>")
async def put(request, foo=None):
...
with pytest.raises(
ServerError,
match="Duplicate route names detected: test_conflicting_body_methods_overload_error\.put.*",
):
await app._startup()
def test_conflicting_body_methods_overload(app: Sanic):
@app.put("/", name="one")
@app.put("/p/", name="two")
@app.put("/p/<foo>", name="three")
async def put(request, foo=None):
return json(
{"name": request.route.name, "body": str(request.body), "foo": foo}
@@ -2219,21 +2235,21 @@ def test_conflicting_body_methods_overload(app):
_, response = app.test_client.put("/", json=payload)
assert response.status == 200
assert response.json == {
"name": "test_conflicting_body_methods_overload.put",
"name": "test_conflicting_body_methods_overload.one",
"foo": None,
"body": data,
}
_, response = app.test_client.put("/p", json=payload)
assert response.status == 200
assert response.json == {
"name": "test_conflicting_body_methods_overload.put",
"name": "test_conflicting_body_methods_overload.two",
"foo": None,
"body": data,
}
_, response = app.test_client.put("/p/test", json=payload)
assert response.status == 200
assert response.json == {
"name": "test_conflicting_body_methods_overload.put",
"name": "test_conflicting_body_methods_overload.three",
"foo": "test",
"body": data,
}
@@ -2246,9 +2262,26 @@ def test_conflicting_body_methods_overload(app):
}
def test_handler_overload(app):
@pytest.mark.asyncio
async def test_handler_overload_error(app: Sanic):
@app.get("/long/sub/route/param_a/<param_a:str>/param_b/<param_b:str>")
@app.post("/long/sub/route/")
def handler(request, **kwargs):
...
with pytest.raises(
ServerError,
match="Duplicate route names detected: test_handler_overload_error\.handler.*",
):
await app._startup()
def test_handler_overload(app: Sanic):
@app.get(
"/long/sub/route/param_a/<param_a:str>/param_b/<param_b:str>",
name="one",
)
@app.post("/long/sub/route/", name="two")
def handler(request, **kwargs):
return json(kwargs)

18
tests/test_routes.py
View File

@@ -12,7 +12,7 @@ from sanic_testing.testing import SanicTestClient
from sanic import Blueprint, Sanic
from sanic.constants import HTTP_METHODS
from sanic.exceptions import NotFound, SanicException
from sanic.exceptions import NotFound, SanicException, ServerError
from sanic.request import Request
from sanic.response import empty, json, text
@@ -744,8 +744,8 @@ def test_route_duplicate(app):
def test_double_stack_route(app):
@app.route("/test/1")
@app.route("/test/2")
@app.route("/test/1", name="test1")
@app.route("/test/2", name="test2")
async def handler1(request):
return text("OK")
@@ -759,8 +759,8 @@ def test_double_stack_route(app):
async def test_websocket_route_asgi(app):
ev = asyncio.Event()
@app.websocket("/test/1")
@app.websocket("/test/2")
@app.websocket("/test/1", name="test1")
@app.websocket("/test/2", name="test2")
async def handler(request, ws):
ev.set()
@@ -1279,7 +1279,7 @@ async def test_added_callable_route_ctx_kwargs(app):
@pytest.mark.asyncio
async def test_duplicate_route_deprecation(app):
async def test_duplicate_route_error(app):
@app.route("/foo", name="duped")
async def handler_foo(request):
return text("...")
@@ -1289,9 +1289,7 @@ async def test_duplicate_route_deprecation(app):
return text("...")
message = (
r"\[DEPRECATION v23\.3\] Duplicate route names detected: "
r"test_duplicate_route_deprecation\.duped\. In the future, "
r"Sanic will enforce uniqueness in route naming\."
"Duplicate route names detected: test_duplicate_route_error.duped."
)
with pytest.warns(DeprecationWarning, match=message):
with pytest.raises(ServerError, match=message):
await app._startup()

4
tests/test_signal_handlers.py
View File

@@ -66,8 +66,8 @@ def test_no_register_system_signals_fails(app):
app.listener("after_server_stop")(after)
message = (
"Cannot run Sanic.serve with register_sys_signals=False. Use "
"either Sanic.serve_single or Sanic.serve_legacy."
r"Cannot run Sanic\.serve with register_sys_signals=False\. Use "
r"Sanic.serve_single\."
)
with pytest.raises(RuntimeError, match=message):
app.prepare(HOST, PORT, register_sys_signals=False)

43
tests/test_static.py
View File

@@ -9,7 +9,7 @@ from time import gmtime, strftime
import pytest
from sanic import Sanic, text
from sanic.exceptions import FileNotFound
from sanic.exceptions import FileNotFound, ServerError
@pytest.fixture(scope="module")
@@ -108,14 +108,9 @@ def test_static_file_pathlib(app, static_file_directory, file_name):
def test_static_file_bytes(app, static_file_directory, file_name):
bsep = os.path.sep.encode("utf-8")
file_path = static_file_directory.encode("utf-8") + bsep + file_name
message = (
"Serving a static directory with a bytes "
"string is deprecated and will be removed in v22.9."
)
with pytest.warns(DeprecationWarning, match=message):
message = "Static file or directory must be a path-like object or string"
with pytest.raises(TypeError, match=message):
app.static("/testing.file", file_path)
request, response = app.test_client.get("/testing.file")
assert response.status == 200
@pytest.mark.parametrize(
@@ -523,10 +518,26 @@ def test_no_stack_trace_on_not_found(app, static_file_directory, caplog):
assert response.text == "No file: /static/non_existing_file.file"
def test_multiple_statics(app, static_file_directory):
@pytest.mark.asyncio
async def test_multiple_statics_error(app, static_file_directory):
app.static("/file", get_file_path(static_file_directory, "test.file"))
app.static("/png", get_file_path(static_file_directory, "python.png"))
message = (
r"Duplicate route names detected: test_multiple_statics_error\.static"
)
with pytest.raises(ServerError, match=message):
await app._startup()
def test_multiple_statics(app, static_file_directory):
app.static(
"/file", get_file_path(static_file_directory, "test.file"), name="file"
)
app.static(
"/png", get_file_path(static_file_directory, "python.png"), name="png"
)
_, response = app.test_client.get("/file")
assert response.status == 200
assert response.body == get_file_content(
@@ -540,10 +551,22 @@ def test_multiple_statics(app, static_file_directory):
)
def test_resource_type_default(app, static_file_directory):
@pytest.mark.asyncio
async def test_resource_type_default_error(app, static_file_directory):
app.static("/static", static_file_directory)
app.static("/file", get_file_path(static_file_directory, "test.file"))
message = r"Duplicate route names detected: test_resource_type_default_error\.static"
with pytest.raises(ServerError, match=message):
await app._startup()
def test_resource_type_default(app, static_file_directory):
app.static("/static", static_file_directory, name="static")
app.static(
"/file", get_file_path(static_file_directory, "test.file"), name="file"
)
_, response = app.test_client.get("/static")
assert response.status == 404

29
tests/worker/test_loader.py
View File

@@ -52,34 +52,23 @@ def test_cwd_in_path():
def test_input_is_dir():
loader = AppLoader(str(STATIC))
message = (
"App not found.\n Please use --simple if you are passing a "
f"directory to sanic.\n eg. sanic {str(STATIC)} --simple"
)
with pytest.raises(ValueError, match=message):
loader.load()
app = loader.load()
assert isinstance(app, Sanic)
def test_input_is_factory():
ns = SimpleNamespace(module="foo")
ns = SimpleNamespace(target="foo")
loader = AppLoader("tests.fake.server:create_app", args=ns)
message = (
"Module is not a Sanic app, it is a function\n If this callable "
"returns a Sanic instance try: \nsanic foo --factory"
)
with pytest.raises(ValueError, match=message):
loader.load()
app = loader.load()
assert isinstance(app, Sanic)
def test_input_is_module():
ns = SimpleNamespace(module="foo")
ns = SimpleNamespace(target="foo")
loader = AppLoader("tests.fake.server", args=ns)
message = (
"Module is not a Sanic app, it is a module\n "
"Perhaps you meant foo:app?"
)
with pytest.raises(ValueError, match=message):
loader.load()
app = loader.load()
assert isinstance(app, Sanic)
@pytest.mark.parametrize("creator", ("mkcert", "trustme"))

18
tests/worker/test_multiplexer.py
View File

@@ -72,24 +72,6 @@ def test_not_have_multiplexer_single(app: Sanic):
assert not event.is_set()
def test_not_have_multiplexer_legacy(app: Sanic):
event = Event()
@app.main_process_start
async def setup(app, _):
app.shared_ctx.event = event
@app.after_server_start
def stop(app):
if hasattr(app, "m") and isinstance(app.m, WorkerMultiplexer):
app.shared_ctx.event.set()
app.stop()
app.run(legacy=True)
assert not event.is_set()
def test_ack(worker_state: Dict[str, Any], m: WorkerMultiplexer):
worker_state["Test"] = {"foo": "bar"}
m.ack()