test: simulation-mode acquisition acceptance suite

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

software/tests/acceptance/harness.py

@@ -0,0 +1,147 @@
+"""
+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() 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.
+    """
+    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}.<ext> 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()
+    )

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)]}"
+    )