解决asyncua客户端线程启动超时，实现单入口Pytest测试部署

核心问题定位 #

超时的根本原因是服务端尚未完成启动初始化、端口监听等操作，客户端就发起了连接请求。要在pytest单入口场景下实现两者协同，关键是确保服务端完全就绪后再触发客户端逻辑，同时合理管理异步任务与线程的调度。

具体实现方案 #

1. 编写带就绪等待的asyncua.Server Fixture #

用pytest的异步fixture管理服务端生命周期，添加短延时确保服务端完成启动：

import pytest
import asyncua
from asyncua import Server, Client
import asyncio
from threading import Thread

@pytest.fixture(scope="module")
async def opcua_server():
    # 初始化服务端
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
    # 注册自定义命名空间
    await server.register_namespace("http://test.namespace")
    
    # 启动服务端并等待就绪
    async with server:
        # 等待服务端完成端口监听与初始化（可根据实际调整延时长度）
        await asyncio.sleep(0.5)
        yield server

2. 线程内隔离运行客户端异步任务 #

客户端的异步代码需要在独立线程的事件循环中执行，避免与pytest主事件循环冲突：

def run_client_async_task(task):
    # 为线程创建独立事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(task)

async def client_connect_and_operate():
    # 客户端连接逻辑，endpoint需与服务端完全一致
    async with Client(url="opc.tcp://localhost:4840/freeopcua/server/", timeout=10) as client:
        # 示例操作：获取根节点并验证
        root_node = client.get_root_node()
        assert root_node.node_id.Identifier == 85
        return root_node

3. 测试用例中协调服务端与客户端 #

在测试用例中借助fixture获取就绪的服务端，再启动线程执行客户端任务：

@pytest.mark.asyncio
async def test_opcua_client_server_interaction(opcua_server):
    # 创建客户端异步任务
    client_task = client_connect_and_operate()
    # 启动独立线程运行客户端任务
    client_thread = Thread(target=run_client_async_task, args=(client_task,))
    client_thread.start()
    # 等待客户端任务完成
    client_thread.join()

关键注意事项 #

• 就绪等待不可少：必须给服务端留足初始化时间，避免连接时服务端未就绪
• 事件循环隔离：每个线程需使用独立的asyncio事件循环，防止与pytest主循环冲突
• 端口与端点一致性：服务端和客户端的endpoint必须完全匹配，否则会直接连接失败
• 超时参数调整：可以给Client增加timeout参数（如示例中的10秒），避免因网络延迟导致误判

替代方案：同一事件循环中并行运行（无需线程） #

如果测试场景简单，也可以在pytest的异步测试中，用asyncio.create_task同时运行服务端和客户端任务：

@pytest.mark.asyncio
async def test_opcua_without_thread():
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
    await server.register_namespace("http://test.namespace")
    
    # 启动服务端后台任务
    server_task = asyncio.create_task(server.run_forever())
    await asyncio.sleep(0.5)  # 等待服务端就绪
    
    try:
        # 执行客户端逻辑
        async with Client(url="opc.tcp://localhost:4840/freeopcua/server/") as client:
            root_node = client.get_root_node()
            assert root_node is not None
    finally:
        # 确保服务端正常停止
        server.stop()
        await server_task

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