''' Typed chart-local gap-overlay regressions. ''' from collections.abc import Iterator import os from types import SimpleNamespace os.environ['QT_QPA_PLATFORM'] = 'offscreen' import msgspec import numpy as np from PyQt6.QtWidgets import QGraphicsScene import pyqtgraph as pg import pytest from piker.ui._annotate import GapAnnotations from piker.ui._gaps import ( GapOverlay, GapOverlayMngr, GapOverlayPayload, GapSpec, SetGapOverlay, gap_specs_from_ohlcv, ) from piker.ui.qt import ( QApplication, QPointF, QRectF, ) FQME: str = 'gap.test' class _ChartStub: ''' Real Qt plot with the small chart API required by `gapman`. ''' def __init__( self, fqme: str, array: np.ndarray, ) -> None: self.fqme: str = fqme self.widget: pg.PlotWidget = pg.PlotWidget() self.viz: SimpleNamespace = SimpleNamespace( plot=self.widget.plotItem, shm=SimpleNamespace(array=array), ) def get_viz( self, fqme: str, ) -> SimpleNamespace: ''' Return this chart's only viz. ''' assert fqme == self.fqme return self.viz def close(self) -> None: ''' Release this test-owned plot widget. ''' self.widget.close() self.widget.deleteLater() def _ohlcv_array( times: tuple[float, ...], ) -> np.ndarray: ''' Build the minimal structured OHLCV used by gap rendering. ''' dtype: np.dtype = np.dtype([ ('index', 'i8'), ('time', 'f8'), ('open', 'f8'), ('close', 'f8'), ]) rows: list[tuple[int, float, float, float]] = [] i: int time_s: float for i, time_s in enumerate(times): price: float = 100 + i rows.append((i, time_s, price, price + 0.5)) return np.array(rows, dtype=dtype) def _display_state( fqme: str, ) -> tuple[ SimpleNamespace, tuple[_ChartStub, _ChartStub], ]: ''' Build independent realtime and history chart stubs. ''' rt_chart: _ChartStub = _ChartStub( fqme, _ohlcv_array((1, 2, 4, 5)), ) hist_chart: _ChartStub = _ChartStub( fqme, _ohlcv_array((60, 120, 300, 360)), ) ds: SimpleNamespace = SimpleNamespace( fqme=fqme, chart=rt_chart, viz=rt_chart.viz, hist_chart=hist_chart, hist_viz=hist_chart.viz, ) return ds, (rt_chart, hist_chart) @pytest.fixture(scope='session') def qapp() -> Iterator[QApplication]: ''' Keep one offscreen Qt application alive for graphics tests. ''' app: QApplication|None = QApplication.instance() if app is None: app = QApplication(['piker-gap-tests']) app.setQuitOnLastWindowClosed(False) yield app app.processEvents() 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' def test_gap_manager_real_qt_lifecycle( qapp: QApplication, ) -> None: ''' Remove overlays from every PyQtGraph registry. Gap layers are inserted through `PlotItem.addItem()`. Removing the item directly from its scene leaves stale `PlotItem.items` and `ViewBox.addedItems` refs which can leak memory and affect range calculations. This test uses real offscreen Qt plots, replaces one remotely owned layer, hides it, and proves each registry releases the old graphics object. ''' ds: SimpleNamespace charts: tuple[_ChartStub, _ChartStub] ds, charts = _display_state(FQME) annots: dict[int, GapAnnotations] = {} gapman: GapOverlayMngr = GapOverlayMngr( dss={FQME: ds}, annots=annots, ) specs: list[GapSpec] = gap_specs_from_ohlcv( ds.hist_viz.shm.array, period_s=60, ) aids: set[int] = set() req: SetGapOverlay = SetGapOverlay( fqme=FQME, timeframe=60, specs=specs, request_id='first', ) try: first: GapOverlay = gapman.apply( req=req, owner='remote-test', aids=aids, ) assert first.aid is not None first_item: GapAnnotations = annots[first.aid] assert first_item.scene() is charts[1].widget.scene() assert first_item in ds.hist_viz.plot.items assert first_item not in ds.hist_viz.plot.vb.addedItems assert aids == {first.aid} second: GapOverlay = gapman.apply( req=SetGapOverlay( fqme=FQME, timeframe=60, specs=specs, request_id='second', ), owner='remote-test', aids=aids, ) assert second.aid is not None assert second.aid != first.aid assert first.aid not in annots assert first_item.scene() is None assert first_item not in ds.hist_viz.plot.items assert first_item not in ds.hist_viz.plot.vb.addedItems assert aids == {second.aid} hidden: GapOverlay = gapman.apply( req=SetGapOverlay( fqme=FQME, timeframe=60, specs=specs, visible=False, request_id='hidden', ), owner='remote-test', aids=aids, ) assert hidden.visible is False assert hidden.aid is None assert aids == set() assert annots == {} assert gapman._layers == {} assert gapman._plots == {} item: object for item in ds.hist_viz.plot.items: assert not isinstance(item, GapAnnotations) local: GapOverlay = gapman.apply( req=SetGapOverlay( fqme=FQME, timeframe=60, specs=specs, request_id='local', ), ) remote: GapOverlay = gapman.apply( req=SetGapOverlay( fqme=FQME, timeframe=60, specs=specs, request_id='remote', ), owner='remote-test', ) assert local.aid is not None assert remote.aid is not None gapman.remove_owner('remote-test') assert local.aid in annots assert remote.aid not in annots assert annots[local.aid].scene() is not None gapman.remove_owner('chart-local') assert annots == {} finally: chart: _ChartStub for chart in charts: chart.close() qapp.processEvents() def test_gap_annotations_reposition_after_prepend( qapp: QApplication, ) -> None: ''' Reposition live gap geometry after history index offsets change. A history prepend preserves timestamps but shifts every absolute chart index. Gap rectangles and arrows otherwise remain attached to their pre-prepend x coordinates. Render one real gap item, shift its source indexes by ten, call `reposition()`, and prove its rectangle bounds and arrow spec move together while the item stays attached to its real Qt scene. ''' ds: SimpleNamespace charts: tuple[_ChartStub, _ChartStub] ds, charts = _display_state(FQME) annots: dict[int, GapAnnotations] = {} gapman: GapOverlayMngr = GapOverlayMngr( dss={FQME: ds}, annots=annots, ) try: gap: GapOverlay = gapman.refresh( ds=ds, timeframe=60, ) assert gap.aid is not None item: GapAnnotations = annots[gap.aid] old_br: QRectF = item.boundingRect() scene: QGraphicsScene = charts[1].widget.scene() old_scene_point: QPointF = item.mapToScene( old_br.center() ) qapp.processEvents() assert item in scene.items(old_scene_point) old_rects: np.ndarray = ( item._rectarray.ndarray().copy() ) shifted: np.ndarray = ds.hist_viz.shm.array.copy() shifted['index'] += 10 item.reposition( array=shifted, fqme=FQME, timeframe=60, ) qapp.processEvents() new_rects: np.ndarray = item._rectarray.ndarray() np.testing.assert_allclose( new_rects[:, 0], old_rects[:, 0] + 10, ) assert item._gap_specs[0]['arrow_x'] == 12 new_br: QRectF = item.boundingRect() new_scene_point: QPointF = item.mapToScene( new_br.center() ) assert new_br.left() == old_br.left() + 10 assert item in scene.items(new_scene_point) assert item not in scene.items(old_scene_point) assert item.scene() is charts[1].widget.scene() finally: gapman.remove_owner('chart-local') chart: _ChartStub for chart in charts: chart.close() qapp.processEvents() def test_duplicate_fqme_layers_use_local_chart_identity( qapp: QApplication, ) -> None: ''' Keep cached displays for one FQME in separate local layers. The actor-global display lookup is keyed by FQME and can represent only its currently registered display. Local cached charts still pass their exact `DisplayState` into `GapOverlayMngr.refresh()`. Build two independent chart pairs with the same FQME, refresh both, and prove their layers survive independently. ''' first_ds: SimpleNamespace first_charts: tuple[_ChartStub, _ChartStub] first_ds, first_charts = _display_state(FQME) second_ds: SimpleNamespace second_charts: tuple[_ChartStub, _ChartStub] second_ds, second_charts = _display_state(FQME) annots: dict[int, GapAnnotations] = {} gapman: GapOverlayMngr = GapOverlayMngr( dss={FQME: second_ds}, annots=annots, ) try: first: GapOverlay = gapman.refresh( ds=first_ds, timeframe=60, ) second: GapOverlay = gapman.refresh( ds=second_ds, timeframe=60, ) assert first.aid is not None assert second.aid is not None assert first.aid != second.aid assert len(gapman._layers) == 2 chart_ids: set[int] = set() key: tuple[str, int, str, float] for key in gapman._layers: chart_ids.add(key[1]) assert chart_ids == { id(first_ds.hist_chart), id(second_ds.hist_chart), } assert annots[first.aid].scene() is not None assert annots[second.aid].scene() is not None finally: gapman.remove_owner('chart-local') chart: _ChartStub for chart in (*first_charts, *second_charts): chart.close() qapp.processEvents()