feat(domain): añade modelos GuardrailViolation, PolicyDefinition y AgentExecution

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Juan
2026-05-10 10:14:13 +02:00
co-authored by Claude Opus 4.7
parent 73114f6731
commit ad77efe366
6 changed files with 241 additions and 0 deletions
+65
View File
@@ -0,0 +1,65 @@
"""Tests de modelos de ejecución."""
from datetime import UTC, datetime
from uuid import uuid4
from agentforge_core.domain.execution import (
AgentExecution,
AgentExecutionSummary,
DecisionStep,
ProposedAction,
)
def test_proposed_action_risk_score_in_rango() -> None:
a = ProposedAction(
id="act-1",
action="rollback_image",
target="cscf-01",
risk_score=4,
rollback_plan="redeploy 4.7.1",
requires_approval=True,
)
assert a.risk_score == 4
def test_decision_step_serializable() -> None:
s = DecisionStep(
step="llm_reason",
timestamp=datetime.now(UTC),
duration_ms=120,
detail={"tokens_in": 100},
)
assert s.model_dump()["detail"]["tokens_in"] == 100
def test_agent_execution_status_running() -> None:
e = AgentExecution(
trace_id=uuid4(),
agent_name="incident_analyzer",
agent_version="v1",
status="running",
started_at=datetime.now(UTC),
finished_at=None,
decision_path=[],
violations=[],
proposed_actions=[],
needs_human_for=None,
final_output=None,
error=None,
)
assert e.status == "running"
def test_agent_execution_summary_no_incluye_decision_path() -> None:
s = AgentExecutionSummary(
trace_id=uuid4(),
agent_name="x",
agent_version="v1",
status="completed",
started_at=datetime.now(UTC),
finished_at=datetime.now(UTC),
n_violations=0,
n_proposed_actions=2,
)
assert "decision_path" not in s.model_dump()
+35
View File
@@ -0,0 +1,35 @@
"""Tests del modelo GuardrailViolation."""
from datetime import UTC, datetime
from uuid import uuid4
import pytest
from pydantic import ValidationError
from agentforge_core.domain.guardrail import GuardrailViolation
def test_guardrail_violation_minimo_valido() -> None:
v = GuardrailViolation(
trace_id=uuid4(),
timestamp=datetime.now(UTC),
stage="input",
validator="DetectPII",
severity="block",
message="DNI detectado",
blocked=True,
)
assert v.blocked is True
def test_guardrail_violation_rechaza_severity_invalido() -> None:
with pytest.raises(ValidationError):
GuardrailViolation(
trace_id=uuid4(),
timestamp=datetime.now(UTC),
stage="input",
validator="x",
severity="critical", # type: ignore[arg-type]
message="x",
blocked=True,
)
+14
View File
@@ -0,0 +1,14 @@
"""Tests del modelo PolicyDefinition."""
from agentforge_core.domain.policy import PolicyDefinition, PolicyValidator
def test_policy_definition_default_fail_closed() -> None:
p = PolicyDefinition(
name="default",
version="v1",
description="x",
input_validators=[PolicyValidator(type="detect_pii", config={})],
output_validators=[],
)
assert p.on_validator_error == "fail_closed"