如何通过等待事件触发的pytest测试（EventEmitter+GCP Pub/Sub场景）

解决方案 #

核心问题分析 #

1. 执行顺序错误：先调用pubsub.listen（会阻塞至超时），之后才注册事件监听器。等监听器注册完成时，事件早已发射完毕，根本无法监听。
2. 监听器注册逻辑错误：received(func)是把func当作参数传入received函数并立即执行，注册给事件的其实是received的返回值（也就是func），完全不符合事件监听的逻辑——监听器应该是接收消息的函数，而非提前执行的结果。
3. 依赖真实GCP服务：直接调用真实Pub/Sub客户端会导致测试阻塞、依赖网络，且无法控制推送的消息内容。
4. 代码小问题：message.ack缺少括号，应改为message.ack()，否则仅引用方法，未实际确认消息。

修正后的测试代码 #

from unittest import mock

def test_pubsub_listen():
    # 1. 准备模拟的Pub/Sub消息
    mock_msg = mock.Mock()
    mock_msg.data = b"expected message"  # Pub/Sub消息数据为bytes类型
    mock_msg.ack = mock.Mock()  # 模拟ack方法

    # 2. 先注册监听器（关键：先注册，再触发事件）
    receive_mock = mock.Mock()
    events.on("subscription_id", receive_mock)

    # 3. Mock GCP Subscriber客户端，避免真实调用
    with mock.patch.object(pubsub.subscriber, "subscribe") as mock_subscribe:
        # 自定义subscribe行为：直接触发传入的callback函数
        def mock_subscribe_behaviour(sub_path, callback):
            # 模拟收到消息，触发回调
            callback(mock_msg)
            # 返回mock的future对象，避免阻塞
            return mock.Mock(result=mock.Mock(side_effect=TimeoutError))
        
        mock_subscribe.side_effect = mock_subscribe_behaviour

        # 4. 调用listen方法，此时会立即触发回调并发射事件
        pubsub.listen(subscription_id, 1)

    # 5. 断言监听器被正确调用
    receive_mock.assert_called_once_with(mock_msg)
    # 验证消息内容是否正确（可选）
    assert receive_mock.call_args[0][0].data == b"expected message"
    # 验证ack方法被调用（可选）
    mock_msg.ack.assert_called_once()

不用Mock的替代方案 #

如果不想依赖mock.Mock()，可以用自定义捕获函数保存消息：

def test_pubsub_listen_with_custom_func():
    captured_msg = None
    
    def capture_message(msg):
        nonlocal captured_msg
        captured_msg = msg
        print(msg)

    events.on("subscription_id", capture_message)

    # 同样mock GCP客户端触发回调...
    pubsub.listen(subscription_id, 1)

    assert captured_msg is not None
    assert captured_msg.data == b"expected message"

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