51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
|
"""
|
||
|
|
测试用模拟对象 — 不连接真实 MQTT Broker
|
||
|
|
|
||
|
|
提供 ``FakeMQTT``（模拟 amqtt ``MQTTClient`` 的最小接口）与辅助函数，
|
||
|
|
用于对 ``chongming_mqtt`` 各客户端做离线单元测试。
|
||
|
|
"""
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
|
||
|
|
class FakeMQTT:
|
||
|
|
"""
|
||
|
|
模拟 amqtt MQTTClient 的最小接口
|
||
|
|
|
||
|
|
记录 publish / subscribe / unsubscribe / disconnect 调用；
|
||
|
|
``on_publish`` 回调可在发布时模拟"对方应用"的行为（如回发响应）。
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, on_publish=None):
|
||
|
|
self.published: list[tuple[str, bytes, object]] = [] # (topic, payload, qos)
|
||
|
|
self.subscribed: list[tuple[str, object]] = [] # (topic, qos)
|
||
|
|
self.unsubscribed: list[str] = []
|
||
|
|
self.disconnected = False
|
||
|
|
self._on_publish = on_publish
|
||
|
|
|
||
|
|
async def publish(self, topic, payload, qos):
|
||
|
|
self.published.append((topic, payload, qos))
|
||
|
|
if self._on_publish is not None:
|
||
|
|
await self._on_publish(topic, payload)
|
||
|
|
|
||
|
|
async def subscribe(self, topics):
|
||
|
|
self.subscribed.extend(list(topics))
|
||
|
|
|
||
|
|
async def unsubscribe(self, topics):
|
||
|
|
self.unsubscribed.extend(list(topics))
|
||
|
|
|
||
|
|
async def disconnect(self):
|
||
|
|
self.disconnected = True
|
||
|
|
|
||
|
|
|
||
|
|
def attach(client, on_publish=None) -> FakeMQTT:
|
||
|
|
"""把 FakeMQTT 挂到被测客户端（跳过真实连接），并标记为运行中"""
|
||
|
|
fake = FakeMQTT(on_publish=on_publish)
|
||
|
|
client._client = fake
|
||
|
|
client._running = True
|
||
|
|
return fake
|
||
|
|
|
||
|
|
|
||
|
|
def run(coro):
|
||
|
|
"""运行一个异步协程（同步测试入口）"""
|
||
|
|
return asyncio.run(coro)
|