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

"""Stage 4 acceptance: the 3D pane in the product.

Four things are being checked here, all of them about integration rather than about
mathematics (the camera, the sampler and the composer have their own files):

* **the pane degrades into a sentence.** No GL context, no package, a refused
  package, the wrong sensor type: each one is a line of text in the operator's
  language, and every one of them leaves the two-dimensional application alone.
* **the pane is one state with two controls.** The View menu entry and the checkbox
  inside the pane are the same flag, and toggling them repeatedly does not drift or
  leak.
* **the record is not affected.** The 3D sink runs on the acquisition thread next to
  the record sink, so the same session is run twice — once with a consumer attached
  to the pane, once without — and the CSV bytes and the valid-sample count must be
  identical. This is the stage-1 note ("开/关三维的 CSV 行序一致") being paid off.
* **the saved settings are repaired, not trusted.** A preferences file is editable by
  hand; it must not be able to make the window fail to open or make a stale file
  relabel a live measurement.
"""

from __future__ import annotations

import itertools
import shutil
import time
from concurrent.futures import Future
from dataclasses import replace
from pathlib import Path

import numpy as np
import pytest

import ui_check
from PyQt5.QtCore import pyqtSignal
from PyQt5.QtWidgets import QWidget
from orisys_app_shared.scene3d import synthetic
from orisys_app_shared.scene3d.contracts import (
    ColorScale,
    DisplaySettings,
    FieldDescriptor,
    UNAVAILABLE_DESCRIPTOR,
)
from orisys_app_shared.scene3d.package import write_package
from orisys_studio.app_window import MIN_PANEL_WIDTHS
from orisys_studio.scene3d import prefs as scene3d_prefs
from orisys_studio.scene3d.asset_source import (
    PACKAGE_SUFFIX,
    SceneSource,
    find_package,
    load_package,
    no_asset_diagnostic,
)
from tests.test_ui_smoke import (  # noqa: F401  (fixtures are imported by name)
    _finish,
    _running_window,
    record_root,
    replay_recording,
)

#: Scratch root for the two-session comparison. One subdirectory per run, because
#: the point of the test is that two recordings are identical, not that one file
#: can be overwritten.
RECORD_SCRATCH = Path(__file__).resolve().parents[1] / "outputs" / "_test_record_3d"


@pytest.fixture
def next_record_root():
    shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)
    RECORD_SCRATCH.mkdir(parents=True, exist_ok=True)
    counter = itertools.count()

    def _next() -> Path:
        path = RECORD_SCRATCH / f"run{next(counter)}"
        path.mkdir(parents=True, exist_ok=True)
        return path

    yield _next
    shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)


@pytest.fixture
def write_poc(tmp_path):
    """A real package on disk, produced by the shipping writer."""

    def _write(name: str = "finger-display-1.0.0") -> "object":
        path = tmp_path / f"{name}{PACKAGE_SUFFIX}"
        write_package(
            synthetic.poc_scene(),
            path,
            sensor_model="finger",
            hardware_revisions=("*",),
        )
        return path

    return _write


def _pump(qt_app, predicate, *, timeout_s: float = 10.0) -> bool:
    """Let Qt deliver queued signals until ``predicate`` holds."""
    import time

    deadline = time.perf_counter() + timeout_s
    while time.perf_counter() < deadline:
        qt_app.processEvents()
        if predicate():
            return True
        time.sleep(0.01)
    return False


# ------------------------------------------------------------------- the pane
def test_a_box_sensor_reports_that_it_has_no_3d_scene(qt_app):
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    panel = Mesh3dPanel()
    try:
        panel.set_sensor("box")
        assert panel.showing_message
        assert panel.view is None, "a box sensor must not build a GL widget at all"
        assert "3D" in panel.message.text() or "三维" in panel.message.text()
    finally:
        panel.deleteLater()


def test_a_missing_package_names_where_it_looked(tmp_path, qt_app, monkeypatch):
    """With no package anywhere, the pane names the directory it looked in.

    The search is stubbed rather than relied on: the repository may or may not carry
    a compiled package (the application's own config directory is where one belongs),
    and a test that only passes on a machine without one is a test that fails the
    first time somebody installs the feature.
    """
    import orisys_studio.ui.mesh3d_panel as panel_module
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    monkeypatch.setattr(panel_module, "find_package", lambda *a, **k: None)
    panel = Mesh3dPanel()
    try:
        panel.set_sensor("finger", package_override=tmp_path / "nope.oryscene")
        assert panel.showing_message
        diagnostic = panel.last_diagnostic
        assert diagnostic is not None and diagnostic.code == "asset_rejected"
        assert "directory" in diagnostic.details
        assert "compile_scene.py" in diagnostic.details["compile"]
        assert panel.message.text(), "the pane showed an empty line instead of the reason"
    finally:
        panel.deleteLater()


def test_the_view_is_built_once_and_only_for_a_finger(qt_app, write_poc):
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    panel = Mesh3dPanel()
    try:
        panel.set_sensor("box")
        assert panel.ensure_view() is None
        panel.set_sensor("finger", package_override=write_poc())
        first = panel.ensure_view()
        second = panel.ensure_view()
        assert first is not None and first is second, "the pane rebuilt its GL widget"
        assert panel.scene_path() is not None
    finally:
        panel.close_view()
        panel.deleteLater()


def test_the_pane_fits_the_width_the_window_reserves_for_it(qt_app):
    """A pane whose content demands more than its declared minimum clips itself."""
    from orisys_studio.app_window import MIN_PANEL_WIDTHS
    from orisys_studio.scene3d.view import Scene3DView

    view = Scene3DView()
    try:
        wanted = view.minimumSizeHint().width()
        assert wanted <= MIN_PANEL_WIDTHS["mesh3d"], (
            f"the 3D pane needs {wanted} px but the window reserves only "
            f"{MIN_PANEL_WIDTHS['mesh3d']}"
        )
        assert view.gl_widget is None or view.gl_widget.minimumHeight() >= 100, (
            "the picture must keep a usable height when the pane is short"
        )
    finally:
        view.close()


def test_a_platform_without_a_context_says_so_instead_of_showing_an_empty_view(
    write_poc, qt_app
):
    """The offscreen plugin cannot create a context; that must not look like "no data".

    Qt reports this by never calling ``initializeGL``, so the widget asks Qt the one
    question that has an answer — is there a context at all — on its first paint. The
    scene comes from a package written for this test, so the case does not depend on
    whether the repository happens to carry a compiled one.
    """
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    panel = Mesh3dPanel()
    try:
        panel.set_sensor("finger", package_override=write_poc())
        view = panel.ensure_view()
        if view is None:
            pytest.skip("the pane did not build a view here")
        # The pane has to be a shown window for Qt to lay out and paint its child: a
        # child of a hidden parent is never painted, which is why the widget only
        # learns that it has no context once something is actually on screen.
        panel.resize(240, 400)
        panel.show()
        view.show()
        panel.set_enabled(True)
        assert _pump(qt_app, lambda: view.assets is not None, timeout_s=15.0), (
            "the test package never loaded"
        )
        widget = view.gl_widget
        for _ in range(5):
            widget.repaint()
            qt_app.processEvents()
        if widget.capabilities() is not None:
            pytest.skip("this platform has a real GL context")
        assert widget.unavailable, "a context-less widget claimed to be usable"
        assert panel.showing_message, "the pane kept showing an empty GL widget"
        assert panel.message.text().strip(), "the pane showed an empty line"
    finally:
        panel.close_view()
        panel.deleteLater()


# --------------------------------------------------------- one state, two controls
def test_the_pane_and_the_view_menu_are_the_same_flag(make_window):
    window = make_window(sensor_type="finger")
    try:
        assert window.mesh3d_panel.enabled() == bool(window._show_finger_3d)
        for _ in range(20):
            window.act_finger_3d.setChecked(not window.act_finger_3d.isChecked())
            assert window.mesh3d_panel.enabled() == bool(window._show_finger_3d)
            assert window.act_finger_3d.isChecked() == bool(window._show_finger_3d)
        # The pane's own checkbox is the other way in, and it drives the same flag.
        view = window.mesh3d_panel.view
        if view is not None and view.panel is not None:
            view.panel.enable_check.setChecked(not window._show_finger_3d)
            assert window.act_finger_3d.isChecked() == bool(window._show_finger_3d)
            assert window.mesh3d_panel.enabled() == bool(window._show_finger_3d)
    finally:
        window.close()


def test_panel_controls_do_not_touch_the_algorithm_side(make_window):
    """The observation camera is display state; nothing else may move with it."""
    window = make_window(sensor_type="finger")
    try:
        before = {
            "units": window.toolbar.units(),
            "image_view_mode": window.arrow_panel.view_mode(),
            "arrow_rotation_deg": window.arrow_panel.rotation_deg(),
            "view_preset": window._view_preset,
            "sensor_type": window.sensor_picker.sensor_type(),
            "splitter": list(window.content_splitter.sizes()),
            "samples": window.session_store.count,
        }
        view = window.mesh3d_panel.view
        if view is None:
            pytest.skip("no 3D view on this platform")
        camera = view._camera
        start = np.asarray(camera.matrices((240, 180)).view, dtype=float)
        camera.orbit((30.0, 10.0), (240, 180))
        camera.pan((5.0, 5.0), (240, 180))
        camera.zoom(2.0)
        moved = np.asarray(camera.matrices((240, 180)).view, dtype=float)
        assert not np.allclose(start, moved), "the camera did not react to the mouse"
        # Still a rigid transform: no mirroring, no scale change.
        rotation = moved[:3, :3]
        assert np.linalg.det(rotation) == pytest.approx(1.0, abs=1e-6)
        assert rotation @ rotation.T == pytest.approx(np.eye(3), abs=1e-6)

        after = {
            "units": window.toolbar.units(),
            "image_view_mode": window.arrow_panel.view_mode(),
            "arrow_rotation_deg": window.arrow_panel.rotation_deg(),
            "view_preset": window._view_preset,
            "sensor_type": window.sensor_picker.sensor_type(),
            "splitter": list(window.content_splitter.sizes()),
            "samples": window.session_store.count,
        }
        assert after == before
    finally:
        window.close()


# ------------------------------------------------------------------ record isolation
class _RecordingView(QWidget):
    """A stand-in for the pane's view that records what it was handed.

    Used to attach a *real* consumer to the real sink path on a machine with no GL:
    what is being tested is whether the window's second sink can change the record,
    not whether the adapter can compute a flow magnitude. It implements the container's
    whole contract (see ``Mesh3dPanel.attach_view``) rather than the one method a
    particular test happens to call, so a change to that contract shows up here. It is
    a real ``QWidget`` because the container puts it in its stack.
    """

    diagnosticRaised = pyqtSignal(object)
    enabledChanged = pyqtSignal(bool)

    def __init__(self, parent=None) -> None:
        super().__init__(parent)
        self.packets: list = []
        self.coordinator = None
        self.assets = None
        self.scenes_used: list = []
        self.accept_scene = True
        self.available = True
        self.reason = ""
        self.source_kind = "live"
        self.sessions = 0
        self.reference_only = 0
        self.closed = 0

    # the sink the acquisition thread calls
    def submit_packet(self, packet) -> None:
        # Deliberately more work than the real one does per packet, so a leak into the
        # record path would show up as a difference rather than as a timing artefact.
        self.packets.append(packet)
        frame = np.asarray(packet.raw_frame)
        float(np.asarray(frame, dtype=np.float64).sum())

    # the container's contract
    def use_scene(self, assets, source_id: str, expected=None) -> bool:
        """The seam ``Mesh3dPanel._on_loaded`` adopts a scene through."""
        self.scenes_used.append((assets, source_id, expected))
        if not self.accept_scene:
            return False
        self.assets = assets
        return True

    def set_enabled(self, enabled: bool) -> None:
        self.enabled_flag = bool(enabled)

    def set_source_kind(self, kind: str) -> None:
        self.source_kind = str(kind)

    def begin_session(self) -> None:
        self.sessions += 1

    def show_reference_only(self) -> None:
        self.reference_only += 1

    def tick(self) -> None:
        self.ticks = getattr(self, "ticks", 0) + 1

    def close(self) -> None:
        self.closed += 1

    def settings(self):
        return DisplaySettings()

    def restore_settings(self, settings) -> None:
        self.settings_value = settings

    def describe(self) -> dict:
        return {"scene": {"asset_id": "double"}, "gl": {}, "display": {}, "stream": {}}

    def render_totals(self) -> dict:
        return {"shown": 0, "dropped_display": 0, "stale": 0, "invalid": 0}

    def render_stats(self):
        return None


def _run_session(make_window, record_root, *, with_3d: bool):
    """One full session with the 3D sink attached, or with no pane at all."""
    import orisys_studio.app_window as app_module

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: record_root
    view = None
    if with_3d:
        view = _RecordingView()
        window.mesh3d_panel.attach_view(view)
    # With no pane the sink still runs (it is registered on the worker) and does
    # nothing, which is the case the comparison below is about.
    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()
    window.toolbar.btn_record.setChecked(False)
    report = window._record_last_report
    rows = ""
    if report is not None and (report.directory / "forces.csv").is_file():
        rows = (report.directory / "forces.csv").read_bytes().decode("utf-8")
    counts = (window.session_store.count, len(list(window.session_store.iter_valid("raw"))))
    _finish(window, stuck)
    window.close()
    return rows, counts, report, view


def test_the_3d_consumer_cannot_change_the_record(make_window, next_record_root):
    """The acceptance row: identical CSV and identical valid samples, 3D on or off."""
    with_3d_csv, with_3d_counts, with_3d_report, view = _run_session(
        make_window, next_record_root(), with_3d=True
    )
    without_csv, without_counts, without_report, _ = _run_session(
        make_window, next_record_root(), with_3d=False
    )
    assert view is not None and len(view.packets) == 96, (
        "the pane's sink did not receive the session's packets"
    )
    assert with_3d_csv, "the recording produced no forces.csv"
    assert with_3d_csv == without_csv, (
        "the record changed when a 3D consumer was attached to the same packets"
    )
    assert with_3d_counts == without_counts, (with_3d_counts, without_counts)
    assert with_3d_report is not None and without_report is not None
    assert len(with_3d_report.problems) == len(without_report.problems)
    header = with_3d_csv.splitlines()[0]
    assert "t_rel_s" in header and len(with_3d_csv.splitlines()) > 10


def test_the_3d_sink_is_registered_and_inert_when_the_pane_is_off(make_window, next_record_root):
    """A sink that has to be added and removed around a toggle is a state machine."""
    import orisys_studio.app_window as app_module

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: next_record_root()
    try:
        assert window._on_packet_recorded in stuck.sinks()
        assert window._on_packet_for_3d in stuck.sinks(), (
            "the 3D sink must be registered with the worker, not attached per toggle"
        )
        assert window.mesh3d_panel.view is None, "a box sensor must not build a view"
        for index in range(10):
            stuck.emit(ui_check._fake_packet(index))
        assert window.session_store.count == 10
        assert window.mesh3d_panel.view is None
    finally:
        _finish(window, stuck)
        window.close()


def test_repeated_closes_are_safe(make_window):
    window = make_window(sensor_type="finger")
    try:
        window.mesh3d_panel.ensure_view()
        window.mesh3d_panel.close_view()
        window.mesh3d_panel.close_view()
        window.close()
        window.close()
    finally:
        pass


# ------------------------------------------------------------------ preferences
def test_the_display_settings_round_trip():
    settings = DisplaySettings(
        display_gain=12.0,
        deform_enabled=True,
        housing_visible=False,
        max_fps=60,
        color_scale=ColorScale(
            descriptor=FieldDescriptor("flow_magnitude", "px", "approximate", False),
            lower=0.25,
            upper=4.5,
            lut_id="sequential",
            mode="fixed",
        ),
    )
    restored = scene3d_prefs.settings_from_prefs(scene3d_prefs.settings_to_prefs(settings))
    assert restored.display_gain == 12.0
    assert restored.deform_enabled is True
    assert restored.housing_visible is False
    assert restored.max_fps == 60
    assert restored.color_scale is not None
    assert (restored.color_scale.lower, restored.color_scale.upper) == (0.25, 4.5)
    assert restored.color_scale.mode == "fixed"


def test_a_saved_descriptor_is_never_restored_as_a_measurement():
    """The colour bar describes the live field, not the last one that was on screen."""
    block = scene3d_prefs.settings_to_prefs(
        DisplaySettings(
            color_scale=ColorScale(
                descriptor=FieldDescriptor("displacement", "m", "calibrated", True),
                lower=0.0,
                upper=1.0,
                lut_id="diverging",
                mode="fixed",
            )
        )
    )
    restored = scene3d_prefs.settings_from_prefs(block)
    assert restored.color_scale is not None
    assert restored.color_scale.descriptor == UNAVAILABLE_DESCRIPTOR

    given = FieldDescriptor("flow_magnitude", "px", "approximate", False)
    restored_with = scene3d_prefs.settings_from_prefs(block, descriptor=given)
    assert restored_with.color_scale.descriptor == given


@pytest.mark.parametrize(
    "block",
    [
        None,
        {},
        {"display_gain": "lots", "max_fps": "fast", "deform_enabled": "maybe"},
        {"display_gain": -5.0, "max_fps": 17},
        {"display_gain": 1e9, "max_fps": 30},
        {"color_scale": {"mode": "sideways", "lower": "a", "upper": "b"}},
        {"color_scale": {"mode": "fixed", "lower": 5.0, "upper": 5.0}},
        {"color_scale": {"mode": "fixed", "lower": float("nan"), "upper": float("inf")}},
        {"color_scale": {"mode": "fixed", "lower": 1e300, "upper": 1e301}},
    ],
)
def test_a_hand_edited_settings_block_is_repaired(block):
    settings = scene3d_prefs.settings_from_prefs(block)
    assert isinstance(settings, DisplaySettings)
    assert scene3d_prefs.MIN_GAIN <= settings.display_gain <= scene3d_prefs.MAX_GAIN
    assert settings.max_fps in scene3d_prefs.FPS_CHOICES
    if settings.color_scale is not None:
        lower, upper = settings.color_scale.lower, settings.color_scale.upper
        assert lower == lower and upper == upper, "a NaN survived the repair"
        assert upper > lower
        assert abs(lower) <= scene3d_prefs.MAX_SCALE_BOUND
        assert abs(upper) <= scene3d_prefs.MAX_SCALE_BOUND
        assert settings.color_scale.mode in scene3d_prefs.SCALE_MODES


def test_the_panel_offers_exactly_the_rates_the_repair_accepts(qt_app):
    from orisys_studio.scene3d.panel import FPS_CHOICES

    assert tuple(FPS_CHOICES) == tuple(scene3d_prefs.FPS_CHOICES)


# ------------------------------------------------------------------- asset source
def test_the_package_search_order_is_override_then_sensor_directory(tmp_path, write_poc):
    explicit = write_poc("explicit")
    sensor_dir = tmp_path / "sensors" / "finger" / "scene"
    sensor_dir.mkdir(parents=True)
    shipped = sensor_dir / f"shipped{PACKAGE_SUFFIX}"
    shipped.write_bytes(explicit.read_bytes())

    found = find_package("finger", override=explicit, root=tmp_path)
    assert found is not None and found.path == explicit
    found_dir = find_package("finger", root=tmp_path)
    assert found_dir is not None and found_dir.path == shipped
    assert find_package("box", root=tmp_path) is None
    assert find_package("finger", override=tmp_path / "gone.oryscene", root=tmp_path) is not None, (
        "a stale override must fall through to the shipped package, not disable 3D"
    )


def test_a_package_is_loaded_off_the_gui_thread(qt_app, write_poc):
    from orisys_studio.scene3d.asset_source import SceneAssetLoader

    path = write_poc()
    loader = SceneAssetLoader()
    loaded: list = []
    failed: list = []
    loader.loaded.connect(lambda assets, description: loaded.append((assets, description)))
    loader.failed.connect(failed.append)
    try:
        loader.load(SceneSource(path=path, kind="package", directory=path.parent))
        assert _pump(qt_app, lambda: bool(loaded or failed)), "the loader never answered"
    finally:
        loader.close()
    assert not failed, [item.message for item in failed]
    assets, description = loaded[0]
    assert assets.geometry_id == synthetic.poc_scene().geometry_id
    assert str(path) in description
    # A second request while one is in flight is not dropped: it is the one the
    # operator asked for *last*, so it waits for the first to finish and then runs.
    loader2 = SceneAssetLoader()
    try:
        loader2._inflight = True
        first = loader2.load(SceneSource(path=path, kind="package", directory=path.parent))
        second = loader2.load(SceneSource(path=path, kind="package", directory=path.parent))
        assert second > first
        assert loader2._executor is None, "a queued request must not open a second worker"
        assert loader2._pending is not None, "the newer request was dropped"
        assert loader2._pending[-1] == second
    finally:
        loader2.close()


class _ManualExecutor:
    """Futures that only run when the test says so, like a loaded thread pool."""

    def __init__(self) -> None:
        self.pending: list = []

    def submit(self, function, *args, **kwargs):
        future: Future = Future()
        self.pending.append((future, function, args, kwargs))
        return future

    def run_next(self) -> None:
        future, function, args, kwargs = self.pending.pop(0)
        if future.set_running_or_notify_cancel():
            try:
                future.set_result(function(*args, **kwargs))
            except BaseException as exc:  # noqa: BLE001 - mirrored into the future
                future.set_exception(exc)

    def shutdown(self, wait: bool = True) -> None:  # noqa: ARG002 - executor contract
        self.pending.clear()


def test_the_newest_load_is_the_one_that_runs(qt_app, write_poc):
    """R7: requests carry an id, and a queue keeps the newest rather than the first."""
    from orisys_studio.scene3d.asset_source import SceneAssetLoader

    path = write_poc()
    loader = SceneAssetLoader()
    loader._executor = _ManualExecutor()
    answers: list = []
    loader.loaded.connect(
        lambda assets, description, request_id, device: answers.append(request_id)
    )
    loader.failed.connect(lambda item, request_id: answers.append(("failed", request_id)))
    try:
        first = loader.load(SceneSource(path=path, kind="package", directory=path.parent))
        second = loader.load(SceneSource(path=path, kind="package", directory=path.parent))
        assert answers == [], "a request was answered before its worker ran"
        loader._executor.run_next()
        assert answers == [first], "the pending request did not start after the first"
        assert loader._pending is None
        loader._executor.run_next()
        assert answers == [first, second]
    finally:
        loader.close()


def test_the_pane_ignores_a_completion_for_a_superseded_load(qt_app):
    """The other half: an old answer must not mark the *new* selection as loaded."""
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    panel = Mesh3dPanel()
    view = _RecordingView()
    panel.attach_view(view)
    assets = synthetic.poc_scene()
    panel._load_request = 2
    panel._on_loaded(assets, "old.oryscene", 1, None)
    assert view.assets is None, "a completion for an older request was adopted"
    panel._on_loaded(assets, "new.oryscene", 2, None)
    assert view.assets is assets
    assert view.scenes_used[-1][1] == "new.oryscene"


def test_a_failed_load_can_be_retried(qt_app, tmp_path, write_poc):
    """A rejection is not a dead end: the next request still runs."""
    from orisys_studio.scene3d.asset_source import SceneAssetLoader

    broken = tmp_path / f"broken{PACKAGE_SUFFIX}"
    broken.write_bytes(b"not a zip archive at all")
    good = write_poc()
    loader = SceneAssetLoader()
    loader._executor = _ManualExecutor()
    answers: list = []
    loader.loaded.connect(lambda assets, description, request_id, device: answers.append(request_id))
    loader.failed.connect(lambda item, request_id: answers.append(("failed", request_id)))
    try:
        bad = loader.load(SceneSource(path=broken, kind="package", directory=tmp_path))
        loader._executor.run_next()
        assert answers == [("failed", bad)]
        again = loader.load(SceneSource(path=good, kind="package", directory=good.parent))
        loader._executor.run_next()
        assert answers[-1] == again
    finally:
        loader.close()


def test_a_corrupt_package_is_reported_not_raised(qt_app, tmp_path):
    from orisys_studio.scene3d.asset_source import SceneAssetLoader

    broken = tmp_path / f"broken{PACKAGE_SUFFIX}"
    broken.write_bytes(b"not a zip archive at all")
    loader = SceneAssetLoader()
    failed: list = []
    loader.failed.connect(failed.append)
    try:
        loader.load(SceneSource(path=broken, kind="package", directory=tmp_path))
        assert _pump(qt_app, lambda: bool(failed)), "a corrupt package produced no answer"
    finally:
        loader.close()
    assert failed[0].code in ("asset_rejected", "asset_auth_failed")
    assert failed[0].message


def test_loading_a_missing_package_raises_a_diagnostic(tmp_path):
    from orisys_app_shared.scene3d.contracts import ThreeDError

    with pytest.raises(ThreeDError) as excinfo:
        load_package(SceneSource(path=tmp_path / "gone.oryscene", kind="package", directory=tmp_path))
    assert excinfo.value.diagnostic.code == "asset_rejected"


def test_the_missing_asset_message_is_not_localised_but_the_pane_is(tmp_path):
    """Diagnostics are logged and quoted in support tickets: they stay English."""
    diagnostic = no_asset_diagnostic("finger")
    assert diagnostic.code == "asset_rejected"
    assert PACKAGE_SUFFIX in diagnostic.message
    assert "compile_scene.py" in diagnostic.details["compile"]


def test_a_finger_scene_uses_the_finger_adapter(tmp_path, write_poc, qt_app):
    """The adapter is chosen by the asset, and a box scene without a device refuses."""
    from orisys_studio.scene3d.view import Scene3DView

    view = Scene3DView()
    try:
        if not view.available:
            pytest.skip("no GL binding here")
        assert view.use_scene(synthetic.poc_scene(), source_id="poc")
        assert view.assets is not None
    finally:
        view.close()


# ------------------------------------------- stage 6: a field that is not this frame's
def _stale_scene_view(qt_app):
    """A real view with the synthetic scene, or a skip when GL is not buildable."""
    from orisys_studio.scene3d.view import Scene3DView

    view = Scene3DView()
    if not view.available:
        view.close()
        pytest.skip("no GL binding here")
    if not view.use_scene(synthetic.poc_scene(), source_id="poc-live"):
        view.close()
        pytest.skip("the synthetic scene was refused")
    view.set_enabled(True)
    return view


def test_a_packet_that_cannot_claim_its_field_marks_the_measurement_stale(qt_app):
    """R2 through the view: the refusal used to be dropped on the floor.

    ``snapshot()`` returning ``None`` left the pane claiming ``live`` about a picture
    nothing was producing. Now the reason travels to the coordinator *and* back out
    to the pane.
    """
    view = _stale_scene_view(qt_app)
    try:
        packet = ui_check._fake_packet(1)
        packet.flow_valid = False
        packet.invalid_reason = "compute: RuntimeError: boom"
        view.submit_packet(packet)

        coordinator = view.coordinator
        assert coordinator is not None
        assert coordinator.state == "stale", "a refused field kept reporting live"
        assert coordinator.last_diagnostic is not None
        assert coordinator.last_diagnostic.code == "field_stale"

        view.tick()
        assert view.last_diagnostic is not None
        assert view.last_diagnostic.code == "field_stale", (
            "the pane never learned why the measurement stopped"
        )
        assert view.panel is not None
        assert view.panel.state_label.text() not in ("", "scene3d.state_stale"), (
            "the stale state has no operator-facing sentence"
        )
    finally:
        view.close()


def test_an_unstated_field_is_refused_rather_than_assumed(qt_app):
    view = _stale_scene_view(qt_app)
    try:
        packet = ui_check._fake_packet(1)
        packet.flow_valid = None
        view.submit_packet(packet)
        coordinator = view.coordinator
        assert coordinator is not None
        assert coordinator.state == "stale"
        assert coordinator.last_diagnostic is not None
        assert coordinator.last_diagnostic.code == "field_unverified"
    finally:
        view.close()


def test_a_fresh_field_reaches_the_screen_and_clears_the_stale_state(qt_app):
    view = _stale_scene_view(qt_app)
    try:
        shape = tuple(view.assets.shape_hw)
        packet = ui_check._fake_packet(1)
        packet.flow = np.full((*shape, 2), 0.5, dtype=np.float32)
        view.submit_packet(packet)
        assert view._snapshot_diagnostic() is None, "a stated, fresh field was refused"
        coordinator = view.coordinator
        for _ in range(60):
            view.tick()
            if coordinator is not None and coordinator.state == "live":
                break
            time.sleep(0.01)
        assert coordinator is not None and coordinator.state == "live", coordinator.state
    finally:
        view.close()


# ---------------------------------- stage 6: the adapter speaks for the *device* (R3)
def test_the_adapter_identity_comes_from_the_device_not_the_asset(qt_app):
    """R3: built from the asset, "the frame matches the scene" is self-certified."""
    from orisys_app_shared.scene3d.contracts import ExpectedGeometry
    from orisys_studio.scene3d.view import Scene3DView

    view = Scene3DView()
    try:
        if not view.available:
            pytest.skip("no GL binding here")
        assets = replace(synthetic.poc_scene(), source_geometry_id="device-src")
        device = ExpectedGeometry(
            "finger",
            "rev",
            "",
            assets.field_domain_id,
            assets.shape_hw,
            source_geometry_id="device-src",
        )
        assert view.use_scene(assets, source_id="poc", expected=device)
        described = view.describe()["scene"]
        assert described["device_verified"] is True, described
        assert described["device"]["source_geometry_id"] == "device-src"
        assert described["source_geometry_id"] == "device-src"

        adapter = view._adapter
        assert adapter is not None
        # The device stated no metre-scaled id, so the asset's is used *and* the pane
        # is only verified on the identities the device did state.
        assert adapter.geometry_id == assets.geometry_id
        assert adapter.field_domain_id == assets.field_domain_id
        assert adapter.shape_hw == tuple(assets.shape_hw)
    finally:
        view.close()


def test_a_device_whose_domain_disagrees_refuses_the_scene(qt_app):
    from orisys_app_shared.scene3d.contracts import ExpectedGeometry
    from orisys_studio.scene3d.view import Scene3DView

    view = Scene3DView()
    try:
        if not view.available:
            pytest.skip("no GL binding here")
        assets = synthetic.poc_scene()
        device = ExpectedGeometry(
            "finger", "", "", "somebody-elses-domain", assets.shape_hw
        )
        assert view.use_scene(assets, source_id="poc", expected=device) is False
        assert view.assets is None
        assert view.last_diagnostic is not None
        assert view.last_diagnostic.code == "field_domain_mismatch"
    finally:
        view.close()


def test_a_scene_without_a_device_description_is_shown_as_unverified(qt_app):
    """Expected unknown: show the surface, but never claim it was checked."""
    from orisys_studio.scene3d.view import Scene3DView

    view = Scene3DView()
    try:
        if not view.available:
            pytest.skip("no GL binding here")
        assert view.use_scene(synthetic.poc_scene(), source_id="poc")
        described = view.describe()["scene"]
        assert described["device_verified"] is False
        assert view.last_diagnostic is not None
        assert view.last_diagnostic.code == "geometry_unverified"
    finally:
        view.close()


# ------------------------------------------------------- stage 5: the record block
def test_a_recording_says_what_the_3d_pane_was(
    make_window, next_record_root, write_poc, qt_app, monkeypatch
):
    """``session.json`` carries the pane's state, and an event carries the totals."""
    import json

    import orisys_studio.app_window as app_module

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: next_record_root()
    try:
        # A real view on a machine with no GL is not possible; the stand-in gives the
        # metadata assembly something to describe (an enabled pane with a scene).
        view = _RecordingView()
        view.assets = None
        window.mesh3d_panel.attach_view(view)
        window.toolbar.btn_record.setChecked(True)
        for index in range(20):
            stuck.emit(ui_check._fake_packet(index, mono_t=index / 30.0))
        window._on_run_tick()
        window.toolbar.btn_record.setChecked(False)
        report = window._record_last_report
        assert report is not None and report.completed or report is not None
        if report is None or not report.completed:
            pytest.skip("the writer had not finished on this machine")
        meta = json.loads((report.directory / "session.json").read_text(encoding="utf-8"))
        block = meta["visualization3d"]
        assert block["block_version"] == 1
        assert block["enabled"] is False, (
            "the pane was not switched on, so the record must say so"
        )
        # The pane has no scene here (no package installed), so the block names the
        # reason instead of pretending to describe a picture.
        assert block.get("reason")
        events = (report.directory / "events.csv").read_text(encoding="utf-8")
        assert "visualization3d" in events, (
            "the display totals were not recorded as an event"
        )
    finally:
        _finish(window, stuck)
        window.close()


def test_the_replay_view_shows_the_pane_and_says_where_the_field_comes_from(
    make_window, replay_recording, qt_app
):
    """History browsing gets the pane, fed by the recompute pass — and says so.

    A recording stores the raw camera image and the forces, not a per-frame
    deformation field, so the pane in replay is either the surface model (no field) or
    what a recompute pass produces from the video. Those are two different claims and
    the label has to distinguish them.
    """
    from tests.test_ui_smoke import _show

    window = _show(make_window(sensor_type="finger"), (1440, 880))
    try:
        assert window._load_recording(Path(replay_recording))
        assert window._view_preset == "replay"
        pane = window.mesh3d_panel
        assert pane.isVisible(), "the 3D pane is hidden in the replay view"
        assert pane.enabled(), "the pane is visible but not switched on"
        assert window.content_splitter.sizes()[0] >= MIN_PANEL_WIDTHS["mesh3d"], (
            "the replay layout left the pane at zero width"
        )
        view = pane.view
        if view is None or view.panel is None:
            pytest.skip("no 3D view on this platform")
        # Wait for the scene to load, then read what the pane claims.
        assert _pump(qt_app, lambda: view.assets is not None, timeout_s=15.0), (
            "the scene never loaded: " + str(pane.last_diagnostic)
        )
        label = view.panel.source_label.text()
        assert "回放" in label or "Replay" in label, label

        # A running pass means the pixels are recomputed, and the label must say so.
        window._recompute_running = True
        window._apply_panel_visibility()
        label = view.panel.source_label.text()
        assert "重算" in label or "Recomputed" in label, label

        # Stopping the pass must clear the picture, not leave the last frame up.
        window._recompute_running = False
        pane.show_reference_only()
        qt_app.processEvents()
        assert view.coordinator.state in ("no_data", "ready", "disabled")
        assert "没有测量值" in view.panel.legend_label.text() or (
            "No measurement" in view.panel.legend_label.text()
        ), view.panel.legend_label.text()
        window._apply_panel_visibility()
        label = view.panel.source_label.text()
        assert "回放" in label or "Replay" in label, label
    finally:
        window.close()


def test_the_live_view_still_says_live(make_window, qt_app):
    window = make_window(sensor_type="finger")
    try:
        window._apply_panel_visibility()
        view = window.mesh3d_panel.view
        if view is None or view.panel is None:
            pytest.skip("no 3D view on this platform")
        label = view.panel.source_label.text()
        assert "实时" in label or "Live" in label, label
        assert view.panel.source_label.toolTip()
    finally:
        window.close()


def test_every_run_gets_a_new_generation(make_window, qt_app):
    """The window tells the pane when a session starts (see the audit's frozen-pane bug).

    Without this the pane accepts the first run's frames and refuses every later run's,
    so the picture stays on the first run's last frame forever.
    """
    from orisys_app_shared.scene3d.contracts import FrameIdentity
    from tests.test_ui_smoke import _show
    import ui_check

    window = _show(make_window(sensor_type="finger"), (1440, 880))
    try:
        view = window.mesh3d_panel.view
        if view is None or view.coordinator is None:
            pytest.skip("no 3D view on this platform")
        assert _pump(qt_app, lambda: view.assets is not None, timeout_s=15.0)

        stuck = ui_check._StuckWorker()
        window._ensure_workers = lambda: None
        window._box_worker = stuck
        window._finger_worker = stuck
        try:
            generations = []
            for _ in range(2):
                window._start_run()
                window._run_timer.stop()
                generations.append(view.coordinator.generation)
                window._stop_run(reason="user_stop")
            assert generations[0] < generations[1], generations

            # And the frames of the second session are accepted, not refused as stale.
            identity = FrameIdentity(
                source_id=view._source_id,
                session_id=window._session_counter,
                frame_id=1,
                t_rel_s=0.0,
                generation=view.coordinator.generation,
            )
            shape = tuple(view.assets.shape_hw)
            packet = ui_check._fake_packet(1)
            packet.flow = np.full((*shape, 2), 0.5, dtype=np.float32)
            view.submit_packet(packet)
            assert view._snapshot_diagnostic() is None, (
                "the adapter refused the second session's packet"
            )
            assert view.coordinator.stale_frames == 0, (
                "the second session's frames were refused as stale"
            )
            del identity
        finally:
            stuck.release()
    finally:
        window.close()


def test_a_platform_locked_scene_is_refused_in_the_application(qt_app):
    """A lock violation is a *pane* failure: the view refuses the scene and says why."""
    from orisys_app_shared.scene3d.contracts import PlatformInfo, PlatformLock
    from orisys_studio.scene3d.view import Scene3DView
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    here = PlatformInfo.here()
    elsewhere = "linux" if here.os_family != "linux" else "windows"
    locked = replace(
        synthetic.poc_scene(),
        platform_lock=PlatformLock(os_family=(elsewhere,), machine=("aarch64",)),
    )
    view = Scene3DView()
    try:
        if not view.available:
            pytest.skip("no GL binding here")
        reported: list = []
        view.diagnosticRaised.connect(reported.append)
        assert view.use_scene(locked, source_id="locked") is False
        assert view.assets is None, "a refused scene must not remain adopted"
        assert any(item.code == "platform_mismatch" for item in reported), reported
    finally:
        view.close()

    panel = Mesh3dPanel()
    try:
        # The pane shows the reason rather than an empty picture: `platform_mismatch`
        # is one of the terminal diagnostics.
        panel._on_diagnostic(
            __import__(
                "orisys_app_shared.scene3d.contracts", fromlist=["diagnostic"]
            ).diagnostic("platform_mismatch", "this package is not released for this OS")
        )
        assert panel.showing_message
        assert "not released" in panel.message.text()
    finally:
        panel.deleteLater()