Files
forja/tests/unit/test_runtime_nodes.py
T

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 agentforge_core.domain.policy import PolicyDefinition, PolicyValidator
from agentforge_core.guardrails.guardrails_ai import GuardrailsAIEngine
from agentforge_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"