242 lines
11 KiB
Python
242 lines
11 KiB
Python
# backend/tests/test_agent/test_agent_confirmation_v2.py
|
|
# #region Test.Agent.ConfirmationV2 [C:3] [TYPE Module] [SEMANTICS test,agent-chat,confirmation,guardrails,hitl]
|
|
# @BRIEF Contract tests for confirmation_metadata_for_tool and build_confirmation_contract_v2 — three-axis risk, env resolution, permission denied.
|
|
# @RELATION BINDS_TO -> [AgentChat.Confirmation]
|
|
# @TEST_FIXTURE: deploy_prod -> guarded + prod env_context
|
|
# @TEST_FIXTURE: deploy_staging -> guarded + staging env_context
|
|
# @TEST_FIXTURE: deploy_dev -> guarded + dev env_context
|
|
# @TEST_FIXTURE: delete_any -> dangerous risk_level
|
|
# @TEST_FIXTURE: read_only -> safe risk_level
|
|
# @TEST_FIXTURE: analyst_deploy -> permission_granted=false
|
|
# @TEST_FIXTURE: env_resolution -> tool_args.env_id > target_env > None
|
|
|
|
|
|
from src.agent._confirmation import (
|
|
_resolve_env_tier,
|
|
build_confirmation_contract_v2,
|
|
confirmation_metadata_for_tool,
|
|
permission_denied_payload,
|
|
)
|
|
|
|
# ── _resolve_env_tier ───────────────────────────────────────────────
|
|
|
|
# #region test_env_resolution_from_tool_args [C:2] [TYPE Function]
|
|
def test_resolve_env_tier_from_tool_args_env_id():
|
|
"""env_id in tool_args takes highest priority."""
|
|
assert _resolve_env_tier({"env_id": "prod-01"}, None) == "prod"
|
|
assert _resolve_env_tier({"env_id": "prod-01"}, "staging") == "prod" # tool_args wins
|
|
# #endregion test_env_resolution_from_tool_args
|
|
|
|
# #region test_env_resolution_from_environment_id [C:2] [TYPE Function]
|
|
def test_resolve_env_tier_from_environment_id():
|
|
"""environment_id in tool_args (alternative key) works."""
|
|
assert _resolve_env_tier({"environment_id": "staging-v2"}, None) == "staging"
|
|
# #endregion test_env_resolution_from_environment_id
|
|
|
|
# #region test_env_resolution_from_target_env [C:2] [TYPE Function]
|
|
def test_resolve_env_tier_from_target_env():
|
|
"""target_env fallback when tool_args has no env."""
|
|
assert _resolve_env_tier({}, "ss-dev") == "dev"
|
|
assert _resolve_env_tier({"query": "test"}, "prod-v2") == "prod"
|
|
# #endregion test_env_resolution_from_target_env
|
|
|
|
# #region test_env_resolution_null_when_no_env [C:2] [TYPE Function]
|
|
def test_resolve_env_tier_null_when_no_env():
|
|
"""When neither tool_args nor target_env provides env, return None."""
|
|
assert _resolve_env_tier({}, None) is None
|
|
assert _resolve_env_tier({"dashboard_id": 42}, None) is None
|
|
# #endregion test_env_resolution_null_when_no_env
|
|
|
|
# #region test_env_resolution_ambiguous_names [C:2] [TYPE Function]
|
|
def test_resolve_env_tier_ambiguous_names():
|
|
"""Environment names containing stag/test/local/dev are correctly tiered."""
|
|
assert _resolve_env_tier({"env_id": "autotest"}, None) == "staging" # "test" in autotest
|
|
assert _resolve_env_tier({"env_id": "localhost"}, None) == "dev" # "local" in localhost
|
|
# #endregion test_env_resolution_ambiguous_names
|
|
|
|
|
|
# ── build_confirmation_contract_v2 ───────────────────────────────────
|
|
|
|
# #region test_deploy_to_prod_is_guarded_with_prod_context [C:2] [TYPE Function]
|
|
def test_deploy_to_prod_is_guarded_with_prod_context():
|
|
"""Deploy to production → guarded risk + prod env_context."""
|
|
contract = build_confirmation_contract_v2(
|
|
"deploy_dashboard", {"env_id": "prod-01"}, "admin", "prod-01",
|
|
)
|
|
assert contract["risk"] == "write"
|
|
assert contract["risk_level"] == "guarded"
|
|
assert contract["dangerous"] is False
|
|
assert contract["env_context"] == "prod"
|
|
assert contract["permission_granted"] is True
|
|
# #endregion test_deploy_to_prod_is_guarded_with_prod_context
|
|
|
|
|
|
# #region test_deploy_to_staging_is_guarded_with_staging_context [C:2] [TYPE Function]
|
|
def test_deploy_to_staging_is_guarded_with_staging_context():
|
|
"""Deploy to staging → guarded risk + staging env_context."""
|
|
contract = build_confirmation_contract_v2(
|
|
"deploy_dashboard", {"env_id": "staging-v2"}, "admin", "staging-v2",
|
|
)
|
|
assert contract["risk"] == "write"
|
|
assert contract["risk_level"] == "guarded"
|
|
assert contract["env_context"] == "staging"
|
|
# #endregion test_deploy_to_staging_is_guarded_with_staging_context
|
|
|
|
|
|
# #region test_deploy_to_dev_is_guarded_with_dev_context [C:2] [TYPE Function]
|
|
def test_deploy_to_dev_is_guarded_with_dev_context():
|
|
"""Deploy to dev → guarded risk + dev env_context."""
|
|
contract = build_confirmation_contract_v2(
|
|
"deploy_dashboard", {"env_id": "my-dev-env"}, "admin", "my-dev-env",
|
|
)
|
|
assert contract["risk"] == "write"
|
|
assert contract["risk_level"] == "guarded"
|
|
assert contract["env_context"] == "dev"
|
|
# #endregion test_deploy_to_dev_is_guarded_with_dev_context
|
|
|
|
|
|
# #region test_delete_operation_is_dangerous [C:2] [TYPE Function]
|
|
def test_delete_operation_is_dangerous():
|
|
"""Delete-prefixed tools should be classified as dangerous."""
|
|
contract = build_confirmation_contract_v2(
|
|
"delete_dashboard", {}, "admin",
|
|
)
|
|
assert contract["risk"] == "write"
|
|
assert contract["risk_level"] == "dangerous"
|
|
assert contract["dangerous"] is True
|
|
# #endregion test_delete_operation_is_dangerous
|
|
|
|
|
|
# #region test_read_only_tool_is_safe [C:2] [TYPE Function]
|
|
def test_read_only_tool_is_safe():
|
|
"""Search/list/get tools should be classified as safe/read."""
|
|
contract = build_confirmation_contract_v2(
|
|
"search_dashboards", {"env_id": "prod"}, "admin", "prod",
|
|
)
|
|
assert contract["risk"] == "read"
|
|
assert contract["risk_level"] == "safe"
|
|
assert contract["dangerous"] is False
|
|
# #endregion test_read_only_tool_is_safe
|
|
|
|
|
|
# #region test_viewer_gets_permission_denied [C:2] [TYPE Function]
|
|
def test_viewer_permission_denied_for_write_tools():
|
|
"""Viewer role should be denied permission for write tools."""
|
|
contract = build_confirmation_contract_v2(
|
|
"deploy_dashboard", {"env_id": "prod"}, "viewer", "prod",
|
|
)
|
|
assert contract["permission_granted"] is False
|
|
assert contract["required_role"] == "admin"
|
|
assert contract["alternatives"] is not None
|
|
assert len(contract["alternatives"]) >= 1
|
|
# #endregion test_viewer_gets_permission_denied
|
|
|
|
|
|
# #region test_analyst_permission_denied_specific_tools [C:2] [TYPE Function]
|
|
def test_analyst_permission_denied_for_admin_tools():
|
|
"""Analyst/editor role should be denied for admin-only tools."""
|
|
for tool_name in ("deploy_dashboard", "execute_migration", "run_backup"):
|
|
contract = build_confirmation_contract_v2(tool_name, {}, "analyst")
|
|
assert contract["permission_granted"] is False, f"{tool_name} should be denied for analyst"
|
|
# #endregion test_analyst_permission_denied_specific_tools
|
|
|
|
|
|
# #region test_admin_write_tool_permission_granted [C:2] [TYPE Function]
|
|
def test_admin_write_tool_permission_granted():
|
|
"""Admin role should always have permission_granted=True for guarded tools."""
|
|
write_tools = ["deploy_dashboard", "commit_changes", "execute_migration"]
|
|
for tool_name in write_tools:
|
|
contract = build_confirmation_contract_v2(tool_name, {}, "admin")
|
|
assert contract["permission_granted"] is True, f"{tool_name} should be allowed for admin"
|
|
# #endregion test_admin_write_tool_permission_granted
|
|
|
|
|
|
# ── confirmation_metadata_for_tool ────────────────────────────────────
|
|
|
|
# #region test_metadata_for_tool_contains_required_fields [C:2] [TYPE Function]
|
|
def test_metadata_for_tool_contains_all_required_fields():
|
|
"""confirmation_metadata_for_tool should produce a complete metadata dict."""
|
|
meta = confirmation_metadata_for_tool(
|
|
"conv-123", "deploy_dashboard",
|
|
{"env_id": "prod-01"}, "admin", "prod-01",
|
|
)
|
|
required_fields = [
|
|
"type", "thread_id", "prompt", "tool_name", "tool_args",
|
|
"risk", "risk_level", "requires_confirmation",
|
|
"dangerous", "env_context",
|
|
]
|
|
for field in required_fields:
|
|
assert field in meta, f"Missing field '{field}' in metadata"
|
|
assert meta["type"] == "confirm_required"
|
|
assert meta["thread_id"] == "conv-123"
|
|
assert meta["tool_name"] == "deploy_dashboard"
|
|
assert meta["risk"] == "write"
|
|
assert meta["risk_level"] == "guarded"
|
|
assert meta["requires_confirmation"] is True
|
|
# #endregion test_metadata_for_tool_contains_required_fields
|
|
|
|
|
|
# ── permission_denied_payload ─────────────────────────────────────────
|
|
|
|
# #region test_permission_denied_payload_structure [C:2] [TYPE Function]
|
|
def test_permission_denied_payload_structure():
|
|
"""permission_denied_payload should produce valid JSON with correct type."""
|
|
import json
|
|
|
|
payload_str = permission_denied_payload("deploy_dashboard", "admin", "viewer")
|
|
payload = json.loads(payload_str)
|
|
|
|
assert payload["content"] == "⛔ Недостаточно прав для deploy_dashboard"
|
|
assert payload["metadata"]["type"] == "permission_denied"
|
|
assert payload["metadata"]["tool_name"] == "deploy_dashboard"
|
|
assert payload["metadata"]["required_role"] == "admin"
|
|
assert payload["metadata"]["user_role"] == "viewer"
|
|
assert payload["metadata"]["alternatives"] == []
|
|
# #endregion test_permission_denied_payload_structure
|
|
|
|
|
|
# #region test_permission_denied_payload_with_alternatives [C:2] [TYPE Function]
|
|
def test_permission_denied_payload_with_alternatives():
|
|
"""Alternatives list should be preserved in the payload."""
|
|
import json
|
|
|
|
alternatives = [{"action": "get_health_summary", "prompt": "Check system health"}]
|
|
payload_str = permission_denied_payload(
|
|
"deploy_dashboard", "admin", "viewer", alternatives,
|
|
)
|
|
payload = json.loads(payload_str)
|
|
|
|
assert payload["metadata"]["alternatives"] == alternatives
|
|
# #endregion test_permission_denied_payload_with_alternatives
|
|
|
|
|
|
# ── Unknown/null tool ─────────────────────────────────────────────────
|
|
|
|
# #region test_null_tool_name_handled [C:2] [TYPE Function]
|
|
def test_null_tool_name_handled_gracefully():
|
|
"""Null tool_name should produce 'unknown_action' fallback."""
|
|
contract = build_confirmation_contract_v2(None)
|
|
assert contract["operation"] == "unknown_action"
|
|
assert contract["risk_level"] == "safe"
|
|
# #endregion test_null_tool_name_handled
|
|
|
|
|
|
# #region test_execute_migration_is_guarded [C:2] [TYPE Function]
|
|
def test_execute_migration_is_guarded():
|
|
"""execute_migration should be classified as guarded (write)."""
|
|
contract = build_confirmation_contract_v2("execute_migration", {})
|
|
assert contract["risk"] == "write"
|
|
assert contract["risk_level"] == "guarded"
|
|
# #endregion test_execute_migration_is_guarded
|
|
|
|
|
|
# #region test_commit_changes_is_guarded [C:2] [TYPE Function]
|
|
def test_commit_changes_is_guarded():
|
|
"""commit_changes should be classified as guarded (write)."""
|
|
contract = build_confirmation_contract_v2("commit_changes", {})
|
|
assert contract["risk"] == "write"
|
|
assert contract["risk_level"] == "guarded"
|
|
# #endregion test_commit_changes_is_guarded
|
|
|
|
# #endregion Test.Agent.ConfirmationV2
|