5cf2144b3f
Change the string passed in the "name" section of the namedtuples in Blueprint to match the name of the Blueprint module attribute name. This allows blueprints to be pickled and unpickled, without errors, which is a requirment of running Sanic in multiprocessing mode in Windows. Added a test for pickling and unpickling blueprints Added a test for pickling and unpickling sanic itself Added a test for enabling multiprocessing on an app with a blueprint (only useful to catch this bug if the tests are run on Windows).
83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
import multiprocessing
|
|
import random
|
|
import signal
|
|
import pickle
|
|
import pytest
|
|
|
|
from sanic.testing import HOST, PORT
|
|
from sanic.response import text
|
|
|
|
|
|
@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
|
|
|
|
|
|
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_pikcle_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'
|