forked from HAL9000/cleveragents-core
freemo
a808c395f9
Add 53 new .feature files and corresponding step definition files targeting uncovered lines identified in build/coverage.xml. Fix AmbiguousStep conflicts in 7 pre-existing step files by disambiguating step text. New tests cover: ACP clients/facade, actor CLI/config, application container, ACMS service/strategies, async worker, automation profile CLI, autonomy guardrail, bridge, change model, config CLI/service, context service, cross-plan correction, database models, decision service, decomposition clustering/service, discovery handler, langchain chat provider, langgraph nodes, materializers, multi-project service, plan apply/CLI/lifecycle/model/ preflight/resume/service, PostgreSQL analyzer, project CLI/context CLI, provider registry, reactive application/route, repositories, resolver handler, resource registry service, resume model, retry patterns, sandbox protocol, server CLI, skill CLI/service, skills registry, subplan execution/service, system CLI, UKO loader, UoW, and YAML template engine. Closes #645
345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""Step definitions for Automation Profile CLI guards coverage boost.
|
|
|
|
Targets uncovered lines 91-97 (_guards_dict non-None branch) and
|
|
lines 194-204 (_print_profile rich guards display block).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.automation_profile import (
|
|
_guards_dict,
|
|
_InMemoryProfileRepository,
|
|
_profile_spec_dict,
|
|
)
|
|
from cleveragents.cli.commands.automation_profile import (
|
|
app as profile_app,
|
|
)
|
|
from cleveragents.domain.models.core.automation_guard import AutomationGuard
|
|
from cleveragents.domain.models.core.automation_profile import AutomationProfile
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_guarded_profile(
|
|
name: str = "acme/guarded",
|
|
description: str = "Profile with guards",
|
|
max_tool_calls: int = 5,
|
|
max_total_cost: float = 100.0,
|
|
tool_allowlist: list[str] | None = None,
|
|
tool_denylist: list[str] | None = None,
|
|
require_approval_for_writes: bool = True,
|
|
require_approval_for_apply: bool = False,
|
|
) -> AutomationProfile:
|
|
"""Create an AutomationProfile with guards attached."""
|
|
guards = AutomationGuard(
|
|
max_tool_calls_per_step=max_tool_calls,
|
|
max_total_cost=max_total_cost,
|
|
tool_allowlist=tool_allowlist,
|
|
tool_denylist=tool_denylist,
|
|
require_approval_for_writes=require_approval_for_writes,
|
|
require_approval_for_apply=require_approval_for_apply,
|
|
)
|
|
return AutomationProfile(
|
|
name=name,
|
|
description=description,
|
|
schema_version="1.0",
|
|
auto_strategize=0.5,
|
|
auto_execute=0.4,
|
|
auto_apply=0.3,
|
|
guards=guards,
|
|
)
|
|
|
|
|
|
def _write_temp_yaml(context: Context, content: str) -> str:
|
|
"""Write content to a temporary YAML file, tracked for cleanup."""
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a fresh automation profile CLI runner for guards coverage")
|
|
def step_fresh_guards_runner(context: Context) -> None:
|
|
context.runner = CliRunner()
|
|
context.result = None
|
|
context.guards_dict_result: dict[str, object] | None = None
|
|
context.profile_spec_result: dict[str, object] | None = None
|
|
context.guard_obj: AutomationGuard | None = None
|
|
context.guarded_profile: AutomationProfile | None = None
|
|
# Reset the module-level in-memory repo
|
|
import cleveragents.cli.commands.automation_profile as ap_mod
|
|
|
|
ap_mod._repo = _InMemoryProfileRepository()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _guards_dict direct tests (lines 91-97)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"an AutomationGuard with max_tool_calls_per_step {calls:d} and max_total_cost {cost:g}"
|
|
)
|
|
def step_create_guard_with_calls_and_cost(
|
|
context: Context,
|
|
calls: int,
|
|
cost: float,
|
|
) -> None:
|
|
context.guard_obj = AutomationGuard(
|
|
max_tool_calls_per_step=calls,
|
|
max_total_cost=cost,
|
|
tool_allowlist=None,
|
|
tool_denylist=None,
|
|
require_approval_for_writes=False,
|
|
require_approval_for_apply=False,
|
|
)
|
|
|
|
|
|
@given(
|
|
'an AutomationGuard with tool_allowlist "{allowlist}" and tool_denylist "{denylist}"'
|
|
)
|
|
def step_create_guard_with_lists(
|
|
context: Context,
|
|
allowlist: str,
|
|
denylist: str,
|
|
) -> None:
|
|
context.guard_obj = AutomationGuard(
|
|
max_tool_calls_per_step=None,
|
|
max_total_cost=None,
|
|
tool_allowlist=allowlist.split(","),
|
|
tool_denylist=denylist.split(","),
|
|
require_approval_for_writes=False,
|
|
require_approval_for_apply=False,
|
|
)
|
|
|
|
|
|
@when("I call _guards_dict with the guard object")
|
|
def step_call_guards_dict(context: Context) -> None:
|
|
context.guards_dict_result = _guards_dict(context.guard_obj)
|
|
|
|
|
|
@when("I call _guards_dict with None")
|
|
def step_call_guards_dict_none(context: Context) -> None:
|
|
context.guards_dict_result = _guards_dict(None)
|
|
|
|
|
|
@then('the guards dict should contain key "{key}" with value {value:d}')
|
|
def step_guards_dict_int_value(context: Context, key: str, value: int) -> None:
|
|
result = context.guards_dict_result
|
|
assert result is not None, "Expected non-None result from _guards_dict"
|
|
assert key in result, f"Key '{key}' not found in guards dict: {result}"
|
|
assert result[key] == value, f"Expected {key}={value}, got {result[key]}"
|
|
|
|
|
|
@then('the guards dict should contain key "{key}" with value {value:g}')
|
|
def step_guards_dict_float_value(context: Context, key: str, value: float) -> None:
|
|
result = context.guards_dict_result
|
|
assert result is not None, "Expected non-None result from _guards_dict"
|
|
assert key in result, f"Key '{key}' not found in guards dict: {result}"
|
|
assert result[key] == value, f"Expected {key}={value}, got {result[key]}"
|
|
|
|
|
|
@then('the guards dict should contain key "{key}"')
|
|
def step_guards_dict_has_key(context: Context, key: str) -> None:
|
|
result = context.guards_dict_result
|
|
assert result is not None, "Expected non-None result from _guards_dict"
|
|
assert key in result, f"Key '{key}' not found in guards dict: {result}"
|
|
|
|
|
|
@then("the guards dict result should be None")
|
|
def step_guards_dict_is_none(context: Context) -> None:
|
|
assert context.guards_dict_result is None, (
|
|
f"Expected None, got {context.guards_dict_result}"
|
|
)
|
|
|
|
|
|
@then('the guards dict tool_allowlist should be "{expected}"')
|
|
def step_guards_dict_allowlist(context: Context, expected: str) -> None:
|
|
result = context.guards_dict_result
|
|
assert result is not None
|
|
expected_list = expected.split(",")
|
|
assert result["tool_allowlist"] == expected_list, (
|
|
f"Expected {expected_list}, got {result['tool_allowlist']}"
|
|
)
|
|
|
|
|
|
@then('the guards dict tool_denylist should be "{expected}"')
|
|
def step_guards_dict_denylist(context: Context, expected: str) -> None:
|
|
result = context.guards_dict_result
|
|
assert result is not None
|
|
expected_list = expected.split(",")
|
|
assert result["tool_denylist"] == expected_list, (
|
|
f"Expected {expected_list}, got {result['tool_denylist']}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _profile_spec_dict with guards (lines 91-97 indirect)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an AutomationProfile with guards having require_approval_for_writes true")
|
|
def step_create_profile_with_guards(context: Context) -> None:
|
|
context.guarded_profile = _make_guarded_profile(
|
|
name="acme/spectest",
|
|
require_approval_for_writes=True,
|
|
)
|
|
|
|
|
|
@when("I call _profile_spec_dict with the guarded profile")
|
|
def step_call_profile_spec_dict(context: Context) -> None:
|
|
context.profile_spec_result = _profile_spec_dict(context.guarded_profile)
|
|
|
|
|
|
@then("the profile spec dict guards should not be None")
|
|
def step_spec_dict_guards_not_none(context: Context) -> None:
|
|
result = context.profile_spec_result
|
|
assert result is not None
|
|
assert result.get("guards") is not None, (
|
|
f"Expected guards to be non-None, got: {result.get('guards')}"
|
|
)
|
|
|
|
|
|
@then('the profile spec dict guards should have "{key}" as true')
|
|
def step_spec_dict_guards_key_true(context: Context, key: str) -> None:
|
|
result = context.profile_spec_result
|
|
assert result is not None
|
|
guards = result["guards"]
|
|
assert guards is not None
|
|
assert guards[key] is True, f"Expected {key}=True, got {guards[key]}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI show with guards - rich format (lines 194-204)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a custom guarded profile "{name}" is stored in the repo')
|
|
def step_store_guarded_profile(context: Context, name: str) -> None:
|
|
import cleveragents.cli.commands.automation_profile as ap_mod
|
|
|
|
profile = _make_guarded_profile(
|
|
name=name,
|
|
max_tool_calls=5,
|
|
max_total_cost=100.0,
|
|
tool_allowlist=["read_file", "search"],
|
|
tool_denylist=["rm_rf"],
|
|
require_approval_for_writes=True,
|
|
require_approval_for_apply=True,
|
|
)
|
|
ap_mod._repo.upsert(profile)
|
|
|
|
|
|
@when('I run automation-profile show "{name}" in rich format')
|
|
def step_show_guarded_rich(context: Context, name: str) -> None:
|
|
context.result = context.runner.invoke(profile_app, ["show", name])
|
|
|
|
|
|
@when('I run automation-profile show "{name}" with format "{fmt}"')
|
|
def step_show_guarded_fmt(context: Context, name: str, fmt: str) -> None:
|
|
context.result = context.runner.invoke(
|
|
profile_app,
|
|
["show", name, "--format", fmt],
|
|
)
|
|
|
|
|
|
@then("the guards coverage show command should succeed")
|
|
def step_guards_coverage_show_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}"
|
|
)
|
|
|
|
|
|
@then('the guards coverage output should contain "{text}"')
|
|
def step_guards_coverage_output_contains(context: Context, text: str) -> None:
|
|
assert context.result is not None
|
|
assert text in context.result.output, (
|
|
f"Expected '{text}' in output. Got:\n{context.result.output}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI add with guards config (lines 194-204 via add path)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a YAML config file for a guarded profile "{name}"')
|
|
def step_guarded_yaml_config(context: Context, name: str) -> None:
|
|
yaml_content = f"""\
|
|
name: {name}
|
|
description: Guarded profile for testing
|
|
schema_version: "1.0"
|
|
auto_strategize: 0.5
|
|
auto_execute: 0.4
|
|
auto_apply: 0.3
|
|
guards:
|
|
max_tool_calls_per_step: 8
|
|
max_total_cost: 200.0
|
|
tool_allowlist:
|
|
- read_file
|
|
- search
|
|
tool_denylist:
|
|
- rm_rf
|
|
require_approval_for_writes: true
|
|
require_approval_for_apply: false
|
|
"""
|
|
context.guarded_yaml_path = _write_temp_yaml(context, yaml_content)
|
|
|
|
|
|
@when("I run automation-profile add with the guarded config file")
|
|
def step_add_guarded_config(context: Context) -> None:
|
|
context.result = context.runner.invoke(
|
|
profile_app,
|
|
["add", "--config", context.guarded_yaml_path],
|
|
)
|
|
|
|
|
|
@then("the guards coverage add command should succeed")
|
|
def step_guards_coverage_add_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}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI remove guarded profile with --format json (lines 91-97 via remove)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I run automation-profile remove "{name}" with --yes and format "{fmt}"')
|
|
def step_remove_guarded_fmt(context: Context, name: str, fmt: str) -> None:
|
|
context.result = context.runner.invoke(
|
|
profile_app,
|
|
["remove", name, "--yes", "--format", fmt],
|
|
)
|
|
|
|
|
|
@then("the guards coverage remove command should succeed")
|
|
def step_guards_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}"
|
|
)
|