orisys-app/apps/studio/tests/test_ui_smoke.py

"""Offscreen UI smoke tests: construction, layout adaptation, theme tokens."""

from __future__ import annotations

from pathlib import Path

import numpy as np

import pytest

import ui_check
from tests.conftest import visible_descendants

RESOLUTIONS = ((1280, 720), (1600, 900), (1920, 1080))


def _show(window, size):
    window.resize(*size)
    window.show()
    from PyQt5.QtWidgets import QApplication

    QApplication.processEvents()
    return window


@pytest.mark.parametrize(("theme", "sensor_type"), [
    ("dark", "box"),
    ("light", "box"),
    ("dark", "finger"),
    ("light", "finger"),
])
def test_window_builds_in_every_theme(make_window, theme, sensor_type):
    window = _show(make_window(theme=theme, sensor_type=sensor_type), (1400, 880))
    assert window.windowTitle()
    assert window.content_splitter.count() == 4


@pytest.mark.parametrize("size", RESOLUTIONS)
def test_panels_fit_within_window(make_window, size):
    window = _show(
        make_window(sensor_type="finger", show_cal_panel=True),
        size,
    )
    sizes = window.content_splitter.sizes()
    panels = [
        window.mesh3d_panel,
        window.arrow_panel,
        window.force_plots,
        window.cal_panel,
    ]
    for panel, width in zip(panels, sizes):
        assert panel.isVisible()
        assert width >= panel.minimumWidth(), (
            f"{panel.objectName() or type(panel).__name__} too narrow: {width}"
        )
    assert window.force_plots.width() >= window.force_plots.minimumWidth()
    assert window.arrow_panel.width() >= window.arrow_panel.minimumWidth()


@pytest.mark.parametrize("size", RESOLUTIONS)
def test_children_stay_inside_their_parent(make_window, size):
    window = _show(make_window(sensor_type="finger", show_cal_panel=True), size)
    oversize = []
    for child in visible_descendants(window):
        if ui_check._inside_scroll_area(child):
            # A scroll area's content is *meant* to exceed its viewport: the
            # calibration panel and the settings area scroll instead of
            # squeezing their controls into an unreadable width.
            continue
        parent = child.parentWidget()
        if parent is None or not parent.isVisible():
            continue
        geometry = child.geometry()
        if geometry.width() > parent.width() + 1 or geometry.height() > parent.height() + 1:
            oversize.append(
                f"{type(child).__name__} {geometry.width()}x{geometry.height()}"
                f" > parent {parent.width()}x{parent.height()}"
            )
    assert not oversize, oversize


def test_theme_tokens_have_identical_keys():
    from orisys_app_shared.theme import build_qss, load_colors

    dark = set(load_colors("dark"))
    light = set(load_colors("light"))
    assert dark == light, {"dark_only": sorted(dark - light), "light_only": sorted(light - dark)}
    assert len(build_qss("dark")) > 1000
    assert len(build_qss("light")) > 1000


def test_arrow_canvas_differs_from_accent():
    """The pure-color arrow canvas must not collide with the accent pen color."""
    from orisys_app_shared.theme import load_colors

    for theme in ("dark", "light"):
        colors = load_colors(theme)
        assert colors["arrow_canvas"] != colors["accent"], theme


def test_palette_rules_hold():
    """WCAG contrast and colour-separation rules from ``tools/ui_check.py``."""
    import ui_check

    assert ui_check.audit_palette() == []


def test_flow_colormap_follows_the_theme_ramp():
    from orisys_app_shared.acquisition.arrows import (
        _lut_from_ramp,
        flow_colormap,
        hex_to_bgr,
    )
    from orisys_app_shared.theme import load_colors

    for theme in ("dark", "light"):
        stops = [s.strip() for s in load_colors(theme)["flow_ramp"].split(",")]
        lut = flow_colormap(theme)
        assert lut.shape == (256, 1, 3)
        assert tuple(int(v) for v in lut[0, 0]) == hex_to_bgr(stops[0])
        assert tuple(int(v) for v in lut[-1, 0]) == hex_to_bgr(stops[-1])

    # Malformed ramps must never raise; they fall back to the built-in ramp.
    fallback = _lut_from_ramp("not,a,ramp")
    assert fallback.shape == (256, 1, 3)


def test_brand_bar_renders_and_localizes(make_window):
    from orisys_app_shared.i18n import t

    window = _show(make_window(language="zh_CN"), (1440, 880))
    bar = window.brand_bar
    assert bar.height() >= 28
    assert bar.name.text()
    assert bar.product.text() == t("app.title")

    zh_tagline = bar.tagline.text()
    window._set_language("en_US")
    assert bar.tagline.text() != zh_tagline
    assert bar.name.text() == "ORISYS"

    logo = bar.logo.pixmap()
    if logo is not None and not logo.isNull():
        image = logo.toImage()
        alphas = [
            image.pixelColor(x, y).alpha()
            for y in range(image.height())
            for x in range(image.width())
        ]
        assert any(a < 32 for a in alphas)
        assert any(a >= 200 for a in alphas)


def test_brand_bar_background_comes_from_tokens(make_window):
    from orisys_app_shared.theme import load_colors

    window = _show(make_window(), (1440, 880))
    for theme in ("dark", "light"):
        window._set_theme(theme)
        colors = load_colors(theme)
        assert colors["brand_bar_bg"] != colors["window_bg"], theme
        assert window.brand_bar._theme == theme


def test_disabled_buttons_use_muted_text_not_hover_background():
    from orisys_app_shared.theme import build_qss, load_colors

    for theme in ("dark", "light"):
        colors = load_colors(theme)
        qss = build_qss(theme)
        assert f"color: {colors['text_muted']}" in qss, theme
        # The old rule painted disabled buttons with the brighter hover color.
        assert f"background-color: {colors['button_hover']};\n    border-color: {colors['panel_border']}" not in qss


def test_splitter_sizes_round_trip_through_prefs(make_window):
    window = _show(make_window(), (1440, 880))
    window.content_splitter.setSizes([300, 420, 520, 0])
    window._persist_prefs()
    stored = window.prefs["splitter_sizes"]
    assert len(stored) == 4
    assert stored[2] > stored[0]


def test_status_chip_tracks_phase(make_window):
    window = _show(make_window(), (1400, 880))
    assert window._session_phase == "idle"
    window._set_phase("running")
    assert window.status_chip.text()
    assert "status_running" not in window.status_chip.text()
    window._set_phase("nonexistent-phase")
    assert window._session_phase == "idle"


def test_run_tooltip_explains_disabled_reason(make_window):
    window = _show(make_window(source_mode="video", video_path=""), (1400, 880))
    assert not window.toolbar.btn_run.isEnabled()
    assert window.toolbar.btn_run.toolTip()

    window.toolbar.set_video_path("C:/tmp/demo.mp4")
    window._update_run_enabled()
    assert window.toolbar.btn_run.isEnabled()


def test_toolbar_run_button_is_labelled(make_window):
    window = _show(make_window(), (1400, 880))
    label = window.toolbar.btn_run.text()
    assert label.strip()
    assert label != "â–¶"
    window.toolbar.set_run_checked(True)
    assert window.toolbar.btn_run.text() != label


def test_stop_cancels_a_hung_connect(make_window):
    """Stop must stay live and take effect instantly while connecting."""
    import time

    import ui_check
    from orisys_app_shared.i18n import t

    window = _show(make_window(sensor_type="box", source_mode="camera"), (1440, 880))
    stuck = ui_check._StuckWorker()
    window._ensure_workers = lambda: None
    window._box_worker = stuck
    window._finger_worker = stuck

    window.toolbar.btn_run.setChecked(True)
    assert window._session_phase == "connecting"
    assert window.toolbar.btn_run.isEnabled(), "Stop is disabled while connecting"
    assert window.toolbar.btn_run.text() == t("toolbar.cancel_label")
    assert window.act_stop.isEnabled()

    started = time.perf_counter()
    window.toolbar.btn_run.setChecked(False)
    elapsed = time.perf_counter() - started
    assert elapsed < 0.5, f"Stop blocked the GUI for {elapsed:.2f}s"
    assert not window._running
    assert stuck.thread_still_owned, "Stop abandoned the still-running worker thread"


def test_escape_cancels_a_hung_connect(make_window):
    import time

    import ui_check

    window = _show(make_window(sensor_type="box", source_mode="camera"), (1440, 880))
    stuck = ui_check._StuckWorker()
    window._ensure_workers = lambda: None
    window._box_worker = stuck
    window._finger_worker = stuck

    window.toolbar.btn_run.setChecked(True)
    assert window.act_stop.shortcut().toString() == "Esc"
    started = time.perf_counter()
    window._on_stop_shortcut()
    assert time.perf_counter() - started < 0.5
    assert not window._running


def test_theme_switch_updates_panels(make_window):
    window = _show(make_window(), (1400, 880))
    window._set_theme("light")
    assert window._theme == "light"
    window._set_theme("dark")
    assert window._theme == "dark"


def test_cal_panel_records_and_lists_samples(make_window):
    window = _show(make_window(), (1400, 880))
    cal = window.cal_panel
    cal.set_running(True)
    assert cal.btn_record.isEnabled()
    assert cal.btn_record.text()
    assert cal.table.columnCount() == 6, "the diagnostic columns changed"

    _add_calibration_point(window, load=1.0, raw=100.2)
    _add_calibration_point(window, load=2.0, raw=400.0)  # off-trend
    _add_calibration_point(window, load=3.0, raw=300.6)
    assert cal.table.rowCount() == 3
    assert not cal.empty_hint.isVisible()

    headers = [
        cal.table.horizontalHeaderItem(i).text() for i in range(cal.table.columnCount())
    ]
    from orisys_app_shared.i18n import t

    assert headers == [
        t("cal.col_weight"),
        t("cal.col_fn"),
        t("cal.col_fx"),
        t("cal.col_fy"),
        t("cal.col_stability"),
        t("cal.col_state"),
    ]


def _add_calibration_point(window, *, load: float, raw: float, samples: int = 40):
    """Drive one point through the recorder and into the panel's table."""
    from orisys_app_shared.force.session import SessionSample

    window.cal_recorder.start(channel="fn", reference_load_n=load, duration_s=1.0)
    for index in range(samples):
        window.cal_recorder.feed(
            SessionSample(
                frame_id=index,
                t_rel_s=index / 100.0,
                fn_raw=raw,
                fx_raw=0.4,
                fy_raw=-0.2,
                valid=True,
            )
        )
    point = window.cal_recorder.finish_now()
    window.cal_panel.on_record_finished(point)
    return point


def test_calibration_diagnostics_drive_the_fit(make_window):
    """The diagnostics must describe the model that would be exported.

    The export builds a piecewise-linear curve, which passes through every point
    it is built from — so the training residual is zero by construction and an
    off-trend point can only show up as a *validation* error. A private
    least-squares residual would have described a model nobody applies.
    """
    window = _show(make_window(), (1400, 880))
    window.cal_panel.set_running(True)
    _add_calibration_point(window, load=1.0, raw=100.2)
    _add_calibration_point(window, load=2.0, raw=400.0)
    _add_calibration_point(window, load=3.0, raw=300.6)
    _add_calibration_point(window, load=5.0, raw=500.9)

    recorder = window.cal_recorder
    fit = recorder.fit("fn")
    assert fit.usable and fit.n_points == 4
    assert fit.method == "piecewise_linear", fit.method
    assert fit.breakpoints == 4, fit.breakpoints
    assert fit.max_residual_n <= 1e-9, "the exported model misses its own points"
    assert window.cal_panel.fit_label.text()

    # Hold the off-trend point (load 2.0 at raw 400) out: the exported model is
    # then built without it and the misfit is measurable.
    window.cal_panel.table.setCurrentCell(1, 0)
    window.cal_panel._on_toggle_validation()
    held_out = recorder.fit("fn")
    assert held_out.validation_points == 1
    assert held_out.validation_error_n > 0.05, held_out.validation_error_n
    assert recorder.points[1].is_validation
    assert np.isnan(recorder.points[1].residual_n), "validation point entered the fit"

    window.cal_panel._on_toggle_validation()
    window.cal_panel._on_toggle_excluded()
    assert recorder.points[1].excluded
    # An excluded point leaves the export store but stays on record.
    assert len(window.cal_panel._store.channel("fn")) == 3

    # The exported model no longer passes through the excluded point.
    from orisys_app_shared.force.calibration_io import (
        build_export_model,
        predict_channel,
    )

    store = window.cal_panel._store
    model = build_export_model({key: store.channel(key) for key in ("fn", "fx", "fy")})
    assert abs(predict_channel(model, "fn", 400.0) - 2.0) > 0.05

    # The store, the label and the export are the same preparation: the panel
    # must not derive the model twice and show numbers for one of them.
    prepared = window.cal_panel._prepared
    assert prepared is not None and prepared.ok
    assert list(store.channel("fn")) == list(prepared.samples["fn"])
    assert prepared.points_in_fit("fn") == 3


def test_the_panel_shows_the_numbers_of_the_file_it_exports(make_window, record_root):
    """Displayed diagnostics must match the written coefficients.

    A repeated reference load makes this sharp: the two samples of one load must
    be merged before the build, so the exported curve and the panel's residual
    and validation figures all come from that single prepared model.
    """
    import json
    from pathlib import Path

    import orisys.calibration as sdk_calibration

    from orisys_app_shared.force.calibration_io import build_force_calibration_v2

    window = _show(make_window(), (1400, 880))
    window.cal_panel.set_running(True)
    window.cal_panel.set_meta_context(sensor_type="box", config_ref=None)
    _add_calibration_point(window, load=1.0, raw=100.0)
    _add_calibration_point(window, load=2.0, raw=200.0)
    _add_calibration_point(window, load=2.0, raw=240.0)  # a deliberate repeat
    _add_calibration_point(window, load=4.0, raw=400.0)
    _add_calibration_point(window, load=3.0, raw=290.0)

    panel = window.cal_panel
    panel.table.setCurrentCell(4, 0)
    panel._on_toggle_validation()
    panel._rebuild_store()
    panel._refresh_fit_label()
    prepared = panel._prepared
    assert prepared is not None and prepared.ok
    assert len(prepared.samples["fn"]) == 3, "the repeated load was not merged"

    # Write the document the export would write, load it back with the SDK and
    # check that its predictions are the prepared ones.
    document, _warnings = build_force_calibration_v2(prepared)
    path = Path(record_root) / "force_calibration.json"
    path.write_text(json.dumps(document), encoding="utf-8")
    loaded = sdk_calibration.load(path, strict=True)

    for point in window.cal_recorder.points:
        raw = float(point.loaded_stats().mean)
        predicted = float(loaded.get("fn").convert(raw))
        assert predicted == pytest.approx(prepared.predict("fn", raw), abs=1e-9)

    # ...and the label describes that same model.
    label = panel.fit_label.text()
    assert "分段线性" in label or "piecewise" in label, label
    assert "3" in label
    assert prepared.max_validation_error("fn") > 0.0


def test_calibration_point_dialog_shows_three_axes(make_window):
    window = _show(make_window(), (1400, 880))
    _add_calibration_point(window, load=1.0, raw=100.2)
    point = window.cal_recorder.points[0]
    assert set(point.trace) == {"fn", "fx", "fy"}
    assert point.trace_dt_s > 0

    from orisys_studio.ui.cal_point_dialog import CalibrationPointDialog

    dialog = CalibrationPointDialog(point, theme="dark")
    assert dialog.windowTitle()
    assert len(dialog.plot_widget.ci.items) == 3, "not one trace per axis"
    dialog.close()


def _write_session_file(root, *, sensor_type, sn, config_ref, points):
    import json

    from orisys_app_shared.force.calibration_session import build_session_document

    document = build_session_document(
        points,
        meta={"sensor_type": sensor_type, "sn": sn, "config_ref": config_ref},
    )
    path = Path(root) / f"calibration_session_{sensor_type}.json"
    path.write_text(json.dumps(document), encoding="utf-8")
    return path


def test_opening_device_a_record_keeps_device_a_identity(make_window, record_root):
    """Device A's record must not be restamped with device B's identity."""
    import json
    from unittest.mock import patch as _patch

    import orisys_studio.ui.cal_panel as cal_panel_module
    from orisys_app_shared.force.calibration_io import build_force_calibration_v2

    window = _show(make_window(), (1400, 880))
    panel = window.cal_panel
    # A record taken on device A, with a full audit series.
    session_points = [
        _add_calibration_point(window, load=1.0, raw=100.0),
        _add_calibration_point(window, load=3.0, raw=300.0),
    ]
    source = _write_session_file(
        record_root,
        sensor_type="box",
        sn="SN-A",
        config_ref="D:/lab/deviceA/sensor.json",
        points=session_points,
    )
    window.cal_recorder.clear_points()

    # ...and the live context is device B.
    panel.set_meta_context(sensor_type="finger", config_ref="D:/lab/deviceB/sensor.json")
    panel.sn_edit.setText("SN-B")

    panel._on_open_record(str(source))
    assert panel.sn_edit.text() == "SN-A", "the record's SN was overwritten"
    meta = panel._export_meta()
    assert meta["sensor_type"] == "box"
    assert meta["sn"] == "SN-A"
    assert meta["config_ref"] == "D:/lab/deviceA/sensor.json"
    assert meta["source_record"].endswith(source.name)

    # Saving the record again still says A.
    import orisys_studio.calibration_service as cal_service_module

    with _patch.object(cal_panel_module, "calibration_export_dir", lambda: Path(record_root)):
        with _patch.object(
            cal_service_module, "calibration_export_dir", lambda: Path(record_root)
        ):
            with _patch.object(cal_panel_module.QMessageBox, "information", lambda *a, **k: None):
                panel._on_save_record()
    saved = sorted(Path(record_root).glob("calibration_session_box_*.json"))
    assert saved, "the record was not saved under device A's type"
    saved_meta = json.loads(saved[-1].read_text(encoding="utf-8"))["meta"]
    assert saved_meta["sn"] == "SN-A"
    assert saved_meta["sensor_type"] == "box"
    assert saved_meta["config_ref"] == "D:/lab/deviceA/sensor.json"

    # ...and so does the exported coefficient document.
    document, _warnings = build_force_calibration_v2(panel._prepared)
    assert document["meta"]["sensor_type"] == "box"
    assert document["meta"]["sn"] == "SN-A"
    assert document["meta"]["config_ref"] == "D:/lab/deviceA/sensor.json"


def test_a_reopened_point_can_be_inspected_from_its_samples(make_window, record_root):
    """A record with samples but no cached trace still opens its diagnostics."""
    window = _show(make_window(), (1400, 880))
    panel = window.cal_panel
    points = [
        _add_calibration_point(window, load=1.0, raw=100.0),
        _add_calibration_point(window, load=2.0, raw=200.0),
    ]
    import json

    from orisys_app_shared.force.calibration_session import build_session_document

    document = build_session_document(points, meta={"sensor_type": "box"})
    for entry in document["points"]:
        entry["trace"] = {}  # only the audit series survived
        entry["trace_dt_s"] = 0.0
    source = Path(record_root) / "no_trace.json"
    source.write_text(json.dumps(document), encoding="utf-8")

    panel._on_open_record(str(source))
    reopened = window.cal_recorder.points[0]
    assert reopened.samples, "the reopened point lost its samples"
    dialog = panel.point_dialog(reopened)
    assert dialog is not None, "the reopened point could not be inspected"
    assert len(dialog.plot_widget.ci.items) == 3
    dialog.close()


def test_applying_another_devices_calibration_is_flagged(make_window, record_root):
    """The apply path states a sensor/config mismatch instead of silently applying."""
    import json

    from orisys_app_shared.force.calibration_io import build_export_model, write_force_calibration
    from orisys_app_shared.force.calibration_io import CalSample, CalSessionStore

    window, stuck = _running_window(make_window)
    store = CalSessionStore()
    store.aggregate("fn", CalSample(weight_n=0.0, raw_mean=0.0, n=10))
    store.aggregate("fn", CalSample(weight_n=5.0, raw_mean=500.0, n=10))
    model = build_export_model({key: store.channel(key) for key in ("fn", "fx", "fy")})
    document = model.to_dict()
    document["identity"] = {"sensor_type": "box", "sn": "SN-A"}
    document["pipeline"] = {"config_ref": "D:/lab/deviceA/sensor.json"}
    path = Path(record_root) / "deviceA_calibration.json"
    path.write_text(json.dumps(document), encoding="utf-8")

    # The window is connected to a finger device.
    mismatch = window.calibration_target_mismatch(
        str(path), sensor_type="finger", config_ref="D:/lab/deviceB/sensor.json"
    )
    assert any("sensor_type" in reason for reason in mismatch), mismatch
    assert any("config_ref" in reason for reason in mismatch), mismatch
    assert not window.calibration_target_mismatch(
        str(path), sensor_type="box", config_ref="D:/lab/deviceA/sensor.json"
    )
    _finish(window, stuck)


def test_force_plots_render_demo_data(make_window):
    window = _show(make_window(), (1440, 880))
    for i in range(60):
        window.force_plots.push(i / 30.0, float(i), -1.0, 1.0, unit="N")
    from PyQt5.QtWidgets import QApplication

    QApplication.processEvents()
    assert window.force_plots.history.sample_count > 0
    assert all(label.text() for label in window.force_plots.value_labels)


# --------------------------------------------------------------- record channel


def _running_window(make_window):
    """A window in the ``running`` phase whose worker never opens a device.

    The Qt timer is stopped deliberately: these tests drive the display path by
    calling ``_on_run_tick`` themselves, and a concurrent 16 ms timer would race
    the assertions (and outlive the window on teardown).
    """
    window = _show(make_window(sensor_type="box", source_mode="camera"), (1440, 880))
    stuck = ui_check._StuckWorker()
    window._ensure_workers = lambda: None
    window._box_worker = stuck
    window._finger_worker = stuck
    window._start_run()
    window._run_timer.stop()
    window._set_phase("running")
    return window, stuck


def _finish(window, stuck):
    window._stop_run(reason="user_stop")
    stuck.release()


def test_record_channel_does_not_depend_on_the_ui_timer(make_window):
    window, stuck = _running_window(make_window)
    for index in range(90):
        stuck.emit(ui_check._fake_packet(index))

    assert window.session_store.count == 90, "samples were lost without a UI tick"
    _finish(window, stuck)


def test_invalid_samples_are_excluded_not_zeroed(make_window):
    import math

    window, stuck = _running_window(make_window)
    for index in range(9):
        stuck.emit(ui_check._fake_packet(index))
    stuck.emit(ui_check._fake_packet(99, valid=False, reason="read failed"))

    slice_ = window.session_store.data()
    assert slice_.valid.tolist()[-1] is False
    assert math.isnan(slice_.raw[-1][0]), "a failed readout became a number"
    assert sum(1 for _ in window.session_store.iter_valid("raw")) == 9
    _finish(window, stuck)


def test_calibration_sampling_survives_a_hidden_panel(make_window):
    window, stuck = _running_window(make_window)
    window._show_cal = False
    window._apply_panel_visibility()
    assert not window.cal_panel.isVisible()

    window.cal_recorder.start(channel="fn", reference_load_n=2.0, duration_s=1.0)
    for index in range(40):
        stuck.emit(ui_check._fake_packet(index))
    point = window.cal_recorder.finish_now()

    assert point is not None
    assert point.n_valid == 40, "hiding the panel stopped the sampling"
    assert point.status == "ok"
    _finish(window, stuck)


def test_record_button_stays_a_cancel_affordance(make_window):
    from orisys_app_shared.i18n import t

    window, stuck = _running_window(make_window)
    window.cal_panel.set_running(True)
    window.cal_panel._on_record_clicked()

    assert window.cal_panel._recording
    assert window.cal_panel.btn_record.isEnabled(), "the recording cannot be stopped"
    assert window.cal_panel.btn_record.text() == t("cal.recording_stop")

    window.cal_panel._on_record_clicked()
    assert not window.cal_panel._recording
    _finish(window, stuck)


def test_unit_switch_rebuilds_from_the_session_store(make_window):
    import math

    window, stuck = _running_window(make_window)
    window.toolbar.units_combo.setCurrentIndex(1)  # raw
    window.force_plots.reset()
    for index in range(60):
        stuck.emit(ui_check._fake_packet(index))
        window._on_run_tick()

    state = window.force_plots.history.state()
    assert float(state.values[:, 0].max()) > 5.0, "raw values were not plotted"

    window.toolbar.units_combo.setCurrentIndex(0)  # newton
    assert window.force_plots._unit == "N"
    newtons = window.force_plots.history.state().values[:, 0]
    finite = [v for v in newtons if math.isfinite(v)]
    assert finite and max(finite) < 5.0, "the buffer still held raw values in N mode"
    _finish(window, stuck)


def test_history_keeps_negative_fn(make_window):
    window, stuck = _running_window(make_window)
    window.force_plots.set_fixed_newton(True)
    for index in range(20):
        window.force_plots.push(index / 30.0, -2.5, 0.0, 0.0, unit="N")
    window.force_plots.push(1.0, -3.0, 0.0, 0.0, unit="N")

    state = window.force_plots.history.state()
    assert float(state.values[:, 0].min()) < 0.0, (
        "negative FN was clamped before entering the buffer"
    )
    _finish(window, stuck)


def test_the_live_axis_is_session_relative_and_survives_a_unit_switch(make_window):
    """The live buffer, the store and a rebuilt history share one time base.

    Feeding the wall clock into the plot put the live axis ~1.7e9 s away from the
    session store's, so a unit switch (which rebuilds from the store) jumped the
    X axis and the chart looked like it had lost its history.
    """
    wall = 1_700_000_000.0
    window, stuck = _running_window(make_window)
    window.toolbar.units_combo.setCurrentIndex(0)  # newton
    window.force_plots.reset()
    for index in range(90):
        stuck.emit(
            ui_check._fake_packet(
                index, timestamp=wall + index / 30.0, mono_t=index / 30.0
            )
        )
        window._on_run_tick()

    before = window.force_plots.history.state()
    assert before is not None
    end_before = float(before.times[-1])
    assert end_before < 60.0, (
        f"the live X axis is not session-relative: it ends at {end_before}"
    )
    assert abs(end_before - 89 / 30.0) < 0.5, end_before

    for units_index in (1, 0, 1, 0):  # raw -> N -> raw -> N
        window.toolbar.units_combo.setCurrentIndex(units_index)

    after = window.force_plots.history.state()
    assert after is not None, "the unit switch emptied the rebuilt history"
    assert abs(float(after.times[-1]) - end_before) < 1.0, (
        f"the X axis jumped across a unit switch: {end_before} -> {after.times[-1]}"
    )
    _finish(window, stuck)


def test_a_single_calibrated_axis_is_shown_and_the_others_are_unavailable(make_window):
    """Only FN calibrated: FN in newtons, FX/FY explicitly unavailable."""
    import math

    from orisys_app_shared.i18n import t

    window, stuck = _running_window(make_window)
    window.toolbar.units_combo.setCurrentIndex(0)  # newton
    window.force_plots.reset()
    for index in range(10):
        stuck.emit(
            ui_check._fake_packet(
                index,
                calibrated_axes=("fn",),
                shear_n=float("nan"),
            )
        )
        window._on_run_tick()

    plots = window.force_plots
    assert plots._available == (True, False, False), plots._available
    assert plots.value_labels[0].text() != t("chart.value_unavailable")
    for index in (1, 2):
        assert plots.value_labels[index].text() == t("chart.value_unavailable"), (
            "an axis with no curve must not show a number"
        )
    state = plots.history.state()
    assert math.isfinite(float(state.values[:, 0].max()))
    assert not any(math.isfinite(v) for v in state.values[:, 1]), (
        "FX carried a value in newton mode without a curve"
    )
    _finish(window, stuck)


def test_a_slow_close_is_not_reported_as_a_finished_recording(make_window):
    """Stopping must not present an unclosed file as a finished experiment."""
    import threading as _threading
    from unittest.mock import patch as _patch

    from orisys_app_shared.i18n import t
    from orisys_app_shared.record.experiment import _SessionWriter

    window, stuck = _running_window(make_window)
    window.toolbar.btn_record.setChecked(True)
    stuck.emit(ui_check._fake_packet(0))
    window._on_run_tick()

    started = _threading.Event()
    release = _threading.Event()
    original = _SessionWriter.write_sample

    def slow(self, *args):
        started.set()
        release.wait(10.0)
        return original(self, *args)

    with _patch.object(_SessionWriter, "write_sample", slow):
        stuck.emit(ui_check._fake_packet(1))
        window._on_run_tick()
        assert started.wait(5.0), "the writer never picked up the frame"

        window._stop_recording(wait=False)
        assert window.recorder.report is None, "the stop claimed a finished recording"
        assert window._record_report_pending, "the pending report was not tracked"
        assert window._record_last_report is None
        assert not window.toolbar.btn_record.isChecked(), (
            "the record button still looks active while the file closes"
        )
        assert window.statusBar().currentMessage() == t("record.closing")

        release.set()
        deadline = __import__("time").monotonic() + 10.0
        while window.recorder.report is None and __import__("time").monotonic() < deadline:
            __import__("time").sleep(0.01)
        window._on_run_tick()

    report = window._record_last_report
    assert report is not None and report.completed, "the terminal report was never published"
    assert not window._record_report_pending
    _finish(window, stuck)


def test_a_failed_recording_is_consumed_even_when_it_is_not_running(make_window):
    """A recording that died on its own still has a report to publish."""
    import time as _time
    from unittest.mock import patch as _patch

    from orisys_app_shared.record.experiment import _SessionWriter

    window, stuck = _running_window(make_window)
    window.toolbar.btn_record.setChecked(True)
    assert window.recorder.is_recording

    with _patch.object(_SessionWriter, "write_sample", side_effect=OSError("disk full")):
        for index in range(8):
            stuck.emit(ui_check._fake_packet(index))
            window._on_run_tick()
        deadline = _time.monotonic() + 3.0
        while window.recorder.stats().write_errors < 5 and _time.monotonic() < deadline:
            _time.sleep(0.01)
        assert window.recorder.stats().failed
        assert not window.recorder.is_recording

        # ``is_recording`` is already False: the stop must still consume the report.
        window._stop_recording()
        report = window._record_last_report
        assert report is not None and report.stats.failed
        assert not window.toolbar.btn_record.isChecked()
        assert window._session_phase == "error"
    _finish(window, stuck)


def test_the_depth_request_belongs_to_the_recording_it_started_in(make_window, record_root):
    """Opening another recording must invalidate the request still running."""
    import threading as _threading
    import time as _time

    from PyQt5.QtWidgets import QApplication

    from orisys_app_shared.record.depth import DepthResult
    from orisys_studio.replay import ReplayController

    window = _show(make_window(), (1200, 700))
    replay = ReplayController()
    replay.open_recording(Path(record_root))
    first_serial = replay._record_serial

    started = _threading.Event()
    release = _threading.Event()
    closed: list[str] = []

    class _Blocking:
        def preflight(self) -> str:
            return ""

        def depth_at(self, _position, **_kwargs):
            started.set()
            release.wait(10.0)
            return DepthResult(
                depth=np.ones((2, 2)), frame_id=0, t_rel_s=0.0, record_id="old"
            )

        def close(self) -> None:
            closed.append("closed")

    blocking = _Blocking()
    replay._recomputer = blocking
    replay.request_depth()
    assert started.wait(5.0), "the depth request never started"

    # Another recording is opened while the request is in flight.
    replay.open_recording(Path(record_root))
    assert replay._record_serial == first_serial + 1
    assert replay._recomputer is not blocking, "the processor was not replaced"
    assert closed == [], "the UI thread closed a processor a worker was reading"

    release.set()
    deadline = _time.monotonic() + 10.0
    while not closed and _time.monotonic() < deadline:
        QApplication.processEvents()
        _time.sleep(0.01)
    assert closed == ["closed"], "the retired processor was never closed"
    assert replay._depth_thread is not None and not replay._depth_thread.is_alive()

    # The answer belongs to the previous recording, so it is dropped, not drawn.
    replay._poll_depth()
    QApplication.processEvents()
    assert replay._depth_ok is False
    assert replay.depth_image() is None, "an answer for another recording was drawn"
    replay.close()
    window.close()


def test_a_busy_depth_processor_is_not_closed_by_the_ui_thread(make_window, record_root):
    import threading as _threading
    import time as _time

    from PyQt5.QtWidgets import QApplication

    from orisys_app_shared.record.depth import DepthResult
    from orisys_studio.replay import ReplayController

    window = _show(make_window(), (1200, 700))
    replay = ReplayController()
    replay.open_recording(Path(record_root))
    started = _threading.Event()
    release = _threading.Event()
    closed: list[str] = []

    class _Blocking:
        def preflight(self) -> str:
            return ""

        def depth_at(self, _position, **_kwargs):
            started.set()
            release.wait(10.0)
            return DepthResult(depth=np.ones((2, 2)), frame_id=0, t_rel_s=0.0)

        def close(self) -> None:
            closed.append("closed")

    replay._recomputer = _Blocking()
    replay.request_depth()
    assert started.wait(5.0)
    replay.close()  # closing must not free a processor in use
    QApplication.processEvents()
    assert closed == [], "the controller closed a processor that was still reading"
    release.set()
    deadline = _time.monotonic() + 10.0
    while not closed and _time.monotonic() < deadline:
        QApplication.processEvents()
        _time.sleep(0.01)
    assert closed == ["closed"], "the owning thread never closed the processor"
    window.close()


def test_a_confirmed_calibration_is_snapshotted_into_the_recording(make_window, record_root):
    """The recording keeps the coefficients themselves, not only their hash."""
    import json
    from pathlib import Path

    import orisys_studio.app_window as app_module
    from orisys_app_shared.i18n import t

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: record_root
    window.toolbar.btn_record.setChecked(True)
    assert window.recorder.is_recording

    coefficients = Path(record_root) / "force_calibration_box.json"
    coefficients.write_text('{"unit": "N", "channels": {}}', encoding="utf-8")
    version = "deadbeefcafe1234"
    window._on_calibration_applied(
        str(coefficients), version, 21, 0.7, int(window.session_store.session_id)
    )
    window._drain_calibration_events()

    directory = window.recorder.directory
    assert directory is not None
    snapshots = sorted(path.name for path in (directory / "calibration").iterdir())
    assert snapshots == [f"force_calibration_{version[:12]}.json"], snapshots
    assert (directory / "calibration" / snapshots[0]).read_text(
        encoding="utf-8"
    ) == '{"unit": "N", "channels": {}}'
    assert window._active_calibration_id == version
    assert window.session_store.calibration_id == version

    window._stop_recording()
    meta = json.loads((directory / "session.json").read_text(encoding="utf-8"))
    entries = meta["calibration_snapshots"]
    assert entries and entries[0]["version"] == version
    assert entries[0]["file"] == f"calibration/{snapshots[0]}"
    assert entries[0]["frame_id"] == 21
    event = next(e for e in meta["events"] if e["kind"] == "calibration_applied")
    assert event["frame_id"] == 21 and event["mono_t"] == pytest.approx(0.7)
    assert snapshots[0] in event["detail"]
    _finish(window, stuck)


def test_a_confirmed_absence_of_calibration_does_not_keep_the_old_version(make_window):
    """An empty version is the honest answer; it must not fall back to the path."""
    from orisys_app_shared.i18n import t

    window = _show(make_window(), (1400, 880))
    window._active_calibration_id = "previous-hash"
    window._on_calibration_applied("/tmp/whatever.json", "", 5, 0.2, 1)
    window._drain_calibration_events()
    assert window._active_calibration_id == "", (
        "an uncalibrated device kept claiming the previous coefficients"
    )
    assert window.statusBar().currentMessage() == t("cal.applied_none")


def test_a_zeroing_is_recorded_only_after_the_worker_confirms_it(make_window, record_root):
    import json

    import orisys_studio.app_window as app_module

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: record_root
    window.toolbar.btn_record.setChecked(True)
    stuck.emit(ui_check._fake_packet(3))
    window._on_run_tick()

    window._on_reset()
    assert stuck.reset_requests == 1, "the zeroing was not requested"
    # The stub confirms from the acquisition side, as a real device does.
    stuck.on_reset_applied = window._on_reset_applied
    window._on_reset()
    window._on_run_tick()
    window._stop_recording()

    directory = window.recorder.directory
    meta = json.loads((directory / "session.json").read_text(encoding="utf-8"))
    zeroing = [e for e in meta["events"] if e["kind"] == "zeroing"]
    assert len(zeroing) == 1, f"unexpected zeroing events: {zeroing}"
    assert zeroing[0]["frame_id"] == 3, zeroing[0]
    assert zeroing[0]["mono_t"] is not None
    _finish(window, stuck)


def test_depth_view_reports_missing_data_and_fixes_its_scale(make_window):
    import numpy as np

    window, stuck = _running_window(make_window)
    window.arrow_panel.set_view_mode("depth")

    window._last_rendered_view_mode = None
    stuck.emit(ui_check._fake_packet(0, depth=None))
    window._on_run_tick()
    from orisys_app_shared.i18n import t

    assert window.arrow_panel.image_label.text() == t("panels.depth_unavailable"), (
        "a missing depth map silently fell back to the processed frame"
    )

    window._last_rendered_view_mode = None
    stuck.emit(ui_check._fake_packet(1, depth=np.linspace(0.0, 1.0, 64).reshape(8, 8)))
    window._on_run_tick()
    latched = window.arrow_panel.depth_scale_label.text()
    assert latched

    window._last_rendered_view_mode = None
    stuck.emit(ui_check._fake_packet(2, depth=np.linspace(0.0, 9.0, 64).reshape(8, 8)))
    window._on_run_tick()
    assert window.arrow_panel.depth_scale_label.text() == latched, (
        "the depth colour scale re-fitted on every frame"
    )
    _finish(window, stuck)


# ------------------------------------------------------------- experiment record

#: Recording scratch goes under the app's gitignored ``outputs`` tree rather than
#: the system temp directory, so the tests also run under a file sandbox that
#: only permits writes inside the workspace.
RECORD_SCRATCH = Path(__file__).resolve().parents[1] / "outputs" / "_test_record"


@pytest.fixture
def record_root():
    import shutil

    shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)
    RECORD_SCRATCH.mkdir(parents=True, exist_ok=True)
    yield RECORD_SCRATCH
    shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)


@pytest.fixture
def replay_recording(record_root):
    """One short finished recording (video + forces + frames + session.json)."""
    from orisys_app_shared.record import ExperimentRecorder, RecordConfig

    recorder = ExperimentRecorder(RecordConfig(video=True, fps=30.0, segment_seconds=30.0))
    recording = recorder.start(
        root=record_root,
        session_id=1,
        metadata={
            "device": {
                "sensor_type": "box",
                "config_ref": str(record_root / "sensor.json"),
                "config_sha256": "recorded",
            },
            "versions": {"sdk": "0.5.1"},
            "depth": {"stored": False, "replay": "recompute"},
        },
    )
    for index in range(60):
        recorder.on_packet(ui_check._fake_packet(index))
    recorder.stop()
    yield Path(recording)


def test_recording_writes_the_documented_experiment_layout(make_window, record_root):
    import csv
    import json

    import orisys_studio.app_window as app_module
    from orisys_app_shared.i18n import t

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: record_root
    assert window.toolbar.btn_record.isEnabled()
    assert window.toolbar.btn_record.text() == t("record.start_label")

    window.toolbar.btn_record.setChecked(True)
    assert window.recorder.is_recording, "the recording did not start"
    assert window.toolbar.btn_record.text() == t("record.stop_label")
    assert not window.toolbar.record_mode_combo.isEnabled()

    for index in range(20):
        stuck.emit(ui_check._fake_packet(index))
        window._on_run_tick()
    window.toolbar.btn_record.setChecked(False)

    report = window._record_last_report
    assert report is not None and report.ok, getattr(report, "problems", None)
    names = sorted(path.name for path in report.directory.iterdir())
    assert names == ["events.csv", "forces.csv", "frames.csv", "raw_0001.mp4", "session.json"]

    with (report.directory / "forces.csv").open(encoding="utf-8") as handle:
        forces = list(csv.DictReader(handle))
    with (report.directory / "frames.csv").open(encoding="utf-8") as handle:
        frames = list(csv.DictReader(handle))
    assert len(forces) == len(frames) == 20
    assert [f["frame_id"] for f in frames] == [r["frame_id"] for r in forces]
    assert [f["t_rel_s"] for f in frames] == [r["t_rel_s"] for r in forces]

    meta = json.loads((report.directory / "session.json").read_text(encoding="utf-8"))
    assert meta["ended_at"]
    assert meta["format_version"]
    assert meta["stats"]["frames"] == 20
    _finish(window, stuck)


def test_recording_leaves_unavailable_values_empty(make_window, record_root):
    import csv

    import orisys_studio.app_window as app_module

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: record_root
    window.toolbar.btn_record.setChecked(True)
    stuck.emit(ui_check._fake_packet(0))
    window._on_run_tick()
    stuck.emit(ui_check._fake_packet(1, valid=False, reason="read failed"))
    window._on_run_tick()
    window.toolbar.btn_record.setChecked(False)

    with (window._record_last_report.directory / "forces.csv").open(encoding="utf-8") as handle:
        rows = list(csv.DictReader(handle))
    failed = rows[-1]
    assert failed["valid"] == "0"
    assert failed["FN_raw"] == "" and failed["FN_N"] == "", (
        "a failed readout was persisted as a number"
    )
    _finish(window, stuck)


# ------------------------------------------------------- top rail / settings


def test_top_area_stays_compact(make_window):
    """The rail is read at a glance; it is not a place to stack configuration."""
    from orisys_app_shared.i18n import t

    window = _show(make_window(), (1440, 880))
    collapsed = window.brand_bar.height() + window.toolbar.height()
    assert collapsed <= 78, f"the collapsed top area is {collapsed}px tall"
    assert window.brand_bar.height() <= 30
    assert not window.toolbar.settings_expanded()
    assert window.brand_bar.btn_live.text() and window.brand_bar.btn_playback.text()
    assert window.toolbar.status_chip.text().strip()
    assert window.toolbar.record_chip.text() == t("record.state_idle")

    window.toolbar.btn_settings.setChecked(True)
    from PyQt5.QtWidgets import QApplication

    QApplication.processEvents()
    assert window.toolbar.settings_expanded()
    assert window.toolbar.units_combo.isVisible()
    assert window.toolbar.btn_preview.isVisible()
    expanded = window.brand_bar.height() + window.toolbar.height()
    assert expanded > collapsed, "opening the settings area changed nothing"
    assert window.act_settings.isChecked(), "the View menu does not follow the toggle"
    assert window.prefs["settings_expanded"] is True

    window.toolbar.btn_settings.setChecked(False)
    QApplication.processEvents()
    assert not window.toolbar.settings_expanded()


@pytest.mark.parametrize("size", RESOLUTIONS)
def test_settings_area_stays_inside_its_container(make_window, size):
    """Opening the settings area must not push a widget past its parent.

    The settings area is the one part of the top rail that may scroll: its own
    content is exempt, everything else must fit.
    """
    from PyQt5.QtWidgets import QApplication

    window = _show(make_window(sensor_type="finger", show_cal_panel=True), size)
    window._set_settings_expanded(True)
    for _ in range(3):
        QApplication.processEvents()

    oversize = []
    for child in visible_descendants(window):
        if ui_check._inside_scroll_area(child):
            continue
        parent = child.parentWidget()
        if parent is None or not parent.isVisible():
            continue
        geometry = child.geometry()
        if geometry.width() > parent.width() + 1 or geometry.height() > parent.height() + 1:
            oversize.append(
                f"{type(child).__name__} {geometry.width()}x{geometry.height()}"
                f" > parent {parent.width()}x{parent.height()}"
            )
    assert not oversize, oversize


def test_idle_run_is_the_only_filled_action(make_window):
    window = _show(make_window(), (1440, 880))
    assert window.toolbar.btn_run.objectName() == "primaryButton"
    assert not window.toolbar.btn_record.isEnabled(), "recording without a session"


def test_connecting_cancel_stays_filled(make_window):
    """Cancel is the way out of a hung device open, so it keeps the weight."""
    from PyQt5.QtWidgets import QApplication

    window = _show(make_window(), (1440, 880))
    stuck = ui_check._StuckWorker()
    window._ensure_workers = lambda: None
    window._box_worker = stuck
    window._finger_worker = stuck
    window.toolbar.btn_run.setChecked(True)
    QApplication.processEvents()
    assert window._session_phase == "connecting"
    assert window.toolbar.btn_run.objectName() == "dangerButton"
    window.toolbar.btn_run.setChecked(False)
    stuck.release()


def test_exactly_one_filled_action_while_acquiring(make_window):
    """Driven through the real entry point: Run, then the first packet."""
    window = _show(make_window(), (1440, 880))
    stuck = ui_check._StuckWorker()
    window._ensure_workers = lambda: None
    window._box_worker = stuck
    window._finger_worker = stuck
    window.toolbar.btn_run.setChecked(True)
    window._run_timer.stop()
    assert window._session_phase == "connecting"
    assert window.toolbar.btn_run.objectName() == "dangerButton", (
        "Cancel lost its weight while a connect can hang"
    )

    stuck.emit(ui_check._fake_packet(1))
    window._on_run_tick()
    assert window._session_phase == "running"
    assert window.toolbar.btn_run.objectName() == "dangerOutline", (
        "Stop is still the filled action while a session merely acquires"
    )
    assert window.toolbar.btn_record.objectName() == "primaryButton"

    window.toolbar.set_record_checked(True)
    assert window.toolbar.btn_record.objectName() == "dangerButton"
    assert window.toolbar.btn_run.objectName() == "dangerOutline", (
        "two filled actions share the rail while recording"
    )

    window.toolbar.set_record_checked(False)
    window.toolbar.btn_run.setChecked(False)
    stuck.release()


def test_record_strip_separates_acquiring_from_recording(make_window):
    from orisys_app_shared.i18n import t
    from orisys_app_shared.record.experiment import RecordStats

    from orisys_studio.ui import record_status

    window, stuck = _running_window(make_window)
    chip = window.toolbar.record_chip
    assert chip.text() == t("record.state_idle")
    assert t("record.state_idle") != t("status.phase_running"), (
        "acquiring and recording share one label"
    )

    real_stats = window._record_stats
    window._record_stats = lambda: (
        record_status.RECORDING,
        RecordStats(duration_s=12.4, frames=372, segments=1, dropped_data=3),
        "20240101T000000Z_session0001_deadbeef",
    )
    window._refresh_record_state()
    assert t("record.state_recording", duration="00:12.4") in chip.text()
    assert t("record.dropped", count=3) in chip.text()
    assert "deadbeef" in chip.toolTip()
    # A missing placeholder would leave the raw template in place: ``t`` falls
    # back to the key instead of raising, so the tooltip is asserted on content.
    assert chip.toolTip() != t("record.strip_tip")

    window._record_stats = lambda: (
        record_status.ERROR,
        RecordStats(frames=10, segments=1, failed=True, write_errors=2, last_error="disk full"),
        "",
    )
    window._refresh_record_state()
    assert chip.text() == t("record.state_error", count=2)
    assert "disk full" in chip.toolTip()

    window._record_stats = real_stats
    _finish(window, stuck)


# --------------------------------------------------- curves / arrow legend


def test_axis_readouts_sit_above_the_charts(make_window):
    window = _show(make_window(), (1440, 880))
    panel = window.force_plots
    assert panel.value_group.y() < panel.plot_frame.y(), (
        "the axis readouts are not above the charts"
    )
    for label in panel.value_labels:
        assert label.text(), "an axis readout is empty"
        assert label.height() >= label.fontMetrics().height(), "a readout is clipped"
        assert label.fontMetrics().height() >= 20, "the readouts are not large digits"
    assert [
        plot.getAxis("bottom").style.get("showValues") for plot in panel.plots
    ] == [False, False, True], "the three charts do not share one bottom time axis"


def test_overview_region_is_the_timeline_control(make_window):
    window, stuck = _running_window(make_window)
    panel = window.force_plots
    for index in range(120):
        stuck.emit(ui_check._fake_packet(index + 100))
    window._on_run_tick()
    assert window.session_store.count == 120
    first, last = window.session_store.extent()
    assert last > first, "the session has no extent to browse"
    assert panel.window_label.text(), "the overview strip has no readout"

    panel.show_all()
    assert panel.mode() == "browse"
    target = (first + (last - first) * 0.25, first + (last - first) * 0.5)
    panel.overview_region.setRegion(target)
    dragged = panel.window()
    assert abs(dragged[0] - target[0]) < 0.05 and abs(dragged[1] - target[1]) < 0.05, (
        f"dragging the overview did not move the window: {dragged} vs {target}"
    )
    assert panel.mode() == "browse"

    window.brand_bar.btn_live.click()
    assert panel.mode() == "follow", "the live nav item did not leave history browsing"
    _finish(window, stuck)


def test_the_charts_start_in_follow_mode(make_window):
    """Every live entry point resumes following; only browsing is a gesture."""
    from PyQt5.QtWidgets import QApplication

    from orisys_app_shared.i18n import t

    window, stuck = _running_window(make_window)
    panel = window.force_plots
    assert panel.mode() == "follow"
    assert panel.mode_label.text() == t("chart.mode_follow")

    for index in range(40):
        stuck.emit(ui_check._fake_packet(index))
        window._on_run_tick()
    QApplication.processEvents()
    assert panel.mode() == "follow", "live samples pushed the charts into browsing"

    panel._unit = "raw"
    window._on_units_changed("newton")
    assert panel.mode() == "follow", "a unit switch left the charts browsing"
    window._on_units_changed("raw")
    assert panel.mode() == "follow"

    # A new session is a live session again.
    panel.show_all()
    assert panel.mode() == "browse"
    window._start_run()
    window._run_timer.stop()
    assert panel.mode() == "follow", "a new run did not resume following"
    _finish(window, stuck)


def test_the_charts_zoom_and_pan_in_every_view(make_window):
    """Zoom/pan must be reachable from the live view, not only while browsing."""
    from PyQt5.QtWidgets import QApplication

    window, stuck = _running_window(make_window)
    panel = window.force_plots
    for index in range(60):
        stuck.emit(ui_check._fake_packet(index))
        window._on_run_tick()
    QApplication.processEvents()

    view_box = panel.plots[0].getViewBox()
    assert panel.mode() == "follow"
    assert view_box.state["mouseEnabled"][0], "the live chart ignores the mouse"

    # The gesture signal is what pyqtgraph emits for a wheel or drag.
    view_box.setXRange(0.5, 1.5, padding=0.0)
    view_box.sigRangeChangedManually.emit([True, False])
    QApplication.processEvents()
    assert panel.mode() == "browse", "a wheel/drag gesture did not start browsing"
    assert panel.window() == (0.5, 1.5), f"the gesture was not applied: {panel.window()}"
    assert all(cursor.movable for cursor in panel.cursors)

    # A programmatic change is not a gesture and must not switch the view.
    panel.follow_latest()
    view_box.setXRange(1.0, 1.5, padding=0.0)
    QApplication.processEvents()
    assert panel.mode() == "follow", "a programmatic range change started browsing"

    # While browsing, dragging the shared cursor moves the playhead.
    panel.show_all()
    moved: list[float] = []
    panel.cursor_moved.connect(moved.append)
    panel.cursors[0].setValue(1.25)
    QApplication.processEvents()
    assert [round(cursor.value(), 2) for cursor in panel.cursors] == [1.25] * 3
    assert moved and moved[-1] == 1.25, "the cursor drag did not notify the replay"
    _finish(window, stuck)


def test_the_arrow_legend_follows_the_view_mode(make_window):
    from orisys_app_shared.acquisition.arrows import arrow_display_scale

    assert arrow_display_scale("box") == 2.0
    assert arrow_display_scale("finger") == 1.0

    window = _show(make_window(), (1440, 880))
    panel = window.arrow_panel
    assert panel.scale_label.isVisible()
    assert "2" in panel.scale_label.text()
    assert panel.pipeline_label.isVisible() and panel.pipeline_label.text()
    assert panel.area_label.isVisible() and panel.area_label.text()
    assert not panel.colorbar.isVisible(), "the depth colour bar is shown in arrows"

    panel.set_view_mode("depth")
    panel.set_depth_scale("0.000 – 1.000")
    assert panel.colorbar.isVisible(), "the depth view has no colour bar"
    assert panel.depth_scale_label.isVisible()
    assert not panel.scale_label.isVisible(), "the arrow magnification leaked into depth"

    panel.set_view_mode("arrows_pure")
    assert not panel.colorbar.isVisible()
    assert panel.scale_label.isVisible()


# ------------------------------------------------- top rail: sensor and views


def test_the_sensor_summary_opens_the_dialog(make_window):
    """The rail states the sensor; the pickers live in the dialog it opens."""
    from PyQt5.QtWidgets import QApplication

    from orisys_app_shared.i18n import t

    window = _show(make_window(), (1440, 880))
    assert window.toolbar.btn_sensor.text() == t(
        "sensor.summary", type="BOX", config="sensor.json"
    )
    assert window.toolbar.btn_sensor.toolTip() == t("sensor.dialog_tip")
    assert not window.sensor_picker.isVisible(), "the pickers are still on the rail"

    window.toolbar.btn_sensor.click()
    QApplication.processEvents()
    dialog = window.sensor_dialog
    assert dialog is not None and dialog.isVisible()
    picker = window.sensor_picker
    assert picker.isVisible(), "the sensor pickers are not in the dialog"
    assert picker.type_label.text() == t("sensor.type")
    assert picker.config_label.text() == t("sensor.config")
    assert dialog.calib_title.text() == t("sensor.calib_section")
    assert dialog.version_label.text() and dialog.axes_label.text()
    assert dialog.export_label.text() == t("sensor.export_none")
    assert not dialog.btn_apply.isEnabled(), "applying with no export"

    # Changing the product in the dialog reaches the window and the rail.
    picker.type_combo.setCurrentIndex(picker.type_combo.findData("finger"))
    QApplication.processEvents()
    assert window.prefs["sensor_type"] == "finger"
    assert "FINGER" in window.toolbar.btn_sensor.text()
    assert picker.sensor_type() == "finger" and picker.config_name()

    picker.type_combo.setCurrentIndex(picker.type_combo.findData("box"))
    QApplication.processEvents()
    dialog.close()


def test_the_source_picker_follows_the_view(make_window):
    """Live offers a camera, a video-file run offers the way back, replay neither."""
    from PyQt5.QtWidgets import QApplication

    window, stuck = _running_window(make_window)
    assert window.toolbar.camera_group.isVisible()
    assert not window.toolbar.video_group.isVisible()
    assert not hasattr(window.toolbar, "source_combo"), "the source is still a combobox"

    window.brand_bar.btn_playback.click()
    QApplication.processEvents()
    assert not window.toolbar.camera_group.isVisible(), "the replay view shows a camera picker"
    window.brand_bar.btn_live.click()
    QApplication.processEvents()
    assert window.toolbar.camera_group.isVisible()

    window.toolbar.set_video_path("C:/tmp/demo.mp4")
    window._set_source_mode("video")
    QApplication.processEvents()
    assert window.toolbar.video_group.isVisible()
    assert "demo.mp4" in window.toolbar.source_summary.text()
    assert not window.toolbar.btn_preview.isEnabled(), "preview is offered for a video file"

    window.toolbar.use_camera_clicked.emit()
    QApplication.processEvents()
    assert window.toolbar.source_mode() == "camera"
    assert window.toolbar.video_path() == ""
    _finish(window, stuck)


def test_the_nav_items_switch_the_main_window_layout(make_window):
    window, stuck = _running_window(make_window)
    for index in range(60):
        stuck.emit(ui_check._fake_packet(index + 100))
    window._on_run_tick()
    assert window.session_store.count == 60

    window.brand_bar.btn_calibration.click()
    assert window.view_preset() == "calibration"
    sizes = window.content_splitter.sizes()
    assert window.cal_panel.isVisible()
    assert not window.arrow_panel.isVisible(), "the calibration view keeps the 2D panel"
    assert sizes[3] >= 500, f"the calibration panel is too narrow: {sizes}"
    assert window.brand_bar.btn_calibration.property("navActive") == "true"

    window.brand_bar.btn_playback.click()
    assert window.view_preset() == "replay"
    assert window.arrow_panel.isVisible() and not window.cal_panel.isVisible()
    # No recording is open yet: the bar says so, and the charts keep following
    # the live session instead of pretending to replay something.
    assert window.replay_bar.isVisible() and not window.replay.is_loaded()
    assert window.force_plots.mode() == "follow", "browsing starts with no recording"

    window.brand_bar.btn_live.click()
    assert window.view_preset() == "live"
    sizes = window.content_splitter.sizes()
    assert sizes[2] > sizes[1], f"the 2D panel is not the smaller one: {sizes}"
    assert window.force_plots.mode() == "follow"
    assert window.brand_bar.btn_live.property("navActive") == "true"
    _finish(window, stuck)


def test_the_calibration_view_does_not_clip_its_controls(make_window):
    from PyQt5.QtWidgets import QHeaderView

    window = _show(make_window(sensor_type="box", show_cal_panel=True), (1440, 880))
    window.brand_bar.btn_calibration.click()
    from PyQt5.QtWidgets import QApplication

    for _ in range(3):
        QApplication.processEvents()

    cal = window.cal_panel
    assert cal.scroll.isVisible(), "the calibration panel cannot scroll"
    assert (
        cal.table.horizontalHeader().sectionResizeMode(0) == QHeaderView.ResizeToContents
    ), "the diagnostic columns can still be elided"
    for button in (cal.btn_apply, cal.btn_export, cal.btn_toggle_validation):
        assert button.height() >= button.fontMetrics().height(), button.text()
        assert button.width() >= button.sizeHint().width(), (
            f"{button.text()!r} is clipped: {button.width()} < {button.sizeHint().width()}"
        )


def test_the_2d_area_keeps_the_output_aspect_ratio(make_window):
    window = _show(
        make_window(sensor_type="finger", show_finger_3d=False), (1440, 880)
    )
    image = window.arrow_panel.image_label
    assert abs(image.aspect() - 16.0 / 9.0) < 0.01
    assert abs(image.height() - image.width() / image.aspect()) <= 2, (
        f"the 2D area does not follow the output aspect: {image.width()}x{image.height()}"
    )


def test_the_replay_bar_labels_its_controls(make_window):
    """The old replay window's controls had no text: it never retranslated."""
    from orisys_app_shared.i18n import t

    window = _show(make_window(), (1440, 880))
    bar = window.replay_bar
    assert bar.btn_open.text() == t("playback.open")
    assert bar.btn_open.toolTip() == t("playback.open_tip")
    assert bar.btn_depth.text() == t("playback.recompute_depth")
    assert bar.btn_depth.toolTip() == t("playback.recompute_depth_tip")
    assert bar.btn_unit.text(), "the unit switch is unlabelled"
    assert bar.slider.toolTip() == t("playback.scrub_tip")
    assert bar.info_label.text() == t("playback.empty")
    assert not bar.slider.isEnabled(), "the progress bar is live with no recording"


def test_the_replay_view_loads_a_recording_and_scrubs_it(make_window, replay_recording):
    """The replay view reads the recording: charts, 2D panel and progress bar."""
    from PyQt5.QtWidgets import QApplication

    from orisys_app_shared.i18n import t

    window = _show(make_window(), (1440, 880))
    window.brand_bar.btn_playback.click()
    QApplication.processEvents()
    assert window.replay_bar.isVisible(), "the replay view has no progress bar"

    assert window._load_recording(replay_recording)
    QApplication.processEvents()
    panel = window.force_plots
    assert panel.is_recording_source(), "the charts did not switch to the recording"
    assert panel.mode() == "browse", "opening a recording did not browse it"
    assert panel.mode_label.text() == t("chart.mode_replay")
    assert not panel.btn_latest.isEnabled(), "a recording offers 'back to latest'"
    drawn = [int(curve.getData()[0].size) for curve in panel.curves]
    assert all(count > 0 for count in drawn), f"the recording drew nothing: {drawn}"

    assert window.arrow_panel.view_modes() == ("recorded", "depth")
    assert window.replay.video_image() is not None, "no frame at the cursor"
    assert window.replay_bar.slider.isEnabled()

    # The progress bar is the playhead.
    window.replay_bar.slider.setValue(window.replay_bar.slider.maximum() // 2)
    QApplication.processEvents()
    assert 0.0 < window.replay.position() < window.replay.duration_s()
    assert abs(panel.cursors[0].value() - window.replay.position()) < 0.05

    # Dragging the chart cursor scrubs the recording too.
    panel.cursors[0].setValue(0.25)
    QApplication.processEvents()
    assert abs(window.replay.position() - 0.25) < 0.05

    # Leaving the view returns the charts to the live session.
    window.brand_bar.btn_live.click()
    QApplication.processEvents()
    assert not panel.is_recording_source()
    assert panel.mode() == "follow"
    assert not window.replay_bar.isVisible()
    assert window.arrow_panel.view_modes() == ("arrows_pure", "arrows_stitched", "depth")


# ------------------------------------------------- correctness regressions


def test_a_late_calibration_does_not_plot_raw_counts_on_the_newton_axis(make_window):
    """A device confirms its calibration a frame or two after the first packets.

    Those frames carry raw counts. ``push`` used to record the new unit and keep
    the old samples, so the first thing drawn on the newton axis was a spike of
    raw counts — the transient at the start of every session. Switching units is
    a switch, never a reinterpretation: the raw samples are dropped from the
    buffer (the store still holds them) and the sample that triggered the switch
    is read out but not appended twice.
    """
    from orisys_app_shared.i18n import t

    window, stuck = _running_window(make_window)
    plots = window.force_plots
    # The preference is newton; an uncalibrated device falls back to raw.
    assert window.toolbar.units() == "newton"
    plots.reset()
    for index in range(3):
        packet = ui_check._fake_packet(index, calibrated_axes=())
        packet.force_calibrated = False
        stuck.emit(packet)
        window._on_run_tick()
    assert plots.unit() == "raw", f"an uncalibrated device is not shown raw: {plots.unit()}"

    for index in range(3, 40):
        stuck.emit(ui_check._fake_packet(index))
        window._on_run_tick()

    state = plots.history.state()
    assert plots.unit() == "N", f"the unit did not follow the calibration: {plots.unit()}"
    assert state is not None
    assert plots.history.sample_count == 37, (
        f"the raw frames stayed in the newton buffer: {plots.history.sample_count} samples"
    )
    peak = float(np.nanmax(np.abs(state.values[:, 0])))
    assert peak < 5.0, f"a raw count is still drawn on the newton axis (peak {peak})"

    # The sample that caused the switch still reaches the readouts; the early
    # return used to leave the cards showing the previous sample.
    assert abs(plots._last_values[0] - (2.0 + 39 * 0.01)) < 1e-9
    assert plots.value_labels[0].text() == "2.390", plots.value_labels[0].text()
    assert plots.value_labels[0].text() != t("chart.value_unavailable")
    _finish(window, stuck)


def test_the_unit_chip_switches_the_charts_and_the_toolbar(make_window):
    """One click on the unit chip changes the unit everywhere, not just locally."""
    from orisys_app_shared.i18n import t
    from orisys_studio.ui.force_plots import unit_display

    from PyQt5.QtWidgets import QPushButton

    window = _show(make_window(), (1440, 880))
    plots = window.force_plots
    assert isinstance(plots.unit_badge, QPushButton), "the unit chip is not clickable"
    assert plots.unit_badge.toolTip() == t("chart.unit_toggle_tip")
    assert plots.unit_badge.accessibleName() == t("chart.unit_toggle")

    before = plots.unit()
    plots.unit_badge.click()
    assert plots.unit() != before, "the unit chip did not switch the charts"
    assert plots.unit_badge.text() == unit_display(plots.unit())
    assert ("newton" if plots.unit() == "N" else "raw") == window.toolbar.units(), (
        "the unit chip and the toolbar disagree about the unit"
    )
    plots.unit_badge.click()
    assert plots.unit() == before, "the unit chip does not switch back"


def test_the_chart_actions_are_icons_that_still_say_what_they_do(make_window):
    """Symbols instead of words — but every one keeps its meaning in words.

    The header actions (zoom to fit, box zoom, back to latest) are drawn at
    runtime in the theme's text colour; a symbol with no tooltip or accessible
    name would be unreadable, so both are part of the contract.
    """
    from orisys_app_shared.i18n import t

    window = _show(make_window(), (1440, 880))
    plots = window.force_plots
    for button, key in (
        (plots.btn_overview, "chart.overview"),
        (plots.btn_box_zoom, "chart.box_zoom"),
        (plots.btn_latest, "chart.return_latest"),
    ):
        assert button.text() == "", f"{key} still shows text: {button.text()!r}"
        assert not button.icon().isNull(), f"{key} has no icon"
        assert button.toolTip() == t(f"{key}_tip")
        assert button.accessibleName() == t(key)

    # The overview strip keeps its word: it labels a region, not an action.
    assert plots.window_caption.text() == t("chart.overview")

    # The arrow footer says what the factor multiplies.
    footer = window.arrow_panel.scale_label.text()
    assert "×" in footer, footer
    assert "位移" in footer or "displacement" in footer, footer

    # A theme switch re-tints the icons instead of leaving stale pixels.
    window._set_theme("light")
    for button in (plots.btn_overview, plots.btn_box_zoom, plots.btn_latest):
        assert not button.icon().isNull(), "an icon vanished across a theme switch"


def test_a_recording_counts_up_while_it_is_running(make_window, record_root):
    """The elapsed time is the span captured so far, not just the final number."""
    import orisys_studio.app_window as app_module
    from orisys_app_shared.i18n import t

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: record_root
    window.toolbar.btn_record.setChecked(True)
    for index in range(96):
        stuck.emit(ui_check._fake_packet(index, mono_t=index / 30.0))
        window._on_run_tick()

    stats = window.recorder.stats()
    assert stats.duration_s > 2.5, f"the live duration stayed at {stats.duration_s}"
    assert window.toolbar.record_chip.text() != t(
        "record.state_recording", duration="00:00.0"
    ), "the record strip does not count up while recording"
    window.toolbar.btn_record.setChecked(False)
    _finish(window, stuck)


def test_the_video_follows_the_sensor_rate_and_keeps_every_row(make_window, record_root):
    """The container rate is the sensor's, and pacing never thins the data.

    The packet's ``fps`` is how fast the pipeline *finished* frames, not how fast
    the sensor produces them: a burst at open (or a file source) reads 48 on a
    30 fps sensor, and writing every delivered frame into a 30 fps container made
    playback run at the wrong speed.
    """
    import csv

    import cv2

    import orisys_studio.app_window as app_module

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: record_root
    assert abs(window._video_fps() - 30.0) < 0.01, window._video_fps()

    window.toolbar.btn_record.setChecked(True)
    for index in range(96):
        # 48 frames per second of acquisition time, above the configured 30.
        packet = ui_check._fake_packet(index, mono_t=index / 48.0)
        packet.fps = 48.0
        stuck.emit(packet)
        window._on_run_tick()
    window.toolbar.btn_record.setChecked(False)

    report = window._record_last_report
    assert report is not None and report.completed
    assert report.stats.resampled_video > 0, "the extra frames were not reported"
    assert any("faster than" in problem for problem in report.problems), report.problems

    with (report.directory / "frames.csv").open(encoding="utf-8") as handle:
        rows = list(csv.DictReader(handle))
    assert len(rows) == 96, f"the video pacing thinned the data rows: {len(rows)}"

    segments = sorted(report.directory.glob("raw_*.mp4"))
    assert segments, "the recording wrote no video segment"
    capture = cv2.VideoCapture(str(segments[0]))
    try:
        written = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
        container_fps = float(capture.get(cv2.CAP_PROP_FPS))
    finally:
        capture.release()
    assert abs(container_fps - 30.0) < 1.0, f"the video declares {container_fps} fps"
    assert 55 <= written <= 65, f"the video holds {written} frames for 2 s at 30 fps"
    _finish(window, stuck)


def test_a_replay_keeps_the_2d_pane_from_the_live_camera(
    make_window, replay_recording
):
    """The pane belongs to the source the view is showing.

    While the replay view is up, its frames own the 2D panel; a live packet used
    to paint over the recording. Leaving the view hands the pane back.
    """
    from PyQt5.QtWidgets import QApplication

    window, stuck = _running_window(make_window)
    window.brand_bar.btn_playback.click()
    QApplication.processEvents()
    assert window._load_recording(replay_recording), "the replay could not open it"
    QApplication.processEvents()
    assert window._replay_owns_image(), "the replay does not own the 2D pane"

    label = window.arrow_panel.image_label
    # Park the playhead at the end of the recording and remember what the pane and
    # the cards show there, so a live packet that overwrote either would show up.
    window.replay_bar.slider.setValue(window.replay_bar.slider.maximum())
    QApplication.processEvents()
    before_key = label.pixmap().cacheKey()
    before_text = label.text()
    before_card = window.force_plots.value_labels[0].text()
    live = np.full((24, 32, 3), 200, np.uint8)
    for index in range(3):
        packet = ui_check._fake_packet(index, mono_t=index / 30.0, frame=live)
        packet.normal_force = 111.0
        packet.normal_force_raw = 999.0
        stuck.emit(packet)
        window._on_run_tick()
    QApplication.processEvents()
    assert label.pixmap().cacheKey() == before_key and label.text() == before_text, (
        "a live frame overwrote the recording in the 2D pane"
    )
    assert window.force_plots.value_labels[0].text() == before_card, (
        "a live sample overwrote the recording's readout"
    )

    # ...and the live view takes it back, which is what makes the check above
    # meaningful rather than a dead pane.
    window.brand_bar.btn_live.click()
    QApplication.processEvents()
    assert not window._replay_owns_image()
    stuck.emit(ui_check._fake_packet(9, mono_t=9 / 30.0, frame=live))
    window._on_run_tick()
    QApplication.processEvents()
    assert label.pixmap().cacheKey() != before_key, "the live view stopped drawing"
    _finish(window, stuck)


def test_the_readout_describes_the_point_the_cursor_is_on(
    make_window, replay_recording
):
    """The three numbers are the cursor's sample, not the last live packet.

    A replay used to state the live acquisition in cards sitting next to a chart
    of 7.5e5 raw counts, which reads as "the replay never opened".
    """
    from PyQt5.QtWidgets import QApplication

    from orisys_app_shared.i18n import t

    window = _show(make_window(), (1440, 880))
    window.brand_bar.btn_playback.click()
    QApplication.processEvents()
    assert window._load_recording(replay_recording)
    QApplication.processEvents()
    plots = window.force_plots
    assert plots.mode() == "browse", "opening a recording did not browse it"

    slider = window.replay_bar.slider
    slider.setValue(slider.minimum())
    QApplication.processEvents()
    first = plots.value_labels[0].text()
    assert first not in ("", t("chart.value_unavailable")), first
    slider.setValue(slider.maximum())
    QApplication.processEvents()
    last = plots.value_labels[0].text()

    # The fixture's forces rise monotonically, so the readout must rise too: the
    # cards are following the playhead rather than a constant.
    assert float(last) > float(first), f"the readout ignores the cursor: {first} -> {last}"
    assert float(first) != 0.0, "a recording is read out as a live zero"
    assert abs(plots.cursor_time() - window.replay.position()) < 0.05


def test_a_raw_only_recording_is_never_shown_in_newton(make_window, replay_recording):
    """A newton preference must not empty a recording that has no calibration.

    The recording's raw counts are always there; its newton triplet only exists
    for the axes it calibrated. Asking for newton anyway drew three blank charts
    and a dash in every card, which looks exactly like "the replay did not load".
    """
    from PyQt5.QtWidgets import QApplication

    from orisys_app_shared.i18n import t

    window = _show(make_window(), (1440, 880))
    window.toolbar.set_units("newton")
    QApplication.processEvents()
    window.brand_bar.btn_playback.click()
    QApplication.processEvents()
    assert window._load_recording(replay_recording)
    QApplication.processEvents()

    panel = window.force_plots
    assert panel.unit() == "raw", "a raw-only recording was shown in newton"
    assert window.toolbar.units() == "raw", "the toolbar disagrees with the charts"
    assert window.replay_bar.btn_unit.text() == t("playback.show_newton"), (
        "the replay bar's unit switch disagrees with the charts"
    )
    drawn = [int(curve.getData()[0].size) for curve in panel.curves]
    assert all(count > 0 for count in drawn), f"the recording drew nothing: {drawn}"
    assert panel.value_labels[0].text() not in ("", t("chart.value_unavailable"))