65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
"""
|
||
MQTTForwardClient 单元测试
|
||
|
||
通过注入 FakeMQTT 验证发布逻辑，不连接真实 Broker。
|
||
"""
|
||
import json
|
||
|
||
from chongming_mqtt.mqtt_forward import MQTTForwardClient
|
||
|
||
from fakes import attach, run
|
||
|
||
|
||
def test_uri_credentials_decoded():
|
||
"""URI 中 percent-encoded 凭据应被解码（%40 -> @）"""
|
||
client = MQTTForwardClient("mqtt://admin:Mixc%402021@dg.datagoo.cn:48081", "/t")
|
||
assert client._username == "admin"
|
||
assert client._password == "Mixc@2021"
|
||
assert client._clean_uri == "mqtt://dg.datagoo.cn:48081"
|
||
|
||
|
||
def test_publish_not_running_returns_false():
|
||
"""未连接时 publish 应返回 False 且不抛异常"""
|
||
client = MQTTForwardClient("mqtt://h:1883", "/t")
|
||
assert run(client.publish({"x": 1})) is False
|
||
|
||
|
||
def test_publish_uses_default_topic():
|
||
"""未指定 topic 时发布到构造时的默认主题"""
|
||
client = MQTTForwardClient("mqtt://h:1883", "/board_ping")
|
||
fake = attach(client)
|
||
assert run(client.publish({"BoardId": "B1"})) is True
|
||
topic, payload, _ = fake.published[0]
|
||
assert topic == "/board_ping"
|
||
assert json.loads(payload) == {"BoardId": "B1"}
|
||
|
||
|
||
def test_publish_specified_topic():
|
||
"""显式 topic 应覆盖默认主题"""
|
||
client = MQTTForwardClient("mqtt://h:1883", "/default")
|
||
fake = attach(client)
|
||
run(client.publish({"a": 1}, topic="/custom"))
|
||
assert fake.published[0][0] == "/custom"
|
||
|
||
|
||
def test_publish_exception_returns_false():
|
||
"""底层发布异常时应返回 False"""
|
||
client = MQTTForwardClient("mqtt://h:1883", "/t")
|
||
|
||
class BoomMQTT:
|
||
async def publish(self, topic, payload, qos):
|
||
raise RuntimeError("boom")
|
||
|
||
client._client = BoomMQTT() # type: ignore
|
||
client._running = True
|
||
assert run(client.publish({"a": 1})) is False
|
||
|
||
|
||
def test_stop_disconnects_and_clears_flag():
|
||
"""stop() 应断开连接并复位运行标志"""
|
||
client = MQTTForwardClient("mqtt://h:1883", "/t")
|
||
fake = attach(client)
|
||
run(client.stop())
|
||
assert fake.disconnected is True
|
||
assert client.is_running is False
|