cursus.validation

Cursus Validation Framework

This module provides comprehensive validation capabilities for pipeline components, including step builders, alignment testing, runtime validation, and script testing.

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

ScriptExecutionResult

alias of ScriptTestResult

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

quick_test_dag(dag, config_path, workspace_dir='test/integration/script_testing')

Quick test function for DAG scripts with default settings.

Parameters:

• dag – PipelineDAG instance

• config_path (str) – Path to configuration file

• workspace_dir (str) – Test workspace directory

Returns:

Dictionary with execution results

Example

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

get_script_testing_info()

Get information about the simplified script testing framework.

Returns:

Dictionary with framework information

Return type:

dict

Modules

alignment	Unified Alignment Tester Module
builders	Streamlined Builders Validation Package
script_testing	Simplified Script Testing Framework
utils