Files
awoooi/apps/api/tests/test_agent_langgraph_adapter.py
Your Name e0a86b6254
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
feat(agents): route high risk through controlled automation
2026-06-26 19:19:40 +08:00

75 lines
2.6 KiB
Python

from __future__ import annotations
import pytest
from src.services.agent_langgraph_adapter import (
LANGGRAPH_CANDIDATE_ID,
build_langgraph_candidate_result,
)
def test_langgraph_adapter_emits_candidate_result_contract():
result = build_langgraph_candidate_result({
"schema_version": "agent_replay_candidate_input_v1",
"run_id": "run",
"incident_id": "INC-1",
"incident_context": {
"severity": "P2",
"alert_category": "host_resource",
"alertname": "HostDiskUsageHigh",
"affected_services": ["node-exporter-110"],
"signals": [
{
"labels": {"instance": "192.168.0.110"},
"annotations": {"summary": "disk usage high"},
}
],
},
"source_metadata": {},
}).to_dict()
assert result["schema_version"] == "agent_candidate_replay_result_v1"
assert result["candidate_id"] == LANGGRAPH_CANDIDATE_ID
assert result["candidate_role"] == "durable_incident_workflow_kernel"
assert result["incident_id"] == "INC-1"
assert "SSH_DIAGNOSE" in result["proposed_action"]
assert result["risk_level"] == "medium"
assert result["requires_human_approval"] is False
assert result["fallback_used"] is False
assert result["trace_complete"] is True
assert result["metadata"]["adapter_mode"] == "deterministic_offline_workflow_kernel"
assert result["metadata"]["sdk_dependency"] == "langgraph_python_package_not_installed"
def test_langgraph_adapter_rejects_label_leak_before_execution():
with pytest.raises(ValueError, match="evaluation label"):
build_langgraph_candidate_result({
"run_id": "run",
"incident_id": "INC-1",
"incident_context": {
"verification_result": "success",
},
"source_metadata": {},
})
def test_langgraph_adapter_preserves_resolved_incidents_as_no_action():
result = build_langgraph_candidate_result({
"schema_version": "agent_replay_candidate_input_v1",
"run_id": "run",
"incident_id": "INC-2",
"incident_context": {
"severity": "P3",
"status": "resolved",
"alert_category": "infrastructure",
"alertname": "DockerContainerUnhealthy",
"affected_services": ["cadvisor"],
},
"source_metadata": {},
}).to_dict()
assert result["proposed_action"].startswith("NO_ACTION:")
assert result["blocked_by_policy"] is True
assert result["trace_complete"] is True
assert result["cost_usd"] == 0