forked from cleveragents/cleveragents-core
CoreRasurae
66b9a4279f
Add SafetyProfile domain model with configurable safety constraints (checkpoint, human approval, cost limits, retry limits, skill categories, sandbox requirement). Integrate into Action model with from_config/as_cli_dict support, and persist via LifecycleActionModel safety_profile_json column. Include resolve_safety_profile() stub that raises NotImplementedError for local mode, signalling that real enforcement is deferred. Add comprehensive test coverage: - 20 BDD scenarios in features/safety_profile.feature - 6 Robot Framework smoke tests - 5 ASV benchmark suites (11 timing functions) Add docs/reference/safety_profile.md reference documentation. ISSUES CLOSED: #332
520 lines
18 KiB
Python
520 lines
18 KiB
Python
"""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
|
|
from cleveragents.domain.models.core.safety_profile import SafetyProfile
|
|
|
|
# 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
|
|
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"
|
|
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,
|
|
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
|
|
# 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}"
|