test(integration): HITL approve/reject + bloqueo por PII (NIF, email)
- test_invoke_hitl: escenario SIP real → awaiting_approval; approve devuelve status=completed con approved_actions; reject deja status=failed con error=rejected_by_human. - test_invoke_pii_block: input con NIF español y email → status=blocked_by_guardrail y violación DetectPII con blocked=True. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,41 @@
|
|||||||
|
"""Integration HITL: escenario SIP (risk>=4) -> awaiting_approval -> approve/reject."""
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
EXAMPLES = (
|
||||||
|
Path(__file__).parent.parent.parent / "agents" / "incident_analyzer" / "examples"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.integration
|
||||||
|
def test_hitl_approve_completa(integration_client) -> None: # type: ignore[no-untyped-def]
|
||||||
|
payload = (EXAMPLES / "01_sip_registration_drop.txt").read_text(encoding="utf-8")
|
||||||
|
r = integration_client.post("/agents/incident_analyzer/invoke", json={"input": payload})
|
||||||
|
body = r.json()
|
||||||
|
assert body["status"] == "awaiting_approval"
|
||||||
|
assert body["needs_human_for"]
|
||||||
|
trace_id = body["trace_id"]
|
||||||
|
pending = [a["id"] for a in body["needs_human_for"]]
|
||||||
|
|
||||||
|
r2 = integration_client.post(
|
||||||
|
f"/executions/{trace_id}/approve",
|
||||||
|
json={"approved_action_ids": pending, "comment": "OK rollback"},
|
||||||
|
)
|
||||||
|
assert r2.status_code == 200
|
||||||
|
assert r2.json()["status"] == "completed"
|
||||||
|
assert r2.json()["final_output"]["approved_actions"]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.integration
|
||||||
|
def test_hitl_reject_marca_failed(integration_client) -> None: # type: ignore[no-untyped-def]
|
||||||
|
payload = (EXAMPLES / "01_sip_registration_drop.txt").read_text(encoding="utf-8")
|
||||||
|
r = integration_client.post("/agents/incident_analyzer/invoke", json={"input": payload})
|
||||||
|
trace_id = r.json()["trace_id"]
|
||||||
|
r2 = integration_client.post(
|
||||||
|
f"/executions/{trace_id}/reject", json={"reason": "rollback no procede"}
|
||||||
|
)
|
||||||
|
assert r2.status_code == 200
|
||||||
|
assert r2.json()["status"] == "failed"
|
||||||
|
assert "rejected_by_human" in r2.json()["error"]
|
||||||
@@ -0,0 +1,13 @@
|
|||||||
|
"""Integration: input con PII (NIF español + email) bloqueado por guardrails de entrada."""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.integration
|
||||||
|
def test_pii_es_nif_bloquea(integration_client) -> None: # type: ignore[no-untyped-def]
|
||||||
|
pii_input = "El cliente con NIF 12345678Z reporta caída de servicio. Email juan@example.com"
|
||||||
|
r = integration_client.post("/agents/incident_analyzer/invoke", json={"input": pii_input})
|
||||||
|
assert r.status_code == 200
|
||||||
|
body = r.json()
|
||||||
|
assert body["status"] == "blocked_by_guardrail"
|
||||||
|
assert any(v["validator"] == "DetectPII" and v["blocked"] for v in body["violations"])
|
||||||
Reference in New Issue
Block a user