83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
"""Tests de los helpers de persistencia JSONL append-only."""
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from agentforge_core.api.persistence import (
|
|
append_execution,
|
|
append_violation,
|
|
read_execution_summaries,
|
|
read_violations,
|
|
)
|
|
from agentforge_core.domain.execution import AgentExecution
|
|
from agentforge_core.domain.guardrail import GuardrailViolation
|
|
|
|
|
|
def _violation() -> GuardrailViolation:
|
|
return GuardrailViolation(
|
|
trace_id=uuid4(),
|
|
timestamp=datetime.now(UTC),
|
|
stage="input",
|
|
validator="DetectPII",
|
|
severity="block",
|
|
message="email detectado",
|
|
blocked=True,
|
|
)
|
|
|
|
|
|
def _execution(status: str = "completed") -> AgentExecution:
|
|
return AgentExecution(
|
|
trace_id=uuid4(),
|
|
agent_name="incident_analyzer",
|
|
agent_version="v1",
|
|
status=status, # type: ignore[arg-type]
|
|
started_at=datetime.now(UTC),
|
|
finished_at=datetime.now(UTC),
|
|
decision_path=[],
|
|
violations=[],
|
|
proposed_actions=[],
|
|
needs_human_for=None,
|
|
final_output={},
|
|
error=None,
|
|
)
|
|
|
|
|
|
def test_append_violation_escribe_jsonl(tmp_path: Path) -> None:
|
|
append_violation(tmp_path, _violation())
|
|
line = (tmp_path / "violations.jsonl").read_text(encoding="utf-8").strip()
|
|
assert json.loads(line)["validator"] == "DetectPII"
|
|
|
|
|
|
def test_append_execution_es_append_only(tmp_path: Path) -> None:
|
|
append_execution(tmp_path, _execution())
|
|
append_execution(tmp_path, _execution())
|
|
lines = (tmp_path / "executions.jsonl").read_text(encoding="utf-8").strip().splitlines()
|
|
assert len(lines) == 2
|
|
|
|
|
|
def test_read_violations_filtra_lineas_vacias(tmp_path: Path) -> None:
|
|
append_violation(tmp_path, _violation())
|
|
with (tmp_path / "violations.jsonl").open("a", encoding="utf-8") as f:
|
|
f.write("\n")
|
|
out = read_violations(tmp_path)
|
|
assert len(out) == 1
|
|
assert out[0].validator == "DetectPII"
|
|
|
|
|
|
def test_read_violations_inexistente_devuelve_vacio(tmp_path: Path) -> None:
|
|
assert read_violations(tmp_path) == []
|
|
assert read_execution_summaries(tmp_path) == []
|
|
|
|
|
|
def test_read_execution_summaries_resume_campos(tmp_path: Path) -> None:
|
|
append_execution(tmp_path, _execution())
|
|
summaries = read_execution_summaries(tmp_path)
|
|
assert len(summaries) == 1
|
|
s = summaries[0]
|
|
assert s.agent_name == "incident_analyzer"
|
|
assert s.status == "completed"
|
|
assert s.n_violations == 0
|
|
assert s.n_proposed_actions == 0
|