diff --git a/README.md b/README.md
index cfc2df9..4223934 100644
--- a/README.md
+++ b/README.md
@@ -117,7 +117,10 @@ await client.stop()
 
 ## 与 aibox_sophon_adapter 插件的集成
 
-插件在 `[workers.config]` 下分别配置这三个客户端：
+插件在 `[workers.config]` 下分别配置这三个客户端，启动逻辑封装在插件的
+`mqtt_clients.py` 能力组件中（`start_mqtt_forward` / `start_mqtt_subscribe` /
+`start_internal_mqtt`，启动失败自动降级为 `None` 并告警），
+由 `AiboxSophonAdapterPlugin.on_init` 调用：
 
 ```toml
 [workers.config.mqtt_forward]       # -> MQTTForwardClient
@@ -159,6 +162,26 @@ async def _cmd_some_op(payload: dict) -> dict:
 
 ---
 
+## 测试
+
+单元测试位于 `utils/mqtt/test/`，通过注入 `FakeMQTT` 模拟底层连接，**不依赖真实 Broker**，可离线快速运行。
+
+```bash
+uv run pytest utils/mqtt/test -v
+```
+
+覆盖范围：
+
+| 测试文件            | 覆盖内容                                                                                           |
+| ------------------- | -------------------------------------------------------------------------------------------------- |
+| `test_forward.py`   | 凭据解码、默认/指定主题发布、未连接与异常时的行为、stop 断开                                       |
+| `test_subscribe.py` | 主题通配符（`+`/`#`）编译、无效路由跳过、消息分发、`reply_topic` 回复、`None` 不回复、`{1}` 占位符 |
+| `test_internal.py`  | 凭据解码、回显 `request_id` 关联、`{rid}` 模板、超时、FIFO 兜底、并发乱序、订阅去重                |
+
+> 若需对真实 Broker 做端到端验证，可参考项目根目录的 `test_mqtt_e2e.py` 模式，自行启动本地 broker 后联测。
+
+---
+
 ## 构建集成
 
 若通过 PyInstaller/Nuitka 打包，需在 `config.toml` 的 `[pyinstaller_build_import] base` 中加入：
diff --git a/test/conftest.py b/test/conftest.py
new file mode 100644
index 0000000..2201ce0
--- /dev/null
+++ b/test/conftest.py
@@ -0,0 +1,12 @@
+"""
+chongming_mqtt 测试公共配置
+
+确保 ``chongming_mqtt`` 可导入：优先使用已安装的包（``uv pip install -e utils/mqtt``），
+若未安装则回退到源码 ``utils/mqtt/src`` 目录。
+"""
+import sys
+from pathlib import Path
+
+_SRC = Path(__file__).resolve().parent.parent / "src"
+if str(_SRC) not in sys.path:
+    sys.path.insert(0, str(_SRC))
diff --git a/test/fakes.py b/test/fakes.py
new file mode 100644
index 0000000..a257a55
--- /dev/null
+++ b/test/fakes.py
@@ -0,0 +1,50 @@
+"""
+测试用模拟对象 — 不连接真实 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)
diff --git a/test/test_forward.py b/test/test_forward.py
new file mode 100644
index 0000000..8e7721d
--- /dev/null
+++ b/test/test_forward.py
@@ -0,0 +1,64 @@
+"""
+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
diff --git a/test/test_internal.py b/test/test_internal.py
new file mode 100644
index 0000000..60a6467
--- /dev/null
+++ b/test/test_internal.py
@@ -0,0 +1,132 @@
+"""
+InternalMQTTClient 单元测试
+
+覆盖：URI 凭据解码、请求-响应关联（回显 request_id / {rid} 模板 / 兜底）、
+超时、并发乱序、订阅去重。
+"""
+import asyncio
+import json
+
+from chongming_mqtt.internal_mqtt import InternalMQTTClient
+
+from fakes import attach, run
+
+
+def test_uri_decoded():
+    """URI 中 percent-encoded 凭据应被解码"""
+    c = InternalMQTTClient("mqtt://u:p%40w@host:1883")
+    assert c._username == "u"
+    assert c._password == "p@w"
+    assert c._clean_uri == "mqtt://host:1883"
+
+
+def test_request_not_running_returns_none():
+    """未连接时 request 返回 None"""
+    c = InternalMQTTClient("mqtt://127.0.0.1:1883")
+    assert run(c.request("/req", "/resp", {})) is None
+
+
+def test_request_echo_rid():
+    """对方回显 request_id → 精确匹配响应"""
+    client = InternalMQTTClient("mqtt://127.0.0.1:1883")
+
+    async def respond(topic, payload):
+        rid = json.loads(payload)["request_id"]
+        await client._handle("/resp", json.dumps({"request_id": rid, "ok": True}).encode())
+
+    attach(client, on_publish=respond)
+    r = run(client.request("/req", "/resp", {"a": 1}, timeout=3))
+    assert r is not None
+    assert r.get("ok") is True
+    assert "request_id" in r
+
+
+def test_request_rid_template():
+    """response_topic 含 {rid} → 订阅唯一主题并按主题匹配"""
+    client = InternalMQTTClient("mqtt://127.0.0.1:1883")
+
+    async def respond(topic, payload):
+        rid = json.loads(payload)["request_id"]
+        await client._handle(f"/resp/{rid}", json.dumps({"request_id": rid, "tpl": True}).encode())
+
+    fake = attach(client, on_publish=respond)
+    r = run(client.request("/req", "/resp/{rid}", {}, timeout=3))
+    assert r is not None
+    assert r.get("tpl") is True
+    # {rid} 模板：订阅的是唯一主题（含实际 rid），而非原模板
+    assert any("/resp/" in t for t, _ in fake.subscribed)
+
+
+def test_request_timeout_returns_none():
+    """对方不回复 → 超时返回 None"""
+    class NoReply:
+        async def publish(self, topic, payload, qos):
+            pass
+
+        async def subscribe(self, topics):
+            pass
+
+    client = InternalMQTTClient("mqtt://127.0.0.1:1883")
+    client._client = NoReply()
+    client._running = True
+    assert run(client.request("/req", "/resp", {}, timeout=0.2)) is None
+
+
+def test_request_fifo_fallback():
+    """对方不回显 request_id（固定主题）→ 兜底 FIFO 取最早挂起请求"""
+    client = InternalMQTTClient("mqtt://127.0.0.1:1883")
+
+    async def respond(topic, payload):
+        await client._handle("/resp", json.dumps({"v": "first"}).encode())
+
+    attach(client, on_publish=respond)
+    r = run(client.request("/req", "/resp", {}, timeout=3))
+    assert r == {"v": "first"}
+
+
+def test_request_concurrent_out_of_order():
+    """并发请求 + 响应乱序 → 每个请求按 request_id 拿回自己的响应"""
+    client = InternalMQTTClient("mqtt://127.0.0.1:1883")
+    attach(client)
+
+    async def do_req(tag):
+        rid = f"r-{tag}"
+        fut = asyncio.get_running_loop().create_future()
+        client._pending[rid] = fut
+        if tag == "x":
+            await asyncio.sleep(0.05)
+            # 模拟响应乱序：先回 y，再回 x
+            await client._handle("/resp", json.dumps({"request_id": "r-y", "v": "y"}).encode())
+            await client._handle("/resp", json.dumps({"request_id": "r-x", "v": "x"}).encode())
+        return await asyncio.wait_for(fut, 2), rid
+
+    async def main():
+        t1 = asyncio.create_task(do_req("x"))
+        t2 = asyncio.create_task(do_req("y"))
+        return await asyncio.gather(t1, t2)
+
+    results = run(main())
+    by_rid = {rid: res for res, rid in results}
+    assert by_rid["r-x"]["v"] == "x"
+    assert by_rid["r-y"]["v"] == "y"
+
+
+def test_subscribe_once_dedup():
+    """同一响应主题重复订阅应去重"""
+    class SpySub:
+        def __init__(self):
+            self.topics = []
+
+        async def publish(self, topic, payload, qos):
+            pass
+
+        async def subscribe(self, topics):
+            self.topics.extend(t for t, _ in topics)
+
+    spy = SpySub()
+    client = InternalMQTTClient("mqtt://127.0.0.1:1883")
+    client._client = spy
+    client._running = True
+    run(client._subscribe_once("/resp"))
+    run(client._subscribe_once("/resp"))
+    assert spy.topics == ["/resp"]
diff --git a/test/test_subscribe.py b/test/test_subscribe.py
new file mode 100644
index 0000000..dfb8b8c
--- /dev/null
+++ b/test/test_subscribe.py
@@ -0,0 +1,124 @@
+"""
+MQTTSubscribeClient 单元测试
+
+覆盖：主题通配符编译、路由构建、消息分发、请求-响应（reply_topic）与不回复约定。
+"""
+import json
+
+from chongming_mqtt.mqtt_subscribe import MQTTSubscribeClient, _compile_topic
+
+from fakes import attach, run
+
+
+# ---------- _compile_topic 通配符匹配 ----------
+
+def test_compile_topic_exact():
+    p = _compile_topic("/edge_app_controller")
+    assert p.match("/edge_app_controller")
+    assert not p.match("/edge_app_controller/other")
+    assert not p.match("/other")
+
+
+def test_compile_topic_plus_wildcard():
+    """+ 只匹配单级"""
+    p = _compile_topic("/dev/+/cmd")
+    assert p.match("/dev/box1/cmd")
+    assert not p.match("/dev/box1/sub/cmd")
+
+
+def test_compile_topic_hash_wildcard():
+    """# 匹配多级且须在末尾"""
+    p = _compile_topic("/dev/#")
+    assert p.match("/dev/a/b/c")
+    assert p.match("/dev/x")
+    assert not p.match("/other/a")
+
+
+# ---------- 路由构建 ----------
+
+def test_routes_skip_invalid():
+    """无效（空 topic / 无 handler）订阅项应被跳过"""
+    client = MQTTSubscribeClient("mqtt://h:1883", routes=[
+        {"topic": "", "handler": lambda d, t: None},
+        {"topic": "/t", "handler": lambda d, t: None},
+        {"topic": "/t2"},  # 无 handler
+    ])
+    assert len(client._routes) == 1
+
+
+# ---------- 消息分发与回复 ----------
+
+async def _echo_handler(payload, topic):
+    return {"echo": payload.get("v")}
+
+
+def test_dispatch_invokes_handler_and_replies():
+    """收到消息 → 调用 handler → 结果发布到 reply_topic"""
+    client = MQTTSubscribeClient("mqtt://h:1883", routes=[
+        {"topic": "/req", "handler": _echo_handler, "reply_topic": "/resp"},
+    ])
+    fake = attach(client)
+    run(client._dispatch("/req", b'{"v": 7}'))
+    assert len(fake.published) == 1
+    topic, payload, _ = fake.published[0]
+    assert topic == "/resp"
+    assert json.loads(payload) == {"echo": 7}
+
+
+def test_dispatch_no_reply_when_handler_returns_none():
+    """handler 返回 None 时不发布回复（静默命令）"""
+    async def silent(payload, topic):
+        return None
+
+    client = MQTTSubscribeClient("mqtt://h:1883", routes=[
+        {"topic": "/req", "handler": silent, "reply_topic": "/resp"},
+    ])
+    fake = attach(client)
+    run(client._dispatch("/req", b'{"v": 1}'))
+    assert fake.published == []
+
+
+def test_dispatch_no_reply_without_reply_topic():
+    """未配置 reply_topic 时不发布"""
+    client = MQTTSubscribeClient("mqtt://h:1883", routes=[
+        {"topic": "/req", "handler": _echo_handler},
+    ])
+    fake = attach(client)
+    run(client._dispatch("/req", b'{"v": 1}'))
+    assert fake.published == []
+
+
+def test_dispatch_ignores_non_json():
+    """非 JSON 消息被忽略，不调用 handler"""
+    calls = []
+
+    async def h(payload, topic):
+        calls.append(payload)
+
+    client = MQTTSubscribeClient("mqtt://h:1883", routes=[
+        {"topic": "/req", "handler": h, "reply_topic": "/resp"},
+    ])
+    fake = attach(client)
+    run(client._dispatch("/req", b"not-json"))
+    assert calls == []
+    assert fake.published == []
+
+
+def test_dispatch_unmatched_topic_ignored():
+    """不匹配任何路由主题的消息被忽略"""
+    client = MQTTSubscribeClient("mqtt://h:1883", routes=[
+        {"topic": "/req", "handler": _echo_handler, "reply_topic": "/resp"},
+    ])
+    fake = attach(client)
+    run(client._dispatch("/other/topic", b'{"v": 1}'))
+    assert fake.published == []
+
+
+def test_reply_topic_placeholder_fill():
+    """reply_topic 的 {1} 占位符填充订阅主题通配符捕获值"""
+    client = MQTTSubscribeClient("mqtt://h:1883", routes=[
+        {"topic": "/dev/+/cmd", "handler": _echo_handler, "reply_topic": "/resp/{1}"},
+    ])
+    fake = attach(client)
+    run(client._dispatch("/dev/box9/cmd", b'{"v": 1}'))
+    assert fake.published[0][0] == "/resp/box9"