From 6252625c0450eafeca80369b52f8f534dee674d3 Mon Sep 17 00:00:00 2001 From: Martin Kersner Date: Sun, 5 Jul 2026 19:18:43 +0900 Subject: [PATCH 1/4] test(ws): reconnect replays active subscription Local websockets-server test: drop the conn mid-stream, assert the client reconnects and replays sorted(active) SUBSCRIBE, then resumes. Backoff monkeypatched tiny for speed. Keyless, offline lane. --- tests/test_ws.py | 47 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) 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) From 672051167db40110330056d086e63d35afd7bc14 Mon Sep 17 00:00:00 2001 From: Martin Kersner Date: Sun, 5 Jul 2026 19:18:52 +0900 Subject: [PATCH 2/4] test(ws): keyed live lane for sporadic channels Opt-in integration lane for /liquidation and /announcement/listing. Tolerates quiet windows (TimeoutError = pass). Announcement is Pro+; skip on handshake/auth rejection so a Basic key doesn't red the lane. --- tests/test_ws_live.py | 74 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 74 insertions(+) create mode 100644 tests/test_ws_live.py diff --git a/tests/test_ws_live.py b/tests/test_ws_live.py new file mode 100644 index 0000000..c90bda2 --- /dev/null +++ b/tests/test_ws_live.py @@ -0,0 +1,74 @@ +"""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 at the handshake (InvalidStatus) or the server closes with an + # auth code (ConnectionClosed) — both subclass WebSocketException. Skip + # rather than fail so a Basic key doesn't red the lane; a Pro key exercises + # connect + auth + SUBSCRIBE for real. + async def run(): + async with AsyncDatamaxiWS(api_key=API_KEY, base_url=BASE_URL) 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 as exc: + pytest.skip(f"/announcement/listing rejected (needs Pro+ tier): {exc!r}") + if msg is not None: + assert isinstance(msg, dict) From 1dea95c9a5b921953ab9368ff8c92f783b7d7525 Mon Sep 17 00:00:00 2001 From: Martin Kersner Date: Sun, 5 Jul 2026 19:20:57 +0900 Subject: [PATCH 3/4] test(ws): reconnect=False on announcement test so auth-close skips Default reconnect=True swallows a post-connect auth close in the reader loop, so the skip-on-rejection path never fired (test just timed out and passed). Disable reconnect so the close propagates to the skip. --- tests/test_ws_live.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/test_ws_live.py b/tests/test_ws_live.py index c90bda2..dc2de6a 100644 --- a/tests/test_ws_live.py +++ b/tests/test_ws_live.py @@ -56,9 +56,13 @@ def test_ws_announcement_subscribe_tolerates_quiet_window(): # rejected at the handshake (InvalidStatus) or the server closes with an # auth code (ConnectionClosed) — both subclass WebSocketException. Skip # rather than fail so a Basic key doesn't red the lane; a Pro key exercises - # connect + auth + SUBSCRIBE for real. + # connect + auth + SUBSCRIBE for real. reconnect=False so a post-connect + # auth close propagates to the skip below instead of being swallowed by the + # reconnect loop (which would just hit the timeout and pass silently). async def run(): - async with AsyncDatamaxiWS(api_key=API_KEY, base_url=BASE_URL) as ws: + 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) From 7bb96a76b2ced584a80d22e6fe7c2a9e95c0f694 Mon Sep 17 00:00:00 2001 From: Martin Kersner Date: Sun, 5 Jul 2026 19:23:17 +0900 Subject: [PATCH 4/4] test(ws): also skip announcement on StopAsyncIteration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With reconnect=False a post-connect auth close ends the stream, so __anext__ raises StopAsyncIteration (not a WebSocketException) — the prior catch missed it and the test would error. Catch both so both rejection modes skip; a quiet Pro conn still times out and passes. --- tests/test_ws_live.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/tests/test_ws_live.py b/tests/test_ws_live.py index dc2de6a..d94c59b 100644 --- a/tests/test_ws_live.py +++ b/tests/test_ws_live.py @@ -53,12 +53,12 @@ async def run(): def test_ws_announcement_subscribe_tolerates_quiet_window(): # /announcement/listing is Pro+ and takes no param. A non-Pro key is - # rejected at the handshake (InvalidStatus) or the server closes with an - # auth code (ConnectionClosed) — both subclass WebSocketException. Skip - # rather than fail so a Basic key doesn't red the lane; a Pro key exercises - # connect + auth + SUBSCRIBE for real. reconnect=False so a post-connect - # auth close propagates to the skip below instead of being swallowed by the - # reconnect loop (which would just hit the timeout and pass silently). + # 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 @@ -72,7 +72,7 @@ async def run(): try: msg = _run(run()) - except websockets.exceptions.WebSocketException as exc: + 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)