161 lines
5.3 KiB
Python
161 lines
5.3 KiB
Python
"""
|
||
RouteRegistry 单元测试（loopback Zenoh 会话 + httpx ASGI 功能断言）
|
||
|
||
验证 process 模式核心链路：
|
||
RoutePublisher.put 描述符 → 网关 RouteRegistry 动态挂载 FastAPI 路由（命中返回 503，worker 桩无实例）
|
||
RoutePublisher.delete → 动态摘除路由（404）
|
||
__meta__ 中的 mqtt_topics → 推送给 MQTT Bridge（update_custom_topics）
|
||
|
||
注：FastAPI 0.139 的 include_router 将路由包装为 _IncludedRouter（惰性挂载），
|
||
不能靠遍历 app.router.routes 的 path 判断，故用 ASGI 请求状态码做功能断言。
|
||
"""
|
||
import asyncio
|
||
import time
|
||
|
||
import httpx
|
||
import zenoh
|
||
from fastapi import FastAPI
|
||
|
||
from src.chongming.app.core.route_registry import RouteRegistry
|
||
from src.chongming.app.core.route_publisher import RoutePublisher
|
||
|
||
failures = []
|
||
|
||
|
||
def check(name, cond, detail=""):
|
||
status = "PASS" if cond else "FAIL"
|
||
print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
|
||
if not cond:
|
||
failures.append(name)
|
||
|
||
|
||
class FakeWorkerManager:
|
||
"""桩：无可用 worker，命中路由后 call 抛 RuntimeError（网关转 503）。"""
|
||
|
||
async def call(self, **kwargs):
|
||
raise RuntimeError("无可用 worker（测试桩）")
|
||
|
||
|
||
class FakeBridge:
|
||
"""记录 update_custom_topics 的调用结果。"""
|
||
|
||
def __init__(self):
|
||
self.topics = None
|
||
|
||
async def update_custom_topics(self, topics):
|
||
self.topics = list(topics)
|
||
|
||
|
||
async def wait_until(cond, timeout=10.0, interval=0.2) -> bool:
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
if cond():
|
||
return True
|
||
await asyncio.sleep(interval)
|
||
return False
|
||
|
||
|
||
HTTP_DESCRIPTOR = {
|
||
"subject": "example.add",
|
||
"method": "GET",
|
||
"path": "/calc/add",
|
||
"params": ["a: float", "b: float"],
|
||
"response_model": {"result": {"type": "float", "required": True}},
|
||
"summary": "加法运算",
|
||
"timeout": 30.0,
|
||
"cacheable": False,
|
||
"auth_required": False,
|
||
"internal": False,
|
||
}
|
||
|
||
MQTT_DESCRIPTOR = {
|
||
"subject": "example.add",
|
||
"method": "MQTT",
|
||
"path": "/calc/add_mqtt",
|
||
"params": ["a: float", "b: float"],
|
||
"summary": "MQTT 加法",
|
||
"timeout": 30.0,
|
||
"cacheable": False,
|
||
"auth_required": False,
|
||
"internal": False,
|
||
}
|
||
|
||
MQTT_TOPICS = [{"topic": "box/aibox_sophon/board_ping"}]
|
||
# registry 会为每个主题补 worker 字段
|
||
MQTT_TOPICS_EXPECTED = [
|
||
{"topic": "box/aibox_sophon/board_ping", "worker": "example"}
|
||
]
|
||
|
||
|
||
async def main():
|
||
zconfig = zenoh.Config()
|
||
try:
|
||
zconfig.from_file("config/zenoh_config.json5")
|
||
except Exception:
|
||
pass # 回退默认配置
|
||
|
||
gw_session = zenoh.open(zconfig)
|
||
pub_session = zenoh.open(zconfig)
|
||
try:
|
||
app = FastAPI()
|
||
registry = RouteRegistry(app, FakeWorkerManager(), prefix="/api/v1")
|
||
await registry.start(gw_session)
|
||
await asyncio.sleep(0.5) # 等待订阅生效
|
||
|
||
transport = httpx.ASGITransport(app=app)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
# 1. 发布 → HTTP 路由动态挂载（命中 503：路由存在但无 worker）
|
||
publisher = RoutePublisher(pub_session, "example")
|
||
publisher.publish_routes(
|
||
[HTTP_DESCRIPTOR, MQTT_DESCRIPTOR], mqtt_topics=MQTT_TOPICS
|
||
)
|
||
|
||
async def mounted():
|
||
resp = await client.get("/api/v1/calc/add?a=1&b=2")
|
||
return resp.status_code == 503
|
||
|
||
ok = await wait_until(mounted)
|
||
check("put 后 HTTP 路由动态挂载（503 命中）", ok)
|
||
|
||
resp = await client.get("/api/v1/calc/add_mqtt?a=1&b=2")
|
||
check("MQTT 端点不挂载 HTTP 路由（404）", resp.status_code == 404,
|
||
f"status={resp.status_code}")
|
||
|
||
# 2. __meta__ 主题推送给 Bridge
|
||
bridge = FakeBridge()
|
||
registry.attach_bridge(bridge)
|
||
ok = await wait_until(lambda: bridge.topics == MQTT_TOPICS_EXPECTED)
|
||
check("mqtt_topics 推送给 Bridge", ok, f"bridge.topics={bridge.topics}")
|
||
|
||
# 3. 重复发布相同描述符 → 幂等（路由不重复、仍可命中）
|
||
publisher.publish_routes([HTTP_DESCRIPTOR])
|
||
await asyncio.sleep(3.5) # 覆盖防抖窗口
|
||
resp = await client.get("/api/v1/calc/add?a=1&b=2")
|
||
check("重复发布幂等（仍正常命中）", resp.status_code == 503,
|
||
f"status={resp.status_code}")
|
||
|
||
# 4. 摘除 → 路由动态移除（404）
|
||
publisher.unpublish()
|
||
|
||
async def unmounted():
|
||
resp = await client.get("/api/v1/calc/add?a=1&b=2")
|
||
return resp.status_code == 404
|
||
|
||
ok = await wait_until(unmounted)
|
||
check("delete 后 HTTP 路由动态摘除（404）", ok)
|
||
ok = await wait_until(lambda: bridge.topics == [])
|
||
check("元信息摘除后 Bridge 主题清空", ok, f"bridge.topics={bridge.topics}")
|
||
finally:
|
||
registry.stop()
|
||
gw_session.close()
|
||
pub_session.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|
||
print("\n" + ("=" * 40))
|
||
if failures:
|
||
print(f"❌ {len(failures)} 项未通过: {failures}")
|
||
raise SystemExit(1)
|
||
print("✅ RouteRegistry 全部通过")
|