theo-agent-dashboard

Unnamed repository; edit this file 'description' to name the repository.
Log | Files | Refs | README

test_ws_manager.py (2899B)

      1 """T022 — Tests for WebSocket connection manager."""
      2 
      3 import asyncio
      4 
      5 import pytest
      6 from app.ws_manager import ConnectionManager
      7 
      8 
      9 # ── helpers ──────────────────────────────────────────────────────────────
     10 
     11 
     12 class FakeWebSocket:
     13     """Minimal async stub that tracks sent JSON."""
     14 
     15     def __init__(self, *, fail: bool = False) -> None:
     16         self.sent: list[dict] = []
     17         self.fail = fail
     18 
     19     async def send_json(self, data: dict) -> None:  # noqa: ANN401
     20         if self.fail:
     21             raise RuntimeError("connection closed")
     22         self.sent.append(data)
     23 
     24     async def close(self) -> None:  # noqa: ANN401
     25         pass
     26 
     27 
     28 # ── tests ────────────────────────────────────────────────────────────────
     29 
     30 
     31 @pytest.mark.asyncio
     32 async def test_connect_stores_connection() -> None:
     33     mgr = ConnectionManager()
     34     ws = FakeWebSocket()
     35     await mgr.connect("s1", ws)
     36     assert mgr.get_connection("s1") is ws
     37 
     38 
     39 @pytest.mark.asyncio
     40 async def test_disconnect_removes_connection() -> None:
     41     mgr = ConnectionManager()
     42     ws = FakeWebSocket()
     43     await mgr.connect("s1", ws)
     44     mgr.disconnect("s1")
     45     assert mgr.get_connection("s1") is None
     46 
     47 
     48 @pytest.mark.asyncio
     49 async def test_send_json_delivers_to_connection() -> None:
     50     mgr = ConnectionManager()
     51     ws = FakeWebSocket()
     52     await mgr.connect("s1", ws)
     53     await mgr.send_json("s1", {"type": "event", "event": "run.started"})
     54     assert ws.sent == [{"type": "event", "event": "run.started"}]
     55 
     56 
     57 @pytest.mark.asyncio
     58 async def test_send_json_removes_dead_connection() -> None:
     59     mgr = ConnectionManager()
     60     ws = FakeWebSocket(fail=True)
     61     await mgr.connect("s1", ws)
     62     await mgr.send_json("s1", {"type": "event"})
     63     assert mgr.get_connection("s1") is None
     64 
     65 
     66 @pytest.mark.asyncio
     67 async def test_get_connection_returns_none_for_unknown() -> None:
     68     mgr = ConnectionManager()
     69     assert mgr.get_connection("nonexistent") is None
     70 
     71 class TestWsManagerEdgeCases:
     72     """Edge cases for WebSocket manager."""
     73 
     74     async def test_send_json_skips_missing_connection(self) -> None:
     75         from app.ws_manager import ConnectionManager
     76         mgr = ConnectionManager()
     77         await mgr.send_json("nonexistent", {"type": "test"})
     78 
     79     def test_disconnect_nonexistent_is_noop(self) -> None:
     80         from app.ws_manager import ConnectionManager
     81         mgr = ConnectionManager()
     82         result = mgr.disconnect("nonexistent")
     83         assert result is None
     84 
     85     async def test_get_connection_returns_none_for_unknown(self) -> None:
     86         from app.ws_manager import ConnectionManager
     87         mgr = ConnectionManager()
     88         assert mgr.get_connection("unknown") is None