ykcheungab
06e06126ca
选点、对准、加载与录像分散在两个面板，状态靠隐藏相位判断。现改由不依赖 Qt 的采集控制器驱动，并加入录像选项、汇总平铺副本、保持到时自动结束、会话结束回到空闲，以及无设备时的模拟采集。已运行 Autocal 非 Qt 测试并通过范围内门禁；界面测试待 Windows 复核，未做实机验收。
265 lines
9.9 KiB
Python
265 lines
9.9 KiB
Python
"""Publishing a summary: one set of files, one pointer, never a partial result."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from evidence_helpers import multipoint_session
|
|
from orisys_autocal.summary import build_summary
|
|
from orisys_autocal.summary_export import (
|
|
POINTER_FORMAT,
|
|
POINTER_NAME,
|
|
SummaryExportError,
|
|
load_summary,
|
|
published_summary_directory,
|
|
publish_summary,
|
|
render_html,
|
|
)
|
|
|
|
|
|
def _sha256(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
@pytest.fixture
|
|
def session(tmp_path):
|
|
root = multipoint_session(
|
|
tmp_path / "rec", sites=("P1", "P2"), repeats=2, levels=(5.0, 10.0))
|
|
return root
|
|
|
|
|
|
def _pointer(root):
|
|
return json.loads((root / POINTER_NAME).read_text(encoding="utf-8"))
|
|
|
|
|
|
def _read_csv(path):
|
|
with path.open(encoding="utf-8-sig", newline="") as handle:
|
|
return list(csv.DictReader(handle))
|
|
|
|
|
|
def test_publishing_writes_all_four_files_and_moves_the_pointer(session):
|
|
summary = build_summary(session)
|
|
|
|
published = publish_summary(session, summary)
|
|
|
|
assert published.is_dir()
|
|
assert published.name == f"summary_{summary['summary_id']}"
|
|
for name in ("summary.json", "summary.html", "baseline_load_data.csv",
|
|
"sensitivity_values.csv"):
|
|
assert (published / name).is_file(), name
|
|
pointer = _pointer(session)
|
|
assert pointer["format_version"] == POINTER_FORMAT
|
|
assert pointer["summary_id"] == summary["summary_id"]
|
|
assert pointer["path"] == f"derived/{published.name}"
|
|
assert set(pointer["files"]) == {
|
|
"summary.json", "summary.html", "baseline_load_data.csv", "sensitivity_values.csv"}
|
|
loaded, pointer_again = load_summary(session)
|
|
assert loaded["summary_id"] == summary["summary_id"]
|
|
assert pointer_again["summary_id"] == summary["summary_id"]
|
|
# The export is a snapshot of the same object, not a second computation.
|
|
stored = json.loads((published / "summary.json").read_text(encoding="utf-8"))
|
|
assert stored["windows"] == summary["windows"]
|
|
assert stored["sensitivity"] == summary["sensitivity"]
|
|
|
|
|
|
def test_the_html_is_self_contained_and_escapes_operator_text(session):
|
|
summary = build_summary(session)
|
|
summary = dict(summary)
|
|
summary["sensor_sn"] = "<script>alert('x')</script>"
|
|
|
|
document = render_html(summary)
|
|
|
|
assert "<script>" not in document
|
|
assert "<script>" in document
|
|
assert "http://" not in document and "https://" not in document
|
|
assert "not a pass/fail certificate" in document
|
|
|
|
|
|
def test_a_write_failure_leaves_the_previous_pointer_in_place(session, monkeypatch):
|
|
import orisys_autocal.summary_export as module
|
|
|
|
first = build_summary(session)
|
|
publish_summary(session, first)
|
|
before = (session / POINTER_NAME).read_text(encoding="utf-8")
|
|
|
|
second = build_summary(session)
|
|
second["generated_at"] = "2999-01-01T00:00:00+00:00"
|
|
|
|
def explode(*args, **kwargs):
|
|
raise OSError("no space left on device")
|
|
|
|
monkeypatch.setattr(module, "_write_csv", explode)
|
|
with pytest.raises(SummaryExportError, match="could not publish"):
|
|
publish_summary(session, second)
|
|
|
|
assert (session / POINTER_NAME).read_text(encoding="utf-8") == before
|
|
assert not (session / "derived" / f"summary_{second['summary_id']}").exists()
|
|
leftovers = [path for path in (session / "derived").iterdir() if path.name.startswith(".tmp-")]
|
|
assert leftovers == []
|
|
|
|
|
|
def test_the_pointer_never_moves_back_to_an_older_summary(session):
|
|
first = build_summary(session)
|
|
first["generated_at"] = "2026-10-18T00:00:00+00:00"
|
|
publish_summary(session, first)
|
|
older = build_summary(session)
|
|
older["generated_at"] = "2026-10-17T00:00:00+00:00"
|
|
|
|
with pytest.raises(SummaryExportError, match="back"):
|
|
publish_summary(session, older)
|
|
|
|
assert _pointer(session)["summary_id"] == first["summary_id"]
|
|
|
|
|
|
def test_changed_inputs_refuse_to_publish(session):
|
|
summary = build_summary(session)
|
|
forces = session / "sites" / "P1" / "repeat-001" / "capture" / "forces.csv"
|
|
text = forces.read_text(encoding="utf-8-sig")
|
|
forces.write_text(text + "\n", encoding="utf-8")
|
|
|
|
with pytest.raises(SummaryExportError, match="changed"):
|
|
publish_summary(session, summary)
|
|
assert not (session / POINTER_NAME).exists()
|
|
|
|
|
|
def test_a_cancelled_export_writes_nothing(session):
|
|
summary = build_summary(session)
|
|
with pytest.raises(SummaryExportError, match="cancelled"):
|
|
publish_summary(session, summary, cancelled=lambda: True)
|
|
assert not (session / POINTER_NAME).exists()
|
|
assert not (session / "derived").exists()
|
|
|
|
|
|
def test_the_same_summary_is_not_published_twice(session):
|
|
summary = build_summary(session)
|
|
publish_summary(session, summary)
|
|
with pytest.raises(SummaryExportError, match="already published"):
|
|
publish_summary(session, summary)
|
|
|
|
|
|
def test_load_refuses_a_tampered_export(session):
|
|
summary = build_summary(session)
|
|
published = publish_summary(session, summary)
|
|
(published / "sensitivity_values.csv").write_text("changed", encoding="utf-8")
|
|
|
|
with pytest.raises(SummaryExportError, match="modified"):
|
|
load_summary(session)
|
|
|
|
|
|
def test_the_csv_keeps_full_precision_and_utf8_text(session):
|
|
summary = build_summary(session)
|
|
summary["windows"][1]["FN_mean"] = 40.1234567890123
|
|
summary["windows"][1]["site_id"] = "测点一"
|
|
|
|
published = publish_summary(session, summary)
|
|
|
|
rows = _read_csv(published / "baseline_load_data.csv")
|
|
assert rows[1]["FN_mean"] == repr(40.1234567890123)
|
|
assert float(rows[1]["FN_mean"]) == pytest.approx(40.1234567890123)
|
|
assert rows[1]["site_id"] == "测点一"
|
|
assert rows[1]["unit"] == "raw"
|
|
assert rows[1]["window_method"] == "hold_second_half_v1"
|
|
assert rows[1]["eligible"] == "1"
|
|
assert rows[1]["samples_FN"] == "2"
|
|
|
|
|
|
def test_the_sensitivity_csv_has_one_row_per_point_trial_axis(session):
|
|
summary = build_summary(session)
|
|
published = publish_summary(session, summary)
|
|
|
|
rows = _read_csv(published / "sensitivity_values.csv")
|
|
|
|
assert len(rows) == len(summary["sensitivity"]) == 2 * 2 * 3
|
|
assert {row["component"] for row in rows} == {"FN", "FX", "FY"}
|
|
assert {row["method"] for row in rows} == {"raw_on_reference_ols_v1"}
|
|
assert {row["review_status"] for row in rows} == {"candidate"}
|
|
fn = [row for row in rows if row["component"] == "FN"]
|
|
assert all(float(row["slope_raw_per_N"]) == pytest.approx(40.0) for row in fn)
|
|
|
|
|
|
def test_the_published_files_are_mirrored_under_the_session_root(session):
|
|
"""The folder the operator opens is the byte-identical copy the pointer names."""
|
|
summary = build_summary(session)
|
|
|
|
published = publish_summary(session, summary)
|
|
|
|
pointer = _pointer(session)
|
|
flat = session / "summary"
|
|
assert flat.is_dir()
|
|
assert sorted(path.name for path in flat.iterdir()) == sorted(pointer["files"])
|
|
for name, digest in pointer["files"].items():
|
|
assert _sha256(flat / name) == digest, name
|
|
assert (flat / name).read_bytes() == (published / name).read_bytes(), name
|
|
# The readable copy is not a second pointer: the authoritative path is derived/.
|
|
assert pointer["path"] == f"derived/{published.name}"
|
|
assert published_summary_directory(session) == flat
|
|
assert load_summary(session)[1]["summary_id"] == summary["summary_id"]
|
|
|
|
|
|
def test_a_failed_copy_reports_without_moving_the_pointer_back(session, monkeypatch, caplog):
|
|
"""Losing the readable copy must not lose the published evidence or the pointer."""
|
|
import orisys_autocal.summary_export as module
|
|
|
|
first = build_summary(session)
|
|
publish_summary(session, first)
|
|
flat = session / "summary"
|
|
before = {path.name: _sha256(path) for path in flat.iterdir()}
|
|
first_pointer = _pointer(session)
|
|
|
|
second = build_summary(session)
|
|
second["generated_at"] = "2999-01-01T00:00:00+00:00"
|
|
reports: list[str] = []
|
|
|
|
def explode(*args, **kwargs):
|
|
raise OSError("no space left on device")
|
|
|
|
monkeypatch.setattr(module, "_write_flat_summary", explode)
|
|
with caplog.at_level(logging.WARNING, logger="orisys_autocal.summary_export"):
|
|
published = publish_summary(session, second, on_warning=reports.append)
|
|
|
|
pointer = _pointer(session)
|
|
assert pointer["summary_id"] == second["summary_id"]
|
|
assert pointer["path"] == f"derived/{published.name}"
|
|
assert first_pointer["path"] != pointer["path"]
|
|
for name, digest in pointer["files"].items():
|
|
assert _sha256(published / name) == digest, name
|
|
# The previous readable copy still stands, unchanged.
|
|
assert {path.name: _sha256(path) for path in flat.iterdir()} == before
|
|
assert reports and "no space left on device" in reports[0]
|
|
assert any("no space left on device" in record.message for record in caplog.records)
|
|
|
|
|
|
def test_a_stale_readable_copy_is_not_offered_as_the_current_summary(session, monkeypatch):
|
|
"""A failed mirror keeps the older copy; opening it must not pass it off as current.
|
|
|
|
The pointer has moved on, so ``summary/`` describes a superseded version. The
|
|
operator must be sent to the directory the pointer names, not to numbers the
|
|
pointer no longer claims.
|
|
"""
|
|
import orisys_autocal.summary_export as module
|
|
|
|
first = build_summary(session)
|
|
publish_summary(session, first)
|
|
|
|
second = build_summary(session)
|
|
second["generated_at"] = "2999-01-01T00:00:00+00:00"
|
|
|
|
def explode(*args, **kwargs):
|
|
raise OSError("no space left on device")
|
|
|
|
monkeypatch.setattr(module, "_write_flat_summary", explode)
|
|
published = publish_summary(session, second)
|
|
|
|
pointer = _pointer(session)
|
|
assert pointer["summary_id"] == second["summary_id"]
|
|
opened = published_summary_directory(session)
|
|
assert opened is not None
|
|
assert opened == published, "the stale summary/ was offered as the current summary"
|
|
stored = json.loads((opened / "summary.json").read_text(encoding="utf-8"))
|
|
assert stored["summary_id"] == second["summary_id"]
|