ccd4c9615c
Update all tests to be compatible with requests-async Cleanup testing client changes with black and isort Remove Python 3.5 and other meta doc cleanup rename pyproject and fix pep517 error Add black config to tox.ini Cleanup tests and remove aiohttp tox.ini change for easier development commands Remove aiohttp from changelog and requirements Cleanup imports and Makefile
91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
import multiprocessing
|
|
import pickle
|
|
import random
|
|
import signal
|
|
|
|
import pytest
|
|
|
|
from sanic.response import text
|
|
from sanic.testing import HOST, PORT
|
|
|
|
|
|
@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 test_multiprocessing(app):
|
|
"""Tests that the number of children we produce is correct"""
|
|
# Selects a number at random so we can spot check
|
|
num_workers = random.choice(range(2, multiprocessing.cpu_count() * 2 + 1))
|
|
process_list = set()
|
|
|
|
def stop_on_alarm(*args):
|
|
for process in multiprocessing.active_children():
|
|
process_list.add(process.pid)
|
|
process.terminate()
|
|
|
|
signal.signal(signal.SIGALRM, stop_on_alarm)
|
|
signal.alarm(3)
|
|
app.run(HOST, PORT, workers=num_workers)
|
|
|
|
assert len(process_list) == num_workers
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
not hasattr(signal, "SIGALRM"),
|
|
reason="SIGALRM is not implemented for this platform",
|
|
)
|
|
def test_multiprocessing_with_blueprint(app):
|
|
from sanic import Blueprint
|
|
|
|
# Selects a number at random so we can spot check
|
|
num_workers = random.choice(range(2, multiprocessing.cpu_count() * 2 + 1))
|
|
process_list = set()
|
|
|
|
def stop_on_alarm(*args):
|
|
for process in multiprocessing.active_children():
|
|
process_list.add(process.pid)
|
|
process.terminate()
|
|
|
|
signal.signal(signal.SIGALRM, stop_on_alarm)
|
|
signal.alarm(3)
|
|
|
|
bp = Blueprint("test_text")
|
|
app.blueprint(bp)
|
|
app.run(HOST, PORT, workers=num_workers)
|
|
|
|
assert len(process_list) == num_workers
|
|
|
|
|
|
# this function must be outside a test function so that it can be
|
|
# able to be pickled (local functions cannot be pickled).
|
|
def handler(request):
|
|
return text("Hello")
|
|
|
|
|
|
# Muliprocessing on Windows requires app to be able to be pickled
|
|
@pytest.mark.parametrize("protocol", [3, 4])
|
|
def test_pickle_app(app, protocol):
|
|
app.route("/")(handler)
|
|
p_app = pickle.dumps(app, protocol=protocol)
|
|
up_p_app = pickle.loads(p_app)
|
|
assert up_p_app
|
|
request, response = app.test_client.get("/")
|
|
assert response.text == "Hello"
|
|
|
|
|
|
@pytest.mark.parametrize("protocol", [3, 4])
|
|
def test_pickle_app_with_bp(app, protocol):
|
|
from sanic import Blueprint
|
|
|
|
bp = Blueprint("test_text")
|
|
bp.route("/")(handler)
|
|
app.blueprint(bp)
|
|
p_app = pickle.dumps(app, protocol=protocol)
|
|
up_p_app = pickle.loads(p_app)
|
|
assert up_p_app
|
|
request, response = app.test_client.get("/")
|
|
assert app.is_request_stream is False
|
|
assert response.text == "Hello"
|