cursus.validation.alignment

Unified Alignment Tester Module

This module provides comprehensive validation of alignment rules between scripts, contracts, specifications, and builders in the pipeline architecture.

The alignment validation covers four levels: 1. Script ↔ Contract Alignment 2. Contract ↔ Specification Alignment 3. Specification ↔ Dependencies Alignment 4. Builder ↔ Configuration Alignment

Consolidated Structure (Post Phase 4): - config/: Configuration and validation rulesets - core/: Core alignment testers for each level - reporting/: Consolidated reporting and scoring (validation_reporter.py) - utils/: Consolidated utilities and models (validation_models.py) - validators/: Remaining validation logic and rules

class UnifiedAlignmentTester(workspace_dirs=None, **kwargs)

Bases: object

Enhanced Unified Alignment Tester with configuration-driven validation.

This class orchestrates validation across all levels based on step-type-aware configuration rules, providing dramatic performance improvements through validation level skipping.

discover_scripts()

Discover scripts - maintained for backward compatibility.

Returns:

List of discovered script names (only steps with actual script files)

Return type:

List[str]

export_report(format='json', output_path=None)

Export validation report - enhanced with configuration insights.

Parameters:

• format (str) – Report format (“json” or “html”)

• output_path (str | None) – Optional output file path

Returns:

Report content as string

Return type:

str

get_component_path_from_catalog(step_name, component_type)

Get component file path from step catalog - maintained for backward compatibility.

Parameters:

• step_name (str) – Name of the step

• component_type (str) – Type of component (‘script’, ‘contract’, ‘spec’, ‘builder’, ‘config’)

Returns:

Path to component file or None if not found

Return type:

Path | None

get_critical_issues()

Get critical validation issues - step-type-aware critical issue analysis.

Returns:

List of critical issues

Return type:

List[Dict[str, Any]]

get_step_info_from_catalog(step_name)

Get step information from step catalog - maintained for backward compatibility.

Parameters:

step_name (str) – Name of the step

Returns:

StepInfo object or None if not found

Return type:

Any | None

get_validation_summary()

Get validation summary - enhanced with step-type-aware metrics.

Returns:

Dictionary containing validation summary

Return type:

Dict[str, Any]

print_summary()

Print enhanced validation summary to console.

run_full_validation(target_scripts=None, skip_levels=None)

Enhanced run_full_validation with configuration-driven approach.

Parameters:

• target_scripts (List[str] | None) – Optional list of specific scripts to validate

• skip_levels (Set[int] | None) – Optional set of validation levels to skip (legacy support)

Returns:

Dictionary containing validation results

Return type:

Dict[str, Any]

run_validation_for_all_steps()

Run validation for all discovered steps.

Returns:

Dictionary containing validation results for all steps

Return type:

Dict[str, Any]

run_validation_for_step(step_name)

Run validation for a specific step based on its ruleset.

Parameters:

step_name (str) – Name of the step to validate

Returns:

Dictionary containing validation results

Return type:

Dict[str, Any]

validate_cross_workspace_compatibility(step_names)

Validate compatibility across workspace components - simplified with configuration.

Parameters:

step_names (List[str]) – List of step names to validate

Returns:

Compatibility validation results

Return type:

Dict[str, Any]

validate_specific_script(step_name, skip_levels=None)

Validate a specific script - maintained for backward compatibility.

Parameters:

• step_name (str) – Name of the step to validate

• skip_levels (Set[int] | None) – Optional set of validation levels to skip (ignored in new system)

Returns:

Dictionary containing validation results

Return type:

Dict[str, Any]

class ScriptContractAlignmentTester(workspace_dirs=None)

Bases: object

Tests alignment between processing scripts and their contracts.

Validates: - Path usage matches contract declarations - Environment variable access matches contract - Script arguments align with contract expectations - File operations match declared inputs/outputs

get_validation_summary(results)

Generate a summary of validation results.

validate_all_scripts(target_scripts=None)

Validate alignment for all scripts or specified target scripts.

Parameters:

target_scripts (List[str] | None) – Specific scripts to validate (None for all)

Returns:

Dictionary mapping script names to validation results

Return type:

Dict[str, Dict[str, Any]]

validate_script(script_name)

Validate alignment for a specific script.

Parameters:

script_name (str) – Name of the script to validate

Returns:

Validation result dictionary

Return type:

Dict[str, Any]

class SpecificationDependencyAlignmentTester(validation_config=None, workspace_dirs=None)

Bases: object

Tests alignment between step specifications and their dependencies.

Validates: - Dependency chains are consistent - All dependencies can be resolved - No circular dependencies exist - Data types match across dependency chains

get_dependency_resolution_report(all_specs)

Generate detailed dependency resolution report using production resolver.

validate_all_specifications(target_scripts=None)

Validate alignment for all specifications or specified target scripts.

This method uses StepCatalog’s bulk loading for efficiency.

Parameters:

target_scripts (List[str] | None) – Specific scripts to validate (None for all)

Returns:

Dictionary mapping specification names to validation results

Return type:

Dict[str, Dict[str, Any]]

validate_specification(spec_name)

Validate alignment for a specific specification.

Parameters:

spec_name (str) – Name of the specification to validate

Returns:

Validation result dictionary

Return type:

Dict[str, Any]

validate_specification_object(specification, spec_name=None)

Validate a pre-loaded specification object.

Parameters:

• specification (Dict[str, Any]) – Serialized specification dictionary

• spec_name (str) – Optional specification name for context

Returns:

Validation result dictionary

Return type:

Dict[str, Any]

class DependencyValidator(validation_config=None)

Bases: object

Validates dependencies between step specifications.

Features: - Enhanced dependency resolution with compatibility scoring - Circular dependency detection - Data type consistency validation - Integration with production registry for canonical name mapping

get_dependency_resolution_report(all_specs)

Generate detailed dependency resolution report using production resolver.

Parameters:

all_specs (Dict[str, Dict[str, Any]]) – Dictionary of all available specifications

Returns:

Detailed resolution report

Return type:

Dict[str, Any]

validate_circular_dependencies(specification, all_specs, spec_name)

Validate that no circular dependencies exist.

Parameters:

• specification (Dict[str, Any]) – The specification to validate

• all_specs (Dict[str, Dict[str, Any]]) – Dictionary of all available specifications

• spec_name (str) – Name of the specification being validated

Returns:

List of validation issues

Return type:

List[Dict[str, Any]]

validate_dependency_data_types(specification, all_specs, spec_name)

Validate data type consistency across dependency chains.

Parameters:

• specification (Dict[str, Any]) – The specification to validate

• all_specs (Dict[str, Dict[str, Any]]) – Dictionary of all available specifications

• spec_name (str) – Name of the specification being validated

Returns:

List of validation issues

Return type:

List[Dict[str, Any]]

validate_dependency_resolution(specification, all_specs, spec_name)

Enhanced dependency validation with compatibility scoring.

Parameters:

• specification (Dict[str, Any]) – The specification to validate dependencies for

• all_specs (Dict[str, Dict[str, Any]]) – Dictionary of all available specifications

• spec_name (str) – Name of the specification being validated

Returns:

List of validation issues

Return type:

List[Dict[str, Any]]

class SageMakerPropertyPathValidator

Bases: object

Validates SageMaker step property paths against official documentation.

This validator ensures that property paths used in step specifications are valid for the specific SageMaker step type, preventing runtime errors in pipeline execution.

get_step_type_documentation(step_type, node_type='')

Get documentation information for a specific step type.

Parameters:

• step_type (str) – The SageMaker step type

• node_type (str) – The node type (optional)

Returns:

Dictionary with documentation information

Return type:

Dict[str, Any]

list_supported_step_types()

List all supported step types and their documentation.

Returns:

List of supported step types with their information

Return type:

List[Dict[str, Any]]

validate_specification_property_paths(specification, contract_name)

Validate all property paths in a specification.

Parameters:

• specification (Dict[str, Any]) – Specification dictionary

• contract_name (str) – Name of the contract being validated

Returns:

List of validation issues

Return type:

List[Dict[str, Any]]

class ValidationLevel(*values)

Bases: Enum

The 3 validation BOUNDARIES — what construction can’t self-check (FZ 31e1d3h / Phase D5).

The unified .step.yaml made Contract<->Spec alignment a construction-time Pydantic invariant (StepInterface._sync_and_align), so the former Level-2 (CONTRACT_SPEC=2) validated a tautology and has been removed. The three surviving levels are renamed in SEMANTICS to the 3 boundaries — SCRIPT_CONTRACT is the Script<->Interface fidelity boundary (B1), SPEC_DEPENDENCY is the cross-step DAG-resolvability boundary (B2, which now also owns the SageMaker property-path check folded in from the old L2), and BUILDER_CONFIG is the registry<->handler<->config binding boundary (B3). The member NAMES + integer VALUES are kept (1/3/4, non-contiguous) so ValidationLevel(1/3/4) coercion and every existing reference keep working; only CONTRACT_SPEC (value 2) is gone.

SCRIPT_CONTRACT = 1

SPEC_DEPENDENCY = 3

BUILDER_CONFIG = 4

class ValidationStatus(*values)

Bases: Enum

Status of validation operations.

PASSED = 'PASSED'

FAILED = 'FAILED'

EXCLUDED = 'EXCLUDED'

ERROR = 'ERROR'

ISSUES_FOUND = 'ISSUES_FOUND'

PASSED_WITH_WARNINGS = 'PASSED_WITH_WARNINGS'

NO_VALIDATOR = 'NO_VALIDATOR'

COMPLETED = 'COMPLETED'

class IssueLevel(*values)

Bases: Enum

Severity levels for validation issues.

ERROR = 'ERROR'

WARNING = 'WARNING'

INFO = 'INFO'

class RuleType(*values)

Bases: Enum

Types of validation rules.

UNIVERSAL = 'universal'

STEP_SPECIFIC = 'step_specific'

METHOD_INTERFACE = 'method_interface'

CONFIGURATION = 'configuration'

class ValidationIssue(level, message, method_name=None, rule_type=None, details=None, step_name=None, file_path=None, line_number=None)

Bases: object

Represents a validation issue found during alignment checking.

details: Dict[str, Any] | None = None

file_path: str | None = None

line_number: int | None = None

method_name: str | None = None

rule_type: RuleType | None = None

step_name: str | None = None

to_dict()

Convert issue to dictionary format.

level: IssueLevel

message: str

class ValidationResult(status, step_name, validation_level=None, issues=None, metadata=None, error_message=None)

Bases: object

Represents the result of a validation operation.

add_issue(issue)

Add a validation issue to the result.

property error_count: int

Count of error-level issues.

error_message: str | None = None

property info_count: int

Count of info-level issues.

issues: List[ValidationIssue] = None

metadata: Dict[str, Any] | None = None

to_dict()

Convert result to dictionary format.

property total_issues: int

Total count of all issues.

validation_level: ValidationLevel | None = None

property warning_count: int

Count of warning-level issues.

status: ValidationStatus

step_name: str

class ValidationSummary

Bases: object

Summary of validation results across multiple steps.

add_result(result)

Add a validation result to the summary.

property success_rate: float

Calculate success rate (passed / total non-excluded).

to_dict()

Convert summary to dictionary format.

class MethodValidationInfo(method_name, is_required, expected_signature=None, return_type=None, purpose=None, validation_rules=None)

Bases: object

Information about method validation requirements.

expected_signature: str | None = None

purpose: str | None = None

return_type: str | None = None

validation_rules: List[str] | None = None

method_name: str

is_required: bool

class StepValidationContext(step_name, step_type, builder_class_name=None, workspace_dirs=None, validation_config=None)

Bases: object

Context information for step validation.

builder_class_name: str | None = None

validation_config: Dict[str, Any] | None = None

workspace_dirs: List[str] | None = None

step_name: str

step_type: str

create_validation_issue(level, message, method_name=None, rule_type=None, details=None, **kwargs)

Create a validation issue with proper type conversion.

create_validation_result(status, step_name, validation_level=None, issues=None, **kwargs)

Create a validation result with proper type conversion.

merge_validation_results(results)

Merge multiple validation results into a single result.

filter_issues_by_level(issues, level)

Filter validation issues by severity level.

group_issues_by_method(issues)

Group validation issues by method name.

format_validation_summary(summary)

Format validation summary as a readable string.

Modules

analyzer	Restored Script Analysis Module
config	Validation Alignment Configuration Module
core	Core Alignment Testers Module
reporting	Reporting and Visualization Module
unified_alignment_tester	Enhanced Unified Alignment Tester
utils	Utilities and Models Module
validators	Validators Module