piker/tests/providers/test_ib_wrapper_timeout.py

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