''' Deterministic history-backfill regressions. ''' from functools import partial from types import SimpleNamespace import numpy as np from pendulum import datetime import pytest import trio from trio import TaskStatus from piker.brokers import ( DataUnavailable, NoData, ) from piker.data._source import def_iohlcv_fields from piker.tsp._history import ( maybe_fill_null_segments, start_backfill, ) def run_exhausted_backfill(get_hist) -> None: ''' Run a provider-exhaustion case without actor services. ''' async def main() -> None: with trio.fail_after(0.5): async with trio.open_nursery() as nursery: await nursery.start(partial( start_backfill, get_hist=get_hist, def_frame_duration=None, mod=SimpleNamespace(name='fake'), mkt=SimpleNamespace(fqme='x.fake'), shm=object(), timeframe=60, backfill_from_shm_index=100, backfill_from_dt=datetime(2026, 1, 2), sampler_stream=object(), backfill_until_dt=datetime(2026, 1, 1), storage=None, write_tsdb=False, )) trio.run(main) def test_data_unavailable_completes() -> None: ''' Provider exhaustion exits without orphaned completion waits. ''' async def get_hist(*args, **kwargs): raise DataUnavailable('history exhausted') run_exhausted_backfill(get_hist) @pytest.mark.parametrize('error_type', [ NoData, DataUnavailable, ]) def test_gap_repair_failure_preserves_sibling( error_type: type[Exception], ) -> None: ''' An unavailable gap query must not cancel sibling feed tasks. Use real null detection and a provider that raises on its first query. After repair returns, release a sibling waiting in the same nursery and require its acknowledgement. This checks local failure containment, not Gateway reset or quote reconnection. ''' frame: np.ndarray = np.zeros( 5, dtype=np.dtype(def_iohlcv_fields), ) frame['index'] = np.arange(5) frame['time'] = datetime(2026, 11, 6).timestamp() + np.arange(5) frame['time'][1:3] = 0 frame['close'][0] = 100 calls: list[dict] = [] async def get_hist(timeframe: float, **kwargs) -> None: assert timeframe == 1 assert kwargs['start_dt'] < kwargs['end_dt'] calls.append(kwargs) raise error_type('provider cannot repair gap') async def main() -> None: repaired: trio.Event = trio.Event() resume: trio.Event = trio.Event() survived: trio.Event = trio.Event() send, receive = trio.open_memory_channel(1) async def sibling( task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED, ) -> None: task_status.started() await resume.wait() survived.set() async def repair() -> None: await maybe_fill_null_segments( shm=SimpleNamespace(array=frame, _array=frame), timeframe=1, get_hist=get_hist, sampler_stream=send, mkt=SimpleNamespace(fqme='mnq.cme.fake'), backfill_until_dt=datetime(2026, 11, 5), ) repaired.set() with trio.fail_after(1): async with trio.open_nursery() as nursery: await nursery.start(sibling) nursery.start_soon(repair) await repaired.wait() resume.set() await survived.wait() nursery.cancel_scope.cancel() assert len(calls) == 1 assert receive.receive_nowait() == { 'broadcast_all': { 'backfilling': ('mnq.cme.fake', 1), }, } assert (frame['time'] > 0).all() trio.run(main) def test_empty_frame_completes() -> None: ''' An empty provider frame is checked before endpoint indexing. ''' frame = np.empty( 0, dtype=np.dtype(def_iohlcv_fields), ) async def get_hist(*args, **kwargs): end_dt = kwargs['end_dt'] return frame, end_dt, end_dt run_exhausted_backfill(get_hist)