cleveragents-core/robot/helper_m3_decision_validation_smoke.py

"""Helper script for m3_decision_validation_smoke.robot E2E tests.

Each subcommand is a self-contained check that prints a sentinel on success.
Uses ``--format plain`` where possible to stabilise assertion strings.
"""

from __future__ import annotations

import os
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path
from unittest.mock import MagicMock, patch

# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from helpers_common import reset_global_state  # noqa: E402
from typer.testing import CliRunner  # noqa: E402

from cleveragents.application.container import Container  # noqa: E402
from cleveragents.cli.commands.invariant import app as invariant_app  # noqa: E402
from cleveragents.cli.commands.plan import app as plan_app  # noqa: E402
from cleveragents.cli.commands.validation import app as validation_app  # noqa: E402
from cleveragents.domain.models.core.correction import (  # noqa: E402
    CorrectionMode,
    CorrectionStatus,
)
from cleveragents.domain.models.core.invariant import (  # noqa: E402
    Invariant,
    InvariantScope,
)

runner = CliRunner()


_PLAN_ULID = "01M3SM0KE00000000000000001"
_DECISION_ULID = "01M3DEC1S10N00000000000001"
_CORRECTION_ULID = "01M3C0RRECT10N000000000001"
_ATTACHMENT_ULID = "01M3ATTACH0000000000000001"
_INVARIANT_ULID = "01M3JNVAR1ANT0000000000001"


def _mock_invariant(
    *,
    text: str = "Never delete production data",
    scope: InvariantScope = InvariantScope.GLOBAL,
    source_name: str = "system",
) -> Invariant:
    """Create an Invariant for M3 smoke tests."""
    return Invariant(
        id=_INVARIANT_ULID,
        text=text,
        scope=scope,
        source_name=source_name,
    )


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def invariant_add_global() -> None:
    """Add a global invariant constraint."""
    mock_svc = MagicMock()
    mock_svc.add_invariant.return_value = _mock_invariant()
    with patch(
        "cleveragents.cli.commands.invariant._get_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            invariant_app,
            ["add", "--global", "Never delete production data", "--format", "plain"],
        )
    if result.exit_code == 0:
        print("m3-invariant-add-global-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def invariant_add_project() -> None:
    """Add a project-scoped invariant constraint."""
    mock_svc = MagicMock()
    mock_svc.add_invariant.return_value = _mock_invariant(
        text="All API changes need tests",
        scope=InvariantScope.PROJECT,
        source_name="local/m3-smoke-proj",
    )
    with patch(
        "cleveragents.cli.commands.invariant._get_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            invariant_app,
            [
                "add",
                "--project",
                "local/m3-smoke-proj",
                "All API changes need tests",
                "--format",
                "plain",
            ],
        )
    if result.exit_code == 0:
        print("m3-invariant-add-project-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def invariant_list() -> None:
    """List invariants."""
    mock_svc = MagicMock()
    mock_svc.list_invariants.return_value = [
        _mock_invariant(),
        _mock_invariant(
            text="All API changes need tests",
            scope=InvariantScope.PROJECT,
            source_name="local/m3-smoke-proj",
        ),
    ]
    with patch(
        "cleveragents.cli.commands.invariant._get_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            invariant_app,
            ["list", "--format", "plain"],
        )
    if result.exit_code == 0:
        print("m3-invariant-list-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def invariant_remove() -> None:
    """Remove an invariant by ID."""
    mock_svc = MagicMock()
    mock_svc.remove_invariant.return_value = _mock_invariant()
    with patch(
        "cleveragents.cli.commands.invariant._get_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            invariant_app,
            ["remove", "--yes", _INVARIANT_ULID, "--format", "plain"],
        )
    if result.exit_code == 0:
        print("m3-invariant-remove-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def validation_add() -> None:
    """Register a validation from YAML config."""
    config_content = (
        "name: local/coverage-check\n"
        "description: Check code coverage meets threshold\n"
        "source: custom\n"
        "mode: required\n"
        "code: |\n"
        "  def run(inputs):\n"
        "      return {'passed': inputs['coverage'] >= 80}\n"
    )
    with tempfile.NamedTemporaryFile(
        mode="w",
        suffix=".yaml",
        delete=False,
    ) as tmp:
        tmp.write(config_content)
        tmp_name = tmp.name

    try:
        mock_svc = MagicMock()
        mock_validation = MagicMock()
        mock_validation.name = "local/coverage-check"
        mock_validation.as_cli_dict.return_value = {
            "name": "local/coverage-check",
            "description": "Check code coverage meets threshold",
            "source": "custom",
            "mode": "required",
        }
        mock_svc.register_tool.return_value = mock_validation
        with patch(
            "cleveragents.cli.commands.validation._get_tool_registry_service",
            return_value=mock_svc,
        ):
            result = runner.invoke(
                validation_app,
                ["add", "--config", tmp_name, "--format", "plain"],
            )
        if result.exit_code == 0 and "coverage-check" in result.output:
            print("m3-validation-add-ok")
        else:
            print(f"FAIL: exit={result.exit_code} output={result.output}")
            sys.exit(1)
    finally:
        os.unlink(tmp_name)


def validation_attach() -> None:
    """Attach a validation to a resource."""
    mock_svc = MagicMock()
    mock_attachment = MagicMock()
    mock_attachment.attachment_id = _ATTACHMENT_ULID
    mock_attachment.validation_name = "local/coverage-check"
    mock_attachment.resource_id = "git-checkout/my-repo"
    mock_attachment.mode = "required"
    mock_attachment.project_name = None
    mock_attachment.plan_id = None
    mock_attachment.created_at = "2026-01-01T00:00:00"
    mock_svc.attach_validation.return_value = mock_attachment
    with patch(
        "cleveragents.cli.commands.validation._get_tool_registry_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            validation_app,
            [
                "attach",
                "git-checkout/my-repo",
                "local/coverage-check",
                "--format",
                "plain",
            ],
        )
    if result.exit_code == 0:
        print("m3-validation-attach-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def validation_detach() -> None:
    """Detach a validation from a resource."""
    mock_svc = MagicMock()
    mock_svc.detach_validation.return_value = True
    with patch(
        "cleveragents.cli.commands.validation._get_tool_registry_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            validation_app,
            ["detach", "--yes", _ATTACHMENT_ULID, "--format", "plain"],
        )
    if result.exit_code == 0:
        print("m3-validation-detach-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def plan_correct_dry_run() -> None:
    """Run plan correct in dry-run mode."""
    mock_correction_svc = MagicMock()
    mock_request = MagicMock()
    mock_request.correction_id = _CORRECTION_ULID
    mock_request.mode = CorrectionMode.REVERT
    mock_request.target_decision_id = _DECISION_ULID
    mock_request.guidance = "Use FastAPI instead"
    mock_request.status = CorrectionStatus.PENDING

    mock_impact = MagicMock()
    mock_impact.affected_decisions = [_DECISION_ULID]
    mock_impact.affected_files = ["src/main.py"]
    mock_impact.estimated_cost = "low"
    mock_impact.risk_level = "low"

    mock_correction_svc.request_correction.return_value = mock_request
    mock_correction_svc.analyze_impact.return_value = mock_impact

    # Mock DecisionService provider on DI container (issue #606 / #647 fix)
    mock_decision_svc = MagicMock()
    mock_decision_svc.list_decisions.return_value = []
    mock_decision_svc.get_influence_edges.return_value = {}
    mock_container = MagicMock(spec=Container)
    mock_container.decision_service.return_value = mock_decision_svc

    with (
        patch(
            "cleveragents.application.services.correction_service.CorrectionService",
            return_value=mock_correction_svc,
        ),
        patch(
            "cleveragents.application.container.get_container",
            return_value=mock_container,
        ),
    ):
        result = runner.invoke(
            plan_app,
            [
                "correct",
                _DECISION_ULID,
                "--mode",
                "revert",
                "--guidance",
                "Use FastAPI instead",
                "--dry-run",
                "--plan",
                _PLAN_ULID,
                "--format",
                "plain",
            ],
        )
    if result.exit_code == 0:
        print("m3-plan-correct-dry-run-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Callable[[], None]] = {
    "invariant-add-global": invariant_add_global,
    "invariant-add-project": invariant_add_project,
    "invariant-list": invariant_list,
    "invariant-remove": invariant_remove,
    "validation-add": validation_add,
    "validation-attach": validation_attach,
    "validation-detach": validation_detach,
    "plan-correct-dry-run": plan_correct_dry_run,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
        sys.exit(1)
    reset_global_state()
    fn = _COMMANDS[sys.argv[1]]
    fn()