cursus.validation.script_testing

Simplified Script Testing Framework

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.

This simplified implementation reduces code from 4,200 lines across 17 modules to 800-1,000 lines across 5 modules while maintaining all functionality and addressing the 3 key user stories:

• US1: Individual Script Functionality Testing

• US2: Data Transfer and Compatibility Testing

• US3: DAG-Guided End-to-End Testing

Key Features: - Maximum infrastructure reuse (95% of existing cursus components) - Config-based phantom script elimination - Package dependency management (the one valid complexity) - Comprehensive result formatting - Interactive input collection extending DAGConfigFactory patterns

Main API Functions:

run_dag_scripts: Main entry point for DAG-guided script testing

Core Components:

ScriptTestingInputCollector: Extends DAGConfigFactory for input collection ResultFormatter: Comprehensive result formatting (preserved from original) ScriptTestResult: 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']}")

class ScriptTestingInputCollector(dag, config_path, registry=None, use_dependency_resolution=True)

Bases: object

Enhanced with Direct Registry Integration for field value population.

This class reuses the existing 600+ lines of proven interactive collection patterns and integrates directly with ScriptExecutionRegistry for: - Registry-coordinated field population - Message passing between script executions - Dynamic environment variable generation - Dependency-aware path resolution

collect_script_inputs_for_dag()

Enhanced collection with Direct Registry Integration.

This method supports three modes: 1. Registry-integrated collection (NEW) 2. Two-phase dependency resolution 3. Manual collection (backward compatibility)

Returns:

Dictionary mapping script names to their input configurations

Return type:

Dict[str, Any]

get_collection_summary()

Get summary of input collection status.

Returns:

Dictionary with collection summary

Return type:

Dict[str, Any]

get_script_requirements(script_name)

Get requirements for a specific script.

This method extends DAGConfigFactory.get_step_requirements() for script testing.

Parameters:

script_name (str) – Name of the script

Returns:

Dictionary with script requirements

Return type:

Dict[str, Any]

class ResultFormatter(format_options=None)

Bases: object

Comprehensive result formatting utilities for script testing results.

Provides multiple output formats and visualization options for script execution results, including console output, JSON, CSV, and HTML reports.

Key features: 1. Multiple output formats (console, JSON, CSV, HTML) 2. Customizable formatting options 3. Summary and detailed reporting 4. Error highlighting and analysis 5. Performance metrics visualization

format_options

Dictionary of formatting configuration options

create_summary_report(results)

Create a comprehensive summary report.

Parameters:

results (Dict[str, Any]) – Dictionary with execution results

Returns:

Formatted summary report

Return type:

str

format_execution_results(results, format_type='console')

Format complete execution results in specified format.

Parameters:

• results (Dict[str, Any]) – Dictionary with execution results from script testing

• format_type (str) – Output format (“console”, “json”, “csv”, “html”)

Returns:

Formatted results string

Return type:

str

format_script_result(script_result, format_type='console')

Format individual script result.

Parameters:

• script_result (ScriptTestResult) – ScriptTestResult to format

• format_type (str) – Output format (“console”, “json”, “summary”)

Returns:

Formatted script result string

Return type:

str

get_formatter_summary()

Get a summary of the formatter configuration and capabilities.

Returns:

Dictionary with formatter summary information

Return type:

Dict[str, Any]

save_results_to_file(results, output_path, format_type='json')

Save results to file in specified format.

Parameters:

• results (Dict[str, Any]) – Dictionary with execution results

• output_path (str) – Path to save results

• format_type (str) – Output format (“json”, “csv”, “html”, “txt”)

Returns:

Path to saved file

Return type:

Path

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

Bases: object

Simple result model for script execution.

class ScriptExecutionRegistry(dag, step_catalog=None)

Bases: object

Central state coordinator that integrates both layers:

Layer 1 (script_dependency_matcher): DAG-level orchestration Layer 2 (script_input_resolver): Script-level resolution

Registry Role: - Maintains DAG execution state - Coordinates message passing between layers - Provides state interface for both layers - Ensures sequential consistency

clear_registry()

Clear registry for testing or new execution.

Resets all state while preserving DAG and step catalog references.

commit_execution_results(node_name, execution_result)

Integration Point 6: Store execution results for message passing.

API layer calls this after script execution to update state.

Parameters:

• node_name (str) – Name of the executed node

• execution_result (ScriptTestResult) – Result of script execution

get_dependency_outputs_for_node(node_name)

Integration Point 2: Provide dependency outputs for message passing.

script_dependency_matcher calls this to get outputs from completed dependencies.

Parameters:

node_name (str) – Name of the node requesting dependency outputs

Returns:

Dictionary mapping output names to actual paths from completed dependencies

Return type:

Dict[str, str]

get_execution_summary()

Get summary of execution state and message passing.

Returns:

Dictionary with execution summary information

Return type:

Dict[str, Any]

get_message_passing_history()

Get complete message passing history for analysis.

Returns:

List of message passing events with timestamps

Return type:

List[Dict[str, Any]]

get_node_config_for_resolver(node_name)

Integration Point 3: Provide node config to script_input_resolver.

script_input_resolver calls this to get base configuration for a node.

Parameters:

node_name (str) – Name of the node requesting configuration

Returns:

Dictionary containing node configuration data

Return type:

Dict[str, Any]

get_node_outputs(node_name)

Get outputs produced by a specific node.

Parameters:

node_name (str) – Name of the node

Returns:

Dictionary of outputs produced by the node

Return type:

Dict[str, str]

get_node_status(node_name)

Get current status of a specific node.

Parameters:

node_name (str) – Name of the node

Returns:

Current status (‘pending’, ‘ready’, ‘completed’, ‘failed’)

Return type:

str

get_ready_node_inputs(node_name)

Integration Point 5: Provide complete inputs for script execution.

API layer calls this to get final inputs for script execution.

Parameters:

node_name (str) – Name of the node requesting inputs

Returns:

Dictionary containing complete input configuration for script execution

Return type:

Dict[str, Any]

initialize_from_dependency_matcher(prepared_data)

Integration Point 1: Receive prepared data from script_dependency_matcher.

script_dependency_matcher calls this to initialize registry state.

Parameters:

prepared_data (Dict[str, Any]) – Output from prepare_script_testing_inputs() containing: - node_specs: Loaded step specifications - dependency_matches: Automatic dependency matches - config_data: Extracted configuration data - execution_order: Topological sort order

sequential_state_update()

Generator that yields nodes in topological order with updated state.

This ensures: - Dependencies are processed before dependents - State updates are sequential and consistent - Message passing happens in correct order

Yields:

Tuple of (node_name, updated_node_state) for each node in execution order

store_resolved_inputs(node_name, resolved_inputs)

Integration Point 4: Store resolved inputs from script_input_resolver.

script_input_resolver calls this to store its resolution results.

Parameters:

• node_name (str) – Name of the node

• resolved_inputs (Dict[str, Any]) – Dictionary containing resolved input configuration

class DAGStateConsistency

Bases: object

Ensures state consistency during sequential message passing.

Guarantees: 1. Dependencies are always processed before dependents (topological order) 2. Node state is only updated when all dependencies are completed 3. Message passing only uses outputs from completed nodes 4. State updates are atomic and consistent

static ensure_state_consistency(registry, node_name)

Ensure node state is consistent before execution.

Parameters:

• registry (ScriptExecutionRegistry) – ScriptExecutionRegistry instance

• node_name (str) – Name of the node to validate

Raises:

RuntimeError – If state is inconsistent

static validate_execution_order(dag, execution_order)

Validate that execution order respects dependency constraints.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance

• execution_order (List[str]) – List of node names in execution order

Raises:

ValueError – If execution order violates dependency constraints

create_script_execution_registry(dag, step_catalog=None)

Factory function to create a ScriptExecutionRegistry instance.

Parameters:

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

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

Returns:

Configured ScriptExecutionRegistry instance

Return type:

ScriptExecutionRegistry

resolve_script_dependencies(dag, config_path, step_catalog, registry=None)

SIMPLIFIED: Two-phase dependency resolution with optional registry integration.

When registry is provided, uses registry coordination for state management. When registry is None, uses legacy standalone mode for backward compatibility.

Parameters:

• dag (PipelineDAG) – PipelineDAG defining script execution order and dependencies

• config_path (str) – Path to configuration file for config-based extraction

• step_catalog (StepCatalog) – For loading specifications and contracts

• registry – Optional ScriptExecutionRegistry for state coordination

Returns:

Complete user inputs ready for script execution

Return type:

Dict[str, Any]

prepare_script_testing_inputs(dag, config_path, step_catalog)

Phase 1: Automatic dependency analysis using pipeline assembler patterns.

DIRECT REUSE of PipelineAssembler._propagate_messages() algorithm.

Parameters:

• dag (PipelineDAG) – PipelineDAG defining script execution order and dependencies

• config_path (str) – Path to configuration file for config-based extraction

• step_catalog (StepCatalog) – For loading specifications and contracts

Returns:

Prepared data structure containing dependency matches and config data

Return type:

Dict[str, Any]

collect_user_inputs_with_dependency_resolution(prepared_data)

Phase 2: Interactive input collection with automatic dependency resolution.

Mirrors PipelineAssembler._propagate_messages() + StepBuilder._get_inputs() patterns.

Parameters:

prepared_data (Dict[str, Any]) – Output from prepare_script_testing_inputs()

Returns:

Complete user inputs ready for script execution

Return type:

Dict[str, Any]

validate_dependency_resolution_result(user_inputs)

Validate the result of dependency resolution.

Parameters:

user_inputs (Dict[str, Any]) – Result from resolve_script_dependencies()

Returns:

True if validation passes, False otherwise

Return type:

bool

get_dependency_resolution_summary(prepared_data, user_inputs)

Generate a summary of the dependency resolution process.

Parameters:

• prepared_data (Dict[str, Any]) – Output from prepare_script_testing_inputs()

• user_inputs (Dict[str, Any]) – Output from collect_user_inputs_with_dependency_resolution()

Returns:

Summary statistics and information

Return type:

Dict[str, Any]

resolve_script_inputs_using_step_patterns(node_name, spec, resolved_dependencies, step_catalog)

Script input resolution adapted from StepBuilder._get_inputs() patterns.

DIRECT ADAPTATION of step builder input resolution logic for script testing. This function mirrors the same patterns used in step builders for input resolution, providing consistent behavior between pipeline steps and script testing.

Parameters:

• node_name (str) – Name of the script/node

• spec (StepInterface) – Step specification with dependencies and outputs

• resolved_dependencies (Dict[str, str]) – Dictionary of resolved dependency paths

• step_catalog (StepCatalog) – Step catalog for contract loading

Returns:

Dictionary mapping logical dependency names to actual script input paths

Return type:

Dict[str, str]

Example

>>> spec = step_catalog.load_spec_class('DataPreprocessing')
>>> resolved_deps = {'training_data': '/data/input/train.csv'}
>>> script_inputs = resolve_script_inputs_using_step_patterns(
...     'DataPreprocessing', spec, resolved_deps, step_catalog
... )
>>> print(script_inputs)
{'training_data': '/data/input/train.csv'}

adapt_step_input_patterns_for_scripts(node_name, inputs, step_catalog)

Adapt step builder input patterns for script testing.

DIRECT ADAPTATION of step builder input resolution patterns. This function provides the same input validation and transformation patterns used in step builders, ensuring consistency across the system.

Parameters:

• node_name (str) – Name of the script/node

• inputs (Dict[str, Any]) – Dictionary of input data to process

• step_catalog (StepCatalog) – Step catalog for specification and contract loading

Returns:

Dictionary mapping logical names to actual script input paths

Raises:

ValueError – If specification or contract not found, or required inputs missing

Return type:

Dict[str, str]

Example

>>> inputs = {'training_data': '/data/train.csv', 'model_config': '/config/model.json'}
>>> script_inputs = adapt_step_input_patterns_for_scripts(
...     'XGBoostTraining', inputs, step_catalog
... )
>>> print(script_inputs)
{'training_data': '/data/train.csv', 'model_config': '/config/model.json'}

validate_script_input_resolution(node_name, script_inputs, step_catalog)

Validate script input resolution using step builder validation patterns.

This function applies the same validation logic used in step builders to ensure script inputs are properly resolved and valid.

Parameters:

• node_name (str) – Name of the script/node

• script_inputs (Dict[str, str]) – Dictionary of resolved script inputs

• step_catalog (StepCatalog) – Step catalog for specification loading

Returns:

True if validation passes, False otherwise

Return type:

bool

Example

>>> script_inputs = {'training_data': '/data/train.csv'}
>>> is_valid = validate_script_input_resolution(
...     'DataPreprocessing', script_inputs, step_catalog
... )
>>> print(is_valid)
True

get_script_input_resolution_summary(node_name, script_inputs, step_catalog)

Generate a summary of script input resolution for debugging and monitoring.

Parameters:

• node_name (str) – Name of the script/node

• script_inputs (Dict[str, str]) – Dictionary of resolved script inputs

• step_catalog (StepCatalog) – Step catalog for specification loading

Returns:

Dictionary with resolution summary information

Return type:

Dict[str, Any]

Example

>>> script_inputs = {'training_data': '/data/train.csv', 'config': '/config/model.json'}
>>> summary = get_script_input_resolution_summary(
...     'XGBoostTraining', script_inputs, step_catalog
... )
>>> print(summary['total_inputs'])
2

transform_logical_names_to_actual_paths(logical_inputs, node_name, step_catalog)

Transform logical dependency names to actual file paths using contract patterns.

This function applies the same logical-to-actual path transformation used in step builders, ensuring consistent path resolution across the system.

Parameters:

• logical_inputs (Dict[str, str]) – Dictionary mapping logical names to paths

• node_name (str) – Name of the script/node

• step_catalog (StepCatalog) – Step catalog for contract loading

Returns:

Dictionary mapping logical names to actual file paths

Return type:

Dict[str, str]

Example

>>> logical_inputs = {'training_data': '/container/input/data'}
>>> actual_paths = transform_logical_names_to_actual_paths(
...     logical_inputs, 'DataPreprocessing', step_catalog
... )
>>> print(actual_paths)
{'training_data': '/opt/ml/input/data/training_data.csv'}

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

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]

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]

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

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]

validate_dag_and_config(dag, config_path)

Validate DAG and config path inputs.

Parameters:

• dag – PipelineDAG instance

• config_path (str) – Path to configuration file

Returns:

Dictionary with validation results

Raises:

ValueError – If validation fails

Return type:

Dict[str, Any]

create_test_workspace(workspace_dir)

Create test workspace directory structure.

Parameters:

workspace_dir (str) – Path to workspace directory

Returns:

Path to created workspace directory

Return type:

Path

load_json_config(config_path)

Load JSON configuration file.

Parameters:

config_path (str) – Path to JSON configuration file

Returns:

Dictionary with configuration data

Return type:

Dict[str, Any]

save_execution_results(results, output_path)

Save execution results to file.

Parameters:

• results (Dict[str, Any]) – Execution results dictionary

• output_path (str) – Path to save results

Returns:

Path to saved file

Return type:

Path

calculate_execution_summary(results)

Calculate execution summary statistics.

Parameters:

results (Dict[str, Any]) – Execution results dictionary

Returns:

Dictionary with summary statistics

Return type:

Dict[str, Any]

format_execution_time(seconds)

Format execution time for display.

Parameters:

seconds (float | None) – Execution time in seconds

Returns:

Formatted time string

Return type:

str

get_script_info(script_path)

Get information about a script file.

Parameters:

script_path (str) – Path to script file

Returns:

Dictionary with script information

Return type:

Dict[str, Any]

check_has_main_function(script_path)

Check if a Python script has a main function.

Parameters:

script_path (str) – Path to Python script

Returns:

True if script has main function, False otherwise

Return type:

bool

create_default_paths(script_name, base_dir='test')

Create default input and output paths for a script.

Parameters:

• script_name (str) – Name of the script

• base_dir (str) – Base directory for paths

Returns:

Dictionary with input and output path mappings

Return type:

Dict[str, Dict[str, str]]

merge_script_configs(base_config, script_config)

Merge base configuration with script-specific configuration.

Parameters:

• base_config (Dict[str, Any]) – Base configuration dictionary

• script_config (Dict[str, Any]) – Script-specific configuration dictionary

Returns:

Merged configuration dictionary

Return type:

Dict[str, Any]

validate_script_inputs(inputs)

Validate script input configuration.

Parameters:

inputs (Dict[str, Any]) – Script input configuration

Returns:

List of validation error messages (empty if valid)

Return type:

List[str]

get_testing_summary()

Get summary of script testing utilities.

Returns:

Dictionary with utility summary information

Return type:

Dict[str, Any]

Modules

api	Simplified Script Testing API
input_collector	Script Testing Input Collector
result_formatter	Result Formatter
script_dependency_matcher	Two-Phase Script Dependency Resolution System
script_execution_registry	Script Execution Registry
script_input_resolver	Script Input Resolution Pattern Adaptation
utils	Script Testing Utilities