149 lines
3.8 KiB
Python
149 lines
3.8 KiB
Python
'''
|
|
Typed chart-local gap-overlay regressions.
|
|
|
|
'''
|
|
import msgspec
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from piker.ui._gaps import (
|
|
GapOverlay,
|
|
GapOverlayMngr,
|
|
GapOverlayPayload,
|
|
GapSpec,
|
|
SetGapOverlay,
|
|
gap_specs_from_ohlcv,
|
|
)
|
|
|
|
|
|
def test_gap_specs_and_wire_roundtrip() -> None:
|
|
'''
|
|
Local detection and remote IPC share one typed request model.
|
|
|
|
Build a small sorted OHLCV array with one positive timestamp gap,
|
|
derive its geometry without a per-row dataframe scan, and
|
|
roundtrip the request/reply through msgpack. This proves callers
|
|
invoke one chart renderer with one concrete schema.
|
|
|
|
'''
|
|
dtype: np.dtype = np.dtype([
|
|
('index', 'i8'),
|
|
('time', 'f8'),
|
|
('open', 'f8'),
|
|
('close', 'f8'),
|
|
])
|
|
array: np.ndarray = np.array([
|
|
(10, 60, 100, 101),
|
|
(11, 120, 101, 102),
|
|
(12, 300, 95, 96),
|
|
(13, 360, 96, 97),
|
|
], dtype=dtype)
|
|
|
|
specs: list[GapSpec] = gap_specs_from_ohlcv(
|
|
array,
|
|
period_s=60,
|
|
)
|
|
assert len(specs) == 1
|
|
spec: GapSpec = specs[0]
|
|
assert spec.start_time == 120
|
|
assert spec.end_time == 300
|
|
assert spec.start_pos == (11.9, 102)
|
|
assert spec.end_pos == (12.1, 95)
|
|
assert spec.pointing == 'down'
|
|
|
|
req: SetGapOverlay = SetGapOverlay(
|
|
fqme='mnq.cme.ib',
|
|
timeframe=60,
|
|
specs=specs,
|
|
request_id='req-616',
|
|
)
|
|
req_wire: bytes = msgspec.msgpack.encode(req)
|
|
assert msgspec.msgpack.decode(
|
|
req_wire,
|
|
type=SetGapOverlay,
|
|
) == req
|
|
|
|
resp: GapOverlay = GapOverlay(
|
|
fqme=req.fqme,
|
|
timeframe=req.timeframe,
|
|
visible=True,
|
|
gap_count=1,
|
|
aid=616,
|
|
request_id=req.request_id,
|
|
)
|
|
resp_wire: bytes = msgspec.msgpack.encode(resp)
|
|
assert msgspec.msgpack.decode(
|
|
resp_wire,
|
|
type=GapOverlay,
|
|
) == resp
|
|
|
|
|
|
def test_gap_overlay_payload_rejects_other_msgs() -> None:
|
|
'''
|
|
Reject payloads outside the strict gap-overlay dialog protocol.
|
|
|
|
The sibling Tractor context installs `GapOverlayPayload` so
|
|
legacy annotation dicts cannot enter its stream as partially
|
|
validated commands. Encode one such dict and prove the same
|
|
msgspec decoder rejects it before endpoint dispatch.
|
|
|
|
'''
|
|
invalid_wire: bytes = msgspec.msgpack.encode({
|
|
'cmd': 'SelectRect',
|
|
})
|
|
|
|
with pytest.raises(msgspec.ValidationError):
|
|
msgspec.msgpack.decode(
|
|
invalid_wire,
|
|
type=GapOverlayPayload,
|
|
)
|
|
|
|
req: SetGapOverlay = SetGapOverlay(
|
|
fqme='mnq.cme.ib',
|
|
timeframe=60,
|
|
specs=[],
|
|
)
|
|
resp: GapOverlay = GapOverlay(
|
|
fqme=req.fqme,
|
|
timeframe=req.timeframe,
|
|
visible=False,
|
|
gap_count=0,
|
|
)
|
|
with pytest.raises(msgspec.ValidationError):
|
|
msgspec.msgpack.decode(
|
|
msgspec.msgpack.encode(resp),
|
|
type=SetGapOverlay,
|
|
)
|
|
with pytest.raises(msgspec.ValidationError):
|
|
msgspec.msgpack.decode(
|
|
msgspec.msgpack.encode(req),
|
|
type=GapOverlay,
|
|
)
|
|
|
|
|
|
def test_gap_overlay_unknown_fqme_returns_typed_error() -> None:
|
|
'''
|
|
Keep a stale remote FQME request inside the typed dialog.
|
|
|
|
A chart can unload or replace display state after a client learns
|
|
its FQME set. Submit a valid request to an empty controller and
|
|
prove it returns `GapOverlay.error` instead of leaking a
|
|
`KeyError` which would terminate the whole Tractor context.
|
|
|
|
'''
|
|
gapman: GapOverlayMngr = GapOverlayMngr(
|
|
dss={},
|
|
annots={},
|
|
)
|
|
req: SetGapOverlay = SetGapOverlay(
|
|
fqme='gone.test',
|
|
timeframe=60,
|
|
specs=[],
|
|
request_id='req-stale',
|
|
)
|
|
|
|
resp: GapOverlay = gapman.apply(req)
|
|
|
|
assert resp.request_id == req.request_id
|
|
assert resp.error == 'No display state for fqme=gone.test'
|