MQTT 设备模拟测试：用一条消息链路验证整套上报协议

为什么值得单独做一套模拟测试

设备侧问题往往到现场才暴露：某个点位主题写错、QoS 设成 0 导致丢包、遗留消息让新订阅者读到脏数据。这些在真实设备上复现成本极高，而模拟器一次运行就能覆盖。

环境起停：一行命令拉起 broker

services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports: ["1883:1883"]
    volumes: ["./mosquitto/config/mosquitto.conf:/mosquitto/config/mosquitto.conf"]

listener 1883
allow_anonymous true
persistence false

客户端封装：三个细节决定稳定性

class MQTTBase:
    def connect(self, timeout: float = 10.0):
        rc = self._client.connect(self.host, self.port, self.keepalive)
        assert rc == mqtt.MQTT_ERR_SUCCESS, f"MQTT 连接失败, rc={rc}"
        self._client.loop_start()
        # CONNACK 是异步回调，必须等连接真正建立再返回
        deadline = time.monotonic() + timeout
        while not self._connected:
            if time.monotonic() >= deadline:
                raise ConnectionError(f"MQTT 未在 {timeout}s 内完成连接")
            time.sleep(0.05)
        return self

• 等待握手：不等待的话 is_connected 断言是竞态，本地必绿、CI 必红。
• QoS > 0 要等发布完成：info.wait_for_publish(timeout=10)，否则断言时消息可能还没出去。
• 消息队列隔离：每个客户端独立 queue，用例间不互相污染。

用例设计：四类协议行为

1. 回路验证（最基础）

for case in mqtt_data["roundtrip"]:
    mqtt_client.subscribe(case["topic"], qos=case["qos"])
    mqtt_client.publish(case["topic"], case["payload"], qos=case["qos"])
    msg = mqtt_client.wait_message(timeout=5)
    assert msg.payload.decode() == case["payload"]

2. 遗留消息（Retained）

新订阅者应立即收到上次保留的消息。同时必须记得清理，否则会污染后续运行：

mqtt_client.publish(topic, payload, retain=True)
late_sub = MQTTBase(host, port).connect()
msg = late_sub.wait_message(timeout=5)
assert msg.retain
# 清理：发空消息覆盖
mqtt_client.publish(topic, "", retain=True)

3. 通配符订阅

/dg/get/# 能匹配任意层级子主题——这是设备上报点位最常用的订阅方式，必须单独验证。

4. 数据驱动的点位表

# TestDatas/mqtt_data.yaml
roundtrip:
  - { topic: "/dg/get/06", payload: '{"soc": 85}', qos: 1 }
  - { topic: "/dg/get/07", payload: '{"pcszt": 42.6}', qos: 0 }
retained:
  { topic: "/dg/device/online", payload: '{"state": 1}' }

接入 CI 的两个坑

1. 端口被系统服务抢占：GitHub Runner 上 apt install mosquitto 会自动启动默认配置的系统服务，占住 1883 且拒绝匿名连接。必须先停：

sudo apt-get install -y -qq mosquitto
sudo systemctl stop mosquitto 2>/dev/null || true
mosquitto -c mosquitto/config/mosquitto.conf -d
sleep 2

2. 版本锁定：requirements.txt 里 paho-mqtt>=1.6.1,<2.0。2.x 改了回调 API 签名，1.x 的 on_connect(client, userdata, flags, rc) 写法不再适用，升级时必须同步改代码。

本地没有 Mosquitto 怎么办

不想装二进制，可以用纯 Python 的 amqtt 临时起一个：

pip install amqtt
amqtt -c broker.yaml   # listeners.default.bind: 127.0.0.1:1883

协议行为一致，适合验证用例逻辑；正式环境还是用 Mosquitto 更接近真实。