piker/tests/test_binance.py

'''
Binance websocket routing and feed-startup regressions.

'''
import pytest
import trio

from piker.brokers._util import DataUnavailable
from piker.brokers.binance.feed import (
    get_quote_routes,
    stream_messages,
    wait_for_live_quotes,
)
from piker.brokers.binance.venues import (
    _futes_market_ws,
    _futes_public_ws,
    _spot_ws,
)
from piker.data import _web_bs


def test_usdtm_websocket_routes() -> None:
    '''
    Binance retired its legacy USD-M ``/ws`` route in April 2026.

    Market and high-frequency public traffic now use separate base
    paths. Verify piker retains both routes so a
    future endpoint edit cannot silently send a valid subscription to
    a socket which never publishes that traffic class.

    '''
    routes: tuple[str, ...] = get_quote_routes(
        mkt_mode='usdtm_futes',
        symbols=['NVDAUSDT'],
    )
    assert routes == (
        f'{_futes_public_ws}/nvdausdt@bookTicker',
        f'{_futes_market_ws}/nvdausdt@aggTrade',
    )
    assert get_quote_routes(
        mkt_mode='spot',
        symbols=['BTCUSDT'],
    ) == (
        f'{_spot_ws}/btcusdt@bookTicker',
        f'{_spot_ws}/btcusdt@aggTrade',
    )


def test_silent_binance_feed_startup_fails() -> None:
    '''
    A retired Binance websocket accepted piker's connection but sent
    no aggregate trades, leaving ``stream_quotes()`` blocked before
    ``task_status.started()`` and hanging every chart consumer.

    Keep an empty receive channel open to reproduce a connected but
    silent socket. The controlled deadline proves startup raises
    ``DataUnavailable`` with the absent stream classes instead of
    waiting indefinitely for a first quote.

    '''
    async def main() -> None:
        send, recv = trio.open_memory_channel(1)
        async with send, recv:
            with pytest.raises(
                DataUnavailable,
                match='no live l1, trade data received',
            ):
                await wait_for_live_quotes(
                    msg_stream=recv,
                    symbols=['NVDAUSDT'],
                    timeout=0.01,
                )

    trio.run(main)


def test_websocket_connection_startup_fails(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    '''
    A websocket handshake which never completed previously blocked
    before Binance's live-data timeout started. Generic feed startup
    then waited 616 seconds, effectively hanging the chart like a
    connected socket which never published quotes.

    Replace the reconnect task with one which reports task startup
    but never sets ``NoBsWs._connected``. The bounded context entry
    proves ``open_autorecon_ws()`` raises a URL-bearing
    ``TimeoutError`` and cancels its nursery instead of waiting
    indefinitely.

    '''
    async def never_connect(
        *args,
        task_status=trio.TASK_STATUS_IGNORED,
        **kwargs,
    ) -> None:
        task_status.started()
        await trio.sleep_forever()

    monkeypatch.setattr(
        _web_bs,
        '_reconnect_forever',
        never_connect,
    )

    async def main() -> None:
        with pytest.raises(
            TimeoutError,
            match='wss://silent.invalid/ws',
        ):
            async with _web_bs.open_autorecon_ws(
                url='wss://silent.invalid/ws',
                connect_timeout=0.01,
            ):
                pytest.fail(
                    'silent websocket unexpectedly connected'
                )

    trio.run(main)


def test_usdtm_aggregate_trade_stream_type() -> None:
    '''
    Binance added ``st`` to USD-M aggregate trades during its June
    2026 UM/CM websocket integration. The first valid NVDAUSDT trade
    then raised ``TypeError`` while constructing piker's strict
    ``AggTrade`` struct, terminating the feed relay during startup.

    Feed one representative USD-M frame through ``stream_messages()``
    and assert its trade quote is emitted. This proves the new stream
    discriminator is accepted without weakening parsing of the fields
    used by downstream consumers.

    '''
    class MockWs:
        def __init__(self) -> None:
            self._msgs = iter([{
                'e': 'aggTrade',
                'E': 1787782452314,
                'a': 8430770,
                's': 'NVDAUSDT',
                'p': '217.91000',
                'q': '0.92',
                'nq': '0.92',
                'f': 20677382,
                'l': 20677382,
                'T': 1787782452162,
                'm': True,
                'st': 1,
            }])

        def __aiter__(self):
            return self

        async def __anext__(self) -> dict:
            try:
                return next(self._msgs)
            except StopIteration:
                raise StopAsyncIteration from None

    async def main() -> None:
        msg_gen = stream_messages(MockWs())
        typ, quote = await anext(msg_gen)
        assert typ == 'trade'
        assert quote['symbol'] == 'NVDAUSDT'
        assert quote['last'] == 217.91

    trio.run(main)