"""Tests for WsBridge reconnect behaviour.
Uses a fake WebSocket client (no network) to drive the supervisor: a socket
delivers a few frames then drops, and the bridge must back off and reconnect,
continuing to forward events and not lose the user's queued input.
"""
from __future__ import annotations
import asyncio
import pytest
from clients.terminal.tui.events import ConnectionStatusChanged, WsEvent
from clients.terminal.tui.ws_bridge import WsBridge
class _FakeApp:
"""Records every posted message — WsBridge only needs ``post_message``."""
def __init__(self) -> None:
self.messages: list = []
def post_message(self, message) -> None:
self.messages.append(message)
class _FakeWs:
"""Async iterator standing in for a real socket.
``frames`` are delivered one per ``__anext__``; when they run out the
socket ``close``s (the supervisor treats that as a drop and reconnects).
``idle=True`` makes the socket stay open with no incoming data — it blocks
until ``close()`` — so a connect can hold ``connected`` True long enough
for the input loop to send.
"""
def __init__(self, frames: list[str] | None = None, *, idle: bool = False, send_fails: bool = False) -> None:
self._frames = list(frames or [])
self.idle = idle
self.send_fails = send_fails
self.sent: list[str] = []
self._closed = asyncio.Event()
def __aiter__(self):
return self
async def __anext__(self):
if self.idle:
# Open socket, no incoming data: block until closed.
await self._closed.wait()
raise StopAsyncIteration
if not self._frames:
raise StopAsyncIteration
return self._frames.pop(0)
async def close(self) -> None:
self._closed.set()
class _FakeClient:
"""Stand-in for NaviWebSocketClient: scripted connect() results, records
sends, exposes the same _input_queue / stop_input surface the bridge uses."""
def __init__(self) -> None:
self._ws: _FakeWs | None = None
self._input_queue: asyncio.Queue = asyncio.Queue()
self._ws_sequence: list[_FakeWs] = []
self.connect_calls = 0
self.sent: list[str] = []
def push_ws(self, ws: _FakeWs) -> None:
self._ws_sequence.append(ws)
async def connect(self) -> None:
self.connect_calls += 1
self._ws = self._ws_sequence.pop(0) if self._ws_sequence else _FakeWs([])
async def close(self) -> None:
if self._ws is not None:
await self._ws.close()
self._ws = None
async def send(self, content: str) -> None:
if self._ws is None:
raise RuntimeError("not connected")
if self._ws.send_fails:
raise RuntimeError("socket dead")
self._ws.sent.append(content)
self.sent.append(content)
def stop_input(self) -> None:
self._input_queue.put_nowait(None)
def enqueue(self, content: str) -> None:
self._input_queue.put_nowait(content)
def _bridge(client: _FakeClient) -> WsBridge:
"""A bridge with a tiny backoff so reconnects happen within the test."""
bridge = WsBridge(_FakeApp(), "sess", client=client)
bridge._backoff_min = 0.01
bridge._backoff_max = 0.05
return bridge
def _ws_payloads(bridge: WsBridge) -> list[dict]:
return [m.payload for m in bridge.app.messages if isinstance(m, WsEvent)]
def _status_details(bridge: WsBridge) -> list[str]:
return [m.detail for m in bridge.app.messages if isinstance(m, ConnectionStatusChanged)]
@pytest.mark.anyio
async def test_reconnects_after_drop_and_continues_forwarding_events() -> None:
"""A socket delivers two frames then closes; the bridge backs off,
reconnects, and frames from the second socket still reach the app."""
client = _FakeClient()
client.push_ws(_FakeWs(['{"type":"status","content":"a"}', '{"type":"status","content":"b"}']))
client.push_ws(_FakeWs(['{"type":"status","content":"c"}']))
bridge = _bridge(client)
await bridge.start()
# connect1 → a, b → drop → backoff → connect2 → c → drop. 50ms is enough
# for two tiny backoffs.
await asyncio.sleep(0.05)
await bridge.stop()
assert client.connect_calls >= 2, "supervisor must reconnect after a drop"
payloads = [p.get("content") for p in _ws_payloads(bridge)]
assert "a" in payloads and "b" in payloads and "c" in payloads
@pytest.mark.anyio
async def test_initial_connect_failure_keeps_retrying_until_success() -> None:
"""If the server is down at startup, the supervisor keeps retrying and
eventually connects when the server comes back."""
client = _FakeClient()
bridge = _bridge(client)
# No ws pushed yet → connect() succeeds but yields an empty (immediately
# closing) socket, which also drops. To simulate a refused connection,
# make connect() fail for the first two attempts.
original_connect = client.connect
calls = {"n": 0}
async def flaky_connect():
calls["n"] += 1
if calls["n"] <= 2:
raise ConnectionRefusedError("server down")
await original_connect()
client.connect = flaky_connect
client.push_ws(_FakeWs(['{"type":"status","content":"recovered"}']))
await bridge.start()
await asyncio.sleep(0.08)
await bridge.stop()
assert calls["n"] >= 3, "must keep retrying past the initial refusals"
assert "recovered" in [p.get("content") for p in _ws_payloads(bridge)]
# The status panel saw the offline→online transition.
details = _status_details(bridge)
assert any("reconnecting" in d for d in details)
assert any(d == "online" for d in details)
@pytest.mark.anyio
async def test_queued_input_is_held_across_disconnect_then_sent_on_reconnect() -> None:
"""A message enqueued while disconnected is not lost — it's sent once the
next socket is live."""
client = _FakeClient()
# ws1 closes immediately (no frames) → drop; ws2 stays open and accepts sends.
client.push_ws(_FakeWs([]))
client.push_ws(_FakeWs(idle=True))
bridge = _bridge(client)
# Long first backoff so the disconnected window is reliably hit.
bridge._backoff_min = 0.2
bridge._backoff_max = 0.2
await bridge.start()
await asyncio.sleep(0.02) # ws1 connected then dropped; now in the 200ms backoff
assert bridge.connected is False # genuinely disconnected
client.enqueue("hello from the dark")
await asyncio.sleep(0.25) # backoff done → ws2 connects → input loop sends
await bridge.stop()
assert "hello from the dark" in client.sent
@pytest.mark.anyio
async def test_send_failure_requeues_message_and_resent_after_reconnect() -> None:
"""If send() fails mid-flight (socket died right after connect), the
message is requeued and delivered after the next reconnect."""
client = _FakeClient()
# ws1: open but send always fails; ws2: open and accepts sends.
client.push_ws(_FakeWs(idle=True, send_fails=True))
client.push_ws(_FakeWs(idle=True))
bridge = _bridge(client)
await bridge.start()
await asyncio.sleep(0.02) # ws1 connected (send will fail)
client.enqueue("survive the blip")
await asyncio.sleep(0.06) # send fails → requeue → close → reconnect → resend
await bridge.stop()
assert "survive the blip" in client.sent
@pytest.mark.anyio
async def test_stop_cancels_backoff_promptly() -> None:
"""Stopping during a backoff sleep ends the supervisor quickly — no hanging
on a long reconnect delay."""
client = _FakeClient()
bridge = _bridge(client)
bridge._backoff_min = 10.0 # long backoff
bridge._backoff_max = 10.0
await bridge.start()
await asyncio.sleep(0.01) # first connect attempt (no ws → empty socket → drop → long backoff)
loop = asyncio.get_event_loop()
t0 = loop.time()
await asyncio.wait_for(bridge.stop(), timeout=2.0)
elapsed = loop.time() - t0
assert elapsed < 2.0, "stop() must cancel the backoff, not wait for it"
@pytest.mark.anyio
async def test_stop_is_idempotent() -> None:
client = _FakeClient()
bridge = _bridge(client)
await bridge.start()
await asyncio.sleep(0.01)
await bridge.stop()
await bridge.stop() # must not raise
assert bridge.connected is False
@pytest.mark.anyio
async def test_supervisor_does_not_reconnect_after_stop() -> None:
"""Once stopped, a dropped socket must not trigger further connects."""
client = _FakeClient()
client.push_ws(_FakeWs([])) # drops immediately
bridge = _bridge(client)
await bridge.start()
await asyncio.sleep(0.02)
await bridge.stop()
calls_after_stop = client.connect_calls
await asyncio.sleep(0.05)
assert client.connect_calls == calls_after_stop, "stop() must end reconnect attempts"
