730 lines
20 KiB
Python
730 lines
20 KiB
Python
'''
|
|
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.QtGui import QKeyEvent
|
|
from PyQt6.QtWidgets import QGraphicsScene
|
|
import pyqtgraph as pg
|
|
import pytest
|
|
import trio
|
|
from trio.testing import wait_all_tasks_blocked
|
|
|
|
from piker.ui._annotate import GapAnnotations
|
|
from piker.ui._display import _register_gap_overlays
|
|
from piker.ui._gaps import (
|
|
GapOverlay,
|
|
GapOverlayMngr,
|
|
GapOverlayPayload,
|
|
GapSpec,
|
|
SetGapOverlay,
|
|
gap_specs_from_ohlcv,
|
|
)
|
|
from piker.ui._interaction import (
|
|
ChartView,
|
|
_toggle_gap_overlays,
|
|
)
|
|
from piker.ui.qt import (
|
|
QApplication,
|
|
QEvent,
|
|
QPointF,
|
|
QRectF,
|
|
Qt,
|
|
QWidget,
|
|
)
|
|
|
|
|
|
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()
|
|
|
|
|
|
def test_startup_and_focused_toggle_use_real_qt(
|
|
qapp: QApplication,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
|
|
) -> None:
|
|
'''
|
|
Register history gaps and toggle the focused chart only.
|
|
|
|
Startup previously lived inside the quote loop and keyboard
|
|
routing treated every state unrelated to the focused history
|
|
chart as a realtime target. Two independent chart pairs reproduce
|
|
that skew. The startup helper must create only 60-second layers;
|
|
then a focused history toggle must hide only its own layer while
|
|
leaving the other cached chart's overlay intact.
|
|
|
|
'''
|
|
from piker.ui import _remote_ctl
|
|
|
|
first_ds: SimpleNamespace
|
|
first_charts: tuple[_ChartStub, _ChartStub]
|
|
first_ds, first_charts = _display_state('first.test')
|
|
second_ds: SimpleNamespace
|
|
second_charts: tuple[_ChartStub, _ChartStub]
|
|
second_ds, second_charts = _display_state('second.test')
|
|
dss: dict[str, SimpleNamespace] = {
|
|
first_ds.fqme: first_ds,
|
|
second_ds.fqme: second_ds,
|
|
}
|
|
annots: dict[int, GapAnnotations] = {}
|
|
gapman: GapOverlayMngr = GapOverlayMngr(
|
|
dss={},
|
|
annots=annots,
|
|
)
|
|
remote_dss: dict[str, SimpleNamespace] = {}
|
|
monkeypatch.setattr(_remote_ctl, '_dss', remote_dss)
|
|
monkeypatch.setattr(_remote_ctl, '_annots', annots)
|
|
monkeypatch.setattr(_remote_ctl, '_gapman', gapman)
|
|
godwidget: SimpleNamespace = SimpleNamespace(gapman=None)
|
|
|
|
try:
|
|
started: GapOverlayMngr = _register_gap_overlays(
|
|
dss=dss,
|
|
godwidget=godwidget,
|
|
)
|
|
assert started is gapman
|
|
assert godwidget.gapman is gapman
|
|
assert remote_dss == dss
|
|
assert len(gapman._layers) == 2
|
|
key: tuple[str, int, str, float]
|
|
for key in gapman._layers:
|
|
assert key[3] == 60
|
|
|
|
view: SimpleNamespace = SimpleNamespace(
|
|
_chart=first_ds.hist_chart,
|
|
)
|
|
toggled: list[GapOverlay] = _toggle_gap_overlays(
|
|
view=view,
|
|
dss=dss,
|
|
gapman=gapman,
|
|
)
|
|
assert len(toggled) == 1
|
|
assert toggled[0].fqme == first_ds.fqme
|
|
assert toggled[0].timeframe == 60
|
|
assert toggled[0].visible is False
|
|
assert (
|
|
'chart-local',
|
|
id(first_ds.hist_chart),
|
|
first_ds.fqme,
|
|
60,
|
|
) not in gapman._layers
|
|
assert (
|
|
'chart-local',
|
|
id(second_ds.hist_chart),
|
|
second_ds.fqme,
|
|
60,
|
|
) in gapman._layers
|
|
|
|
view._chart = first_ds.chart
|
|
rt_toggled: list[GapOverlay] = _toggle_gap_overlays(
|
|
view=view,
|
|
dss=dss,
|
|
gapman=gapman,
|
|
)
|
|
assert len(rt_toggled) == 1
|
|
assert rt_toggled[0].fqme == first_ds.fqme
|
|
assert rt_toggled[0].timeframe == 1
|
|
assert rt_toggled[0].visible is True
|
|
assert (
|
|
'chart-local',
|
|
id(first_ds.chart),
|
|
first_ds.fqme,
|
|
1,
|
|
) in gapman._layers
|
|
|
|
unrelated: list[GapOverlay] = _toggle_gap_overlays(
|
|
view=SimpleNamespace(_chart=object()),
|
|
dss=dss,
|
|
gapman=gapman,
|
|
)
|
|
assert unrelated == []
|
|
finally:
|
|
gapman.remove_owner('chart-local')
|
|
chart: _ChartStub
|
|
for chart in (*first_charts, *second_charts):
|
|
chart.close()
|
|
qapp.processEvents()
|
|
|
|
|
|
def test_ctrl_g_event_renders_real_history_overlay(
|
|
qapp: QApplication,
|
|
|
|
) -> None:
|
|
'''
|
|
Route a real Qt Ctrl-G event into a rendered history gap layer.
|
|
|
|
Calling the toggle helper directly does not prove that Qt event
|
|
filtering, `KeyboardMsg` conversion or the asynchronous view-mode
|
|
handler recognizes the configured binding. Install the production
|
|
`EventRelay` on a real widget, send one offscreen `QKeyEvent`,
|
|
and wait until the handler blocks again. The resulting manager
|
|
and scene state prove the complete local keyboard path ran.
|
|
|
|
'''
|
|
ds: SimpleNamespace
|
|
charts: tuple[_ChartStub, _ChartStub]
|
|
ds, charts = _display_state(FQME)
|
|
annots: dict[int, GapAnnotations] = {}
|
|
gapman: GapOverlayMngr = GapOverlayMngr(
|
|
dss={FQME: ds},
|
|
annots=annots,
|
|
)
|
|
nav: SimpleNamespace = SimpleNamespace(
|
|
hide_info=lambda: None,
|
|
)
|
|
lines: SimpleNamespace = SimpleNamespace(
|
|
unstage_line=lambda: None,
|
|
)
|
|
godwidget: SimpleNamespace = SimpleNamespace(gapman=gapman)
|
|
order_mode: SimpleNamespace = SimpleNamespace(
|
|
godw=godwidget,
|
|
cancel_all_orders=lambda: None,
|
|
current_pp=SimpleNamespace(nav=nav),
|
|
lines=lines,
|
|
active=False,
|
|
)
|
|
source: QWidget = QWidget()
|
|
source.order_mode = order_mode
|
|
source._chart = ds.hist_chart
|
|
source.linked = SimpleNamespace(
|
|
cursor=SimpleNamespace(in_query_mode=False),
|
|
)
|
|
source.setMouseMode = lambda mode: None
|
|
|
|
async def drive_key_event() -> None:
|
|
'''
|
|
Run the production input context around one key event.
|
|
|
|
'''
|
|
async with ChartView.open_async_input_handler(
|
|
source,
|
|
dss={FQME: ds},
|
|
):
|
|
key_event: QKeyEvent = QKeyEvent(
|
|
QEvent.Type.KeyPress,
|
|
Qt.Key.Key_G,
|
|
Qt.KeyboardModifier.ControlModifier,
|
|
'g',
|
|
)
|
|
assert QApplication.sendEvent(source, key_event)
|
|
await wait_all_tasks_blocked()
|
|
|
|
try:
|
|
trio.run(drive_key_event)
|
|
assert len(gapman._layers) == 1
|
|
layer_key: tuple[str, int, str, float] = next(
|
|
iter(gapman._layers)
|
|
)
|
|
assert layer_key == (
|
|
'chart-local',
|
|
id(ds.hist_chart),
|
|
FQME,
|
|
60,
|
|
)
|
|
aid: int = gapman._layers[layer_key]
|
|
assert annots[aid].scene() is charts[1].widget.scene()
|
|
finally:
|
|
gapman.remove_owner('chart-local')
|
|
source.close()
|
|
source.deleteLater()
|
|
chart: _ChartStub
|
|
for chart in charts:
|
|
chart.close()
|
|
qapp.processEvents()
|