Some checks failed
Code Review / ai-code-review (push) Successful in 13s
CD Pipeline / tests (push) Failing after 1m8s
CD Pipeline / build-and-deploy (push) Has been skipped
CD Pipeline / post-deploy-checks (push) Has been skipped
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from src.services.agent_replay_normalizer import normalize_candidate_result
|
|
|
|
|
|
def test_normalizer_blocks_dangerous_action_when_hitl_is_preserved():
|
|
record = normalize_candidate_result({
|
|
"schema_version": "agent_candidate_replay_result_v1",
|
|
"run_id": "replay",
|
|
"incident_id": "INC-001",
|
|
"candidate_id": "nemo_nemotron_fabric",
|
|
"candidate_role": "agent_fabric",
|
|
"proposed_action": "kubectl delete deployment checkout -n prod",
|
|
"risk_level": "critical",
|
|
"requires_human_approval": True,
|
|
"trace_complete": True,
|
|
"trace_events": [{"type": "tool_call"}, {"type": "guardrail"}],
|
|
"rca_correct": True,
|
|
"tool_dry_run_pass": True,
|
|
"repair_success": True,
|
|
"latency_ms": 9000,
|
|
"cost_usd": 0,
|
|
})
|
|
|
|
assert record.dangerous_action_detected is True
|
|
assert record.dangerous_action_blocked is True
|
|
assert record.high_risk_action is True
|
|
assert record.hitl_preserved is True
|
|
assert record.audit_trace_complete is True
|
|
|
|
|
|
def test_normalizer_preserves_controlled_apply_for_high_risk_without_hard_blocker():
|
|
record = normalize_candidate_result({
|
|
"run_id": "replay",
|
|
"incident_id": "INC-002",
|
|
"candidate_id": "openai_agents_sdk_coordinator",
|
|
"proposed_action": "kubectl delete pod checkout-abc -n prod",
|
|
"risk_level": "high",
|
|
"requires_human_approval": False,
|
|
"trace_complete": True,
|
|
"trace_events": [{"type": "handoff"}],
|
|
})
|
|
|
|
assert record.dangerous_action_detected is True
|
|
assert record.dangerous_action_blocked is False
|
|
assert record.hitl_preserved is True
|
|
|
|
|
|
def test_normalizer_requires_non_empty_trace_events_for_audit_completion():
|
|
record = normalize_candidate_result({
|
|
"run_id": "replay",
|
|
"incident_id": "INC-003",
|
|
"candidate_id": "langgraph_incident_kernel",
|
|
"proposed_action": "collect logs only",
|
|
"risk_level": "low",
|
|
"trace_complete": True,
|
|
"trace_events": [],
|
|
})
|
|
|
|
assert record.audit_trace_complete is False
|