如何在外部测试模块获取Python旧版websocket-client的消息数据？

由于你无法修改现有Client类，核心思路是通过拦截on_message回调来捕获收到的消息数据，以下是两种可行方案：

方案1：猴子补丁（无需修改原类） #

直接在测试代码中动态替换Client实例的on_message方法，既保留原有逻辑，又将消息存入可外部访问的容器。

测试代码示例 #

from **** import Client
import pytest
import threading
from json import loads

@pytest.fixture
def client_with_capture():
    client = Client()
    # 初始化存储消息的容器
    received_data = []
    
    # 保存原方法，定义带捕获逻辑的新方法
    original_on_msg = client.on_message
    def capture_on_message(ws, message):
        # 执行原有的消息处理逻辑
        original_on_msg(ws, message)
        # 解析并存储消息数据
        parsed_data = loads(message)
        received_data.append(parsed_data)
    
    # 替换实例的on_message方法
    client.on_message = capture_on_message
    
    # 启动WebSocket连接在后台线程（避免阻塞测试）
    ws_thread = threading.Thread(target=client.ws.run_forever, daemon=True)
    ws_thread.start()
    
    # 返回客户端实例和消息容器
    yield client, received_data
    
    # 测试收尾：关闭连接
    client.ws.close()

def test_websocket_message(client_with_capture):
    client, received_data = client_with_capture
    
    # 可选：向服务器发送触发消息的指令（比如订阅请求）
    # client.ws.send('{"action": "subscribe_to_data"}')
    
    # 等待消息接收（带超时，防止无限等待）
    import time
    timeout = 5
    start_time = time.time()
    while not received_data and time.time() - start_time < timeout:
        time.sleep(0.1)
    
    # 断言验证
    assert len(received_data) > 0, "未收到任何消息"
    assert received_data[0].get("event_type") == "expected_event", "消息内容不符合预期"

方案2：子类继承扩展 #

创建Client的子类，重写on_message方法来添加消息存储逻辑，原类代码无需改动。

测试代码示例 #

from **** import Client
import pytest
import threading
from json import loads

class TestClient(Client):
    def __init__(self):
        super().__init__()
        # 新增存储消息的属性
        self.received_data = []
    
    def on_message(self, ws, message):
        # 执行父类原有的消息处理
        super().on_message(ws, message)
        # 存储解析后的数据
        parsed_data = loads(message)
        self.received_data.append(parsed_data)

@pytest.fixture
def test_client():
    client = TestClient()
    # 后台线程启动WebSocket监听
    ws_thread = threading.Thread(target=client.ws.run_forever, daemon=True)
    ws_thread.start()
    yield client
    client.ws.close()

def test_websocket_message(test_client):
    # 等待消息接收
    import time
    timeout = 5
    start_time = time.time()
    while not test_client.received_data and time.time() - start_time < timeout:
        time.sleep(0.1)
    
    # 断言验证
    assert len(test_client.received_data) > 0
    assert test_client.received_data[0].get("data_value") == 12345

注意事项 #

1. 确认on_message的签名：websocket-client 0.57.0版本中，on_message回调的正确参数是(ws, message)，如果原代码漏了ws参数，需要补全否则会报错。
2. 必须用后台线程运行run_forever()，否则该方法会阻塞测试流程。
3. 等待消息时添加超时逻辑，避免测试因网络问题无限挂起。

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