31 lines
870 B
Python
31 lines
870 B
Python
# 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}) # 不抛
|