6ffc4d9756
* First attempt at new Websockets implementation based on websockets >= 9.0, with sans-i/o features. Requires more work. * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * wip, update websockets code to new Sans/IO API * Refactored new websockets impl into own modules Incorporated other suggestions made by team * Another round of work on the new websockets impl * Added websocket_timeout support (matching previous/legacy support) * Lots more comments * Incorporated suggested changes from previous round of review * Changed RuntimeError usage to ServerError * Changed SanicException usage to ServerError * Removed some redundant asserts * Change remaining asserts to ServerErrors * Fixed some timeout handling issues * Fixed websocket.close() handling, and made it more robust * Made auto_close task smarter and more error-resilient * Made fail_connection routine smarter and more error-resilient * Further new websockets impl fixes * Update compatibility with Websockets v10 * Track server connection state in a more precise way * Try to handle the shutdown process more gracefully * Add a new end_connection() helper, to use as an alterative to close() or fail_connection() * Kill the auto-close task and keepalive-timeout task when sanic is shutdown * Deprecate WEBSOCKET_READ_LIMIT and WEBSOCKET_WRITE_LIMIT configs, they are not used in this implementation. * Change a warning message to debug level Remove default values for deprecated websocket parameters * Fix flake8 errors * Fix a couple of missed failing tests * remove websocket bench from examples * Integrate suggestions from code reviews Use Optional[T] instead of union[T,None] Fix mypy type logic errors change "is not None" to truthy checks where appropriate change "is None" to falsy checks were appropriate Add more debug logging when debug mode is on Change to using sanic.logger for debug logging rather than error_logger. * Fix long line lengths of debug messages Add some new debug messages when websocket IO is paused and unpaused for flow control Fix websocket example to use app.static() * remove unused import in websocket example app * re-run isort after Flake8 fixes Co-authored-by: Adam Hopkins <adam@amhopkins.com> Co-authored-by: Adam Hopkins <admhpkns@gmail.com>
195 lines
5.2 KiB
Python
195 lines
5.2 KiB
Python
import asyncio
|
|
import json
|
|
import shlex
|
|
import subprocess
|
|
import time
|
|
import urllib.request
|
|
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from sanic_testing.testing import ASGI_PORT as PORT
|
|
|
|
from sanic.app import Sanic
|
|
from sanic.worker import GunicornWorker
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def gunicorn_worker():
|
|
command = (
|
|
"gunicorn "
|
|
f"--bind 127.0.0.1:{PORT} "
|
|
"--worker-class sanic.worker.GunicornWorker "
|
|
"examples.simple_server:app"
|
|
)
|
|
worker = subprocess.Popen(shlex.split(command))
|
|
time.sleep(3)
|
|
yield
|
|
worker.kill()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def gunicorn_worker_with_access_logs():
|
|
command = (
|
|
"gunicorn "
|
|
f"--bind 127.0.0.1:{PORT + 1} "
|
|
"--worker-class sanic.worker.GunicornWorker "
|
|
"examples.simple_server:app"
|
|
)
|
|
worker = subprocess.Popen(shlex.split(command), stdout=subprocess.PIPE)
|
|
time.sleep(2)
|
|
return worker
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def gunicorn_worker_with_env_var():
|
|
command = (
|
|
'env SANIC_ACCESS_LOG="False" '
|
|
"gunicorn "
|
|
f"--bind 127.0.0.1:{PORT + 2} "
|
|
"--worker-class sanic.worker.GunicornWorker "
|
|
"--log-level info "
|
|
"examples.simple_server:app"
|
|
)
|
|
worker = subprocess.Popen(shlex.split(command), stdout=subprocess.PIPE)
|
|
time.sleep(2)
|
|
return worker
|
|
|
|
|
|
def test_gunicorn_worker(gunicorn_worker):
|
|
with urllib.request.urlopen(f"http://localhost:{PORT}/") as f:
|
|
res = json.loads(f.read(100).decode())
|
|
assert res["test"]
|
|
|
|
|
|
def test_gunicorn_worker_no_logs(gunicorn_worker_with_env_var):
|
|
"""
|
|
if SANIC_ACCESS_LOG was set to False do not show access logs
|
|
"""
|
|
with urllib.request.urlopen(f"http://localhost:{PORT + 2}/") as _:
|
|
gunicorn_worker_with_env_var.kill()
|
|
assert not gunicorn_worker_with_env_var.stdout.read()
|
|
|
|
|
|
def test_gunicorn_worker_with_logs(gunicorn_worker_with_access_logs):
|
|
"""
|
|
default - show access logs
|
|
"""
|
|
with urllib.request.urlopen(f"http://localhost:{PORT + 1}/") as _:
|
|
gunicorn_worker_with_access_logs.kill()
|
|
assert (
|
|
b"(sanic.access)[INFO][127.0.0.1"
|
|
in gunicorn_worker_with_access_logs.stdout.read()
|
|
)
|
|
|
|
|
|
class GunicornTestWorker(GunicornWorker):
|
|
def __init__(self):
|
|
self.app = mock.Mock()
|
|
self.app.callable = Sanic("test_gunicorn_worker")
|
|
self.servers = {}
|
|
self.exit_code = 0
|
|
self.cfg = mock.Mock()
|
|
self.notify = mock.Mock()
|
|
|
|
|
|
@pytest.fixture
|
|
def worker():
|
|
return GunicornTestWorker()
|
|
|
|
|
|
def test_worker_init_process(worker):
|
|
with mock.patch("sanic.worker.asyncio") as mock_asyncio:
|
|
try:
|
|
worker.init_process()
|
|
except TypeError:
|
|
pass
|
|
|
|
assert mock_asyncio.get_event_loop.return_value.close.called
|
|
assert mock_asyncio.new_event_loop.called
|
|
assert mock_asyncio.set_event_loop.called
|
|
|
|
|
|
def test_worker_init_signals(worker):
|
|
worker.loop = mock.Mock()
|
|
worker.init_signals()
|
|
assert worker.loop.add_signal_handler.called
|
|
|
|
|
|
def test_handle_abort(worker):
|
|
with mock.patch("sanic.worker.sys") as mock_sys:
|
|
worker.handle_abort(object(), object())
|
|
assert not worker.alive
|
|
assert worker.exit_code == 1
|
|
mock_sys.exit.assert_called_with(1)
|
|
|
|
|
|
def test_handle_quit(worker):
|
|
worker.handle_quit(object(), object())
|
|
assert not worker.alive
|
|
assert worker.exit_code == 0
|
|
|
|
|
|
async def _a_noop(*a, **kw):
|
|
pass
|
|
|
|
|
|
def test_run_max_requests_exceeded(worker):
|
|
loop = asyncio.new_event_loop()
|
|
worker.ppid = 1
|
|
worker.alive = True
|
|
sock = mock.Mock()
|
|
sock.cfg_addr = ("localhost", 8080)
|
|
worker.sockets = [sock]
|
|
worker.wsgi = mock.Mock()
|
|
worker.connections = set()
|
|
worker.log = mock.Mock()
|
|
worker.loop = loop
|
|
worker.servers = {
|
|
"server1": {"requests_count": 14},
|
|
"server2": {"requests_count": 15},
|
|
}
|
|
worker.max_requests = 10
|
|
worker._run = mock.Mock(wraps=_a_noop)
|
|
|
|
# exceeding request count
|
|
_runner = asyncio.ensure_future(worker._check_alive(), loop=loop)
|
|
loop.run_until_complete(_runner)
|
|
|
|
assert not worker.alive
|
|
worker.notify.assert_called_with()
|
|
worker.log.info.assert_called_with(
|
|
"Max requests exceeded, shutting " "down: %s", worker
|
|
)
|
|
|
|
|
|
def test_worker_close(worker):
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.sleep = mock.Mock(wraps=_a_noop)
|
|
worker.ppid = 1
|
|
worker.pid = 2
|
|
worker.cfg.graceful_timeout = 1.0
|
|
worker.signal = mock.Mock()
|
|
worker.signal.stopped = False
|
|
worker.wsgi = mock.Mock()
|
|
conn = mock.Mock()
|
|
conn.websocket = mock.Mock()
|
|
conn.websocket.fail_connection = mock.Mock(wraps=_a_noop)
|
|
worker.connections = set([conn])
|
|
worker.log = mock.Mock()
|
|
worker.loop = loop
|
|
server = mock.Mock()
|
|
server.close = mock.Mock(wraps=lambda *a, **kw: None)
|
|
server.wait_closed = mock.Mock(wraps=_a_noop)
|
|
worker.servers = {server: {"requests_count": 14}}
|
|
worker.max_requests = 10
|
|
|
|
# close worker
|
|
_close = asyncio.ensure_future(worker.close(), loop=loop)
|
|
loop.run_until_complete(_close)
|
|
|
|
assert worker.signal.stopped
|
|
assert conn.websocket.fail_connection.called
|
|
assert len(worker.servers) == 0
|