piker/tests/test_ib_method_proxy.py

'''
IB asyncio method-proxy regressions.

'''
from types import SimpleNamespace
from typing import Any

import pytest
import trio

from piker.brokers.ib import api as ib_api
from piker.brokers.ib.api import (
    MethodProxy,
    relay_client_proxy_messages,
)


class FakeChannel:
    '''
    Model the Trio side of an asyncio linked-task channel.

    '''
    def __init__(self) -> None:
        self.requests: list[tuple] = []
        (
            self._tx,
            self._rx,
        ) = trio.open_memory_channel[Any](10)

    async def send(self, msg: tuple) -> None:
        '''
        Capture one Trio-to-asyncio method request.

        '''
        self.requests.append(msg)

    async def receive(self) -> Any:
        '''
        Receive one simulated asyncio-to-Trio response.

        '''
        return await self._rx.receive()

    def closed(self) -> bool:
        '''
        Report an open channel for the lifetime of each test.

        '''
        return False

    async def respond(self, msg: Any) -> None:
        '''
        Send one response or status event to the proxy relay.

        '''
        await self._tx.send(msg)

    async def close_responses(self) -> None:
        '''
        Close the simulated asyncio-to-Trio response channel.

        '''
        await self._tx.aclose()


def test_method_proxy_routes_after_idle_status_events() -> None:
    '''
    Idle IB status traffic must not break first-symbol qualification.

    ``open_client_proxy()`` previously broadcast a one-slot channel
    to an always-running status relay and an idle method receiver.
    Status events advanced only the relay, so a first order for an
    uncached symbol raised ``Lagged`` before receiving
    ``get_sym_details()``. Feed several idle events through the
    sole-reader relay, then start two method calls and return their
    responses in reverse order. Both exact results prove status
    traffic cannot overrun a dormant caller and concurrent responses
    remain correlated by request ID.

    '''
    async def main() -> None:
        chan = FakeChannel()
        proxy = MethodProxy(
            chan,
            event_table={},
            asyncio_ns=SimpleNamespace(),
        )
        results: dict[str, str] = {}

        async def call(name: str) -> None:
            results[name] = await proxy._run_method(meth=name)

        async with trio.open_nursery() as nursery:
            nursery.start_soon(
                relay_client_proxy_messages,
                chan,
                proxy,
            )
            for i in range(3):
                await chan.respond((
                    'event',
                    {'reason': f'farm-status-{i}'},
                ))

            nursery.start_soon(call, 'first')
            nursery.start_soon(call, 'second')
            with trio.fail_after(0.1):
                while len(chan.requests) < 2:
                    await trio.lowlevel.checkpoint()

            mids: dict[str, int] = {
                meth: mid
                for meth, _kwargs, mid in chan.requests
            }
            await chan.respond({
                'mid': mids['second'],
                'result': 'second-result',
            })
            await chan.respond({
                'mid': mids['first'],
                'result': 'first-result',
            })
            with trio.fail_after(0.1):
                while len(results) < 2:
                    await trio.lowlevel.checkpoint()
            nursery.cancel_scope.cancel()

        assert results == {
            'first': 'first-result',
            'second': 'second-result',
        }

    trio.run(main)


def test_method_proxy_relay_accepts_clean_eof() -> None:
    '''
    Normal asyncio relay completion must not fail proxy teardown.

    The sole-reader relay blocks in ``LinkedTaskChannel.receive()``
    until the asyncio task exits and closes its Trio send channel.
    Close the fake response sender before entering the relay and
    prove the resulting ``EndOfChannel`` is graceful completion.

    '''
    async def main() -> None:
        chan = FakeChannel()
        proxy = MethodProxy(
            chan,
            event_table={},
            asyncio_ns=SimpleNamespace(),
        )
        await chan.close_responses()
        with trio.fail_after(0.1):
            await relay_client_proxy_messages(chan, proxy)

    trio.run(main)


def test_method_proxy_drops_cancelled_call_response() -> None:
    '''
    A cancelled symbol search must not skew the next method response.

    Search timeouts can cancel a Trio caller after its request
    reaches the asyncio client. Wait until one request is sent,
    cancel its caller, and then deliver the orphaned response. The
    sole-reader relay must drop that stale ``mid`` without failing; a
    subsequent call must still receive its own exact response.

    '''
    async def main() -> None:
        chan = FakeChannel()
        proxy = MethodProxy(
            chan,
            event_table={},
            asyncio_ns=SimpleNamespace(),
        )
        call_scope = trio.CancelScope()
        cancelled = trio.Event()
        result: list[str] = []

        async def cancelled_call() -> None:
            with call_scope:
                try:
                    await proxy._run_method(meth='cancelled')
                finally:
                    cancelled.set()

        async def next_call() -> None:
            value = await proxy._run_method(meth='next')
            result.append(value)

        async with trio.open_nursery() as nursery:
            nursery.start_soon(
                relay_client_proxy_messages,
                chan,
                proxy,
            )
            nursery.start_soon(cancelled_call)
            with trio.fail_after(0.1):
                while len(chan.requests) < 1:
                    await trio.lowlevel.checkpoint()
            cancelled_mid: int = chan.requests[0][2]
            call_scope.cancel()
            with trio.fail_after(0.1):
                await cancelled.wait()
            assert proxy._pending == {}
            assert (
                proxy._request_methods[cancelled_mid]
                ==
                'cancelled'
            )

            await chan.respond({
                'mid': cancelled_mid,
                'result': 'stale',
            })
            with trio.fail_after(0.1):
                while cancelled_mid in proxy._request_methods:
                    await trio.lowlevel.checkpoint()
            nursery.start_soon(next_call)
            with trio.fail_after(0.1):
                while len(chan.requests) < 2:
                    await trio.lowlevel.checkpoint()
            next_mid: int = chan.requests[1][2]
            await chan.respond({
                'mid': next_mid,
                'result': 'fresh',
            })
            with trio.fail_after(0.1):
                while not result:
                    await trio.lowlevel.checkpoint()
            nursery.cancel_scope.cancel()

        assert result == ['fresh']

    trio.run(main)


def test_method_proxy_warns_when_metadata_is_evicted(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    '''
    Unresolved metadata eviction must remain operator-visible.

    Cancelled calls whose asyncio responses never arrive retain a
    bounded method name solely for later stale-response diagnostics.
    Fill that table beyond its limit and prove the oldest entry is
    evicted with its request ID, method name, and configured bound in
    the warning.

    '''
    proxy = MethodProxy(
        FakeChannel(),
        event_table={},
        asyncio_ns=SimpleNamespace(),
    )
    warnings: list[str] = []
    monkeypatch.setattr(ib_api.log, 'warning', warnings.append)

    for mid in range(257):
        proxy._track_request_method(mid, f'method-{mid}')

    assert len(proxy._request_methods) == 256
    assert 0 not in proxy._request_methods
    assert warnings == [
        'Evicting unresolved method-request metadata,\n'
        'mid: 0\n'
        "meth: 'method-0'\n"
        'pending metadata limit: 256\n'
    ]