使用pytest+multiprocessing测试时的孤儿进程销毁问题

问题根源 #

你遇到的孤儿进程，大概率是因为Flask在debug模式下会自动启动重载子进程（reloader），而你用multiprocessing.Process启动的主进程被terminate()杀掉后，这个重载子进程失去父进程，就变成了孤儿进程。另外，单纯用terminate()可能没给进程足够时间清理资源，也会导致残留。

可行解决方案 #

1. 关闭Flask的重载与debug模式 #

测试环境完全不需要debug和自动重载，直接关闭这两个选项，让Flask只启动单个进程：

def run_server():
    app.run(
        host='127.0.0.1',
        port=5000,
        debug=False,  # 关闭debug
        use_reloader=False  # 强制关闭重载子进程
    )

这样multiprocessing.Process启动的就是唯一的服务进程，terminate()可以彻底杀掉它。

2. 优化进程销毁逻辑 #

在terminate()之后加上join()，给进程足够时间退出；同时设置daemon=True，让子进程随测试主进程自动退出：

import pytest
from multiprocessing import Process
from my_flask_app import app

def run_server():
    app.run(host='127.0.0.1', port=5000, debug=False, use_reloader=False)

@pytest.fixture(scope="module")
def flask_server():
    # 设置daemon=True，主进程退出时子进程自动终止
    process = Process(target=run_server, daemon=True)
    process.start()
    yield process
    # 先终止进程，再等待退出，避免残留
    process.terminate()
    process.join(timeout=5)  # 超时5秒强制结束

3. 用pytest-flask插件替代手动进程管理 #

pytest-flask有内置的fixture，会自动帮你管理Flask服务的生命周期，完全不用手动处理进程：

1. 安装插件：pip install pytest-flask
2. 在conftest.py里配置：

import pytest
from my_flask_app import app

@pytest.fixture
def client():
    app.config['TESTING'] = True
    # 用test_client自动启动/关闭服务
    with app.test_client() as client:
        yield client

3. 测试用例直接用client fixture，不用管服务启动：

def test_example(client):
    response = client.get('/')
    assert response.data == b"Hello World"

4. 改用subprocess控制进程组 #

如果必须用独立进程，subprocess可以通过进程组一次性杀掉所有子进程，避免残留：

import pytest
import subprocess
import time

@pytest.fixture(scope="module")
def flask_server():
    # start_new_session=True创建新进程组
    proc = subprocess.Popen(
        ["python", "-m", "flask", "run", "--host=127.0.0.1", "--port=5000", "--no-debug", "--no-reload"],
        start_new_session=True
    )
    time.sleep(1)  # 等待服务启动完成
    yield proc
    # 杀掉整个进程组，包括所有子进程
    proc.terminate()
    proc.wait(timeout=5)

问题复现代码 #

# conftest.py
import pytest
from multiprocessing import Process
from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello():
    return "Hello World"

def run_server():
    # debug=True会触发Flask的重载子进程
    app.run(host='127.0.0.1', port=5000, debug=True)

@pytest.fixture(scope="session")
def server():
    p = Process(target=run_server)
    p.start()
    yield
    p.terminate()

# test_client.py
def test_client(server):
    import requests
    response = requests.get('http://127.0.0.1:5000/')
    assert response.text == "Hello World"

内容的提问来源于stack exchange，提问作者Uuruurrgh