File size: 3,801 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 | """Contract tests for core 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 TestHealthEndpoints:
"""Contract tests for health check endpoints."""
def test_health_endpoint_contracts(self):
"""Test /health endpoint conforms to OpenAPI spec."""
# Get the API operation for /health
operation = schema["/health"]["GET"]
with TestClient(app) as client:
response = client.get("/health")
# Validate response against OpenAPI schema
operation.validate_response(response)
# Schemathesis validates schema, we check business logic
# Health endpoint should return 200 (healthy) or 503 (unhealthy)
assert response.status_code in [200, 503]
def test_api_v1_health_contracts(self):
"""Test /api/v1/health endpoint conforms to OpenAPI spec."""
# Get the API operation for /api/v1/health
operation = schema["/api/v1/health"]["GET"]
with TestClient(app) as client:
response = client.get("/api/v1/health")
# Validate response against OpenAPI schema
operation.validate_response(response)
# May return 200 or 404 depending on setup
assert response.status_code in [200, 404]
def test_root_endpoint_contracts(self):
"""Test root endpoint conforms to OpenAPI spec."""
# Get the API operation for /
operation = schema["/"]["GET"]
with TestClient(app) as client:
response = client.get("/")
# Validate response against OpenAPI schema
operation.validate_response(response)
# Root endpoint should return 200
assert response.status_code == 200
class TestAgentEndpoints:
"""Contract tests for agent endpoints."""
def test_list_agents_contracts(self):
"""Test GET /api/agents/ conforms to OpenAPI spec."""
# Get the API operation for GET /api/agents/
operation = schema["/api/agents/"]["GET"]
with TestClient(app) as client:
response = client.get("/api/agents/")
# Validate response against OpenAPI schema
operation.validate_response(response)
# May return 200, 401, 403 for auth, or 404 if route doesn't exist
assert response.status_code in [200, 401, 403, 404]
def test_get_agent_contracts(self):
"""Test GET /api/agents/{id} conforms to OpenAPI spec."""
# Get the API operation for GET /api/agents/{agent_id}
operation = schema["/api/agents/{agent_id}"]["GET"]
with TestClient(app) as client:
response = client.get("/api/agents/test-agent-id")
# Validate response against OpenAPI schema
operation.validate_response(response)
# May return 200, 401, 403, or 404 if agent not found or route doesn't exist
assert response.status_code in [200, 401, 403, 404]
def test_create_agent_contracts(self):
"""Test POST /api/agents/spawn conforms to OpenAPI spec."""
# Get the API operation for POST /api/agents/spawn
operation = schema["/api/agents/spawn"]["POST"]
with TestClient(app) as client:
response = client.post(
"/api/agents/spawn",
json={
"agent_id": "test-spawn-agent",
"config": {}
}
)
# Validate response against OpenAPI schema
operation.validate_response(response)
# May return 200, 400, 401, 403, or 404 if route doesn't exist
assert response.status_code in [200, 400, 401, 403, 404]
|