sanic/tests/test_server_events.py

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(f"DEBUG MESSAGE FOR PYTEST for {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