"""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]}" )