DataOps-platform/tests/test_agent_governance_contract.py

from pathlib import Path

from app.core.system.permissions import (
    AGENTS_MANAGE,
    AGENTS_OPERATE,
    AGENTS_READ,
    permission_for_request,
    permissions_for_roles,
)

ROOT = Path(__file__).resolve().parents[1]


def test_agent_governance_migration_is_additive_scoped_and_evidence_preserving():
    migration = (ROOT / "migrations/versions/20260802_450_agent_governance.py").read_text()
    assert 'revision = "20260802_450"' in migration
    assert 'down_revision = "20260802_440"' in migration
    for table in (
        "governed_agents", "governed_agent_versions", "agent_tool_grants",
        "agent_credentials", "agent_action_requests", "agent_governance_events",
    ):
        assert f"CREATE TABLE public.{table}" in migration
    assert "INTERVAL '15 minutes'" in migration
    assert "automatic_execution_allowed" in migration
    assert "REFERENCES public.governance_tasks(uid)" in migration
    assert "raise RuntimeError" in migration
    assert "DROP TABLE" not in migration.upper()


def test_agent_permissions_are_separated_and_deny_by_default_router_knows_them():
    assert AGENTS_READ in permissions_for_roles(["viewer"])
    assert AGENTS_OPERATE not in permissions_for_roles(["viewer"])
    assert AGENTS_OPERATE in permissions_for_roles(["editor"])
    assert AGENTS_MANAGE not in permissions_for_roles(["editor"])
    assert AGENTS_MANAGE in permissions_for_roles(["admin"])
    assert permission_for_request("/api/knowledge/agents", "GET") == (AGENTS_READ,)
    assert permission_for_request("/api/knowledge/agents", "POST") == (AGENTS_MANAGE,)
    assert permission_for_request(
        "/api/knowledge/agents/x/actions/authorize", "POST"
    ) == (AGENTS_OPERATE,)
    assert permission_for_request(
        "/api/knowledge/agents/actions/x/reconcile", "POST"
    ) == (AGENTS_MANAGE,)


def test_agent_api_and_operator_surface_expose_identity_grants_decisions_and_replay():
    api = (ROOT / "app/api/knowledge_base/agent_governance_routes.py").read_text()
    client = (ROOT / "frontend/src/api/agentGovernance.js").read_text()
    view = (ROOT / "frontend/src/views/systemManage/agentGovernance/index.vue").read_text()
    routes = (ROOT / "frontend/src/router/routes.js").read_text()
    for surface in ("/agents", "/grants", "/credentials", "/actions/authorize", "/reconcile", "/complete", "/replay", "/dashboard"):
        assert surface in api
    for operation in (
        "registerGovernedAgent", "createAgentToolGrant", "issueAgentCredential",
        "revokeAgentCredentials", "reconcileAgentAction", "replayAgentAction",
    ):
        assert operation in client
    for boundary in (
        "高风险和关键风险操作", "平台不会自动执行", "凭证最长 15 分钟",
        "凭证仅本次返回", "决策证据回放",
    ):
        assert boundary in view or boundary in api
    assert "/systemManage/agent-governance" in routes
    assert "agents:read" in routes


def test_repository_never_persists_raw_prompt_token_or_output():
    migration = (ROOT / "migrations/versions/20260802_450_agent_governance.py").read_text()
    repository = (ROOT / "app/core/llm/agent_governance_repository.py").read_text()
    service = (ROOT / "app/core/llm/agent_governance.py").read_text()
    assert "raw_prompt" not in migration
    assert "token_digest" in migration and "token_digest" in repository
    assert "input_digest" in migration and "output_digest" in migration
    assert '"raw_prompt_retained": False' in service
    assert "high-risk automatic execution is disabled" in service
    assert "UnifiedWorkCenterService" in repository