HAL9000
848bdc47bb
CI / lint (pull_request) Successful in 40s
CI / helm (pull_request) Successful in 33s
CI / quality (pull_request) Successful in 56s
CI / typecheck (pull_request) Successful in 1m1s
CI / push-validation (pull_request) Successful in 38s
CI / build (pull_request) Successful in 1m17s
CI / security (pull_request) Successful in 2m14s
CI / unit_tests (pull_request) Successful in 5m27s
CI / docker (pull_request) Successful in 2m49s
CI / coverage (pull_request) Successful in 11m50s
CI / integration_tests (pull_request) Successful in 18m6s
CI / status-check (pull_request) Successful in 15s
795 lines
30 KiB
Python
795 lines
30 KiB
Python
"""Step definitions for Safety Profile domain model tests."""
|
|
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.domain.models.core.safety_profile import (
|
|
DEFAULT_SAFETY_PROFILE,
|
|
SafetyProfile,
|
|
SafetyProfileProvenance,
|
|
SafetyProfileRef,
|
|
get_builtin_safety_profile,
|
|
resolve_safety_profile,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_safety(**overrides: Any) -> SafetyProfile:
|
|
"""Create a SafetyProfile with sensible defaults."""
|
|
return SafetyProfile(**overrides)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Default construction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a default safety profile")
|
|
def step_create_default_safety(context: Context) -> None:
|
|
"""Create a default safety profile."""
|
|
context.safety_model = _make_safety()
|
|
context.safety_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Category parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a safety profile with categories "{cats}"')
|
|
def step_create_safety_with_cats(context: Context, cats: str) -> None:
|
|
"""Create a safety profile with specific categories."""
|
|
cat_list = [c for c in cats.split(",")]
|
|
context.safety_model = _make_safety(allowed_skill_categories=cat_list)
|
|
context.safety_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constraint validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I try to create a safety profile with max_cost_per_plan {value:g}")
|
|
def step_try_safety_cost(context: Context, value: float) -> None:
|
|
"""Try creating a safety profile with max_cost_per_plan."""
|
|
context.safety_error = None
|
|
context.safety_model = None
|
|
try:
|
|
context.safety_model = _make_safety(max_cost_per_plan=value)
|
|
except ValidationError as e:
|
|
context.safety_error = e
|
|
|
|
|
|
@when("I try to create a safety profile with max_total_cost {value:g}")
|
|
def step_try_safety_total_cost(context: Context, value: float) -> None:
|
|
"""Try creating with invalid max_total_cost."""
|
|
context.safety_error = None
|
|
context.safety_model = None
|
|
try:
|
|
context.safety_model = _make_safety(max_total_cost=value)
|
|
except ValidationError as e:
|
|
context.safety_error = e
|
|
|
|
|
|
@when("I try to create a safety profile with max_retries_per_step {value:d}")
|
|
def step_try_safety_retries(context: Context, value: int) -> None:
|
|
"""Try creating with invalid max_retries_per_step."""
|
|
context.safety_error = None
|
|
context.safety_model = None
|
|
try:
|
|
context.safety_model = _make_safety(max_retries_per_step=value)
|
|
except ValidationError as e:
|
|
context.safety_error = e
|
|
|
|
|
|
@when("I create a safety profile with max_cost_per_plan {value:g}")
|
|
def step_create_safety_cost(context: Context, value: float) -> None:
|
|
"""Create a safety profile with specific max_cost_per_plan."""
|
|
context.safety_model = _make_safety(max_cost_per_plan=value)
|
|
context.safety_error = None
|
|
|
|
|
|
@when("I create a safety profile with max_retries_per_step {value:d}")
|
|
def step_create_safety_retries(context: Context, value: int) -> None:
|
|
"""Create a safety profile with specific max_retries_per_step."""
|
|
context.safety_model = _make_safety(max_retries_per_step=value)
|
|
context.safety_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cross-field validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I try to create a safety profile with cost {cost:g} and total {total:g}")
|
|
def step_try_safety_cost_exceeds_total(
|
|
context: Context, cost: float, total: float
|
|
) -> None:
|
|
"""Try creating a safety profile where max_cost_per_plan > max_total_cost."""
|
|
context.safety_error = None
|
|
context.safety_model = None
|
|
try:
|
|
context.safety_model = _make_safety(
|
|
max_cost_per_plan=cost, max_total_cost=total
|
|
)
|
|
except ValidationError as e:
|
|
context.safety_error = e
|
|
|
|
|
|
@when("I create a safety profile with cost {cost:g} and total {total:g}")
|
|
def step_create_safety_cost_and_total(
|
|
context: Context, cost: float, total: float
|
|
) -> None:
|
|
"""Create a safety profile with both cost limits."""
|
|
context.safety_model = _make_safety(max_cost_per_plan=cost, max_total_cost=total)
|
|
context.safety_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Category type guard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I try to create a safety profile with non-string categories")
|
|
def step_try_safety_non_string_cats(context: Context) -> None:
|
|
"""Try creating a safety profile with non-string categories."""
|
|
context.safety_error = None
|
|
context.safety_model = None
|
|
try:
|
|
context.safety_model = _make_safety(
|
|
allowed_skill_categories=["valid", 123, None]
|
|
)
|
|
except ValidationError as e:
|
|
context.safety_error = e
|
|
|
|
|
|
@when("I try to create a safety profile with non-list categories")
|
|
def step_try_safety_non_list_cats(context: Context) -> None:
|
|
"""Try creating a safety profile with a non-list categories value."""
|
|
context.safety_error = None
|
|
context.safety_model = None
|
|
try:
|
|
context.safety_model = _make_safety(allowed_skill_categories="not-a-list")
|
|
except ValidationError as e:
|
|
context.safety_error = e
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# from_config factory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I load a safety profile from config with require_sandbox false")
|
|
def step_load_safety_config_sandbox(context: Context) -> None:
|
|
"""Load from config with require_sandbox false."""
|
|
config = {"require_sandbox": False}
|
|
context.safety_model = SafetyProfile.from_config(config)
|
|
context.safety_error = None
|
|
|
|
|
|
@when("I load a safety profile from full config")
|
|
def step_load_safety_full_config(context: Context) -> None:
|
|
"""Load from config with all fields."""
|
|
config = {
|
|
"require_sandbox": False,
|
|
"require_checkpoints": False,
|
|
"require_human_approval": True,
|
|
"allow_unsafe_tools": True,
|
|
"max_cost_per_plan": 50.0,
|
|
"max_retries_per_step": 5,
|
|
"max_total_cost": 200.0,
|
|
"allowed_skill_categories": ["code", "test"],
|
|
}
|
|
context.safety_model = SafetyProfile.from_config(config)
|
|
context.safety_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DEFAULT_SAFETY_PROFILE constant
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I load the default safety profile constant")
|
|
def step_load_default_constant(context: Context) -> None:
|
|
"""Load DEFAULT_SAFETY_PROFILE."""
|
|
context.safety_model = DEFAULT_SAFETY_PROFILE
|
|
context.safety_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SafetyProfileRef
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a safety profile ref with name "{name}" and provenance "{prov}"')
|
|
def step_create_safety_ref(context: Context, name: str, prov: str) -> None:
|
|
"""Create a SafetyProfileRef."""
|
|
context.safety_ref = SafetyProfileRef(
|
|
profile_name=name,
|
|
provenance=SafetyProfileProvenance(prov),
|
|
)
|
|
context.safety_error = None
|
|
|
|
|
|
@when("I try to create a safety profile ref with empty name")
|
|
def step_try_create_empty_ref(context: Context) -> None:
|
|
"""Try creating ref with empty name."""
|
|
context.safety_error = None
|
|
try:
|
|
SafetyProfileRef(
|
|
profile_name="",
|
|
provenance=SafetyProfileProvenance.PLAN,
|
|
)
|
|
except ValidationError as e:
|
|
context.safety_error = e
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resolve_safety_profile precedence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a plan safety profile with allow_unsafe_tools {val}")
|
|
def step_given_plan_profile(context: Context, val: str) -> None:
|
|
"""Set plan-level safety profile."""
|
|
context.resolve_plan = SafetyProfile(allow_unsafe_tools=val.lower() == "true")
|
|
|
|
|
|
@given("an action safety profile with allow_unsafe_tools {val}")
|
|
def step_given_action_profile(context: Context, val: str) -> None:
|
|
"""Set action-level safety profile."""
|
|
context.resolve_action = SafetyProfile(allow_unsafe_tools=val.lower() == "true")
|
|
|
|
|
|
@given("a project safety profile with allow_unsafe_tools {val}")
|
|
def step_given_project_profile(context: Context, val: str) -> None:
|
|
"""Set project-level safety profile."""
|
|
context.resolve_project = SafetyProfile(allow_unsafe_tools=val.lower() == "true")
|
|
|
|
|
|
@given("a global safety profile with allow_unsafe_tools {val}")
|
|
def step_given_global_profile(context: Context, val: str) -> None:
|
|
"""Set global-level safety profile."""
|
|
context.resolve_global = SafetyProfile(allow_unsafe_tools=val.lower() == "true")
|
|
|
|
|
|
@given("a full plan safety profile with all fields customized")
|
|
def step_given_full_plan_profile(context: Context) -> None:
|
|
"""Set plan-level safety profile with all 8 fields customized."""
|
|
context.resolve_plan = SafetyProfile(
|
|
require_sandbox=False,
|
|
require_checkpoints=False,
|
|
allow_unsafe_tools=True,
|
|
require_human_approval=True,
|
|
allowed_skill_categories=["code", "test"],
|
|
max_cost_per_plan=50.0,
|
|
max_retries_per_step=7,
|
|
max_total_cost=200.0,
|
|
)
|
|
|
|
|
|
@given("a full action safety profile with defaults")
|
|
def step_given_full_action_defaults(context: Context) -> None:
|
|
"""Set action-level safety profile with all defaults."""
|
|
context.resolve_action = SafetyProfile()
|
|
|
|
|
|
@given("a full action safety profile with all fields customized")
|
|
def step_given_full_action_profile(context: Context) -> None:
|
|
"""Set action-level safety profile with all 8 fields customized."""
|
|
context.resolve_action = SafetyProfile(
|
|
require_sandbox=False,
|
|
require_checkpoints=False,
|
|
allow_unsafe_tools=True,
|
|
require_human_approval=True,
|
|
allowed_skill_categories=["code", "test"],
|
|
max_cost_per_plan=50.0,
|
|
max_retries_per_step=7,
|
|
max_total_cost=200.0,
|
|
)
|
|
|
|
|
|
@when("I resolve the safety profile")
|
|
def step_resolve_profile(context: Context) -> None:
|
|
"""Resolve safety profile with whatever levels are set."""
|
|
plan = getattr(context, "resolve_plan", None)
|
|
action = getattr(context, "resolve_action", None)
|
|
project = getattr(context, "resolve_project", None)
|
|
global_ = getattr(context, "resolve_global", None)
|
|
profile, provenance = resolve_safety_profile(
|
|
plan_profile=plan,
|
|
action_profile=action,
|
|
project_profile=project,
|
|
global_profile=global_,
|
|
)
|
|
context.resolved_profile = profile
|
|
context.resolved_provenance = provenance
|
|
|
|
|
|
@when("I resolve the safety profile with no levels")
|
|
def step_resolve_profile_none(context: Context) -> None:
|
|
"""Resolve safety profile with all levels as None."""
|
|
profile, provenance = resolve_safety_profile()
|
|
context.resolved_profile = profile
|
|
context.resolved_provenance = provenance
|
|
|
|
|
|
@then('the resolved provenance should be "{expected}"')
|
|
def step_check_resolved_provenance(context: Context, expected: str) -> None:
|
|
"""Check resolved provenance value."""
|
|
actual = context.resolved_provenance.value
|
|
assert actual == expected, f"Expected provenance '{expected}', got '{actual}'"
|
|
|
|
|
|
@then("the resolved profile allow_unsafe_tools should be {expected}")
|
|
def step_check_resolved_allow_unsafe(context: Context, expected: str) -> None:
|
|
"""Check resolved profile allow_unsafe_tools."""
|
|
exp_bool = expected.lower() == "true"
|
|
actual = context.resolved_profile.allow_unsafe_tools
|
|
assert actual is exp_bool, f"Expected allow_unsafe_tools {exp_bool}, got {actual}"
|
|
|
|
|
|
@then("the resolved profile require_sandbox should be {expected}")
|
|
def step_check_resolved_require_sandbox(context: Context, expected: str) -> None:
|
|
"""Check resolved profile require_sandbox."""
|
|
exp_bool = expected.lower() == "true"
|
|
actual = context.resolved_profile.require_sandbox
|
|
assert actual is exp_bool, f"Expected require_sandbox {exp_bool}, got {actual}"
|
|
|
|
|
|
@then("the resolved profile require_checkpoints should be {expected}")
|
|
def step_check_resolved_require_checkpoints(context: Context, expected: str) -> None:
|
|
"""Check resolved profile require_checkpoints."""
|
|
exp_bool = expected.lower() == "true"
|
|
actual = context.resolved_profile.require_checkpoints
|
|
assert actual is exp_bool, f"Expected require_checkpoints {exp_bool}, got {actual}"
|
|
|
|
|
|
@then("the resolved profile require_human_approval should be {expected}")
|
|
def step_check_resolved_require_human_approval(context: Context, expected: str) -> None:
|
|
"""Check resolved profile require_human_approval."""
|
|
exp_bool = expected.lower() == "true"
|
|
actual = context.resolved_profile.require_human_approval
|
|
assert actual is exp_bool, (
|
|
f"Expected require_human_approval {exp_bool}, got {actual}"
|
|
)
|
|
|
|
|
|
@then("the resolved profile max_cost_per_plan should be {expected:g}")
|
|
def step_check_resolved_max_cost(context: Context, expected: float) -> None:
|
|
"""Check resolved profile max_cost_per_plan."""
|
|
actual = context.resolved_profile.max_cost_per_plan
|
|
assert actual == expected, f"Expected max_cost_per_plan {expected}, got {actual}"
|
|
|
|
|
|
@then("the resolved profile max_retries_per_step should be {expected:d}")
|
|
def step_check_resolved_max_retries(context: Context, expected: int) -> None:
|
|
"""Check resolved profile max_retries_per_step."""
|
|
actual = context.resolved_profile.max_retries_per_step
|
|
assert actual == expected, f"Expected max_retries_per_step {expected}, got {actual}"
|
|
|
|
|
|
@then("the resolved profile max_total_cost should be {expected:g}")
|
|
def step_check_resolved_max_total_cost(context: Context, expected: float) -> None:
|
|
"""Check resolved profile max_total_cost."""
|
|
actual = context.resolved_profile.max_total_cost
|
|
assert actual == expected, f"Expected max_total_cost {expected}, got {actual}"
|
|
|
|
|
|
@then('the resolved profile allowed_skill_categories should be "{expected}"')
|
|
def step_check_resolved_categories(context: Context, expected: str) -> None:
|
|
"""Check resolved profile allowed_skill_categories."""
|
|
expected_cats = [c.strip() for c in expected.split(",") if c.strip()]
|
|
actual = context.resolved_profile.allowed_skill_categories
|
|
assert actual == expected_cats, (
|
|
f"Expected allowed_skill_categories {expected_cats}, got {actual}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# model_dump
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a safety profile and dump it")
|
|
def step_create_and_dump(context: Context) -> None:
|
|
"""Create and dump a safety profile."""
|
|
context.safety_model = _make_safety()
|
|
context.safety_dump = context.safety_model.model_dump()
|
|
context.safety_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action with safety profile
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create an action with safety profile from config")
|
|
def step_create_action_with_safety(context: Context) -> None:
|
|
"""Create an Action from config with safety_profile."""
|
|
from cleveragents.domain.models.core.action import Action
|
|
|
|
config = {
|
|
"name": "local/test-safety-action",
|
|
"description": "Test action with safety profile",
|
|
"definition_of_done": "Tests pass",
|
|
"strategy_actor": "local/default",
|
|
"execution_actor": "local/default",
|
|
"safety_profile": {
|
|
"require_sandbox": False,
|
|
"max_cost_per_plan": 100.0,
|
|
},
|
|
}
|
|
context.action_model = Action.from_config(config)
|
|
context.safety_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# from_yaml edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I try to load a safety profile from a non-existent yaml file")
|
|
def step_try_load_nonexistent_yaml(context: Context) -> None:
|
|
"""Try loading from a non-existent YAML file."""
|
|
context.safety_error = None
|
|
try:
|
|
SafetyProfile.from_yaml("/tmp/nonexistent_safety_profile_test.yaml")
|
|
except FileNotFoundError as e:
|
|
context.safety_error = e
|
|
|
|
|
|
@when("I try to load a safety profile from yaml with non-dict content")
|
|
def step_try_load_non_dict_yaml(context: Context) -> None:
|
|
"""Try loading from a YAML file that contains a non-dict."""
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
yaml_content = "- item1\n- item2\n"
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
|
f.write(yaml_content)
|
|
yaml_path = f.name
|
|
|
|
context.safety_error = None
|
|
try:
|
|
SafetyProfile.from_yaml(yaml_path)
|
|
except ValueError as e:
|
|
context.safety_error = e
|
|
finally:
|
|
Path(yaml_path).unlink(missing_ok=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# validate_assignment (frozen model raises on mutation)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I try to assign max_retries_per_step {value:d} on the safety profile")
|
|
def step_try_assign_retries(context: Context, value: int) -> None:
|
|
"""Try assigning on a frozen safety profile (should raise)."""
|
|
context.safety_error = None
|
|
try:
|
|
context.safety_model.max_retries_per_step = value
|
|
except ValidationError as e:
|
|
context.safety_error = e
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the safety profile should be created")
|
|
def step_safety_created(context: Context) -> None:
|
|
"""Verify safety profile was created."""
|
|
assert context.safety_model is not None, "Safety profile should exist"
|
|
|
|
|
|
@then("the safety require_sandbox should be {expected}")
|
|
def step_check_require_sandbox(context: Context, expected: str) -> None:
|
|
"""Check require_sandbox."""
|
|
exp_bool = expected.lower() == "true"
|
|
actual = context.safety_model.require_sandbox
|
|
assert actual is exp_bool, f"Expected require_sandbox {exp_bool}, got {actual}"
|
|
|
|
|
|
@then("the safety require_checkpoints should be {expected}")
|
|
def step_check_require_checkpoints(context: Context, expected: str) -> None:
|
|
"""Check require_checkpoints."""
|
|
exp_bool = expected.lower() == "true"
|
|
actual = context.safety_model.require_checkpoints
|
|
assert actual is exp_bool, f"Expected require_checkpoints {exp_bool}, got {actual}"
|
|
|
|
|
|
@then("the safety require_human_approval should be {expected}")
|
|
def step_check_human_approval(context: Context, expected: str) -> None:
|
|
"""Check require_human_approval."""
|
|
exp_bool = expected.lower() == "true"
|
|
actual = context.safety_model.require_human_approval
|
|
assert actual is exp_bool, (
|
|
f"Expected require_human_approval {exp_bool}, got {actual}"
|
|
)
|
|
|
|
|
|
@then("the safety allow_unsafe_tools should be {expected}")
|
|
def step_check_allow_unsafe_tools(context: Context, expected: str) -> None:
|
|
"""Check allow_unsafe_tools."""
|
|
exp_bool = expected.lower() == "true"
|
|
actual = context.safety_model.allow_unsafe_tools
|
|
assert actual is exp_bool, f"Expected allow_unsafe_tools {exp_bool}, got {actual}"
|
|
|
|
|
|
@then("the safety max_retries_per_step should be {expected:d}")
|
|
def step_check_max_retries(context: Context, expected: int) -> None:
|
|
"""Check max_retries_per_step."""
|
|
actual = context.safety_model.max_retries_per_step
|
|
assert actual == expected, f"Expected max_retries {expected}, got {actual}"
|
|
|
|
|
|
@then("the safety max_cost_per_plan should be none")
|
|
def step_check_cost_none(context: Context) -> None:
|
|
"""Check max_cost_per_plan is None."""
|
|
assert context.safety_model.max_cost_per_plan is None
|
|
|
|
|
|
@then("the safety max_cost_per_plan should be {expected:g}")
|
|
def step_check_cost_value(context: Context, expected: float) -> None:
|
|
"""Check max_cost_per_plan value."""
|
|
actual = context.safety_model.max_cost_per_plan
|
|
assert actual == expected, f"Expected max_cost_per_plan {expected}, got {actual}"
|
|
|
|
|
|
@then("the safety max_total_cost should be none")
|
|
def step_check_total_cost_none(context: Context) -> None:
|
|
"""Check max_total_cost is None."""
|
|
assert context.safety_model.max_total_cost is None
|
|
|
|
|
|
@then("the safety max_total_cost should be {expected:g}")
|
|
def step_check_total_cost_value(context: Context, expected: float) -> None:
|
|
"""Check max_total_cost value."""
|
|
actual = context.safety_model.max_total_cost
|
|
assert actual == expected, f"Expected max_total_cost {expected}, got {actual}"
|
|
|
|
|
|
@then("the safety allowed_skill_categories should be empty")
|
|
def step_check_cats_empty(context: Context) -> None:
|
|
"""Check allowed_skill_categories is empty."""
|
|
assert context.safety_model.allowed_skill_categories == []
|
|
|
|
|
|
@then("the safety allowed_skill_categories count should be {expected:d}")
|
|
def step_check_cats_count(context: Context, expected: int) -> None:
|
|
"""Check allowed_skill_categories count."""
|
|
actual = len(context.safety_model.allowed_skill_categories)
|
|
assert actual == expected, f"Expected {expected} categories, got {actual}"
|
|
|
|
|
|
@then("a safety validation error should be raised")
|
|
def step_check_safety_error(context: Context) -> None:
|
|
"""Verify validation error was raised."""
|
|
assert context.safety_error is not None, "Expected a validation error"
|
|
|
|
|
|
@then("a safety FileNotFoundError should be raised")
|
|
def step_check_safety_file_not_found(context: Context) -> None:
|
|
"""Verify FileNotFoundError was raised."""
|
|
assert context.safety_error is not None, "Expected a FileNotFoundError"
|
|
assert isinstance(context.safety_error, FileNotFoundError)
|
|
|
|
|
|
@then("a safety ValueError should be raised")
|
|
def step_check_safety_value_error(context: Context) -> None:
|
|
"""Verify ValueError was raised."""
|
|
assert context.safety_error is not None, "Expected a ValueError"
|
|
assert isinstance(context.safety_error, ValueError)
|
|
|
|
|
|
@then('the safety profile ref should have name "{expected}"')
|
|
def step_check_ref_name(context: Context, expected: str) -> None:
|
|
"""Check safety profile ref name."""
|
|
assert context.safety_ref.profile_name == expected
|
|
|
|
|
|
@then('the safety profile ref should have provenance "{expected}"')
|
|
def step_check_ref_provenance(context: Context, expected: str) -> None:
|
|
"""Check safety profile ref provenance."""
|
|
actual = context.safety_ref.provenance.value
|
|
assert actual == expected, f"Expected provenance '{expected}', got '{actual}'"
|
|
|
|
|
|
@then("a NotImplementedError should be raised")
|
|
def step_check_not_implemented(context: Context) -> None:
|
|
"""Verify NotImplementedError was raised."""
|
|
assert context.resolve_error is not None, "Expected NotImplementedError"
|
|
|
|
|
|
@then('the resolve error should mention "{text}"')
|
|
def step_check_resolve_error_text(context: Context, text: str) -> None:
|
|
"""Check resolve error message contains text."""
|
|
error_str = str(context.resolve_error)
|
|
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
|
|
|
|
|
|
# NOTE: The above two steps are retained for backward compatibility.
|
|
# They are no longer used by safety_profile.feature (the resolve stub was
|
|
# replaced with real implementation tests) but may be referenced elsewhere.
|
|
|
|
|
|
@then('the safety dump should have key "{key}"')
|
|
def step_check_safety_dump_key(context: Context, key: str) -> None:
|
|
"""Check dump has key."""
|
|
assert key in context.safety_dump, f"Expected key '{key}' in dump"
|
|
|
|
|
|
@then("the safety dump require_sandbox should be true")
|
|
def step_check_safety_dump_sandbox(context: Context) -> None:
|
|
"""Check dump require_sandbox."""
|
|
assert context.safety_dump["require_sandbox"] is True
|
|
|
|
|
|
@then("the action safety profile should not be none")
|
|
def step_check_action_safety_not_none(context: Context) -> None:
|
|
"""Check action has safety profile."""
|
|
assert context.action_model.safety_profile is not None
|
|
|
|
|
|
@then("the action safety profile require_sandbox should be false")
|
|
def step_check_action_safety_sandbox_false(context: Context) -> None:
|
|
"""Check action safety profile require_sandbox."""
|
|
assert context.action_model.safety_profile.require_sandbox is False
|
|
|
|
|
|
@then("the action cli dict should contain safety_profile key")
|
|
def step_check_action_cli_dict_safety(context: Context) -> None:
|
|
"""Check action cli dict has safety_profile."""
|
|
cli_dict = context.action_model.as_cli_dict()
|
|
assert "safety_profile" in cli_dict, (
|
|
f"Expected 'safety_profile' in cli dict, keys: {list(cli_dict)}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# allowed_tools / denied_tools steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a safety profile with allowed_tools "{tools}"')
|
|
def step_create_safety_allowed_tools(context: Context, tools: str) -> None:
|
|
"""Create a safety profile with specific allowed_tools."""
|
|
tool_list = [t.strip() for t in tools.split(",") if t.strip()]
|
|
try:
|
|
context.safety_model = SafetyProfile(
|
|
allowed_tools=tool_list,
|
|
require_sandbox=False,
|
|
require_checkpoints=False,
|
|
)
|
|
context.safety_error = None
|
|
except Exception as exc:
|
|
context.safety_model = None
|
|
context.safety_error = exc
|
|
|
|
|
|
@when('I create a safety profile with denied_tools "{tools}"')
|
|
def step_create_safety_denied_tools(context: Context, tools: str) -> None:
|
|
"""Create a safety profile with specific denied_tools."""
|
|
tool_list = [t.strip() for t in tools.split(",") if t.strip()]
|
|
try:
|
|
context.safety_model = SafetyProfile(
|
|
denied_tools=tool_list,
|
|
require_sandbox=False,
|
|
require_checkpoints=False,
|
|
)
|
|
context.safety_error = None
|
|
except Exception as exc:
|
|
context.safety_model = None
|
|
context.safety_error = exc
|
|
|
|
|
|
@then("the safety allowed_tools count should be {expected:d}")
|
|
def step_check_allowed_tools_count(context: Context, expected: int) -> None:
|
|
"""Check the number of allowed_tools."""
|
|
actual = len(context.safety_model.allowed_tools)
|
|
assert actual == expected, f"Expected allowed_tools count {expected}, got {actual}"
|
|
|
|
|
|
@then("the safety denied_tools count should be {expected:d}")
|
|
def step_check_denied_tools_count(context: Context, expected: int) -> None:
|
|
"""Check the number of denied_tools."""
|
|
actual = len(context.safety_model.denied_tools)
|
|
assert actual == expected, f"Expected denied_tools count {expected}, got {actual}"
|
|
|
|
|
|
@then("the safety allowed_tools should be empty")
|
|
def step_check_allowed_tools_empty(context: Context) -> None:
|
|
"""Check that allowed_tools is empty."""
|
|
assert context.safety_model.allowed_tools == [], (
|
|
f"Expected empty allowed_tools, got {context.safety_model.allowed_tools}"
|
|
)
|
|
|
|
|
|
@then("the safety denied_tools should be empty")
|
|
def step_check_denied_tools_empty(context: Context) -> None:
|
|
"""Check that denied_tools is empty."""
|
|
assert context.safety_model.denied_tools == [], (
|
|
f"Expected empty denied_tools, got {context.safety_model.denied_tools}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Built-in safety profile steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I load the unrestricted built-in safety profile")
|
|
def step_load_builtin_unrestricted(context: Context) -> None:
|
|
"""Load the unrestricted built-in safety profile."""
|
|
try:
|
|
context.safety_model = get_builtin_safety_profile("unrestricted")
|
|
context.safety_error = None
|
|
except Exception as exc:
|
|
context.safety_model = None
|
|
context.safety_error = exc
|
|
|
|
|
|
@when("I load the read-only built-in safety profile")
|
|
def step_load_builtin_read_only(context: Context) -> None:
|
|
"""Load the read-only built-in safety profile."""
|
|
try:
|
|
context.safety_model = get_builtin_safety_profile("read-only")
|
|
context.safety_error = None
|
|
except Exception as exc:
|
|
context.safety_model = None
|
|
context.safety_error = exc
|
|
|
|
|
|
@when("I load the no-network built-in safety profile")
|
|
def step_load_builtin_no_network(context: Context) -> None:
|
|
"""Load the no-network built-in safety profile."""
|
|
try:
|
|
context.safety_model = get_builtin_safety_profile("no-network")
|
|
context.safety_error = None
|
|
except Exception as exc:
|
|
context.safety_model = None
|
|
context.safety_error = exc
|
|
|
|
|
|
@when("I load the sandboxed built-in safety profile")
|
|
def step_load_builtin_sandboxed(context: Context) -> None:
|
|
"""Load the sandboxed built-in safety profile."""
|
|
try:
|
|
context.safety_model = get_builtin_safety_profile("sandboxed")
|
|
context.safety_error = None
|
|
except Exception as exc:
|
|
context.safety_model = None
|
|
context.safety_error = exc
|
|
|
|
|
|
@when('I try to load an unknown built-in safety profile "{name}"')
|
|
def step_try_load_unknown_builtin(context: Context, name: str) -> None:
|
|
"""Try to load an unknown built-in safety profile."""
|
|
try:
|
|
context.safety_model = get_builtin_safety_profile(name)
|
|
context.safety_error = None
|
|
except KeyError as exc:
|
|
context.safety_model = None
|
|
context.safety_error = exc
|
|
|
|
|
|
@then("a safety KeyError should be raised")
|
|
def step_check_safety_key_error(context: Context) -> None:
|
|
"""Check that a KeyError was raised."""
|
|
assert context.safety_error is not None, (
|
|
"Expected a KeyError but no error was raised"
|
|
)
|
|
assert isinstance(context.safety_error, KeyError), (
|
|
f"Expected KeyError, got {type(context.safety_error).__name__}: "
|
|
f"{context.safety_error}"
|
|
)
|