Qa Api Testing Contracts

QA API Testing and Contracts

Use this skill to turn an API schema into enforceable checks (lint, diff, contracts, and negative/security cases) and wire them into CI so breaking changes cannot ship silently.

Ask For Inputs

• API type and canonical schema artifact (OpenAPI 3.1, SDL, proto) and where it lives in-repo.
• Environments, auth method(s), and how to provision stable test identities/keys.
• Critical endpoints/operations and business flows (rank by risk and revenue impact).
• Data constraints (idempotency keys, pagination, ordering), rate limits, and error format (prefer RFC 7807 application/problem+json for REST).
• Versioning + deprecation policy, consumer inventory, and release cadence.
• Current test tooling/CI and what “blocking” means for your org.

Outputs (What to Produce)

• A minimal CI gate set (lint + breaking diff + contract suite) wired to PRs.
• A coverage map derived from the schema (critical operations first).
• A negative/security baseline aligned to OWASP API risks.

Quick Start

1. Lint the schema (syntax + best-practice rules) and fix issues before writing tests. 2. Add breaking-change checks against the base branch on every PR. 3. Pick a contract strategy (CDC, schema-driven, or both) and run it in CI against an ephemeral environment. 4. Add negative/security cases for auth, validation, and error handling. 5. Make gates explicit (what blocks merge/release) and publish results.

Workflow

1) Establish Contract Artifacts (Source of Truth)

• REST: single OpenAPI 3.1 file or a compiled artifact; avoid drift across fragments.
• GraphQL: checked-in SDL (and federation/composition config if relevant).
• gRPC: checked-in .proto + buf.yaml (or equivalent) with a stable module layout.

2) Validate the Schema (Fast, Deterministic)

• Run spec linting (Spectral / GraphQL Inspector / buf lint).
• Enforce a small, explicit ruleset (naming, descriptions, auth annotations, consistent error model).

3) Detect Breaking Changes (PR Gate)

• REST: OpenAPI diff with a breaking-change policy (remove/rename/type change/requiredness).
• GraphQL: schema diff with breaking checks (field removals, type changes, non-null tightening).
• gRPC: buf breaking (do not reuse/renumber fields; avoid changing request/response shapes incompatibly).

4) Execute Contract Tests (CI Gate)

Choose one or combine:

• CDC (Pact): best when many independent consumers exist and behavior matters beyond schema.
• Schema-driven (Specmatic): best when schema is the contract and you want fast coverage across operations.
• Property-based (Schemathesis): best when you want systematic edge cases and server hardening.

5) Add Negative + Security Cases (Minimum Set)

• AuthN/AuthZ: missing/expired token (401), insufficient scope/role (403), tenant isolation.
• Validation: missing required fields, invalid types, boundary values, empty strings, large payloads.
• Error handling: stable error shape, safe messages, correct status codes, correlation IDs.
• Abuse & limits: rate limiting (429), pagination limits, idempotency replay, retry-safe semantics.
• For GraphQL, also validate operations checks (known/persisted queries) if you have an operation registry (GraphOS/Hive/etc.).

6) Define CI Quality Gates (Merge + Release)

• Pre-merge: schema lint + breaking-change diff (blocking).
• Pre-release: contract suite (blocking), plus smoke/functional tests for critical flows.
• Reporting: publish artifacts (diff report, contract verification, failing cases) and link in PR.

Quality Checks

• Fail fast: schema violations and breaking changes block merge.
• Determinism: isolate data, freeze time where needed, avoid shared mutable fixtures.
• Flake hygiene: separate network instability from contract failures; retry only for known-transient classes.
• Alignment: contracts reflect versioning/deprecation policy and consumer inventory.
• Scope control: keep load/resilience tests separate unless explicitly requested.

Use the Bundled Templates

• Coverage plan: assets/api-test-plan.md
• Release review: assets/contract-change-checklist.md
• Tooling map: assets/schema-validation-matrix.md

AI Assistance (Use Carefully)

• Use AI to draft tests, suggest missing edge cases, and tighten matchers.
• Treat AI output as untrusted until verified against the schema and real behavior.
• Avoid uploading sensitive payloads; sanitize examples and logs.
• For a tool comparison and workflows, read references/ai-contract-testing.md.

Read These When Needed

• Change safety and CDC patterns: references/contract-testing-patterns.md
• AI-assisted tooling and decision matrix: references/ai-contract-testing.md
• API versioning and backward compatibility: references/api-versioning-strategies.md
• Schema-driven and property-based testing: references/schema-driven-testing.md
• OWASP API security testing: references/api-security-testing.md
• Curated authoritative links: data/sources.json

Related Skills

• Use dev-api-design for API design decisions.
• Use qa-testing-strategy for overall testing strategy.
• Use qa-resilience for chaos and reliability testing.
• Use software-security-appsec for API security review.

Fact-Checking

• Use web search/web fetch to verify current external facts, versions, pricing, deadlines, regulations, or platform behavior before final answers.
• Prefer primary sources; report source links and dates for volatile information.
• If web access is unavailable, state the limitation and mark guidance as unverified.

API Test Plan

Overview

Environments

Coverage Map

Test Types Checklist

• [ ] Schema validation (response matches OpenAPI/SDL/Proto)
• [ ] Happy path (valid inputs, expected outputs)
• [ ] Negative testing (invalid inputs, error responses)
• [ ] Authentication and authorization
• [ ] Idempotency (POST/PUT/DELETE safety)
• [ ] Rate limiting and throttling
• [ ] Pagination and cursors
• [ ] Timeout and retry behavior
• [ ] Backward compatibility

Data Strategy

CI Quality Gates

Tools

Contract Change Checklist

Use this checklist before releasing API changes to production.

Change Summary

Change Classification

Breaking Changes (Require version bump)

• [ ] Removing fields or endpoints
• [ ] Changing field types (string to int, etc.)
• [ ] Changing required fields (optional to required)
• [ ] Renaming enum values
• [ ] Changing default behavior
• [ ] Reducing allowed values in enums
• [ ] Changing authentication requirements

Non-Breaking Changes (Safe to release)

• [ ] Adding optional fields
• [ ] Adding new endpoints
• [ ] Adding new enum values (with backwards-compatible defaults)
• [ ] Deprecating fields (with migration period)
• [ ] Performance improvements (same contract)

Compatibility Assessment

Validation Checklist

• [ ] Schema updated (OpenAPI/SDL/Proto)
• [ ] Schema validation passes
• [ ] Consumer contract tests pass
• [ ] Provider contract tests pass
• [ ] Existing integration tests pass
• [ ] Mocks updated to match new schema
• [ ] Documentation updated
• [ ] Release notes drafted
• [ ] Consumers notified (if breaking)

Rollback Plan

Approvals

Schema Validation Matrix

Map your APIs to validation tools and CI stages.

API Inventory

Validation Levels

Tool Configuration

REST (OpenAPI)

GraphQL (SDL)

If you need mocked GraphQL responses in tests, prefer schema-aware mocks (for example, @graphql-tools/mock) or request-level mocking (for example, msw) so the mock stays aligned with the SDL.

gRPC (Proto)

CI Pipeline Stages

# Example GitHub Actions
validate-api:
  steps:
    - uses: actions/checkout@v4
      with:
        fetch-depth: 0

    - name: Materialize base spec
      run: git show "origin/${{ github.base_ref }}:specs/api.yaml" > /tmp/api.base.yaml

    - name: Lint OpenAPI
      run: spectral lint specs/*.yaml --ruleset .spectral.yaml

    - name: Check breaking changes
      run: oasdiff breaking /tmp/api.base.yaml specs/api.yaml

    - name: Contract tests
      run: pact-verifier --provider-base-url=$API_URL

    - name: Property tests
      run: schemathesis run specs/api.yaml --base-url=$API_URL

Ruleset Examples

Spectral (.spectral.yaml)

extends: ["spectral:oas", "spectral:asyncapi"]
rules:
  operation-operationId: error
  operation-description: warn
  info-contact: warn

Buf (buf.yaml)

version: v1
breaking:
  use:
    - FILE
lint:
  use:
    - DEFAULT

AI-Powered Validation (2026)

See ../references/ai-contract-testing.md for setup guides.

AI-Powered Contract Testing (2026)

Overview

AI-powered tools now automate contract test generation, maintenance, and validation. This guide covers the leading approaches and when to use each.

Guardrails

• Keep deterministic gates (schema lint + breaking diff + contract verification) as the source of truth; AI suggestions must pass them.
• Treat generated tests as a starting point; tighten matchers to avoid flaky or overfitted contracts.
• Sanitize payloads/logs before sharing with third-party tools; never include credentials or PII.

Tool Comparison

PactFlow AI Code Review

PactFlow's AI inspects Pact tests and suggests improvements.

What It Checks

• Contract completeness (missing scenarios)
• Matcher usage (overly strict vs too loose)
• State handler coverage
• Best practice violations

Setup

// Enable AI review in PactFlow settings
// Runs automatically on contract publish

// Example: AI detects overly strict matcher
// Before (flagged)
willRespondWith: {
  body: { id: "123", name: "John" }  // Exact match
}

// After (AI suggestion)
willRespondWith: {
  body: {
    id: Matchers.string("123"),
    name: Matchers.string("John")
  }
}

Notes

• Feature availability, pricing, and supported languages change frequently; confirm current capabilities in vendor docs.
• Treat AI output as untrusted until it passes contract verification and your own review.
• Avoid uploading sensitive payloads or credentials; sanitize examples and logs.

Bi-Directional Contract Testing

New paradigm combining provider-driven and consumer-driven approaches.

Traditional CDC Flow

Consumer → Contract → Broker → Provider Verification

Bi-Directional Flow

Consumer → Contract ─┐
                     ├─► Broker ◄─► Comparison
Provider → OpenAPI ──┘

When to Use

Setup Example

# pact-config.yml for bi-directional
provider:
  name: UserService
  specification:
    type: openapi
    path: ./specs/user-api.yaml

consumer:
  name: WebApp
  contracts:
    - path: ./pacts/webapp-userservice.json

Keploy: Traffic-Based Test Generation

Keploy captures live API traffic and generates tests automatically.

How It Works

1. Attach to running service
2. Capture real requests/responses
3. Generate test cases with assertions
4. Replay in CI without test environment

Setup

# Install Keploy
curl -O https://raw.githubusercontent.com/keploy/keploy/main/keploy.sh
chmod +x keploy.sh

# Record tests from running app
keploy record -c "npm start"

# Replay in CI
keploy test -c "npm start" --delay 10

Generated Test Example

# keploy/tests/test-1.yaml (auto-generated)
version: api.keploy.io/v1beta2
kind: Http
name: test-1
spec:
  request:
    method: GET
    url: /api/users/123
    headers:
      Authorization: Bearer ***
  response:
    status_code: 200
    body: |
      {"id": "123", "name": "John", "email": "john@example.com"}
  assertions:
    - type: status_code
      expected: 200
    - type: body_contains
      expected: '"id": "123"'

Best Use Cases

• Legacy APIs with no existing tests
• Complex integration flows
• Rapid test coverage bootstrap
• Regression test generation

Postman Postbot

AI assistant for test generation within Postman.

Capabilities

• Analyze request/response and generate assertions
• Suggest edge cases and negative scenarios
• Auto-generate documentation
• Debug failing requests

Example Workflow

// 1. Send request in Postman
// 2. Postbot analyzes response
// 3. Suggests tests:

pm.test("Status is 200", () => {
  pm.response.to.have.status(200);
});

pm.test("Response has required fields", () => {
  const json = pm.response.json();
  pm.expect(json).to.have.property("id");
  pm.expect(json).to.have.property("name");
  pm.expect(json).to.have.property("email");
});

pm.test("Email format is valid", () => {
  const json = pm.response.json();
  pm.expect(json.email).to.match(/^[^\s@]+@[^\s@]+\.[^\s@]+$/);
});

// Postbot also suggests edge cases:
// - What if user ID doesn't exist? (404)
// - What if ID is malformed? (400)
// - What if auth token is expired? (401)

Specmatic: Schema-Driven Contracts

Converts OpenAPI/GraphQL specs into executable contracts.

Key Features

• No code required for basic contract tests
• Intelligent service virtualisation
• Backward compatibility checking
• Works with OpenAPI 3.x and GraphQL SDL

Setup

# Install
npm install -g specmatic

# Run contract tests from OpenAPI
specmatic test --contract specs/user-api.yaml --host localhost --port 3000

# Generate stub server
specmatic stub --contract specs/user-api.yaml --port 9000

CI Integration

# .github/workflows/contracts.yml
- name: Contract Test
  run: |
    specmatic test \
      --contract specs/api.yaml \
      --host ${{ env.API_HOST }}

- name: Backward Compatibility
  run: |
    specmatic compatible \
      --older origin/main:specs/api.yaml \
      --newer specs/api.yaml

Decision Matrix: When to Use What

AI vs Manual Contract Tests

Use AI When

• Bootstrapping test coverage
• Generating boilerplate assertions
• Catching obvious missing scenarios
• Maintaining large test suites

Use Manual When

• Complex business logic validation
• Security-critical flows
• Custom matcher requirements
• Semantic contract rules

CI Integration Patterns

Combined AI + Traditional Pipeline

# .github/workflows/api-contracts.yml
jobs:
  ai-generated:
    steps:
      - name: Run Keploy tests
        run: keploy test -c "npm start"

  pact-contracts:
    steps:
      - name: Run consumer contracts
        run: npm run test:pact

      - name: Publish to PactFlow
        run: |
          pact-broker publish ./pacts \
            --consumer-app-version=${{ github.sha }}

  ai-review:
    needs: pact-contracts
    steps:
      - name: PactFlow AI Review
        run: |
          # Triggered automatically on publish
          # Review results in PactFlow dashboard

Metrics to Track

API Security Testing

Systematic API security testing aligned with the OWASP API Security Top 10 (2023), covering authentication, authorization, injection, and automated scanning patterns.

---

Contents

• OWASP API Security Top 10 Overview
• BOLA Testing (Broken Object Level Authorization)
• BFLA Testing (Broken Function Level Authorization)
• Authentication Testing
• Rate Limiting Verification
• Injection Testing
• Mass Assignment Testing
• SSRF Prevention Testing
• Security Headers Validation
• API Key Management Testing
• Automated Security Scanning
• Security Testing Checklist
• Related Resources

---

OWASP API Security Top 10 Overview

---

BOLA Testing

Broken Object Level Authorization: accessing another user's resources by changing the object ID.

Test Patterns

import pytest
import requests

class TestBOLA:
    """Test that users cannot access other users' resources."""

    def setup_method(self):
        self.user_a_token = get_token("user_a@example.com")
        self.user_b_token = get_token("user_b@example.com")

        # Create a resource owned by user_a
        r = requests.post(
            f"{BASE_URL}/orders",
            json={"item": "widget", "quantity": 1},
            headers={"Authorization": f"Bearer {self.user_a_token}"},
        )
        self.user_a_order_id = r.json()["id"]

    def test_cannot_read_other_users_order(self):
        """User B must not access User A's order."""
        r = requests.get(
            f"{BASE_URL}/orders/{self.user_a_order_id}",
            headers={"Authorization": f"Bearer {self.user_b_token}"},
        )
        assert r.status_code == 403 or r.status_code == 404

    def test_cannot_update_other_users_order(self):
        """User B must not modify User A's order."""
        r = requests.patch(
            f"{BASE_URL}/orders/{self.user_a_order_id}",
            json={"quantity": 999},
            headers={"Authorization": f"Bearer {self.user_b_token}"},
        )
        assert r.status_code in (403, 404)

    def test_cannot_delete_other_users_order(self):
        """User B must not delete User A's order."""
        r = requests.delete(
            f"{BASE_URL}/orders/{self.user_a_order_id}",
            headers={"Authorization": f"Bearer {self.user_b_token}"},
        )
        assert r.status_code in (403, 404)

    def test_id_enumeration_does_not_leak_data(self):
        """Sequential ID enumeration must not reveal other users' data."""
        for offset in range(-5, 6):
            test_id = self.user_a_order_id + offset
            r = requests.get(
                f"{BASE_URL}/orders/{test_id}",
                headers={"Authorization": f"Bearer {self.user_b_token}"},
            )
            if r.status_code == 200:
                # If accessible, it must belong to user_b
                assert r.json()["owner_id"] == "user_b"

BOLA Test Matrix

---

BFLA Testing

Broken Function Level Authorization: accessing admin or privileged endpoints as a regular user.

class TestBFLA:
    """Test that role-based access control is enforced."""

    def setup_method(self):
        self.admin_token = get_token("admin@example.com")
        self.user_token = get_token("user@example.com")
        self.readonly_token = get_token("viewer@example.com")

    ADMIN_ENDPOINTS = [
        ("GET", "/admin/users"),
        ("POST", "/admin/users"),
        ("DELETE", "/admin/users/1"),
        ("POST", "/admin/settings"),
        ("GET", "/admin/audit-logs"),
        ("POST", "/admin/export"),
    ]

    @pytest.mark.parametrize("method,path", ADMIN_ENDPOINTS)
    def test_regular_user_cannot_access_admin(self, method, path):
        """Regular users must be denied access to admin endpoints."""
        r = requests.request(
            method, f"{BASE_URL}{path}",
            headers={"Authorization": f"Bearer {self.user_token}"},
            json={} if method in ("POST", "PUT") else None,
        )
        assert r.status_code in (401, 403), (
            f"User accessed admin endpoint: {method} {path} -> {r.status_code}"
        )

    WRITE_ENDPOINTS = [
        ("POST", "/orders"),
        ("PATCH", "/orders/1"),
        ("DELETE", "/orders/1"),
        ("POST", "/documents"),
    ]

    @pytest.mark.parametrize("method,path", WRITE_ENDPOINTS)
    def test_readonly_cannot_write(self, method, path):
        """Read-only users must be denied write operations."""
        r = requests.request(
            method, f"{BASE_URL}{path}",
            headers={"Authorization": f"Bearer {self.readonly_token}"},
            json={} if method in ("POST", "PUT", "PATCH") else None,
        )
        assert r.status_code in (401, 403)

    def test_privilege_escalation_via_body(self):
        """Users cannot escalate their own role via request body."""
        r = requests.patch(
            f"{BASE_URL}/users/me",
            json={"role": "admin"},
            headers={"Authorization": f"Bearer {self.user_token}"},
        )
        # Either rejected or role field ignored
        if r.status_code == 200:
            assert r.json()["role"] != "admin"

---

Authentication Testing

JWT Validation Tests

import jwt
import time

class TestJWTAuthentication:

    def test_expired_token_rejected(self):
        """Expired JWTs must be rejected."""
        expired_token = jwt.encode(
            {"sub": "user_1", "exp": int(time.time()) - 3600},
            SECRET_KEY, algorithm="HS256",
        )
        r = requests.get(
            f"{BASE_URL}/users/me",
            headers={"Authorization": f"Bearer {expired_token}"},
        )
        assert r.status_code == 401

    def test_tampered_token_rejected(self):
        """Tokens signed with wrong key must be rejected."""
        bad_token = jwt.encode(
            {"sub": "user_1", "exp": int(time.time()) + 3600},
            "wrong_secret_key", algorithm="HS256",
        )
        r = requests.get(
            f"{BASE_URL}/users/me",
            headers={"Authorization": f"Bearer {bad_token}"},
        )
        assert r.status_code == 401

    def test_none_algorithm_rejected(self):
        """JWT 'none' algorithm attack must be rejected."""
        header = {"alg": "none", "typ": "JWT"}
        payload = {"sub": "admin", "exp": int(time.time()) + 3600}
        # Manually construct unsigned token
        import base64, json
        h = base64.urlsafe_b64encode(json.dumps(header).encode()).rstrip(b"=")
        p = base64.urlsafe_b64encode(json.dumps(payload).encode()).rstrip(b"=")
        none_token = f"{h.decode()}.{p.decode()}."

        r = requests.get(
            f"{BASE_URL}/users/me",
            headers={"Authorization": f"Bearer {none_token}"},
        )
        assert r.status_code == 401

    def test_missing_token_returns_401(self):
        """Requests without auth token must return 401."""
        r = requests.get(f"{BASE_URL}/users/me")
        assert r.status_code == 401

    def test_refresh_token_rotation(self):
        """Used refresh tokens must be invalidated after rotation."""
        # Get initial token pair
        r = requests.post(f"{BASE_URL}/auth/login", json={
            "email": "user@example.com", "password": "password123"
        })
        refresh_token = r.json()["refresh_token"]

        # Use refresh token
        r = requests.post(f"{BASE_URL}/auth/refresh", json={
            "refresh_token": refresh_token
        })
        assert r.status_code == 200

        # Reuse same refresh token (must fail)
        r = requests.post(f"{BASE_URL}/auth/refresh", json={
            "refresh_token": refresh_token
        })
        assert r.status_code == 401

---

Rate Limiting Verification

import time
from concurrent.futures import ThreadPoolExecutor

class TestRateLimiting:

    def test_rate_limit_enforced(self):
        """Verify rate limit headers and enforcement."""
        responses = []
        for _ in range(110):  # Exceed 100/min limit
            r = requests.get(
                f"{BASE_URL}/users",
                headers={"Authorization": f"Bearer {self.token}"},
            )
            responses.append(r)

        # Verify headers present
        assert "X-RateLimit-Limit" in responses[0].headers
        assert "X-RateLimit-Remaining" in responses[0].headers

        # Some requests should be rate-limited
        status_codes = [r.status_code for r in responses]
        assert 429 in status_codes, "Rate limit was never triggered"

        # 429 response must include Retry-After
        limited = [r for r in responses if r.status_code == 429]
        assert "Retry-After" in limited[0].headers

    def test_rate_limit_per_user_not_global(self):
        """Rate limiting one user must not affect another."""
        # Exhaust user_a's limit
        for _ in range(110):
            requests.get(
                f"{BASE_URL}/users",
                headers={"Authorization": f"Bearer {self.user_a_token}"},
            )

        # User_b should still have quota
        r = requests.get(
            f"{BASE_URL}/users",
            headers={"Authorization": f"Bearer {self.user_b_token}"},
        )
        assert r.status_code == 200

    def test_brute_force_login_protection(self):
        """Login endpoint must have strict rate limiting."""
        for i in range(20):
            r = requests.post(f"{BASE_URL}/auth/login", json={
                "email": "target@example.com",
                "password": f"wrong_password_{i}",
            })

        # Should be rate-limited or locked
        assert r.status_code in (429, 423)

---

Injection Testing

SQL Injection

SQL_INJECTION_PAYLOADS = [
    "' OR '1'='1",
    "'; DROP TABLE users; --",
    "1 UNION SELECT username, password FROM users--",
    "admin'--",
    "1; WAITFOR DELAY '0:0:5'--",
]

@pytest.mark.parametrize("payload", SQL_INJECTION_PAYLOADS)
def test_sql_injection_blocked(api_client, payload):
    """SQL injection payloads must not alter query behavior."""
    r = api_client.get(f"/users?search={payload}")
    assert r.status_code in (200, 400, 422)  # Never 500
    # Verify no unexpected data returned
    if r.status_code == 200:
        data = r.json()
        # Should not return all users (injection success indicator)
        assert len(data.get("data", [])) <= 10

NoSQL Injection

NOSQL_INJECTION_PAYLOADS = [
    {"email": {"$gt": ""}, "password": {"$gt": ""}},
    {"email": {"$ne": "nonexistent"}, "password": {"$ne": "wrong"}},
    {"email": {"$regex": ".*"}, "password": {"$regex": ".*"}},
    {"$where": "this.email == this.email"},
]

@pytest.mark.parametrize("payload", NOSQL_INJECTION_PAYLOADS)
def test_nosql_injection_blocked(api_client, payload):
    """NoSQL injection payloads must be rejected."""
    r = api_client.post("/auth/login", json=payload)
    assert r.status_code != 200, "NoSQL injection may have succeeded"

Command Injection

COMMAND_INJECTION_PAYLOADS = [
    "; ls -la",
    "| cat /etc/passwd",
    "$(whoami)",
    "`id`",
    "& ping -c 10 127.0.0.1",
]

@pytest.mark.parametrize("payload", COMMAND_INJECTION_PAYLOADS)
def test_command_injection_blocked(api_client, payload):
    """Command injection via API parameters must be blocked."""
    r = api_client.post("/tools/convert", json={"filename": payload})
    assert r.status_code in (400, 422)
    # Verify no command execution indicators in response
    assert "/root" not in r.text
    assert "uid=" not in r.text

---

Mass Assignment Testing

class TestMassAssignment:
    """Test that APIs reject unexpected fields that could modify protected attributes."""

    def test_cannot_set_admin_via_registration(self):
        """Registration must ignore role/admin fields."""
        r = requests.post(f"{BASE_URL}/auth/register", json={
            "email": "newuser@example.com",
            "password": "SecureP@ss123",
            "name": "New User",
            "role": "admin",             # Mass assignment attempt
            "is_admin": True,            # Mass assignment attempt
            "subscription": "enterprise", # Mass assignment attempt
        })
        if r.status_code == 201:
            user = r.json()
            assert user.get("role") != "admin"
            assert user.get("is_admin") is not True
            assert user.get("subscription") != "enterprise"

    def test_cannot_modify_protected_fields_via_update(self):
        """Profile update must ignore protected fields."""
        r = requests.patch(
            f"{BASE_URL}/users/me",
            json={
                "name": "Updated Name",
                "id": 1,                    # Cannot change own ID
                "created_at": "2020-01-01", # Cannot backdate
                "email_verified": True,      # Cannot self-verify
            },
            headers={"Authorization": f"Bearer {self.user_token}"},
        )
        if r.status_code == 200:
            user = r.json()
            assert user["id"] != 1 or user["id"] == self.original_user_id
            assert user.get("email_verified") != True  # noqa

---

SSRF Prevention Testing

SSRF_PAYLOADS = [
    "http://127.0.0.1/admin",
    "http://localhost/admin",
    "http://0.0.0.0/",
    "http://169.254.169.254/latest/meta-data/",  # AWS metadata
    "http://[::1]/admin",
    "http://metadata.google.internal/",           # GCP metadata
    "http://100.100.100.200/latest/meta-data/",   # Alibaba metadata
    "file:///etc/passwd",
    "gopher://127.0.0.1:6379/_INFO",
]

@pytest.mark.parametrize("url", SSRF_PAYLOADS)
def test_ssrf_blocked(api_client, url):
    """Server must not fetch internal/metadata URLs."""
    r = api_client.post("/tools/fetch-url", json={"url": url})
    assert r.status_code in (400, 403, 422), (
        f"SSRF may have succeeded for {url}: status={r.status_code}"
    )
    # Verify no internal data leaked
    assert "ami-id" not in r.text        # AWS metadata indicator
    assert "root:" not in r.text         # /etc/passwd indicator
    assert "instance-id" not in r.text   # Cloud metadata indicator

---

Security Headers Validation

def test_security_headers_present(api_client):
    """Verify required security headers on all API responses."""
    r = api_client.get("/users")
    headers = r.headers

    # Required headers
    assert headers.get("X-Content-Type-Options") == "nosniff"
    assert headers.get("X-Frame-Options") in ("DENY", "SAMEORIGIN")
    assert "strict-transport-security" in {k.lower() for k in headers}
    assert headers.get("Cache-Control") in (
        "no-store", "no-cache, no-store, must-revalidate"
    )

    # Must NOT expose
    assert "Server" not in headers or headers["Server"] == ""
    assert "X-Powered-By" not in headers
    assert "X-AspNet-Version" not in headers

    # CORS headers (if applicable)
    assert headers.get("Access-Control-Allow-Origin") != "*" or \
        "authenticated endpoint should not use wildcard CORS"

Security Headers Reference

---

API Key Management Testing

class TestAPIKeyManagement:

    def test_key_in_url_rejected(self):
        """API keys in URL query params should be rejected (log exposure risk)."""
        r = requests.get(f"{BASE_URL}/users?api_key=test_key_123")
        assert r.status_code in (400, 401), \
            "API key in URL should be rejected"

    def test_revoked_key_rejected(self):
        """Revoked API keys must be immediately rejected."""
        # Create and revoke a key
        key = create_api_key("test_user")
        revoke_api_key(key)

        r = requests.get(
            f"{BASE_URL}/users",
            headers={"X-API-Key": key},
        )
        assert r.status_code == 401

    def test_key_scoping_enforced(self):
        """API keys must only access their permitted scopes."""
        readonly_key = create_api_key("test_user", scopes=["read"])

        # Read should work
        r = requests.get(
            f"{BASE_URL}/users",
            headers={"X-API-Key": readonly_key},
        )
        assert r.status_code == 200

        # Write should be rejected
        r = requests.post(
            f"{BASE_URL}/users",
            json={"name": "test"},
            headers={"X-API-Key": readonly_key},
        )
        assert r.status_code == 403

    def test_key_not_leaked_in_error_responses(self):
        """API key must never appear in error response bodies."""
        key = "sk_test_secret_key_12345"
        r = requests.get(
            f"{BASE_URL}/nonexistent",
            headers={"X-API-Key": key},
        )
        assert key not in r.text

---

Automated Security Scanning

OWASP ZAP Integration

# Pull ZAP Docker image
docker pull zaproxy/zap-stable

# API scan using OpenAPI spec
docker run --rm -v $(pwd):/zap/wrk zaproxy/zap-stable \
    zap-api-scan.py \
    -t http://host.docker.internal:8000/openapi.json \
    -f openapi \
    -r /zap/wrk/reports/zap-report.html \
    -J /zap/wrk/reports/zap-report.json \
    -c /zap/wrk/zap-config.conf

# Baseline scan (passive only)
docker run --rm -v $(pwd):/zap/wrk zaproxy/zap-stable \
    zap-baseline.py \
    -t http://host.docker.internal:8000 \
    -r /zap/wrk/reports/baseline.html

CI Pipeline Integration

# .github/workflows/security-scan.yml
name: API Security Scan

on:
  push:
    branches: [main]
  schedule:
    - cron: '0 2 * * 1'  # Weekly Monday 2AM

jobs:
  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Start API
        run: docker compose up -d api
      - name: Wait for API
        run: |
          for i in {1..30}; do
            curl -s http://localhost:8000/health && break
            sleep 2
          done
      - name: Run OWASP ZAP
        run: |
          docker run --rm --network host \
            -v ${{ github.workspace }}:/zap/wrk \
            zaproxy/zap-stable \
            zap-api-scan.py \
            -t http://localhost:8000/openapi.json \
            -f openapi \
            -J /zap/wrk/zap-results.json
      - name: Check for high severity findings
        run: |
          HIGH=$(jq '[.site[].alerts[] | select(.riskcode >= 3)] | length' zap-results.json)
          if [ "$HIGH" -gt 0 ]; then
            echo "HIGH/CRITICAL findings detected: $HIGH"
            jq '.site[].alerts[] | select(.riskcode >= 3) | .name' zap-results.json
            exit 1
          fi

---

Security Testing Checklist

• [ ] BOLA: Cross-user access tested for all resource endpoints
• [ ] BFLA: Admin endpoints blocked for regular users
• [ ] BFLA: Write endpoints blocked for read-only users
• [ ] JWT: Expired, tampered, and none-algorithm tokens rejected
• [ ] JWT: Refresh token rotation and reuse detection verified
• [ ] Rate limiting: Enforced per-user on all endpoints
• [ ] Rate limiting: Login endpoint has strict limits
• [ ] SQL injection: Parameterized queries verified, payloads tested
• [ ] NoSQL injection: Operator injection payloads rejected
• [ ] Command injection: Payloads in all user-controlled parameters tested
• [ ] Mass assignment: Protected fields not modifiable via API
• [ ] SSRF: Internal and cloud metadata URLs blocked
• [ ] Security headers: All required headers present
• [ ] API keys: Not accepted in URL params, revocation works
• [ ] Automated scan: OWASP ZAP or Burp Suite run in CI
• [ ] No sensitive data in error responses
• [ ] HTTPS enforced, no HTTP fallback

---

Related Resources

• [contract-testing-patterns.md](contract-testing-patterns.md) - Contract testing fundamentals
• [schema-driven-testing.md](schema-driven-testing.md) - Schema-based fuzzing and validation
• [api-versioning-strategies.md](api-versioning-strategies.md) - Versioning and deprecation
• [SKILL.md](../SKILL.md) - QA API Testing & Contracts skill overview

API Versioning Strategies

Patterns for versioning APIs, detecting breaking changes, and automating backward compatibility verification across release cycles.

---

Contents

• Versioning Schemes
• Semantic Versioning for APIs
• Breaking vs Non-Breaking Changes
• Backward Compatibility Verification
• Deprecation Policy Design
• Sunset Header Implementation
• Consumer Notification Workflows
• Migration Testing Patterns
• Version Matrix Testing
• OpenAPI Diff Tooling
• Versioning Checklist
• Related Resources

---

Versioning Schemes

Scheme Comparison

URL Path Versioning (Recommended for Public APIs)

# FastAPI example
from fastapi import FastAPI, APIRouter

app = FastAPI()

# Version 1
v1_router = APIRouter(prefix="/v1")

@v1_router.get("/users/{user_id}")
async def get_user_v1(user_id: int):
    return {"id": user_id, "name": "Alice", "email": "alice@example.com"}

# Version 2 - added 'role' field, restructured 'name'
v2_router = APIRouter(prefix="/v2")

@v2_router.get("/users/{user_id}")
async def get_user_v2(user_id: int):
    return {
        "id": user_id,
        "name": {"first": "Alice", "last": "Smith"},
        "email": "alice@example.com",
        "role": "admin",
    }

app.include_router(v1_router)
app.include_router(v2_router)

Header-Based Versioning

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    accept: str = Header(default="application/vnd.myapi.v2+json"),
):
    if "v1" in accept:
        return {"id": user_id, "name": "Alice"}
    elif "v2" in accept:
        return {"id": user_id, "name": {"first": "Alice", "last": "Smith"}}
    else:
        raise HTTPException(406, "Unsupported API version")

# Client usage
curl -H "Accept: application/vnd.myapi.v1+json" https://api.example.com/users/42
curl -H "Accept: application/vnd.myapi.v2+json" https://api.example.com/users/42

---

Semantic Versioning for APIs

SemVer Applied to APIs

MAJOR.MINOR.PATCH

MAJOR (v1 → v2): Breaking changes
  - Removed endpoints or fields
  - Changed field types
  - Changed authentication scheme
  - Restructured response shape

MINOR (v1.1 → v1.2): New features, backward compatible
  - Added new endpoints
  - Added optional fields to responses
  - Added optional query parameters
  - New enum values in responses

PATCH (v1.1.0 → v1.1.1): Bug fixes, no API changes
  - Fixed incorrect status codes
  - Fixed validation logic
  - Performance improvements
  - Documentation corrections

Version Lifecycle

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│  Alpha   │───▶│   Beta   │───▶│  Stable  │───▶│  Sunset  │───▶ Removed
│ (v3-alpha)│   │(v3-beta) │    │   (v3)   │    │  (v3)    │
└──────────┘    └──────────┘    └──────────┘    └──────────┘
                                     │
                                     │  v4 released
                                     ▼
                                ┌──────────┐
                                │Deprecated│
                                │   (v3)   │
                                └──────────┘

---

Breaking vs Non-Breaking Changes

Breaking Changes Catalog

Automated Breaking Change Detection

def detect_breaking_changes(old_spec: dict, new_spec: dict) -> list[dict]:
    """Detect breaking changes between two OpenAPI specs."""
    breaking = []

    # Check removed endpoints
    old_paths = set(old_spec.get("paths", {}).keys())
    new_paths = set(new_spec.get("paths", {}).keys())
    for removed in old_paths - new_paths:
        breaking.append({
            "type": "endpoint_removed",
            "path": removed,
            "severity": "critical",
        })

    # Check each shared endpoint
    for path in old_paths & new_paths:
        old_methods = set(old_spec["paths"][path].keys())
        new_methods = set(new_spec["paths"][path].keys())

        for removed_method in old_methods - new_methods:
            breaking.append({
                "type": "method_removed",
                "path": path,
                "method": removed_method,
                "severity": "critical",
            })

        for method in old_methods & new_methods:
            # Check response schema changes
            changes = compare_response_schemas(
                old_spec["paths"][path][method],
                new_spec["paths"][path][method],
            )
            breaking.extend(changes)

    return breaking

---

Backward Compatibility Verification

Contract Test Suite

import pytest
import requests

BASE_V1 = "https://api.example.com/v1"
BASE_V2 = "https://api.example.com/v2"

class TestBackwardCompatibility:
    """Verify v2 does not break v1 consumers."""

    def test_v1_response_fields_still_present(self):
        """All v1 response fields must still exist in v1 endpoint."""
        response = requests.get(f"{BASE_V1}/users/1")
        data = response.json()
        # These fields must always be present in v1
        assert "id" in data
        assert "name" in data
        assert "email" in data
        assert isinstance(data["name"], str)  # v1 uses flat string

    def test_v1_status_codes_unchanged(self):
        """v1 must return same status codes for same scenarios."""
        # Existing resource
        r = requests.get(f"{BASE_V1}/users/1")
        assert r.status_code == 200

        # Missing resource
        r = requests.get(f"{BASE_V1}/users/999999")
        assert r.status_code == 404

        # Invalid input
        r = requests.get(f"{BASE_V1}/users/abc")
        assert r.status_code in (400, 422)

    def test_v1_pagination_contract_preserved(self):
        """Pagination structure must not change in v1."""
        r = requests.get(f"{BASE_V1}/users?page=1&per_page=10")
        data = r.json()
        assert "data" in data
        assert "meta" in data
        assert "total" in data["meta"]
        assert "page" in data["meta"]

Compatibility Test Matrix

#!/bin/bash
# run_compat_matrix.sh - Test all supported version combinations

VERSIONS=("v1" "v2" "v3")
RESULTS=()

for version in "${VERSIONS[@]}"; do
    echo "Testing $version compatibility..."
    pytest tests/compatibility/ \
        --api-version="$version" \
        --junitxml="reports/compat_${version}.xml" \
        2>&1

    if [ $? -eq 0 ]; then
        RESULTS+=("$version: PASS")
    else
        RESULTS+=("$version: FAIL")
    fi
done

echo ""
echo "=== Compatibility Matrix Results ==="
for result in "${RESULTS[@]}"; do
    echo "  $result"
done

---

Deprecation Policy Design

Deprecation Timeline Template

Phase 1: Announce (Month 0)
  - Add Deprecation header to responses
  - Update API documentation
  - Notify consumers via email/changelog
  - Provide migration guide

Phase 2: Warning Period (Months 1-3)
  - Log usage of deprecated endpoints
  - Send targeted notifications to active consumers
  - Add Sunset header with target date

Phase 3: Soft Sunset (Month 4-5)
  - Return warning in response body
  - Throttle deprecated endpoint (optional)
  - Final migration reminders

Phase 4: Hard Sunset (Month 6)
  - Return 410 Gone status
  - Include migration URL in response body
  - Remove from documentation

Deprecation Headers

HTTP/1.1 200 OK
Deprecation: true
Sunset: Sat, 01 Mar 2026 00:00:00 GMT
Link: <https://api.example.com/v2/users>; rel="successor-version"

{
  "data": { ... },
  "_deprecation": {
    "message": "This endpoint is deprecated. Use /v2/users instead.",
    "sunset_date": "2026-03-01",
    "migration_guide": "https://docs.example.com/migration/v1-to-v2"
  }
}

---

Sunset Header Implementation

from fastapi import FastAPI, Response
from datetime import datetime

app = FastAPI()

SUNSET_DATES = {
    "v1": datetime(2026, 3, 1),
    "v2": None,  # Current version, no sunset
}

def add_deprecation_headers(response: Response, version: str):
    """Add standard deprecation and sunset headers."""
    sunset = SUNSET_DATES.get(version)
    if sunset:
        response.headers["Deprecation"] = "true"
        response.headers["Sunset"] = sunset.strftime("%a, %d %b %Y %H:%M:%S GMT")
        response.headers["Link"] = (
            f'<https://api.example.com/v{int(version[1:])+1}/users>; '
            f'rel="successor-version"'
        )

@app.get("/v1/users/{user_id}")
async def get_user_v1(user_id: int, response: Response):
    add_deprecation_headers(response, "v1")
    return {"id": user_id, "name": "Alice"}

---

Consumer Notification Workflows

Consumer Usage Tracking

def track_version_usage(version: str, consumer_id: str):
    """Track which consumers are using which versions."""
    redis_client.hincrby(f"api_usage:{version}", consumer_id, 1)
    redis_client.sadd(f"active_consumers:{version}", consumer_id)
    redis_client.expire(f"active_consumers:{version}", 86400 * 30)

def get_consumers_needing_migration(deprecated_version: str) -> set:
    """Find consumers still using a deprecated version."""
    return redis_client.smembers(f"active_consumers:{deprecated_version}")

---

Migration Testing Patterns

Response Shape Migration Test

def test_v1_to_v2_migration_path(api_client):
    """Verify consumers can migrate from v1 to v2."""
    # Get v1 response
    v1_response = api_client.get("/v1/users/1").json()

    # Get v2 response
    v2_response = api_client.get("/v2/users/1").json()

    # All v1 data must be derivable from v2
    assert v2_response["id"] == v1_response["id"]
    assert v2_response["email"] == v1_response["email"]

    # v1 'name' (string) maps to v2 'name.first' + 'name.last'
    full_name = f"{v2_response['name']['first']} {v2_response['name']['last']}"
    assert full_name == v1_response["name"]

def test_v1_to_v2_request_compatibility(api_client):
    """Verify v1 request format works or returns helpful error in v2."""
    v1_payload = {"name": "Alice Smith", "email": "alice@example.com"}

    response = api_client.post("/v2/users", json=v1_payload)

    # v2 should either accept the old format or return actionable error
    if response.status_code == 422:
        error = response.json()
        assert "migration" in str(error).lower() or "name.first" in str(error)

---

Version Matrix Testing

Multi-Version CI Pipeline

# .github/workflows/version-compat.yml
name: API Version Compatibility

on:
  push:
    paths: ['src/api/**']

jobs:
  compatibility:
    strategy:
      matrix:
        api_version: [v1, v2, v3]
        client_version: [v1, v2, v3]
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Start API server
        run: docker compose up -d api
      - name: Run compatibility tests
        run: |
          pytest tests/compatibility/ \
            --server-version=${{ matrix.api_version }} \
            --client-version=${{ matrix.client_version }} \
            --junitxml=reports/${{ matrix.api_version }}_${{ matrix.client_version }}.xml
      - name: Upload results
        uses: actions/upload-artifact@v4
        with:
          name: compat-${{ matrix.api_version }}-${{ matrix.client_version }}
          path: reports/

Compatibility Matrix Visualization

           Server Version
           v1      v2      v3
Client v1  PASS    PASS    FAIL*
Client v2  N/A     PASS    PASS
Client v3  N/A     N/A     PASS

* v1 client + v3 server: Expected failure (v1 sunset)

---

OpenAPI Diff Tooling

oasdiff (Recommended)

# Install
go install github.com/tufin/oasdiff@latest

# Detect breaking changes
oasdiff breaking old-api.yaml new-api.yaml

# Full changelog diff
oasdiff changelog old-api.yaml new-api.yaml

# Output as JSON for CI
oasdiff breaking old-api.yaml new-api.yaml --format json

# Fail CI on breaking changes
oasdiff breaking old-api.yaml new-api.yaml --fail-on ERR

Optic

# Install
npm install -g @useoptic/optic

# Compare specs
optic diff old-api.yaml new-api.yaml

# CI integration - check for breaking changes
optic diff old-api.yaml new-api.yaml --check

CI Integration Example

#!/bin/bash
# check_api_compat.sh - Run in CI before merge

set -euo pipefail

OLD_SPEC="main:openapi/spec.yaml"
NEW_SPEC="openapi/spec.yaml"

echo "Checking for breaking changes..."
BREAKING=$(oasdiff breaking "$OLD_SPEC" "$NEW_SPEC" --format json 2>&1)

if [ "$(echo "$BREAKING" | jq length)" -gt 0 ]; then
    echo "BREAKING CHANGES DETECTED:"
    echo "$BREAKING" | jq -r '.[].message'
    echo ""
    echo "If intentional, bump the major version and update migration guide."
    exit 1
fi

echo "No breaking changes detected. Safe to merge."

---

Versioning Checklist

• [ ] Versioning scheme selected and documented
• [ ] Breaking change detection automated in CI (oasdiff or optic)
• [ ] Backward compatibility test suite covers all supported versions
• [ ] Deprecation policy written with clear timelines
• [ ] Sunset headers implemented for deprecated versions
• [ ] Consumer usage tracking in place
• [ ] Migration guide published for each major version bump
• [ ] Version matrix testing runs in CI
• [ ] Notification workflow configured for deprecation events
• [ ] 410 Gone responses configured for sunset versions
• [ ] OpenAPI spec versioned alongside code

---

Related Resources

• [contract-testing-patterns.md](contract-testing-patterns.md) - Consumer-driven contract testing
• [schema-driven-testing.md](schema-driven-testing.md) - Schema-based test generation
• [ai-contract-testing.md](ai-contract-testing.md) - AI-specific contract testing
• [SKILL.md](../SKILL.md) - QA API Testing & Contracts skill overview

Contract Testing Patterns

Breaking vs Non-Breaking Changes

Breaking Changes (Require Version Bump)

Non-Breaking Changes (Safe to Release)

GraphQL Change Safety (SDL)

Breaking Changes

• Remove a field/type/enum value used by consumers.
• Change a field return type or input type incompatibly.
• Tighten nullability (String → String!) or list nullability in a way that can change runtime results.
• Add a required argument to an existing field.
• Change a directive or federation composition in a way that breaks query planning.

Non-Breaking Changes

• Add a field (clients ignore what they do not request).
• Add an optional argument.
• Add an enum value (as long as consumers handle unknowns defensively).
• Deprecate fields with a published sunset policy.

gRPC / Protobuf Change Safety (Proto3)

Breaking Changes

• Reuse or renumber field numbers (wire incompatibility).
• Change a field type incompatibly (for example, int32 → string).
• Change request/response shapes in a way that breaks existing clients (for example, moving fields between messages without compatibility rules).
• Remove an RPC or change streaming semantics.
• Rename packages/services without aliases (client generation and routing break).

Non-Breaking Changes

• Add a new field with a new field number.
• Add a new RPC.
• Add an enum value (clients should handle unknown values).
• Remove a field only if you reserve the field number and name and keep behavior compatible.

Rule of thumb: never reuse field numbers; use reserved for removed numbers/names; run buf breaking in CI.

Consumer-Driven Contract Testing

Traditional CDC Workflow

Consumer                    Broker                    Provider
   │                          │                          │
   ├── Generate contract ────►│                          │
   │   (Pact file)            │                          │
   │                          │◄── Fetch contracts ──────┤
   │                          │                          │
   │                          │    Verify against        │
   │                          │    provider ────────────►│
   │                          │                          │
   │◄── Results ──────────────┤◄── Publish results ──────┤

Bi-Directional Contract Testing (2026)

New paradigm that combines consumer contracts with provider specifications.

Consumer ─── Pact Contract ───┐
                              ├──► Broker ◄──► Comparison Engine
Provider ─── OpenAPI Spec ────┘

When to Use Bi-Directional:

Setup Example:

# pactflow.yml
provider:
  name: UserService
  specification:
    type: openapi
    path: ./specs/user-api.yaml

consumer:
  name: WebApp
  contracts:
    - path: ./pacts/webapp-userservice.json

Benefits:

• Provider doesn't need to run Pact verification
• Works with existing OpenAPI specs
• Faster adoption for teams new to contract testing
• Supports provider-driven development

Consumer Test Example (Pact)

// consumer.pact.spec.js
const { Pact } = require('@pact-foundation/pact');

describe('User API Consumer', () => {
  const provider = new Pact({
    consumer: 'WebApp',
    provider: 'UserService',
  });

  it('fetches a user by ID', async () => {
    await provider.addInteraction({
      state: 'user 123 exists',
      uponReceiving: 'a request for user 123',
      withRequest: {
        method: 'GET',
        path: '/users/123',
      },
      willRespondWith: {
        status: 200,
        body: {
          id: '123',
          name: Matchers.string('John'),
          email: Matchers.email(),
        },
      },
    });

    const user = await userClient.getUser('123');
    expect(user.id).toBe('123');
  });
});

Provider Verification Example

// provider.pact.spec.js
const { Verifier } = require('@pact-foundation/pact');

describe('User API Provider', () => {
  it('validates consumer contracts', async () => {
    await new Verifier({
      providerBaseUrl: 'http://localhost:3000',
      pactBrokerUrl: process.env.PACT_BROKER_URL,
      provider: 'UserService',
      publishVerificationResult: true,
      stateHandlers: {
        'user 123 exists': async () => {
          await seedUser({ id: '123', name: 'John' });
        },
      },
    }).verifyProvider();
  });
});

Schema Validation Patterns

Four Levels of Validation

Property-Based Testing (Schemathesis)

# Run property-based tests against OpenAPI spec
schemathesis run https://api.example.com/openapi.yaml \
  --checks all \
  --hypothesis-max-examples=100 \
  --base-url https://staging.api.example.com

Schemathesis automatically generates test cases to find:

• 500 errors from edge case inputs
• Schema violations in responses
• Security issues (auth bypass, injection)

CI Integration Patterns

Pre-merge Gates

# .github/workflows/api-contracts.yml
name: API Contract Validation

on: [pull_request]

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: Lint OpenAPI
        run: npx @stoplight/spectral-cli lint specs/*.yaml

      - name: Check breaking changes
        run: |
          git show "origin/${{ github.base_ref }}:specs/api.yaml" > /tmp/api.base.yaml
          oasdiff breaking /tmp/api.base.yaml specs/api.yaml

  contract-test:
    runs-on: ubuntu-latest
    steps:
      - name: Run consumer contract tests
        run: npm run test:pact

      - name: Publish contracts to broker
        run: npx pact-broker publish ./pacts \
          --consumer-app-version=${{ github.sha }} \
          --broker-base-url=${{ secrets.PACT_BROKER_URL }}

Post-merge Provider Verification

# Run after provider changes merge
verify-contracts:
  runs-on: ubuntu-latest
  steps:
    - name: Start provider
      run: docker compose up -d api

    - name: Verify all consumer contracts
      run: |
        npx pact-verifier \
          --provider-base-url=http://localhost:3000 \
          --pact-broker-url=${{ secrets.PACT_BROKER_URL }} \
          --provider=UserService \
          --publish-verification-results

Deployment Gate (Pact Broker)

Run a deploy gate (separate from tests) so you only deploy versions proven compatible:

# In the deploy pipeline (example: staging)
npx pact-broker can-i-deploy \
  --pacticipant "UserService" \
  --version "$GITHUB_SHA" \
  --to-environment "staging" \
  --broker-base-url "$PACT_BROKER_URL"

# After a successful deploy
npx pact-broker record-deployment \
  --pacticipant "UserService" \
  --version "$GITHUB_SHA" \
  --environment "staging" \
  --broker-base-url "$PACT_BROKER_URL"

Mock Strategies

Schema-Based Mocks (Prism)

# Generate mock server from OpenAPI spec
npx @stoplight/prism-cli mock specs/api.yaml --port 4010

# Responses match schema with realistic fake data
curl http://localhost:4010/users/123
# {"id": "abc123", "name": "John Doe", "email": "john@example.com"}

Contract-Based Mocks (Pact Stub)

# Start stub server from Pact contracts
npx pact-stub-server --pact-dir ./pacts --port 4011

# Only returns interactions defined in contracts
curl http://localhost:4011/users/123

When to Use Each

Versioning Strategies

URL Versioning

GET /v1/users/123
GET /v2/users/123

• Clear version visibility
• Easy routing
• URL pollution over time

Header Versioning

GET /users/123
Accept: application/vnd.api+json; version=2

• Clean URLs
• Harder to test/debug
• Proxy complexity

Contract Testing with Versions

// Test multiple versions simultaneously
const versions = ['v1', 'v2'];

versions.forEach(version => {
  describe(`User API ${version}`, () => {
    it('maintains backward compatibility', async () => {
      const response = await fetch(`/${version}/users/123`);
      expect(response.status).toBe(200);
      // v1 and v2 should both return user data
    });
  });
});

Contract Versioning Best Practices (2026)

Semantic Versioning for Contracts

Apply SemVer principles to API contracts:

Contract Version in Pact

// Include version in consumer contract
const provider = new Pact({
  consumer: 'WebApp',
  provider: 'UserService',
  pactfileWriteMode: 'update',
});

// Publish with version
await pactBroker.publishPacts({
  pactFilesOrDirs: ['./pacts'],
  consumerVersion: process.env.GIT_COMMIT,
  tags: ['main', 'v2'],
});

Deprecation Strategy

# OpenAPI deprecation example
paths:
  /v1/users/{id}:
    get:
      deprecated: true
      x-deprecation-date: "2026-06-01"
      x-sunset-date: "2026-12-01"
      description: |
        DEPRECATED: Use /v2/users/{id} instead.
        Will be removed on 2026-12-01.

Multi-Version Contract Matrix

Consumer v1.0 ──► Provider v1.x ✓
Consumer v1.0 ──► Provider v2.x ✓ (backward compatible)
Consumer v2.0 ──► Provider v1.x ✗ (requires v2 features)
Consumer v2.0 ──► Provider v2.x ✓

Related Resources

• See ai-contract-testing.md for AI-powered test generation
• See ../assets/contract-change-checklist.md for release validation

Schema-Driven Testing

Property-based and schema-driven API testing using OpenAPI specifications as the source of truth for automated test generation, fuzzing, and validation.

---

Contents

• Schema-Driven Testing Principles
• Schemathesis for OpenAPI Fuzzing
• Property-Based Testing with Hypothesis
• Property-Based Testing with fast-check
• Negative Test Generation from Schemas
• Boundary Value Analysis
• Stateful Testing
• Response Schema Validation
• Schema Evolution Testing
• Custom Strategies for Domain Types
• Schema Testing Checklist
• Related Resources

---

Schema-Driven Testing Principles

Schema ──▶ Test Generator ──▶ Requests ──▶ API ──▶ Response Validator
  │                                                        │
  └─── Constraints (types, ranges, patterns) ──────────────┘
                    Validates against schema

---

Schemathesis for OpenAPI Fuzzing

Schemathesis generates test cases automatically from OpenAPI/GraphQL specifications.

Installation and Basic Usage

# Install
pip install schemathesis

# Run against a live API
schemathesis run https://api.example.com/openapi.json

# Run against a local spec file
schemathesis run ./openapi.yaml --base-url http://localhost:8000

# Target specific endpoints
schemathesis run ./openapi.yaml \
    --base-url http://localhost:8000 \
    --endpoint "/users" \
    --method POST

# Set authentication
schemathesis run ./openapi.yaml \
    --base-url http://localhost:8000 \
    --header "Authorization: Bearer $TOKEN"

Advanced Schemathesis Configuration

# Increase test cases per endpoint
schemathesis run ./openapi.yaml \
    --base-url http://localhost:8000 \
    --hypothesis-max-examples=500

# Enable stateful testing (link-based)
schemathesis run ./openapi.yaml \
    --base-url http://localhost:8000 \
    --stateful=links

# Output as JUnit XML for CI
schemathesis run ./openapi.yaml \
    --base-url http://localhost:8000 \
    --junit-xml=reports/schemathesis.xml

# Reproduce a specific failure
schemathesis replay ./cassette.yaml

Schemathesis in Python Tests

import schemathesis

schema = schemathesis.from_uri("http://localhost:8000/openapi.json")

@schema.parametrize()
def test_api(case):
    """Schemathesis generates and runs test cases from schema."""
    response = case.call()
    case.validate_response(response)

@schema.parametrize(endpoint="/users", method="POST")
def test_create_user(case):
    """Test user creation with generated payloads."""
    response = case.call()
    case.validate_response(response)
    # Custom assertions
    if response.status_code == 201:
        data = response.json()
        assert "id" in data
        assert isinstance(data["id"], int)

# Add custom checks
@schema.parametrize()
def test_no_server_errors(case):
    """API must never return 500."""
    response = case.call()
    assert response.status_code < 500, (
        f"Server error on {case.method} {case.path}: {response.text}"
    )

---

Property-Based Testing with Hypothesis

Hypothesis generates random inputs satisfying constraints and finds minimal failing examples.

Basic API Property Tests

from hypothesis import given, settings, strategies as st
import requests

@given(
    name=st.text(min_size=1, max_size=100, alphabet=st.characters(
        whitelist_categories=("L", "N", "Zs")
    )),
    email=st.emails(),
    age=st.integers(min_value=0, max_value=150),
)
@settings(max_examples=200)
def test_create_user_properties(name, email, age):
    """
    Property: Creating a user with valid data always succeeds
    and returns the same data back.
    """
    payload = {"name": name, "email": email, "age": age}
    response = requests.post("http://localhost:8000/v1/users", json=payload)

    assert response.status_code == 201
    data = response.json()
    assert data["name"] == name
    assert data["email"] == email
    assert data["age"] == age
    assert "id" in data

@given(
    page=st.integers(min_value=1, max_value=1000),
    per_page=st.integers(min_value=1, max_value=100),
)
def test_pagination_properties(page, per_page):
    """
    Properties:
    - Response always contains 'data' array
    - Array length <= per_page
    - Meta includes correct page number
    """
    r = requests.get(
        f"http://localhost:8000/v1/users?page={page}&per_page={per_page}"
    )
    assert r.status_code == 200
    body = r.json()
    assert isinstance(body["data"], list)
    assert len(body["data"]) <= per_page
    assert body["meta"]["page"] == page

Strategies from OpenAPI Schema

from hypothesis import strategies as st
import json

def strategy_from_json_schema(schema: dict) -> st.SearchStrategy:
    """Generate Hypothesis strategy from JSON Schema definition."""
    schema_type = schema.get("type")

    if schema_type == "string":
        min_len = schema.get("minLength", 0)
        max_len = schema.get("maxLength", 256)
        pattern = schema.get("pattern")
        enum = schema.get("enum")

        if enum:
            return st.sampled_from(enum)
        if schema.get("format") == "email":
            return st.emails()
        if schema.get("format") == "date":
            return st.dates().map(str)
        if schema.get("format") == "uuid":
            return st.uuids().map(str)
        if pattern:
            return st.from_regex(pattern, fullmatch=True)
        return st.text(min_size=min_len, max_size=max_len)

    elif schema_type == "integer":
        minimum = schema.get("minimum", -2**31)
        maximum = schema.get("maximum", 2**31)
        return st.integers(min_value=minimum, max_value=maximum)

    elif schema_type == "number":
        minimum = schema.get("minimum", -1e10)
        maximum = schema.get("maximum", 1e10)
        return st.floats(
            min_value=minimum, max_value=maximum,
            allow_nan=False, allow_infinity=False,
        )

    elif schema_type == "boolean":
        return st.booleans()

    elif schema_type == "array":
        items_strategy = strategy_from_json_schema(schema.get("items", {}))
        min_items = schema.get("minItems", 0)
        max_items = schema.get("maxItems", 10)
        return st.lists(items_strategy, min_size=min_items, max_size=max_items)

    elif schema_type == "object":
        properties = schema.get("properties", {})
        required = set(schema.get("required", []))
        fixed = {
            k: strategy_from_json_schema(v)
            for k, v in properties.items()
            if k in required
        }
        optional = {
            k: strategy_from_json_schema(v)
            for k, v in properties.items()
            if k not in required
        }
        return st.fixed_dictionaries(fixed, optional=optional)

    return st.none()

---

Property-Based Testing with fast-check

JavaScript/TypeScript API Testing

import fc from 'fast-check';
import axios from 'axios';

const BASE_URL = 'http://localhost:8000/v1';

describe('User API Properties', () => {
  it('creating a user with valid data returns 201', async () => {
    await fc.assert(
      fc.asyncProperty(
        fc.record({
          name: fc.string({ minLength: 1, maxLength: 100 }),
          email: fc.emailAddress(),
          age: fc.integer({ min: 0, max: 150 }),
        }),
        async (user) => {
          const res = await axios.post(`${BASE_URL}/users`, user, {
            validateStatus: () => true,
          });
          expect(res.status).toBe(201);
          expect(res.data.name).toBe(user.name);
          expect(res.data.email).toBe(user.email);
        },
      ),
      { numRuns: 100 },
    );
  });

  it('GET /users always returns valid pagination', async () => {
    await fc.assert(
      fc.asyncProperty(
        fc.integer({ min: 1, max: 100 }),
        fc.integer({ min: 1, max: 50 }),
        async (page, perPage) => {
          const res = await axios.get(
            `${BASE_URL}/users?page=${page}&per_page=${perPage}`,
          );
          expect(res.status).toBe(200);
          expect(Array.isArray(res.data.data)).toBe(true);
          expect(res.data.data.length).toBeLessThanOrEqual(perPage);
        },
      ),
    );
  });

  it('invalid input always returns 4xx, never 5xx', async () => {
    await fc.assert(
      fc.asyncProperty(
        fc.anything(),
        async (randomPayload) => {
          const res = await axios.post(`${BASE_URL}/users`, randomPayload, {
            validateStatus: () => true,
          });
          // Server must never crash on random input
          expect(res.status).toBeLessThan(500);
        },
      ),
      { numRuns: 200 },
    );
  });
});

---

Negative Test Generation from Schemas

Systematic Invalid Input Generation

def generate_negative_cases(schema: dict) -> list[dict]:
    """Generate invalid inputs that should trigger validation errors."""
    negative = []
    properties = schema.get("properties", {})
    required = schema.get("required", [])

    # Missing required fields
    for field in required:
        case = {k: generate_valid(v) for k, v in properties.items() if k != field}
        negative.append({
            "payload": case,
            "description": f"Missing required field: {field}",
            "expected_status": 422,
        })

    # Wrong types for each field
    for field, field_schema in properties.items():
        wrong_type_values = get_wrong_type_values(field_schema["type"])
        for wrong_val in wrong_type_values:
            case = {k: generate_valid(v) for k, v in properties.items()}
            case[field] = wrong_val
            negative.append({
                "payload": case,
                "description": f"Wrong type for {field}: {type(wrong_val).__name__}",
                "expected_status": 422,
            })

    # Boundary violations
    for field, field_schema in properties.items():
        if "maxLength" in field_schema:
            case = {k: generate_valid(v) for k, v in properties.items()}
            case[field] = "x" * (field_schema["maxLength"] + 1)
            negative.append({
                "payload": case,
                "description": f"Exceeds maxLength for {field}",
                "expected_status": 422,
            })
        if "minimum" in field_schema:
            case = {k: generate_valid(v) for k, v in properties.items()}
            case[field] = field_schema["minimum"] - 1
            negative.append({
                "payload": case,
                "description": f"Below minimum for {field}",
                "expected_status": 422,
            })

    return negative

def get_wrong_type_values(expected_type: str) -> list:
    """Return values of incorrect types."""
    all_types = {
        "string": [123, True, None, [], {}],
        "integer": ["abc", True, None, [], {}, 3.14],
        "number": ["abc", True, None, [], {}],
        "boolean": ["abc", 0, None, [], {}],
        "array": ["abc", 123, None, {}],
        "object": ["abc", 123, None, []],
    }
    return all_types.get(expected_type, [None])

Running Negative Tests

@pytest.mark.parametrize(
    "case",
    generate_negative_cases(USER_SCHEMA),
    ids=lambda c: c["description"],
)
def test_validation_rejects_invalid_input(api_client, case):
    """API must reject all invalid inputs with proper error response."""
    response = api_client.post("/v1/users", json=case["payload"])
    assert response.status_code == case["expected_status"], (
        f"Expected {case['expected_status']} for: {case['description']}, "
        f"got {response.status_code}"
    )
    # Error response should include field-level detail
    error = response.json()
    assert "detail" in error or "errors" in error

---

Boundary Value Analysis

Boundary Matrix from Schema

Automated Boundary Testing

def generate_boundary_values(schema: dict) -> list[tuple]:
    """Generate boundary test values from schema constraints."""
    boundaries = []

    for field, props in schema.get("properties", {}).items():
        if props["type"] == "integer":
            lo = props.get("minimum")
            hi = props.get("maximum")
            if lo is not None:
                boundaries.extend([
                    (field, lo - 1, "below_minimum", 422),
                    (field, lo, "at_minimum", 200),
                    (field, lo + 1, "above_minimum", 200),
                ])
            if hi is not None:
                boundaries.extend([
                    (field, hi - 1, "below_maximum", 200),
                    (field, hi, "at_maximum", 200),
                    (field, hi + 1, "above_maximum", 422),
                ])

        elif props["type"] == "string":
            min_len = props.get("minLength", 0)
            max_len = props.get("maxLength")
            if min_len > 0:
                boundaries.extend([
                    (field, "x" * (min_len - 1), "below_minLength", 422),
                    (field, "x" * min_len, "at_minLength", 200),
                ])
            if max_len:
                boundaries.extend([
                    (field, "x" * max_len, "at_maxLength", 200),
                    (field, "x" * (max_len + 1), "above_maxLength", 422),
                ])

    return boundaries

---

Stateful Testing

Link-Based Stateful Testing (Schemathesis)

# Schemathesis follows OpenAPI links to test multi-step workflows
# e.g., POST /users → GET /users/{id} → PATCH /users/{id} → DELETE /users/{id}
schemathesis run ./openapi.yaml \
    --base-url http://localhost:8000 \
    --stateful=links \
    --hypothesis-max-examples=100

Custom State Machine Testing

import hypothesis.stateful as stateful
from hypothesis import strategies as st

class APIStateMachine(stateful.RuleBasedStateMachine):
    """Test API through stateful sequences of operations."""

    created_user_ids = stateful.Bundle("user_ids")

    @stateful.rule(
        target=created_user_ids,
        name=st.text(min_size=1, max_size=50),
        email=st.emails(),
    )
    def create_user(self, name, email):
        r = requests.post(BASE_URL + "/users", json={
            "name": name, "email": email
        })
        assert r.status_code == 201
        user_id = r.json()["id"]
        return user_id

    @stateful.rule(user_id=created_user_ids)
    def get_user(self, user_id):
        r = requests.get(f"{BASE_URL}/users/{user_id}")
        assert r.status_code == 200
        assert r.json()["id"] == user_id

    @stateful.rule(
        user_id=created_user_ids,
        new_name=st.text(min_size=1, max_size=50),
    )
    def update_user(self, user_id, new_name):
        r = requests.patch(
            f"{BASE_URL}/users/{user_id}",
            json={"name": new_name},
        )
        assert r.status_code == 200
        assert r.json()["name"] == new_name

    @stateful.rule(user_id=stateful.consumes(created_user_ids))
    def delete_user(self, user_id):
        r = requests.delete(f"{BASE_URL}/users/{user_id}")
        assert r.status_code == 204

        # Verify deletion
        r = requests.get(f"{BASE_URL}/users/{user_id}")
        assert r.status_code == 404

TestAPIStateMachine = APIStateMachine.TestCase

---

Response Schema Validation

Runtime Response Validation

import jsonschema

def validate_response_against_schema(
    response_data: dict,
    openapi_spec: dict,
    path: str,
    method: str,
    status_code: int,
) -> list[str]:
    """Validate API response matches OpenAPI schema."""
    errors = []

    try:
        response_schema = (
            openapi_spec["paths"][path][method]["responses"]
            [str(status_code)]["content"]["application/json"]["schema"]
        )
    except KeyError:
        errors.append(f"No schema defined for {method.upper()} {path} {status_code}")
        return errors

    # Resolve $ref if present
    response_schema = resolve_refs(response_schema, openapi_spec)

    validator = jsonschema.Draft7Validator(response_schema)
    for error in validator.iter_errors(response_data):
        errors.append(f"{error.json_path}: {error.message}")

    return errors

Middleware Validation (Python)

from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import json

class SchemaValidationMiddleware(BaseHTTPMiddleware):
    """Validate all responses match OpenAPI schema in dev/test."""

    def __init__(self, app, openapi_spec: dict):
        super().__init__(app)
        self.spec = openapi_spec

    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)

        if request.url.path.startswith("/api/"):
            body = b""
            async for chunk in response.body_iterator:
                body += chunk

            errors = validate_response_against_schema(
                response_data=json.loads(body),
                openapi_spec=self.spec,
                path=request.url.path,
                method=request.method.lower(),
                status_code=response.status_code,
            )

            if errors:
                # Log schema violations (do not block in prod)
                logger.warning(f"Schema violation: {errors}")

            return Response(
                content=body,
                status_code=response.status_code,
                headers=dict(response.headers),
            )
        return response

---

Schema Evolution Testing

Track Schema Changes Over Time

def test_schema_evolution_is_backward_compatible(
    current_schema: dict,
    previous_schema: dict,
):
    """Verify schema changes are backward compatible."""
    current_required = set(current_schema.get("required", []))
    previous_required = set(previous_schema.get("required", []))

    # New required fields = breaking change
    new_required = current_required - previous_required
    assert not new_required, (
        f"New required fields break backward compatibility: {new_required}"
    )

    # Removed fields = breaking change
    previous_fields = set(previous_schema.get("properties", {}).keys())
    current_fields = set(current_schema.get("properties", {}).keys())
    removed_fields = previous_fields - current_fields
    assert not removed_fields, (
        f"Removed fields break backward compatibility: {removed_fields}"
    )

    # Type changes = breaking change
    for field in previous_fields & current_fields:
        prev_type = previous_schema["properties"][field].get("type")
        curr_type = current_schema["properties"][field].get("type")
        assert prev_type == curr_type, (
            f"Type change for '{field}': {prev_type} -> {curr_type}"
        )

---

Custom Strategies for Domain Types

Building Domain-Specific Generators

from hypothesis import strategies as st

# Currency amount: positive, 2 decimal places
currency = st.decimals(
    min_value="0.01", max_value="999999.99", places=2
).map(float)

# Phone number: E.164 format
phone_number = st.from_regex(r"\+[1-9]\d{6,14}", fullmatch=True)

# ISO country code
country_code = st.sampled_from(["US", "GB", "DE", "FR", "JP", "AU", "CA"])

# Slug (URL-safe string)
slug = st.from_regex(r"[a-z0-9]+(-[a-z0-9]+)*", fullmatch=True).filter(
    lambda s: 3 <= len(s) <= 60
)

# Composite domain object
order_strategy = st.fixed_dictionaries({
    "customer_id": st.uuids().map(str),
    "items": st.lists(
        st.fixed_dictionaries({
            "sku": st.from_regex(r"[A-Z]{3}-\d{4}", fullmatch=True),
            "quantity": st.integers(min_value=1, max_value=100),
            "unit_price": currency,
        }),
        min_size=1,
        max_size=20,
    ),
    "currency": st.sampled_from(["USD", "EUR", "GBP"]),
    "shipping_country": country_code,
})

---

Schema Testing Checklist

• [ ] OpenAPI spec is the single source of truth for all tests
• [ ] Schemathesis runs against all endpoints in CI
• [ ] Property-based tests cover core creation/read/update/delete flows
• [ ] Negative tests generated from schema for all required fields
• [ ] Boundary values tested for all constrained fields
• [ ] Stateful testing validates multi-step workflows
• [ ] Response schema validation runs in test/dev environments
• [ ] Schema evolution checked for backward compatibility
• [ ] Custom strategies built for domain-specific types
• [ ] Failing examples are minimized and saved for regression
• [ ] CI gates on server errors (5xx) from fuzzed input

---

Related Resources

• [contract-testing-patterns.md](contract-testing-patterns.md) - Consumer-driven contract testing
• [api-versioning-strategies.md](api-versioning-strategies.md) - Versioning and compatibility
• [api-security-testing.md](api-security-testing.md) - Security-focused API testing
• [SKILL.md](../SKILL.md) - QA API Testing & Contracts skill overview