File size: 5,685 Bytes
cc036ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | """Contract tests for governance API endpoints using Schemathesis schema validation."""
import pytest
from fastapi.testclient import TestClient
from main_api_app import app
from tests.contract.conftest import schema
class TestGovernanceEndpoints:
"""Contract tests for agent governance endpoints."""
def test_list_governance_agents_contracts(self):
"""Test GET /api/agent-governance/agents conforms to OpenAPI spec."""
operation = schema["/api/agent-governance/agents"]["GET"]
with TestClient(app) as client:
response = client.get("/api/agent-governance/agents")
operation.validate_response(response)
# Schemathesis validates schema
assert response.status_code in [200, 401, 403, 404]
def test_governance_check_deployment_contracts(self):
"""Test POST /api/agent-governance/check-deployment conforms to OpenAPI spec."""
operation = schema["/api/agent-governance/check-deployment"]["POST"]
with TestClient(app) as client:
response = client.post(
"/api/agent-governance/check-deployment",
json={
"agent_id": "test-agent",
"workflow_name": "Test Workflow",
"actions": ["present_canvas"]
}
)
operation.validate_response(response)
assert response.status_code in [200, 400, 401, 422]
def test_governance_enforce_action_contracts(self):
"""Test POST /api/agent-governance/enforce-action conforms to OpenAPI spec."""
operation = schema["/api/agent-governance/enforce-action"]["POST"]
with TestClient(app) as client:
response = client.post(
"/api/agent-governance/enforce-action",
json={
"agent_id": "test-agent",
"action_type": "present_canvas",
"context": {}
}
)
operation.validate_response(response)
assert response.status_code in [200, 400, 401, 422]
class TestApprovalEndpoints:
"""Contract tests for approval workflow endpoints."""
def test_pending_approvals_contracts(self):
"""Test GET /api/agent-governance/pending-approvals conforms to OpenAPI spec."""
operation = schema["/api/agent-governance/pending-approvals"]["GET"]
with TestClient(app) as client:
response = client.get("/api/agent-governance/pending-approvals")
operation.validate_response(response)
assert response.status_code in [200, 401, 403, 404]
def test_submit_for_approval_contracts(self):
"""Test POST /api/agent-governance/submit-for-approval conforms to OpenAPI spec."""
operation = schema["/api/agent-governance/submit-for-approval"]["POST"]
with TestClient(app) as client:
response = client.post(
"/api/agent-governance/submit-for-approval",
json={
"agent_id": "test-agent",
"workflow_name": "Test Workflow",
"workflow_definition": {},
"trigger_type": "manual",
"actions": ["present_canvas"],
"requested_by": "test-user"
}
)
operation.validate_response(response)
assert response.status_code in [200, 201, 400, 401, 422, 500]
def test_approve_workflow_contracts(self):
"""Test POST /api/agent-governance/approve/{approval_id} conforms to OpenAPI spec."""
operation = schema["/api/agent-governance/approve/{approval_id}"]["POST"]
with TestClient(app) as client:
response = client.post("/api/agent-governance/approve/test-approval-id")
operation.validate_response(response)
assert response.status_code in [200, 401, 404, 422]
def test_reject_workflow_contracts(self):
"""Test POST /api/agent-governance/reject/{approval_id} conforms to OpenAPI spec."""
operation = schema["/api/agent-governance/reject/{approval_id}"]["POST"]
with TestClient(app) as client:
response = client.post("/api/agent-governance/reject/test-approval-id")
operation.validate_response(response)
assert response.status_code in [200, 401, 404, 422]
class TestFeedbackEndpoints:
"""Contract tests for agent feedback endpoints."""
def test_submit_feedback_contracts(self):
"""Test POST /api/agent-governance/feedback conforms to OpenAPI spec."""
operation = schema["/api/agent-governance/feedback"]["POST"]
with TestClient(app) as client:
response = client.post(
"/api/agent-governance/feedback",
json={
"agent_id": "test-agent",
"original_output": "Test output",
"user_correction": "Corrected output",
"input_context": "Test context"
}
)
operation.validate_response(response)
assert response.status_code in [200, 400, 401, 422, 500]
class TestGovernanceRulesEndpoints:
"""Contract tests for governance rules endpoints."""
def test_governance_rules_contracts(self):
"""Test GET /api/agent-governance/rules conforms to OpenAPI spec."""
operation = schema["/api/agent-governance/rules"]["GET"]
with TestClient(app) as client:
response = client.get("/api/agent-governance/rules")
operation.validate_response(response)
assert response.status_code in [200, 401, 403, 404]
|