cursus.validation.script_testing.api

Simplified Script Testing API

This module provides a streamlined script testing framework that extends existing cursus infrastructure instead of reimplementing it. The approach eliminates over-engineering by directly reusing DAGConfigFactory, StepCatalog, and UnifiedDependencyResolver components.

Key Functions:

run_dag_scripts: Main entry point for DAG-guided script testing execute_single_script: Execute individual scripts with dependency management install_script_dependencies: Handle package dependencies (valid complexity)

class ScriptTestResult(success, output_files=None, error_message=None, execution_time=None)

Bases: object

Simple result model for script execution.

run_dag_scripts(dag, config_path, test_workspace_dir='test/integration/script_testing', step_catalog=None, use_dependency_resolution=True)

ENHANCED: Run scripts with ScriptExecutionRegistry integration and message passing.

This function now uses the ScriptExecutionRegistry for central state coordination, enabling intelligent dependency resolution and automatic message passing between script executions.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance defining the pipeline structure

• config_path (str) – Path to pipeline configuration JSON file for script validation

• test_workspace_dir (str) – Directory for test workspace and script discovery

• step_catalog (StepCatalog | None) – Optional StepCatalog instance (will create if not provided)

• use_dependency_resolution (bool) – Whether to use two-phase dependency resolution

Returns:

Dictionary with execution results and metadata

Return type:

Dict[str, Any]

Example

>>> from cursus.validation.script_testing import run_dag_scripts
>>> from cursus.api.dag.base_dag import PipelineDAG
>>>
>>> dag = PipelineDAG.from_json("configs/xgboost_training.json")
>>> results = run_dag_scripts(
...     dag=dag,
...     config_path="pipeline_config/config_NA_xgboost_AtoZ.json"
... )
>>> print(f"Pipeline success: {results['pipeline_success']}")

collect_script_inputs_using_dag_factory(dag, config_path)

Collect script inputs by extending DAGConfigFactory patterns.

This function reuses the existing 600+ lines of proven interactive collection patterns instead of reimplementing them.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance

• config_path (str) – Path to configuration file for script validation

Returns:

Dictionary mapping script names to their input configurations

Return type:

Dict[str, Any]

get_validated_scripts_from_config(dag, configs)

Get only scripts with actual entry points from config (eliminates phantom scripts).

This addresses the phantom script issue by using config-based validation to ensure only scripts with actual entry points are discovered.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance

• configs (Dict[str, Any]) – Loaded configuration instances

Returns:

List of validated script names with actual entry points

Return type:

List[str]

collect_script_inputs(config)

Extract script path, environment variables, and job arguments from config.

This function uses proper config field access patterns instead of direct __dict__ access. It focuses on config-to-script transformation, not path management (that’s InputCollector’s job).

Parameters:

config – Populated config instance (BasePipelineConfig or derived)

Returns:

Dictionary with script_path, environment_variables, and job_arguments

Return type:

Dict[str, Any]

extract_script_path_from_config(config)

Extract script path from config entry point fields using proper config access.

Parameters:

config – Config instance with entry point fields

Returns:

Resolved script path or None if not found

Return type:

str | None

extract_environment_variables_from_config(config)

Extract environment variables from config using proper field access.

Parameters:

config – Config instance

Returns:

Dictionary of environment variables

Return type:

Dict[str, str]

extract_job_arguments_from_config(config)

Extract job arguments from config using proper field access.

Parameters:

config – Config instance

Returns:

argparse.Namespace with job arguments

execute_scripts_in_order(execution_order, user_inputs)

DRAMATICALLY SIMPLIFIED: Execute scripts with complete pre-resolved inputs.

All complexity (message passing, dependency matching, config extraction) is handled in input collection phase.

Parameters:

• execution_order (List[str]) – List of script names in topological order

• user_inputs (Dict[str, Any]) – Complete inputs from two-phase dependency resolution

Returns:

Dictionary with execution results

Return type:

Dict[str, Any]

execute_single_script(script_path, input_paths, output_paths, environ_vars, job_args)

Execute a single script with the fixed signature and dependency management.

This function handles the one legitimate complexity in script testing: package dependency management (scripts import packages that need installation).

Parameters:

• script_path (str) – Path to the script file

• input_paths (Dict[str, str]) – Input paths from InputCollector (contract-based logical names)

• output_paths (Dict[str, str]) – Output paths from InputCollector (contract-based logical names)

• environ_vars (Dict[str, str]) – Environment variables from config

• job_args – Job arguments from config (argparse.Namespace)

Returns:

ScriptTestResult with execution outcome

Return type:

ScriptTestResult

install_script_dependencies(script_path)

Install package dependencies for script execution.

This is the ONE valid complexity in script testing - scripts import packages that need to be installed before execution. In SageMaker pipeline, this was isolated as an environment.

Parameters:

script_path (str) – Path to the script file

parse_script_imports(script_path)

Parse script file to extract required packages.

Parameters:

script_path (str) – Path to the script file

Returns:

List of required package names

Return type:

List[str]

is_package_installed(package_name)

Check if a package is installed.

Parameters:

package_name (str) – Name of the package to check

Returns:

True if package is installed, False otherwise

Return type:

bool

install_package(package_name)

Install a package using pip.

Parameters:

package_name (str) – Name of the package to install

execute_scripts_with_registry_coordination(dag, registry)

Execute scripts with registry coordination and message passing.

This function uses the ScriptExecutionRegistry to coordinate script execution with automatic message passing between nodes.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance

• registry – ScriptExecutionRegistry instance

Returns:

Dictionary with execution results and metadata

Return type:

Dict[str, Any]

import_and_execute_script(script_path, input_paths, output_paths, environ_vars, job_args)

Import and execute a script with the fixed signature.

Uses the testability pattern: main(input_paths, output_paths, environ_vars, job_args)

Parameters:

• script_path (str) – Path to the script file

• input_paths (Dict[str, str]) – Input paths from InputCollector (contract-based logical names)

• output_paths (Dict[str, str]) – Output paths from InputCollector (contract-based logical names)

• environ_vars (Dict[str, str]) – Environment variables from config

• job_args – Job arguments from config (argparse.Namespace)

Returns:

Dictionary with execution results

Return type:

Dict[str, Any]