temp/features/steps/automation_profile_cli_coverage_steps.py

"""Step definitions for Automation Profile CLI coverage boost."""

from __future__ import annotations

import os
import tempfile
import warnings
from pathlib import Path
from unittest.mock import patch

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

from cleveragents.cli.commands.automation_profile import (
    _threshold_summary,
    emit_automation_level_deprecation_warning,
)
from cleveragents.cli.commands.automation_profile import (
    app as profile_app,
)
from cleveragents.core.exceptions import (
    CleverAgentsError,
    ValidationError,
)
from cleveragents.domain.models.core.automation_profile import (
    BUILTIN_PROFILES,
    AutomationProfile,
)
from cleveragents.infrastructure.database.repositories import (
    AutomationProfileNotFoundError,
)


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)


def _make_profile(
    name: str = "acme/strict",
    description: str = "Strict review",
) -> AutomationProfile:
    return AutomationProfile(
        name=name,
        description=description,
        schema_version="1.0",
        decompose_task=0.8,
        create_tool=0.7,
        select_tool=0.6,
    )


def _write_temp(context: Context, content: str) -> str:
    fd, path = tempfile.mkstemp(suffix=".yaml")
    with os.fdopen(fd, "w") as fh:
        fh.write(content)
    if not hasattr(context, "_temp_files"):
        context._temp_files = []
    context._temp_files.append(path)
    return path


@given("an automation profile coverage CLI runner")
def step_coverage_runner(context: Context) -> None:
    context.runner = CliRunner()
    context.result = None
    # 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)


@when("I delete a non-existent profile from the in-memory repo")
def step_delete_nonexistent_from_repo(context: Context) -> None:
    context.repo_error = None
    try:
        context._ap_service._repo.delete("nonexistent/profile")
    except AutomationProfileNotFoundError as exc:
        context.repo_error = exc


@then("a NotFoundError should be raised from the repo")
def step_assert_not_found_error(context: Context) -> None:
    assert context.repo_error is not None, "Expected NotFoundError but none was raised"
    assert isinstance(context.repo_error, AutomationProfileNotFoundError)


@when("I call threshold summary on a built-in profile")
def step_call_threshold_summary(context: Context) -> None:
    profile = BUILTIN_PROFILES["manual"]
    context.threshold_result = _threshold_summary(profile)


@then("the threshold summary should contain strategize execute and apply values")
def step_assert_threshold_summary(context: Context) -> None:
    result = context.threshold_result
    assert "decompose_task=" in result
    assert "create_tool=" in result
    assert "select_tool=" in result


@when("I validate a guarded profile config against docs schema and model")
def step_validate_guarded_profile_schema_and_model(context: Context) -> None:
    schema_path = Path("docs/schema/automation_profile.schema.yaml")
    with open(schema_path) as fh:
        schema = yaml.safe_load(fh)

    config = {
        "name": "acme/guarded",
        "guards": {
            "max_tool_calls_per_step": 5,
            "max_total_cost": 25.0,
            "tool_allowlist": ["read_file"],
            "tool_denylist": ["shell_exec"],
            "require_approval_for_writes": True,
            "require_approval_for_apply": True,
        },
    }

    context.guarded_profile_schema_error = None
    context.guarded_profile_model_error = None
    context.guarded_profile_model = None

    try:
        jsonschema.validate(instance=config, schema=schema)
    except jsonschema.ValidationError as exc:
        context.guarded_profile_schema_error = exc

    try:
        context.guarded_profile_model = AutomationProfile.from_config(config)
    except Exception as exc:  # pragma: no cover - assertion reports details
        context.guarded_profile_model_error = exc


@then("the guarded profile config should pass schema and model validation")
def step_guarded_profile_schema_and_model_ok(context: Context) -> None:
    assert context.guarded_profile_schema_error is None, (
        "Expected docs schema validation to pass, got: "
        f"{context.guarded_profile_schema_error}"
    )
    assert context.guarded_profile_model_error is None, (
        "Expected AutomationProfile.from_config validation to pass, got: "
        f"{context.guarded_profile_model_error}"
    )
    assert context.guarded_profile_model is not None
    assert context.guarded_profile_model.guards is not None
    assert context.guarded_profile_model.guards.max_tool_calls_per_step == 5


@given("a YAML file containing a list instead of a dict")
def step_non_dict_yaml(context: Context) -> None:
    context.non_dict_yaml_path = _write_temp(context, "- item1\n- item2\n")


@when("I run automation-profile add with that non-dict YAML file")
def step_run_add_non_dict(context: Context) -> None:
    context.result = context.runner.invoke(
        profile_app, ["add", "--config", context.non_dict_yaml_path]
    )


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


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


@when("I run automation-profile add with a config that triggers FileNotFoundError")
def step_run_add_file_not_found(context: Context) -> None:
    # Create a file that exists at check time but causes FileNotFoundError on open
    path = _write_temp(
        context,
        "name: acme/test\ndescription: test\nschema_version: '1.0'\n",
    )
    original_open = open

    def patched_open(p, *args, **kwargs):
        if str(p) == path:
            raise FileNotFoundError(f"File vanished: {p}")
        return original_open(p, *args, **kwargs)

    with patch("builtins.open", side_effect=patched_open):
        context.result = context.runner.invoke(profile_app, ["add", "--config", path])


@given("a YAML file that triggers a ValidationError from the service")
def step_yaml_validation_error(context: Context) -> None:
    context.validation_yaml_path = _write_temp(
        context,
        "name: acme/valerr\ndescription: test\nschema_version: '1.0'\n",
    )


@when("I run automation-profile add with that validation-error YAML file")
def step_run_add_validation_error(context: Context) -> None:
    with patch(
        "cleveragents.cli.commands.automation_profile.AutomationProfileService.create_profile",
        side_effect=ValidationError("Invalid profile data"),
    ):
        context.result = context.runner.invoke(
            profile_app, ["add", "--config", context.validation_yaml_path]
        )


@given("a YAML file that triggers a CleverAgentsError from the service")
def step_yaml_ca_error(context: Context) -> None:
    context.ca_error_yaml_path = _write_temp(
        context,
        "name: acme/caerr\ndescription: test\nschema_version: '1.0'\n",
    )


@when("I run automation-profile add with that cleveragents-error YAML file")
def step_run_add_ca_error(context: Context) -> None:
    with patch(
        "cleveragents.cli.commands.automation_profile.AutomationProfileService.create_profile",
        side_effect=CleverAgentsError("Something went wrong"),
    ):
        context.result = context.runner.invoke(
            profile_app, ["add", "--config", context.ca_error_yaml_path]
        )


@given('a custom coverage profile "{name}" has been added')
def step_add_custom_coverage_profile(context: Context, name: str) -> None:
    profile = _make_profile(name=name)
    context._ap_service._repo.upsert(profile)


@when('I run automation-profile remove "{name}" without --yes and decline')
def step_remove_no_yes_decline(context: Context, name: str) -> None:
    context.result = context.runner.invoke(profile_app, ["remove", name], input="n\n")


@when('I run automation-profile remove "{name}" without --yes and confirm')
def step_remove_no_yes_confirm(context: Context, name: str) -> None:
    context.result = context.runner.invoke(profile_app, ["remove", name], input="y\n")


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


@when('I run automation-profile remove "{name}" with validation error')
def step_remove_validation_error(context: Context, name: str) -> None:
    with patch(
        "cleveragents.cli.commands.automation_profile.AutomationProfileService.delete_profile",
        side_effect=ValidationError("Cannot delete"),
    ):
        context.result = context.runner.invoke(profile_app, ["remove", name, "--yes"])


@when('I run automation-profile remove "{name}" with CleverAgentsError')
def step_remove_ca_error(context: Context, name: str) -> None:
    with patch(
        "cleveragents.cli.commands.automation_profile.AutomationProfileService.delete_profile",
        side_effect=CleverAgentsError("Delete failed"),
    ):
        context.result = context.runner.invoke(profile_app, ["remove", name, "--yes"])


@when("I run automation-profile list with a CleverAgentsError from service")
def step_list_ca_error(context: Context) -> None:
    with patch(
        "cleveragents.cli.commands.automation_profile.AutomationProfileService.list_profiles",
        side_effect=CleverAgentsError("List failed"),
    ):
        context.result = context.runner.invoke(profile_app, ["list"])


@when("I run automation-profile show with a CleverAgentsError from service")
def step_show_ca_error(context: Context) -> None:
    with patch(
        "cleveragents.cli.commands.automation_profile.AutomationProfileService.get_profile",
        side_effect=CleverAgentsError("Show failed"),
    ):
        context.result = context.runner.invoke(profile_app, ["show", "manual"])


@when("I call emit_automation_level_deprecation_warning")
def step_call_deprecation_warning(context: Context) -> None:
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        emit_automation_level_deprecation_warning()
        context.warnings = w


@then("a DeprecationWarning should be emitted")
def step_assert_deprecation_warning(context: Context) -> None:
    assert len(context.warnings) >= 1, "Expected at least one warning"
    assert any(issubclass(w.category, DeprecationWarning) for w in context.warnings), (
        f"Expected DeprecationWarning, got: {[w.category for w in context.warnings]}"
    )