"""Tests for BingxVenueAdapter.reconcile() — the async non-blocking path. Regression suite for the 30s event-loop deadlock caused by the old sync reconcile calling asyncio.run() in a thread pool while the aiohttp session was bound to the main event loop. Run with: python -m pytest prod/clean_arch/dita_v2/test_venue_reconcile.py -v """ from __future__ import annotations import asyncio import inspect import sys import time from typing import Any, List from unittest.mock import AsyncMock, MagicMock sys.path.insert(0, "/mnt/dolphinng5_predict") import pytest from prod.clean_arch.dita_v2.bingx_venue import BingxVenueAdapter from prod.clean_arch.dita_v2.contracts import ( KernelCommandType, KernelEventKind, VenueEvent, VenueEventStatus, ) # ── helpers ────────────────────────────────────────────────────────────────── def _flat_snapshot(**kwargs) -> MagicMock: """Build a minimal ExchangeStateSnapshot-like mock.""" snap = MagicMock() snap.open_orders = kwargs.get("open_orders", []) snap.all_orders = kwargs.get("all_orders", []) snap.all_fills = kwargs.get("all_fills", []) snap.open_positions = kwargs.get("open_positions", {}) snap.capital = kwargs.get("capital", 25000.0) return snap def _make_venue(snapshot=None, raises=None) -> BingxVenueAdapter: """Return a BingxVenueAdapter with a mock backend.""" backend = MagicMock() if raises is not None: backend.refresh_state = AsyncMock(side_effect=raises) else: backend.refresh_state = AsyncMock(return_value=snapshot or _flat_snapshot()) venue = BingxVenueAdapter.__new__(BingxVenueAdapter) venue.backend = backend venue._event_seq = iter(range(10_000)) venue._snap_lock = __import__("threading").Lock() venue._snapshot_ready = __import__("threading").Event() venue._snapshot_ready.set() venue._last_snapshot = None return venue # ── 1. Contract: reconcile() must be async ─────────────────────────────────── class TestReconcileIsAsync: def test_reconcile_returns_coroutine(self): """reconcile() must return a coroutine, not a plain list.""" venue = _make_venue() result = venue.reconcile() assert inspect.isawaitable(result), ( "reconcile() must be async so pump_venue_events can await it " "without blocking the event loop" ) asyncio.run(result) # clean up def test_pump_venue_events_isawaitable_path(self): """pump_venue_events uses inspect.isawaitable — verify it fires for our async reconcile (i.e. we didn't accidentally break the contract).""" venue = _make_venue() coro = venue.reconcile() assert inspect.isawaitable(coro) asyncio.run(coro) # ── 2. Correctness: returns events from snapshot ───────────────────────────── class TestReconcileEvents: def test_flat_exchange_returns_empty(self): """No open orders → no events.""" venue = _make_venue(_flat_snapshot()) events = asyncio.run(venue.reconcile()) assert events == [] def test_open_order_produces_event(self): """A single open (NEW) BingX order must produce an ORDER_ACK event.""" order_row = { "orderId": "V-001", "clientOrderId": "t1:i1", "symbol": "BTC-USDT", "side": "BUY", "status": "NEW", "origQty": "1.0", "executedQty": "0.0", "avgPrice": "0.0", } snap = _flat_snapshot(open_orders=[order_row]) venue = _make_venue(snap) events = asyncio.run(venue.reconcile()) assert len(events) == 1 assert events[0].kind == KernelEventKind.ORDER_ACK def test_filled_order_produces_fill_event(self): """A FILLED order must produce a FULL_FILL event.""" order_row = { "orderId": "V-002", "clientOrderId": "t1:i1", "symbol": "BTC-USDT", "side": "SELL", "status": "FILLED", "origQty": "1.0", "executedQty": "1.0", "avgPrice": "50000.0", } snap = _flat_snapshot(all_orders=[order_row]) venue = _make_venue(snap) events = asyncio.run(venue.reconcile()) assert any(e.kind == KernelEventKind.FULL_FILL for e in events) def test_deduplication_across_open_and_all_orders(self): """Same order in both open_orders and all_orders must appear once.""" row = { "orderId": "V-003", "clientOrderId": "t1:i1", "symbol": "BTC-USDT", "side": "BUY", "status": "NEW", "origQty": "1.0", "executedQty": "0.0", "avgPrice": "0.0", } snap = _flat_snapshot(open_orders=[row], all_orders=[row]) venue = _make_venue(snap) events = asyncio.run(venue.reconcile()) assert len(events) == 1, "Duplicate entry must be deduped" def test_backend_called_with_include_history_false(self): """reconcile() must call refresh_state with include_history=False. include_history=True with symbol=None is a no-op but wastes a round-trip building the param; False is explicit and cheaper.""" venue = _make_venue() asyncio.run(venue.reconcile()) venue.backend.refresh_state.assert_called_once_with(None, include_history=False) # ── 3. Fault tolerance: exception → empty list, not crash ─────────────────── class TestReconcileFaultTolerance: def test_backend_exception_returns_empty_list(self): """If refresh_state raises, reconcile must return [] not propagate.""" venue = _make_venue(raises=RuntimeError("BingX 503")) events = asyncio.run(venue.reconcile()) assert events == [], "Exception must be swallowed and return empty list" def test_timeout_error_returns_empty_list(self): venue = _make_venue(raises=TimeoutError("VST timeout")) events = asyncio.run(venue.reconcile()) assert events == [] def test_connection_error_returns_empty_list(self): venue = _make_venue(raises=ConnectionError("VST unreachable")) events = asyncio.run(venue.reconcile()) assert events == [] def test_n_consecutive_failures_never_raise(self): """10 consecutive backend failures must never propagate an exception.""" venue = _make_venue(raises=RuntimeError("persistent failure")) for _ in range(10): events = asyncio.run(venue.reconcile()) assert events == [] # ── 4. Non-blocking: reconcile must complete in << 30s ─────────────────────── class TestReconcileNonBlocking: def test_completes_well_under_timeout(self): """The async reconcile must complete in under 2s (backend is mocked as instant). The old sync path blocked for 30s minimum on the deadlock. This guards against any regression to blocking behaviour.""" venue = _make_venue() t0 = time.perf_counter() asyncio.run(venue.reconcile()) elapsed = time.perf_counter() - t0 assert elapsed < 2.0, ( f"reconcile() took {elapsed:.2f}s — blocking regression detected. " f"Must complete in < 2s (mocked backend = instant)." ) def test_exception_path_completes_quickly(self): """Even when backend raises, reconcile must return quickly.""" venue = _make_venue(raises=TimeoutError("30s deadlock simulation")) t0 = time.perf_counter() asyncio.run(venue.reconcile()) elapsed = time.perf_counter() - t0 assert elapsed < 2.0 def test_awaitable_in_asyncio_gather(self): """reconcile() must compose cleanly with asyncio.gather — critical for future parallelisation of reconcile + other venue calls.""" venues = [_make_venue() for _ in range(3)] async def _run(): results = await asyncio.gather(*[v.reconcile() for v in venues]) return results results = asyncio.run(_run()) assert len(results) == 3 assert all(isinstance(r, list) for r in results) def test_concurrent_reconciles_do_not_deadlock(self): """10 concurrent reconcile coroutines must all complete.""" async def _run(): venues = [_make_venue() for _ in range(10)] return await asyncio.gather(*[v.reconcile() for v in venues]) results = asyncio.run(_run()) assert len(results) == 10 # ── 5. Integration: pump_venue_events awaits async reconcile ───────────────── class TestPumpIntegration: """Verify the pump_venue_events → reconcile() async chain end-to-end.""" def test_pump_venue_events_awaits_async_reconcile(self): """pump_venue_events must detect isawaitable and await the reconcile coroutine — verifying the inspect.isawaitable() contract is met.""" import inspect as _inspect venue = _make_venue() coro = venue.reconcile() assert _inspect.isawaitable(coro), ( "pump_venue_events checks inspect.isawaitable(events); " "reconcile() must return a coroutine for the await path to fire" ) asyncio.run(coro) def test_reconcile_result_is_list_of_venue_events(self): """The awaited result must be a list (possibly empty) — the type pump_venue_events iterates over.""" venue = _make_venue() result = asyncio.run(venue.reconcile()) assert isinstance(result, list) for item in result: assert isinstance(item, VenueEvent)