test(ws): reconnect-resubscribe + sporadic-channel live lane (Closes #169)

tests/test_ws.py

@@ -234,3 +234,50 @@ def test_ws_repr_no_key_leak():
     assert ws.ws_url == "wss://api.datamaxiplus.com"
     assert "secret" not in repr(ws)
     assert "has_key=True" in repr(ws)
+
+
+def test_ws_reconnect_replays_active_subscription(monkeypatch):
+    # Drop the connection mid-stream and assert the client reconnects and
+    # replays its active SUBSCRIBE, then resumes streaming. `_RECONNECT_BACKOFF`
+    # is a module global read at call time in `_reader`, so shrink it to keep
+    # the test fast.
+    monkeypatch.setattr("datamaxi.aio.ws._RECONNECT_BACKOFF", 0.01)
+
+    seen = []  # SUBSCRIBE params observed, one entry per connection
+
+    async def handler(conn):
+        idx = len(seen)
+        seen.append(None)
+        async for raw in conn:
+            m = json.loads(raw)
+            if m.get("method") == "SUBSCRIBE":
+                seen[idx] = m["params"]
+                await conn.send(json.dumps({"result": m["params"], "id": m["id"]}))
+                if idx == 0:
+                    # first connection: stream one message then drop it
+                    await conn.send(json.dumps({"s": "USD-KRW", "d": 1, "r": 1.0}))
+                    return  # returning closes the conn -> client reconnects
+                # reconnect: send a distinguishable message, keep the conn open
+                await conn.send(json.dumps({"s": "USD-KRW", "d": 2, "r": 2.0}))
+
+    async def run():
+        async with _serve(handler) as server:
+            async with AsyncDatamaxiWS(
+                api_key="k",
+                ws_url=f"ws://localhost:{_port(server)}",
+                keepalive=0,  # no PING noise
+            ) as ws:
+                stream = await ws.forex.subscribe("USD-KRW")
+                first = await _first(stream)
+                # reading the second message forces the reconnect path to
+                # complete (resubscribe + resume)
+                second = await _first(stream)
+                return first, second
+
+    first, second = _run(run())
+    assert first == {"s": "USD-KRW", "d": 1, "r": 1.0}
+    assert second == {"s": "USD-KRW", "d": 2, "r": 2.0}  # post-reconnect payload
+    # the server saw two connections and the reconnect replayed the active params
+    assert len(seen) >= 2
+    assert seen[0] == ["USD-KRW"]
+    assert seen[1] == ["USD-KRW"]  # `_open` replays sorted(self._active)

tests/test_ws_live.py

@@ -0,0 +1,78 @@
+"""Keyed live/integration lane for sporadic WS channels.
+
+The offline `test_ws.py` covers the protocol against a local server. These
+tests hit the real DataMaxi+ WS API for the two event-driven channels that
+only got ack-level live checks — `/liquidation` (subscribe-only) and
+`/announcement/listing` (Pro+). Both are sporadic, so a quiet window (no event
+in the timeout) is a PASS: the value is that connect + auth + SUBSCRIBE succeed
+without raising. Opt-in and deselected from the keyless CI lane via the
+`integration` marker; needs DATAMAXI_API_KEY. Skipped when `websockets` absent.
+"""
+
+import asyncio
+
+import pytest
+
+websockets = pytest.importorskip("websockets")
+
+from datamaxi.aio.ws import AsyncDatamaxiWS  # noqa: E402
+from tests.conftest import API_KEY, BASE_URL  # noqa: E402
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.skipif(
+        not API_KEY,
+        reason="API key not provided. Set DATAMAXI_API_KEY environment variable.",
+    ),
+]
+
+# Sporadic channels may be silent for long stretches; a quiet window is a pass.
+_QUIET_TIMEOUT = 10.0
+
+
+def _run(coro):
+    return asyncio.run(coro)
+
+
+def test_ws_liquidation_subscribe_tolerates_quiet_window():
+    # /liquidation is subscribe-only and event-driven (basic tier). Param format
+    # is "SYMBOL@exchange" per the generated registry.
+    async def run():
+        async with AsyncDatamaxiWS(api_key=API_KEY, base_url=BASE_URL) as ws:
+            stream = await ws.liquidation.subscribe("BTC-USDT@binance")
+            try:
+                return await asyncio.wait_for(stream.__anext__(), _QUIET_TIMEOUT)
+            except asyncio.TimeoutError:
+                # quiet window: connect + auth + SUBSCRIBE ok, just no event
+                return None
+
+    msg = _run(run())
+    if msg is not None:
+        assert isinstance(msg, dict)
+
+
+def test_ws_announcement_subscribe_tolerates_quiet_window():
+    # /announcement/listing is Pro+ and takes no param. A non-Pro key is
+    # rejected either at the handshake (InvalidStatus, a WebSocketException) or
+    # by a close right after connect. reconnect=False makes that post-connect
+    # close deterministic: the reader stops instead of reconnect-looping, so the
+    # stream ends and __anext__ raises StopAsyncIteration. Skip on either so a
+    # Basic key doesn't red the lane; a Pro key streams (or hits the quiet-window
+    # timeout) for real. A quiet Pro connection stays open -> TimeoutError -> pass.
+    async def run():
+        async with AsyncDatamaxiWS(
+            api_key=API_KEY, base_url=BASE_URL, reconnect=False
+        ) as ws:
+            stream = await ws.announcement.subscribe()
+            try:
+                return await asyncio.wait_for(stream.__anext__(), _QUIET_TIMEOUT)
+            except asyncio.TimeoutError:
+                # quiet window: subscribed fine, just no listing event
+                return None
+
+    try:
+        msg = _run(run())
+    except (websockets.exceptions.WebSocketException, StopAsyncIteration) as exc:
+        pytest.skip(f"/announcement/listing rejected (needs Pro+ tier): {exc!r}")
+    if msg is not None:
+        assert isinstance(msg, dict)