86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
"""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
|