DataOps-platform/tests/test_agent_governance_api.py

from __future__ import annotations

USER_UID = "01900000-0000-7000-8000-000000009801"
AGENT_UID = "01900000-0000-7000-8000-000000009802"
REQUEST_UID = "01900000-0000-7000-8000-000000009803"


class FakeAgentGovernanceService:
    def __init__(self):
        self.calls = []

    def list_agents(self, **filters):
        self.calls.append(("list_agents", filters))
        return [{"uid": AGENT_UID, "status": "active", "current_version": 2}]

    def register_agent(self, payload, actor_uid):
        self.calls.append(("register_agent", payload, actor_uid))
        return {"uid": AGENT_UID, "status": "draft", "current_version": 1}

    def create_tool_grant(self, uid, payload, actor_uid):
        self.calls.append(("create_tool_grant", uid, payload, actor_uid))
        return {"uid": "grant-1", "agent_uid": uid, **payload}

    def issue_credential(self, uid, payload, actor_uid):
        self.calls.append(("issue_credential", uid, payload, actor_uid))
        return {"token": "only-once", "expires_at": "2026-08-02T12:05:00+08:00"}

    def authorize_action(self, uid, token, payload):
        self.calls.append(("authorize_action", uid, token, payload))
        return {"uid": REQUEST_UID, "decision": "denied", "current_version": 1}

    def list_requests(self, **filters):
        return [{"uid": REQUEST_UID, "decision": "denied", **filters}]

    def reconcile_action(self, uid, expected_version, actor_uid):
        self.calls.append(("reconcile_action", uid, expected_version, actor_uid))
        return {"uid": uid, "decision": "authorized", "current_version": 2}

    def replay(self, uid):
        return {"request": {"uid": uid, "input_digest": "a" * 64}, "events": []}

    def dashboard(self):
        return {"agent_count": 1, "denied_count": 1}


def _headers(role, **extra):
    return {"Authorization": f"Bearer {role}", **extra}


def _client(monkeypatch):
    from app import create_app
    from app.api.knowledge_base import agent_governance_routes

    service = FakeAgentGovernanceService()
    monkeypatch.setattr(agent_governance_routes, "_service", lambda: service)
    monkeypatch.setattr(
        "app.core.system.auth.load_identity_from_token",
        lambda token, secret: (
            {"id": USER_UID, "username": token, "roles": [token]}
            if token in {"viewer", "editor", "admin"}
            else None
        ),
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


def test_viewer_reads_agent_inventory_dashboard_and_replay(monkeypatch):
    client, _service = _client(monkeypatch)
    listing = client.get(
        "/api/knowledge/agents?status=active", headers=_headers("viewer")
    )
    assert listing.status_code == 200
    assert listing.get_json()["data"][0]["uid"] == AGENT_UID
    dashboard = client.get("/api/knowledge/agents/dashboard", headers=_headers("viewer"))
    assert dashboard.get_json()["data"]["denied_count"] == 1
    replay = client.get(
        f"/api/knowledge/agents/actions/{REQUEST_UID}/replay",
        headers=_headers("viewer"),
    )
    assert replay.status_code == 200
    assert replay.headers["Cache-Control"] == "no-store"


def test_editor_operates_grants_and_one_time_credentials_but_cannot_register(monkeypatch):
    client, service = _client(monkeypatch)
    forbidden = client.post(
        "/api/knowledge/agents", json={"code": "NOPE"}, headers=_headers("editor")
    )
    assert forbidden.status_code == 403
    grant = client.post(
        f"/api/knowledge/agents/{AGENT_UID}/grants",
        json={"tool_name": "knowledge.search"}, headers=_headers("editor"),
    )
    assert grant.status_code == 201
    credential = client.post(
        f"/api/knowledge/agents/{AGENT_UID}/credentials",
        json={"ttl_seconds": 300}, headers=_headers("editor"),
    )
    assert credential.status_code == 201
    assert credential.headers["Cache-Control"] == "no-store"
    assert credential.get_json()["data"]["token"] == "only-once"
    assert service.calls[-1][0] == "issue_credential"


def test_admin_registers_and_reconciles_while_machine_credential_is_explicit(monkeypatch):
    client, service = _client(monkeypatch)
    created = client.post(
        "/api/knowledge/agents", json={"code": "READ_GOV"}, headers=_headers("admin")
    )
    assert created.status_code == 201
    assert created.headers["ETag"] == '"1"'

    decision = client.post(
        f"/api/knowledge/agents/{AGENT_UID}/actions/authorize",
        json={"prompt": "read"},
        headers=_headers("editor", **{"X-Agent-Credential": "machine-token"}),
    )
    assert decision.status_code == 201
    assert decision.get_json()["data"]["decision"] == "denied"
    assert service.calls[-1][2] == "machine-token"

    missing = client.post(
        f"/api/knowledge/agents/actions/{REQUEST_UID}/reconcile",
        headers=_headers("admin"),
    )
    assert missing.status_code == 428
    reconciled = client.post(
        f"/api/knowledge/agents/actions/{REQUEST_UID}/reconcile",
        headers=_headers("admin", **{"If-Match": '"1"'}),
    )
    assert reconciled.status_code == 200
    assert reconciled.headers["ETag"] == '"2"'


def test_production_requires_a_dedicated_machine_credential_secret():
    import pytest

    from app import create_app
    from app.api.knowledge_base.agent_governance_routes import (
        AgentGovernanceUnavailable,
        _effective_credential_secret,
        _require_credential_secret,
    )

    app = create_app()
    app.config.update(TESTING=False, FLASK_ENV="production", AGENT_CREDENTIAL_SECRET="")
    with app.app_context():
        fallback, dedicated = _effective_credential_secret()
        assert len(fallback) == 64 and dedicated is False
        with pytest.raises(AgentGovernanceUnavailable):
            _require_credential_secret()
        app.config["AGENT_CREDENTIAL_SECRET"] = "production-agent-secret-with-at-least-32-bytes"
        _secret, dedicated = _effective_credential_secret()
        assert dedicated is True
        _require_credential_secret()