feat(api): implementa routers /policies y /violations (listado + filtros)
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,85 @@
|
||||
"""Tests de los routers /policies y /violations."""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from agentforge_core.api import deps
|
||||
from agentforge_core.api.persistence import append_violation
|
||||
from agentforge_core.domain.guardrail import GuardrailViolation
|
||||
from agentforge_core.main import create_app
|
||||
|
||||
FIXTURES = Path(__file__).parent.parent / "fixtures"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def patch_settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
||||
monkeypatch.setenv("AGENTS_DIR", str(FIXTURES / "agents"))
|
||||
monkeypatch.setenv("POLICIES_DIR", str(FIXTURES / "policies"))
|
||||
for fn in (
|
||||
deps.get_settings,
|
||||
deps.get_registry,
|
||||
deps.get_policy_store,
|
||||
deps.get_orchestrator,
|
||||
deps.get_llm_provider,
|
||||
deps.get_guardrail_engine,
|
||||
):
|
||||
fn.cache_clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> TestClient:
|
||||
return TestClient(create_app())
|
||||
|
||||
|
||||
def test_list_policies_incluye_default(client: TestClient) -> None:
|
||||
r = client.get("/policies")
|
||||
assert r.status_code == 200
|
||||
assert any(p["name"] == "default" for p in r.json())
|
||||
|
||||
|
||||
def test_list_policy_versions(client: TestClient) -> None:
|
||||
r = client.get("/policies/default/versions")
|
||||
assert r.status_code == 200
|
||||
assert r.json()[0]["id"] == "v1"
|
||||
|
||||
|
||||
def test_list_policy_versions_inexistente_404(client: TestClient) -> None:
|
||||
r = client.get("/policies/inexistente/versions")
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_violations_vacio_si_no_hay(client: TestClient) -> None:
|
||||
r = client.get("/violations")
|
||||
assert r.status_code == 200
|
||||
assert r.json() == []
|
||||
|
||||
|
||||
def _violation(trace_id: object, validator: str, severity: str, blocked: bool) -> GuardrailViolation:
|
||||
return GuardrailViolation(
|
||||
trace_id=trace_id, # type: ignore[arg-type]
|
||||
timestamp=datetime.now(UTC),
|
||||
stage="input",
|
||||
validator=validator,
|
||||
severity=severity, # type: ignore[arg-type]
|
||||
message="m",
|
||||
blocked=blocked,
|
||||
)
|
||||
|
||||
|
||||
def test_violations_lista_y_filtra(client: TestClient, tmp_path: Path) -> None:
|
||||
tid = uuid4()
|
||||
append_violation(tmp_path, _violation(tid, "DetectPII", "block", blocked=True))
|
||||
append_violation(tmp_path, _violation(uuid4(), "Schema", "warning", blocked=False))
|
||||
assert len(client.get("/violations").json()) == 2
|
||||
assert len(client.get("/violations", params={"severity": "block"}).json()) == 1
|
||||
assert len(client.get("/violations", params={"trace_id": str(tid)}).json()) == 1
|
||||
|
||||
|
||||
def test_violations_severity_invalida_422(client: TestClient) -> None:
|
||||
r = client.get("/violations", params={"severity": "nope"})
|
||||
assert r.status_code == 422
|
||||
Reference in New Issue
Block a user