cbb985627d
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 16s
CI / quality (pull_request) Successful in 21s
CI / lint (pull_request) Successful in 22s
CI / security (pull_request) Successful in 36s
CI / typecheck (pull_request) Successful in 37s
CI / integration_tests (pull_request) Successful in 4m49s
CI / unit_tests (pull_request) Successful in 5m54s
CI / docker (pull_request) Successful in 38s
CI / benchmark-regression (pull_request) Successful in 15m25s
CI / coverage (pull_request) Successful in 41m50s
147 lines
4.4 KiB
Python
147 lines
4.4 KiB
Python
"""Helper script for Robot Framework automation profile guard tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
import yaml
|
|
|
|
from cleveragents.application.services.automation_profile_service import (
|
|
AutomationProfileService,
|
|
)
|
|
from cleveragents.domain.models.core.automation_guard import (
|
|
AutomationGuard,
|
|
)
|
|
from cleveragents.domain.models.core.automation_profile import (
|
|
AutomationProfile,
|
|
)
|
|
|
|
|
|
def _test_guard_allow() -> None:
|
|
"""Guard allows tool within limits."""
|
|
g = AutomationGuard(max_tool_calls_per_step=5)
|
|
p = AutomationProfile(name="test/allow", guards=g)
|
|
r = p.check_guard("read_file", calls_so_far=3)
|
|
assert r.allowed is True
|
|
print("guard-allow-ok")
|
|
|
|
|
|
def _test_guard_deny() -> None:
|
|
"""Guard blocks tool on denylist."""
|
|
g = AutomationGuard(tool_denylist=["shell_exec"])
|
|
p = AutomationProfile(name="test/deny", guards=g)
|
|
r = p.check_guard("shell_exec")
|
|
assert r.allowed is False
|
|
assert "denylist" in (r.reason or "")
|
|
print("guard-deny-ok")
|
|
|
|
|
|
def _test_guard_max_calls() -> None:
|
|
"""Guard blocks when call limit exceeded."""
|
|
g = AutomationGuard(max_tool_calls_per_step=3)
|
|
p = AutomationProfile(name="test/max-calls", guards=g)
|
|
r = p.check_guard("tool", calls_so_far=3)
|
|
assert r.allowed is False
|
|
assert r.requires_approval is True
|
|
print("guard-max-calls-ok")
|
|
|
|
|
|
def _test_guard_cost() -> None:
|
|
"""Guard blocks when cost budget exceeded."""
|
|
g = AutomationGuard(max_total_cost=10.0)
|
|
p = AutomationProfile(name="test/cost", guards=g)
|
|
r = p.check_guard("tool", cost_so_far=10.0)
|
|
assert r.allowed is False
|
|
assert "Cost" in (r.reason or "")
|
|
print("guard-cost-ok")
|
|
|
|
|
|
def _test_guard_write() -> None:
|
|
"""Guard requires approval for writes."""
|
|
g = AutomationGuard(require_approval_for_writes=True)
|
|
p = AutomationProfile(name="test/write", guards=g)
|
|
r = p.check_guard("write_file", is_write=True)
|
|
assert r.allowed is False
|
|
assert r.requires_approval is True
|
|
print("guard-write-ok")
|
|
|
|
|
|
def _test_guard_apply() -> None:
|
|
"""Guard requires approval for apply."""
|
|
g = AutomationGuard(require_approval_for_apply=True)
|
|
p = AutomationProfile(name="test/apply", guards=g)
|
|
r = p.check_guard("__apply__")
|
|
assert r.allowed is False
|
|
assert r.requires_approval is True
|
|
print("guard-apply-ok")
|
|
|
|
|
|
def _test_from_yaml() -> None:
|
|
"""Load profile from YAML."""
|
|
cfg: dict[str, Any] = {
|
|
"name": "test/yaml",
|
|
"guards": {"max_tool_calls_per_step": 5},
|
|
}
|
|
tmp_path = tempfile.mktemp(suffix=".yaml")
|
|
with open(tmp_path, "w") as fh:
|
|
yaml.dump(cfg, fh)
|
|
p = AutomationProfile.from_yaml(tmp_path)
|
|
assert p.guards is not None
|
|
assert p.guards.max_tool_calls_per_step == 5
|
|
print("from-yaml-ok")
|
|
|
|
|
|
def _test_resolution() -> None:
|
|
"""Effective profile resolution precedence."""
|
|
svc = AutomationProfileService()
|
|
p1 = svc.get_effective_profile(plan_profile="cautious")
|
|
assert p1.name == "cautious"
|
|
p2 = svc.get_effective_profile(action_profile="manual")
|
|
assert p2.name == "manual"
|
|
p3 = svc.get_effective_profile(default_profile="auto")
|
|
assert p3.name == "auto"
|
|
print("resolution-ok")
|
|
|
|
|
|
def _test_summary() -> None:
|
|
"""Print guard feature summary."""
|
|
g = AutomationGuard(
|
|
max_tool_calls_per_step=10,
|
|
max_total_cost=50.0,
|
|
tool_denylist=["dangerous"],
|
|
require_approval_for_writes=True,
|
|
require_approval_for_apply=True,
|
|
)
|
|
p = AutomationProfile(name="test/summary", guards=g)
|
|
r1 = p.check_guard("safe_tool")
|
|
assert r1.allowed is True
|
|
r2 = p.check_guard("dangerous")
|
|
assert r2.allowed is False
|
|
print("summary-ok")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "summary"
|
|
dispatch: dict[str, Any] = {
|
|
"guard-allow": _test_guard_allow,
|
|
"guard-deny": _test_guard_deny,
|
|
"guard-max-calls": _test_guard_max_calls,
|
|
"guard-cost": _test_guard_cost,
|
|
"guard-write": _test_guard_write,
|
|
"guard-apply": _test_guard_apply,
|
|
"from-yaml": _test_from_yaml,
|
|
"resolution": _test_resolution,
|
|
"summary": _test_summary,
|
|
}
|
|
fn = dispatch.get(cmd)
|
|
if fn:
|
|
fn()
|
|
else:
|
|
print(f"Unknown command: {cmd}", file=sys.stderr)
|
|
sys.exit(1)
|