feat(api): WS 连接池 ConnectionManager

tests/api/test_ws.py

@@ -0,0 +1,30 @@
+# tests/api/test_ws.py
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+
+
+@pytest.mark.asyncio
+async def test_connect_and_broadcast():
+    from sanguo_api.ws import ConnectionManager
+    mgr = ConnectionManager()
+    ws = AsyncMock()
+    mgr.connect("t1", ws)
+    assert "t1" in mgr._connections
+    await mgr.broadcast("t1", {"status": "running"})
+    ws.send_json.assert_called_with({"status": "running"})
+
+
+def test_disconnect_removes_ws():
+    from sanguo_api.ws import ConnectionManager
+    mgr = ConnectionManager()
+    ws = MagicMock()
+    mgr.connect("t1", ws)
+    mgr.disconnect("t1", ws)
+    assert ws not in mgr._connections.get("t1", set())
+
+
+@pytest.mark.asyncio
+async def test_broadcast_no_subscribers_no_error():
+    from sanguo_api.ws import ConnectionManager
+    mgr = ConnectionManager()
+    await mgr.broadcast("nope", {"x": 1})  # 不抛