81494453b0
Change auto reloader enviroment varible name to SANIC_SERVER_RUNNING Fix some typo mistakes, flake uncompatibilities and such problems. Raise NotImplementedError for operating systems except posix systems for auto reloading.
109 lines
3.0 KiB
Python
109 lines
3.0 KiB
Python
import os
|
|
import sys
|
|
import subprocess
|
|
import signal
|
|
from threading import Thread
|
|
from time import sleep
|
|
from json.decoder import JSONDecodeError
|
|
import aiohttp
|
|
import asyncio
|
|
import async_timeout
|
|
|
|
sanic_project_content_one = '''
|
|
from sanic import Sanic
|
|
from sanic import response
|
|
|
|
app = Sanic(__name__)
|
|
|
|
|
|
@app.route("/")
|
|
async def test(request):
|
|
return response.json({"test": 1})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host="127.0.0.1", port=8000, auto_reload=True)
|
|
'''
|
|
|
|
sanic_project_content_two = '''
|
|
from sanic import Sanic
|
|
from sanic import response
|
|
|
|
app = Sanic(__name__)
|
|
|
|
|
|
@app.route("/")
|
|
async def test(request):
|
|
return response.json({"test": 2})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host="127.0.0.1", port=8000, auto_reload=True)
|
|
'''
|
|
|
|
process_id = None
|
|
|
|
|
|
def execute_cmd(command):
|
|
process = subprocess.Popen(command, shell=True)
|
|
global process_id
|
|
process_id = process.pid
|
|
process.communicate()
|
|
|
|
|
|
class TestAutoReloading:
|
|
|
|
def check_response(self, url, response):
|
|
"""Send http request and tries to take it's response as json.
|
|
Returns a dictionary.
|
|
"""
|
|
async def req(url, excepted_response):
|
|
async with aiohttp.ClientSession() as session:
|
|
with async_timeout.timeout(10):
|
|
async with session.get(url) as response:
|
|
try:
|
|
result = await response.json()
|
|
except JSONDecodeError:
|
|
result = {}
|
|
return result == excepted_response
|
|
|
|
loop = asyncio.get_event_loop()
|
|
return loop.run_until_complete(req(url, response))
|
|
|
|
def test_reloading_after_change_file(self, capsys):
|
|
if os.name != 'posix':
|
|
return
|
|
|
|
with capsys.disabled():
|
|
pass
|
|
sanic_app_file_path = "simple_sanic_app.py"
|
|
with open(sanic_app_file_path, "w") as _file:
|
|
_file.write(sanic_project_content_one)
|
|
|
|
cmd = ' '.join([sys.executable, sanic_app_file_path])
|
|
thread = Thread(target=execute_cmd, args=(cmd,))
|
|
thread.start()
|
|
|
|
sleep(2) # wait for completing server start process
|
|
assert self.check_response("http://127.0.0.1:8000/", {"test": 1})
|
|
|
|
with open(sanic_app_file_path, "w") as _file:
|
|
_file.write(sanic_project_content_two)
|
|
|
|
sleep(2) # wait for completing server start process
|
|
assert self.check_response("http://127.0.0.1:8000/", {"test": 2})
|
|
|
|
thread.join(1)
|
|
os.remove(sanic_app_file_path)
|
|
|
|
def teardown_method(self, method):
|
|
if process_id:
|
|
root_proc_path = \
|
|
"/proc/{pid}/task/{pid}/children".format(pid=process_id)
|
|
if not os.path.isfile(root_proc_path):
|
|
return
|
|
with open(root_proc_path) as children_list_file:
|
|
children_list_pid = children_list_file.read().split()
|
|
for child_pid in children_list_pid:
|
|
os.kill(int(child_pid), signal.SIGTERM)
|