用 Mock WebSocket 服务解耦 App 与后端的联调

App 端测试最大的依赖不是后端接口，而是长连接——客服会话、消息推送、状态同步全走 WebSocket。后端一挂或环境一脏，App 用例全陪葬。解法是做一个 Mock WebSocket 服务：协议兼容、消息可编程、客户端可批量制造。

一、协议先行：把真实协议固化成契约

Mock 的前提是有"真"的参照。第一步不是写代码，而是从真实后端抓一周的流量，沉淀出协议契约：

# config/mock_templates.yaml
templates:
  reply-message:
    desc: 买家消息回复
    params:
      user_name: { label: 买家昵称, type: string }
      conversation_id: { label: 会话ID, type: string }
      content: { label: 回复内容, type: string, default: "您好，请问需要什么帮助？" }
    body: |
      {"type":"reply-message","request_id":"{{uuid}}","task_id":"{{task}}",
       "priority":2,"data":{"user_name":"{{user_name}}",
       "conversation_id":"{{conversation_id}}","content":"{{content}}"}}

  read-receipt:
    desc: 已读回执
    params: { conversation_id: { type: string } }
    body: '{"type":"read-receipt","data":{"conversation_id":"{{conversation_id}}"}}'

  typing:
    desc: 对方正在输入
    params: { user_name: { type: string } }
    body: '{"type":"typing","data":{"user_name":"{{user_name}}"}}'

模板 + 参数占位符 + {{uuid}}/{{task}} 内置变量，Web 控制台填参数即出报文——Mock 的可编程性来自模板，不是来自硬编码。

二、服务设计：广播 + 定向 + 录制回放

class MockWsServer:
    """核心就三件事：管理连接、定向下发、广播、记录流水"""

    def __init__(self):
        self.clients: dict[str, WebSocket] = {}   # conn_id -> ws
        self.history: deque[dict] = deque(maxlen=2000)

    async def broadcast(self, payload: dict):
        """广播给所有活跃连接（模拟服务端主动推送）"""
        for conn in self.clients.values():
            await conn.send_json(payload)
        self.history.append({"dir": "s2c", "conn": "*", **payload})

    async def send_to(self, conn_id: str, payload: dict):
        """定向下发（模拟"给某买家回消息"）"""
        ...

    async def on_client_message(self, conn_id: str, payload: dict):
        """c2s 方向：App 发来的帧，记录 + 按规则自动应答"""
        self.history.append({"dir": "c2s", "conn": conn_id, **payload})
        await self._auto_reply(conn_id, payload)

三个设计点：

自动应答规则表：App 发 login 帧必须回 login-ok，否则后续全断。把这类"握手型应答"写成规则表自动触发，模板下发留给业务帧——Mock 既要"像服务端"，又要"可控"；
双方向流水（c2s/s2c）全部进环形缓冲，Web 控制台实时滚动展示——联调时"App 到底发了什么"一眼可见，比在 App 里抓包快十倍；
客户端可批量制造：Mock 客户端脚本可以一次连 20 个"假买家"，模拟群发消息、多会话并发，这是真实环境造不出的数据。

三、与 UI 自动化联动

Mock 服务的价值在"场景编排"：

@pytest.fixture(scope="session")
def mock():
    server = MockWsClient(base_url="http://127.0.0.1:8900")
    yield server
    server.reset()

def test_unread_badge(mock, app_driver):
    mock.connect_fake_buyer("buyer_001")
    open_chat_list(app_driver)
    mock.send_to(app_conn, make("reply-message", user_name="buyer_001",
                                content="在吗？订单什么时候发货？"))
    # 断言：角标 +1、列表顶部出现新会话、红点样式
    assert unread_badge(app_driver) == 1

测试矩阵：同一套 UI 用例 × 三类服务端行为（正常推送 / 延迟推送 / 乱序推送），乱序推送是真实网络下 WebSocket 帧乱序的模拟——typing 事件晚于 reply-message 到达时，UI 不能闪烁、不能把"正在输入"残留到消息之后。这种 case 只在 Mock 环境造得出。

四、为什么不用 WireMock / 现成工具

需求	现成工具	自研 Mock WS
HTTP 接口 Mock	✅ 强	不需要
WS 帧级控制（延迟/乱序/断连）	❌ 基本没有	✅
业务模板参数化	❌	✅
与 UI 用例 fixture 级联动	⚠️ 要自己桥	✅ 原生 Python

现成工具解决 80% 的"接口 Mock"，但 WS 的帧级时序控制是自动化测试的刚需，这部分只能自己写——好在核心逻辑（连接管理 + 广播 + 流水）不到 300 行，Web 控制台只是它的一个客户端。

五、小结

Mock WebSocket 服务让 App 测试从"依赖环境"变成"编排环境"：协议契约 YAML 化、帧时序可编程、双方向流水透明。联调效率的提升是量级的——过去一个"消息角标不更新"的排查要拉后端抓包两小时，现在 Mock 里重放一遍 5 分钟定位。

ScofieldYu