diff --git a/openhcs/core/steps/function_io.py b/openhcs/core/steps/function_io.py index bb788d5ee0..3a4da75d75 100644 --- a/openhcs/core/steps/function_io.py +++ b/openhcs/core/steps/function_io.py @@ -425,21 +425,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/integration/conftest.py b/tests/integration/conftest.py index a853fdb530..8a9ffb821e 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -2,10 +2,20 @@ from __future__ import annotations -from collections.abc import Callable, Sequence +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 + +from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG from tests.integration.helpers.fixture_utils import ( BACKEND_CONFIGS, @@ -112,3 +122,105 @@ 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, + ) + + +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) diff --git a/tests/integration/test_main.py b/tests/integration/test_main.py index 90cd0ef786..7fb5f35e63 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, @@ -116,6 +115,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 +140,7 @@ class TestConfig: backend_config: str execution_mode: str microscope_config: Dict + ports: TestPorts use_threading: bool = False def __post_init__(self): @@ -211,31 +228,28 @@ 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( + 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 +280,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 +293,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 +359,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 +716,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 +830,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 +849,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 +900,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." - ) 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, 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}