'''
IB quote restart regressions using the real Tractor asyncio bridge.

'''
import asyncio
from contextlib import asynccontextmanager as acm
from types import SimpleNamespace

from async_generator import aclosing
from ib_async import Contract, IB, RequestError, Ticker
import pytest
import tractor
import trio

from piker.brokers.ib import feed
from piker.brokers.ib import _util
from piker.brokers.ib.api import Client, open_client_proxy


async def exercise_quote_restart() -> list[str]:
    '''
    Cancel an active quote consumer, then reopen in the same actor.

    Only the external IB connection is replaced. The production quote
    producer, eventkit callback, bridge, broadcaster and context cache
    all run unchanged in an asyncio-infected Tractor child. The farm
    OK event in stream_quotes() cancels its local scope after a reset;
    reproduce that cancellation after an actual ticker delivery.

    A cached owner can turn this into a normal bridge exit, feeding
    TrioTaskExited back into the consumer and killing brokerd. Two
    generations must close their IB subscriptions and asyncio tasks.
    Return callback failures separately so actor survival cannot be
    mistaken for clean teardown. This proves local restart semantics,
    not connectivity to a real Gateway.

    '''
    contract = Contract(symbol='QQQ')
    tasks: list[asyncio.Task] = []
    tickers: list[Ticker] = []
    cancelled: list[Contract] = []
    callback_errors: list[dict] = []

    class ReplayIB:
        '''
        Deliver one real eventkit ticker per subscription.

        '''
        def reqMktData(self, con: Contract, opts: str) -> Ticker:
            '''
            Emit once the production producer installs its callback.

            '''
            task = asyncio.current_task()
            assert task is not None
            tasks.append(task)
            ticker = Ticker(contract=con)
            tickers.append(ticker)
            asyncio.get_running_loop().call_soon(
                ticker.updateEvent.emit,
                ticker,
            )
            return ticker

        def cancelMktData(self, con: Contract) -> None:
            '''
            Record the production producer's subscription teardown.

            '''
            cancelled.append(con)

    client = SimpleNamespace(ib=ReplayIB())

    @acm
    async def load_clients(**kwargs):
        '''
        Supply an offline IB peer without opening broker connections.

        '''
        yield {'ib.replay': client}

    loop = asyncio.get_running_loop()
    previous_handler = loop.get_exception_handler()
    loop.set_exception_handler(
        lambda loop, context: callback_errors.append(context)
    )
    try:
        with pytest.MonkeyPatch.context() as patch:
            patch.setattr(feed, 'load_aio_clients', load_clients)
            patch.setattr(
                feed,
                'get_preferred_data_client',
                lambda clients: ('ib.replay', client),
            )
            for generation in range(2):
                with trio.CancelScope() as scope:
                    async with feed.open_aio_quote_stream(
                        'qqq.nasdaq',
                        contract=contract,
                    ) as quotes:
                        async with aclosing(quotes):
                            ticker = await quotes.receive()
                            assert ticker is tickers[generation]
                            scope.cancel()
                            await trio.lowlevel.checkpoint()

                assert scope.cancelled_caught
                assert len(cancelled) == generation + 1
                assert tasks[generation].done()
                assert len(tickers[generation].updateEvent) == 0

            # Flush done callbacks before accepting a clean restart.
            await tractor.to_asyncio.run_task(asyncio.sleep, delay=0)
            return [
                repr(context['exception'])
                for context in callback_errors
            ]
    finally:
        loop.set_exception_handler(previous_handler)


@pytest.mark.parametrize(
    'check_callbacks',
    [False, True],
    ids=['actor-survival', 'callback-cleanliness'],
)
def test_quote_restart_preserves_actor(
    reg_addr,
    check_callbacks: bool,
) -> None:
    '''
    A farm-style quote restart must leave its broker actor usable.

    Use a real TCP actor nursery and an asyncio-infected child. Run
    two subscription generations, then invoke another RPC on the same
    actor. Both cases reject a propagated bridge exception, leaked
    quote task or lost actor. The strict case also rejects asyncio
    callback failures; the survival case reports those independently.
    The backout can restore restart survival while the strict case
    still exposes Tractor's completion-callback defect.

    '''
    async def main() -> None:
        with trio.fail_after(10):
            async with (
                tractor.open_root_actor(registry_addrs=[reg_addr]),
                tractor.open_nursery() as nursery,
            ):
                portal = await nursery.start_actor(
                    'ib_quote_replay',
                    enable_modules=[__name__],
                    infect_asyncio=True,
                )
                errors = await portal.run(exercise_quote_restart)
                errors += await portal.run(exercise_quote_restart)
                await portal.cancel_actor()
            if check_callbacks:
                assert not errors, errors

    trio.run(main)


async def exercise_history_recovery(timeframe: int) -> None:
    '''
    Replay pacing and farm recovery across the production IB adapter.

    Keep Client.bars(), MethodProxy, both relay tasks, get_bars(),
    wait_on_data_reset() and data_reset_hack() intact. Replace only
    external contract discovery, historical replies and VNC input.
    Deliver the farm event through IB.errorEvent after the real reset
    task is waiting, and reject a retry before that recovery signal.

    This covers the bridge missing from the imported policy tests;
    it neither sends VNC keys nor connects to a broker or storage.

    '''
    requests: int = 0
    reset_calls: int = 0
    recovery_sent: bool = False
    reset_started = trio.Event()
    contract = Contract(symbol='QQQ')
    reason = 'HMDS data farm connection is OK:ushmds'

    class ReplayIB(IB):
        '''
        Use real IB events with scripted historical responses.

        '''
        async def reqHistoricalDataAsync(self, **kwargs) -> list:
            '''
            Reject the first request; return empty data after reset.

            '''
            nonlocal requests
            requests += 1
            if requests == 1:
                raise RequestError(1, 162, feed._pacing)
            assert recovery_sent
            assert requests == 2
            return []

    ib = ReplayIB()
    client = Client(ib=ib, config={'vnc_addrs': []})

    async def find_contracts(fqme: str) -> list[Contract]:
        '''
        Return the replay contract without network qualification.

        '''
        return [contract]

    async def vnc_input(client: Client) -> None:
        '''
        Record completion of external reset input on asyncio.

        '''
        nonlocal reset_calls
        reset_calls += 1
        assert reset_calls == 1
        reset_started.set()
        await asyncio.sleep(0)

    async def emit_recovery() -> None:
        '''
        Deliver the Gateway status through Client.inline_errors().

        '''
        nonlocal recovery_sent
        recovery_sent = True
        ib.errorEvent.emit(-1, 2106, reason, contract)

    with pytest.MonkeyPatch.context() as patch:
        patch.setattr(client, 'find_contracts', find_contracts)
        patch.setattr(_util, '_reset_tech', 'vnc')
        patch.setattr(_util, 'vnc_click_hack', vnc_input)
        async with open_client_proxy(client) as proxy:
            async def gateway_status() -> None:
                '''
                Wait for the actual farm-event consumer to be ready.

                '''
                await reset_started.wait()
                event = proxy.event_table[reason]
                while not event.statistics().tasks_waiting:
                    await trio.lowlevel.checkpoint()
                await tractor.to_asyncio.run_task(emit_recovery)

            async with trio.open_nursery() as nursery:
                nursery.start_soon(gateway_status)
                result, reset_used = await feed.get_bars(
                    proxy,
                    'qqq.nasdaq',
                    timeframe,
                )
                assert result is None
                assert reset_used
                assert not feed._data_reset_lock.locked()

    assert requests == 2
    assert reset_calls == 1


def test_history_pacing_through_real_bridge(reg_addr) -> None:
    '''
    Pacing recovery must complete before retrying either bar period.

    Run the production proxy and reset tasks in an infected actor for
    both 1s and 60s history. A returned empty frame must release the
    request rather than trigger more resets. Repeat on the same actor
    to expose leaked ownership from the first request.

    '''
    async def main() -> None:
        with trio.fail_after(10):
            async with (
                tractor.open_root_actor(registry_addrs=[reg_addr]),
                tractor.open_nursery() as nursery,
            ):
                portal = await nursery.start_actor(
                    'ib_history_replay',
                    enable_modules=[__name__],
                    infect_asyncio=True,
                )
                for timeframe in (1, 60):
                    await portal.run(
                        exercise_history_recovery,
                        timeframe=timeframe,
                    )
                await portal.cancel_actor()

    trio.run(main)