''' Offline packet-idle timer regressions using real eventkit events. ''' import asyncio from datetime import datetime from types import SimpleNamespace from eventkit import Event from ib_async import wrapper as upstream import pytest from piker.brokers.ib import api @pytest.fixture def timer(monkeypatch): ''' Drive real asyncio TimerHandles with a deterministic wall clock. ''' class ClockLoop: now = 1_800_000_000.0 def __init__(self): self.handles = [] def get_debug(self): return False def _timer_handle_cancelled(self, handle): pass def call_exception_handler(self, context): raise context['exception'] def call_later(self, delay, callback): handle = asyncio.TimerHandle( self.now + delay, callback, (), self, ) self.handles.append(handle) return handle def advance(self, seconds): target = self.now + seconds while self.handles: handle = min(self.handles, key=lambda h: h.when()) if handle.when() > target: break self.handles.remove(handle) self.now = handle.when() if not handle.cancelled(): handle._run() self.now = target loop = ClockLoop() class ClockDateTime(datetime): @classmethod def now(cls, tz=None): return cls.fromtimestamp(loop.now, tz) monkeypatch.setattr(upstream, 'datetime', ClockDateTime) monkeypatch.setattr(upstream, 'getLoop', lambda: loop) monkeypatch.setattr(api, 'getLoop', lambda: loop) monkeypatch.setattr( api.time, 'time_ns', lambda: int(loop.now * 1_000_000_000), ) ib = SimpleNamespace(timeoutEvent=Event('timeoutEvent')) wrapper = api.NonShittyWrapper(ib) events = [] def record(diff): events.append((diff, wrapper.lastTime)) ib.timeoutEvent.connect(record) return loop, wrapper, events def test_packet_timeout_reschedules_and_emits_once(timer): ''' Upstream datetime subtraction raised TypeError after a packet replaced NonShittyWrapper.lastTime with epoch nanoseconds. Arm the real wrapper, deliver packets around the first deadline, and drive its TimerHandles without sleeping or opening sockets. New traffic must postpone expiry by the remaining idle interval; the real eventkit listener must see seconds and the intact ns timestamp exactly once, with the timer already disabled. ''' loop, wrapper, events = timer wrapper.setTimeout(5) loop.advance(1) wrapper.tcpDataArrived() assert isinstance(wrapper.lastTime, int) loop.advance(3) wrapper.tcpDataArrived() packet_time = wrapper.lastTime loop.advance(1) assert events == [] assert wrapper.lastTime == packet_time assert wrapper._timeoutHandle.when() == loop.now + 4 loop.advance(4) assert events == [(5.0, packet_time)] assert wrapper._timeoutHandle is None assert wrapper._timeout == 0 assert wrapper.lastTime == packet_time wrapper._setTimer() loop.advance(20) assert events == [(5.0, packet_time)] def test_datetime_timeout_and_sentinel(timer): ''' Before any packet, Wrapper.setTimeout uses datetime timestamps. The ns fix must preserve that upstream path and datetime.min's sentinel short circuit. Run an armed pre-traffic timer to expiry, then invoke the sentinel path and assert no callback is queued. ''' loop, wrapper, events = timer assert isinstance(wrapper.lastTime, datetime) assert wrapper._timeoutHandle is None wrapper.setTimeout(3) loop.advance(3) assert len(events) == 1 assert events[0][0] == 3.0 assert isinstance(events[0][1], datetime) assert wrapper._timeoutHandle is None wrapper.lastTime = datetime.min wrapper._setTimer(2) loop.advance(10) assert len(events) == 1 assert wrapper.lastTime == datetime.min assert wrapper._timeoutHandle is None def test_rearm_disable_and_reentrant_listener(timer): ''' Replacing or disabling a packet-idle timer must cancel its old handle. Expiry must also finish teardown before timeoutEvent listeners rearm it, otherwise the new timer can be lost. Advance cancelled deadlines deterministically, then rearm from a real eventkit listener and deliver another packet. Both arms must emit once, with packet timestamps intact during emission. ''' loop, wrapper, events = timer wrapper.setTimeout(2) wrapper.tcpDataArrived() old = wrapper._timeoutHandle wrapper.setTimeout(4) assert old.cancelled() replacement = wrapper._timeoutHandle wrapper.tcpDataArrived() wrapper.setTimeout(0) assert replacement.cancelled() assert wrapper._timeoutHandle is None wrapper.tcpDataArrived() loop.advance(10) assert events == [] observed = [] def rearm(diff): observed.append((wrapper._timeout, wrapper._timeoutHandle)) if len(observed) == 1: wrapper.setTimeout(3) wrapper.tcpDataArrived() wrapper.ib.timeoutEvent.connect(rearm) wrapper.setTimeout(2) wrapper.tcpDataArrived() first_packet = wrapper.lastTime loop.advance(2) second_packet = wrapper.lastTime assert wrapper._timeout == 3 assert wrapper._timeoutHandle is not None loop.advance(3) loop.advance(10) assert events == [(2.0, first_packet), (3.0, second_packet)] assert observed == [(0, None), (0, None)] assert wrapper.lastTime == second_packet