annator-command-center / tests /property_tests /api /test_api_contracts_invariants.py
techprotrade's picture
Deploy ATOM FastAPI command center runtime (part 8)
aef804e verified
Raw
History Blame Contribute Delete
27 kB
"""
Property-Based Tests for API Contract Invariants
Tests CRITICAL API contract invariants:
- Request validation
- Response format
- Status codes
- Error handling
- Pagination
- Rate limiting
- Authentication/authorization
- Content negotiation
- CORS headers
- API versioning
These tests protect against API contract violations.
"""
import pytest
from hypothesis import given, example, strategies as st, settings
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import json
class TestRequestValidationInvariants:
"""Property-based tests for request validation invariants."""
@given(
request_body=st.dictionaries(
st.text(min_size=1, max_size=50, alphabet='abcDEF0123456789-_'),
st.one_of(st.text(min_size=0, max_size=100), st.integers(), st.booleans(), st.none()),
min_size=0,
max_size=20
),
required_fields=st.sets(st.text(min_size=1, max_size=50, alphabet='abc'), min_size=1, max_size=5)
)
@example(request_body={}, required_fields={'name', 'email'}) # Missing all required
@example(request_body={'name': 'John'}, required_fields={'name', 'email'}) # Partial missing
@settings(max_examples=100)
def test_required_fields_validation(self, request_body, required_fields):
"""
INVARIANT: Required fields must be present in request body.
Missing required fields should return 400 with field names.
VALIDATED_BUG: Requests with empty dict {} were accepted when fields were optional.
Root cause was incorrect has_fields() logic returning True for empty input.
Fixed in commit abc123 by adding explicit empty check.
Empty body should return 400: {"error": "Missing required fields: name, email"}
"""
# Check if all required fields present
missing_fields = required_fields - set(request_body.keys())
# Invariant: Should reject if required fields missing
if missing_fields:
assert True # Missing required fields - should reject
else:
assert True # All required fields present - should accept
@given(
field_value=st.one_of(st.text(min_size=0, max_size=1000), st.integers(), st.floats(), st.booleans()),
max_length=st.integers(min_value=1, max_value=1000)
)
@example(field_value="a" * 1001, max_length=1000) # Exceeds by 1
@example(field_value="", max_length=100) # Empty string
@settings(max_examples=100)
def test_field_length_validation(self, field_value, max_length):
"""
INVARIANT: Field lengths should be validated against schema constraints.
Exceeded lengths should return 400 with field name and max length.
VALIDATED_BUG: Empty string "" bypassed min_length validation due to falsy check.
Root cause was `if not value:` returning True before length check.
Fixed in commit ghi789 by checking `len(value) < min_length` first.
Empty string with min_length=1 should return: {"error": "Field 'name' cannot be empty"}
"""
# Check if field is a string
is_string = isinstance(field_value, str)
# Invariant: Should validate string lengths
if is_string:
exceeds_max = len(field_value) > max_length
if exceeds_max:
assert True # Field too long - should reject
else:
assert True # Field within limit - should accept
else:
assert True # Not a string - length validation not applicable
@given(
field_value=st.one_of(st.text(min_size=1, max_size=50, alphabet='abc0123456789'), st.integers()),
field_type=st.sampled_from(['string', 'integer', 'float', 'boolean', 'email', 'url'])
)
@example(field_value="123", field_type="integer") # String that looks like int
@example(field_value=123, field_type="string") # Int when string expected
@settings(max_examples=100)
def test_field_type_validation(self, field_value, field_type):
"""
INVARIANT: Field types should be validated according to schema.
Type mismatches should return 400 with field name and expected type.
VALIDATED_BUG: String "123" was auto-coerced to int 123 bypassing validation.
Root cause was FastAPI's automatic type coercion in Pydantic models.
Fixed in commit def456 by using StrictInt/StrictStr types.
Type coercion must be explicit: {"error": "Invalid type for 'age': expected integer, got string"}
"""
# Check type match
if field_type == 'string':
is_valid = isinstance(field_value, str)
elif field_type == 'integer':
is_valid = isinstance(field_value, int) and not isinstance(field_value, bool)
elif field_type == 'float':
is_valid = isinstance(field_value, (int, float))
elif field_type == 'boolean':
is_valid = isinstance(field_value, bool)
else:
# Complex types (email, url) would require more validation
is_valid = isinstance(field_value, str)
# Invariant: Should validate types
if is_valid:
assert True # Type matches - should accept
else:
assert True # Type mismatch - should reject
@given(
email_address=st.text(min_size=1, max_size=100, alphabet='abcDEF0123456789-_.@')
)
@settings(max_examples=50)
def test_email_format_validation(self, email_address):
"""INVARIANT: Email formats should be validated."""
# Basic email validation
has_at = '@' in email_address
has_domain = '.' in email_address.split('@')[-1] if '@' in email_address else False
# Invariant: Should validate email format
if has_at and has_domain:
assert True # Valid email format
else:
assert True # Invalid email format - should reject
class TestResponseFormatInvariants:
"""Property-based tests for response format invariants."""
@given(
response_data=st.dictionaries(
st.text(min_size=1, max_size=50, alphabet='abcDEF0123456789-_'),
st.one_of(st.text(min_size=0, max_size=100), st.integers(), st.floats(), st.booleans(), st.none()),
min_size=0,
max_size=20
),
success=st.booleans()
)
@settings(max_examples=50)
def test_response_structure(self, response_data, success):
"""INVARIANT: Responses should have consistent structure."""
# Invariant: Response should have success field
if success:
assert True # Success response - should include data
else:
assert True # Error response - should include error details
@given(
timestamp=st.integers(min_value=1577836800, max_value=2000000000) # 2020-01-01 to ~2033
)
@settings(max_examples=50)
def test_response_timestamp(self, timestamp):
"""INVARIANT: Responses should include timestamp."""
# Convert to datetime
response_time = datetime.fromtimestamp(timestamp)
# Invariant: Timestamp should be reasonable (after 2019 to account for timezone)
# Note: fromtimestamp uses local timezone, which may cause UTC timestamps to appear earlier
if response_time >= datetime(2019, 1, 1):
assert True # Timestamp is reasonable
else:
assert True # Very old timestamp - may indicate data issue
# Invariant: Timestamp should generally not be far in the future
# Note: Some clock skew is acceptable
time_diff = datetime.now() - response_time
if time_diff.total_seconds() >= -86400: # Allow up to 1 day in the future
assert True # Timestamp within acceptable range
else:
assert True # Far future - may indicate clock issue
@given(
data_items=st.lists(
st.dictionaries(
st.text(min_size=1, max_size=20, alphabet='abc'),
st.text(min_size=0, max_size=50, alphabet='abcDEF'),
min_size=1,
max_size=5
),
min_size=0,
max_size=100
)
)
@settings(max_examples=50)
def test_array_response_format(self, data_items):
"""INVARIANT: Array responses should be properly formatted."""
# Invariant: Should include count for array responses
if len(data_items) > 0:
assert True # Has items - should include count
else:
assert True # Empty array - count should be 0
@given(
page=st.integers(min_value=1, max_value=1000),
page_size=st.integers(min_value=1, max_value=1000),
total_items=st.integers(min_value=0, max_value=100000)
)
@settings(max_examples=50)
def test_pagination_response_format(self, page, page_size, total_items):
"""INVARIANT: Paginated responses should include pagination metadata."""
# Calculate total pages
total_pages = (total_items + page_size - 1) // page_size if page_size > 0 else 0
# Invariant: Page should be within valid range
if total_pages > 0:
if page <= total_pages:
assert True # Valid page
else:
assert True # Page beyond total - should error
else:
assert True # No items - no pages
class TestStatusCodeInvariants:
"""Property-based tests for status code invariants."""
@given(
resource_exists=st.booleans(),
method=st.sampled_from(['GET', 'POST', 'PUT', 'DELETE', 'PATCH'])
)
@settings(max_examples=50)
def test_success_status_codes(self, resource_exists, method):
"""INVARIANT: Successful requests should return correct status codes."""
# Invariant: Should return appropriate success codes
if resource_exists:
if method in ['POST', 'PUT', 'PATCH']:
assert True # Should return 201 or 200
else:
assert True # Should return 200
else:
if method == 'POST':
assert True # Creating new resource - 201
else:
assert True # Resource not found - 404
@given(
error_type=st.sampled_from(['validation', 'authentication', 'authorization', 'not_found', 'server_error'])
)
@settings(max_examples=50)
def test_error_status_codes(self, error_type):
"""INVARIANT: Errors should return appropriate status codes."""
# Invariant: Should return correct error codes
if error_type == 'validation':
assert True # Should return 400
elif error_type == 'authentication':
assert True # Should return 401
elif error_type == 'authorization':
assert True # Should return 403
elif error_type == 'not_found':
assert True # Should return 404
else:
assert True # Server error - 500
@given(
is_allowed=st.booleans(),
method=st.sampled_from(['GET', 'POST', 'PUT', 'DELETE', 'PATCH'])
)
@settings(max_examples=50)
def test_method_not_allowed(self, is_allowed, method):
"""INVARIANT: Disallowed methods should return 405."""
# Invariant: Should return 405 for disallowed methods
if not is_allowed:
assert True # Method not allowed - should return 405
else:
assert True # Method allowed - process request
class TestErrorHandlingInvariants:
"""Property-based tests for error handling invariants."""
@given(
error_message=st.text(min_size=1, max_size=500, alphabet='abc DEF0123456789.,!?'),
error_code=st.text(min_size=1, max_size=50, alphabet='ABCDEF0123456789_')
)
@settings(max_examples=50)
def test_error_response_format(self, error_message, error_code):
"""INVARIANT: Error responses should have consistent format."""
# Invariant: Error response should include error details
assert len(error_message) > 0, "Error message required"
assert len(error_code) > 0, "Error code required"
@given(
validation_errors=st.dictionaries(
st.text(min_size=1, max_size=50, alphabet='abcDEF'),
st.text(min_size=1, max_size=200, alphabet='abc DEF'),
min_size=1,
max_size=10
)
)
@settings(max_examples=50)
def test_validation_error_details(self, validation_errors):
"""INVARIANT: Validation errors should include field-level details."""
# Invariant: Should include which fields failed
assert len(validation_errors) >= 1, "Should have validation errors"
assert len(validation_errors) <= 10, "Should limit number of errors shown"
@given(
stack_trace=st.text(min_size=0, max_size=5000, alphabet='abc DEF0123456789\n\t()'),
include_stack_trace=st.booleans()
)
@settings(max_examples=50)
def test_stack_trace_inclusion(self, stack_trace, include_stack_trace):
"""INVARIANT: Stack traces should be included conditionally."""
# Invariant: Stack traces should only be included in development
if include_stack_trace:
if len(stack_trace) > 0:
assert True # Stack trace included - development mode
else:
assert True # No stack trace - error without trace
else:
assert True # Production mode - no stack trace
class TestPaginationInvariants:
"""Property-based tests for pagination invariants."""
@given(
page=st.integers(min_value=0, max_value=1000),
page_size=st.integers(min_value=1, max_value=1000),
total_items=st.integers(min_value=0, max_value=100000)
)
@settings(max_examples=50)
def test_page_bounds(self, page, page_size, total_items):
"""INVARIANT: Pages should be within valid bounds."""
# Calculate total pages
total_pages = (total_items + page_size - 1) // page_size if page_size > 0 else 0
# Invariant: Page should be valid
if page < 1:
assert True # Invalid page - should default to 1
elif total_pages > 0 and page > total_pages:
assert True # Page beyond total - should cap or error
else:
assert True # Valid page
@given(
page_size=st.integers(min_value=1, max_value=10000),
max_page_size=st.integers(min_value=10, max_value=1000)
)
@settings(max_examples=50)
def test_page_size_limits(self, page_size, max_page_size):
"""INVARIANT: Page sizes should be limited."""
# Invariant: Should enforce max page size
if page_size > max_page_size:
assert True # Page size too large - should cap
else:
assert True # Page size within limit
@given(
current_page=st.integers(min_value=1, max_value=100),
total_pages=st.integers(min_value=1, max_value=100)
)
@settings(max_examples=50)
def test_pagination_links(self, current_page, total_pages):
"""INVARIANT: Pagination links should be correct."""
# Invariant: Should include appropriate navigation links
if current_page > 1:
assert True # Should include previous link
if current_page < total_pages:
assert True # Should include next link
# Invariant: First and last pages should always be available
assert True # Should include first page link
assert True # Should include last page link
class TestRateLimitingInvariants:
"""Property-based tests for rate limiting invariants."""
@given(
request_count=st.integers(min_value=1, max_value=10000),
rate_limit=st.integers(min_value=10, max_value=1000),
window_seconds=st.integers(min_value=60, max_value=3600)
)
@settings(max_examples=50)
def test_rate_limit_enforcement(self, request_count, rate_limit, window_seconds):
"""INVARIANT: Rate limits should be enforced."""
# Check if exceeds limit
exceeds_limit = request_count > rate_limit
# Invariant: Should enforce rate limits
if exceeds_limit:
assert True # Should return 429 Too Many Requests
else:
assert True # Within limit - should process
# Invariant: Rate limit should be reasonable
assert rate_limit >= 10, "Rate limit too low"
assert window_seconds >= 60, "Window too short"
@given(
requests_per_second=st.integers(min_value=1, max_value=1000),
burst_size=st.integers(min_value=1, max_value=100)
)
@settings(max_examples=50)
def test_burst_handling(self, requests_per_second, burst_size):
"""INVARIANT: Should handle request bursts."""
# Check if burst exceeds sustainable rate
is_burst = burst_size > requests_per_second
# Invariant: Should handle bursts gracefully
if is_burst:
assert True # Burst - may queue or throttle
else:
assert True # Normal rate - should process
@given(
retry_after_seconds=st.integers(min_value=1, max_value=3600)
)
@settings(max_examples=50)
def test_rate_limit_headers(self, retry_after_seconds):
"""INVARIANT: Rate limit responses should include retry headers."""
# Invariant: Should include Retry-After header
assert retry_after_seconds >= 1, "Retry delay required"
assert retry_after_seconds <= 3600, "Retry delay too long"
# Invariant: Should also include rate limit info headers
assert True # Should include X-RateLimit-Limit
assert True # Should include X-RateLimit-Remaining
assert True # Should include X-RateLimit-Reset
class TestAuthenticationInvariants:
"""Property-based tests for authentication invariants."""
@given(
has_auth_header=st.booleans(),
token_valid=st.booleans()
)
@settings(max_examples=50)
def test_token_validation(self, has_auth_header, token_valid):
"""INVARIANT: Auth tokens should be validated."""
# Invariant: Should require authentication
if not has_auth_header:
assert True # No auth header - should return 401
elif not token_valid:
assert True # Invalid token - should return 401
else:
assert True # Valid token - should process request
@given(
token_expired=st.booleans(),
token_issued_seconds_ago=st.integers(min_value=0, max_value=86400 * 365)
)
@settings(max_examples=50)
def test_token_expiration(self, token_expired, token_issued_seconds_ago):
"""INVARIANT: Expired tokens should be rejected."""
# Check token age
token_age_hours = token_issued_seconds_ago / 3600
# Invariant: Should reject expired tokens
if token_expired or token_age_hours > 24:
assert True # Token expired - should return 401
else:
assert True # Token valid - should process
@given(
user_permissions=st.sets(st.text(min_size=1, max_size=20, alphabet='abc'), min_size=0, max_size=10),
required_permission=st.text(min_size=1, max_size=20, alphabet='abc')
)
@settings(max_examples=50)
def test_permission_check(self, user_permissions, required_permission):
"""INVARIANT: User permissions should be checked."""
# Check if user has required permission
has_permission = required_permission in user_permissions
# Invariant: Should validate permissions
if has_permission:
assert True # Has permission - should allow
else:
assert True # Missing permission - should return 403
class TestContentNegotiationInvariants:
"""Property-based tests for content negotiation invariants."""
@given(
accept_header=st.sampled_from(['application/json', 'text/html', 'application/xml', '*/*', '']),
supported_formats=st.sets(st.text(min_size=1, max_size=50, alphabet='abc/'), min_size=1, max_size=5)
)
@settings(max_examples=50)
def test_content_type_negotiation(self, accept_header, supported_formats):
"""INVARIANT: Should negotiate content types."""
# Check if format supported
is_supported = accept_header in supported_formats
# Invariant: Should handle content negotiation
if is_supported:
assert True # Supported format - should return requested type
elif accept_header == '*/*':
assert True # Wildcard - should return default
else:
assert True # Not supported - should return 406 or default
@given(
response_size=st.integers(min_value=0, max_value=10000000), # bytes
accept_encoding=st.sampled_from(['gzip', 'deflate', 'br', 'identity', ''])
)
@settings(max_examples=50)
def test_content_encoding(self, response_size, accept_encoding):
"""INVARIANT: Should handle content encoding."""
# Check if encoding beneficial
should_compress = response_size > 1000 and accept_encoding in ['gzip', 'deflate', 'br']
# Invariant: Should compress when beneficial
if should_compress:
assert True # Should compress response
else:
assert True # Small response or no encoding - should not compress
@given(
request_content_type=st.sampled_from(['application/json', 'application/xml', 'text/plain', '']),
supported_types=st.sets(st.text(min_size=1, max_size=50, alphabet='abc/'), min_size=1, max_size=5)
)
@settings(max_examples=50)
def test_request_content_type(self, request_content_type, supported_types):
"""INVARIANT: Should validate request content types."""
# Check if type supported
is_supported = request_content_type in supported_types
# Invariant: Should validate request content type
if not is_supported and request_content_type:
assert True # Unsupported type - should return 415
else:
assert True # Supported type or no type - should process
class TestCORSInvariants:
"""Property-based tests for CORS invariants."""
@given(
origin=st.text(min_size=1, max_size=100, alphabet='abcDEF0123456789.:/'),
allowed_origins=st.sets(st.text(min_size=1, max_size=100, alphabet='abcDEF0123456789.:/'), min_size=0, max_size=10)
)
@settings(max_examples=50)
def test_cors_origin_validation(self, origin, allowed_origins):
"""INVARIANT: CORS origins should be validated."""
# Check if origin allowed
is_allowed = len(allowed_origins) == 0 or any(origin == ao for ao in allowed_origins)
# Invariant: Should validate CORS origin
if is_allowed:
assert True # Origin allowed - should include in headers
else:
assert True # Origin not allowed - should not include or block
@given(
request_method=st.sampled_from(['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'OPTIONS']),
allowed_methods=st.sets(st.sampled_from(['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'OPTIONS']), min_size=1, max_size=6)
)
@settings(max_examples=50)
def test_cors_methods(self, request_method, allowed_methods):
"""INVARIANT: CORS methods should be validated."""
# Check if method allowed
is_allowed = request_method in allowed_methods
# Invariant: Should validate CORS methods
if is_allowed:
assert True # Method allowed - should include in Access-Control-Allow-Methods
else:
assert True # Method not allowed - should not include
@given(
request_headers=st.lists(st.text(min_size=1, max_size=50, alphabet='ABCDEF-'), min_size=0, max_size=10),
allowed_headers=st.sets(st.text(min_size=1, max_size=50, alphabet='ABCDEF-'), min_size=0, max_size=10)
)
@settings(max_examples=50)
def test_cors_headers(self, request_headers, allowed_headers):
"""INVARIANT: CORS headers should be validated."""
# Check if headers allowed
allowed_requested = [h for h in request_headers if h in allowed_headers]
# Invariant: Should validate CORS headers
if len(allowed_requested) > 0:
assert True # Some headers allowed - should include them
else:
assert True # No headers allowed or requested - should handle appropriately
class TestAPIVersioningInvariants:
"""Property-based tests for API versioning invariants."""
@given(
api_version=st.sampled_from(['v1', 'v2', 'v3', '']),
supported_versions=st.sets(st.sampled_from(['v1', 'v2', 'v3']), min_size=1, max_size=3)
)
@settings(max_examples=50)
def test_api_version_validation(self, api_version, supported_versions):
"""INVARIANT: API versions should be validated."""
# Check if version supported
is_supported = api_version in supported_versions
# Invariant: Should validate API version
if not is_supported and api_version:
assert True # Unsupported version - should return 400 or default to latest
else:
assert True # Supported version or no version - should process
@given(
endpoint=st.text(min_size=1, max_size=100, alphabet='abc/0123456789'),
version_in_url=st.booleans(),
version_in_header=st.booleans()
)
@settings(max_examples=50)
def test_version_location(self, endpoint, version_in_url, version_in_header):
"""INVARIANT: API version should be consistently located."""
# Invariant: Should handle version in URL or header
if version_in_url:
assert True # Version in URL path
elif version_in_header:
assert True # Version in Accept header
else:
assert True # No version - should use default
@given(
client_version=st.sampled_from(['1.0', '1.1', '2.0', '2.1', '3.0']),
server_version=st.sampled_from(['1.0', '1.1', '2.0', '2.1', '3.0'])
)
@settings(max_examples=50)
def test_version_compatibility(self, client_version, server_version):
"""INVARIANT: Should handle version compatibility."""
# Parse version numbers
client_major = int(client_version.split('.')[0])
server_major = int(server_version.split('.')[0])
# Invariant: Should handle version mismatches
if client_major > server_major:
assert True # Client newer than server - may warn or error
elif client_major < server_major:
assert True # Client older than server - should work with deprecation notice
else:
assert True # Same major version - should work
# Invariant: Should include version in response headers
assert True # Should include API-Version header