forked from HAL9000/cleveragents-core
aditya
2764fcef5c
Address review-driven fixes across actor schema, preflight guardrails, docs/examples, and Behave/Robot coverage: unify preflight warning behavior with shared role-warning logic, resolve actor-name to config payloads in production preflight flow, harden response_format validation/coercion edge cases, extract duplicated helper logic, and expand negative-path test coverage. Also fix cross-scenario patcher leakage in step modules to eliminate full-run-only coverage failures.
364 lines
13 KiB
Python
364 lines
13 KiB
Python
"""Step definitions for Plan Generation Pre-flight Guardrails scenarios.
|
|
|
|
All step patterns are prefixed with ``pfg-`` to avoid AmbiguousStep
|
|
collisions with other feature step definitions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.actor.schema import ActorConfigSchema
|
|
from cleveragents.application.services.plan_preflight_guardrail import (
|
|
PlanPreflightGuardrail,
|
|
PreflightRejection,
|
|
)
|
|
|
|
# -- Givens ----------------------------------------------------------------
|
|
|
|
|
|
@given("a plan preflight guardrail")
|
|
def step_pfg_given_guardrail(context: Context) -> None:
|
|
context.guardrail = PlanPreflightGuardrail()
|
|
# Reset state between scenarios
|
|
context.pfg_action_registry = {}
|
|
context.pfg_actor_registry = {}
|
|
context.pfg_tool_registry = {}
|
|
context.pfg_skill_registry = {}
|
|
context.pfg_automation_profile = None
|
|
context.pfg_require_checkpoints = False
|
|
context.pfg_tool_capabilities = {}
|
|
context.pfg_resource_paths = ()
|
|
context.pfg_validation_registry = {}
|
|
context.pfg_check_result = None
|
|
context.pfg_report = None
|
|
context.pfg_rejection = None
|
|
|
|
|
|
@given('an action registry containing "{name}"')
|
|
def step_pfg_action_registry_with(context: Context, name: str) -> None:
|
|
context.pfg_action_registry[name] = {"name": name}
|
|
|
|
|
|
@given("an empty action registry")
|
|
def step_pfg_empty_action_registry(context: Context) -> None:
|
|
context.pfg_action_registry = {}
|
|
|
|
|
|
@given("an actor registry with all 4 roles registered")
|
|
def step_pfg_actor_all_roles(context: Context) -> None:
|
|
for role in PlanPreflightGuardrail.ACTOR_ROLES:
|
|
context.pfg_actor_registry[role] = {"role": role}
|
|
|
|
|
|
@given('an actor registry missing "{role}" role')
|
|
def step_pfg_actor_missing_role(context: Context, role: str) -> None:
|
|
for r in PlanPreflightGuardrail.ACTOR_ROLES:
|
|
if r != role:
|
|
context.pfg_actor_registry[r] = {"role": r}
|
|
|
|
|
|
@given('a tool registry containing "{t1}" and "{t2}"')
|
|
def step_pfg_tool_registry_two(context: Context, t1: str, t2: str) -> None:
|
|
context.pfg_tool_registry[t1] = {"name": t1}
|
|
context.pfg_tool_registry[t2] = {"name": t2}
|
|
|
|
|
|
@given('a tool registry containing "{name}"')
|
|
def step_pfg_tool_registry_one(context: Context, name: str) -> None:
|
|
context.pfg_tool_registry[name] = {"name": name}
|
|
|
|
|
|
@given("a valid automation profile")
|
|
def step_pfg_valid_automation(context: Context) -> None:
|
|
context.pfg_automation_profile = {"name": "auto"}
|
|
|
|
|
|
@given("no automation profile")
|
|
def step_pfg_no_automation(context: Context) -> None:
|
|
context.pfg_automation_profile = None
|
|
|
|
|
|
@given("require_checkpoints is true")
|
|
def step_pfg_require_checkpoints_true(context: Context) -> None:
|
|
context.pfg_require_checkpoints = True
|
|
|
|
|
|
@given("require_checkpoints is false")
|
|
def step_pfg_require_checkpoints_false(context: Context) -> None:
|
|
context.pfg_require_checkpoints = False
|
|
|
|
|
|
@given("tool capabilities showing all tools checkpointable")
|
|
def step_pfg_all_checkpointable(context: Context) -> None:
|
|
context.pfg_tool_capabilities = {"tool-a": True, "tool-b": True}
|
|
|
|
|
|
@given('tool capabilities with "{name}" not checkpointable')
|
|
def step_pfg_non_checkpointable(context: Context, name: str) -> None:
|
|
context.pfg_tool_capabilities = {"safe-tool": True, name: False}
|
|
|
|
|
|
@given("resource paths that exist on filesystem")
|
|
def step_pfg_existing_resources(context: Context) -> None:
|
|
# Use a temp file to guarantee it exists
|
|
with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as tmp:
|
|
tmp_path = tmp.name
|
|
context.pfg_resource_paths = (tmp_path,)
|
|
context._pfg_tmp_file = tmp_path
|
|
|
|
|
|
@given('resource path "{path}"')
|
|
def step_pfg_resource_path(context: Context, path: str) -> None:
|
|
context.pfg_resource_paths = (path,)
|
|
|
|
|
|
@given('a validation registry containing "{v1}" and "{v2}"')
|
|
def step_pfg_validation_registry_two(context: Context, v1: str, v2: str) -> None:
|
|
context.pfg_validation_registry[v1] = {"name": v1}
|
|
context.pfg_validation_registry[v2] = {"name": v2}
|
|
|
|
|
|
@given('a validation registry containing "{name}"')
|
|
def step_pfg_validation_registry_one(context: Context, name: str) -> None:
|
|
context.pfg_validation_registry[name] = {"name": name}
|
|
|
|
|
|
@given("a plan preflight guardrail with all registries populated")
|
|
def step_pfg_all_populated(context: Context) -> None:
|
|
context.guardrail = PlanPreflightGuardrail()
|
|
context.pfg_action_registry = {"test-action": {"name": "test-action"}}
|
|
context.pfg_actor_registry = {
|
|
role: {"role": role} for role in PlanPreflightGuardrail.ACTOR_ROLES
|
|
}
|
|
context.pfg_tool_registry = {"tool-a": {"name": "tool-a"}}
|
|
context.pfg_skill_registry = {}
|
|
context.pfg_automation_profile = {"name": "auto"}
|
|
context.pfg_require_checkpoints = False
|
|
context.pfg_tool_capabilities = {}
|
|
context.pfg_resource_paths = ()
|
|
context.pfg_validation_registry = {}
|
|
context.pfg_rejection = None
|
|
context.pfg_report = None
|
|
|
|
|
|
@given("an estimation actor registry entry without response_format")
|
|
def step_pfg_estimation_actor_missing_response_format(context: Context) -> None:
|
|
context.pfg_actor_registry["estimation"] = {
|
|
"name": "local/estimator",
|
|
"context_view": "strategist",
|
|
}
|
|
|
|
|
|
@given("an estimation actor registry entry with non-strategist context_view")
|
|
def step_pfg_estimation_actor_wrong_context_view(context: Context) -> None:
|
|
context.pfg_actor_registry["estimation"] = {
|
|
"name": "local/estimator",
|
|
"context_view": "executor",
|
|
"response_format": {"type": "object"},
|
|
}
|
|
|
|
|
|
@given(
|
|
"an estimation actor registry entry as ActorConfigSchema without response_format"
|
|
)
|
|
def step_pfg_estimation_actor_model_missing_response_format(context: Context) -> None:
|
|
context.pfg_actor_registry["estimation"] = ActorConfigSchema(
|
|
name="local/estimator-model",
|
|
type="llm",
|
|
description="Model-based estimation actor",
|
|
model="gpt-4",
|
|
role_hint="estimation",
|
|
context_view="strategist",
|
|
)
|
|
|
|
|
|
@given("a plan preflight guardrail with missing action")
|
|
def step_pfg_missing_action(context: Context) -> None:
|
|
context.guardrail = PlanPreflightGuardrail()
|
|
context.pfg_action_registry = {}
|
|
context.pfg_actor_registry = {
|
|
role: {"role": role} for role in PlanPreflightGuardrail.ACTOR_ROLES
|
|
}
|
|
context.pfg_tool_registry = {}
|
|
context.pfg_skill_registry = {}
|
|
context.pfg_automation_profile = {"name": "auto"}
|
|
context.pfg_require_checkpoints = False
|
|
context.pfg_tool_capabilities = {}
|
|
context.pfg_resource_paths = ()
|
|
context.pfg_validation_registry = {}
|
|
context.pfg_rejection = None
|
|
context.pfg_report = None
|
|
|
|
|
|
# -- Whens ----------------------------------------------------------------
|
|
|
|
|
|
@when('I check action schema for "{name}"')
|
|
def step_pfg_check_action(context: Context, name: str) -> None:
|
|
context.pfg_check_result = context.guardrail.check_action_schema(
|
|
name, context.pfg_action_registry
|
|
)
|
|
|
|
|
|
@when("I check actor availability")
|
|
def step_pfg_check_actors(context: Context) -> None:
|
|
context.pfg_check_result = context.guardrail.check_actor_availability(
|
|
context.pfg_actor_registry
|
|
)
|
|
|
|
|
|
@when('I check skill/tool existence for tools "{tools_csv}"')
|
|
def step_pfg_check_tools(context: Context, tools_csv: str) -> None:
|
|
tool_names = tuple(t.strip() for t in tools_csv.split(","))
|
|
context.pfg_check_result = context.guardrail.check_skill_tool_existence(
|
|
tool_names,
|
|
context.pfg_tool_registry,
|
|
(),
|
|
context.pfg_skill_registry,
|
|
)
|
|
|
|
|
|
@when("I check automation policy")
|
|
def step_pfg_check_automation(context: Context) -> None:
|
|
context.pfg_check_result = context.guardrail.check_automation_policy(
|
|
context.pfg_automation_profile
|
|
)
|
|
|
|
|
|
@when("I check rollback feasibility")
|
|
def step_pfg_check_rollback(context: Context) -> None:
|
|
context.pfg_check_result = context.guardrail.check_rollback_feasibility(
|
|
context.pfg_require_checkpoints,
|
|
context.pfg_tool_capabilities,
|
|
)
|
|
|
|
|
|
@when("I check resource accessibility")
|
|
def step_pfg_check_resources(context: Context) -> None:
|
|
context.pfg_check_result = context.guardrail.check_resource_accessibility(
|
|
context.pfg_resource_paths
|
|
)
|
|
|
|
|
|
@when('I check validation resolution for "{names_csv}"')
|
|
def step_pfg_check_validation(context: Context, names_csv: str) -> None:
|
|
validation_names = tuple(v.strip() for v in names_csv.split(","))
|
|
context.pfg_check_result = context.guardrail.check_validation_resolution(
|
|
validation_names,
|
|
context.pfg_validation_registry,
|
|
)
|
|
|
|
|
|
@when("I run all preflight checks")
|
|
def step_pfg_run_all(context: Context) -> None:
|
|
try:
|
|
context.pfg_report = context.guardrail.run_all_checks(
|
|
action_name="test-action",
|
|
action_registry=context.pfg_action_registry,
|
|
actor_registry=context.pfg_actor_registry,
|
|
tool_names=tuple(context.pfg_tool_registry.keys()),
|
|
tool_registry=context.pfg_tool_registry,
|
|
skill_names=tuple(context.pfg_skill_registry.keys()),
|
|
skill_registry=context.pfg_skill_registry,
|
|
automation_profile=context.pfg_automation_profile,
|
|
require_checkpoints=context.pfg_require_checkpoints,
|
|
tool_capabilities=context.pfg_tool_capabilities,
|
|
resource_paths=context.pfg_resource_paths,
|
|
validation_names=(),
|
|
validation_registry=context.pfg_validation_registry,
|
|
)
|
|
context.pfg_rejection = None
|
|
except PreflightRejection as exc:
|
|
context.pfg_rejection = exc
|
|
context.pfg_report = None
|
|
|
|
|
|
@when("I run all preflight checks expecting rejection")
|
|
def step_pfg_run_all_expecting_rejection(context: Context) -> None:
|
|
try:
|
|
context.pfg_report = context.guardrail.run_all_checks(
|
|
action_name="missing-action",
|
|
action_registry=context.pfg_action_registry,
|
|
actor_registry=context.pfg_actor_registry,
|
|
automation_profile=context.pfg_automation_profile,
|
|
)
|
|
context.pfg_rejection = None
|
|
except PreflightRejection as exc:
|
|
context.pfg_rejection = exc
|
|
context.pfg_report = None
|
|
|
|
|
|
# -- Thens ----------------------------------------------------------------
|
|
|
|
|
|
@then('the check should pass with message containing "{text}"')
|
|
def step_pfg_then_pass_with_msg(context: Context, text: str) -> None:
|
|
result = context.pfg_check_result
|
|
assert result is not None, "No check result recorded"
|
|
assert result.passed is True, f"Expected PASS but got FAIL: {result.message}"
|
|
assert text in result.message, (
|
|
f"Expected '{text}' in message, got: {result.message}"
|
|
)
|
|
|
|
|
|
@then("the check should pass")
|
|
def step_pfg_then_pass(context: Context) -> None:
|
|
result = context.pfg_check_result
|
|
assert result is not None, "No check result recorded"
|
|
assert result.passed is True, f"Expected PASS but got FAIL: {result.message}"
|
|
|
|
|
|
@then('the check should fail with message containing "{text}"')
|
|
def step_pfg_then_fail_with_msg(context: Context, text: str) -> None:
|
|
result = context.pfg_check_result
|
|
assert result is not None, "No check result recorded"
|
|
assert result.passed is False, f"Expected FAIL but got PASS: {result.message}"
|
|
assert text in result.message, (
|
|
f"Expected '{text}' in message, got: {result.message}"
|
|
)
|
|
|
|
|
|
@then("the report should show all checks passed")
|
|
def step_pfg_then_report_all_passed(context: Context) -> None:
|
|
report = context.pfg_report
|
|
assert report is not None, "No report recorded"
|
|
assert report.all_passed, (
|
|
f"Expected all checks passed, failures: {[r.message for r in report.failures]}"
|
|
)
|
|
|
|
|
|
@then("no PreflightRejection should be raised")
|
|
def step_pfg_then_no_rejection(context: Context) -> None:
|
|
assert context.pfg_rejection is None, (
|
|
f"Expected no rejection but got: {context.pfg_rejection}"
|
|
)
|
|
|
|
|
|
@then("a PreflightRejection should be raised")
|
|
def step_pfg_then_rejection(context: Context) -> None:
|
|
assert context.pfg_rejection is not None, (
|
|
"Expected PreflightRejection but none was raised"
|
|
)
|
|
|
|
|
|
@then('the rejection should identify "{check_name}"')
|
|
def step_pfg_then_rejection_check(context: Context, check_name: str) -> None:
|
|
rejection = context.pfg_rejection
|
|
assert rejection is not None, "No rejection recorded"
|
|
assert rejection.check.value == check_name, (
|
|
f"Expected check '{check_name}', got '{rejection.check.value}'"
|
|
)
|
|
|
|
|
|
@then('the preflight warnings should contain "{text}"')
|
|
def step_pfg_then_warnings_contain(context: Context, text: str) -> None:
|
|
report = context.pfg_report
|
|
assert report is not None, "No report recorded"
|
|
assert any(text in warning.message for warning in report.warnings), (
|
|
f"Expected warning containing '{text}', got {report.warnings}"
|
|
)
|