cleveragents-core/robot/helper_validation_pipeline.py

"""Helper script for Robot Framework validation pipeline tests.

Self-contained Python helper that exercises the ValidationPipeline
and its models, printing sentinel strings on success and exiting
with code 1 on failure.
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from cleveragents.application.services.validation_pipeline import (
    ValidationCommand,
    ValidationPipeline,
    ValidationResult,
    ValidationSummary,
)
from cleveragents.domain.models.core.tool import ValidationMode

# ---------------------------------------------------------------------------
# Mock executor
# ---------------------------------------------------------------------------


def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    """Simple mock executor that always passes."""
    return {"passed": True, "message": f"{validation_name} passed"}


def _failing_executor(
    validation_name: str, arguments: dict[str, Any]
) -> dict[str, Any]:
    """Mock executor that always fails."""
    return {"passed": False, "message": f"{validation_name} failed"}


# ---------------------------------------------------------------------------
# Sub-commands
# ---------------------------------------------------------------------------


def test_models() -> None:
    """Verify ValidationCommand, ValidationResult, ValidationSummary models."""
    cmd = ValidationCommand(
        validation_name="check-syntax",
        resource_id="R00001",
        resource_name="repo-main",
        mode=ValidationMode.REQUIRED,
        arguments={"key": "value"},
        timeout_seconds=15.0,
    )
    assert cmd.validation_name == "check-syntax"
    assert cmd.mode == ValidationMode.REQUIRED
    assert cmd.timeout_seconds == 15.0

    result = ValidationResult(
        validation_name="check-syntax",
        resource_id="R00001",
        resource_name="repo-main",
        mode=ValidationMode.REQUIRED,
        passed=True,
        message="All good",
        data={"detail": "ok"},
        duration_ms=42.5,
        error=None,
        timed_out=False,
    )
    assert result.passed is True
    assert result.duration_ms == 42.5

    summary = ValidationSummary(
        total=2,
        required_passed=1,
        required_failed=1,
        informational_passed=0,
        informational_failed=0,
        results=[],
    )
    assert summary.all_required_passed is False
    print("models-ok")


def test_empty_pipeline() -> None:
    """Verify empty pipeline returns passing summary."""
    pipeline = ValidationPipeline(
        commands=[],
        executor=_mock_executor,
    )
    summary = pipeline.run()
    assert summary.total == 0
    assert summary.all_required_passed is True
    print("empty-pipeline-ok")


def test_passing_pipeline() -> None:
    """Verify pipeline with passing validations."""
    commands = [
        ValidationCommand(
            validation_name="check-a",
            resource_id="R1",
            resource_name="repo",
            mode=ValidationMode.REQUIRED,
        ),
        ValidationCommand(
            validation_name="check-b",
            resource_id="R1",
            resource_name="repo",
            mode=ValidationMode.INFORMATIONAL,
        ),
    ]
    pipeline = ValidationPipeline(
        commands=commands,
        executor=_mock_executor,
        max_workers=1,
    )
    summary = pipeline.run()
    assert summary.total == 2
    assert summary.required_passed == 1
    assert summary.informational_passed == 1
    assert summary.all_required_passed is True
    print("passing-pipeline-ok")


def test_failing_required() -> None:
    """Verify failing required validation blocks."""
    commands = [
        ValidationCommand(
            validation_name="check-fail",
            resource_id="R1",
            resource_name="repo",
            mode=ValidationMode.REQUIRED,
        ),
    ]
    pipeline = ValidationPipeline(
        commands=commands,
        executor=_failing_executor,
        max_workers=1,
    )
    summary = pipeline.run()
    assert summary.total == 1
    assert summary.required_failed == 1
    assert summary.all_required_passed is False
    print("failing-required-ok")


def test_run_for_plan() -> None:
    """Verify run_for_plan stores summary in metadata."""
    commands = [
        ValidationCommand(
            validation_name="plan-check",
            resource_id="R1",
            resource_name="repo",
            mode=ValidationMode.REQUIRED,
        ),
    ]
    pipeline = ValidationPipeline(
        commands=commands,
        executor=_mock_executor,
        max_workers=1,
    )
    metadata: dict[str, Any] = {}
    summary = pipeline.run_for_plan(plan_metadata=metadata)
    assert summary.total == 1
    assert "validation_summary" in metadata
    assert metadata["validation_summary"]["total"] == 1
    assert metadata["validation_summary"]["all_required_passed"] is True
    print("run-for-plan-ok")


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "models"
    commands_map = {
        "models": test_models,
        "empty_pipeline": test_empty_pipeline,
        "passing_pipeline": test_passing_pipeline,
        "failing_required": test_failing_required,
        "run_for_plan": test_run_for_plan,
    }
    handler = commands_map.get(cmd)
    if handler is None:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)
    try:
        handler()
    except Exception as exc:
        print(f"FAIL: {exc}", file=sys.stderr)
        sys.exit(1)