From ac7317fee4da79ff93a9366f7efc8af876b29f8e Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sun, 12 Jul 2026 16:31:49 -0700 Subject: [PATCH 1/3] test: simulation-mode acquisition acceptance suite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit End-to-end acceptance tests that drive MultiPointController directly against simulated hardware (in-process, no GUI, no QApplication) and assert on observable artifacts only: image files, coordinates.csv rows, 'acquisition parameters.json', .done markers, the per-acquisition log, and subprocess state. New tests/acceptance/ package (pytest marker: acceptance, registered in its conftest; runs under the default pytest invocation, so CI picks it up with no workflow change): - test_full_acquisition.py — full small acquisition (2x2 FOV x 2ch x 3z = 24 images + CSV/JSON/log assertions); OME_TIFF vs ZARR_V3 format parity (16 planes, identical plane shape/dtype; zarr read back via tensorstore, skipped where not installed); 2-timepoint smoke. - test_abort.py — abort mid-acquisition: clean stop, a second acquisition succeeds in the same process, and no JobRunner subprocess survives close() (PR #582 regression surface). - test_backpressure.py — z-stack under a ~1.5-frame byte limit completes instead of deadlocking (pins the byte-release-on-job- completion design; throttling verified engaged via logs). - harness.py/conftest.py — shared AcquisitionHarness around tests/control/test_stubs.py wiring, artifact helpers, generous CI-safe timeouts, def-knob pinning (incl. patching the worker's star-imported FILE_SAVING_OPTION binding). Suite runtime: 5 passed in ~28s locally. Co-Authored-By: Claude Fable 5 --- software/tests/acceptance/__init__.py | 0 software/tests/acceptance/conftest.py | 93 ++++++++ software/tests/acceptance/harness.py | 146 ++++++++++++ software/tests/acceptance/test_abort.py | 129 +++++++++++ .../tests/acceptance/test_backpressure.py | 99 ++++++++ .../tests/acceptance/test_full_acquisition.py | 214 ++++++++++++++++++ 6 files changed, 681 insertions(+) create mode 100644 software/tests/acceptance/__init__.py create mode 100644 software/tests/acceptance/conftest.py create mode 100644 software/tests/acceptance/harness.py create mode 100644 software/tests/acceptance/test_abort.py create mode 100644 software/tests/acceptance/test_backpressure.py create mode 100644 software/tests/acceptance/test_full_acquisition.py diff --git a/software/tests/acceptance/__init__.py b/software/tests/acceptance/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/software/tests/acceptance/conftest.py b/software/tests/acceptance/conftest.py new file mode 100644 index 000000000..48861c9e4 --- /dev/null +++ b/software/tests/acceptance/conftest.py @@ -0,0 +1,93 @@ +""" +Fixtures for the simulation-mode acquisition acceptance suite. + +These tests run full acquisitions end-to-end against simulated hardware and +assert on observable artifacts (files on disk, process state), not internals. +They are headless-safe: no QApplication is created (MultiPointController uses +plain threads and plain-callable callbacks), so they run under Xvfb in CI. +""" + +import logging +from unittest.mock import patch + +import pytest + +import control._def +import control.core.multi_point_worker +import control.microcontroller + +logger = logging.getLogger(__name__) + + +def pytest_configure(config): + config.addinivalue_line( + "markers", + "acceptance: end-to-end simulation-mode acquisition acceptance tests", + ) + + +def _make_tracking_init(original_init, instances_list): + """Create a wrapper that tracks Microcontroller instances.""" + + def _tracking_init(self, *args, **kwargs): + original_init(self, *args, **kwargs) + instances_list.append(self) + + return _tracking_init + + +@pytest.fixture(autouse=True) +def cleanup_microcontrollers(): + """ + Automatically close all Microcontroller instances after each test. + + Same rationale as tests/control/conftest.py: the packet-reading background + thread must be stopped or subsequent tests can segfault. + """ + active_microcontrollers = [] + original_init = control.microcontroller.Microcontroller.__init__ + + with patch.object( + control.microcontroller.Microcontroller, + "__init__", + _make_tracking_init(original_init, active_microcontrollers), + ): + yield + + for micro in active_microcontrollers: + try: + if hasattr(micro, "terminate_reading_received_packet_thread"): + if not micro.terminate_reading_received_packet_thread: + micro.close() + except Exception as e: + logger.warning(f"Failed to close Microcontroller in test cleanup: {e}") + + +@pytest.fixture(autouse=True) +def _watchdog_state_to_tmp(tmp_path, monkeypatch): + # Keep acquisition breadcrumbs out of the real user state dir during tests. + monkeypatch.setenv("SQUID_WATCHDOG_STATE_DIR", str(tmp_path / "watchdog")) + + +def set_file_saving_option(monkeypatch, option): + """ + Point acquisition file saving at `option` (a control._def.FileSavingOption). + + multi_point_worker star-imports FILE_SAVING_OPTION, freezing its own + binding at import time, while job_processing reads _def.FILE_SAVING_OPTION + at runtime — so both locations must be patched together. + """ + monkeypatch.setattr(control._def, "FILE_SAVING_OPTION", option) + monkeypatch.setattr(control.core.multi_point_worker, "FILE_SAVING_OPTION", option) + + +@pytest.fixture +def acquisition_defaults(monkeypatch): + """ + Pin the control._def knobs the acceptance scenarios depend on to known + values, restoring them afterwards. Individual tests override per-scenario + (e.g. FILE_SAVING_OPTION, backpressure limits) with the same monkeypatch. + """ + monkeypatch.setattr(control._def, "MERGE_CHANNELS", False) + set_file_saving_option(monkeypatch, control._def.FileSavingOption.INDIVIDUAL_IMAGES) + return monkeypatch diff --git a/software/tests/acceptance/harness.py b/software/tests/acceptance/harness.py new file mode 100644 index 000000000..0d2d4eba3 --- /dev/null +++ b/software/tests/acceptance/harness.py @@ -0,0 +1,146 @@ +""" +Shared harness for the simulation-mode acquisition acceptance suite. + +Drives MultiPointController directly (in-process, no GUI, no QApplication), +reusing the wiring helpers from tests/control/test_stubs.py. Assertions in the +scenario files operate on observable artifacts: files on disk and process +state. + +Timing thresholds are deliberately generous — CI runners are slow. +""" + +import csv +import threading +from dataclasses import dataclass, field +from pathlib import Path +from typing import List, Optional + +import control._def +import control.microscope +from control.core.multi_point_controller import MultiPointController + +import tests.control.test_stubs as ts +from control.core.multi_point_utils import MultiPointControllerFunctions + +# Generous CI-safe waits (seconds). +START_TIMEOUT_S = 60 +FINISH_TIMEOUT_S = 300 + + +class AcquisitionTracker: + """Collects acquisition callbacks so tests can wait on observable events.""" + + def __init__(self): + self.started_event = threading.Event() + self.finished_event = threading.Event() + self.first_image_event = threading.Event() + self.image_count = 0 + self._lock = threading.Lock() + + def get_callbacks(self) -> MultiPointControllerFunctions: + return MultiPointControllerFunctions( + signal_acquisition_start=lambda params: self.started_event.set(), + signal_acquisition_finished=lambda: self.finished_event.set(), + signal_new_image=self._receive_image, + signal_current_configuration=lambda config: None, + signal_current_fov=lambda x, y: None, + signal_overall_progress=lambda progress: None, + signal_region_progress=lambda progress: None, + ) + + def _receive_image(self, frame, info): + with self._lock: + self.image_count += 1 + self.first_image_event.set() + + +@dataclass +class AcquisitionHarness: + """A simulated microscope + MultiPointController pair with cleanup.""" + + scope: "control.microscope.Microscope" + mpc: MultiPointController + tracker: AcquisitionTracker + experiment_dirs: List[Path] = field(default_factory=list) + + def close(self): + try: + self.mpc.close() + finally: + self.scope.close() + + def new_experiment(self, base_path: Path, experiment_id: str) -> None: + """Start a fresh experiment; the timestamped directory it creates is + recorded in self.experiment_dirs. + + start_new_experiment() serializes 'acquisition parameters.json' from + the controller's CURRENT settings — configure NZ/Nt/deltaZ/channels + before calling this if the test asserts on that file's contents.""" + base_path.mkdir(parents=True, exist_ok=True) + before = set(base_path.iterdir()) + self.mpc.set_base_path(str(base_path)) + self.mpc.start_new_experiment(experiment_id) + created = [p for p in base_path.iterdir() if p not in before and p.is_dir()] + assert len(created) == 1, f"expected exactly one new experiment dir, found {created}" + self.experiment_dirs.append(created[0]) + + @property + def experiment_dir(self) -> Path: + assert self.experiment_dirs, "new_experiment() has not been called" + return self.experiment_dirs[-1] + + def add_fov_grid(self, region_id: str = "region0", nx: int = 2, ny: int = 2) -> None: + """Add an nx x ny FOV grid at a stage position guaranteed to be within + limits (coordinates outside stage limits are silently dropped).""" + stage_config = self.mpc.stage.get_config() + x = stage_config.X_AXIS.MIN_POSITION + 1.0 + y = stage_config.Y_AXIS.MIN_POSITION + 1.0 + z = (stage_config.Z_AXIS.MIN_POSITION + stage_config.Z_AXIS.MAX_POSITION) / 2.0 + self.mpc.scanCoordinates.add_flexible_region(region_id, x, y, z, nx, ny, 0) + fovs = self.mpc.scanCoordinates.region_fov_coordinates.get(region_id, []) + assert len(fovs) == nx * ny, f"region {region_id}: expected {nx * ny} FOVs, got {len(fovs)}" + + def select_channels(self, count: int) -> List[str]: + objective = self.scope.objective_store.current_objective + names = [c.name for c in self.mpc.liveController.get_channels(objective)] + assert len(names) >= count, f"need {count} channels, simulation config provides {names}" + selected = names[:count] + self.mpc.set_selected_configurations(selected) + return selected + + def run_and_wait(self, timeout_s: float = FINISH_TIMEOUT_S) -> None: + self.mpc.run_acquisition() + assert self.tracker.started_event.wait(START_TIMEOUT_S), "acquisition did not start" + assert self.tracker.finished_event.wait(timeout_s), f"acquisition did not finish within {timeout_s}s" + + +def make_harness() -> AcquisitionHarness: + """Build a fresh simulated microscope wired to a MultiPointController. + + Callers own cleanup: call harness.close() (use the `harness` fixture or + try/finally) so JobRunner subprocesses and semaphores are released. + """ + control._def.MERGE_CHANNELS = False + scope = control.microscope.Microscope.build_from_global_config(True) + tracker = AcquisitionTracker() + mpc = ts.get_test_multi_point_controller(microscope=scope, callbacks=tracker.get_callbacks()) + mpc.scanCoordinates.clear_regions() + return AcquisitionHarness(scope=scope, mpc=mpc, tracker=tracker) + + +def read_coordinates_csv(path: Path) -> List[dict]: + with open(path, newline="") as f: + return list(csv.DictReader(f)) + + +def timepoint_dir(experiment_dir: Path, timepoint: int) -> Path: + # FILE_ID_PADDING is 0 by default, so per-timepoint dirs are "0", "1", ... + return experiment_dir / str(timepoint).zfill(control._def.FILE_ID_PADDING) + + +def list_image_files(directory: Path) -> List[Path]: + """Individual-images mode: per-image files named + {region}_{fov}_{z}_{channel}. directly in the timepoint dir.""" + return sorted(p for p in directory.glob("*.tiff") if p.is_file()) + sorted( + p for p in directory.glob("*.bmp") if p.is_file() + ) diff --git a/software/tests/acceptance/test_abort.py b/software/tests/acceptance/test_abort.py new file mode 100644 index 000000000..a5da63db2 --- /dev/null +++ b/software/tests/acceptance/test_abort.py @@ -0,0 +1,129 @@ +""" +Acceptance scenario 3: abort mid-acquisition, then re-acquire in the same process. + +Drives MultiPointController directly against a simulated microscope (no GUI). A +multi-FOV / multi-channel / multi-Z acquisition is started with enough runway +that a mid-flight abort lands after the first image but before completion. We +then assert: + + * abort stops the run cleanly (finished callback fires, acquisition_in_progress + drops, fewer than the full image count were produced), + * the same process/harness can immediately run a second, smaller acquisition to + completion, and + * once the harness is closed, no orphaned JobRunner subprocesses linger. + +Context (PR #582): the app previously leaked orphaned JobRunner ``spawn_main`` +subprocesses because ``os._exit()`` skipped multiprocessing's atexit hook. +Per-acquisition cleanup now shuts runners down in non-blocking daemon threads; +this test pins that no JobRunner survives ``close()`` at the acceptance level. +""" + +import multiprocessing +import time + +import pytest + +from control.core.job_processing import JobRunner + +from tests.acceptance.harness import make_harness, timepoint_dir, list_image_files + +pytestmark = pytest.mark.acceptance + + +def _wait_until(predicate, timeout_s: float, interval_s: float = 0.2) -> bool: + """Poll ``predicate`` until it is truthy or the deadline passes.""" + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(interval_s) + return predicate() + + +def test_abort_mid_acquisition_then_reacquire(tmp_path, acquisition_defaults): + h = make_harness() + tracker = h.tracker + try: + # ---- Run 1: large enough to guarantee abort lands mid-flight ---------- + # 3x3 FOVs x 2 channels x NZ=2 = 36 images of runway. + h.new_experiment(tmp_path, "abort_run") + h.add_fov_grid("region0", nx=3, ny=3) + h.select_channels(2) + h.mpc.set_NZ(2) + expected_full = 3 * 3 * 2 * 2 + assert expected_full == 36 + + h.mpc.run_acquisition() + + assert tracker.started_event.wait(60), "run 1 did not start" + assert tracker.first_image_event.wait(120), "run 1 produced no image before abort" + + # Abort mid-flight (real API misspelling: request_abort_aquisition). + h.mpc.request_abort_aquisition() + + assert tracker.finished_event.wait(120), "abort did not fire finished callback" + # acquisition_in_progress() reflects the worker thread and may briefly lag + # the finished callback, so poll it down rather than asserting immediately. + assert _wait_until( + lambda: not h.mpc.acquisition_in_progress(), timeout_s=30 + ), "acquisition_in_progress stayed True after abort" + + aborted_count = tracker.image_count + assert aborted_count < expected_full, ( + f"abort produced {aborted_count} images; expected fewer than {expected_full} " + "(abort did not land mid-flight -- increase the grid)" + ) + assert aborted_count >= 1, "expected at least one image before abort" + + # Partial artifacts are allowed; only pin what is stable: the first + # timepoint directory was created for the partial run. + assert timepoint_dir(h.experiment_dir, 0).is_dir(), "run 1 timepoint dir 0/ was not created" + + # ---- Run 2: second experiment in the same process/harness ------------- + # The tracker events are latched from run 1; clear them and remember the + # run-1 image count so we can assert run 2 emits exactly 2 new callbacks. + tracker.started_event.clear() + tracker.finished_event.clear() + tracker.first_image_event.clear() + count_before_run2 = tracker.image_count + + h.mpc.scanCoordinates.clear_regions() + h.new_experiment(tmp_path, "after_abort") + h.add_fov_grid("region0", nx=2, ny=1) + h.select_channels(1) + h.mpc.set_NZ(1) + + h.run_and_wait(timeout_s=300) + + new_callbacks = tracker.image_count - count_before_run2 + assert new_callbacks == 2, ( + f"run 2 emitted {new_callbacks} new-image callbacks, expected 2 " "(2 FOVs x 1 channel x NZ=1)" + ) + + run2_tp0 = timepoint_dir(h.experiment_dir, 0) + run2_images = list_image_files(run2_tp0) + assert len(run2_images) == 2, f"run 2 timepoint 0/ has {run2_images}, expected 2 files" + assert (h.experiment_dir / ".done").exists(), "run 2 .done marker missing" + + finally: + h.close() + + # ---- Orphan assertion (after close) -------------------------------------- + # Formulation: type-based (isinstance(child, JobRunner)) rather than a raw + # active_children() count. JobRunner is a multiprocessing.Process subclass, so + # in the parent process a leaked runner shows up as a JobRunner instance in + # active_children(). This is more robust than a count baseline because: + # (a) the MultiPointController PRE-WARMS a JobRunner at construction and + # after each acquisition start, so a live runner is EXPECTED while the + # harness is open -- a count taken "before the test" would need careful + # timing to exclude it, whereas after close() the invariant is simply + # "zero JobRunners", and + # (b) type-matching ignores unrelated children the test framework or other + # fixtures might spawn, targeting exactly the leak PR #582 is about. + def _no_job_runners() -> bool: + return not [c for c in multiprocessing.active_children() if isinstance(c, JobRunner)] + + assert _wait_until(_no_job_runners, timeout_s=15), ( + "JobRunner subprocess(es) still alive after close(): " + f"{[c for c in multiprocessing.active_children() if isinstance(c, JobRunner)]}" + ) diff --git a/software/tests/acceptance/test_backpressure.py b/software/tests/acceptance/test_backpressure.py new file mode 100644 index 000000000..d1edca011 --- /dev/null +++ b/software/tests/acceptance/test_backpressure.py @@ -0,0 +1,99 @@ +"""Acceptance test pinning the backpressure byte-release deadlock class. + +This test guards against a historical PERMANENT DEADLOCK in acquisition +throttling (see project CLAUDE.md, "Backpressure" design note): + + Backpressure tracks bytes in the MAIN PROCESS queue. Bytes are incremented + when a job is dispatched and decremented when a job completes. For + DownsampledViewJob, bytes are released *immediately on job completion* + (not when the well/region completes), because the image data moves to + subprocess memory when processed. + + Why this matters: if bytes were only released at well completion, a z-stack + acquisition (FOVs x z-levels x channels) could accumulate more pending bytes + than the byte limit *before any well finishes*. The acquisition would then + block forever in wait_for_capacity() waiting for capacity that only well + completion could free -- a permanent deadlock. + +The scenario below drives a small z-stack under a deliberately tight byte limit +(~1.5 camera frames) so that the region's total pending bytes vastly exceed the +limit before the single region completes. Under correct per-job byte release the +acquisition throttles but drains and finishes; under the regressed +release-at-well-completion behavior it would hang. Completion within the timeout +is therefore the regression assertion. +""" + +import logging + +import pytest + +import control._def +from tests.acceptance.harness import ( + make_harness, + list_image_files, + timepoint_dir, +) + +pytestmark = pytest.mark.acceptance + +# Empirically measured simulated-camera frame: 4168 x 4168 uint16 = 34,744,448 +# bytes = 33.135 MiB (backpressure accounts in binary MiB, 1 MiB = 1024*1024). +_IMAGE_SIZE_MIB = 33.135 + + +def test_zstack_completes_under_tight_byte_limit(tmp_path, acquisition_defaults, caplog): + """A z-stack whose total bytes far exceed the byte limit still completes. + + Pins the DownsampledViewJob byte-release-on-job-completion design: if bytes + were only released at region completion, the 12-image z-stack (~8x the byte + limit) would deadlock before the single region finished. Completion is the + regression assertion; a hang manifests as the run_and_wait timeout. + """ + monkeypatch = acquisition_defaults + + # Byte limit ~1.5 frames: the second dispatched frame already exceeds it, so + # throttling must engage well before the region's 12 frames are all in flight. + byte_limit_mib = 1.5 * _IMAGE_SIZE_MIB # ~49.7 MiB + monkeypatch.setattr(control._def, "ACQUISITION_MAX_PENDING_MB", byte_limit_mib) + monkeypatch.setattr(control._def, "ACQUISITION_MAX_PENDING_JOBS", 3) + monkeypatch.setattr(control._def, "ACQUISITION_THROTTLING_ENABLED", True) + + harness = make_harness() + try: + harness.new_experiment(tmp_path / "backpressure", "zstack_tight_limit") + harness.add_fov_grid("region0", nx=2, ny=2) # 4 FOVs + harness.select_channels(1) # 1 channel + + harness.mpc.set_Nt(1) + harness.mpc.set_NZ(3) # 4 FOVs * 1 channel * 3 z = 12 images + harness.mpc.set_deltaZ(1.0) + harness.mpc.set_af_flag(False) + harness.mpc.set_reflection_af_flag(False) + + # 12 frames * 33.135 MiB = ~397.6 MiB pending if none released, + # ~8x the ~49.7 MiB byte limit -- the historical deadlock trigger. + with caplog.at_level(logging.INFO): + # Deadlock (regression) manifests as this timeout. + harness.run_and_wait(timeout_s=300) + + # Completion assertions: all 12 images landed and the region drained. + tp0 = timepoint_dir(harness.experiment_dir, 0) + images = list_image_files(tp0) + assert len(images) == 12, f"expected 12 image files in {tp0}, found {len(images)}: {images}" + assert harness.tracker.image_count == 12, f"tracker saw {harness.tracker.image_count} images, expected 12" + + # Root acquisition-complete marker. + done_marker = harness.experiment_dir / ".done" + assert done_marker.exists(), f"expected .done marker at {done_marker}" + + # Informational: confirm throttling actually engaged (not asserted -- + # timing-dependent on CI). Reported by the runner via captured logs. + throttled = any("Backpressure throttling" in r.getMessage() for r in caplog.records) + logging.getLogger(__name__).info( + "backpressure throttling engaged during acquisition: %s (byte limit=%.1f MiB, image=%.3f MiB)", + throttled, + byte_limit_mib, + _IMAGE_SIZE_MIB, + ) + finally: + harness.close() diff --git a/software/tests/acceptance/test_full_acquisition.py b/software/tests/acceptance/test_full_acquisition.py new file mode 100644 index 000000000..3e9ff22fd --- /dev/null +++ b/software/tests/acceptance/test_full_acquisition.py @@ -0,0 +1,214 @@ +""" +End-to-end simulation-mode acquisition acceptance tests. + +These drive a full MultiPointController acquisition against simulated hardware +(no GUI, no QApplication) and assert on observable artifacts only: files on +disk, their counts/shapes, coordinate CSVs, the acquisition-parameters JSON, +``.done`` markers, and the per-acquisition log. They intentionally pin CURRENT +master behavior; where the product writes something surprising the assertion +documents it with a ``# pins current behavior:`` note rather than "fixing" it. + +Ordering note: ``acquisition parameters.json`` is written by +``start_new_experiment()`` (i.e. inside ``harness.new_experiment``), so any NZ / +Nt / channel / region configuration that must appear in that JSON has to be set +*before* ``new_experiment`` is called. The helpers below configure the +controller first, then create the experiment folder, then run. +""" + +import glob +import json + +import pytest + +import tifffile + +from control._def import FileSavingOption +from tests.acceptance.conftest import set_file_saving_option +from tests.acceptance.harness import ( + list_image_files, + make_harness, + read_coordinates_csv, + timepoint_dir, +) + +pytestmark = pytest.mark.acceptance + + +def _open_zarr3(array_path): + """Open a Zarr v3 array written by the product's TensorStore writer. + + The store is true Zarr v3 (zarr.json + ``c/`` chunk keys); zarr-python + 2.15.0 cannot read that layout, so we use TensorStore -- the same library + the product uses to write it. + """ + import tensorstore as ts + + return ts.open({"driver": "zarr3", "kvstore": {"driver": "file", "path": str(array_path)}}).result() + + +def _plane_count_and_shape(full_shape): + """Split an (..., Y, X) shape into (num_planes, (Y, X)).""" + plane_shape = tuple(full_shape[-2:]) + num_planes = 1 + for dim in full_shape[:-2]: + num_planes *= dim + return num_planes, plane_shape + + +def test_full_small_acquisition(tmp_path, acquisition_defaults): + """Scenario 1: 2x2 FOVs x 2 channels x NZ=3, Nt=1, individual images.""" + h = make_harness() + try: + # Configure fully BEFORE new_experiment so the params JSON captures NZ/Nt. + h.add_fov_grid("region0", 2, 2) + h.select_channels(2) + h.mpc.set_NZ(3) + h.mpc.set_deltaZ(1.0) # micrometers + h.mpc.set_Nt(1) + h.mpc.set_deltat(0.0) + h.new_experiment(tmp_path / "exp", "small") + + h.run_and_wait(timeout_s=300) + + ed = h.experiment_dir + + # Image counts: 4 FOVs x 3 z x 2 channels. + assert h.mpc.get_acquisition_image_count() == 24 + assert h.tracker.image_count == 24 + + # Timepoint 0 holds exactly 24 individual image files. + tp0 = timepoint_dir(ed, 0) + assert len(list_image_files(tp0)) == 24 + + # Per-timepoint coordinates.csv: one row per region x fov x z_level = 4x3. + inner = read_coordinates_csv(tp0 / "coordinates.csv") + assert len(inner) == 12 + for col in ("region", "fov", "z_level"): + assert col in inner[0] + + # Top-level coordinates.csv: one row per FOV. + top = read_coordinates_csv(ed / "coordinates.csv") + assert len(top) == 4 + + # acquisition parameters.json (literal space in the filename). + with open(ed / "acquisition parameters.json") as f: + params = json.load(f) + assert params["Nz"] == 3 + assert params["Nt"] == 1 + assert params["dz(um)"] == 1.0 # set_deltaZ(1.0) is micrometers + + # .done markers at experiment root and in the timepoint dir. + assert (ed / ".done").exists() + assert (tp0 / ".done").exists() + + # Per-acquisition log exists and is clean of ERROR-level records. + log_text = (ed / "acquisition.log").read_text() + error_lines = [ln for ln in log_text.splitlines() if " - ERROR - " in ln] + assert error_lines == [], f"unexpected ERROR log lines: {error_lines[:5]}" + finally: + h.close() + + +def test_format_parity_ome_tiff_vs_zarr(tmp_path, acquisition_defaults): + """Scenario 2: same geometry (2x2 FOVs x 2 ch x NZ=2, Nt=1) in OME-TIFF and + Zarr v3 represents identical logical content (16 planes, same per-plane + shape and dtype).""" + # The product's ZARR_V3 writer requires tensorstore (and so does reading + # the store back) — skip where it isn't installed, like test_zarr_writer.py. + pytest.importorskip("tensorstore") + + def _run(option, subdir): + set_file_saving_option(acquisition_defaults, option) + h = make_harness() + try: + h.add_fov_grid("region0", 2, 2) + h.select_channels(2) + h.mpc.set_NZ(2) + h.mpc.set_deltaZ(1.0) + h.mpc.set_Nt(1) + h.mpc.set_deltat(0.0) + h.new_experiment(tmp_path / subdir, subdir) + h.run_and_wait(timeout_s=300) + return h.experiment_dir + finally: + h.close() + + # --- OME-TIFF --- + ome_dir = _run(FileSavingOption.OME_TIFF, "ome") + # pins current behavior: OME-TIFF writes one file per FOV under + # {experiment}/ome_tiff/, NOT under the timepoint dir. + ome_files = sorted(glob.glob(str(ome_dir / "**" / "*.ome.tiff"), recursive=True)) + assert len(ome_files) == 4, f"expected 4 per-FOV OME-TIFF files, found {ome_files}" + ome_total_planes = 0 + ome_plane_shapes = set() + ome_dtypes = set() + for path in ome_files: + with tifffile.TiffFile(path) as tf: + series = tf.series[0] + # pins current behavior: per-FOV series is ZCYX (Z=2, C=2). + planes, plane_shape = _plane_count_and_shape(series.shape) + ome_total_planes += planes + ome_plane_shapes.add(plane_shape) + ome_dtypes.add(series.dtype) + assert ome_total_planes == 16 # 4 FOVs x 2 z x 2 ch + assert len(ome_plane_shapes) == 1 + (ome_plane_shape,) = ome_plane_shapes + assert len(ome_dtypes) == 1 + (ome_dtype,) = ome_dtypes + + # --- Zarr v3 --- + zarr_dir = _run(FileSavingOption.ZARR_V3, "zarr") + # pins current behavior: non-well flexible region writes per-FOV stores at + # {experiment}/zarr/{region_id}/fov_{n}.ome.zarr/0 (experiment ROOT, not + # under the timepoint dir as ZarrWriterInfo's docstring path suggests). + zarr_arrays = sorted(glob.glob(str(zarr_dir / "**" / "fov_*.ome.zarr" / "0"), recursive=True)) + assert len(zarr_arrays) == 4, f"expected 4 per-FOV zarr stores, found {zarr_arrays}" + zarr_total_planes = 0 + zarr_plane_shapes = set() + zarr_dtypes = set() + for array_path in zarr_arrays: + arr = _open_zarr3(array_path) + # pins current behavior: per-FOV array is 5D TCZYX (T=1, C=2, Z=2). + planes, plane_shape = _plane_count_and_shape(tuple(arr.shape)) + zarr_total_planes += planes + zarr_plane_shapes.add(plane_shape) + zarr_dtypes.add(arr.dtype.numpy_dtype) + assert zarr_total_planes == 16 # 4 FOVs x 2 z x 2 ch + assert len(zarr_plane_shapes) == 1 + (zarr_plane_shape,) = zarr_plane_shapes + assert len(zarr_dtypes) == 1 + (zarr_dtype,) = zarr_dtypes + + # --- Parity between formats --- + assert ome_total_planes == zarr_total_planes == 16 + assert ome_plane_shape == zarr_plane_shape + assert ome_dtype == zarr_dtype + + +def test_multi_timepoint_smoke(tmp_path, acquisition_defaults): + """Scenario 5: Nt=2, 2x1 FOVs, 1 channel, NZ=1, individual images.""" + h = make_harness() + try: + h.add_fov_grid("region0", 2, 1) + h.select_channels(1) + h.mpc.set_NZ(1) + h.mpc.set_Nt(2) + h.mpc.set_deltat(0.0) + h.new_experiment(tmp_path / "exp", "multi_t") + + h.run_and_wait(timeout_s=300) + + ed = h.experiment_dir + assert h.tracker.image_count == 4 # 2 FOVs x 1 z x 1 ch x 2 timepoints + + for t in (0, 1): + tp = timepoint_dir(ed, t) + assert tp.is_dir(), f"timepoint dir {t} missing" + assert len(list_image_files(tp)) == 2 + rows = read_coordinates_csv(tp / "coordinates.csv") + assert len(rows) == 2 + assert (tp / ".done").exists() + + assert (ed / ".done").exists() + finally: + h.close() From 4b33050fd6d6cdc131847a54efb26dcbd6fabc61 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sun, 12 Jul 2026 17:57:07 -0700 Subject: [PATCH 2/3] test: harden acceptance suite per review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review findings (Opus, all nits): fix harness docstring referencing a nonexistent fixture; drop the redundant un-restored MERGE_CHANNELS module write (acquisition_defaults owns it with restore); guard the backpressure test's premise — assert on-disk acquisition bytes exceed 5x the byte limit so a shrunken simulated frame fails loudly instead of passing without exercising the deadlock path. Co-Authored-By: Claude Fable 5 --- software/tests/acceptance/harness.py | 7 ++++--- software/tests/acceptance/test_backpressure.py | 14 ++++++++++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/software/tests/acceptance/harness.py b/software/tests/acceptance/harness.py index 0d2d4eba3..c7bc80c93 100644 --- a/software/tests/acceptance/harness.py +++ b/software/tests/acceptance/harness.py @@ -117,10 +117,11 @@ def run_and_wait(self, timeout_s: float = FINISH_TIMEOUT_S) -> None: def make_harness() -> AcquisitionHarness: """Build a fresh simulated microscope wired to a MultiPointController. - Callers own cleanup: call harness.close() (use the `harness` fixture or - try/finally) so JobRunner subprocesses and semaphores are released. + Callers own cleanup: call harness.close() in a try/finally so JobRunner + subprocesses and semaphores are released. Combine with the + `acquisition_defaults` fixture, which pins MERGE_CHANNELS and the file + saving option with automatic restore. """ - control._def.MERGE_CHANNELS = False scope = control.microscope.Microscope.build_from_global_config(True) tracker = AcquisitionTracker() mpc = ts.get_test_multi_point_controller(microscope=scope, callbacks=tracker.get_callbacks()) diff --git a/software/tests/acceptance/test_backpressure.py b/software/tests/acceptance/test_backpressure.py index d1edca011..a7b57041b 100644 --- a/software/tests/acceptance/test_backpressure.py +++ b/software/tests/acceptance/test_backpressure.py @@ -80,6 +80,20 @@ def test_zstack_completes_under_tight_byte_limit(tmp_path, acquisition_defaults, tp0 = timepoint_dir(harness.experiment_dir, 0) images = list_image_files(tp0) assert len(images) == 12, f"expected 12 image files in {tp0}, found {len(images)}: {images}" + + # Guard the premise, not just the outcome: the deadlock path is only + # exercised while total acquisition bytes far exceed the byte limit. + # If the simulated frame ever shrinks, fail loudly instead of going + # green without testing anything. (Uncompressed TIFF size ~= frame + # bytes, so on-disk size is a faithful proxy.) + total_bytes = sum(p.stat().st_size for p in images) + limit_bytes = byte_limit_mib * 1024 * 1024 + assert total_bytes > 5 * limit_bytes, ( + f"acquisition totalled {total_bytes / 1024 / 1024:.1f} MiB, not >5x the " + f"{byte_limit_mib:.1f} MiB byte limit — the simulated frame size changed and " + "this test no longer exercises the backpressure deadlock path; re-derive " + "_IMAGE_SIZE_MIB and the limit" + ) assert harness.tracker.image_count == 12, f"tracker saw {harness.tracker.image_count} images, expected 12" # Root acquisition-complete marker. From fd156084a788db07e8c2a1e23f4507c84de12a5b Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sun, 12 Jul 2026 18:13:11 -0700 Subject: [PATCH 3/3] test: derive backpressure byte limit from a measured frame, not a constant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The premise guard added in the previous commit fired in CI: the pristine CI config crops the simulated camera to a quarter of the local frame (~8.3 MiB vs ~33.1 MiB), so the hardcoded frame size overstated the pressure and the guard correctly refused to pass. Snap one frame from the simulated camera at test start and size the byte limit (1.5 frames) from its actual nbytes — the 12-frame acquisition is then ~8x the limit under any machine config, and the on-disk guard still verifies the premise held. Co-Authored-By: Claude Fable 5 --- .../tests/acceptance/test_backpressure.py | 38 ++++++++++--------- 1 file changed, 21 insertions(+), 17 deletions(-) diff --git a/software/tests/acceptance/test_backpressure.py b/software/tests/acceptance/test_backpressure.py index a7b57041b..b8a609e2a 100644 --- a/software/tests/acceptance/test_backpressure.py +++ b/software/tests/acceptance/test_backpressure.py @@ -36,10 +36,6 @@ pytestmark = pytest.mark.acceptance -# Empirically measured simulated-camera frame: 4168 x 4168 uint16 = 34,744,448 -# bytes = 33.135 MiB (backpressure accounts in binary MiB, 1 MiB = 1024*1024). -_IMAGE_SIZE_MIB = 33.135 - def test_zstack_completes_under_tight_byte_limit(tmp_path, acquisition_defaults, caplog): """A z-stack whose total bytes far exceed the byte limit still completes. @@ -51,15 +47,23 @@ def test_zstack_completes_under_tight_byte_limit(tmp_path, acquisition_defaults, """ monkeypatch = acquisition_defaults - # Byte limit ~1.5 frames: the second dispatched frame already exceeds it, so - # throttling must engage well before the region's 12 frames are all in flight. - byte_limit_mib = 1.5 * _IMAGE_SIZE_MIB # ~49.7 MiB - monkeypatch.setattr(control._def, "ACQUISITION_MAX_PENDING_MB", byte_limit_mib) - monkeypatch.setattr(control._def, "ACQUISITION_MAX_PENDING_JOBS", 3) - monkeypatch.setattr(control._def, "ACQUISITION_THROTTLING_ENABLED", True) - harness = make_harness() try: + # Frame size is machine-config dependent (CI's pristine INI crops to a + # quarter of the local default), so measure it instead of hardcoding: + # snap one frame from the simulated camera and size the limit off it. + harness.scope.camera.send_trigger() + frame = harness.scope.camera.read_frame() + assert frame is not None, "simulated camera returned no frame" + image_size_mib = frame.nbytes / (1024 * 1024) + + # Byte limit ~1.5 frames: the second dispatched frame already exceeds + # it, so throttling must engage well before the region's 12 frames are + # all in flight (12 frames = ~8x the limit). + byte_limit_mib = 1.5 * image_size_mib + monkeypatch.setattr(control._def, "ACQUISITION_MAX_PENDING_MB", byte_limit_mib) + monkeypatch.setattr(control._def, "ACQUISITION_MAX_PENDING_JOBS", 3) + monkeypatch.setattr(control._def, "ACQUISITION_THROTTLING_ENABLED", True) harness.new_experiment(tmp_path / "backpressure", "zstack_tight_limit") harness.add_fov_grid("region0", nx=2, ny=2) # 4 FOVs harness.select_channels(1) # 1 channel @@ -70,8 +74,8 @@ def test_zstack_completes_under_tight_byte_limit(tmp_path, acquisition_defaults, harness.mpc.set_af_flag(False) harness.mpc.set_reflection_af_flag(False) - # 12 frames * 33.135 MiB = ~397.6 MiB pending if none released, - # ~8x the ~49.7 MiB byte limit -- the historical deadlock trigger. + # 12 frames = ~8x the ~1.5-frame byte limit if none were released -- + # the historical deadlock trigger. with caplog.at_level(logging.INFO): # Deadlock (regression) manifests as this timeout. harness.run_and_wait(timeout_s=300) @@ -90,9 +94,9 @@ def test_zstack_completes_under_tight_byte_limit(tmp_path, acquisition_defaults, limit_bytes = byte_limit_mib * 1024 * 1024 assert total_bytes > 5 * limit_bytes, ( f"acquisition totalled {total_bytes / 1024 / 1024:.1f} MiB, not >5x the " - f"{byte_limit_mib:.1f} MiB byte limit — the simulated frame size changed and " - "this test no longer exercises the backpressure deadlock path; re-derive " - "_IMAGE_SIZE_MIB and the limit" + f"{byte_limit_mib:.1f} MiB byte limit — saved images are much smaller than " + "the snapped frame this test sized the limit from, so the backpressure " + "deadlock path is no longer being exercised" ) assert harness.tracker.image_count == 12, f"tracker saw {harness.tracker.image_count} images, expected 12" @@ -107,7 +111,7 @@ def test_zstack_completes_under_tight_byte_limit(tmp_path, acquisition_defaults, "backpressure throttling engaged during acquisition: %s (byte limit=%.1f MiB, image=%.3f MiB)", throttled, byte_limit_mib, - _IMAGE_SIZE_MIB, + image_size_mib, ) finally: harness.close()