sanic/tests/test_multiprocessing.py

26 lines
731 B
Python
Raw Permalink Normal View History

import multiprocessing
import random
import signal
2016-10-18 09:22:49 +01:00
from sanic import Sanic
2018-03-16 04:28:52 +00:00
from sanic.testing import HOST, PORT
2016-10-18 09:22:49 +01:00
def test_multiprocessing():
"""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))
app = Sanic('test_multiprocessing')
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)
2018-03-16 05:06:58 +00:00
signal.alarm(3)
2018-03-16 04:28:52 +00:00
app.run(HOST, PORT, workers=num_workers)
assert len(process_list) == num_workers