diff --git a/tests/test_ws.py b/tests/test_ws.py index 61de1e3..da450a4 100644 --- a/tests/test_ws.py +++ b/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) diff --git a/tests/test_ws_live.py b/tests/test_ws_live.py new file mode 100644 index 0000000..d94c59b --- /dev/null +++ b/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)