Python Socket服务端Pytest单元测试卡顿问题排查求助

核心问题 #

Socket服务端的阻塞式监听循环是罪魁祸首——如果测试代码直接触发这个循环，会把pytest的测试收集进程卡死，一直停在collecting状态。

具体修复方案 #

1. 拆分服务端代码，避免模块加载时阻塞 #

把服务端的初始化和启动逻辑分开，只在直接运行服务端脚本时才启动循环：
原SERVER.py如果是模块级直接启动循环，改成：

import socket

def create_server(port=8080):
    """创建并初始化socket服务端"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # 允许端口复用，避免测试后端口占用
    sock.bind(('localhost', port))
    sock.listen(5)
    return sock

def run_server(sock):
    """启动阻塞式监听循环"""
    while True:
        conn, addr = sock.accept()
        # 处理客户端请求的逻辑
        conn.send(b"Hello from server")
        conn.close()

if __name__ == '__main__':
    # 只有直接运行脚本时才启动服务
    server_sock = create_server()
    run_server(server_sock)

2. 用后台线程运行服务端 #

测试代码里绝对不能直接调用run_server，必须把服务端放到后台线程，避免阻塞测试主线程：

import pytest
import threading
import socket
from SERVER import create_server, run_server

@pytest.fixture(scope="module")
def server():
    """启动后台服务端的fixture"""
    sock = create_server()
    # 用守护线程运行，测试结束后自动退出
    server_thread = threading.Thread(target=run_server, args=(sock,), daemon=True)
    server_thread.start()
    yield sock
    # 测试完成后关闭socket
    sock.close()

def test_server_response(server):
    """测试服务端响应"""
    client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client_sock.connect(('localhost', 8080))
    client_sock.send(b"Test request")
    response = client_sock.recv(1024)
    assert response == b"Hello from server"
    client_sock.close()

3. 调整pytest-cov参数，避免不必要的覆盖率收集 #

如果是覆盖率工具导致的额外阻塞，精准指定要收集的模块，不要全局扫描：

pytest test_server.py --cov=SERVER --cov-report=xml

用--cov=SERVER只收集服务端代码的覆盖率，避免扫描测试文件或其他无关模块。

4. 检查测试文件的模块级代码 #

确保test_server.py里没有在模块顶部直接执行服务端启动逻辑——比如不要在测试文件开头写run_server(create_server())，这种代码会在pytest收集测试用例时就触发阻塞。

5. 处理端口占用问题 #

测试前检查端口是否被占用，避免服务端启动失败导致卡住：

def check_port_available(port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        return s.connect_ex(('localhost', port)) != 0

@pytest.fixture(scope="module")
def server():
    port = 8080
    if not check_port_available(port):
        pytest.skip(f"Port {port} is occupied, skip test")
    sock = create_server(port)
    # ... 后续启动线程逻辑

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