e506c89304
* ➖ deprecate None value support for app name * 🚨 cleanup linter issues across the codebase
143 lines
3.8 KiB
Python
143 lines
3.8 KiB
Python
import asyncio
|
|
import signal
|
|
|
|
import pytest
|
|
|
|
from sanic.testing import HOST, PORT
|
|
|
|
|
|
AVAILABLE_LISTENERS = [
|
|
"before_server_start",
|
|
"after_server_start",
|
|
"before_server_stop",
|
|
"after_server_stop",
|
|
]
|
|
|
|
skipif_no_alarm = 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 create_listener(listener_name, in_list):
|
|
async def _listener(app, loop):
|
|
print("DEBUG MESSAGE FOR PYTEST for {}".format(listener_name))
|
|
in_list.insert(0, app.name + listener_name)
|
|
|
|
return _listener
|
|
|
|
|
|
def start_stop_app(random_name_app, **run_kwargs):
|
|
def stop_on_alarm(signum, frame):
|
|
raise KeyboardInterrupt("SIGINT for sanic to stop gracefully")
|
|
|
|
signal.signal(signal.SIGALRM, stop_on_alarm)
|
|
signal.alarm(1)
|
|
try:
|
|
random_name_app.run(HOST, PORT, **run_kwargs)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
|
|
@skipif_no_alarm
|
|
@pytest.mark.parametrize("listener_name", AVAILABLE_LISTENERS)
|
|
def test_single_listener(app, listener_name):
|
|
"""Test that listeners on their own work"""
|
|
output = []
|
|
# Register listener
|
|
app.listener(listener_name)(create_listener(listener_name, output))
|
|
start_stop_app(app)
|
|
assert app.name + listener_name == output.pop()
|
|
|
|
|
|
@skipif_no_alarm
|
|
@pytest.mark.parametrize("listener_name", AVAILABLE_LISTENERS)
|
|
def test_register_listener(app, listener_name):
|
|
"""
|
|
Test that listeners on their own work with
|
|
app.register_listener method
|
|
"""
|
|
output = []
|
|
# Register listener
|
|
listener = create_listener(listener_name, output)
|
|
app.register_listener(listener, event=listener_name)
|
|
start_stop_app(app)
|
|
assert app.name + listener_name == output.pop()
|
|
|
|
|
|
@skipif_no_alarm
|
|
def test_all_listeners(app):
|
|
output = []
|
|
for listener_name in AVAILABLE_LISTENERS:
|
|
listener = create_listener(listener_name, output)
|
|
app.listener(listener_name)(listener)
|
|
start_stop_app(app)
|
|
for listener_name in AVAILABLE_LISTENERS:
|
|
assert app.name + listener_name == output.pop()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_before_events_create_server(app):
|
|
class MySanicDb:
|
|
pass
|
|
|
|
@app.listener("before_server_start")
|
|
async def init_db(app, loop):
|
|
app.db = MySanicDb()
|
|
|
|
await app.create_server(debug=True, return_asyncio_server=True, port=PORT)
|
|
|
|
assert hasattr(app, "db")
|
|
assert isinstance(app.db, MySanicDb)
|
|
|
|
|
|
def test_create_server_trigger_events(app):
|
|
"""Test if create_server can trigger server events"""
|
|
|
|
flag1 = False
|
|
flag2 = False
|
|
flag3 = False
|
|
|
|
async def stop(app, loop):
|
|
nonlocal flag1
|
|
flag1 = True
|
|
await asyncio.sleep(0.1)
|
|
app.stop()
|
|
|
|
async def before_stop(app, loop):
|
|
nonlocal flag2
|
|
flag2 = True
|
|
|
|
async def after_stop(app, loop):
|
|
nonlocal flag3
|
|
flag3 = True
|
|
|
|
app.listener("after_server_start")(stop)
|
|
app.listener("before_server_stop")(before_stop)
|
|
app.listener("after_server_stop")(after_stop)
|
|
|
|
loop = asyncio.get_event_loop()
|
|
serv_coro = app.create_server(return_asyncio_server=True)
|
|
serv_task = asyncio.ensure_future(serv_coro, loop=loop)
|
|
server = loop.run_until_complete(serv_task)
|
|
server.after_start()
|
|
try:
|
|
loop.run_forever()
|
|
except KeyboardInterrupt as e:
|
|
loop.stop()
|
|
finally:
|
|
# Run the on_stop function if provided
|
|
server.before_stop()
|
|
|
|
# Wait for server to close
|
|
close_task = server.close()
|
|
loop.run_until_complete(close_task)
|
|
|
|
# Complete all tasks on the loop
|
|
signal.stopped = True
|
|
for connection in server.connections:
|
|
connection.close_if_idle()
|
|
server.after_stop()
|
|
assert flag1 and flag2 and flag3
|