70 lines
1.8 KiB
Python
70 lines
1.8 KiB
Python
'''
|
|
Deterministic history-backfill regressions.
|
|
|
|
'''
|
|
from functools import partial
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
from pendulum import datetime
|
|
import trio
|
|
|
|
from piker.brokers import DataUnavailable
|
|
from piker.data._source import def_iohlcv_fields
|
|
from piker.tsp._history import 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)
|
|
|
|
|
|
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)
|