From fe031150769d77272e8340df77c868b869f16fcf Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Sat, 7 Mar 2026 21:14:40 +0000 Subject: [PATCH 01/15] Add workflow context manager to wrap parsl.load --- .github/workflows/test-suite.yaml | 4 +- docs/configuration.rst | 56 +++++-- docs/container.rst | 4 +- docs/data.rst | 16 +- docs/quickstart.rst | 86 ++++++---- docs/testing.rst | 4 +- src/chiltepin/__init__.py | 14 ++ src/chiltepin/workflow.py | 196 ++++++++++++++++++++++ tests/test_data.py | 32 ++-- tests/test_globus_compute_hello.py | 29 +--- tests/test_globus_compute_mpi.py | 29 +--- tests/test_parsl_hello.py | 28 +--- tests/test_parsl_mpi.py | 28 +--- tests/test_tasks.py | 32 +--- tests/test_workflow.py | 256 +++++++++++++++++++++++++++++ 15 files changed, 623 insertions(+), 191 deletions(-) create mode 100644 src/chiltepin/workflow.py create mode 100644 tests/test_workflow.py diff --git a/.github/workflows/test-suite.yaml b/.github/workflows/test-suite.yaml index c7cb134b..423ddfbf 100644 --- a/.github/workflows/test-suite.yaml +++ b/.github/workflows/test-suite.yaml @@ -94,7 +94,7 @@ jobs: GLOBUS_COMPUTE_CLIENT_ID: ${{ secrets.GC_CLIENT_ID }} GLOBUS_COMPUTE_CLIENT_SECRET: ${{ secrets.GC_CLIENT_SECRET }} run: | - docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ matrix.python-version }}/bin/activate; pytest -s -v --assert=plain --cov=src/chiltepin --cov-report=term-missing --cov-report=markdown:coverage.md --cov-report=xml --cov-report=html --cov-fail-under=0 --config=tests/configs/docker.yaml" + docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ matrix.python-version }}/bin/activate; pytest -s -v --cov=src/chiltepin --cov-report=term-missing --cov-report=markdown:coverage.md --cov-report=xml --cov-report=html --cov-fail-under=0 --config=tests/configs/docker.yaml" - name: Copy coverage reports from container if: always() @@ -233,7 +233,7 @@ jobs: GLOBUS_COMPUTE_CLIENT_ID: ${{ secrets.GC_CLIENT_ID }} GLOBUS_COMPUTE_CLIENT_SECRET: ${{ secrets.GC_CLIENT_SECRET }} run: | - docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ env.LATEST_PYTHON_VERSION }}/bin/activate; pytest -s -v --assert=plain --config=tests/configs/docker.yaml" + docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ env.LATEST_PYTHON_VERSION }}/bin/activate; pytest -s -v --config=tests/configs/docker.yaml" - name: Debug session if: ${{ failure() }} diff --git a/docs/configuration.rst b/docs/configuration.rst index 302dcd79..9b4ef76e 100644 --- a/docs/configuration.rst +++ b/docs/configuration.rst @@ -415,26 +415,58 @@ Loading Configurations Parse and Load ^^^^^^^^^^^^^^ +**Loading from a file:** + .. code-block:: python - import chiltepin.configure - import parsl + from chiltepin import workflow - # Parse YAML configuration file - config_dict = chiltepin.configure.parse_file("my_config.yaml") - - # Create Parsl configuration - parsl_config = chiltepin.configure.load( - config_dict, + # Load configuration from YAML file and run workflow + with workflow( + "my_config.yaml", include=["compute", "mpi"], # Only load specific resources run_dir="./runinfo" # Directory for Parsl runtime files - ) + ): + # Run your tasks here + result = my_task(executor=["compute"]).result() + +**Loading from a dict:** - # Initialize Parsl with configuration - parsl.load(parsl_config) +.. code-block:: python + + from chiltepin import workflow + + # Define configuration as a dictionary + config_dict = { + "compute": { + "provider": "slurm", + "account": "my-account", + "partition": "compute", + "nodes_per_block": 1, + "cores_per_node": 40, + "walltime": "01:00:00", + }, + "mpi": { + "provider": "slurm", + "account": "my-account", + "partition": "compute", + "nodes_per_block": 2, + "launcher": "mpi", + "walltime": "02:00:00", + } + } + + # Load configuration from dict and run workflow + with workflow( + config_dict, + include=["compute", "mpi"], # Only load specific resources + run_dir="./runinfo" + ): + # Run your tasks here + result = my_task(executor=["compute"]).result() The ``include`` parameter lets you selectively load only specific resources from your -configuration file. If omitted, all resources are loaded. +configuration. If omitted, all resources are loaded. Configuration Best Practices ----------------------------- diff --git a/docs/container.rst b/docs/container.rst index 5d514008..58c3ed58 100644 --- a/docs/container.rst +++ b/docs/container.rst @@ -66,13 +66,13 @@ After installation, you can run the test suite using the Docker-specific configu .. code-block:: console - (container) $ pytest --assert=plain --config=tests/configs/docker.yaml + (container) $ pytest --config=tests/configs/docker.yaml For more verbose output: .. code-block:: console - (container) $ pytest -s -vvv --assert=plain --config=tests/configs/docker.yaml + (container) $ pytest -s -vvv --config=tests/configs/docker.yaml Adjusting Core Count -------------------- diff --git a/docs/data.rst b/docs/data.rst index 60c7e33e..43e5c373 100644 --- a/docs/data.rst +++ b/docs/data.rst @@ -238,8 +238,7 @@ A common pattern is to stage data, process it, then clean up: .. code-block:: python - import parsl - import chiltepin.configure + from chiltepin import workflow from chiltepin.tasks import python_task from chiltepin.data import transfer_task, delete_task @@ -251,24 +250,21 @@ A common pattern is to stage data, process it, then clean up: result = df.mean().to_dict() return result - # Load configuration and start Parsl - config_dict = chiltepin.configure.parse_file("config.yaml") - parsl_config = chiltepin.configure.load(config_dict) - - with parsl.load(parsl_config): + # Load configuration and start workflow + with workflow("config.yaml"): # Stage data to compute resource stage_in = transfer_task( src_ep="my-laptop", dst_ep="hpc-scratch", src_path="/Users/me/data/dataset.csv", dst_path="/scratch/project/dataset.csv", - executor="local" + executor=["local"] ) # Process the staged data (waits for transfer via inputs) analysis = analyze_data( "/scratch/project/dataset.csv", - executor="compute", + executor=["compute"], inputs=[stage_in] # Non-blocking dependency ) @@ -276,7 +272,7 @@ A common pattern is to stage data, process it, then clean up: cleanup = delete_task( src_ep="hpc-scratch", src_path="/scratch/project/dataset.csv", - executor="local", + executor=["local"], inputs=[analysis] # Non-blocking dependency ) diff --git a/docs/quickstart.rst b/docs/quickstart.rst index a9e6c214..95513c65 100644 --- a/docs/quickstart.rst +++ b/docs/quickstart.rst @@ -116,8 +116,7 @@ Create ``my_workflow.py``: .. code-block:: python - import parsl - import chiltepin.configure + from chiltepin import workflow from chiltepin.tasks import bash_task, python_task # Define tasks @@ -137,23 +136,16 @@ Create ``my_workflow.py``: return result if __name__ == "__main__": - # Load configuration - config_dict = chiltepin.configure.parse_file("my_config.yaml") - parsl_config = chiltepin.configure.load( - config_dict, - include=["local", "remote"], - run_dir="./runinfo" - ) - - with parsl.load(parsl_config): + # Load configuration and run workflow + with chiltepin.workflow("my_config.yaml", include=["local", "remote"], run_dir="./runinfo"): # Run local task on "local" resource - local_future = hello_local(executor="local") + local_future = hello_local(executor=["local"]) # Run remote bash task on "remote" resource (returns exit code: 0 = success) - remote_future = hello_remote(executor="remote") + remote_future = hello_remote(executor=["remote"]) # Run multiple compute tasks on "remote" resource - futures = [compute_task(i, executor="remote") for i in range(1, 5)] + futures = [compute_task(i, executor=["remote"]) for i in range(1, 5)] # Get the results print(f"Local: {local_future.result()}") @@ -213,8 +205,7 @@ Simple Workflow (``simple_workflow.py``) .. code-block:: python - import parsl - import chiltepin.configure + from chiltepin import workflow from chiltepin.tasks import bash_task, python_task # Define tasks @@ -227,15 +218,12 @@ Simple Workflow (``simple_workflow.py``) return "echo 'Task completed successfully'" if __name__ == "__main__": - # Load configuration - config_dict = chiltepin.configure.parse_file("local_config.yaml") - parsl_config = chiltepin.configure.load(config_dict, run_dir="./runinfo") - - with parsl.load(parsl_config): - result = multiply(6, 7, executor="local").result() + # Load configuration and run workflow + with chiltepin.workflow("local_config.yaml", run_dir="./runinfo"): + result = multiply(6, 7, executor=["local"]).result() print(f"6 * 7 = {result}") - exit_code = system_info(executor="local").result() + exit_code = system_info(executor=["local"]).result() print(f"Bash task exit code: {exit_code}") Run it: @@ -274,8 +262,7 @@ MPI Workflow .. code-block:: python - import parsl - import chiltepin.configure + from chiltepin import workflow from chiltepin.tasks import bash_task @bash_task @@ -287,18 +274,15 @@ MPI Workflow return f"srun -n {ranks} ./mpi_app" if __name__ == "__main__": - config_dict = chiltepin.configure.parse_file("mpi_config.yaml") - parsl_config = chiltepin.configure.load(config_dict, run_dir="./runinfo") - - with parsl.load(parsl_config): + with chiltepin.workflow("mpi_config.yaml", run_dir="./runinfo"): # Compile MPI application on the MPI resource (returns exit code) - compile_result = compile_mpi(executor="mpi-resource-name").result() + compile_result = compile_mpi(executor=["mpi-resource-name"]).result() print(f"Compilation exit code: {compile_result}") # Run with different rank counts on the MPI resource results = [] for ranks in [4, 8, 16]: - future = run_mpi(ranks, executor="mpi-resource-name") + future = run_mpi(ranks, executor=["mpi-resource-name"]) results.append(future.result()) for i, result in enumerate(results, 1): @@ -347,16 +331,46 @@ The ``executor`` value must match a resource name from your configuration file. Configuration Loading ^^^^^^^^^^^^^^^^^^^^^ -The ``include`` parameter selects specific resources to load from the configuration: +The ``include`` parameter selects specific resources to load from the configuration. + +**Loading from a file:** + +.. code-block:: python + + # Load only specific resources from YAML file + with workflow( + "my_config.yaml", + include=["local", "compute"], # Only these resources + run_dir="./runinfo" + ): + # Run tasks using selected resources + result = my_task(executor=["compute"]).result() + +**Loading from a dict:** .. code-block:: python - # Load only specific resources - parsl_config = chiltepin.configure.load( - config_dict, + # Define configuration as a dictionary + config = { + "local": { + "provider": "localhost", + "cores_per_node": 4, + }, + "compute": { + "provider": "slurm", + "partition": "compute", + "nodes_per_block": 1, + } + } + + # Load only specific resources from dict + with workflow( + config, include=["local", "compute"], # Only these resources run_dir="./runinfo" - ) + ): + # Run tasks using selected resources + result = my_task(executor=["compute"]).result() If ``include`` is omitted, all resources in the configuration are loaded. diff --git a/docs/testing.rst b/docs/testing.rst index 417731e4..627185fb 100644 --- a/docs/testing.rst +++ b/docs/testing.rst @@ -34,7 +34,7 @@ To run the full test suite: .. code-block:: console - $ pytest --assert=plain --config=tests/configs/.yaml + $ pytest --config=tests/configs/.yaml Where ```` is one of: @@ -53,7 +53,7 @@ For more detailed information during testing: .. code-block:: console - $ pytest -s -vvv --assert=plain --config=tests/configs/.yaml + $ pytest -s -vvv --config=tests/configs/.yaml Running Specific Tests ~~~~~~~~~~~~~~~~~~~~~~ diff --git a/src/chiltepin/__init__.py b/src/chiltepin/__init__.py index 98813136..850cdec6 100644 --- a/src/chiltepin/__init__.py +++ b/src/chiltepin/__init__.py @@ -1 +1,15 @@ # SPDX-License-Identifier: Apache-2.0 + +"""Chiltepin: Federated NWP Workflow Tools. + +This package provides tools for building scientific workflows that can execute +on distributed computing resources using Parsl and Globus services. +""" + +from chiltepin.workflow import workflow, workflow_from_dict, workflow_from_file + +__all__ = [ + "workflow", + "workflow_from_file", + "workflow_from_dict", +] diff --git a/src/chiltepin/workflow.py b/src/chiltepin/workflow.py new file mode 100644 index 00000000..363e12a2 --- /dev/null +++ b/src/chiltepin/workflow.py @@ -0,0 +1,196 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Workflow context managers for Chiltepin. + +This module provides context managers that wrap Parsl configuration and lifecycle +management, eliminating the need for users to directly import or interact with Parsl. +""" + +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Dict, List, Optional, Union + +import parsl +from globus_compute_sdk import Client + +from chiltepin import configure + + +@contextmanager +def workflow( + config: Union[str, Path, Dict[str, Any]], + *, + include: Optional[List[str]] = None, + run_dir: Optional[str] = None, + client: Optional[Client] = None, + log_file: Optional[str] = None, + log_level: Optional[int] = None, +): + """Context manager for Chiltepin workflows. + + This wraps Parsl configuration and lifecycle management, eliminating the need + for users to directly import or interact with Parsl. + + Parameters + ---------- + config : str, Path, or dict + Either a path to a YAML configuration file or a configuration dictionary + include : list of str, optional + List of resource labels to load. If None, all resources are loaded. + run_dir : str, optional + Directory for Parsl runtime files. If None, uses Parsl's default. + client : globus_compute_sdk.Client, optional + Globus Compute client for Globus Compute resources. If None, one will + be created automatically if needed. + log_file : str, optional + Path to Parsl log file. If None, no file logging is configured. + log_level : int, optional + Logging level (e.g., logging.DEBUG). Only used if log_file is provided. + + Yields + ------ + None + The context manager doesn't yield anything. Tasks can be submitted + within the context. + + Examples + -------- + From a configuration file: + + >>> from chiltepin import workflow + >>> from chiltepin.tasks import python_task + >>> + >>> @python_task + >>> def my_task(): + ... return "Hello from workflow!" + >>> + >>> with workflow("config.yaml"): + ... result = my_task(executor=["compute"]) + ... print(result.result()) + + From a configuration dictionary: + + >>> config = {"local": {"provider": "localhost", "cores_per_node": 4}} + >>> with workflow(config): + ... result = my_task(executor=["local"]) + ... print(result.result()) + + With logging and selective resources: + + >>> import logging + >>> with workflow("config.yaml", include=["compute"], + ... log_file="workflow.log", log_level=logging.DEBUG): + ... # only "compute" resource is available + ... result = my_task(executor=["compute"]) + """ + # Parse config if it's a file path + if isinstance(config, (str, Path)): + config_dict = configure.parse_file(str(config)) + else: + config_dict = config + + # Set up logging if requested + logger_handler = None + if log_file is not None: + import logging as log_module + + level = log_level if log_level is not None else log_module.INFO + logger_handler = parsl.set_file_logger(filename=log_file, level=level) + + # Load configuration + parsl_config = configure.load( + config_dict, + include=include, + client=client, + run_dir=run_dir, + ) + + # Load Parsl with the configuration + dfk = parsl.load(parsl_config) + + try: + yield + finally: + # Cleanup - match the pattern used in tests + dfk.cleanup() + dfk = None + parsl.clear() + if logger_handler is not None: + logger_handler() + + +# Convenience aliases for clarity +@contextmanager +def workflow_from_file( + config_file: Union[str, Path], + **kwargs, +): + """Context manager for workflows using a YAML configuration file. + + This is an alias for workflow() that makes it explicit that a file + path is expected. + + Parameters + ---------- + config_file : str or Path + Path to YAML configuration file + **kwargs + Additional arguments passed to workflow() + + Examples + -------- + >>> from chiltepin import workflow_from_file + >>> from chiltepin.tasks import python_task + >>> + >>> @python_task + >>> def my_task(): + ... return 42 + >>> + >>> with workflow_from_file("my_config.yaml", include=["compute"]): + ... result = my_task(executor=["compute"]) + ... print(result.result()) + """ + with workflow(config_file, **kwargs): + yield + + +@contextmanager +def workflow_from_dict( + config_dict: Dict[str, Any], + **kwargs, +): + """Context manager for workflows using a configuration dictionary. + + This is an alias for workflow() that makes it explicit that a dict + is expected. + + Parameters + ---------- + config_dict : dict + Configuration dictionary + **kwargs + Additional arguments passed to workflow() + + Examples + -------- + >>> from chiltepin import workflow_from_dict + >>> from chiltepin.tasks import python_task + >>> + >>> @python_task + >>> def my_task(): + ... return 42 + >>> + >>> config = {"local": {"provider": "localhost"}} + >>> with workflow_from_dict(config): + ... result = my_task(executor=["local"]) + ... print(result.result()) + """ + with workflow(config_dict, **kwargs): + yield + + +__all__ = [ + "workflow", + "workflow_from_file", + "workflow_from_dict", +] diff --git a/tests/test_data.py b/tests/test_data.py index 5aacd1b7..1c978ef0 100644 --- a/tests/test_data.py +++ b/tests/test_data.py @@ -8,9 +8,9 @@ import parsl import pytest -import chiltepin.configure import chiltepin.data as data import chiltepin.endpoint as endpoint +from chiltepin import workflow # Set up fixture to initialize and cleanup Parsl @@ -30,30 +30,18 @@ def config(config_file): clients = endpoint.login() transfer_client = clients["transfer"] - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_data_parsl.log"), - level=logging.DEBUG, - ) - - # Load the default resource configuration - resources = chiltepin.configure.load( - {}, run_dir=str(output_dir / "test_data_runinfo") - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - # Generate unique destination filename to avoid conflicts in concurrent tests unique_dst = f"1MB.to_ursa.{uuid.uuid4()}" - # Run the tests with the loaded resources - yield {"client": transfer_client, "unique_dst": unique_dst} - - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + # Use workflow context manager with default (empty) resource configuration + with workflow( + {}, + run_dir=str(output_dir / "test_data_runinfo"), + log_file=str(output_dir / "test_data_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"client": transfer_client, "unique_dst": unique_dst} def test_data_task_basic(config): diff --git a/tests/test_globus_compute_hello.py b/tests/test_globus_compute_hello.py index cc1aed4f..83fadbe3 100644 --- a/tests/test_globus_compute_hello.py +++ b/tests/test_globus_compute_hello.py @@ -9,6 +9,7 @@ import chiltepin.configure import chiltepin.endpoint as endpoint +from chiltepin import workflow from chiltepin.tasks import bash_task, python_task @@ -58,31 +59,17 @@ def config(config_file): assert len(endpoint_id) == 36 yaml_config["gc-service"]["endpoint"] = f"{endpoint_id}" - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_globus_compute_hello_parsl.log"), - level=logging.DEBUG, - ) - - # Load the finalized resource configuration - resources = chiltepin.configure.load( + # Use workflow context manager for Parsl lifecycle + with workflow( yaml_config, include=["gc-service"], client=compute_client, run_dir=str(output_dir / "test_globus_compute_hello_runinfo"), - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests with the loaded resources - yield {"output_dir": output_dir} - - # Cleanup Parsl after tests are done - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + log_file=str(output_dir / "test_globus_compute_hello_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"output_dir": output_dir} # Stop the test endpoint now that tests are done endpoint.stop("test", config_dir=f"{output_dir}/.globus_compute", timeout=15) diff --git a/tests/test_globus_compute_mpi.py b/tests/test_globus_compute_mpi.py index 3f8033e7..740d55df 100644 --- a/tests/test_globus_compute_mpi.py +++ b/tests/test_globus_compute_mpi.py @@ -11,6 +11,7 @@ import chiltepin.configure import chiltepin.endpoint as endpoint +from chiltepin import workflow from chiltepin.tasks import bash_task @@ -65,31 +66,17 @@ def config(config_file): yaml_config["gc-compute"]["endpoint"] = f"{endpoint_id}" yaml_config["gc-mpi"]["endpoint"] = f"{endpoint_id}" - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_globus_compute_mpi_parsl.log"), - level=logging.DEBUG, - ) - - # Load the finalized resource configuration - resources = chiltepin.configure.load( + # Use workflow context manager for Parsl lifecycle + with workflow( yaml_config, include=["gc-compute", "gc-mpi"], client=compute_client, run_dir=str(output_dir / "test_globus_compute_mpi_runinfo"), - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests with the loaded resources - yield {"output_dir": output_dir} - - # Cleanup Parsl after tests are done - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + log_file=str(output_dir / "test_globus_compute_mpi_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"output_dir": output_dir} # Stop the test endpoint now that tests are done endpoint.stop("test", config_dir=f"{output_dir}/.globus_compute", timeout=15) diff --git a/tests/test_parsl_hello.py b/tests/test_parsl_hello.py index 1578db93..3f805e34 100755 --- a/tests/test_parsl_hello.py +++ b/tests/test_parsl_hello.py @@ -8,6 +8,7 @@ import pytest import chiltepin.configure +from chiltepin import workflow from chiltepin.tasks import bash_task, python_task @@ -26,29 +27,16 @@ def config(config_file): f"export PYTHONPATH=${{PYTHONPATH}}:{pwd.parent.resolve()}" ) - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_parsl_hello_parsl.log"), - level=logging.DEBUG, - ) - - resources = chiltepin.configure.load( + # Use workflow context manager for Parsl lifecycle + with workflow( yaml_config, include=["service"], run_dir=str(output_dir / "test_parsl_hello_runinfo"), - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests with the loaded resources - yield {"output_dir": output_dir} - - # Cleanup Parsl after tests are done - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + log_file=str(output_dir / "test_parsl_hello_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"output_dir": output_dir} # Test bash hello world diff --git a/tests/test_parsl_mpi.py b/tests/test_parsl_mpi.py index 486fa4c9..45458271 100644 --- a/tests/test_parsl_mpi.py +++ b/tests/test_parsl_mpi.py @@ -11,6 +11,7 @@ import pytest import chiltepin.configure +from chiltepin import workflow from chiltepin.tasks import bash_task @@ -33,29 +34,16 @@ def config(config_file): f"export PYTHONPATH=${{PYTHONPATH}}:{pwd.parent.resolve()}" ) - # Set Parsl logging to DEBUG and redirect to a file in the output directory - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_parsl_mpi_parsl.log"), - level=logging.DEBUG, - ) - - resources = chiltepin.configure.load( + # Use workflow context manager for Parsl lifecycle + with workflow( yaml_config, include=["compute", "mpi"], run_dir=str(output_dir / "test_parsl_mpi_runinfo"), - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests with the loaded resources - yield {"output_dir": output_dir} - - # Cleanup Parsl after tests are done - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + log_file=str(output_dir / "test_parsl_mpi_parsl.log"), + log_level=logging.DEBUG, + ): + # Run the tests with the loaded resources + yield {"output_dir": output_dir} def test_parsl_hello_mpi(config): diff --git a/tests/test_tasks.py b/tests/test_tasks.py index 66a210c7..1fbac23a 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -14,7 +14,7 @@ import parsl import pytest -import chiltepin.configure +from chiltepin import workflow from chiltepin.tasks import bash_task, join_task, python_task @@ -28,12 +28,6 @@ def parsl_config(): output_dir = pwd / "test_output" output_dir.mkdir(exist_ok=True) - # Set Parsl logging to DEBUG and redirect to a file - logger_handler = parsl.set_file_logger( - filename=str(output_dir / "test_tasks_parsl.log"), - level=logging.DEBUG, - ) - # Configure local executor with PYTHONPATH that includes the project root # This allows Parsl workers to import test modules project_root = pwd.parent.resolve() @@ -46,22 +40,14 @@ def parsl_config(): } } - # Load configuration with the environment setup - resources = chiltepin.configure.load( - local_config, run_dir=str(output_dir / "test_tasks_runinfo") - ) - - # Load the resources in Parsl - dfk = parsl.load(resources) - - # Run the tests - yield - - # Cleanup - dfk.cleanup() - dfk = None - parsl.clear() - logger_handler() + # Use workflow context manager for Parsl lifecycle + with workflow( + local_config, + run_dir=str(output_dir / "test_tasks_runinfo"), + log_file=str(output_dir / "test_tasks_parsl.log"), + log_level=logging.DEBUG, + ): + yield # ===== Python Task Tests - Standalone Functions ===== diff --git a/tests/test_workflow.py b/tests/test_workflow.py new file mode 100644 index 00000000..fbb97d5c --- /dev/null +++ b/tests/test_workflow.py @@ -0,0 +1,256 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for chiltepin.workflow module. + +This test suite validates that workflow context managers work correctly +for managing Parsl lifecycle without requiring users to directly interact +with Parsl. +""" + +import pathlib + +import parsl +import pytest +import yaml + +import chiltepin.configure +from chiltepin import workflow, workflow_from_dict, workflow_from_file +from chiltepin.tasks import python_task + + +# Helper function to add PYTHONPATH to config +def add_pythonpath_to_config(config_dict, resource_name): + """Add project root to PYTHONPATH for a specific resource.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + if resource_name in config_dict: + env = config_dict[resource_name].get("environment", []) + # Make a copy to avoid modifying shared references + env = env.copy() if env else [] + env.append(f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}") + config_dict[resource_name]["environment"] = env + return config_dict + + +# Cleanup any existing Parsl state before tests +@pytest.fixture(scope="function", autouse=True) +def cleanup_parsl(): + """Ensure Parsl is cleaned up before and after each test.""" + # Cleanup before test + try: + dfk = parsl.dfk() + if dfk: + dfk.cleanup() + parsl.clear() + except Exception: + pass # No DFK loaded, which is fine + + yield + + # Cleanup after test + try: + dfk = parsl.dfk() + if dfk: + dfk.cleanup() + parsl.clear() + except Exception: + pass + + +class TestWorkflowContextManager: + """Test workflow() context manager with different configurations.""" + + def test_workflow_with_dict_config(self, tmp_path): + """Test workflow context manager with a dictionary config.""" + + @python_task + def add_numbers(a, b): + """Simple task for testing.""" + return a + b + + @python_task + def multiply(x, y): + """Another simple task for testing.""" + return x * y + + # Get project root for PYTHONPATH + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-executor": { + "provider": "localhost", + "cores_per_node": 2, + "max_workers_per_node": 2, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + with workflow(config, run_dir=str(tmp_path / "runinfo1")): + # Submit tasks + future1 = add_numbers(10, 32, executor=["test-executor"]) + future2 = multiply(6, 7, executor=["test-executor"]) + + # Get results + result1 = future1.result() + result2 = future2.result() + + assert result1 == 42 + assert result2 == 42 + + def test_workflow_with_file_config(self, config_file, tmp_path): + """Test workflow context manager with a YAML config file.""" + + @python_task + def greet(name): + """Task that returns a greeting.""" + return f"Hello, {name}!" + + # Load config from file and add PYTHONPATH dynamically + config_dict = chiltepin.configure.parse_file(config_file) + add_pythonpath_to_config(config_dict, "service") + + # Write modified config to temp file + temp_config_file = tmp_path / "temp_config.yaml" + with open(temp_config_file, "w") as f: + yaml.dump(config_dict, f) + + # Test with file path argument + with workflow( + str(temp_config_file), + include=["service"], + run_dir=str(tmp_path / "runinfo2"), + ): + future = greet("Workflow", executor=["service"]) + result = future.result() + assert result == "Hello, Workflow!" + + +class TestWorkflowAliases: + """Test workflow_from_dict and workflow_from_file convenience aliases.""" + + def test_workflow_from_dict(self, tmp_path): + """Test workflow_from_dict convenience function.""" + + @python_task + def add_numbers(a, b): + return a + b + + # Get project root for PYTHONPATH + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "my-executor": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + with workflow_from_dict(config, run_dir=str(tmp_path / "runinfo5")): + future = add_numbers(100, 200, executor=["my-executor"]) + result = future.result() + assert result == 300 + + def test_workflow_from_file(self, config_file, tmp_path): + """Test workflow_from_file convenience function.""" + + @python_task + def multiply(x, y): + return x * y + + # Load config from file and add PYTHONPATH dynamically + config_dict = chiltepin.configure.parse_file(config_file) + add_pythonpath_to_config(config_dict, "service") + + # Write modified config to temp file + temp_config_file = tmp_path / "temp_config.yaml" + with open(temp_config_file, "w") as f: + yaml.dump(config_dict, f) + + # Test with file path argument using workflow_from_file + with workflow_from_file( + str(temp_config_file), + include=["service"], + run_dir=str(tmp_path / "runinfo6"), + ): + future = multiply(21, 2, executor=["service"]) + result = future.result() + assert result == 42 + + +class TestWorkflowCleanup: + """Test that workflow context managers properly cleanup resources.""" + + def test_sequential_workflows(self, tmp_path): + """Test that multiple workflows can be created sequentially.""" + + @python_task + def add_numbers(a, b): + return a + b + + @python_task + def multiply(x, y): + return x * y + + # Get project root for PYTHONPATH + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config1 = { + "exec1": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + config2 = { + "exec2": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # First workflow + with workflow(config1, run_dir=str(tmp_path / "runinfo7")): + future = add_numbers(1, 2, executor=["exec1"]) + assert future.result() == 3 + + # Second workflow should work without conflicts + with workflow(config2, run_dir=str(tmp_path / "runinfo8")): + future = multiply(3, 4, executor=["exec2"]) + assert future.result() == 12 + + def test_workflow_cleanup_on_exception(self, tmp_path): + """Test that workflow cleans up even if an exception occurs.""" + + @python_task + def add_numbers(a, b): + return a + b + + # Get project root for PYTHONPATH + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Workflow should cleanup even if exception occurs + with pytest.raises(ValueError): + with workflow(config, run_dir=str(tmp_path / "runinfo9")): + future = add_numbers(5, 5, executor=["test-exec"]) + result = future.result() + assert result == 10 + raise ValueError("Intentional test error") + + # Should be able to create another workflow after exception + with workflow(config, run_dir=str(tmp_path / "runinfo10")): + future = add_numbers(7, 8, executor=["test-exec"]) + assert future.result() == 15 From 89ee77ea055aa010747a0453b79575a935f599fb Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Sat, 7 Mar 2026 21:16:41 +0000 Subject: [PATCH 02/15] Fix formatting --- tests/test_data.py | 1 - tests/test_globus_compute_hello.py | 1 - tests/test_globus_compute_mpi.py | 1 - tests/test_parsl_hello.py | 1 - tests/test_parsl_mpi.py | 1 - 5 files changed, 5 deletions(-) diff --git a/tests/test_data.py b/tests/test_data.py index 1c978ef0..15a1301c 100644 --- a/tests/test_data.py +++ b/tests/test_data.py @@ -5,7 +5,6 @@ import uuid from unittest import mock -import parsl import pytest import chiltepin.data as data diff --git a/tests/test_globus_compute_hello.py b/tests/test_globus_compute_hello.py index 83fadbe3..c53803ca 100644 --- a/tests/test_globus_compute_hello.py +++ b/tests/test_globus_compute_hello.py @@ -4,7 +4,6 @@ import os.path import pathlib -import parsl import pytest import chiltepin.configure diff --git a/tests/test_globus_compute_mpi.py b/tests/test_globus_compute_mpi.py index 740d55df..375f9ae1 100644 --- a/tests/test_globus_compute_mpi.py +++ b/tests/test_globus_compute_mpi.py @@ -6,7 +6,6 @@ import re from datetime import datetime as dt -import parsl import pytest import chiltepin.configure diff --git a/tests/test_parsl_hello.py b/tests/test_parsl_hello.py index 3f805e34..1187b64c 100755 --- a/tests/test_parsl_hello.py +++ b/tests/test_parsl_hello.py @@ -4,7 +4,6 @@ import os.path import pathlib -import parsl import pytest import chiltepin.configure diff --git a/tests/test_parsl_mpi.py b/tests/test_parsl_mpi.py index 45458271..9e2a6ab2 100644 --- a/tests/test_parsl_mpi.py +++ b/tests/test_parsl_mpi.py @@ -7,7 +7,6 @@ import re from datetime import datetime as dt -import parsl import pytest import chiltepin.configure From 2d433d8773a068de893575ff7a5692c67cfd3a30 Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Sat, 7 Mar 2026 21:43:51 +0000 Subject: [PATCH 03/15] Fix possible uncaught exception in workflow context manager --- docs/quickstart.rst | 2 +- src/chiltepin/workflow.py | 32 ++++++++++++++++++-------------- 2 files changed, 19 insertions(+), 15 deletions(-) diff --git a/docs/quickstart.rst b/docs/quickstart.rst index 95513c65..ab70475f 100644 --- a/docs/quickstart.rst +++ b/docs/quickstart.rst @@ -274,7 +274,7 @@ MPI Workflow return f"srun -n {ranks} ./mpi_app" if __name__ == "__main__": - with chiltepin.workflow("mpi_config.yaml", run_dir="./runinfo"): + with workflow("mpi_config.yaml", run_dir="./runinfo"): # Compile MPI application on the MPI resource (returns exit code) compile_result = compile_mpi(executor=["mpi-resource-name"]).result() print(f"Compilation exit code: {compile_result}") diff --git a/src/chiltepin/workflow.py b/src/chiltepin/workflow.py index 363e12a2..7a6fb0aa 100644 --- a/src/chiltepin/workflow.py +++ b/src/chiltepin/workflow.py @@ -97,24 +97,28 @@ def workflow( level = log_level if log_level is not None else log_module.INFO logger_handler = parsl.set_file_logger(filename=log_file, level=level) - # Load configuration - parsl_config = configure.load( - config_dict, - include=include, - client=client, - run_dir=run_dir, - ) - - # Load Parsl with the configuration - dfk = parsl.load(parsl_config) + # Initialize dfk to None before attempting to load + dfk = None try: + # Load configuration + parsl_config = configure.load( + config_dict, + include=include, + client=client, + run_dir=run_dir, + ) + + # Load Parsl with the configuration + dfk = parsl.load(parsl_config) + yield finally: - # Cleanup - match the pattern used in tests - dfk.cleanup() - dfk = None - parsl.clear() + # Cleanup only if dfk was successfully created + if dfk is not None: + dfk.cleanup() + dfk = None + parsl.clear() if logger_handler is not None: logger_handler() From a7173d1617b9a7b534ebd80876e3363223f6ba17 Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Sun, 8 Mar 2026 00:23:32 +0000 Subject: [PATCH 04/15] Fix inconsistent executor args and improve workflow context cleanup resilience --- docs/data.rst | 26 +++++----- docs/quickstart.rst | 6 +-- docs/tasks.rst | 106 +++++++++++++++++++------------------- src/chiltepin/__init__.py | 19 ++++++- src/chiltepin/data.py | 6 +-- src/chiltepin/tasks.py | 4 +- src/chiltepin/workflow.py | 32 ++++++++++-- tests/test_tasks.py | 2 +- 8 files changed, 119 insertions(+), 82 deletions(-) diff --git a/docs/data.rst b/docs/data.rst index 43e5c373..c8d88034 100644 --- a/docs/data.rst +++ b/docs/data.rst @@ -45,7 +45,7 @@ Basic Usage dst_ep="hpc-scratch", src_path="/Users/me/data/input.dat", dst_path="/scratch/project/input.dat", - executor="local" + executor=["local"] ) # Wait for transfer to complete @@ -115,7 +115,7 @@ Transfer entire directories recursively: src_path="/Users/me/project/data/", dst_path="/scratch/project/data/", recursive=True, - executor="local" + executor=["local"] ) Endpoint Names vs UUIDs @@ -159,7 +159,7 @@ Basic Usage delete_future = delete_task( src_ep="hpc-scratch", src_path="/scratch/project/temp.dat", - executor="local" + executor=["local"] ) # Wait for deletion to complete @@ -219,7 +219,7 @@ Delete entire directories: src_ep="hpc-scratch", src_path="/scratch/project/temp_data/", recursive=True, - executor="local" + executor=["local"] ) .. warning:: @@ -302,7 +302,7 @@ Transfer multiple files in parallel: dst_ep="hpc-scratch", src_path=f"/Users/me/data/{filename}", dst_path=f"/scratch/project/{filename}", - executor="local" + executor=["local"] ) transfers.append(future) @@ -336,8 +336,8 @@ To run a transfer or deletion after multiple tasks complete, pass them via the return True # Generate files in parallel - config_ready = generate_config(executor="compute") - input_ready = generate_input(executor="compute") + config_ready = generate_config(executor=["compute"]) + input_ready = generate_input(executor=["compute"]) # Wait for both files before transferring (non-blocking dependency) transfer = transfer_task( @@ -346,7 +346,7 @@ To run a transfer or deletion after multiple tasks complete, pass them via the src_path="/scratch/", dst_path="/results/", recursive=True, - executor="local", + executor=["local"], inputs=[config_ready, input_ready] # Multiple dependencies ) @@ -382,21 +382,21 @@ dependencies: dst_ep="hpc-scratch", src_path="/data/raw.csv", dst_path="/scratch/raw.csv", - executor="local" + executor=["local"] ) # Preprocess (waits for transfer via inputs) preprocess_future = preprocess( "/scratch/raw.csv", "/scratch/clean.csv", - executor="compute", + executor=["compute"], inputs=[stage_raw] # Wait for transfer non-blocking ) # Analyze (waits for preprocess completing and passing output path) analysis_future = analyze( preprocess_future, # Parsl waits for this future and passes the output path - executor="compute" + executor=["compute"] ) # Stage results back (waits for analysis via inputs) @@ -405,7 +405,7 @@ dependencies: dst_ep="my-laptop", src_path="/scratch/clean.csv", dst_path="/data/cleaned_output.csv", - executor="local", + executor=["local"], inputs=[analysis_future] # Wait for analysis non-blocking ) @@ -414,7 +414,7 @@ dependencies: src_ep="hpc-scratch", src_path="/scratch/", recursive=True, - executor="local", + executor=["local"], inputs=[stage_out] # Wait for stage out non-blocking ) diff --git a/docs/quickstart.rst b/docs/quickstart.rst index ab70475f..0787c38a 100644 --- a/docs/quickstart.rst +++ b/docs/quickstart.rst @@ -137,7 +137,7 @@ Create ``my_workflow.py``: if __name__ == "__main__": # Load configuration and run workflow - with chiltepin.workflow("my_config.yaml", include=["local", "remote"], run_dir="./runinfo"): + with workflow("my_config.yaml", include=["local", "remote"], run_dir="./runinfo"): # Run local task on "local" resource local_future = hello_local(executor=["local"]) @@ -219,7 +219,7 @@ Simple Workflow (``simple_workflow.py``) if __name__ == "__main__": # Load configuration and run workflow - with chiltepin.workflow("local_config.yaml", run_dir="./runinfo"): + with workflow("local_config.yaml", run_dir="./runinfo"): result = multiply(6, 7, executor=["local"]).result() print(f"6 * 7 = {result}") @@ -320,7 +320,7 @@ When calling a task, use the ``executor`` parameter to specify which resource to return "result" # Specify which resource to use - result = my_task(executor="compute").result() + result = my_task(executor=["compute"]).result() The ``executor`` value must match a resource name from your configuration file. diff --git a/docs/tasks.rst b/docs/tasks.rst index 6b5b6c7c..aa3e57d6 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -47,7 +47,7 @@ Basic Usage return "Hello from a Chiltepin task!" # Call the task and specify where to run it - future = hello_world(executor="my-resource") + future = hello_world(executor=["my-resource"]) result = future.result() # Wait for completion and get result print(result) # "Hello from a Chiltepin task!" @@ -63,11 +63,11 @@ Python tasks can accept both positional and keyword arguments: return (a + b) * multiply_by # Use positional arguments - future1 = add_numbers(5, 3, executor="compute") + future1 = add_numbers(5, 3, executor=["compute"]) print(future1.result()) # 8 # Use keyword arguments - future2 = add_numbers(5, 3, multiply_by=2, executor="compute") + future2 = add_numbers(5, 3, multiply_by=2, executor=["compute"]) print(future2.result()) # 16 Importing Modules @@ -111,9 +111,9 @@ Python tasks can return any serializable Python object: import pandas as pd return pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]}) - list_result = get_list(5, executor="local").result() - dict_result = get_dict("temperature", 72.5, executor="local").result() - df_result = get_dataframe(executor="compute").result() + list_result = get_list(5, executor=["local"]).result() + dict_result = get_dict("temperature", 72.5, executor=["local"]).result() + df_result = get_dataframe(executor=["compute"]).result() Bash Tasks ---------- @@ -133,7 +133,7 @@ Basic Usage return "echo 'Hello from bash!'" # Bash tasks return the exit code (0 = success) - future = echo_hello(executor="my-resource") + future = echo_hello(executor=["my-resource"]) exit_code = future.result() print(exit_code) # 0 @@ -152,7 +152,7 @@ Use function arguments to dynamically construct commands: def run_simulation(config_file, num_steps): return f"./my_simulator --config {config_file} --steps {num_steps}" - exit_code = process_file("data.txt", "sorted.txt", executor="compute").result() + exit_code = process_file("data.txt", "sorted.txt", executor=["compute"]).result() .. warning:: Be careful with shell injection vulnerabilities. Validate and sanitize inputs @@ -172,7 +172,7 @@ By default, bash tasks return the exit code. To capture stdout or stderr, use th # Capture stdout to a file future = get_hostname( - executor="compute", + executor=["compute"], stdout="hostname_output.txt" ) exit_code = future.result() @@ -196,7 +196,7 @@ You can also capture stderr for debugging: return "some_command_that_might_fail" future = risky_command( - executor="compute", + executor=["compute"], stdout="output.txt", stderr="errors.txt" ) @@ -245,9 +245,9 @@ Join tasks call other tasks and return futures: @join_task def process_list(values, factor): # Launch multiple tasks in parallel - futures = [multiply(v, factor, executor="compute") for v in values] + futures = [multiply(v, factor, executor=["compute"]) for v in values] # Aggregate results with another task - return add_values(*futures, executor="compute") + return add_values(*futures, executor=["compute"]) # Process [1, 2, 3] with factor 2: (1*2) + (2*2) + (3*2) = 12 result = process_list([1, 2, 3], 2).result() @@ -277,9 +277,9 @@ Example - Processing Multiple Files: @join_task def process_all_files(file_list): # Process all files in parallel - futures = [process_file(f, executor="compute") for f in file_list] + futures = [process_file(f, executor=["compute"]) for f in file_list] # Check if all succeeded - return check_all_success(futures, executor="local") + return check_all_success(futures, executor=["local"]) files = ["data1.txt", "data2.txt", "data3.txt"] success = process_all_files(files).result() @@ -307,15 +307,15 @@ Join tasks can coordinate both python and bash tasks: @join_task def compile_and_run(input_file): # First compile - compile_future = compile_code(executor="compute") + compile_future = compile_code(executor=["compute"]) compile_future.result() # Wait for compilation # Then run - run_future = run_app(input_file, executor="compute", stdout="output.txt") + run_future = run_app(input_file, executor=["compute"], stdout="output.txt") run_future.result() # Wait for execution # Parse results - return parse_results("output.txt", executor="local") + return parse_results("output.txt", executor=["local"]) result = compile_and_run("input.dat").result() @@ -354,9 +354,9 @@ object-oriented workflow design: # Create instance and use tasks processor = DataProcessor({'normalize': True, 'export_format': 'json'}) - data = processor.load_data("input.csv", executor="compute").result() - transformed = processor.transform_data(data, executor="compute").result() - exit_code = processor.export_data("output.json", executor="compute").result() + data = processor.load_data("input.csv", executor=["compute"]).result() + transformed = processor.transform_data(data, executor=["compute"]).result() + exit_code = processor.export_data("output.json", executor=["compute"]).result() .. warning:: **Mutable Object State**: When using class methods as tasks, be aware that mutable @@ -389,7 +389,7 @@ Specify a single resource by name: return "result" # Run on the "compute" resource from your config - future = my_task(executor="compute") + future = my_task(executor=["compute"]) Multiple Resources ^^^^^^^^^^^^^^^^^^ @@ -433,7 +433,7 @@ Call ``.result()`` to wait for task completion and retrieve the result: time.sleep(2) return 42 - future = compute_value(executor="compute") + future = compute_value(executor=["compute"]) print("Task submitted, doing other work...") # This blocks until the task completes @@ -447,7 +447,7 @@ Check if a task is done without blocking: .. code-block:: python - future = compute_value(executor="compute") + future = compute_value(executor=["compute"]) if future.done(): print("Task completed!") @@ -463,7 +463,7 @@ Wait for multiple tasks efficiently: .. code-block:: python # Launch multiple tasks - futures = [compute_value(executor="compute") for _ in range(10)] + futures = [compute_value(executor=["compute"]) for _ in range(10)] # Wait for all to complete results = [f.result() for f in futures] @@ -480,7 +480,7 @@ Exceptions raised in tasks are re-raised when calling ``.result()``: def failing_task(): raise ValueError("Something went wrong!") - future = failing_task(executor="compute") + future = failing_task(executor=["compute"]) try: result = future.result() @@ -539,17 +539,17 @@ transfer and deletion tasks that can be incorporated into your workflows: dst_ep="my-dest-endpoint", src_path="/data/input.dat", dst_path="/scratch/input.dat", - executor="local" + executor=["local"] ) # Process the transferred data (waits for transfer by passing its future) - result = process_file(transfer, "/scratch/input.dat", executor="compute") + result = process_file(transfer, "/scratch/input.dat", executor=["compute"]) # Clean up after processing cleanup = delete_task( src_ep="my-dest-endpoint", src_path="/scratch/input.dat", - executor="local", + executor=["local"], inputs=[result] # Waits for processing to complete ) cleanup.result() @@ -574,8 +574,8 @@ Pass data directly through futures: return sum(data) # Data flows through futures - data_future = generate_data(100, executor="compute") - sum_future = sum_data(data_future, executor="compute") + data_future = generate_data(100, executor=["compute"]) + sum_future = sum_data(data_future, executor=["compute"]) result = sum_future.result() For large data, consider files or data staging strategies. @@ -625,9 +625,9 @@ The most common approach is to pass a future from one task as an argument to ano return f"final_{input_data}" # Chain tasks - data flows through futures - future1 = step1(executor="compute") - future2 = step2(future1, executor="compute") # Waits for future1 - future3 = step3(future2, executor="compute") # Waits for future2 + future1 = step1(executor=["compute"]) + future2 = step2(future1, executor=["compute"]) # Waits for future1 + future3 = step3(future2, executor=["compute"]) # Waits for future2 final_result = future3.result() @@ -649,7 +649,7 @@ parameter. This is automatically supported by all Chiltepin task decorators dst_ep="hpc-scratch", src_path="/data/input.dat", dst_path="/scratch/input.dat", - executor="local" + executor=["local"] ) # Process the data - waits for transfer without passing its result @@ -658,13 +658,13 @@ parameter. This is automatically supported by all Chiltepin task decorators with open(filepath) as f: return len(f.read()) - result = process_data("/scratch/input.dat", executor="compute", inputs=[stage]) + result = process_data("/scratch/input.dat", executor=["compute"], inputs=[stage]) # Clean up - waits for processing to complete cleanup = delete_task( src_ep="hpc-scratch", src_path="/scratch/input.dat", - executor="local", + executor=["local"], inputs=[result] ) @@ -700,12 +700,12 @@ You can combine both approaches and specify multiple dependencies: def combine(data1, data2): return f"{data1}_{data2}" - a = task_a(executor="compute") - b = task_b(executor="compute") - c = task_c(executor="compute") + a = task_a(executor=["compute"]) + b = task_b(executor=["compute"]) + c = task_c(executor=["compute"]) # Combine waits for a and b (via arguments) and c (via inputs) - result = combine(a, b, executor="compute", inputs=[c]) + result = combine(a, b, executor=["compute"], inputs=[c]) .. tip:: **Avoid premature .result() calls**: In this example, notice that ``.result()`` is only @@ -717,17 +717,17 @@ You can combine both approaches and specify multiple dependencies: .. code-block:: python - result1 = step1(executor="compute").result() # Blocks here - result2 = step2(result1, executor="compute").result() # Blocks here - result3 = step3(result2, executor="compute").result() # Blocks here + result1 = step1(executor=["compute"]).result() # Blocks here + result2 = step2(result1, executor=["compute"]).result() # Blocks here + result3 = step3(result2, executor=["compute"]).result() # Blocks here **Good practice** (maximizes parallelism): .. code-block:: python - future1 = step1(executor="compute") - future2 = step2(future1, executor="compute") # Scheduled, doesn't block - future3 = step3(future2, executor="compute") # Scheduled, doesn't block + future1 = step1(executor=["compute"]) + future2 = step2(future1, executor=["compute"]) # Scheduled, doesn't block + future3 = step3(future2, executor=["compute"]) # Scheduled, doesn't block result = future3.result() # Only block when you need the final result Timeout Handling @@ -745,7 +745,7 @@ Handle long-running tasks with timeouts: time.sleep(100) return "done" - future = long_task(executor="compute") + future = long_task(executor=["compute"]) try: result = future.result(timeout=10) # Wait max 10 seconds @@ -812,9 +812,9 @@ Map-Reduce @join_task def map_reduce(items): # Map phase - futures = [map_task(item, executor="compute") for item in items] + futures = [map_task(item, executor=["compute"]) for item in items] # Reduce phase - return reduce_task(futures, executor="compute") + return reduce_task(futures, executor=["compute"]) result = map_reduce([1, 2, 3, 4, 5]).result() # 55 @@ -836,9 +836,9 @@ Pipeline Processing return data ** 2 # Create pipeline - data1 = stage1(5, executor="compute") - data2 = stage2(data1, executor="compute") - result = stage3(data2, executor="compute").result() # ((5*2)+10)^2 = 400 + data1 = stage1(5, executor=["compute"]) + data2 = stage2(data1, executor=["compute"]) + result = stage3(data2, executor=["compute"]).result() # ((5*2)+10)^2 = 400 Parameter Sweep ^^^^^^^^^^^^^^^ @@ -855,7 +855,7 @@ Parameter Sweep futures = [] for p1 in [1, 2, 3]: for p2 in [10, 20, 30]: - future = run_experiment(p1, p2, executor="compute") + future = run_experiment(p1, p2, executor=["compute"]) futures.append(future) # Collect all results diff --git a/src/chiltepin/__init__.py b/src/chiltepin/__init__.py index 850cdec6..bfdebb32 100644 --- a/src/chiltepin/__init__.py +++ b/src/chiltepin/__init__.py @@ -6,10 +6,25 @@ on distributed computing resources using Parsl and Globus services. """ -from chiltepin.workflow import workflow, workflow_from_dict, workflow_from_file - __all__ = [ "workflow", "workflow_from_file", "workflow_from_dict", ] + + +def __getattr__(name): + """Lazy import of workflow functions to avoid loading Parsl unnecessarily. + + This allows users to import other chiltepin submodules (like configure, tasks) + without triggering the import of Parsl and its dependencies unless they + actually use the workflow context managers. + """ + if name in __all__: + from chiltepin.workflow import workflow, workflow_from_dict, workflow_from_file + + globals()[name] = locals()[name] + return locals()[name] + + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + diff --git a/src/chiltepin/data.py b/src/chiltepin/data.py index a853f12e..0e07ed4e 100644 --- a/src/chiltepin/data.py +++ b/src/chiltepin/data.py @@ -37,17 +37,17 @@ def process_data(input_path): dst_ep="hpc-scratch", src_path="/data/input.dat", dst_path="/scratch/input.dat", - executor="local" + executor=["local"] ) # Process the staged data (waits for stage via inputs parameter) - result = process_data("/scratch/input.dat", executor="compute", inputs=[stage]) + result = process_data("/scratch/input.dat", executor=["compute"], inputs=[stage]) # Clean up after processing completes (waits for result via inputs) cleanup = delete_task( src_ep="hpc-scratch", src_path="/scratch/input.dat", - executor="local", + executor=["local"], inputs=[result] ) diff --git a/src/chiltepin/tasks.py b/src/chiltepin/tasks.py index af4f9593..d1050db6 100644 --- a/src/chiltepin/tasks.py +++ b/src/chiltepin/tasks.py @@ -24,7 +24,7 @@ def add_numbers(a, b): return a + b # Execute on a specific resource - result = add_numbers(5, 3, executor="compute").result() + result = add_numbers(5, 3, executor=["compute"]).result() Define a bash task:: @@ -35,7 +35,7 @@ def list_files(directory): return f"ls -la {directory}" # Returns exit code (0 = success) - exit_code = list_files("/tmp", executor="compute").result() + exit_code = list_files("/tmp", executor=["compute"]).result() """ from functools import wraps diff --git a/src/chiltepin/workflow.py b/src/chiltepin/workflow.py index 7a6fb0aa..268c079f 100644 --- a/src/chiltepin/workflow.py +++ b/src/chiltepin/workflow.py @@ -114,13 +114,35 @@ def workflow( yield finally: - # Cleanup only if dfk was successfully created + # Cleanup operations - each wrapped in try/except to ensure all are attempted + # even if some fail + cleanup_errors = [] + + # Cleanup DataFlowKernel if dfk is not None: - dfk.cleanup() - dfk = None - parsl.clear() + try: + dfk.cleanup() + except Exception as e: + cleanup_errors.append(f"dfk.cleanup() failed: {e}") + + try: + parsl.clear() + except Exception as e: + cleanup_errors.append(f"parsl.clear() failed: {e}") + + # Cleanup logger handler if logger_handler is not None: - logger_handler() + try: + logger_handler() + except Exception as e: + cleanup_errors.append(f"logger_handler() failed: {e}") + + # If any cleanup operations failed, raise an exception with all errors + if cleanup_errors: + error_msg = "Errors during workflow cleanup:\n" + "\n".join( + f" - {err}" for err in cleanup_errors + ) + raise RuntimeError(error_msg) # Convenience aliases for clarity diff --git a/tests/test_tasks.py b/tests/test_tasks.py index 1fbac23a..b149f29f 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -589,7 +589,7 @@ def task2(): return 2 f1 = task1(executor=["test-local"]) - f2 = task2(executor="all") + f2 = task2(executor="all") # Using "all" to test that it works as an alias for all executors assert f1.result() == 1 assert f2.result() == 2 From 1450b1475d5946e3d69e0aadd0ac4cbd80956ab0 Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Sun, 8 Mar 2026 00:30:36 +0000 Subject: [PATCH 05/15] Fix linting issues --- src/chiltepin/__init__.py | 7 +++++-- tests/test_tasks.py | 4 +++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/src/chiltepin/__init__.py b/src/chiltepin/__init__.py index bfdebb32..4530aacc 100644 --- a/src/chiltepin/__init__.py +++ b/src/chiltepin/__init__.py @@ -21,10 +21,13 @@ def __getattr__(name): actually use the workflow context managers. """ if name in __all__: - from chiltepin.workflow import workflow, workflow_from_dict, workflow_from_file + from chiltepin.workflow import ( # noqa: F401 + workflow, + workflow_from_dict, + workflow_from_file, + ) globals()[name] = locals()[name] return locals()[name] raise AttributeError(f"module {__name__!r} has no attribute {name!r}") - diff --git a/tests/test_tasks.py b/tests/test_tasks.py index b149f29f..4249daa0 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -589,7 +589,9 @@ def task2(): return 2 f1 = task1(executor=["test-local"]) - f2 = task2(executor="all") # Using "all" to test that it works as an alias for all executors + f2 = task2( + executor="all" + ) # Using "all" to test that it works as an alias for all executors assert f1.result() == 1 assert f2.result() == 2 From fd765b2e692735a2a3954462944dedc98bb637ec Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Sun, 8 Mar 2026 01:22:00 +0000 Subject: [PATCH 06/15] Don't use "local" executor in examples, update workflow exception tests --- docs/quickstart.rst | 22 ++-- src/chiltepin/workflow.py | 48 ++++---- tests/test_workflow.py | 246 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 285 insertions(+), 31 deletions(-) diff --git a/docs/quickstart.rst b/docs/quickstart.rst index 0787c38a..022a8ddc 100644 --- a/docs/quickstart.rst +++ b/docs/quickstart.rst @@ -89,8 +89,8 @@ Create ``my_config.yaml`` with your endpoint UUID: .. code-block:: yaml - # Local resource for small tasks - local: + # Laptop/workstation resource for small tasks + laptop: provider: "localhost" init_blocks: 1 max_blocks: 1 @@ -137,9 +137,9 @@ Create ``my_workflow.py``: if __name__ == "__main__": # Load configuration and run workflow - with workflow("my_config.yaml", include=["local", "remote"], run_dir="./runinfo"): - # Run local task on "local" resource - local_future = hello_local(executor=["local"]) + with workflow("my_config.yaml", include=["laptop", "remote"], run_dir="./runinfo"): + # Run local task on "laptop" resource + local_future = hello_local(executor=["laptop"]) # Run remote bash task on "remote" resource (returns exit code: 0 = success) remote_future = hello_remote(executor=["remote"]) @@ -195,7 +195,7 @@ Configuration File (``local_config.yaml``) .. code-block:: yaml - local: + laptop: provider: "localhost" init_blocks: 1 max_blocks: 1 @@ -220,10 +220,10 @@ Simple Workflow (``simple_workflow.py``) if __name__ == "__main__": # Load configuration and run workflow with workflow("local_config.yaml", run_dir="./runinfo"): - result = multiply(6, 7, executor=["local"]).result() + result = multiply(6, 7, executor=["laptop"]).result() print(f"6 * 7 = {result}") - exit_code = system_info(executor=["local"]).result() + exit_code = system_info(executor=["laptop"]).result() print(f"Bash task exit code: {exit_code}") Run it: @@ -340,7 +340,7 @@ The ``include`` parameter selects specific resources to load from the configurat # Load only specific resources from YAML file with workflow( "my_config.yaml", - include=["local", "compute"], # Only these resources + include=["laptop", "compute"], # Only these resources run_dir="./runinfo" ): # Run tasks using selected resources @@ -352,7 +352,7 @@ The ``include`` parameter selects specific resources to load from the configurat # Define configuration as a dictionary config = { - "local": { + "laptop": { "provider": "localhost", "cores_per_node": 4, }, @@ -366,7 +366,7 @@ The ``include`` parameter selects specific resources to load from the configurat # Load only specific resources from dict with workflow( config, - include=["local", "compute"], # Only these resources + include=["laptop", "compute"], # Only these resources run_dir="./runinfo" ): # Run tasks using selected resources diff --git a/src/chiltepin/workflow.py b/src/chiltepin/workflow.py index 268c079f..f3e36d17 100644 --- a/src/chiltepin/workflow.py +++ b/src/chiltepin/workflow.py @@ -70,9 +70,9 @@ def workflow( From a configuration dictionary: - >>> config = {"local": {"provider": "localhost", "cores_per_node": 4}} + >>> config = {"laptop": {"provider": "localhost", "cores_per_node": 4}} >>> with workflow(config): - ... result = my_task(executor=["local"]) + ... result = my_task(executor=["laptop"]) ... print(result.result()) With logging and selective resources: @@ -115,34 +115,42 @@ def workflow( yield finally: # Cleanup operations - each wrapped in try/except to ensure all are attempted - # even if some fail - cleanup_errors = [] + # even if some fail. Exceptions are chained together. + cleanup_exception = None - # Cleanup DataFlowKernel + # Cleanup DataFlowKernel if it was created if dfk is not None: try: dfk.cleanup() except Exception as e: - cleanup_errors.append(f"dfk.cleanup() failed: {e}") - - try: - parsl.clear() - except Exception as e: - cleanup_errors.append(f"parsl.clear() failed: {e}") + cleanup_exception = e + + # Always call parsl.clear() regardless of dfk state + try: + parsl.clear() + except Exception as e: + if cleanup_exception is None: + cleanup_exception = e + else: + # Chain this exception to the previous one + e.__context__ = cleanup_exception + cleanup_exception = e # Cleanup logger handler if logger_handler is not None: try: logger_handler() except Exception as e: - cleanup_errors.append(f"logger_handler() failed: {e}") + if cleanup_exception is None: + cleanup_exception = e + else: + # Chain this exception to the previous one + e.__context__ = cleanup_exception + cleanup_exception = e - # If any cleanup operations failed, raise an exception with all errors - if cleanup_errors: - error_msg = "Errors during workflow cleanup:\n" + "\n".join( - f" - {err}" for err in cleanup_errors - ) - raise RuntimeError(error_msg) + # If any cleanup failed, raise the last exception (with others chained via __context__) + if cleanup_exception is not None: + raise cleanup_exception # Convenience aliases for clarity @@ -206,9 +214,9 @@ def workflow_from_dict( >>> def my_task(): ... return 42 >>> - >>> config = {"local": {"provider": "localhost"}} + >>> config = {"laptop": {"provider": "localhost"}} >>> with workflow_from_dict(config): - ... result = my_task(executor=["local"]) + ... result = my_task(executor=["laptop"]) ... print(result.result()) """ with workflow(config_dict, **kwargs): diff --git a/tests/test_workflow.py b/tests/test_workflow.py index fbb97d5c..7bbc40cf 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -8,6 +8,7 @@ """ import pathlib +from unittest import mock import parsl import pytest @@ -254,3 +255,248 @@ def add_numbers(a, b): with workflow(config, run_dir=str(tmp_path / "runinfo10")): future = add_numbers(7, 8, executor=["test-exec"]) assert future.result() == 15 + + +class TestWorkflowExceptionHandling: + """Test exception handling during workflow cleanup.""" + + def test_dfk_cleanup_exception(self, tmp_path): + """Test that exceptions during dfk.cleanup() are properly raised.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + + with pytest.raises(RuntimeError, match="Cleanup failed"): + with workflow(config, run_dir=str(tmp_path / "runinfo_exc1")): + pass + + def test_parsl_clear_exception(self, tmp_path): + """Test that exceptions during parsl.clear() are properly raised.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + with mock.patch("parsl.clear") as mock_clear: + mock_clear.side_effect = RuntimeError("Clear failed") + + with pytest.raises(RuntimeError, match="Clear failed"): + with workflow(config, run_dir=str(tmp_path / "runinfo_exc2")): + pass + + def test_logger_handler_exception(self, tmp_path): + """Test that exceptions during logger_handler() are properly raised.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Mock the logger handler to raise an exception + def mock_set_file_logger(*args, **kwargs): + def failing_handler(): + raise RuntimeError("Logger cleanup failed") + + return failing_handler + + with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): + with pytest.raises(RuntimeError, match="Logger cleanup failed"): + with workflow( + config, + run_dir=str(tmp_path / "runinfo_exc3"), + log_file=str(tmp_path / "test.log"), + ): + pass + + def test_chained_exceptions_cleanup_then_clear(self, tmp_path): + """Test exception chaining when dfk.cleanup() and parsl.clear() both fail.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + with mock.patch("parsl.clear") as mock_clear: + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + mock_clear.side_effect = RuntimeError("Clear failed") + + with pytest.raises(RuntimeError) as exc_info: + with workflow(config, run_dir=str(tmp_path / "runinfo_exc4")): + pass + + # The last exception (clear) should be raised + assert "Clear failed" in str(exc_info.value) + # And the previous exception (cleanup) should be in the chain + assert exc_info.value.__context__ is not None + assert "Cleanup failed" in str(exc_info.value.__context__) + + def test_chained_exceptions_all_three(self, tmp_path): + """Test exception chaining when all three cleanup operations fail.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Mock the logger handler to raise an exception + def mock_set_file_logger(*args, **kwargs): + def failing_handler(): + raise RuntimeError("Logger cleanup failed") + + return failing_handler + + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + with mock.patch("parsl.clear") as mock_clear: + with mock.patch( + "parsl.set_file_logger", side_effect=mock_set_file_logger + ): + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + mock_clear.side_effect = RuntimeError("Clear failed") + + with pytest.raises(RuntimeError) as exc_info: + with workflow( + config, + run_dir=str(tmp_path / "runinfo_exc5"), + log_file=str(tmp_path / "test.log"), + ): + pass + + # The last exception (logger) should be raised + assert "Logger cleanup failed" in str(exc_info.value) + # Check the exception chain + assert exc_info.value.__context__ is not None + assert "Clear failed" in str(exc_info.value.__context__) + assert exc_info.value.__context__.__context__ is not None + assert "Cleanup failed" in str( + exc_info.value.__context__.__context__ + ) + + def test_parsl_clear_called_when_dfk_is_none(self, tmp_path): + """Test that parsl.clear() is called even when dfk is None.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Mock parsl.load to fail, so dfk stays None + with mock.patch("parsl.load") as mock_load: + with mock.patch("parsl.clear") as mock_clear: + mock_load.side_effect = RuntimeError("Load failed") + + with pytest.raises(RuntimeError, match="Load failed"): + with workflow(config, run_dir=str(tmp_path / "runinfo_exc6")): + pass + + # parsl.clear() should still be called even though dfk is None + mock_clear.assert_called_once() + + def test_parsl_clear_exception_without_cleanup_exception(self, tmp_path): + """Test parsl.clear() exception when dfk.cleanup() succeeds.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # This tests the branch where parsl.clear() fails but dfk.cleanup() succeeded + # We need to patch the real parsl.clear in the workflow module's finally block + original_clear = parsl.clear + call_count = [0] + + def mock_clear_that_fails_once(): + call_count[0] += 1 + # Fail on first call (from workflow cleanup), succeed on second (from test cleanup) + if call_count[0] == 1: + raise RuntimeError("Clear failed") + original_clear() + + with mock.patch("parsl.clear", side_effect=mock_clear_that_fails_once): + with pytest.raises(RuntimeError, match="Clear failed"): + with workflow(config, run_dir=str(tmp_path / "runinfo_exc7")): + pass + + +@pytest.fixture +def config_file_fixture(tmp_path): + """Create a temporary config file for testing file-based workflows.""" + import yaml + + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "service": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + config_file = tmp_path / "test_config.yaml" + with open(config_file, "w") as f: + yaml.dump(config, f) + + return str(config_file) + + +class TestWorkflowFromFileCoverage: + """Tests specifically for workflow_from_file function coverage.""" + + def test_workflow_from_file_with_local_fixture(self, config_file_fixture, tmp_path): + """Test workflow_from_file with a local config file fixture.""" + + @python_task + def simple_task(): + return "success" + + with workflow_from_file( + config_file_fixture, + include=["service"], + run_dir=str(tmp_path / "runinfo_file"), + ): + future = simple_task(executor=["service"]) + result = future.result() + assert result == "success" From 50642ef21265722864a22ade1a1a952792c9d007 Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Sun, 8 Mar 2026 18:54:18 +0000 Subject: [PATCH 07/15] Inprove workflow context testing --- src/chiltepin/workflow.py | 42 ++++-- tests/test_workflow.py | 287 ++++++++++++++++++++++++++++++++++++-- 2 files changed, 310 insertions(+), 19 deletions(-) diff --git a/src/chiltepin/workflow.py b/src/chiltepin/workflow.py index f3e36d17..477ee7af 100644 --- a/src/chiltepin/workflow.py +++ b/src/chiltepin/workflow.py @@ -6,6 +6,8 @@ management, eliminating the need for users to directly import or interact with Parsl. """ +import logging +import sys from contextlib import contextmanager from pathlib import Path from typing import Any, Dict, List, Optional, Union @@ -15,6 +17,9 @@ from chiltepin import configure +# Module-level logger for cleanup warnings +_logger = logging.getLogger(__name__) + @contextmanager def workflow( @@ -114,25 +119,36 @@ def workflow( yield finally: - # Cleanup operations - each wrapped in try/except to ensure all are attempted - # even if some fail. Exceptions are chained together. + # Check if we're cleaning up during exception handling + # If so, don't mask the user's exception with cleanup exceptions + user_exception = sys.exc_info()[0] is not None cleanup_exception = None - # Cleanup DataFlowKernel if it was created + # Attempt all cleanup operations, catching exceptions if dfk is not None: try: dfk.cleanup() except Exception as e: - cleanup_exception = e + if user_exception: + _logger.warning( + "Exception during dfk.cleanup() while handling user exception", + exc_info=True, + ) + else: + cleanup_exception = e - # Always call parsl.clear() regardless of dfk state + # Always call parsl.clear() try: parsl.clear() except Exception as e: - if cleanup_exception is None: + if user_exception: + _logger.warning( + "Exception during parsl.clear() while handling user exception", + exc_info=True, + ) + elif cleanup_exception is None: cleanup_exception = e else: - # Chain this exception to the previous one e.__context__ = cleanup_exception cleanup_exception = e @@ -141,15 +157,19 @@ def workflow( try: logger_handler() except Exception as e: - if cleanup_exception is None: + if user_exception: + _logger.warning( + "Exception during logger cleanup while handling user exception", + exc_info=True, + ) + elif cleanup_exception is None: cleanup_exception = e else: - # Chain this exception to the previous one e.__context__ = cleanup_exception cleanup_exception = e - # If any cleanup failed, raise the last exception (with others chained via __context__) - if cleanup_exception is not None: + # Only raise cleanup exceptions if there wasn't a user exception + if cleanup_exception is not None and not user_exception: raise cleanup_exception diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 7bbc40cf..8e8de749 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -273,13 +273,26 @@ def test_dfk_cleanup_exception(self, tmp_path): } } + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: mock_cleanup.side_effect = RuntimeError("Cleanup failed") with pytest.raises(RuntimeError, match="Cleanup failed"): with workflow(config, run_dir=str(tmp_path / "runinfo_exc1")): + dfk_ref = parsl.dfk() # Capture dfk reference pass + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + def test_parsl_clear_exception(self, tmp_path): """Test that exceptions during parsl.clear() are properly raised.""" project_root = pathlib.Path(__file__).parent.parent.resolve() @@ -301,7 +314,18 @@ def test_parsl_clear_exception(self, tmp_path): pass def test_logger_handler_exception(self, tmp_path): - """Test that exceptions during logger_handler() are properly raised.""" + """Test that exceptions during logger_handler() are properly raised. + + Note: Logger cleanup happens both inside dfk.cleanup() and in our explicit + logger_handler() call. When it fails during dfk.cleanup(), that exception + is what gets raised. + """ + # Ensure clean state before test + try: + parsl.clear() + except Exception: + pass + project_root = pathlib.Path(__file__).parent.parent.resolve() config = { @@ -321,7 +345,8 @@ def failing_handler(): return failing_handler with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): - with pytest.raises(RuntimeError, match="Logger cleanup failed"): + # The logger cleanup exception will be raised (either from dfk.cleanup or our call) + with pytest.raises(RuntimeError): with workflow( config, run_dir=str(tmp_path / "runinfo_exc3"), @@ -342,6 +367,10 @@ def test_chained_exceptions_cleanup_then_clear(self, tmp_path): } } + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: with mock.patch("parsl.clear") as mock_clear: mock_cleanup.side_effect = RuntimeError("Cleanup failed") @@ -349,6 +378,7 @@ def test_chained_exceptions_cleanup_then_clear(self, tmp_path): with pytest.raises(RuntimeError) as exc_info: with workflow(config, run_dir=str(tmp_path / "runinfo_exc4")): + dfk_ref = parsl.dfk() # Capture dfk reference pass # The last exception (clear) should be raised @@ -357,6 +387,14 @@ def test_chained_exceptions_cleanup_then_clear(self, tmp_path): assert exc_info.value.__context__ is not None assert "Cleanup failed" in str(exc_info.value.__context__) + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + def test_chained_exceptions_all_three(self, tmp_path): """Test exception chaining when all three cleanup operations fail.""" project_root = pathlib.Path(__file__).parent.parent.resolve() @@ -370,6 +408,10 @@ def test_chained_exceptions_all_three(self, tmp_path): } } + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + # Mock the logger handler to raise an exception def mock_set_file_logger(*args, **kwargs): def failing_handler(): @@ -391,6 +433,7 @@ def failing_handler(): run_dir=str(tmp_path / "runinfo_exc5"), log_file=str(tmp_path / "test.log"), ): + dfk_ref = parsl.dfk() # Capture dfk reference pass # The last exception (logger) should be raised @@ -403,6 +446,14 @@ def failing_handler(): exc_info.value.__context__.__context__ ) + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + def test_parsl_clear_called_when_dfk_is_none(self, tmp_path): """Test that parsl.clear() is called even when dfk is None.""" project_root = pathlib.Path(__file__).parent.parent.resolve() @@ -442,18 +493,26 @@ def test_parsl_clear_exception_without_cleanup_exception(self, tmp_path): } # This tests the branch where parsl.clear() fails but dfk.cleanup() succeeded - # We need to patch the real parsl.clear in the workflow module's finally block + # Need to account for cleanup_parsl fixture calling clear before test runs original_clear = parsl.clear call_count = [0] - def mock_clear_that_fails_once(): + def mock_clear_selective(): call_count[0] += 1 - # Fail on first call (from workflow cleanup), succeed on second (from test cleanup) - if call_count[0] == 1: + # The cleanup_parsl fixture may have already called clear before this test + # We want to fail on the workflow's clear call + # After seeing how many times it's been called, fail on the next-to-last call + # For safety, let's fail on call 2 (which should be from workflow cleanup) + # and succeed on all others + if call_count[0] == 2: raise RuntimeError("Clear failed") - original_clear() + else: + try: + original_clear() + except Exception: + pass # Ignore errors from fixture cleanup - with mock.patch("parsl.clear", side_effect=mock_clear_that_fails_once): + with mock.patch("parsl.clear", side_effect=mock_clear_selective): with pytest.raises(RuntimeError, match="Clear failed"): with workflow(config, run_dir=str(tmp_path / "runinfo_exc7")): pass @@ -500,3 +559,215 @@ def simple_task(): future = simple_task(executor=["service"]) result = future.result() assert result == "success" + + +class TestUserExceptionPrecedence: + """Test that user exceptions are not masked by cleanup exceptions.""" + + def test_user_exception_not_masked_by_cleanup_exception(self, tmp_path, caplog): + """Test that user exceptions take precedence over cleanup exceptions.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + + # Mock dfk.cleanup to fail + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + + # User exception should be raised, not cleanup exception + with pytest.raises(ValueError, match="User error"): + with workflow(config, run_dir=str(tmp_path / "runinfo_user1")): + dfk_ref = parsl.dfk() # Capture dfk reference + raise ValueError("User error") + + # Cleanup exception should be logged as a warning + assert any( + "Exception during dfk.cleanup() while handling user exception" + in record.message + for record in caplog.records + ) + + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + def test_all_cleanup_operations_attempted(self, tmp_path, caplog): + """Test that all cleanup operations are attempted even when some fail.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + + # Mock dfk.cleanup and parsl.clear to fail + original_clear = parsl.clear + call_count = [0] + + def mock_clear_selective(): + call_count[0] += 1 + # Fail on first call (our explicit call after dfk.cleanup fails) + if call_count[0] == 1: + raise RuntimeError("Clear failed") + else: + # Subsequent calls use original + original_clear() + + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + with mock.patch("parsl.clear", side_effect=mock_clear_selective): + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + + # User exception should be raised, not any cleanup exception + with pytest.raises(ValueError, match="User error"): + with workflow(config, run_dir=str(tmp_path / "runinfo_user2")): + dfk_ref = parsl.dfk() # Capture dfk reference + raise ValueError("User error") + + # Both cleanup exceptions should be logged as warnings + assert any( + "Exception during dfk.cleanup() while handling user exception" + in record.message + for record in caplog.records + ) + assert any( + "Exception during parsl.clear() while handling user exception" + in record.message + for record in caplog.records + ) + + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + def test_cleanup_exception_raised_when_no_user_exception(self, tmp_path): + """Test that cleanup exceptions are raised when there's no user exception.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Store reference to real cleanup for later + real_cleanup = parsl.DataFlowKernel.cleanup + dfk_ref = None + + # Mock dfk.cleanup to fail + with mock.patch("parsl.DataFlowKernel.cleanup") as mock_cleanup: + mock_cleanup.side_effect = RuntimeError("Cleanup failed") + + # Cleanup exception should be raised when there's no user exception + with pytest.raises(RuntimeError, match="Cleanup failed"): + with workflow(config, run_dir=str(tmp_path / "runinfo_user3")): + dfk_ref = parsl.dfk() # Capture dfk reference + pass # No user exception + + # Actually clean up the DFK now that we're done testing + if dfk_ref: + try: + real_cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + + def test_user_exception_with_logger_cleanup_failure(self, tmp_path, caplog): + """Test that user exceptions take precedence when logger cleanup fails.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Mock the logger handler to raise an exception + def mock_set_file_logger(*args, **kwargs): + def failing_handler(): + raise RuntimeError("Logger cleanup failed") + + return failing_handler + + with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): + # User exception should be raised, not logger cleanup exception + with pytest.raises(ValueError, match="User error"): + with workflow( + config, + run_dir=str(tmp_path / "runinfo_user4"), + log_file=str(tmp_path / "test.log"), + ): + raise ValueError("User error") + + # Logger cleanup exception should be logged as a warning + assert any( + "Exception during logger cleanup while handling user exception" + in record.message + for record in caplog.records + ) + + def test_logger_handler_exception_standalone(self, tmp_path): + """Test logger cleanup failure when dfk.cleanup and parsl.clear succeed.""" + project_root = pathlib.Path(__file__).parent.parent.resolve() + + config = { + "test-exec": { + "provider": "localhost", + "cores_per_node": 1, + "max_workers_per_node": 1, + "environment": [f"export PYTHONPATH=${{PYTHONPATH}}:{project_root}"], + } + } + + # Create a handler that succeeds first time (dfk cleanup) but fails second time (our explicit call) + call_count = [0] + + def mock_set_file_logger(*args, **kwargs): + def conditional_failing_handler(): + call_count[0] += 1 + if call_count[0] == 2: # Fail on second call (our explicit cleanup) + raise RuntimeError("Logger cleanup failed") + + return conditional_failing_handler + + with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): + # Logger cleanup exception should be raised + with pytest.raises(RuntimeError, match="Logger cleanup failed"): + with workflow( + config, + run_dir=str(tmp_path / "runinfo_logger_standalone"), + log_file=str(tmp_path / "test.log"), + ): + pass From 8bb6da0ebd3e5f451e2b6f8cdea9b789ae60b421 Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Sun, 8 Mar 2026 20:27:09 +0000 Subject: [PATCH 08/15] Attempt to fix workflow context testing issues --- tests/test_tasks.py | 4 +--- tests/test_workflow.py | 42 +++++++++++++++++++++++++++++++----------- 2 files changed, 32 insertions(+), 14 deletions(-) diff --git a/tests/test_tasks.py b/tests/test_tasks.py index 4249daa0..0ecd279a 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -589,9 +589,7 @@ def task2(): return 2 f1 = task1(executor=["test-local"]) - f2 = task2( - executor="all" - ) # Using "all" to test that it works as an alias for all executors + f2 = task2(executor="all") # "all" is an alias for all executors assert f1.result() == 1 assert f2.result() == 2 diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 8e8de749..f14cf4bb 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -492,31 +492,51 @@ def test_parsl_clear_exception_without_cleanup_exception(self, tmp_path): } } - # This tests the branch where parsl.clear() fails but dfk.cleanup() succeeded - # Need to account for cleanup_parsl fixture calling clear before test runs + # This tests the branch where parsl.clear() fails but dfk.cleanup() succeeds + # Explicitly ensure dfk.cleanup() succeeds so cleanup_exception stays None original_clear = parsl.clear + dfk_ref = None + + def mock_cleanup_success(self): + """Mock dfk.cleanup() to succeed without doing anything.""" + # Don't actually clean up - we'll do it manually after + call_count = [0] def mock_clear_selective(): call_count[0] += 1 - # The cleanup_parsl fixture may have already called clear before this test - # We want to fail on the workflow's clear call - # After seeing how many times it's been called, fail on the next-to-last call - # For safety, let's fail on call 2 (which should be from workflow cleanup) - # and succeed on all others if call_count[0] == 2: + # Second call (from workflow) - make it fail raise RuntimeError("Clear failed") else: + # First and subsequent calls should succeed try: original_clear() except Exception: - pass # Ignore errors from fixture cleanup + pass - with mock.patch("parsl.clear", side_effect=mock_clear_selective): - with pytest.raises(RuntimeError, match="Clear failed"): - with workflow(config, run_dir=str(tmp_path / "runinfo_exc7")): + with mock.patch.object(parsl.DataFlowKernel, "cleanup", mock_cleanup_success): + with mock.patch("parsl.clear", side_effect=mock_clear_selective): + # Explicitly make the first call to parsl.clear() to "use up" count==1 + try: + parsl.clear() + except Exception: pass + # Now the workflow's call will be count==2 and will raise + with pytest.raises(RuntimeError, match="Clear failed"): + with workflow(config, run_dir=str(tmp_path / "runinfo_exc7")): + dfk_ref = parsl.dfk() # Capture for manual cleanup + pass + + # Manually clean up since we mocked cleanup + if dfk_ref: + try: + parsl.DataFlowKernel.cleanup(dfk_ref) + parsl.clear() + except Exception: + pass + @pytest.fixture def config_file_fixture(tmp_path): From 1db807de6c32b900fd9bec582eb4571d04d088ed Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Mon, 9 Mar 2026 16:06:21 +0000 Subject: [PATCH 09/15] Rename workflow context manager and enhance its exception handling --- docs/api.rst | 7 ++++ docs/configuration.rst | 8 ++-- docs/data.rst | 4 +- docs/quickstart.rst | 16 ++++---- src/chiltepin/__init__.py | 12 +++--- src/chiltepin/workflow.py | 53 ++++++++++++++------------ tests/test_data.py | 4 +- tests/test_globus_compute_hello.py | 4 +- tests/test_globus_compute_mpi.py | 4 +- tests/test_parsl_hello.py | 4 +- tests/test_parsl_mpi.py | 4 +- tests/test_tasks.py | 4 +- tests/test_workflow.py | 60 +++++++++++++++--------------- 13 files changed, 98 insertions(+), 86 deletions(-) diff --git a/docs/api.rst b/docs/api.rst index 319e403d..0830f700 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -11,6 +11,13 @@ Configuration Module :members: :show-inheritance: +Workflow Module +--------------- + +.. automodule:: chiltepin.workflow + :members: + :show-inheritance: + Endpoint Management Module --------------------------- diff --git a/docs/configuration.rst b/docs/configuration.rst index 9b4ef76e..dc220c8e 100644 --- a/docs/configuration.rst +++ b/docs/configuration.rst @@ -419,10 +419,10 @@ Parse and Load .. code-block:: python - from chiltepin import workflow + from chiltepin import run_workflow # Load configuration from YAML file and run workflow - with workflow( + with run_workflow( "my_config.yaml", include=["compute", "mpi"], # Only load specific resources run_dir="./runinfo" # Directory for Parsl runtime files @@ -434,7 +434,7 @@ Parse and Load .. code-block:: python - from chiltepin import workflow + from chiltepin import run_workflow # Define configuration as a dictionary config_dict = { @@ -457,7 +457,7 @@ Parse and Load } # Load configuration from dict and run workflow - with workflow( + with run_workflow( config_dict, include=["compute", "mpi"], # Only load specific resources run_dir="./runinfo" diff --git a/docs/data.rst b/docs/data.rst index c8d88034..5e6e44ca 100644 --- a/docs/data.rst +++ b/docs/data.rst @@ -238,7 +238,7 @@ A common pattern is to stage data, process it, then clean up: .. code-block:: python - from chiltepin import workflow + from chiltepin import run_workflow from chiltepin.tasks import python_task from chiltepin.data import transfer_task, delete_task @@ -251,7 +251,7 @@ A common pattern is to stage data, process it, then clean up: return result # Load configuration and start workflow - with workflow("config.yaml"): + with run_workflow("config.yaml"): # Stage data to compute resource stage_in = transfer_task( src_ep="my-laptop", diff --git a/docs/quickstart.rst b/docs/quickstart.rst index 022a8ddc..80fe39cc 100644 --- a/docs/quickstart.rst +++ b/docs/quickstart.rst @@ -116,7 +116,7 @@ Create ``my_workflow.py``: .. code-block:: python - from chiltepin import workflow + from chiltepin import run_workflow from chiltepin.tasks import bash_task, python_task # Define tasks @@ -137,7 +137,7 @@ Create ``my_workflow.py``: if __name__ == "__main__": # Load configuration and run workflow - with workflow("my_config.yaml", include=["laptop", "remote"], run_dir="./runinfo"): + with run_workflow("my_config.yaml", include=["laptop", "remote"], run_dir="./runinfo"): # Run local task on "laptop" resource local_future = hello_local(executor=["laptop"]) @@ -205,7 +205,7 @@ Simple Workflow (``simple_workflow.py``) .. code-block:: python - from chiltepin import workflow + from chiltepin import run_workflow from chiltepin.tasks import bash_task, python_task # Define tasks @@ -219,7 +219,7 @@ Simple Workflow (``simple_workflow.py``) if __name__ == "__main__": # Load configuration and run workflow - with workflow("local_config.yaml", run_dir="./runinfo"): + with run_workflow("local_config.yaml", run_dir="./runinfo"): result = multiply(6, 7, executor=["laptop"]).result() print(f"6 * 7 = {result}") @@ -262,7 +262,7 @@ MPI Workflow .. code-block:: python - from chiltepin import workflow + from chiltepin import run_workflow from chiltepin.tasks import bash_task @bash_task @@ -274,7 +274,7 @@ MPI Workflow return f"srun -n {ranks} ./mpi_app" if __name__ == "__main__": - with workflow("mpi_config.yaml", run_dir="./runinfo"): + with run_workflow("mpi_config.yaml", run_dir="./runinfo"): # Compile MPI application on the MPI resource (returns exit code) compile_result = compile_mpi(executor=["mpi-resource-name"]).result() print(f"Compilation exit code: {compile_result}") @@ -338,7 +338,7 @@ The ``include`` parameter selects specific resources to load from the configurat .. code-block:: python # Load only specific resources from YAML file - with workflow( + with run_workflow( "my_config.yaml", include=["laptop", "compute"], # Only these resources run_dir="./runinfo" @@ -364,7 +364,7 @@ The ``include`` parameter selects specific resources to load from the configurat } # Load only specific resources from dict - with workflow( + with run_workflow( config, include=["laptop", "compute"], # Only these resources run_dir="./runinfo" diff --git a/src/chiltepin/__init__.py b/src/chiltepin/__init__.py index 4530aacc..c725beb8 100644 --- a/src/chiltepin/__init__.py +++ b/src/chiltepin/__init__.py @@ -7,9 +7,9 @@ """ __all__ = [ - "workflow", - "workflow_from_file", - "workflow_from_dict", + "run_workflow", + "run_workflow_from_file", + "run_workflow_from_dict", ] @@ -22,9 +22,9 @@ def __getattr__(name): """ if name in __all__: from chiltepin.workflow import ( # noqa: F401 - workflow, - workflow_from_dict, - workflow_from_file, + run_workflow, + run_workflow_from_dict, + run_workflow_from_file, ) globals()[name] = locals()[name] diff --git a/src/chiltepin/workflow.py b/src/chiltepin/workflow.py index 477ee7af..ec42defb 100644 --- a/src/chiltepin/workflow.py +++ b/src/chiltepin/workflow.py @@ -22,7 +22,7 @@ @contextmanager -def workflow( +def run_workflow( config: Union[str, Path, Dict[str, Any]], *, include: Optional[List[str]] = None, @@ -62,28 +62,28 @@ def workflow( -------- From a configuration file: - >>> from chiltepin import workflow + >>> from chiltepin import run_workflow >>> from chiltepin.tasks import python_task >>> >>> @python_task >>> def my_task(): ... return "Hello from workflow!" >>> - >>> with workflow("config.yaml"): + >>> with run_workflow("config.yaml"): ... result = my_task(executor=["compute"]) ... print(result.result()) From a configuration dictionary: >>> config = {"laptop": {"provider": "localhost", "cores_per_node": 4}} - >>> with workflow(config): + >>> with run_workflow(config): ... result = my_task(executor=["laptop"]) ... print(result.result()) With logging and selective resources: >>> import logging - >>> with workflow("config.yaml", include=["compute"], + >>> with run_workflow("config.yaml", include=["compute"], ... log_file="workflow.log", log_level=logging.DEBUG): ... # only "compute" resource is available ... result = my_task(executor=["compute"]) @@ -123,6 +123,7 @@ def workflow( # If so, don't mask the user's exception with cleanup exceptions user_exception = sys.exc_info()[0] is not None cleanup_exception = None + cleanup_tb = None # Preserve original traceback # Attempt all cleanup operations, catching exceptions if dfk is not None: @@ -136,6 +137,7 @@ def workflow( ) else: cleanup_exception = e + cleanup_tb = sys.exc_info()[2] # Always call parsl.clear() try: @@ -148,9 +150,11 @@ def workflow( ) elif cleanup_exception is None: cleanup_exception = e + cleanup_tb = sys.exc_info()[2] else: - e.__context__ = cleanup_exception + e.__cause__ = cleanup_exception cleanup_exception = e + cleanup_tb = sys.exc_info()[2] # Cleanup logger handler if logger_handler is not None: @@ -164,24 +168,27 @@ def workflow( ) elif cleanup_exception is None: cleanup_exception = e + cleanup_tb = sys.exc_info()[2] else: - e.__context__ = cleanup_exception + e.__cause__ = cleanup_exception cleanup_exception = e + cleanup_tb = sys.exc_info()[2] # Only raise cleanup exceptions if there wasn't a user exception + # Preserve the original traceback so the error location is clear if cleanup_exception is not None and not user_exception: - raise cleanup_exception + raise cleanup_exception.with_traceback(cleanup_tb) # Convenience aliases for clarity @contextmanager -def workflow_from_file( +def run_workflow_from_file( config_file: Union[str, Path], **kwargs, ): """Context manager for workflows using a YAML configuration file. - This is an alias for workflow() that makes it explicit that a file + This is an alias for run_workflow() that makes it explicit that a file path is expected. Parameters @@ -189,33 +196,33 @@ def workflow_from_file( config_file : str or Path Path to YAML configuration file **kwargs - Additional arguments passed to workflow() + Additional arguments passed to run_workflow() Examples -------- - >>> from chiltepin import workflow_from_file + >>> from chiltepin import run_workflow_from_file >>> from chiltepin.tasks import python_task >>> >>> @python_task >>> def my_task(): ... return 42 >>> - >>> with workflow_from_file("my_config.yaml", include=["compute"]): + >>> with run_workflow_from_file("my_config.yaml", include=["compute"]): ... result = my_task(executor=["compute"]) ... print(result.result()) """ - with workflow(config_file, **kwargs): + with run_workflow(config_file, **kwargs): yield @contextmanager -def workflow_from_dict( +def run_workflow_from_dict( config_dict: Dict[str, Any], **kwargs, ): """Context manager for workflows using a configuration dictionary. - This is an alias for workflow() that makes it explicit that a dict + This is an alias for run_workflow() that makes it explicit that a dict is expected. Parameters @@ -223,11 +230,11 @@ def workflow_from_dict( config_dict : dict Configuration dictionary **kwargs - Additional arguments passed to workflow() + Additional arguments passed to run_workflow() Examples -------- - >>> from chiltepin import workflow_from_dict + >>> from chiltepin import run_workflow_from_dict >>> from chiltepin.tasks import python_task >>> >>> @python_task @@ -235,16 +242,16 @@ def workflow_from_dict( ... return 42 >>> >>> config = {"laptop": {"provider": "localhost"}} - >>> with workflow_from_dict(config): + >>> with run_workflow_from_dict(config): ... result = my_task(executor=["laptop"]) ... print(result.result()) """ - with workflow(config_dict, **kwargs): + with run_workflow(config_dict, **kwargs): yield __all__ = [ - "workflow", - "workflow_from_file", - "workflow_from_dict", + "run_workflow", + "run_workflow_from_file", + "run_workflow_from_dict", ] diff --git a/tests/test_data.py b/tests/test_data.py index 15a1301c..9412812b 100644 --- a/tests/test_data.py +++ b/tests/test_data.py @@ -9,7 +9,7 @@ import chiltepin.data as data import chiltepin.endpoint as endpoint -from chiltepin import workflow +from chiltepin import run_workflow # Set up fixture to initialize and cleanup Parsl @@ -33,7 +33,7 @@ def config(config_file): unique_dst = f"1MB.to_ursa.{uuid.uuid4()}" # Use workflow context manager with default (empty) resource configuration - with workflow( + with run_workflow( {}, run_dir=str(output_dir / "test_data_runinfo"), log_file=str(output_dir / "test_data_parsl.log"), diff --git a/tests/test_globus_compute_hello.py b/tests/test_globus_compute_hello.py index c53803ca..1a9b1cf9 100644 --- a/tests/test_globus_compute_hello.py +++ b/tests/test_globus_compute_hello.py @@ -8,7 +8,7 @@ import chiltepin.configure import chiltepin.endpoint as endpoint -from chiltepin import workflow +from chiltepin import run_workflow from chiltepin.tasks import bash_task, python_task @@ -59,7 +59,7 @@ def config(config_file): yaml_config["gc-service"]["endpoint"] = f"{endpoint_id}" # Use workflow context manager for Parsl lifecycle - with workflow( + with run_workflow( yaml_config, include=["gc-service"], client=compute_client, diff --git a/tests/test_globus_compute_mpi.py b/tests/test_globus_compute_mpi.py index 375f9ae1..64247a57 100644 --- a/tests/test_globus_compute_mpi.py +++ b/tests/test_globus_compute_mpi.py @@ -10,7 +10,7 @@ import chiltepin.configure import chiltepin.endpoint as endpoint -from chiltepin import workflow +from chiltepin import run_workflow from chiltepin.tasks import bash_task @@ -66,7 +66,7 @@ def config(config_file): yaml_config["gc-mpi"]["endpoint"] = f"{endpoint_id}" # Use workflow context manager for Parsl lifecycle - with workflow( + with run_workflow( yaml_config, include=["gc-compute", "gc-mpi"], client=compute_client, diff --git a/tests/test_parsl_hello.py b/tests/test_parsl_hello.py index 1187b64c..79d3bf7c 100755 --- a/tests/test_parsl_hello.py +++ b/tests/test_parsl_hello.py @@ -7,7 +7,7 @@ import pytest import chiltepin.configure -from chiltepin import workflow +from chiltepin import run_workflow from chiltepin.tasks import bash_task, python_task @@ -27,7 +27,7 @@ def config(config_file): ) # Use workflow context manager for Parsl lifecycle - with workflow( + with run_workflow( yaml_config, include=["service"], run_dir=str(output_dir / "test_parsl_hello_runinfo"), diff --git a/tests/test_parsl_mpi.py b/tests/test_parsl_mpi.py index 9e2a6ab2..099ec8d5 100644 --- a/tests/test_parsl_mpi.py +++ b/tests/test_parsl_mpi.py @@ -10,7 +10,7 @@ import pytest import chiltepin.configure -from chiltepin import workflow +from chiltepin import run_workflow from chiltepin.tasks import bash_task @@ -34,7 +34,7 @@ def config(config_file): ) # Use workflow context manager for Parsl lifecycle - with workflow( + with run_workflow( yaml_config, include=["compute", "mpi"], run_dir=str(output_dir / "test_parsl_mpi_runinfo"), diff --git a/tests/test_tasks.py b/tests/test_tasks.py index 0ecd279a..2ec39faa 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -14,7 +14,7 @@ import parsl import pytest -from chiltepin import workflow +from chiltepin import run_workflow from chiltepin.tasks import bash_task, join_task, python_task @@ -41,7 +41,7 @@ def parsl_config(): } # Use workflow context manager for Parsl lifecycle - with workflow( + with run_workflow( local_config, run_dir=str(output_dir / "test_tasks_runinfo"), log_file=str(output_dir / "test_tasks_parsl.log"), diff --git a/tests/test_workflow.py b/tests/test_workflow.py index f14cf4bb..bb3b64be 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -15,7 +15,7 @@ import yaml import chiltepin.configure -from chiltepin import workflow, workflow_from_dict, workflow_from_file +from chiltepin import run_workflow, run_workflow_from_dict, run_workflow_from_file from chiltepin.tasks import python_task @@ -58,7 +58,7 @@ def cleanup_parsl(): class TestWorkflowContextManager: - """Test workflow() context manager with different configurations.""" + """Test run_workflow() context manager with different configurations.""" def test_workflow_with_dict_config(self, tmp_path): """Test workflow context manager with a dictionary config.""" @@ -85,7 +85,7 @@ def multiply(x, y): } } - with workflow(config, run_dir=str(tmp_path / "runinfo1")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo1")): # Submit tasks future1 = add_numbers(10, 32, executor=["test-executor"]) future2 = multiply(6, 7, executor=["test-executor"]) @@ -115,7 +115,7 @@ def greet(name): yaml.dump(config_dict, f) # Test with file path argument - with workflow( + with run_workflow( str(temp_config_file), include=["service"], run_dir=str(tmp_path / "runinfo2"), @@ -147,7 +147,7 @@ def add_numbers(a, b): } } - with workflow_from_dict(config, run_dir=str(tmp_path / "runinfo5")): + with run_workflow_from_dict(config, run_dir=str(tmp_path / "runinfo5")): future = add_numbers(100, 200, executor=["my-executor"]) result = future.result() assert result == 300 @@ -169,7 +169,7 @@ def multiply(x, y): yaml.dump(config_dict, f) # Test with file path argument using workflow_from_file - with workflow_from_file( + with run_workflow_from_file( str(temp_config_file), include=["service"], run_dir=str(tmp_path / "runinfo6"), @@ -215,12 +215,12 @@ def multiply(x, y): } # First workflow - with workflow(config1, run_dir=str(tmp_path / "runinfo7")): + with run_workflow(config1, run_dir=str(tmp_path / "runinfo7")): future = add_numbers(1, 2, executor=["exec1"]) assert future.result() == 3 # Second workflow should work without conflicts - with workflow(config2, run_dir=str(tmp_path / "runinfo8")): + with run_workflow(config2, run_dir=str(tmp_path / "runinfo8")): future = multiply(3, 4, executor=["exec2"]) assert future.result() == 12 @@ -245,14 +245,14 @@ def add_numbers(a, b): # Workflow should cleanup even if exception occurs with pytest.raises(ValueError): - with workflow(config, run_dir=str(tmp_path / "runinfo9")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo9")): future = add_numbers(5, 5, executor=["test-exec"]) result = future.result() assert result == 10 raise ValueError("Intentional test error") # Should be able to create another workflow after exception - with workflow(config, run_dir=str(tmp_path / "runinfo10")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo10")): future = add_numbers(7, 8, executor=["test-exec"]) assert future.result() == 15 @@ -281,7 +281,7 @@ def test_dfk_cleanup_exception(self, tmp_path): mock_cleanup.side_effect = RuntimeError("Cleanup failed") with pytest.raises(RuntimeError, match="Cleanup failed"): - with workflow(config, run_dir=str(tmp_path / "runinfo_exc1")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc1")): dfk_ref = parsl.dfk() # Capture dfk reference pass @@ -310,7 +310,7 @@ def test_parsl_clear_exception(self, tmp_path): mock_clear.side_effect = RuntimeError("Clear failed") with pytest.raises(RuntimeError, match="Clear failed"): - with workflow(config, run_dir=str(tmp_path / "runinfo_exc2")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc2")): pass def test_logger_handler_exception(self, tmp_path): @@ -347,7 +347,7 @@ def failing_handler(): with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): # The logger cleanup exception will be raised (either from dfk.cleanup or our call) with pytest.raises(RuntimeError): - with workflow( + with run_workflow( config, run_dir=str(tmp_path / "runinfo_exc3"), log_file=str(tmp_path / "test.log"), @@ -377,15 +377,15 @@ def test_chained_exceptions_cleanup_then_clear(self, tmp_path): mock_clear.side_effect = RuntimeError("Clear failed") with pytest.raises(RuntimeError) as exc_info: - with workflow(config, run_dir=str(tmp_path / "runinfo_exc4")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc4")): dfk_ref = parsl.dfk() # Capture dfk reference pass # The last exception (clear) should be raised assert "Clear failed" in str(exc_info.value) # And the previous exception (cleanup) should be in the chain - assert exc_info.value.__context__ is not None - assert "Cleanup failed" in str(exc_info.value.__context__) + assert exc_info.value.__cause__ is not None + assert "Cleanup failed" in str(exc_info.value.__cause__) # Actually clean up the DFK now that we're done testing if dfk_ref: @@ -428,7 +428,7 @@ def failing_handler(): mock_clear.side_effect = RuntimeError("Clear failed") with pytest.raises(RuntimeError) as exc_info: - with workflow( + with run_workflow( config, run_dir=str(tmp_path / "runinfo_exc5"), log_file=str(tmp_path / "test.log"), @@ -439,12 +439,10 @@ def failing_handler(): # The last exception (logger) should be raised assert "Logger cleanup failed" in str(exc_info.value) # Check the exception chain - assert exc_info.value.__context__ is not None - assert "Clear failed" in str(exc_info.value.__context__) - assert exc_info.value.__context__.__context__ is not None - assert "Cleanup failed" in str( - exc_info.value.__context__.__context__ - ) + assert exc_info.value.__cause__ is not None + assert "Clear failed" in str(exc_info.value.__cause__) + assert exc_info.value.__cause__.__cause__ is not None + assert "Cleanup failed" in str(exc_info.value.__cause__.__cause__) # Actually clean up the DFK now that we're done testing if dfk_ref: @@ -473,7 +471,7 @@ def test_parsl_clear_called_when_dfk_is_none(self, tmp_path): mock_load.side_effect = RuntimeError("Load failed") with pytest.raises(RuntimeError, match="Load failed"): - with workflow(config, run_dir=str(tmp_path / "runinfo_exc6")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc6")): pass # parsl.clear() should still be called even though dfk is None @@ -525,7 +523,7 @@ def mock_clear_selective(): # Now the workflow's call will be count==2 and will raise with pytest.raises(RuntimeError, match="Clear failed"): - with workflow(config, run_dir=str(tmp_path / "runinfo_exc7")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_exc7")): dfk_ref = parsl.dfk() # Capture for manual cleanup pass @@ -571,7 +569,7 @@ def test_workflow_from_file_with_local_fixture(self, config_file_fixture, tmp_pa def simple_task(): return "success" - with workflow_from_file( + with run_workflow_from_file( config_file_fixture, include=["service"], run_dir=str(tmp_path / "runinfo_file"), @@ -607,7 +605,7 @@ def test_user_exception_not_masked_by_cleanup_exception(self, tmp_path, caplog): # User exception should be raised, not cleanup exception with pytest.raises(ValueError, match="User error"): - with workflow(config, run_dir=str(tmp_path / "runinfo_user1")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_user1")): dfk_ref = parsl.dfk() # Capture dfk reference raise ValueError("User error") @@ -662,7 +660,7 @@ def mock_clear_selective(): # User exception should be raised, not any cleanup exception with pytest.raises(ValueError, match="User error"): - with workflow(config, run_dir=str(tmp_path / "runinfo_user2")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_user2")): dfk_ref = parsl.dfk() # Capture dfk reference raise ValueError("User error") @@ -709,7 +707,7 @@ def test_cleanup_exception_raised_when_no_user_exception(self, tmp_path): # Cleanup exception should be raised when there's no user exception with pytest.raises(RuntimeError, match="Cleanup failed"): - with workflow(config, run_dir=str(tmp_path / "runinfo_user3")): + with run_workflow(config, run_dir=str(tmp_path / "runinfo_user3")): dfk_ref = parsl.dfk() # Capture dfk reference pass # No user exception @@ -744,7 +742,7 @@ def failing_handler(): with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): # User exception should be raised, not logger cleanup exception with pytest.raises(ValueError, match="User error"): - with workflow( + with run_workflow( config, run_dir=str(tmp_path / "runinfo_user4"), log_file=str(tmp_path / "test.log"), @@ -785,7 +783,7 @@ def conditional_failing_handler(): with mock.patch("parsl.set_file_logger", side_effect=mock_set_file_logger): # Logger cleanup exception should be raised with pytest.raises(RuntimeError, match="Logger cleanup failed"): - with workflow( + with run_workflow( config, run_dir=str(tmp_path / "runinfo_logger_standalone"), log_file=str(tmp_path / "test.log"), From 4a19094770ecf48b64c2e0615e020c4ac12d8368 Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Mon, 9 Mar 2026 16:46:55 +0000 Subject: [PATCH 10/15] make workflow parsl cleanup fixture more robust --- tests/test_workflow.py | 61 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 53 insertions(+), 8 deletions(-) diff --git a/tests/test_workflow.py b/tests/test_workflow.py index bb3b64be..fbac9654 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -8,6 +8,7 @@ """ import pathlib +import warnings from unittest import mock import parsl @@ -40,10 +41,31 @@ def cleanup_parsl(): try: dfk = parsl.dfk() if dfk: - dfk.cleanup() - parsl.clear() - except Exception: - pass # No DFK loaded, which is fine + try: + dfk.cleanup() + except Exception as e: + warnings.warn( + f"DFK cleanup failed in pre-test fixture cleanup: {e}", + RuntimeWarning, + stacklevel=2, + ) + try: + parsl.clear() + except Exception as e: + warnings.warn( + f"parsl.clear() failed in pre-test fixture cleanup: {e}", + RuntimeWarning, + stacklevel=2, + ) + except Exception as e: + # Expected: "Must first load config" when no DFK loaded + # Unexpected: Other errors + if "Must first load config" not in str(e) and "No DataFlowKernel" not in str(e): + warnings.warn( + f"Unexpected error in pre-test cleanup: {e}", + RuntimeWarning, + stacklevel=2, + ) yield @@ -51,10 +73,33 @@ def cleanup_parsl(): try: dfk = parsl.dfk() if dfk: - dfk.cleanup() - parsl.clear() - except Exception: - pass + try: + dfk.cleanup() + except Exception as e: + # Don't warn about double-cleanup attempts (expected in some tests) + if "already been cleaned-up" not in str(e): + warnings.warn( + f"DFK cleanup failed in post-test fixture cleanup: {e}", + RuntimeWarning, + stacklevel=2, + ) + try: + parsl.clear() + except Exception as e: + warnings.warn( + f"parsl.clear() failed in post-test fixture cleanup: {e}", + RuntimeWarning, + stacklevel=2, + ) + except Exception as e: + # Expected: "Must first load config" when no DFK loaded + # Unexpected: Other errors + if "Must first load config" not in str(e) and "No DataFlowKernel" not in str(e): + warnings.warn( + f"Unexpected error in post-test cleanup: {e}", + RuntimeWarning, + stacklevel=2, + ) class TestWorkflowContextManager: From 4247ac1b078a34b34754545f0b724fd5900ad69d Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Mon, 9 Mar 2026 17:35:38 +0000 Subject: [PATCH 11/15] Fix workflow tests to use more reliable log message retrieval --- tests/test_workflow.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_workflow.py b/tests/test_workflow.py index fbac9654..94a3597d 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -657,7 +657,7 @@ def test_user_exception_not_masked_by_cleanup_exception(self, tmp_path, caplog): # Cleanup exception should be logged as a warning assert any( "Exception during dfk.cleanup() while handling user exception" - in record.message + in record.getMessage() for record in caplog.records ) @@ -712,12 +712,12 @@ def mock_clear_selective(): # Both cleanup exceptions should be logged as warnings assert any( "Exception during dfk.cleanup() while handling user exception" - in record.message + in record.getMessage() for record in caplog.records ) assert any( "Exception during parsl.clear() while handling user exception" - in record.message + in record.getMessage() for record in caplog.records ) @@ -797,7 +797,7 @@ def failing_handler(): # Logger cleanup exception should be logged as a warning assert any( "Exception during logger cleanup while handling user exception" - in record.message + in record.getMessage() for record in caplog.records ) From 6dec7c0ff7a5f3efe24c52a0fe22aff1de7ee82e Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Mon, 9 Mar 2026 18:03:32 +0000 Subject: [PATCH 12/15] Improve workflow context manager testing and fix empty configuration bug --- src/chiltepin/configure.py | 3 +- tests/test_configure.py | 26 +++++++++++++ tests/test_workflow.py | 80 ++++++++++++++++---------------------- 3 files changed, 61 insertions(+), 48 deletions(-) diff --git a/src/chiltepin/configure.py b/src/chiltepin/configure.py index 8fcc09d3..ba95e37d 100644 --- a/src/chiltepin/configure.py +++ b/src/chiltepin/configure.py @@ -41,7 +41,8 @@ def parse_file(filename: str) -> Dict[str, Any]: except yaml.YAMLError as e: print("Invalid yaml configuration") raise (e) - return yaml_config + # yaml.safe_load returns None for empty files; return empty dict instead + return yaml_config if yaml_config is not None else {} def create_provider(config: Dict[str, Any]) -> ExecutionProvider: diff --git a/tests/test_configure.py b/tests/test_configure.py index 4c5bd668..8e413213 100644 --- a/tests/test_configure.py +++ b/tests/test_configure.py @@ -64,6 +64,32 @@ def test_parse_invalid_yaml(self): finally: pathlib.Path(tmp_path).unlink() + def test_parse_empty_yaml(self): + """Test parsing an empty YAML file returns an empty dict.""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: + # Write nothing (empty file) + tmp_path = tmp.name + + try: + result = configure.parse_file(tmp_path) + assert result == {} + assert isinstance(result, dict) + finally: + pathlib.Path(tmp_path).unlink() + + def test_parse_yaml_comments_only(self): + """Test parsing a YAML file with only comments returns an empty dict.""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: + tmp.write("# This is just a comment\n# Another comment\n") + tmp_path = tmp.name + + try: + result = configure.parse_file(tmp_path) + assert result == {} + assert isinstance(result, dict) + finally: + pathlib.Path(tmp_path).unlink() + class TestCreateProvider: """Test create_provider() function for all provider types.""" diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 94a3597d..b6e8577a 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -12,6 +12,7 @@ from unittest import mock import parsl +import parsl.errors import pytest import yaml @@ -33,73 +34,58 @@ def add_pythonpath_to_config(config_dict, resource_name): return config_dict -# Cleanup any existing Parsl state before tests -@pytest.fixture(scope="function", autouse=True) -def cleanup_parsl(): - """Ensure Parsl is cleaned up before and after each test.""" - # Cleanup before test +def _cleanup_parsl_state(phase, ignore_already_cleaned=False): + """Helper function to cleanup Parsl DFK state. + + Args: + phase: Description of when cleanup is happening (e.g., "pre-test", "post-test") + ignore_already_cleaned: If True, suppress warnings about already cleaned DFK + """ try: dfk = parsl.dfk() if dfk: try: dfk.cleanup() except Exception as e: + # Don't warn about double-cleanup attempts if requested + if ignore_already_cleaned and "already been cleaned-up" in str(e): + return warnings.warn( - f"DFK cleanup failed in pre-test fixture cleanup: {e}", + f"DFK cleanup failed in {phase} fixture cleanup: {e}", RuntimeWarning, - stacklevel=2, + stacklevel=3, ) try: parsl.clear() except Exception as e: warnings.warn( - f"parsl.clear() failed in pre-test fixture cleanup: {e}", + f"parsl.clear() failed in {phase} fixture cleanup: {e}", RuntimeWarning, - stacklevel=2, + stacklevel=3, ) + except parsl.errors.NoDataFlowKernelError: + # Expected when no DFK loaded + pass except Exception as e: - # Expected: "Must first load config" when no DFK loaded - # Unexpected: Other errors - if "Must first load config" not in str(e) and "No DataFlowKernel" not in str(e): - warnings.warn( - f"Unexpected error in pre-test cleanup: {e}", - RuntimeWarning, - stacklevel=2, - ) + # Unexpected errors + warnings.warn( + f"Unexpected error in {phase} cleanup: {e}", + RuntimeWarning, + stacklevel=3, + ) + + +# Cleanup any existing Parsl state before tests +@pytest.fixture(scope="function", autouse=True) +def cleanup_parsl(): + """Ensure Parsl is cleaned up before and after each test.""" + # Cleanup before test + _cleanup_parsl_state("pre-test") yield # Cleanup after test - try: - dfk = parsl.dfk() - if dfk: - try: - dfk.cleanup() - except Exception as e: - # Don't warn about double-cleanup attempts (expected in some tests) - if "already been cleaned-up" not in str(e): - warnings.warn( - f"DFK cleanup failed in post-test fixture cleanup: {e}", - RuntimeWarning, - stacklevel=2, - ) - try: - parsl.clear() - except Exception as e: - warnings.warn( - f"parsl.clear() failed in post-test fixture cleanup: {e}", - RuntimeWarning, - stacklevel=2, - ) - except Exception as e: - # Expected: "Must first load config" when no DFK loaded - # Unexpected: Other errors - if "Must first load config" not in str(e) and "No DataFlowKernel" not in str(e): - warnings.warn( - f"Unexpected error in post-test cleanup: {e}", - RuntimeWarning, - stacklevel=2, - ) + _cleanup_parsl_state("post-test", ignore_already_cleaned=True) class TestWorkflowContextManager: From 878d1df07a4524ebaf0ad7f51dd4ac7838978fcf Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Mon, 9 Mar 2026 18:57:01 +0000 Subject: [PATCH 13/15] Document default local resource and allow it to be overridden --- docs/configuration.rst | 51 ++++++++++++++++++++++++ src/chiltepin/configure.py | 39 +++++++++++-------- src/chiltepin/workflow.py | 10 ++++- tests/test_configure.py | 79 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 161 insertions(+), 18 deletions(-) diff --git a/docs/configuration.rst b/docs/configuration.rst index dc220c8e..7d112716 100644 --- a/docs/configuration.rst +++ b/docs/configuration.rst @@ -17,6 +17,45 @@ Resources can be: The configuration file defines named resources, where each resource represents a pool of nodes and/or cores on which tasks can be executed. +Default "local" Resource +^^^^^^^^^^^^^^^^^^^^^^^^ + +Chiltepin automatically provides a default resource named **"local"** that is always +available, even if you don't define any resources in your configuration file. This +default resource: + +- Uses the current machine (localhost provider) +- Has 1 core per worker and 1 maximum worker +- Starts with 0 blocks and can scale up to 1 block +- Is useful for testing and light computational tasks + +You can use this default resource without any configuration: + +.. code-block:: python + + from chiltepin import run_workflow + from chiltepin.tasks import python_task + + @python_task + def my_task(): + return "Hello from local!" + + # Works even with an empty config + with run_workflow({}): + result = my_task(executor=["local"]).result() + +You can override the default "local" resource by defining your own resource with +that name in your configuration file. This is useful if you need different settings +for local execution: + +.. code-block:: yaml + + local: + provider: "localhost" + max_blocks: 2 + max_workers_per_node: 4 + cores_per_worker: 2 + Basic Structure --------------- @@ -468,6 +507,18 @@ Parse and Load The ``include`` parameter lets you selectively load only specific resources from your configuration. If omitted, all resources are loaded. +.. note:: + The default **"local"** resource is always available, regardless of the ``include`` + parameter. You do not need to add "local" to the include list to use it. This ensures + you always have a fallback resource available for tasks. + + .. code-block:: python + + # "local" is available even though not in include list + with run_workflow("config.yaml", include=["compute"]): + local_result = my_task(executor=["local"]).result() # Works! + compute_result = my_task(executor=["compute"]).result() # Works! + Configuration Best Practices ----------------------------- diff --git a/src/chiltepin/configure.py b/src/chiltepin/configure.py index ba95e37d..0f3fe683 100644 --- a/src/chiltepin/configure.py +++ b/src/chiltepin/configure.py @@ -382,23 +382,7 @@ def load( # Get project root directory for setting PYTHONPATH project_base = Path(__file__).parent.parent.parent.resolve() - # Define a default HTEX Executor with a local provider - # This includes adding project root directory to PYTHONPATH - executors = [ - HighThroughputExecutor( - label="local", - worker_debug=True, - cores_per_worker=1, - max_workers_per_node=1, - provider=LocalProvider( - init_blocks=0, - max_blocks=1, - worker_init=f"export PYTHONPATH=${{PYTHONPATH}}:{project_base}", - ), - ) - ] - - # Add an Executor for each resource + # Determine which resources to load if include is None: resources = config else: @@ -409,6 +393,27 @@ def load( f"Resources specified in include list not found in config: {missing}" ) resources = {key: config[key] for key in include} + + # Create executors list, starting with default "local" if not overridden + executors = [] + + # Only add default "local" executor if user hasn't defined their own + if "local" not in resources: + executors.append( + HighThroughputExecutor( + label="local", + worker_debug=True, + cores_per_worker=1, + max_workers_per_node=1, + provider=LocalProvider( + init_blocks=0, + max_blocks=1, + worker_init=f"export PYTHONPATH=${{PYTHONPATH}}:{project_base}", + ), + ) + ) + + # Add an Executor for each resource for resource_name, resource_config in resources.items(): executors.append( create_executor( diff --git a/src/chiltepin/workflow.py b/src/chiltepin/workflow.py index ec42defb..bdc9c8e4 100644 --- a/src/chiltepin/workflow.py +++ b/src/chiltepin/workflow.py @@ -42,6 +42,8 @@ def run_workflow( Either a path to a YAML configuration file or a configuration dictionary include : list of str, optional List of resource labels to load. If None, all resources are loaded. + Note: The default "local" resource is always available, regardless of + this parameter. run_dir : str, optional Directory for Parsl runtime files. If None, uses Parsl's default. client : globus_compute_sdk.Client, optional @@ -85,8 +87,10 @@ def run_workflow( >>> import logging >>> with run_workflow("config.yaml", include=["compute"], ... log_file="workflow.log", log_level=logging.DEBUG): - ... # only "compute" resource is available + ... # "compute" resource from config is available ... result = my_task(executor=["compute"]) + ... # "local" resource is also always available + ... local_result = my_task(executor=["local"]) """ # Parse config if it's a file path if isinstance(config, (str, Path)): @@ -209,6 +213,8 @@ def run_workflow_from_file( >>> >>> with run_workflow_from_file("my_config.yaml", include=["compute"]): ... result = my_task(executor=["compute"]) + ... # "local" resource is also always available + ... local_result = my_task(executor=["local"]) ... print(result.result()) """ with run_workflow(config_file, **kwargs): @@ -244,6 +250,8 @@ def run_workflow_from_dict( >>> config = {"laptop": {"provider": "localhost"}} >>> with run_workflow_from_dict(config): ... result = my_task(executor=["laptop"]) + ... # "local" resource is also always available + ... local_result = my_task(executor=["local"]) ... print(result.result()) """ with run_workflow(config_dict, **kwargs): diff --git a/tests/test_configure.py b/tests/test_configure.py index 8e413213..72ec97ce 100644 --- a/tests/test_configure.py +++ b/tests/test_configure.py @@ -638,6 +638,85 @@ def test_load_with_client(self): ] assert len(gc_executors) == 1 + def test_load_overrides_default_local(self): + """Test that user-defined 'local' resource overrides the default.""" + resources = { + "local": { + "provider": "localhost", + "cores_per_worker": 4, + "max_workers_per_node": 8, + } + } + config = configure.load(resources) + + # Should have only 1 executor (user's local, not default + user's) + assert len(config.executors) == 1 + assert config.executors[0].label == "local" + # Verify it's the user's config (4 cores per worker, not default 1) + assert config.executors[0].cores_per_worker == 4 + assert config.executors[0].max_workers_per_node == 8 + + def test_load_local_in_include_overrides_default(self): + """Test that including 'local' in config overrides default even with include filter.""" + resources = { + "local": { + "provider": "localhost", + "cores_per_worker": 2, + }, + "compute": { + "provider": "slurm", + "partition": "compute", + }, + } + config = configure.load(resources, include=["local"]) + + # Should have only user's local (not default) + assert len(config.executors) == 1 + assert config.executors[0].label == "local" + assert config.executors[0].cores_per_worker == 2 + + def test_load_include_without_local_uses_default(self): + """Test that default local is available when not in include list.""" + resources = { + "local": { + "provider": "localhost", + "cores_per_worker": 4, + }, + "compute": { + "provider": "slurm", + "partition": "compute", + }, + } + # Include only compute, not local + config = configure.load(resources, include=["compute"]) + + # Should have default local + compute (user's local not loaded) + assert len(config.executors) == 2 + labels = [ex.label for ex in config.executors] + assert "local" in labels + assert "compute" in labels + + # Verify it's the default local (1 core per worker, not user's 4) + local_ex = [ex for ex in config.executors if ex.label == "local"][0] + assert local_ex.cores_per_worker == 1 + + def test_load_duplicate_in_include_is_harmless(self): + """Test that duplicate resource names in include list are harmless.""" + resources = { + "compute": {"provider": "slurm", "partition": "compute"}, + "mpi": {"provider": "slurm", "mpi": True}, + } + + # Duplicates in include list should just be ignored + config = configure.load(resources, include=["compute", "mpi", "compute"]) + + # Should have local + compute + mpi (compute only loaded once despite being in list twice) + assert len(config.executors) == 3 + labels = [ex.label for ex in config.executors] + assert "local" in labels + assert "compute" in labels + assert "mpi" in labels + class TestIntegrationScenarios: """Integration tests for realistic configuration scenarios.""" From 956d430b3cf14e24f48d07bdb326e0376ea72582 Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Mon, 9 Mar 2026 20:29:03 +0000 Subject: [PATCH 14/15] Fix local config include bug --- src/chiltepin/configure.py | 40 +++++++++++++++++++----------- tests/test_configure.py | 50 +++++++++++++++++++++++++++++++++++--- 2 files changed, 73 insertions(+), 17 deletions(-) diff --git a/src/chiltepin/configure.py b/src/chiltepin/configure.py index 0f3fe683..d7d98be2 100644 --- a/src/chiltepin/configure.py +++ b/src/chiltepin/configure.py @@ -386,19 +386,30 @@ def load( if include is None: resources = config else: - # Validate that all requested resources exist - missing = [key for key in include if key not in config] + # Validate that all requested resources exist (except "local" which is always available) + missing = [key for key in include if key != "local" and key not in config] if missing: raise RuntimeError( f"Resources specified in include list not found in config: {missing}" ) - resources = {key: config[key] for key in include} + resources = {key: config[key] for key in include if key in config} - # Create executors list, starting with default "local" if not overridden + # Create executors list executors = [] - # Only add default "local" executor if user hasn't defined their own - if "local" not in resources: + # Add "local" executor - use user's definition if provided, otherwise use default + # This happens regardless of the include filter + if "local" in config: + # User defined their own "local", use it as the default + executors.append( + create_executor( + "local", + config["local"], + client, + ), + ) + else: + # Use built-in default "local" executors.append( HighThroughputExecutor( label="local", @@ -413,15 +424,16 @@ def load( ) ) - # Add an Executor for each resource + # Add an Executor for each resource (skip "local" since we already added it) for resource_name, resource_config in resources.items(): - executors.append( - create_executor( - resource_name, - resource_config, - client, - ), - ) + if resource_name != "local": + executors.append( + create_executor( + resource_name, + resource_config, + client, + ), + ) config_kwargs = {"executors": executors} if run_dir is not None: diff --git a/tests/test_configure.py b/tests/test_configure.py index 72ec97ce..2a190bde 100644 --- a/tests/test_configure.py +++ b/tests/test_configure.py @@ -676,7 +676,7 @@ def test_load_local_in_include_overrides_default(self): assert config.executors[0].cores_per_worker == 2 def test_load_include_without_local_uses_default(self): - """Test that default local is available when not in include list.""" + """Test that user's local is used even when not in include list.""" resources = { "local": { "provider": "localhost", @@ -690,13 +690,38 @@ def test_load_include_without_local_uses_default(self): # Include only compute, not local config = configure.load(resources, include=["compute"]) - # Should have default local + compute (user's local not loaded) + # Should have user's local + compute (user's local always used when defined) assert len(config.executors) == 2 labels = [ex.label for ex in config.executors] assert "local" in labels assert "compute" in labels - # Verify it's the default local (1 core per worker, not user's 4) + # Verify it's the user's local (4 cores per worker, not default 1) + local_ex = [ex for ex in config.executors if ex.label == "local"][0] + assert local_ex.cores_per_worker == 4 + + def test_load_include_no_user_local_gets_default(self): + """Test that default local is used when user doesn't define one.""" + resources = { + "compute": { + "provider": "slurm", + "partition": "compute", + }, + "mpi": { + "provider": "slurm", + "mpi": True, + }, + } + # Include only compute (no local defined by user) + config = configure.load(resources, include=["compute"]) + + # Should have default local + compute + assert len(config.executors) == 2 + labels = [ex.label for ex in config.executors] + assert "local" in labels + assert "compute" in labels + + # Verify it's the default local (1 core per worker) local_ex = [ex for ex in config.executors if ex.label == "local"][0] assert local_ex.cores_per_worker == 1 @@ -717,6 +742,25 @@ def test_load_duplicate_in_include_is_harmless(self): assert "compute" in labels assert "mpi" in labels + def test_load_include_local_without_defining_it(self): + """Test that including 'local' works even if user didn't define it in config.""" + resources = { + "compute": {"provider": "slurm", "partition": "compute"}, + } + + # Including "local" in the list should work even though it's not in config + config = configure.load(resources, include=["local", "compute"]) + + # Should have default local + compute + assert len(config.executors) == 2 + labels = [ex.label for ex in config.executors] + assert "local" in labels + assert "compute" in labels + + # Verify it's the default local (1 core per worker) + local_ex = [ex for ex in config.executors if ex.label == "local"][0] + assert local_ex.cores_per_worker == 1 + class TestIntegrationScenarios: """Integration tests for realistic configuration scenarios.""" From 9e8aaacfaac1836ec852c4cccb9800c9a692ac50 Mon Sep 17 00:00:00 2001 From: Christopher Harrop Date: Mon, 9 Mar 2026 20:47:07 +0000 Subject: [PATCH 15/15] Fix test name --- tests/test_configure.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_configure.py b/tests/test_configure.py index 2a190bde..10e96347 100644 --- a/tests/test_configure.py +++ b/tests/test_configure.py @@ -675,7 +675,7 @@ def test_load_local_in_include_overrides_default(self): assert config.executors[0].label == "local" assert config.executors[0].cores_per_worker == 2 - def test_load_include_without_local_uses_default(self): + def test_load_include_without_local_uses_local_override(self): """Test that user's local is used even when not in include list.""" resources = { "local": {