sanic/tests/test_multiprocessing.py

import multiprocessing
import random
import signal
import pytest

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