532 lines
14 KiB
Python
532 lines
14 KiB
Python
import asyncio
|
|
import logging
|
|
import re
|
|
|
|
from inspect import isawaitable
|
|
from os import environ
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
try:
|
|
import uvloop # noqa
|
|
except ImportError:
|
|
pass
|
|
|
|
from sanic import Sanic
|
|
from sanic.compat import OS_IS_WINDOWS
|
|
from sanic.config import Config
|
|
from sanic.exceptions import SanicException
|
|
from sanic.response import text
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_app_registry():
|
|
Sanic._app_registry = {}
|
|
|
|
|
|
def uvloop_installed():
|
|
try:
|
|
import uvloop # noqa
|
|
|
|
return True
|
|
except ImportError:
|
|
return False
|
|
|
|
|
|
def test_app_loop_running(app):
|
|
@app.get("/test")
|
|
async def handler(request):
|
|
assert isinstance(app.loop, asyncio.AbstractEventLoop)
|
|
return text("pass")
|
|
|
|
request, response = app.test_client.get("/test")
|
|
assert response.text == "pass"
|
|
|
|
|
|
def test_create_asyncio_server(app):
|
|
if not uvloop_installed():
|
|
loop = asyncio.get_event_loop()
|
|
asyncio_srv_coro = app.create_server(return_asyncio_server=True)
|
|
assert isawaitable(asyncio_srv_coro)
|
|
srv = loop.run_until_complete(asyncio_srv_coro)
|
|
assert srv.is_serving() is True
|
|
|
|
|
|
def test_asyncio_server_no_start_serving(app):
|
|
if not uvloop_installed():
|
|
loop = asyncio.get_event_loop()
|
|
asyncio_srv_coro = app.create_server(
|
|
port=43123,
|
|
return_asyncio_server=True,
|
|
asyncio_server_kwargs=dict(start_serving=False),
|
|
)
|
|
srv = loop.run_until_complete(asyncio_srv_coro)
|
|
assert srv.is_serving() is False
|
|
|
|
|
|
def test_asyncio_server_start_serving(app):
|
|
if not uvloop_installed():
|
|
loop = asyncio.get_event_loop()
|
|
asyncio_srv_coro = app.create_server(
|
|
port=43124,
|
|
return_asyncio_server=True,
|
|
asyncio_server_kwargs=dict(start_serving=False),
|
|
)
|
|
srv = loop.run_until_complete(asyncio_srv_coro)
|
|
assert srv.is_serving() is False
|
|
loop.run_until_complete(srv.start_serving())
|
|
assert srv.is_serving() is True
|
|
wait_close = srv.close()
|
|
loop.run_until_complete(wait_close)
|
|
# Looks like we can't easily test `serve_forever()`
|
|
|
|
|
|
def test_create_server_main(app, caplog):
|
|
app.listener("main_process_start")(lambda *_: ...)
|
|
loop = asyncio.get_event_loop()
|
|
with caplog.at_level(logging.INFO):
|
|
asyncio_srv_coro = app.create_server(return_asyncio_server=True)
|
|
loop.run_until_complete(asyncio_srv_coro)
|
|
assert (
|
|
"sanic.root",
|
|
30,
|
|
"Listener events for the main process are not available with "
|
|
"create_server()",
|
|
) in caplog.record_tuples
|
|
|
|
|
|
def test_create_server_main_convenience(app, caplog):
|
|
app.main_process_start(lambda *_: ...)
|
|
loop = asyncio.get_event_loop()
|
|
with caplog.at_level(logging.INFO):
|
|
asyncio_srv_coro = app.create_server(return_asyncio_server=True)
|
|
loop.run_until_complete(asyncio_srv_coro)
|
|
assert (
|
|
"sanic.root",
|
|
30,
|
|
"Listener events for the main process are not available with "
|
|
"create_server()",
|
|
) in caplog.record_tuples
|
|
|
|
|
|
def test_app_loop_not_running(app):
|
|
with pytest.raises(SanicException) as excinfo:
|
|
app.loop
|
|
|
|
assert str(excinfo.value) == (
|
|
"Loop can only be retrieved after the app has started "
|
|
"running. Not supported with `create_server` function"
|
|
)
|
|
|
|
|
|
def test_app_run_raise_type_error(app):
|
|
|
|
with pytest.raises(TypeError) as excinfo:
|
|
app.run(loop="loop")
|
|
|
|
assert str(excinfo.value) == (
|
|
"loop is not a valid argument. To use an existing loop, "
|
|
"change to create_server().\nSee more: "
|
|
"https://sanic.readthedocs.io/en/latest/sanic/deploying.html"
|
|
"#asynchronous-support"
|
|
)
|
|
|
|
|
|
def test_app_route_raise_value_error(app):
|
|
|
|
with pytest.raises(ValueError) as excinfo:
|
|
|
|
@app.route("/test")
|
|
async def handler():
|
|
return text("test")
|
|
|
|
assert (
|
|
str(excinfo.value)
|
|
== "Required parameter `request` missing in the handler() route?"
|
|
)
|
|
|
|
|
|
def test_app_handle_request_handler_is_none(app, monkeypatch):
|
|
def mockreturn(*args, **kwargs):
|
|
return Mock(), None, {}
|
|
|
|
# Not sure how to make app.router.get() return None, so use mock here.
|
|
monkeypatch.setattr(app.router, "get", mockreturn)
|
|
|
|
@app.get("/test")
|
|
def handler(request):
|
|
return text("test")
|
|
|
|
_, response = app.test_client.get("/test")
|
|
|
|
assert (
|
|
"'None' was returned while requesting a handler from the router"
|
|
in response.text
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("websocket_enabled", [True, False])
|
|
@pytest.mark.parametrize("enable", [True, False])
|
|
def test_app_enable_websocket(app, websocket_enabled, enable):
|
|
app.websocket_enabled = websocket_enabled
|
|
app.enable_websocket(enable=enable)
|
|
|
|
assert app.websocket_enabled == enable
|
|
|
|
@app.websocket("/ws")
|
|
async def handler(request, ws):
|
|
await ws.send("test")
|
|
|
|
assert app.websocket_enabled == True
|
|
|
|
|
|
@patch("sanic.app.WebSocketProtocol")
|
|
def test_app_websocket_parameters(websocket_protocol_mock, app):
|
|
app.config.WEBSOCKET_MAX_SIZE = 44
|
|
app.config.WEBSOCKET_PING_TIMEOUT = 48
|
|
app.config.WEBSOCKET_PING_INTERVAL = 50
|
|
|
|
@app.websocket("/ws")
|
|
async def handler(request, ws):
|
|
await ws.send("test")
|
|
|
|
try:
|
|
# This will fail because WebSocketProtocol is mocked and only the call kwargs matter
|
|
app.test_client.get("/ws")
|
|
except:
|
|
pass
|
|
|
|
websocket_protocol_call_args = websocket_protocol_mock.call_args
|
|
ws_kwargs = websocket_protocol_call_args[1]
|
|
assert ws_kwargs["websocket_max_size"] == app.config.WEBSOCKET_MAX_SIZE
|
|
assert (
|
|
ws_kwargs["websocket_ping_timeout"]
|
|
== app.config.WEBSOCKET_PING_TIMEOUT
|
|
)
|
|
assert (
|
|
ws_kwargs["websocket_ping_interval"]
|
|
== app.config.WEBSOCKET_PING_INTERVAL
|
|
)
|
|
|
|
|
|
def test_handle_request_with_nested_exception(app, monkeypatch):
|
|
|
|
err_msg = "Mock Exception"
|
|
|
|
# Not sure how to raise an exception in app.error_handler.response(), use mock here
|
|
def mock_error_handler_response(*args, **kwargs):
|
|
raise Exception(err_msg)
|
|
|
|
monkeypatch.setattr(
|
|
app.error_handler, "response", mock_error_handler_response
|
|
)
|
|
|
|
@app.get("/")
|
|
def handler(request):
|
|
raise Exception
|
|
|
|
request, response = app.test_client.get("/")
|
|
assert response.status == 500
|
|
assert response.text == "An error occurred while handling an error"
|
|
|
|
|
|
def test_handle_request_with_nested_exception_debug(app, monkeypatch):
|
|
|
|
err_msg = "Mock Exception"
|
|
|
|
# Not sure how to raise an exception in app.error_handler.response(), use mock here
|
|
def mock_error_handler_response(*args, **kwargs):
|
|
raise Exception(err_msg)
|
|
|
|
monkeypatch.setattr(
|
|
app.error_handler, "response", mock_error_handler_response
|
|
)
|
|
|
|
@app.get("/")
|
|
def handler(request):
|
|
raise Exception
|
|
|
|
request, response = app.test_client.get("/", debug=True)
|
|
assert response.status == 500
|
|
assert response.text.startswith(
|
|
f"Error while handling error: {err_msg}\nStack: Traceback (most recent call last):\n"
|
|
)
|
|
|
|
|
|
def test_handle_request_with_nested_sanic_exception(app, monkeypatch, caplog):
|
|
|
|
# Not sure how to raise an exception in app.error_handler.response(), use mock here
|
|
def mock_error_handler_response(*args, **kwargs):
|
|
raise SanicException("Mock SanicException")
|
|
|
|
monkeypatch.setattr(
|
|
app.error_handler, "response", mock_error_handler_response
|
|
)
|
|
|
|
@app.get("/")
|
|
def handler(request):
|
|
raise Exception
|
|
|
|
with caplog.at_level(logging.ERROR):
|
|
request, response = app.test_client.get("/")
|
|
port = request.server_port
|
|
assert port > 0
|
|
assert response.status == 500
|
|
assert "Mock SanicException" in response.text
|
|
assert (
|
|
"sanic.error",
|
|
logging.ERROR,
|
|
f"Exception occurred while handling uri: 'http://127.0.0.1:{port}/'",
|
|
) in caplog.record_tuples
|
|
|
|
|
|
def test_app_name_required():
|
|
with pytest.raises(SanicException):
|
|
Sanic()
|
|
|
|
|
|
def test_app_has_test_mode_sync():
|
|
app = Sanic("test")
|
|
|
|
@app.get("/")
|
|
def handler(request):
|
|
assert request.app.test_mode
|
|
return text("test")
|
|
|
|
_, response = app.test_client.get("/")
|
|
assert response.status == 200
|
|
|
|
|
|
def test_app_registry():
|
|
instance = Sanic("test")
|
|
assert Sanic._app_registry["test"] is instance
|
|
|
|
|
|
def test_app_registry_wrong_type():
|
|
with pytest.raises(
|
|
SanicException, match="Registered app must be an instance of Sanic"
|
|
):
|
|
Sanic.register_app(1)
|
|
|
|
|
|
def test_app_registry_name_reuse():
|
|
Sanic("test")
|
|
Sanic.test_mode = False
|
|
with pytest.raises(
|
|
SanicException, match='Sanic app name "test" already in use.'
|
|
):
|
|
Sanic("test")
|
|
Sanic.test_mode = True
|
|
Sanic("test")
|
|
|
|
|
|
def test_app_registry_retrieval():
|
|
instance = Sanic("test")
|
|
assert Sanic.get_app("test") is instance
|
|
|
|
|
|
def test_app_registry_retrieval_from_multiple():
|
|
instance = Sanic("test")
|
|
Sanic("something_else")
|
|
assert Sanic.get_app("test") is instance
|
|
|
|
|
|
def test_get_app_does_not_exist():
|
|
with pytest.raises(
|
|
SanicException, match='Sanic app name "does-not-exist" not found.'
|
|
):
|
|
Sanic.get_app("does-not-exist")
|
|
|
|
|
|
def test_get_app_does_not_exist_force_create():
|
|
assert isinstance(
|
|
Sanic.get_app("does-not-exist", force_create=True), Sanic
|
|
)
|
|
|
|
|
|
def test_get_app_default():
|
|
instance = Sanic("test")
|
|
assert Sanic.get_app() is instance
|
|
|
|
|
|
def test_get_app_no_default():
|
|
with pytest.raises(
|
|
SanicException, match="No Sanic apps have been registered."
|
|
):
|
|
Sanic.get_app()
|
|
|
|
|
|
def test_get_app_default_ambiguous():
|
|
Sanic("test1")
|
|
Sanic("test2")
|
|
with pytest.raises(
|
|
SanicException,
|
|
match=re.escape(
|
|
'Multiple Sanic apps found, use Sanic.get_app("app_name")'
|
|
),
|
|
):
|
|
Sanic.get_app()
|
|
|
|
|
|
def test_app_no_registry():
|
|
Sanic("no-register", register=False)
|
|
with pytest.raises(
|
|
SanicException, match='Sanic app name "no-register" not found.'
|
|
):
|
|
Sanic.get_app("no-register")
|
|
|
|
|
|
def test_app_no_registry_env():
|
|
environ["SANIC_REGISTER"] = "False"
|
|
Sanic("no-register")
|
|
with pytest.raises(
|
|
SanicException, match='Sanic app name "no-register" not found.'
|
|
):
|
|
Sanic.get_app("no-register")
|
|
del environ["SANIC_REGISTER"]
|
|
|
|
|
|
def test_app_set_attribute_warning(app):
|
|
with pytest.warns(DeprecationWarning) as record:
|
|
app.foo = 1
|
|
|
|
assert len(record) == 1
|
|
assert record[0].message.args[0] == (
|
|
"Setting variables on Sanic instances is deprecated "
|
|
"and will be removed in version 21.12. You should change your "
|
|
"Sanic instance to use instance.ctx.foo instead."
|
|
)
|
|
|
|
|
|
def test_app_set_context(app):
|
|
app.ctx.foo = 1
|
|
|
|
retrieved = Sanic.get_app(app.name)
|
|
assert retrieved.ctx.foo == 1
|
|
|
|
|
|
def test_subclass_initialisation():
|
|
class CustomSanic(Sanic):
|
|
pass
|
|
|
|
CustomSanic("test_subclass_initialisation")
|
|
|
|
|
|
def test_bad_custom_config():
|
|
with pytest.raises(
|
|
SanicException,
|
|
match=(
|
|
"When instantiating Sanic with config, you cannot also pass "
|
|
"load_env or env_prefix"
|
|
),
|
|
):
|
|
Sanic("test", config=1, load_env=1)
|
|
with pytest.raises(
|
|
SanicException,
|
|
match=(
|
|
"When instantiating Sanic with config, you cannot also pass "
|
|
"load_env or env_prefix"
|
|
),
|
|
):
|
|
Sanic("test", config=1, env_prefix=1)
|
|
|
|
|
|
def test_custom_config():
|
|
class CustomConfig(Config):
|
|
...
|
|
|
|
config = CustomConfig()
|
|
app = Sanic("custom", config=config)
|
|
|
|
assert app.config == config
|
|
|
|
|
|
def test_custom_context():
|
|
class CustomContext:
|
|
...
|
|
|
|
ctx = CustomContext()
|
|
app = Sanic("custom", ctx=ctx)
|
|
|
|
assert app.ctx == ctx
|
|
|
|
|
|
def test_app_uvloop_config(caplog):
|
|
if uvloop_installed():
|
|
asyncio.set_event_loop_policy(None)
|
|
environ["SANIC_USE_UVLOOP"] = "false"
|
|
|
|
app = Sanic("dont_use_uvloop")
|
|
|
|
@app.get("/")
|
|
def _(request):
|
|
assert not isinstance(
|
|
asyncio.get_event_loop_policy(),
|
|
uvloop.EventLoopPolicy
|
|
)
|
|
return text("ok")
|
|
|
|
app.test_client.get("/")
|
|
del environ["SANIC_USE_UVLOOP"]
|
|
|
|
app = Sanic("use_uvloop")
|
|
|
|
@app.get("/")
|
|
async def _(request):
|
|
assert isinstance(
|
|
asyncio.get_event_loop_policy(),
|
|
uvloop.EventLoopPolicy
|
|
)
|
|
return text("ok")
|
|
|
|
app.test_client.get("/")
|
|
|
|
asyncio.set_event_loop_policy(None)
|
|
|
|
environ["SANIC_NO_UVLOOP"] = "true"
|
|
app = Sanic("use_uvloop_with_no_uvloop_set")
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
@app.get("/")
|
|
async def _(request):
|
|
assert isinstance(
|
|
asyncio.get_event_loop_policy(),
|
|
uvloop.EventLoopPolicy
|
|
)
|
|
return text("ok")
|
|
|
|
app.test_client.get("/")
|
|
|
|
del environ["SANIC_NO_UVLOOP"]
|
|
asyncio.set_event_loop_policy(None)
|
|
|
|
assert (
|
|
"sanic.error",
|
|
logging.WARNING,
|
|
"You are running the app using uvloop, but the "
|
|
"'SANIC_NO_UVLOOP' environment variable (used to opt-out "
|
|
"of installing uvloop with Sanic) is set to true. If you "
|
|
"want to disable uvloop with Sanic, set the 'USE_UVLOOP' "
|
|
"configuration value to false."
|
|
) in caplog.record_tuples
|
|
|
|
elif not OS_IS_WINDOWS:
|
|
app = Sanic("wants_but_cant_use_uvloop")
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
@app.get("/")
|
|
async def _(request):
|
|
return text("ok")
|
|
app.test_client.get("/")
|
|
|
|
assert (
|
|
"sanic.error",
|
|
logging.WARNING,
|
|
"You are trying to use uvloop, but uvloop is not "
|
|
"installed in your system. In order to use uvloop "
|
|
"you must first install it. Otherwise, you can disable "
|
|
"uvloop completely by setting the 'USE_UVLOOP' "
|
|
"configuration value to false. Sanic will now continue "
|
|
"to run without using uvloop."
|
|
) in caplog.record_tuples
|