orisys-app/apps/autocal/tests/test_summary_lifecycle.py

"""The candidate export, the reviewed export, and what happens when it fails."""

from __future__ import annotations

import hashlib
import time

from PySide6.QtCore import QCoreApplication

from review_test_helpers import wait_review

from evidence_helpers import multipoint_session
from orisys_autocal.summary_export import POINTER_NAME, load_summary


def _pump(panel, timeout: float = 30.0) -> None:
    deadline = time.monotonic() + timeout
    while panel.summary_publishing and time.monotonic() < deadline:
        QCoreApplication.processEvents()
        time.sleep(0.005)
    assert not panel.summary_publishing, "the summary export never finished"


def _finish_simulated_campaign(campaign) -> None:
    while campaign.phase in {"load", "hold"}:
        if campaign.phase == "load":
            branch, target = campaign._steps[campaign._step_index]
            campaign.confirm_reference(target)
        campaign.complete_hold(
            now_ns=campaign._hold_started_ns + int(campaign._hold_s * 1e9)
        )


def _acquired_hashes(root) -> dict[str, str]:
    """Hashes of the recorded evidence only: derived results are not evidence."""
    return {
        str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
        for path in root.rglob("*")
        if path.is_file() and "derived" not in path.parts and path.name != POINTER_NAME
    }


def test_closing_the_session_publishes_a_candidate_summary(make_window, tmp_path):
    root = multipoint_session(
        tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0, 10.0))
    before = _acquired_hashes(root)
    window = make_window()
    panel = window.calibration_panel

    assert panel.publish_summary_for(root)
    _pump(panel)

    assert (root / POINTER_NAME).is_file()
    summary, pointer = load_summary(root)
    assert summary["formal_conclusion"] is False
    assert summary["windows"]
    assert all(row["review_status"] == "candidate" for row in summary["windows"])
    assert pointer["path"].startswith("derived/summary_")
    # Raw evidence is never rewritten by an export.
    assert _acquired_hashes(root) == before
    window.close()


def test_a_failed_export_is_reported_and_can_be_retried(
    make_window, tmp_path, monkeypatch
):
    import orisys_autocal.ui.calibration_panel as panel_module

    root = multipoint_session(tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0,))
    window = make_window()
    panel = window.calibration_panel

    def explode(*args, **kwargs):
        raise RuntimeError("disk full")

    monkeypatch.setattr(panel_module, "publish_summary", explode)
    assert panel.publish_summary_for(root)
    _pump(panel)

    assert not (root / POINTER_NAME).exists()
    assert "disk full" in panel.result_label.text()
    assert panel.trial.status == "idle"  # the acquisition result is untouched

    monkeypatch.undo()
    assert panel.publish_summary_for(root)
    _pump(panel)
    assert (root / POINTER_NAME).is_file()
    window.close()


def test_a_published_candidate_is_republished_when_the_review_is_saved(qt_app, tmp_path):
    from orisys_autocal.session_review import review_session
    from orisys_autocal.ui.record_overview_dialog import RecordOverviewDialog

    root = multipoint_session(
        tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0, 10.0))
    review = review_session(root)
    window = review["sites"][0]["windows"][2]
    selections = {f"{window['trial_id']}:{window['step_index']}": {
        "t_start_host_mono_ns": window["hold_start_host_mono_ns"] + 1,
        "t_end_host_mono_ns": window["hold_end_host_mono_ns"],
        "stability_verified": True,
    }}
    reviewed = review_session(root, selections=selections)
    dialog = RecordOverviewDialog(root, review=reviewed)
    try:
        dialog._save()
        wait_review(dialog)
        assert (root / POINTER_NAME).is_file()
        summary, pointer = load_summary(root)
        confirmed = [row for row in summary["windows"] if row["stability_verified"]]
        assert confirmed and confirmed[0]["review_status"] == "confirmed"
        assert pointer["summary_id"] == summary["summary_id"]
        assert summary["review_relative_path"].startswith("derived/review_")
        assert summary["sensitivity"]
        # A second explicit publish (the retry action) makes a new version.
        first_id = pointer["summary_id"]
        dialog._publish_summary()
        wait_review(dialog)
        second, pointer_again = load_summary(root)
        assert pointer_again["summary_id"] != first_id
        assert second["summary_id"] == pointer_again["summary_id"]
    finally:
        dialog.close()


def test_the_multipoint_end_of_campaign_starts_the_export(make_window, tmp_path):
    window = make_window()
    panel = window.calibration_panel
    multipoint = panel.multipoint
    multipoint.campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
    directory = multipoint.campaign.start(
        tmp_path, sensor_model="finger", sensor_sn="SN1",
        coordinate_frame="sensor", max_force_n=20, simulate=True,
    )
    _finish_simulated_campaign(multipoint.campaign)
    assert multipoint.campaign.phase == "done"

    multipoint._poll()

    _pump(panel)
    assert (directory / POINTER_NAME).is_file()
    summary, _pointer = load_summary(directory)
    assert summary["session_id"] == directory.name
    # A simulated campaign produces no frames, so its windows must not read as
    # measurements: they are excluded with the reason the review recorded.
    assert all(row["eligible"] is False for row in summary["windows"])
    assert all(row["excluded_reason"] == "no_frame_source" for row in summary["windows"])
    assert any(issue["reason"] == "no_frame_source" for issue in summary["issues"])
    window.close()