Add test for coverage, avoid crash on missing _stopping.
This commit is contained in:
parent
6ea818395e
commit
68fb963539
|
@ -31,9 +31,10 @@ else:
|
|||
def ctrlc_workaround_for_windows(app):
|
||||
async def stay_active(app):
|
||||
"""Frequently poll asyncio to allow *receiving* any signals in Python"""
|
||||
loop = asyncio.get_running_loop()
|
||||
while not die:
|
||||
# If someone else stopped the app, just exit
|
||||
if asyncio.get_running_loop()._stopping:
|
||||
if getattr(loop, "_stopping", False):
|
||||
return
|
||||
await asyncio.sleep(0.1)
|
||||
# Can't be called from signal handler, so call it from here
|
||||
|
|
|
@ -5,6 +5,7 @@ import signal
|
|||
from queue import Queue
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from sanic.compat import ctrlc_workaround_for_windows
|
||||
from sanic.response import HTTPResponse
|
||||
from sanic.testing import HOST, PORT
|
||||
|
||||
|
@ -58,3 +59,24 @@ def test_dont_register_system_signals(app):
|
|||
|
||||
app.run(HOST, PORT, register_sys_signals=False)
|
||||
assert calledq.get() is False
|
||||
|
||||
|
||||
def test_windows_workaround(app):
|
||||
"""Test Windows workaround (on any OS)"""
|
||||
too_slow = False
|
||||
|
||||
@app.add_task
|
||||
async def signaler(app):
|
||||
ctrlc_workaround_for_windows(app)
|
||||
await asyncio.sleep(0.1)
|
||||
os.kill(os.getpid(), signal.SIGINT)
|
||||
|
||||
@app.add_task
|
||||
async def timeout(app):
|
||||
nonlocal too_slow
|
||||
await asyncio.sleep(1)
|
||||
too_slow = True
|
||||
app.stop()
|
||||
|
||||
app.run(HOST, PORT)
|
||||
assert not too_slow
|
||||
|
|
Loading…
Reference in New Issue
Block a user