67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
"""Tests de los nodos del grafo (lógica pura, sin compilación de grafo)."""
|
|
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from forja_core.domain.policy import PolicyDefinition, PolicyValidator
|
|
from forja_core.guardrails.guardrails_ai import GuardrailsAIEngine
|
|
from forja_core.runtime.nodes import build_node_validate_input
|
|
|
|
|
|
@pytest.fixture
|
|
def engine() -> GuardrailsAIEngine:
|
|
return GuardrailsAIEngine()
|
|
|
|
|
|
@pytest.fixture
|
|
def policy_block_email() -> PolicyDefinition:
|
|
return PolicyDefinition(
|
|
name="t",
|
|
version="v1",
|
|
description="t",
|
|
input_validators=[
|
|
PolicyValidator(
|
|
type="detect_pii",
|
|
config={"entities": ["EMAIL_ADDRESS"], "severity_on_match": "block"},
|
|
)
|
|
],
|
|
output_validators=[],
|
|
)
|
|
|
|
|
|
def _base_state(user_input: str) -> dict:
|
|
return {
|
|
"trace_id": str(uuid4()),
|
|
"agent_name": "x",
|
|
"agent_version": "v1",
|
|
"user_input": user_input,
|
|
"messages": [],
|
|
"raw_llm_output": None,
|
|
"parsed_output": None,
|
|
"proposed_actions": [],
|
|
"violations": [],
|
|
"decision_path": [],
|
|
"status": "running",
|
|
"error": None,
|
|
"human_decision": None,
|
|
}
|
|
|
|
|
|
async def test_validate_input_marca_blocked_si_pii(
|
|
engine: GuardrailsAIEngine, policy_block_email: PolicyDefinition
|
|
) -> None:
|
|
node = build_node_validate_input(engine, policy_block_email)
|
|
out = await node(_base_state("manda correo a juan@example.com"))
|
|
assert out["status"] == "blocked_by_guardrail"
|
|
assert any(v["blocked"] for v in out["violations"])
|
|
|
|
|
|
async def test_validate_input_pasa_sin_pii(
|
|
engine: GuardrailsAIEngine, policy_block_email: PolicyDefinition
|
|
) -> None:
|
|
node = build_node_validate_input(engine, policy_block_email)
|
|
out = await node(_base_state("incidente sin pii"))
|
|
# El nodo solo escribe `status` cuando bloquea; si no, el estado sigue "running".
|
|
assert out.get("status", "running") == "running"
|