230941ff4f
* Fix watchdog reload worker repeatedly if there are multiple changed files * Simplify autoreloader, don't need multiprocessing.Process. Now works on OSX py38. * Allow autoreloader with multiple workers and run it earlier. * This works OK on Windows too. * I don't see how cwd could be different here. * app.run and app.create_server argument fixup. * Add test for auto_reload (coverage not working unfortunately). * Reloader cleanup, don't use external kill commands and exit normally. * Strip newlines on test output (Windows-compat). * Report failures in test_auto_reload to avoid timeouts. * Use different test server ports to avoid binding problems on Windows. * Fix previous commit * Listen on same port after reload. * Show Goin' Fast banner on reloads. * More robust testing, also -m sanic. * Add a timeout to terminate process * Try a workaround for tmpdir deletion on Windows. * Join process also on error (context manager doesn't). * Cleaner autoreloader termination on Windows. * Remove unused code. * Rename test. * Longer timeout on test exit. Co-authored-by: Hùng X. Lê <lexhung@gmail.com> Co-authored-by: L. Kärkkäinen <tronic@users.noreply.github.com> Co-authored-by: Adam Hopkins <admhpkns@gmail.com>
89 lines
2.5 KiB
Python
89 lines
2.5 KiB
Python
import os
|
|
import secrets
|
|
import sys
|
|
|
|
from subprocess import PIPE, Popen
|
|
from tempfile import TemporaryDirectory
|
|
from textwrap import dedent
|
|
from threading import Timer
|
|
from time import sleep
|
|
|
|
import pytest
|
|
|
|
|
|
# We need to interrupt the autoreloader without killing it, so that the server gets terminated
|
|
# https://stefan.sofa-rockers.org/2013/08/15/handling-sub-process-hierarchies-python-linux-os-x/
|
|
|
|
try:
|
|
from signal import CTRL_BREAK_EVENT
|
|
from subprocess import CREATE_NEW_PROCESS_GROUP
|
|
|
|
flags = CREATE_NEW_PROCESS_GROUP
|
|
except ImportError:
|
|
flags = 0
|
|
|
|
def terminate(proc):
|
|
if flags:
|
|
proc.send_signal(CTRL_BREAK_EVENT)
|
|
else:
|
|
proc.terminate()
|
|
|
|
def write_app(filename, **runargs):
|
|
text = secrets.token_urlsafe()
|
|
with open(filename, "w") as f:
|
|
f.write(dedent(f"""\
|
|
import os
|
|
from sanic import Sanic
|
|
|
|
app = Sanic(__name__)
|
|
|
|
@app.listener("after_server_start")
|
|
def complete(*args):
|
|
print("complete", os.getpid(), {text!r})
|
|
|
|
if __name__ == "__main__":
|
|
app.run(**{runargs!r})
|
|
"""
|
|
))
|
|
return text
|
|
|
|
def scanner(proc):
|
|
for line in proc.stdout:
|
|
line = line.decode().strip()
|
|
print(">", line)
|
|
if line.startswith("complete"):
|
|
yield line
|
|
|
|
|
|
argv = dict(
|
|
script=[sys.executable, "reloader.py"],
|
|
module=[sys.executable, "-m", "reloader"],
|
|
sanic=[sys.executable, "-m", "sanic", "--port", "42104", "--debug", "reloader.app"],
|
|
)
|
|
|
|
@pytest.mark.parametrize("runargs, mode", [
|
|
(dict(port=42102, auto_reload=True), "script"),
|
|
(dict(port=42103, debug=True), "module"),
|
|
(dict(), "sanic"),
|
|
])
|
|
async def test_reloader_live(runargs, mode):
|
|
with TemporaryDirectory() as tmpdir:
|
|
filename = os.path.join(tmpdir, "reloader.py")
|
|
text = write_app(filename, **runargs)
|
|
proc = Popen(argv[mode], cwd=tmpdir, stdout=PIPE, creationflags=flags)
|
|
try:
|
|
timeout = Timer(5, terminate, [proc])
|
|
timeout.start()
|
|
# Python apparently keeps using the old source sometimes if
|
|
# we don't sleep before rewrite (pycache timestamp problem?)
|
|
sleep(1)
|
|
line = scanner(proc)
|
|
assert text in next(line)
|
|
# Edit source code and try again
|
|
text = write_app(filename, **runargs)
|
|
assert text in next(line)
|
|
finally:
|
|
timeout.cancel()
|
|
terminate(proc)
|
|
proc.wait(timeout=3)
|