""" Shared fixtures for API contract property-based tests. Provides API clients and Hypothesis strategies for testing: - Malformed JSON generation - Oversized payload strategies - API client wrapper with authenticated requests - Response validation helpers Fixture reuse: Imports authenticated_client from e2e_ui fixtures (10-100x faster than UI login) per FIXTURE_REUSE_GUIDE.md. """ import pytest from hypothesis import strategies as st from hypothesis import settings, HealthCheck # ============================================================================ # HYPOTHESIS SETTINGS FOR API CONTRACT TESTS # ============================================================================ # # Uses tiered settings based on test criticality and IO cost: # - CRITICAL: max_examples=200 (response validation, schema conformance) # - STANDARD: max_examples=100 (malformed JSON, input validation) # - IO_BOUND: max_examples=50 (oversized payloads, network operations) # # Per INVARIANTS.md guidelines for API contract testing. # ============================================================================ HYPOTHESIS_SETTINGS_CRITICAL = { "suppress_health_check": [HealthCheck.function_scoped_fixture, HealthCheck.too_slow], "max_examples": 200, # Critical invariants (response validation) "deadline": None # No timeout for critical tests } HYPOTHESIS_SETTINGS_STANDARD = { "suppress_health_check": [HealthCheck.function_scoped_fixture, HealthCheck.too_slow], "max_examples": 100, # Standard invariants (malformed JSON) "deadline": 10000 # 10 seconds per test } HYPOTHESIS_SETTINGS_IO = { "suppress_health_check": [HealthCheck.function_scoped_fixture, HealthCheck.too_slow], "max_examples": 50, # IO-bound operations (oversized payloads) "deadline": 30000 # 30 seconds per test } # ============================================================================ # API CLIENT FIXTURES (reuse from e2e_ui) # ============================================================================ # # Per FIXTURE_REUSE_GUIDE.md lines 117-174, we MUST import and reuse # existing auth fixtures from tests.e2e_ui.fixtures.auth_fixtures. # # DO NOT create new auth fixtures - reuse authenticated_user for # 10-100x faster authentication vs UI login. # ============================================================================ import sys from pathlib import Path # Add backend to path for imports sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent)) from tests.e2e_ui.fixtures.auth_fixtures import authenticated_user @pytest.fixture(scope="function") def api_auth_client(authenticated_user): """ Create an API client with authentication headers pre-configured. Reuses authenticated_user fixture from e2e_ui for 10-100x faster authentication (JWT token in headers vs UI login flow). Args: authenticated_user: Tuple[User, JWT token] from e2e_ui fixtures Returns: dict: Client config with headers and base URL Example: def test_api_call(api_auth_client): headers = api_auth_client["headers"] response = client.post("/api/v1/agents/execute", headers=headers, json={"test": "data"}) """ from fastapi.testclient import TestClient from main_api_app import app user, token = authenticated_user # Create TestClient with auth headers client_config = { "client": TestClient(app), "headers": {"Authorization": f"Bearer {token}"}, "user_id": str(user.id), "token": token } return client_config # ============================================================================ # MALFORMED JSON STRATEGIES # ============================================================================ # # Generates malformed JSON inputs for fuzzing API endpoints. # Tests invariant: Malformed JSON returns 400/422 (not 500) # ============================================================================ @pytest.fixture(scope="session") def malformed_json_strategy(): """ Strategy for generating malformed JSON inputs. Returns: st.SearchStrategy: Strategy generating various malformed JSON patterns Patterns: - Random text (not valid JSON) - Dict with None values (invalid JSON serialization) - Specifically malformed JSON strings - Empty/null values - Truncated JSON RADII: 100 examples covers common malformed patterns (text, None, incomplete JSON, empty payloads) without exhaustively testing all possible malformed inputs. """ return st.one_of( # Random text (not valid JSON) st.text(min_size=0, max_size=10000), # Dict with None values (invalid JSON) st.dictionaries(st.text(), st.none()), # Specifically malformed JSON strings st.just('{"invalid": json}'), st.just('{"unterminated":'), st.just('{"extra": "comma",}'), # Empty/null payloads st.just(''), st.just('null'), st.just('[]'), st.just('{}'), # Invalid UTF-8 sequences st.binary().filter(lambda b: b.decode('utf-8', errors='ignore') != b.decode('utf-8', errors='replace')) ) @pytest.fixture(scope="session") def injection_payload_strategy(): """ Strategy for generating injection attempt payloads. Returns: st.SearchStrategy: Strategy generating SQL/injection patterns Tests invariant: Injection attempts are sanitized (not 500 crash) RADII: 100 examples covers common SQL injection and XSS patterns. """ injection_patterns = [ # SQL injection "'; DROP TABLE users; --", "1' OR '1'='1", "admin'--", "' UNION SELECT * FROM users--", # XSS attempts "", "javascript:alert('xss')", "", # Command injection "; rm -rf /", "| cat /etc/passwd", "$(whoami)", # Path traversal "../../../etc/passwd", "..\\..\\..\\..\\windows\\system32\\", ] return st.one_of( st.sampled_from(injection_patterns), st.text().map(lambda s: s + "\x00") # Null bytes ) # ============================================================================ # OVERSIZED PAYLOAD STRATEGIES # ============================================================================ # # Generates oversized payloads for testing 413 responses. # Tests invariant: Oversized payloads return 413 (not OOM/crash) # ============================================================================ @pytest.fixture(scope="session") def oversized_payload_strategy(): """ Strategy for generating oversized payload sizes. Returns: st.SearchStrategy: Strategy generating payload sizes (1MB to 100MB) RADII: 50 examples covers size range from 1MB to 100MB (tests OOM protection without exhausting test machine memory). """ return st.integers(min_value=1_000_000, max_value=100_000_000) @pytest.fixture(scope="session") def oversized_json_strategy(): """ Strategy for generating oversized JSON payloads. Returns: st.SearchStrategy: Strategy generating oversized JSON dictionaries RADII: 50 examples covers various oversized structures (deep nesting, wide objects, large strings) without excessive test time. """ return st.one_of( # Large string value st.dictionaries( st.just("data"), st.text(min_size=1_000_000, max_size=10_000_000) ), # Deeply nested structure (stack overflow protection) st.recursive( st.just({}), lambda s: st.dictionaries(st.text(), s), max_leaves=50 ), # Wide object (many keys) st.dictionaries( st.text(min_size=5, max_size=20), st.text(min_size=100, max_size=1000), min_size=1000, max_size=10000 ) ) # ============================================================================ # RESPONSE VALIDATION HELPERS # ============================================================================ @pytest.fixture(scope="function") def assert_client_error(): """ Assertion helper for client error responses (not server errors). Args: response: FastAPI/requests Response object Raises: AssertionError: If response is 500 (server error) or 2xx (success) Example: def test_malformed_json(api_auth_client, assert_client_error): response = api_auth_client["client"].post( "/api/v1/agents/execute", headers=api_auth_client["headers"], json={"invalid": "json"} ) assert_client_error(response) # Must be 4xx, not 500 """ def _assert(response): # Must be client error (4xx), not server error (5xx) assert 400 <= response.status_code < 500, \ f"Expected client error (4xx), got {response.status_code}: {response.text[:200]}" # Specifically NOT 500 Internal Server Error assert response.status_code != 500, \ f"API returned 500 Internal Server Error (invariant violation): {response.text[:200]}" # Must not be 2xx success assert not (200 <= response.status_code < 300), \ f"Expected client error, got success {response.status_code}: {response.text[:200]}" return _assert @pytest.fixture(scope="function") def assert_payload_too_large(): """ Assertion helper for oversized payload responses (413). Args: response: FastAPI/requests Response object Raises: AssertionError: If response is not 413 or is 500 Example: def test_oversized_payload(api_auth_client, assert_payload_too_large): large_data = {"data": "x" * 50_000_000} response = api_auth_client["client"].post( "/api/v1/agents/execute", headers=api_auth_client["headers"], json=large_data ) assert_payload_too_large(response) # Must be 413 """ def _assert(response): # Must be 413 Payload Too Large or 400 Bad Request assert response.status_code in [400, 413], \ f"Expected 413 Payload Too Large or 400, got {response.status_code}: {response.text[:200]}" # Specifically NOT 500 Internal Server Error assert response.status_code != 500, \ f"API returned 500 Internal Server Error (invariant violation): {response.text[:200]}" return _assert @pytest.fixture(scope="function") def assert_response_schema(): """ Assertion helper for response schema validation. Args: response: FastAPI/requests Response object required_fields: List of required field names Raises: AssertionError: If response missing required fields or invalid content-type Example: def test_response_schema(api_auth_client, assert_response_schema): response = api_auth_client["client"].get( "/api/v1/agents", headers=api_auth_client["headers"] ) assert_response_schema(response, required_fields=["agents", "total"]) """ def _assert(response, required_fields=None): # Check content-type is JSON assert "application/json" in response.headers.get("content-type", ""), \ f"Response content-type must be JSON, got: {response.headers.get('content-type')}" # Check response can be parsed as JSON try: data = response.json() except Exception as e: raise AssertionError(f"Response body is not valid JSON: {e}") # Check required fields present if required_fields: missing_fields = [f for f in required_fields if f not in data] assert len(missing_fields) == 0, \ f"Response missing required fields: {missing_fields}. Got keys: {list(data.keys())}" return _assert