cleveragents-core/features/steps/automation_profile_cli_steps.py

"""Step definitions for the Automation Profile CLI feature."""

from __future__ import annotations

import os
import re
import tempfile
from unittest.mock import MagicMock, patch

from behave import given, then, when
from behave.runner import Context
from typer.testing import CliRunner

from cleveragents.cli.commands.automation_profile import (
    _InMemoryProfileRepository,
)
from cleveragents.cli.commands.automation_profile import (
    app as profile_app,
)
from cleveragents.domain.models.core.automation_profile import AutomationProfile

# Regex to strip ANSI escape sequences that Rich may emit.
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]")


def _normalize_cli_output(text: str) -> str:
    """Strip ANSI codes and collapse whitespace so assertions are immune to
    Rich panel wrapping caused by varying terminal widths (CI vs local)."""
    text = _ANSI_RE.sub("", text)
    return " ".join(text.split())


_VALID_YAML = """\
name: acme/strict
description: Strict review for production
schema_version: "1.0"
auto_strategize: 1.0
auto_execute: 1.0
auto_apply: 1.0
auto_decisions_strategize: 1.0
auto_decisions_execute: 1.0
auto_validation_fix: 1.0
auto_strategy_revision: 1.0
auto_reversion_from_apply: 1.0
auto_child_plans: 1.0
auto_retry_transient: 1.0
auto_checkpoint_restore: 1.0
require_sandbox: true
require_checkpoints: true
allow_unsafe_tools: false
"""

_INVALID_NAME_YAML = """\
name: "invalid name with spaces!"
description: Bad profile
schema_version: "1.0"
"""

_INVALID_THRESHOLD_YAML = """\
name: acme/bad-threshold
description: Bad threshold
schema_version: "1.0"
auto_strategize: 2.0
"""

_INVALID_YAML_CONTENT = """\
name: [this is not valid yaml
  because: it has bad syntax
"""


def _make_custom_profile(
    name: str = "acme/strict",
    description: str = "Strict review for production",
) -> AutomationProfile:
    """Create a custom AutomationProfile for testing."""
    return AutomationProfile(
        name=name,
        description=description,
        schema_version="1.0",
        auto_strategize=1.0,
        auto_execute=1.0,
        auto_apply=1.0,
        auto_decisions_strategize=1.0,
        auto_decisions_execute=1.0,
        auto_validation_fix=1.0,
        auto_strategy_revision=1.0,
        auto_reversion_from_apply=1.0,
        auto_child_plans=1.0,
        auto_retry_transient=1.0,
        auto_checkpoint_restore=1.0,
        require_sandbox=True,
        require_checkpoints=True,
        allow_unsafe_tools=False,
    )


def _write_temp_yaml(context: Context, content: str) -> str:
    """Write YAML content to a temporary file and register cleanup."""
    fd, path = tempfile.mkstemp(suffix=".yaml")
    with os.fdopen(fd, "w") as fh:
        fh.write(content)
    if not hasattr(context, "_cleanup_handlers"):
        context._cleanup_handlers = []
    context._cleanup_handlers.append(
        lambda p=path: os.unlink(p) if os.path.exists(p) else None
    )
    return path


@given("an automation profile CLI runner with mocks")
def step_automation_profile_cli_runner(context: Context) -> None:
    """Set up the CLI runner for testing."""
    context.runner = CliRunner()
    context.result = None
    # Reset the module-level in-memory repo for each scenario
    import cleveragents.cli.commands.automation_profile as ap_mod

    ap_mod._repo = _InMemoryProfileRepository()


@given("a valid automation profile config YAML file")
def step_valid_profile_config(context: Context) -> None:
    """Write a valid profile config to a temp file."""
    context.yaml_path = _write_temp_yaml(context, _VALID_YAML)


@given('the custom profile "{name}" already exists')
def step_custom_profile_exists(context: Context, name: str) -> None:
    """Pre-register a custom profile in the in-memory repo."""
    import cleveragents.cli.commands.automation_profile as ap_mod

    profile = _make_custom_profile(name=name)
    ap_mod._repo.upsert(profile)


@given("an invalid YAML automation profile config file")
def step_invalid_yaml_config(context: Context) -> None:
    """Write an invalid YAML to a temp file."""
    context.invalid_yaml_path = _write_temp_yaml(context, _INVALID_YAML_CONTENT)


@given("an automation profile config YAML file with invalid name")
def step_invalid_name_config(context: Context) -> None:
    """Write a profile config with invalid name to a temp file."""
    context.invalid_name_yaml_path = _write_temp_yaml(context, _INVALID_NAME_YAML)


@given("an automation profile config YAML file with invalid threshold")
def step_invalid_threshold_config(context: Context) -> None:
    """Write a profile config with invalid threshold to a temp file."""
    context.invalid_threshold_yaml_path = _write_temp_yaml(
        context, _INVALID_THRESHOLD_YAML
    )


@given('an automation profile config YAML file with built-in name "{name}"')
def step_builtin_name_config(context: Context, name: str) -> None:
    """Write a profile config with a built-in name to a temp file."""
    yaml_content = f"""\
name: {name}
description: Attempt to overwrite built-in
schema_version: "1.0"
auto_strategize: 0.5
"""
    context.builtin_name_yaml_path = _write_temp_yaml(context, yaml_content)


@given('an automation profile config YAML file with schema_version "{version}"')
def step_unsupported_schema_config(context: Context, version: str) -> None:
    """Write a profile config with unsupported schema version."""
    yaml_content = f"""\
name: acme/unsupported
description: Unsupported schema
schema_version: "{version}"
"""
    context.unsupported_schema_yaml_path = _write_temp_yaml(context, yaml_content)


@given('a custom profile "{name}" has been added')
def step_custom_profile_added(context: Context, name: str) -> None:
    """Add a custom profile via the in-memory repo."""
    import cleveragents.cli.commands.automation_profile as ap_mod

    profile = _make_custom_profile(name=name)
    ap_mod._repo.upsert(profile)


# ---- When steps ---- #


@when("I run automation-profile add with --config pointing to the YAML file")
def step_run_add_config(context: Context) -> None:
    """Run the add command with --config."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.yaml_path]
    )


@when("I run automation-profile add with --config and --update")
def step_run_add_config_update(context: Context) -> None:
    """Run the add command with --config and --update."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.yaml_path, "--update"]
    )


@when("I run automation-profile add with --config without --update")
def step_run_add_config_no_update(context: Context) -> None:
    """Run the add command with --config but without --update."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.yaml_path]
    )


@when("I run automation-profile add with --config pointing to a missing file")
def step_run_add_missing_file(context: Context) -> None:
    """Run the add command with a missing config file."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", "/nonexistent/path/profile.yaml"]
    )


@when("I run automation-profile add with --config pointing to the invalid YAML file")
def step_run_add_invalid_yaml(context: Context) -> None:
    """Run the add command with invalid YAML."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.invalid_yaml_path]
    )


@when(
    "I run automation-profile add with --config pointing to the invalid name YAML file"
)
def step_run_add_invalid_name(context: Context) -> None:
    """Run the add command with invalid name."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.invalid_name_yaml_path]
    )


@when(
    "I run automation-profile add with --config pointing to the invalid threshold "
    "YAML file"
)
def step_run_add_invalid_threshold(context: Context) -> None:
    """Run the add command with invalid threshold."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.invalid_threshold_yaml_path]
    )


@when(
    "I run automation-profile add with --config pointing to the built-in name YAML file"
)
def step_run_add_builtin_name(context: Context) -> None:
    """Run the add command with a built-in name."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.builtin_name_yaml_path]
    )


@when(
    "I run automation-profile add with --config pointing to the unsupported "
    "schema YAML file"
)
def step_run_add_unsupported_schema(context: Context) -> None:
    """Run the add command with unsupported schema version."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.unsupported_schema_yaml_path]
    )


@when("I run automation-profile list")
def step_run_list(context: Context) -> None:
    """Run the list command."""
    context.result = context.runner.invoke(profile_app, ["list"])


@when('I run automation-profile list with namespace filter "{namespace}"')
def step_run_list_namespace(context: Context, namespace: str) -> None:
    """Run the list command with a namespace filter."""
    context.result = context.runner.invoke(
        profile_app, ["list", "--namespace", namespace]
    )


@when('I run automation-profile list with regex "{regex}"')
def step_run_list_regex(context: Context, regex: str) -> None:
    """Run the list command with a regex filter."""
    context.result = context.runner.invoke(profile_app, ["list", regex])


@when("I run automation-profile list with --format json")
def step_run_list_format_json(context: Context) -> None:
    """Run the list command with --format json."""
    context.result = context.runner.invoke(profile_app, ["list", "--format", "json"])


@when('I run automation-profile show "{name}"')
def step_run_show(context: Context, name: str) -> None:
    """Run the show command."""
    context.result = context.runner.invoke(profile_app, ["show", name])


@when('I run automation-profile show "{name}" with --format yaml')
def step_run_show_format_yaml(context: Context, name: str) -> None:
    """Run the show command with --format yaml."""
    context.result = context.runner.invoke(
        profile_app, ["show", name, "--format", "yaml"]
    )


@when('I run automation-profile show "{name}" with --format json')
def step_run_show_format_json(context: Context, name: str) -> None:
    """Run the show command with --format json."""
    context.result = context.runner.invoke(
        profile_app, ["show", name, "--format", "json"]
    )


@when('I run automation-profile show "{name}" with --format plain')
def step_run_show_format_plain(context: Context, name: str) -> None:
    """Run the show command with --format plain."""
    context.result = context.runner.invoke(
        profile_app, ["show", name, "--format", "plain"]
    )


@when('I run automation-profile show "{name}" with --format table')
def step_run_show_format_table(context: Context, name: str) -> None:
    """Run the show command with --format table."""
    context.result = context.runner.invoke(
        profile_app, ["show", name, "--format", "table"]
    )


@when('I run automation-profile remove "{name}" with --yes')
def step_run_remove_yes(context: Context, name: str) -> None:
    """Run the remove command with --yes."""
    context.result = context.runner.invoke(profile_app, ["remove", name, "--yes"])


@when('I run automation-profile remove "{name}" with --yes and --format json')
def step_run_remove_yes_json(context: Context, name: str) -> None:
    """Run the remove command with --yes and --format json."""
    context.result = context.runner.invoke(
        profile_app, ["remove", name, "--yes", "--format", "json"]
    )


@when('I invoke plan use with --automation-level "{level}"')
def step_invoke_plan_use_automation_level(context: Context, level: str) -> None:
    """Invoke plan use with the deprecated --automation-level flag."""
    from cleveragents.cli.commands.plan import app as plan_app

    # Mock the lifecycle service so we don't need real infra
    mock_service = MagicMock()
    mock_plan = MagicMock()
    mock_plan.identity.plan_id = "test-plan-id"
    mock_plan.namespaced_name = "test/plan"
    mock_plan.phase.value = "strategize"
    mock_plan.processing_state.value = "queued"
    mock_plan.state.value = "queued"
    mock_plan.project_links = []
    mock_plan.arguments = {}
    mock_plan.automation_profile = None
    mock_plan.action_name = "test/action"
    mock_plan.description = "test"
    mock_plan.definition_of_done = "test"
    mock_plan.strategy_actor = "test"
    mock_plan.execution_actor = "test"
    mock_plan.estimation_actor = None
    mock_plan.invariant_actor = None
    mock_plan.arguments_order = None
    mock_plan.timestamps.created_at.isoformat.return_value = "2024-01-01T00:00:00"
    mock_plan.timestamps.updated_at.isoformat.return_value = "2024-01-01T00:00:00"
    mock_plan.timestamps.strategize_started_at = None
    mock_plan.timestamps.strategize_completed_at = None
    mock_plan.timestamps.execute_started_at = None
    mock_plan.timestamps.execute_completed_at = None
    mock_plan.timestamps.applied_at = None
    mock_plan.error_message = None
    mock_plan.is_terminal = False
    mock_service.get_action_by_name.return_value = MagicMock()
    mock_service.get_action_by_name.return_value.namespaced_name = "test/action"
    mock_service.use_action.return_value = mock_plan

    # Widen the virtual terminal so Rich does not wrap text inside error
    # panels.  Without this, narrow terminals (e.g. CI containers with
    # capture_output=True) cause the error message to be split across
    # lines, breaking substring assertions.
    prev_columns = os.environ.get("COLUMNS")
    os.environ["COLUMNS"] = "200"
    try:
        with patch(
            "cleveragents.cli.commands.plan._get_lifecycle_service",
            return_value=mock_service,
        ):
            context.result = context.runner.invoke(
                plan_app,
                [
                    "use",
                    "test/action",
                    "--automation-level",
                    level,
                    "--format",
                    "json",
                ],
            )
    finally:
        if prev_columns is None:
            os.environ.pop("COLUMNS", None)
        else:
            os.environ["COLUMNS"] = prev_columns


# ---- Then steps ---- #


@then("the automation-profile add should succeed")
def step_add_should_succeed(context: Context) -> None:
    """Assert the add command succeeded."""
    assert context.result is not None
    assert context.result.exit_code == 0, (
        f"Expected exit code 0, got {context.result.exit_code}. "
        f"Output: {context.result.output}"
    )


@then('the automation-profile output should contain "{text}"')
def step_output_contains(context: Context, text: str) -> None:
    """Assert the command output contains specific text."""
    assert context.result is not None
    assert text.lower() in context.result.output.lower(), (
        f"Expected '{text}' in output. Got: {context.result.output}"
    )


@then("the automation-profile command should abort")
def step_command_should_abort(context: Context) -> None:
    """Assert the command aborted with non-zero exit code."""
    assert context.result is not None
    assert context.result.exit_code != 0, (
        f"Expected non-zero exit code, got {context.result.exit_code}. "
        f"Output: {context.result.output}"
    )


@then("the automation-profile list should succeed")
def step_list_should_succeed(context: Context) -> None:
    """Assert the list command succeeded."""
    assert context.result is not None
    assert context.result.exit_code == 0, (
        f"Expected exit code 0, got {context.result.exit_code}. "
        f"Output: {context.result.output}"
    )


@then('the automation-profile json output should contain "{text}"')
def step_json_output_contains(context: Context, text: str) -> None:
    """Assert the JSON output contains specific text."""
    assert context.result is not None
    assert text in context.result.output, (
        f"Expected '{text}' in JSON output. Got: {context.result.output}"
    )


@then('the automation-profile yaml output should contain "{text}"')
def step_yaml_output_contains(context: Context, text: str) -> None:
    """Assert the YAML output contains specific text."""
    assert context.result is not None
    assert text in context.result.output, (
        f"Expected '{text}' in YAML output. Got: {context.result.output}"
    )


@then("the automation-profile show should succeed")
def step_show_should_succeed(context: Context) -> None:
    """Assert the show command succeeded."""
    assert context.result is not None
    assert context.result.exit_code == 0, (
        f"Expected exit code 0, got {context.result.exit_code}. "
        f"Output: {context.result.output}"
    )


@then("the automation-profile remove should succeed")
def step_remove_should_succeed(context: Context) -> None:
    """Assert the remove command succeeded."""
    assert context.result is not None
    assert context.result.exit_code == 0, (
        f"Expected exit code 0, got {context.result.exit_code}. "
        f"Output: {context.result.output}"
    )


@then("the plan output should contain deprecation warning for automation-level")
def step_plan_output_deprecation_warning(context: Context) -> None:
    """Assert the plan output contains deprecation warning."""
    assert context.result is not None
    output = context.result.output.lower()
    assert "deprecated" in output, (
        f"Expected 'deprecated' in output. Got: {context.result.output}"
    )


@then('the plan output should contain "{text}"')
def step_plan_output_should_contain(context: Context, text: str) -> None:
    """Assert the plan output contains the given text."""
    assert context.result is not None
    output = _normalize_cli_output(context.result.output)
    assert text in output, f"Expected '{text}' in output. Got: {context.result.output}"