211 lines
6.6 KiB
Python
211 lines
6.6 KiB
Python
"""Incremental requests retain business rows and survive query changes/errors."""
|
|
|
|
import os
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
import pytest
|
|
from PySide6.QtCore import Qt
|
|
from PySide6.QtTest import QTest
|
|
from PySide6.QtWidgets import QApplication, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget
|
|
|
|
from doctor_workstation.ui.infinite_list import InfiniteList
|
|
from doctor_workstation.ui.widgets import get_value, page_items
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def app():
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
class Requests:
|
|
def __init__(self):
|
|
self.pending = []
|
|
|
|
def __call__(self, function, **callbacks):
|
|
self.pending.append((function, callbacks))
|
|
|
|
def finish(self, index=0, error=None):
|
|
fn, cb = self.pending.pop(index)
|
|
if error:
|
|
cb["on_error"](error)
|
|
else:
|
|
cb["on_success"](fn())
|
|
cb["on_finished"]()
|
|
|
|
|
|
def setup(app, size=2):
|
|
host = QWidget()
|
|
layout = QVBoxLayout(host)
|
|
table = QTableWidget(0, 2)
|
|
footer = InfiniteList(size)
|
|
footer.bind(table)
|
|
layout.addWidget(table)
|
|
layout.addWidget(footer)
|
|
host.resize(420, 220)
|
|
seen = []
|
|
errors = []
|
|
jobs = Requests()
|
|
|
|
def apply(result):
|
|
rows = page_items(result)
|
|
seen.append(result)
|
|
table.setRowCount(len(rows))
|
|
for i, row in enumerate(rows):
|
|
for col in range(2):
|
|
item = QTableWidgetItem(str(row["id"]))
|
|
item.setData(Qt.ItemDataRole.UserRole, row)
|
|
if col == 1:
|
|
item.setFlags(item.flags() | Qt.ItemFlag.ItemIsUserCheckable)
|
|
item.setCheckState(Qt.CheckState.Unchecked)
|
|
table.setItem(i, col, item)
|
|
|
|
return host, table, footer, seen, errors, jobs, apply
|
|
|
|
|
|
def test_append_deduplicates_and_preserves_selection_checks_metadata(app):
|
|
host, table, f, seen, errors, jobs, apply = setup(app)
|
|
pages = {
|
|
1: {"items": [{"id": 1}, {"id": 2}], "total": 4, "extend": {"scope": "mine"}},
|
|
2: {"items": [{"id": 2}, {"id": 3}], "total": 4},
|
|
3: {"items": [{"id": 4}], "total": 4},
|
|
}
|
|
f.reload(lambda p: pages[p], apply, errors.append, runner=jobs, query_key="a")
|
|
jobs.finish()
|
|
table.selectRow(1)
|
|
table.item(1, 1).setCheckState(Qt.CheckState.Checked)
|
|
f.load_more()
|
|
f.load_more()
|
|
assert len(jobs.pending) == 1
|
|
jobs.finish()
|
|
assert [r["id"] for r in f.rows] == [1, 2, 3]
|
|
assert table.currentRow() == 1 and table.item(1, 1).checkState() == Qt.CheckState.Checked
|
|
assert get_value(seen[-1], "extend.scope") == "mine"
|
|
f.load_more()
|
|
jobs.finish()
|
|
assert not f.has_more
|
|
assert table.rowCount() == 4 and not errors
|
|
host.close()
|
|
|
|
|
|
def test_changed_query_discards_old_success_and_error(app):
|
|
host, t, f, seen, errors, jobs, apply = setup(app)
|
|
f.reload(
|
|
lambda p: {"items": [{"id": 1}], "total": 1},
|
|
apply,
|
|
errors.append,
|
|
runner=jobs,
|
|
query_key="a",
|
|
)
|
|
f.reload(
|
|
lambda p: {"items": [{"id": 2}], "total": 1},
|
|
apply,
|
|
errors.append,
|
|
runner=jobs,
|
|
query_key="b",
|
|
)
|
|
jobs.finish(1)
|
|
jobs.finish(0, error=RuntimeError("stale"))
|
|
assert f.rows == [{"id": 2}] and not errors and not f.loading
|
|
f.invalidate()
|
|
assert not f.has_more
|
|
host.close()
|
|
|
|
|
|
def test_append_failure_retry_and_empty_end_are_bounded(app):
|
|
host, t, f, seen, errors, jobs, apply = setup(app)
|
|
calls = []
|
|
|
|
def fetch(p):
|
|
calls.append(p)
|
|
return {"items": [{"id": 1}, {"id": 2}] if p == 1 else [], "total": 8}
|
|
|
|
f.reload(fetch, apply, errors.append, runner=jobs, query_key="a")
|
|
jobs.finish()
|
|
f.load_more()
|
|
jobs.finish(error=RuntimeError("offline"))
|
|
assert t.rowCount() == 2 and f.page == 1 and not f.retry_button.isHidden()
|
|
f.load_more()
|
|
assert not jobs.pending
|
|
f.retry()
|
|
jobs.finish()
|
|
assert f.page == 2 and not f.has_more
|
|
assert calls == [1, 2] and len(errors) == 1
|
|
host.close()
|
|
|
|
|
|
def test_refresh_retains_loaded_prefix_until_complete(app):
|
|
host, t, f, seen, errors, jobs, apply = setup(app)
|
|
calls = []
|
|
|
|
def fetch(p):
|
|
calls.append(p)
|
|
return {"items": [{"id": p * 2 - 1}, {"id": p * 2}], "total": 6}
|
|
|
|
f.reload(fetch, apply, errors.append, runner=jobs, query_key="a")
|
|
jobs.finish()
|
|
f.load_more()
|
|
jobs.finish()
|
|
assert t.rowCount() == 4
|
|
f.reload(fetch, apply, errors.append, runner=jobs, query_key="a")
|
|
jobs.finish()
|
|
assert t.rowCount() == 4 and f.loading
|
|
jobs.finish()
|
|
assert t.rowCount() == 4 and f.page == 2
|
|
assert calls == [1, 2, 1, 2]
|
|
host.close()
|
|
|
|
|
|
def test_scroll_autofill_then_load_at_bottom(app):
|
|
host, t, f, seen, errors, jobs, apply = setup(app, size=12)
|
|
|
|
def fetch(p):
|
|
return {"items": [{"id": i} for i in range((p - 1) * 12, p * 12)], "total": 36}
|
|
|
|
host.show()
|
|
app.processEvents()
|
|
f.reload(fetch, apply, errors.append, runner=jobs, query_key="a")
|
|
jobs.finish()
|
|
app.processEvents()
|
|
QTest.qWait(60)
|
|
assert not jobs.pending
|
|
t.verticalScrollBar().setValue(t.verticalScrollBar().maximum())
|
|
QTest.qWait(60)
|
|
assert len(jobs.pending) == 1
|
|
old = t.verticalScrollBar().value()
|
|
jobs.finish()
|
|
assert t.verticalScrollBar().value() == old and t.rowCount() == 24
|
|
host.close()
|
|
|
|
|
|
def test_poll_during_slow_request_keeps_work_and_updates_consumer_callbacks(app):
|
|
host, table, footer, seen, errors, jobs, apply = setup(app)
|
|
completed = []
|
|
|
|
def fetch(page):
|
|
return {"items": [{"id": page * 2 - 1}, {"id": page * 2}], "total": 6}
|
|
|
|
footer.reload(fetch, apply, errors.append, runner=jobs, query_key="a")
|
|
footer.reload(
|
|
fetch,
|
|
apply,
|
|
errors.append,
|
|
runner=jobs,
|
|
query_key="a",
|
|
on_finished=lambda: completed.append("latest"),
|
|
)
|
|
assert len(jobs.pending) == 1
|
|
jobs.finish()
|
|
assert completed == ["latest"] and table.rowCount() == 2
|
|
footer.load_more()
|
|
footer.reload(fetch, apply, errors.append, runner=jobs, query_key="a")
|
|
assert len(jobs.pending) == 1
|
|
jobs.finish()
|
|
assert table.rowCount() == 4 and footer.page == 2
|
|
footer.reload(fetch, apply, errors.append, runner=jobs, query_key="a")
|
|
jobs.finish() # Prefix page 2 now pending.
|
|
footer.reload(fetch, apply, errors.append, runner=jobs, query_key="a")
|
|
assert len(jobs.pending) == 1
|
|
jobs.finish()
|
|
assert footer.page == 2 and not footer.loading and table.rowCount() == 4
|
|
host.close()
|