使用线程运行pytest导致后续测试失败的问题排查

核心原因分析 #

你的问题本质是FlightServer测试残留的子进程/进程组干扰了后续测试的PID复用与进程管理，具体有几个可能的触发点：

1. 未彻底清理服务器子进程：pa.start(self.arrow_server)可能启动了独立子进程处理请求，但self.arrow_server.shutdown()仅终止了主线程逻辑，没递归终止这些子进程。残留的孤儿进程会占用PID，后续测试启动的新进程可能复用这些已被系统回收的PID，导致tearDown时执行kill操作抛出ProcessLookupError。
2. 进程组操作冲突：后续测试使用os.killpg杀死进程组，如果FlightServer启动时修改了当前测试进程的进程组设置，os.killpg会误杀无关进程（包括FlightServer残留进程或其他测试进程），导致后续测试的目标进程提前终止，最终出现PID不存在的错误。
3. 线程终止不彻底：虽然设置了daemon=True，但pa.start()可能是阻塞循环，shutdown()未正确触发循环退出，线程持续存活导致服务器资源无法释放。

具体解决步骤 #

1. 强制清理服务器子进程 #

在tearDown中补充子进程清理逻辑，通过端口筛选并终止FlightServer相关的子进程：

import psutil
import platform

def tearDown(self):
    self.arrow_server.shutdown()
    self.t.join(timeout=5)  # 给线程预留退出时间
    
    # 清理占用目标端口的子进程
    if platform.system() != "Windows":
        current_process = psutil.Process()
        for child in current_process.children(recursive=True):
            try:
                # 通过端口判断是否为FlightServer子进程
                for conn in child.connections():
                    if conn.laddr.port == 13620:
                        child.terminate()
                        child.wait(timeout=3)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                pass
    
    self.arrow_server = None

（需提前安装psutil：pip install psutil）

2. 隔离进程组避免冲突 #

在服务器线程启动时，显式将其放入独立进程组，避免后续测试的killpg操作误杀无关进程：

import os

def setUp(self):
    host = "localhost"
    port = 13620
    scheme = "grpc+tcp"
    location = "{}://{}:{}".format(scheme, host, port)
    self.arrow_server = FlightServer(host, location)
    
    def start_server():
        # 让服务器子进程在独立进程组运行
        if platform.system() != "Windows":
            os.setpgid(0, 0)
        pa.start(self.arrow_server)
    
    self.t = threading.Thread(target=start_server, args=(), daemon=True)
    self.t.start()

3. 改用pytest Fixture管理资源 #

将服务器的启动/终止逻辑改为pytest的function级fixture，确保资源在每个测试前后被严格清理，避免类级setUp/tearDown的潜在泄漏：

import pytest
import threading
import psutil
import pa

@pytest.fixture(scope="function")
def flight_server():
    host = "localhost"
    port = 13620
    scheme = "grpc+tcp"
    location = "{}://{}:{}".format(scheme, host, port)
    arrow_server = pa.FlightServer(host, location)
    server_thread = None
    
    try:
        def start_server():
            if platform.system() != "Windows":
                os.setpgid(0, 0)
            pa.start(arrow_server)
        
        server_thread = threading.Thread(target=start_server, daemon=True)
        server_thread.start()
        yield arrow_server
    finally:
        arrow_server.shutdown()
        if server_thread:
            server_thread.join(timeout=5)
        
        # 清理残留子进程
        current_process = psutil.Process()
        for child in current_process.children(recursive=True):
            try:
                for conn in child.connections():
                    if conn.laddr.port == port:
                        child.terminate()
                        child.wait(timeout=3)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                pass

测试时直接注入fixture即可：

def test_flight_server_function(flight_server):
    # 测试逻辑代码
    pass

验证方法 #

• 运行FlightServer测试后，用netstat -tulpn | grep 13620检查端口是否被释放；
• 用ps aux | grep <服务器进程名>确认无残留子进程。

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