''' 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' ]