From 64ba745b5c1effa8116fb823f9b7d0f37ba0cff3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 08:13:53 -0400 Subject: [PATCH 1/7] Give each test_main run its own transport endpoints test_main hard-coded the execution server (7777) and viewer ports (5555/5556/5557/5559, plus Fiji's default 5565). On Linux these name IPC sockets under ~/.openhcs/ipc, shared by every checkout, so concurrent runs and agents connected to each other's servers and viewers: an ImageXpress run reported an OperaPhenix plate, and a viewer from another checkout rejected this checkout's metadata. A free_port_pair fixture in the integration conftest now allocates pairs through DataControlPortPairAuthority, starting from an OS-assigned port rather than the configured default every run would pick first, and shuts down whatever still listens on those pairs at teardown (viewers kept alive by a spawned execution server). test_main takes all its ports from it. Also delete the disabled _test_main_with_code_serialization reference function and create_test_pipeline's ignored sequential_config parameter. Co-Authored-By: Claude Opus 5.5 --- tests/integration/conftest.py | 58 +++++++- tests/integration/test_main.py | 248 ++++++--------------------------- 2 files changed, 98 insertions(+), 208 deletions(-) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index a853fdb530..304e248179 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -2,10 +2,16 @@ from __future__ import annotations -from collections.abc import Callable, Sequence +import socket +from collections.abc import Callable, Iterator, Sequence +from dataclasses import replace from typing import Any import pytest +from zmqruntime import EndpointShutdownMode, ZMQClient +from zmqruntime.transport import DataControlPortPairAuthority + +from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG from tests.integration.helpers.fixture_utils import ( BACKEND_CONFIGS, @@ -112,3 +118,53 @@ def enable_fiji( return bool( request.config.getoption("--enable-fiji") or visualizer_config["enable_fiji"] ) + + +def _os_assigned_port() -> int: + """Ask the OS for a currently unused port to start this run's search from.""" + while True: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + if port + OPENHCS_ZMQ_CONFIG.control_port_offset <= 65535: + return port + + +@pytest.fixture +def free_port_pair() -> Iterator[Callable[[], int]]: + """Allocate this test's own data/control endpoint pairs. + + Every server, viewer and client a test starts takes its port from here, so + concurrent runs never share an endpoint. Pairs come from the transport owner + that will bind them; the search starts at an OS-assigned port instead of the + configured default, which every concurrent run would otherwise pick first. + Whatever still listens on an allocated pair at teardown, such as a viewer + a spawned execution server kept alive, is shut down. + """ + allocated: set[int] = set() + data_ports: list[int] = [] + + def allocate() -> int: + pair = DataControlPortPairAuthority.acquire( + replace(OPENHCS_ZMQ_CONFIG, default_port=_os_assigned_port()), + transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, + excluded=allocated, + ) + allocated.update(pair.ports) + data_ports.append(pair.data_port) + return pair.data_port + + yield allocate + + for port in data_ports: + endpoint = OPENHCS_ZMQ_CONFIG.client_endpoint(port) + for mode in (EndpointShutdownMode.GRACEFUL, EndpointShutdownMode.FORCE): + if not endpoint.occupied_ports(OPENHCS_ZMQ_CONFIG): + break + ZMQClient.shutdown_endpoint_on_port( + port, + mode, + transport_mode=endpoint.transport_mode, + host=endpoint.host, + config=OPENHCS_ZMQ_CONFIG, + ) diff --git a/tests/integration/test_main.py b/tests/integration/test_main.py index b726853983..cc9b6105d4 100644 --- a/tests/integration/test_main.py +++ b/tests/integration/test_main.py @@ -116,6 +116,23 @@ def __post_init__(self): ) +@dataclass(frozen=True) +class TestPorts: + """This test's own endpoints; concurrent runs never share one.""" + + execution: int + napari: tuple[int, int, int] + fiji: tuple[int, int] + + @classmethod + def allocate(cls, free_port_pair) -> "TestPorts": + return cls( + execution=free_port_pair(), + napari=(free_port_pair(), free_port_pair(), free_port_pair()), + fiji=(free_port_pair(), free_port_pair()), + ) + + @dataclass class TestConfig: """Configuration for test execution.""" @@ -124,6 +141,7 @@ class TestConfig: backend_config: str execution_mode: str microscope_config: Dict + ports: TestPorts use_threading: bool = False def __post_init__(self): @@ -220,22 +238,16 @@ def test_function_dir(base_test_dir, microscope_config, request): def create_test_pipeline( + ports: TestPorts, enable_napari: bool = False, enable_fiji: bool = False, - sequential_config: dict = None, ) -> list[Step]: """Create test pipeline with materialization configuration. Args: + ports: This test's viewer endpoints enable_napari: Enable Napari streaming enable_fiji: Enable Fiji streaming - sequential_config: Sequential processing configuration dict with keys: - - sequential_components: List of component names to process sequentially - - should_fail: Whether this config should fail validation - - expected_error: Expected error message substring (if should_fail=True) - - Note: sequential_config is NOT used in this function - it should be set in PipelineConfig instead. - This parameter is kept for backward compatibility but is ignored. """ cpu_only_mode = os.getenv("OPENHCS_CPU_ONLY", "false").lower() == "true" if cpu_only_mode: @@ -266,9 +278,11 @@ def create_test_pipeline( ), step_materialization_config=LazyStepMaterializationConfig(), napari_streaming_config=LazyNapariStreamingConfig( - port=5555, enabled=enable_napari + port=ports.napari[0], enabled=enable_napari + ), + fiji_streaming_config=LazyFijiStreamingConfig( + port=ports.fiji[0], enabled=enable_fiji ), - fiji_streaming_config=LazyFijiStreamingConfig(enabled=enable_fiji), ), Step( func=create_composite, @@ -277,10 +291,10 @@ def create_test_pipeline( group_by=Ungrouped, ), napari_streaming_config=LazyNapariStreamingConfig( - port=5557, enabled=enable_napari + port=ports.napari[1], enabled=enable_napari ), fiji_streaming_config=LazyFijiStreamingConfig( - port=5556, enabled=enable_fiji + port=ports.fiji[1], enabled=enable_fiji ), ), Step( @@ -343,11 +357,13 @@ def create_test_pipeline( } ), napari_streaming_config=LazyNapariStreamingConfig( - port=5559, + port=ports.napari[2], variable_size_handling=NapariVariableSizeHandling.PAD_TO_MAX, enabled=enable_napari, ), - fiji_streaming_config=LazyFijiStreamingConfig(enabled=enable_fiji), + fiji_streaming_config=LazyFijiStreamingConfig( + port=ports.fiji[0], enabled=enable_fiji + ), ), ] @@ -698,7 +714,7 @@ def _execute_pipeline_zmq( # Create ZMQ client and connect to server # The server will create its own orchestrator and get the well list - client = ZMQExecutionClient(port=7777, persistent=False) + client = ZMQExecutionClient(port=test_config.ports.execution, persistent=False) try: # Connect to server (spawns if needed) @@ -812,17 +828,18 @@ def test_main( enable_napari: bool, enable_fiji: bool, sequential_config: Dict, + free_port_pair, ): """Unified test for all combinations of microscope types, backends, data types, execution modes, and sequential processing.""" + ports = TestPorts.allocate(free_port_pair) # Handle both Path and int (OMERO plate_id) - if isinstance(plate_dir, int): - test_config = TestConfig( - plate_dir, backend_config, execution_mode, microscope_config - ) - else: - test_config = TestConfig( - Path(plate_dir), backend_config, execution_mode, microscope_config - ) + test_config = TestConfig( + plate_dir if isinstance(plate_dir, int) else Path(plate_dir), + backend_config, + execution_mode, + microscope_config, + ports, + ) print( f"{CONSTANTS.START_INDICATOR} with plate: {plate_dir}, backend: {backend_config}, microscope: {microscope_config['format']}, mode: {execution_mode}, zmq: {zmq_execution_mode}, sequential: {sequential_config['name']}" @@ -830,9 +847,9 @@ def test_main( # Create pipeline with sequential configuration pipeline = create_test_pipeline( + ports, enable_napari=enable_napari, enable_fiji=enable_fiji, - sequential_config=sequential_config, ) # If this configuration should fail validation, expect ValueError during compilation @@ -881,186 +898,3 @@ def test_main( print(f"{'=' * 80}\n") print(f"{CONSTANTS.SUCCESS_INDICATOR} ({len(results)} wells processed)") - - -def _test_main_with_code_serialization( - plate_dir: Union[Path, str, int], - backend_config: str, - data_type_config: Dict, - execution_mode: str, - zmq_execution_mode: str, - microscope_config: Dict, -): - """ - DISABLED: Code serialization test (not run as pytest test). - - This function tests the code serializer for code-based object serialization, - but is disabled because: - 1. It's redundant with test_main (which tests the actual integration) - 2. Code serialization is already tested in the PyQt UI - 3. It breaks with OMERO (plate_dir is int, not Path) - - The function is kept for reference but prefixed with _ to exclude from pytest. - - Original purpose: - - Test using the serializer for code-based object serialization - - Mirror the UI's approach: create objects ā†’ convert to code ā†’ exec ā†’ use - - Prove code-based serialization works for remote execution - """ - # Handle both Path and int (OMERO plate_id) - if isinstance(plate_dir, int): - test_config = TestConfig( - plate_dir, backend_config, execution_mode, microscope_config - ) - else: - test_config = TestConfig( - Path(plate_dir), backend_config, execution_mode, microscope_config - ) - - print( - f"{CONSTANTS.START_INDICATOR} [CODE SERIALIZATION TEST] with plate: {plate_dir}, backend: {backend_config}, mode: {execution_mode}, zmq: {zmq_execution_mode}" - ) - - # Step 1: Create objects normally - from polystore.base import reset_memory_backend - - reset_memory_backend() - - global_config = _create_pipeline_config(test_config) - - # Create PipelineConfig with lazy configs for proper hierarchical inheritance - pipeline_config = PipelineConfig( - path_planning_config=LazyPathPlanningConfig( - output_dir_suffix=CONSTANTS.OUTPUT_SUFFIX - ), - step_well_filter_config=LazyStepWellFilterConfig( - well_filter=CONSTANTS.PIPELINE_STEP_WELL_FILTER_TEST - ), - ) - - pipeline = create_test_pipeline() - - print("šŸ“¦ Step 1: Created original objects") - print(f" - GlobalPipelineConfig: {type(global_config).__name__}") - print(f" - PipelineConfig: {type(pipeline_config).__name__}") - print(f" - Pipeline: {len(pipeline)} steps") - - # Step 2: Convert to Python code using the serializer - import openhcs.serialization.pycodify_formatters # noqa: F401 - from pycodify import Assignment, generate_python_source - - print("\nšŸ”„ Step 2: Converting objects to Python code...") - - # Generate code for GlobalPipelineConfig - global_config_code = generate_python_source( - Assignment("config", global_config), - header="# Configuration Code", - clean_mode=True, - ) - - # Generate code for PipelineConfig - pipeline_config_code = generate_python_source( - Assignment("config", pipeline_config), - header="# Configuration Code", - clean_mode=True, - ) - - # Generate code for Pipeline steps through the canonical transport authority. - pipeline_steps_code = FunctionStepTransportAuthority.source_from_pipeline(pipeline) - - print(f" - GlobalPipelineConfig code: {len(global_config_code)} chars") - print(f" - PipelineConfig code: {len(pipeline_config_code)} chars") - print(f" - Pipeline steps code: {len(pipeline_steps_code)} chars") - - # Save the generated code to files for inspection - code_output_dir = test_config.plate_dir / "generated_code" - code_output_dir.mkdir(exist_ok=True) - - (code_output_dir / "global_config.py").write_text(global_config_code) - (code_output_dir / "pipeline_config.py").write_text(pipeline_config_code) - (code_output_dir / "pipeline_steps.py").write_text(pipeline_steps_code) - - print(f" - Saved code to: {code_output_dir}") - - # Step 3: Exec the code to recreate objects - print("\nāš™ļø Step 3: Recreating objects from Python code using exec()...") - - # Recreate GlobalPipelineConfig - global_config_namespace = {} - exec(global_config_code, global_config_namespace) - recreated_global_config = global_config_namespace["config"] - - # Recreate PipelineConfig - pipeline_config_namespace = {} - exec(pipeline_config_code, pipeline_config_namespace) - recreated_pipeline_config = pipeline_config_namespace["config"] - - # Recreate Pipeline steps - pipeline_steps_namespace = {} - exec(pipeline_steps_code, pipeline_steps_namespace) - recreated_pipeline_steps = ( - FunctionStepTransportAuthority.pipeline_steps_from_namespace( - pipeline_steps_namespace - ) - ) - - print( - f" - Recreated GlobalPipelineConfig: {type(recreated_global_config).__name__}" - ) - print(f" - Recreated PipelineConfig: {type(recreated_pipeline_config).__name__}") - print(f" - Recreated Pipeline steps: {len(recreated_pipeline_steps)} steps") - - # Verify the recreated objects match the originals - print("\nšŸ” Step 4: Verifying recreated objects...") - - # Check GlobalPipelineConfig fields - assert recreated_global_config.num_workers == global_config.num_workers - assert recreated_global_config.use_threading == global_config.use_threading - print(f" āœ… GlobalPipelineConfig fields match") - - # Check PipelineConfig fields - assert type(recreated_pipeline_config) == type(pipeline_config) - print(f" āœ… PipelineConfig type matches") - - # Check Pipeline steps - assert len(recreated_pipeline_steps) == len(pipeline) - for i, (orig_step, recreated_step) in enumerate( - zip(pipeline, recreated_pipeline_steps) - ): - assert type(orig_step) == type(recreated_step) - assert orig_step.name == recreated_step.name - print(f" āœ… Pipeline steps match ({len(recreated_pipeline_steps)} steps)") - - # Step 5: Use recreated objects for execution - print("\nšŸš€ Step 5: Executing pipeline with recreated objects...") - - # Copy reconstructed steps into the mutable execution list. - recreated_pipeline = list(recreated_pipeline_steps) - - # Execute using the specified mode (direct or zmq) - if zmq_execution_mode == "zmq": - # For ZMQ mode, use the recreated configs directly - results = _execute_pipeline_zmq( - test_config, - recreated_pipeline, - recreated_global_config, - recreated_pipeline_config, - ) - else: - # For direct mode, set up global context and use orchestrator - ensure_global_config_context(GlobalPipelineConfig, recreated_global_config) - orchestrator = PipelineOrchestrator( - test_config.plate_dir, pipeline_config=recreated_pipeline_config - ) - orchestrator.initialize() - results = _execute_pipeline_phases(orchestrator, recreated_pipeline) - - validate_separate_materialization(test_config.plate_dir) - - print( - f"\n{CONSTANTS.SUCCESS_INDICATOR} [CODE SERIALIZATION TEST] ({len(results)} wells processed)" - ) - print("āœ… Code-based serialization works perfectly!") - print( - " This proves we can use Python code instead of pickling for remote execution." - ) From 44d3bcd277a8d36186a62caee78625d6cf3267ab Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 08:13:53 -0400 Subject: [PATCH 2/7] Key zarr conversion outputs by their plate-relative virtual path update_metadata_for_zarr_conversion looked up each converted store file by its path relative to the zarr store ("A01_....tif"), but source projections are keyed by plate-relative virtual paths ("TimePoint_1/A01_....tif"). Conversion mirrors the source subdirectory's layout inside the store, so the lookup now prefixes the original subdirectory. Any prepared workspace whose images live in a subdirectory failed step 0 with "Zarr conversion output has no declared source projection" since ccfef5f6d; the existing unit test used the plate root (".") as the subdirectory, where both keys coincide. Co-Authored-By: Claude Opus 5.5 --- openhcs/core/steps/function_io.py | 12 ++- ...test_zarr_input_conversion_subdirectory.py | 101 ++++++++++++++++++ 2 files changed, 108 insertions(+), 5 deletions(-) create mode 100644 tests/unit/test_zarr_input_conversion_subdirectory.py diff --git a/openhcs/core/steps/function_io.py b/openhcs/core/steps/function_io.py index f0bcaddd61..19dc524a6c 100644 --- a/openhcs/core/steps/function_io.py +++ b/openhcs/core/steps/function_io.py @@ -427,21 +427,23 @@ def update_metadata_for_zarr_conversion( str(zarr_dir), Backend.ZARR.value ): try: - source_virtual_path = Path(output_path).relative_to(zarr_dir) + store_path = PurePosixPath( + Path(output_path).relative_to(zarr_dir).as_posix() + ) except ValueError as exc: raise ValueError( "Zarr conversion output lies outside its declared store: " f"{output_path!r}." ) from exc - source_virtual_text = source_virtual_path.as_posix() + # Conversion mirrors the source subdirectory's layout inside the + # store, while virtual paths are plate-relative. + source_virtual_text = str(PurePosixPath(original_subdir) / store_path) if source_virtual_text not in source_projections.entries: raise ValueError( "Zarr conversion output has no declared source projection: " f"{source_virtual_text!r}." ) - materialized_path = str( - PurePosixPath(zarr_subdir) / source_virtual_path - ) + materialized_path = str(PurePosixPath(zarr_subdir) / store_path) materialized_projections.append( replace( source_projections.entries[source_virtual_text], diff --git a/tests/unit/test_zarr_input_conversion_subdirectory.py b/tests/unit/test_zarr_input_conversion_subdirectory.py new file mode 100644 index 0000000000..f1607a57db --- /dev/null +++ b/tests/unit/test_zarr_input_conversion_subdirectory.py @@ -0,0 +1,101 @@ +"""Zarr input conversion of a prepared workspace whose images live in a subdirectory.""" + +from __future__ import annotations + +import json +import queue +from pathlib import Path + +from objectstate.lazy_factory import ensure_global_config_context + +from openhcs.constants.constants import Backend +from openhcs.core.config import ( + GlobalPipelineConfig, + LazyVFSConfig, + MaterializationBackend, + PipelineConfig, + VFSConfig, +) +from openhcs.core.dataset_sources.choice import DatasetSourceChoice +from openhcs.core.orchestrator.orchestrator import PipelineOrchestrator +from openhcs.core.progress import set_progress_queue +from openhcs.core.steps import FunctionStep +from openhcs.core.axes import AxisFamily +from openhcs.demo.synthetic_data import SyntheticMicroscopyGenerator +from openhcs.processing.backends.processors.numpy_processor import ( + stack_percentile_normalize, +) + + +def _prepared_workspace_plate(root: Path) -> Path: + generator = SyntheticMicroscopyGenerator( + output_dir=str(root), + grid_size=(1, 1), + tile_size=(16, 16), + overlap_percent=0, + wavelengths=1, + z_stack_levels=2, + num_cells=0, + wells=["A01"], + format="ImageXpress", + include_all_components=True, + random_seed=1, + ) + generator.generate_dataset() + generator.generate_openhcs_metadata(sub_dir="TimePoint_1") + return root + + +def test_zarr_conversion_keys_store_outputs_by_plate_relative_virtual_path(tmp_path): + """Store paths are relative to the source subdirectory; virtual paths are not.""" + plate = _prepared_workspace_plate(tmp_path / "plate") + ensure_global_config_context( + GlobalPipelineConfig, + GlobalPipelineConfig( + num_workers=1, + use_threading=True, + dataset_source=DatasetSourceChoice.named("openhcsdata"), + vfs_config=VFSConfig(materialization_backend=MaterializationBackend.ZARR), + ), + ) + orchestrator = PipelineOrchestrator( + plate, + pipeline_config=PipelineConfig( + vfs_config=LazyVFSConfig(materialization_backend=MaterializationBackend.ZARR) + ), + ).initialize() + wells = orchestrator.get_component_keys(AxisFamily.active().partition_axis()) + progress_queue = queue.Queue() + set_progress_queue(progress_queue) + try: + bundle = orchestrator.compile_pipelines( + pipeline_definition=[FunctionStep(func=stack_percentile_normalize)], + well_filter=wells, + ) + results = orchestrator.execute_compiled_plate( + execution_bundle=bundle, + progress_queue=progress_queue, + progress_context={ + "execution_id": "zarr-conversion-subdirectory", + "plate_id": str(plate), + "axis_id": "", + }, + ) + finally: + set_progress_queue(None) + assert all(result.is_success() for result in results.values()), results + + subdirectories = json.loads((plate / "openhcs_metadata.json").read_text())[ + "subdirectories" + ] + assert subdirectories["TimePoint_1"]["main"] is False + converted = subdirectories["zarr"] + assert converted["main"] is True + source_files = { + Path(path).name for path in subdirectories["TimePoint_1"]["image_files"] + } + assert {Path(path).name for path in converted["image_files"]} == source_files + assert all(path.startswith("zarr/") for path in converted["image_files"]) + assert { + record["ref"]["backend"] for record in converted["source_projection"] + } == {Backend.ZARR.value} From f8a55013bd5f6217926370936261b21bf574e1c3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 08:13:53 -0400 Subject: [PATCH 3/7] Publish complete metadata for zarr-mode result-only directories Bump PolyStore so the zarr backend lists and reports the CSV, JSON and ROI archive files it saves through disk passthrough. Since 1f1525046 the step-time publication admits a result directory when exists() finds its saved files, which zarr answers through passthrough, while completion writes the entry's header fields only when is_dir()/list_files() find outputs, which zarr did not. Zarr runs therefore left a "results" entry with projection fields only and no image_files, available_backends or microscope_handler_name. Co-Authored-By: Claude Opus 5.5 --- external/PolyStore | 2 +- .../test_persisted_roi_only_publication.py | 44 +++++++++++++++++-- 2 files changed, 41 insertions(+), 5 deletions(-) diff --git a/external/PolyStore b/external/PolyStore index 3938236ecd..ac230c46a1 160000 --- a/external/PolyStore +++ b/external/PolyStore @@ -1 +1 @@ -Subproject commit 3938236ecd1ed09e73c31c89304247b1409eff9b +Subproject commit ac230c46a12d3d8dbd586669f788b84562c87373 diff --git a/tests/unit/test_persisted_roi_only_publication.py b/tests/unit/test_persisted_roi_only_publication.py index cdb09d7bf3..f402f4784e 100644 --- a/tests/unit/test_persisted_roi_only_publication.py +++ b/tests/unit/test_persisted_roi_only_publication.py @@ -86,7 +86,7 @@ def write(self, context, *, produced_plan=None): context.publication_audit.append(("after", self.sub_dir)) -def plan_for(plate): +def plan_for(plate, backend=Backend.DISK.value): return CompiledStepPlan( step_index=0, step_name="synthetic saved result", @@ -97,7 +97,7 @@ def plan_for(plate): analysis_results_dir=str(plate / "compiled_default"), runtime_artifact_materialization=RuntimeArtifactMaterializationPlan( persistent_enabled=True, - persistent_backend=Backend.DISK.value, + persistent_backend=backend, ), ) @@ -105,6 +105,7 @@ def plan_for(plate): def save_archive_outcome( context, plate, destination, monkeypatch, *, filename="independent-outline.roi.zip", content=b"synthetic admission only", + backend=Backend.DISK.value, ): """Use the original declared bundle writer, batch and disk save boundary.""" spec = MaterializationSpec(FileBundleOptions()) @@ -115,10 +116,10 @@ def save_archive_outcome( payload, str(base_path), context.filemanager, - (Backend.DISK.value,), + (backend,), ) saved = batch.save() - (output,) = saved.outputs_for_backend(Backend.DISK.value) + (output,) = saved.outputs_for_backend(backend) assert output is batch.outputs[0] # Source qualification must consume the original successful save, not render # a mutable logical payload again or use planned names as proof of a save. @@ -244,6 +245,41 @@ def test_saved_roi_only_batch_publishes_and_reconciles_without_rendering( assert result_paths(plate, publication_context) == (output.path,) +def test_result_only_destination_saved_through_zarr_passthrough_publishes_complete_entry( + tmp_path, + publication_context, + monkeypatch, +): + """Zarr saves archives through disk passthrough; publication must still see them.""" + from polystore.zarr import ZarrStorageBackend + + publication_context.filemanager = FileManager( + { + Backend.DISK.value: DiskStorageBackend(), + Backend.ZARR.value: ZarrStorageBackend(), + } + ) + monkeypatch.setattr( + "polystore.backend_registry.get_backend_instance", + lambda key: publication_context.filemanager._get_backend(key), + ) + plate = tmp_path / "plate" + outcome, output = save_archive_outcome( + publication_context, plate, "results", monkeypatch, backend=Backend.ZARR.value + ) + plan = plan_for(plate, Backend.ZARR.value) + publication_context.step_plans = {0: plan} + _publish_saved_step(publication_context, plan, artifact_materializations=(outcome,)) + OpenHCSMetadataTarget.finalize_completed_plate({"A01": publication_context}) + + published = json.loads(METADATA_CONFIG.metadata_path(plate).read_text()) + entry = published["subdirectories"]["results"] + assert entry["image_files"] == [] + assert entry["microscope_handler_name"] == "openhcsdata" + assert entry["available_backends"] == {Backend.ZARR.value: True} + assert Path(output.path).is_file() + + def test_shared_writer_accepts_result_only_but_not_unaddressed_images( tmp_path, publication_context, From 950c0b906177e9b68e1fd5e5e3021ab4d792ea4d Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 08:13:54 -0400 Subject: [PATCH 4/7] Compare native Napari slices through the layer slicing state 7bc78166a called Layer._make_slice_input, which the installed napari (0.9.2, slicing moved to Layer._slicing_state) does not have. Every scheduled layer update raised AttributeError inside the viewer, so it never acked and producers timed out after 30 s. The failed steps then left saved images without produced records, which surfaced as "Saved images lack typed produced addresses". Use _slicing_state.make_slice_input. Co-Authored-By: Claude Opus 5.5 --- openhcs/runtime/napari_viewer_server.py | 3 +- .../unit/test_napari_native_frame_applied.py | 33 +++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_napari_native_frame_applied.py diff --git a/openhcs/runtime/napari_viewer_server.py b/openhcs/runtime/napari_viewer_server.py index e75033bcc9..33f733f3f8 100644 --- a/openhcs/runtime/napari_viewer_server.py +++ b/openhcs/runtime/napari_viewer_server.py @@ -2472,7 +2472,8 @@ def native_frame_applied(self, *, include_hidden: bool = False) -> bool: viewer = self.server.require_viewer() return self._native_frame_mutation_depth == 0 and all( layer.loaded and ( - not layer.visible or layer._slice_input == layer._make_slice_input(viewer.dims) + not layer.visible or layer._slice_input + == layer._slicing_state.make_slice_input(viewer.dims) ) for layer in viewer.layers if layer.visible or include_hidden ) diff --git a/tests/unit/test_napari_native_frame_applied.py b/tests/unit/test_napari_native_frame_applied.py new file mode 100644 index 0000000000..0752954937 --- /dev/null +++ b/tests/unit/test_napari_native_frame_applied.py @@ -0,0 +1,33 @@ +"""Native frame completion against the installed Napari layer slicing API.""" + +from __future__ import annotations + +from types import SimpleNamespace + +import numpy as np +import pytest + +napari_components = pytest.importorskip("napari.components") + +from openhcs.runtime.napari_viewer_server import NapariLayerDisplayPipeline # noqa: E402 + + +def _pipeline_for(viewer) -> SimpleNamespace: + return SimpleNamespace( + server=SimpleNamespace(require_viewer=lambda: viewer), + _native_frame_mutation_depth=0, + ) + + +def test_native_frame_applied_compares_slices_from_real_napari_layers(): + viewer = napari_components.ViewerModel() + layer = viewer.add_image(np.zeros((3, 8, 8), dtype=np.uint16)) + pipeline = _pipeline_for(viewer) + + assert NapariLayerDisplayPipeline.native_frame_applied(pipeline) + + applied_slice = layer._slice_input + viewer.dims.set_current_step(0, 2) + assert layer._slice_input != applied_slice + layer._slicing_state._slice_input = applied_slice + assert not NapariLayerDisplayPipeline.native_frame_applied(pipeline) From ca52e4bf6ac464af3f1ee48e89ae9db2d058f072 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 08:40:54 -0400 Subject: [PATCH 5/7] Delete zarr-saved ROI archives when re-running over existing outputs Bump PolyStore: the zarr backend's delete() used its own .json/.csv/.txt list instead of the declared passthrough extensions, so replacing a .roi.zip saved through disk passthrough failed in Cell Counting. Co-Authored-By: Claude Opus 5.5 --- external/PolyStore | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/external/PolyStore b/external/PolyStore index ac230c46a1..82bf0b064c 160000 --- a/external/PolyStore +++ b/external/PolyStore @@ -1 +1 @@ -Subproject commit ac230c46a12d3d8dbd586669f788b84562c87373 +Subproject commit 82bf0b064c12625b4c60a9b88d851c382782585c From 76155c8709a6198d4b0c9f4247304a3d60b567cd Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 08:40:54 -0400 Subject: [PATCH 6/7] Give every test_main parametrization a fresh plate directory All parametrizations of one microscope regenerated their plate into the same tests/integration/tests_data directory. A zarr run converts its input plate in place (a second subdirectory marked main) and leaves saved outputs, so the next parametrization started from the previous one's state and failed with "Multiple OpenHCS metadata subdirectories are marked main". Plates now live under the test's own tmp_path. Co-Authored-By: Claude Opus 5.5 --- tests/integration/test_main.py | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/tests/integration/test_main.py b/tests/integration/test_main.py index cc9b6105d4..60f74de60a 100644 --- a/tests/integration/test_main.py +++ b/tests/integration/test_main.py @@ -64,7 +64,6 @@ # Test utilities and fixtures from tests.integration.helpers.fixture_utils import ( backend_config, - base_test_dir, data_type_config, execution_mode, microscope_config, @@ -229,12 +228,15 @@ def _fiji_enabled() -> bool: @pytest.fixture -def test_function_dir(base_test_dir, microscope_config, request): - """Create test directory for a specific test function.""" - test_name = request.node.originalname or request.node.name.split("[")[0] - test_dir = base_test_dir / f"{test_name}[{microscope_config['format']}]" - test_dir.mkdir(parents=True, exist_ok=True) - yield test_dir +def test_function_dir(tmp_path, microscope_config): + """Give every parametrization a fresh plate directory. + + A shared directory let one run's converted input (a second ``main`` + subdirectory) and saved outputs leak into the next parametrization. + """ + test_dir = tmp_path / microscope_config["format"] + test_dir.mkdir() + return test_dir def create_test_pipeline( From a922eb9b2172a0b144685173ed36d388edf5af54 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 10 Oct 2026 09:13:28 -0400 Subject: [PATCH 7/7] Kill every process an integration test spawned, detached ones included Execution and viewer servers start their own sessions (setsid) and outlive the test process tree; ViewerStateManager only stops viewers the test process itself launched. With per-test ports each zmq or Fiji run left new viewer servers (and Fiji windows) behind. An autouse fixture now marks each test's environment, which every descendant inherits, and at teardown kills the process group of every marked process still alive, pass or fail. Co-Authored-By: Claude Opus 5.5 --- tests/integration/conftest.py | 56 +++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 304e248179..8a9ffb821e 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -2,11 +2,15 @@ from __future__ import annotations +import os +import signal import socket +import uuid from collections.abc import Callable, Iterator, Sequence from dataclasses import replace from typing import Any +import psutil import pytest from zmqruntime import EndpointShutdownMode, ZMQClient from zmqruntime.transport import DataControlPortPairAuthority @@ -168,3 +172,55 @@ def allocate() -> int: host=endpoint.host, config=OPENHCS_ZMQ_CONFIG, ) + + +SPAWN_MARKER_VARIABLE = "OPENHCS_TEST_SPAWN_MARKER" + + +def _marked_processes(marker: str) -> list[psutil.Process]: + found = [] + for process in psutil.process_iter(["pid"]): + if process.pid == os.getpid(): + continue + try: + if process.environ().get(SPAWN_MARKER_VARIABLE) == marker: + found.append(process) + except (psutil.AccessDenied, psutil.NoSuchProcess, psutil.ZombieProcess): + continue + return found + + +@pytest.fixture(autouse=True) +def reap_spawned_processes(monkeypatch) -> Iterator[None]: + """Kill every process this test spawned, detached ones included. + + Execution and viewer servers start their own sessions (setsid), so they + outlive the test and its process tree. Each test marks its environment; + every descendant inherits the marker, and teardown kills the process group + of each marked process still alive, whether the test passed or failed. + """ + marker = uuid.uuid4().hex + monkeypatch.setenv(SPAWN_MARKER_VARIABLE, marker) + yield + own_group = os.getpgid(0) + survivors = _marked_processes(marker) + for process in survivors: + try: + group = os.getpgid(process.pid) + if group != own_group: + os.killpg(group, signal.SIGTERM) + else: + process.terminate() + except (ProcessLookupError, psutil.NoSuchProcess): + continue + _gone, alive = psutil.wait_procs(survivors, timeout=10) + for process in alive: + try: + group = os.getpgid(process.pid) + if group != own_group: + os.killpg(group, signal.SIGKILL) + else: + process.kill() + except (ProcessLookupError, psutil.NoSuchProcess): + continue + psutil.wait_procs(alive, timeout=10)