使用 pytest 进行多处理时正确拆卸进程

Question

我正在开发一个包，其中包括带有相应客户端的服务器。 我想我可以节省大量的精力来模拟所有的交互，只需要编写一些同时测试双方的测试用例即可。

所以我编写了一个固定装置，它将产生一个 multiprocessing.Process 实例，服务器在其中运行。

测试完成后，我使用 process.terminate() 拆除进程。不幸的是，还有另一个“孤立”进程，我似乎无法处理。 如何删除其他进程或者如何避免首先启动它？ 如果您知道使用与多处理不同的库来执行此操作的方法，我完全愿意。 为了重现该问题，我使用了 how-to-make-a-simple-python-rest-server-and-client 中的代码来创建一个简单的 Rest 服务器/客户端对。

import asyncio
import logging
import multiprocessing as mp

import pytest
import pytest_asyncio
import requests
from flask import Flask, jsonify


def server():
    app = Flask(__name__)

    tasks = [
        {
            "id": 1,
            "title": "Buy groceries",
            "description": "Milk, Cheese, Pizza, Fruit, Tylenol",
            "done": False,
        },
        {
            "id": 2,
            "title": "Learn Python",
            "description": "Need to find a good Python tutorial on the web",
            "done": False,
        },
    ]

    @app.route("/todo/api/v1.0/tasks", methods=["GET"])
    def get_tasks():
        return jsonify({"tasks": tasks})

    app.run()


class TestMultiprocessingTester:

    @pytest.fixture(scope="class")
    def server_process(self):
        server_process = mp.Process(target=server)
        server_process.start()
        yield server_process
        server_process.terminate()
        if server_process.is_alive():
            server_process.kill()
        logging.getLogger("Process").warning("shut down server process")

    @pytest.mark.asyncio
    async def test_client(self, server_process: mp.Process):
        await asyncio.sleep(10)
        url = "http://localhost:5000/todo/api/v1.0/tasks"
        response = requests.get(url)
        assert response.status_code == 200
        assert response.json()["tasks"][0]["done"] is False

Answer 1

为了能够骑僵尸，医生说的是：

在 Unix 上，当一个进程完成但尚未加入时，它会变成 僵尸。永远不应该有太多，因为每次都有一个新的 进程启动（或调用 active_children()）全部完成 尚未加入的进程将被加入。还打电话 已完成进程的 Process.is_alive 将加入该进程。即便如此 显式加入所有进程可能是一个好习惯 你开始吧。

所以为了避免僵尸，你需要在杀死进程后调用 join() 方法。