cleveragents-core/features/steps/automation_profile_cli_steps.py

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

from __future__ import annotations

import json
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 (
    app as profile_app,
)
from cleveragents.domain.models.core.automation_profile import AutomationProfile
from cleveragents.domain.models.core.safety_profile import SafetyProfile


def _create_in_memory_profile_service():
    """Create an AutomationProfileService backed by an in-memory SQLite DB."""
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker

    from cleveragents.application.services.automation_profile_service import (
        AutomationProfileService,
    )
    from cleveragents.infrastructure.database.models import Base
    from cleveragents.infrastructure.database.repositories import (
        AutomationProfileRepository,
    )

    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    repo = AutomationProfileRepository(session_factory=factory, auto_commit=True)
    return AutomationProfileService(repo=repo)


# 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"
decompose_task: 1.0
create_tool: 1.0
select_tool: 1.0
edit_code: 1.0
execute_command: 1.0
create_file: 1.0
delete_content: 1.0
access_network: 1.0
install_dependency: 1.0
modify_config: 1.0
approve_plan: 1.0
safety:
  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"
decompose_task: 2.0
"""

_LEGACY_THRESHOLD_YAML = """\
name: acme/legacy-thresholds
description: Uses pre-rename threshold keys
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
"""

_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",
        decompose_task=1.0,
        create_tool=1.0,
        select_tool=1.0,
        edit_code=1.0,
        execute_command=1.0,
        create_file=1.0,
        delete_content=1.0,
        access_network=1.0,
        install_dependency=1.0,
        modify_config=1.0,
        approve_plan=1.0,
        safety=SafetyProfile(
            require_sandbox=True,
            require_checkpoints=True,
            allow_unsafe_tools=False,
            require_human_approval=False,
            max_retries_per_step=3,
        ),
    )


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
    # Widen the virtual terminal so Rich does not truncate profile names
    # in the table output.  The wider column headers introduced by the
    # task-type field rename (e.g. "Create Tool") need extra room.
    prev_columns = os.environ.get("COLUMNS")
    os.environ["COLUMNS"] = "200"
    # Create an in-memory service and patch _get_service for this scenario
    context._ap_service = _create_in_memory_profile_service()
    context._ap_patcher = patch(
        "cleveragents.cli.commands.automation_profile._get_service",
        return_value=context._ap_service,
    )
    context._ap_patcher.start()
    if not hasattr(context, "_cleanup_handlers"):
        context._cleanup_handlers = []
    context._cleanup_handlers.append(context._ap_patcher.stop)

    def _restore_columns() -> None:
        if prev_columns is None:
            os.environ.pop("COLUMNS", None)
        else:
            os.environ["COLUMNS"] = prev_columns

    context._cleanup_handlers.append(_restore_columns)


@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."""
    profile = _make_custom_profile(name=name)
    context._ap_service._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 legacy threshold keys")
def step_legacy_threshold_config(context: Context) -> None:
    """Write a profile config with legacy pre-rename threshold keys."""
    context.legacy_threshold_yaml_path = _write_temp_yaml(
        context, _LEGACY_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"
decompose_task: 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."""
    profile = _make_custom_profile(name=name)
    context._ap_service._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 legacy threshold "
    "YAML file"
)
def step_run_add_legacy_threshold(context: Context) -> None:
    """Run the add command with legacy threshold keys."""
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.legacy_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 list with --format yaml")
def step_run_list_format_yaml(context: Context) -> None:
    """Run the list command with --format yaml."""
    context.result = context.runner.invoke(profile_app, ["list", "--format", "yaml"])


@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}"


@then("the automation-profile list json output has profiles wrapper with summary")
def step_list_json_has_profiles_wrapper_with_summary(context: Context) -> None:
    """Assert the list JSON output has the spec-required profiles wrapper and summary.

    Per docs/specification.md lines 16998-17017, the JSON output must be a dict
    with a ``profiles`` key (list of simplified profile entries) and a ``summary``
    key with ``built_in``, ``custom``, and ``total`` counts.  Each profile entry
    must contain only: name, source, select_tool, sandbox, description.
    """
    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}"
    )
    # Extract JSON from output (may have leading log lines)
    output = context.result.output
    parsed: object = None
    for i, ch in enumerate(output):
        if ch == "{":
            try:
                parsed = json.loads(output[i:])
                break
            except json.JSONDecodeError:
                continue
    assert parsed is not None, f"No valid JSON found in output:\n{output[:500]}"
    assert isinstance(parsed, dict), f"Expected dict at top level, got {type(parsed)}"

    # Validate profiles key
    assert "profiles" in parsed, (
        f"Missing 'profiles' key in JSON output. Keys: {list(parsed.keys())}"
    )
    profiles = parsed["profiles"]
    assert isinstance(profiles, list), (
        f"Expected 'profiles' to be a list, got {type(profiles)}"
    )
    assert len(profiles) >= 8, (
        f"Expected at least 8 built-in profiles, got {len(profiles)}"
    )

    # Validate each profile entry has only the spec-required fields
    required_fields = {"name", "source", "select_tool", "sandbox", "description"}
    for entry in profiles:
        assert isinstance(entry, dict), f"Profile entry is not a dict: {entry}"
        assert required_fields.issubset(entry.keys()), (
            f"Profile entry missing required fields. "
            f"Expected {required_fields}, got {set(entry.keys())}"
        )
        # Ensure no full profile dict fields leak through
        assert "phase_transitions" not in entry, (
            "Profile entry must not contain 'phase_transitions' (full profile dict leaked)"
        )
        assert "decompose_task" not in entry, (
            "Profile entry must not contain 'decompose_task' (full profile dict leaked)"
        )

    # Validate summary key
    assert "summary" in parsed, (
        f"Missing 'summary' key in JSON output. Keys: {list(parsed.keys())}"
    )
    summary = parsed["summary"]
    assert isinstance(summary, dict), (
        f"Expected 'summary' to be a dict, got {type(summary)}"
    )
    assert "built_in" in summary, f"Missing 'built_in' in summary: {summary}"
    assert "custom" in summary, f"Missing 'custom' in summary: {summary}"
    assert "total" in summary, f"Missing 'total' in summary: {summary}"
    assert summary["built_in"] >= 8, (
        f"Expected at least 8 built-in profiles in summary, got {summary['built_in']}"
    )
    assert summary["total"] == summary["built_in"] + summary["custom"], (
        f"summary.total ({summary['total']}) != built_in ({summary['built_in']}) "
        f"+ custom ({summary['custom']})"
    )