API 测试自动化：REST / GraphQL / gRPC 统一策略

┌────────────────────────────────────────────────────────────┐
│                    API 测试分层模型                         │
├────────────────────────────────────────────────────────────┤
│                                                            │
│   契约验证层    ──►  Schema / Spec 一致性                   │
│   (Contract)       OpenAPI, Protobuf, GraphQL SDL            │
│                                                            │
│   功能验证层    ──►  状态码、响应体、业务逻辑               │
│   (Functional)     CRUD, 边界值, 错误场景                   │
│                                                            │
│   非功能验证层  ──►  性能、安全、可靠性                     │
│   (Non-functional) 延迟基线, 认证, 速率限制                  │
│                                                            │
└────────────────────────────────────────────────────────────┘

import pytest
import requests
from pydantic import BaseModel, Field
from typing import List

# --- 1. 定义响应 Schema（比 dict 更安全） ---
class OrderItem(BaseModel):
    product_id: str
    quantity: int = Field(gt=0)
    price: float = Field(gt=0)

class OrderResponse(BaseModel):
    id: str
    customer_id: str
    status: str  # pending, confirmed, shipped
    total_amount: float = Field(ge=0)
    items: List[OrderItem]
    created_at: str

class ErrorResponse(BaseModel):
    error: str
    message: str
    details: dict | None = None

BASE_URL = "http://localhost:8000"

@pytest.fixture
def auth_headers():
    """获取认证Token。"""
    resp = requests.post(f"{BASE_URL}/api/auth/login", json={
        "email": "test@example.com",
        "password": "testpass123"
    })
    token = resp.json()["access_token"]
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

class TestOrderAPI:
    def test_create_order_success(self, auth_headers):
        response = requests.post(
            f"{BASE_URL}/api/orders",
            headers=auth_headers,
            json={
                "customer_id": "cust-001",
                "items": [
                    {"product_id": "p1", "quantity": 2, "price": 29.99}
                ]
            }
        )
        assert response.status_code == 201

        # Pydantic 自动校验整个响应结构
        order = OrderResponse(**response.json())
        assert order.status == "pending"
        assert order.total_amount == 59.98
        assert len(order.items) == 1

        # 验证 Header
        assert "/api/orders/" in response.headers.get("Location", "")

    @pytest.mark.parametrize("payload,expected_error", [
        # 边界值测试
        ({"customer_id": "", "items": []}, "VALIDATION_ERROR"),
        ({"customer_id": "x", "items": [{"product_id": "p1", "quantity": 0}]}, "VALIDATION_ERROR"),
        ({"customer_id": "x" * 1000, "items": [{"product_id": "p1", "quantity": 1, "price": -1}]}, "VALIDATION_ERROR"),
        # 缺少必填字段
        ({"items": []}, "VALIDATION_ERROR"),
    ])
    def test_create_order_validation_errors(self, auth_headers, payload, expected_error):
        response = requests.post(
            f"{BASE_URL}/api/orders",
            headers=auth_headers,
            json=payload
        )
        assert response.status_code == 422
        error = ErrorResponse(**response.json())
        assert error.error == expected_error

    def test_get_order_not_found(self, auth_headers):
        response = requests.get(
            f"{BASE_URL}/api/orders/nonexistent-id",
            headers=auth_headers
        )
        assert response.status_code == 404

    def test_unauthorized_access(self):
        """未认证请求应返回 401。"""
        response = requests.get(f"{BASE_URL}/api/orders")
        assert response.status_code == 401
        assert "Unauthorized" in response.json()["message"]

    def test_pagination(self, auth_headers):
        """分页参数边界测试。"""
        # 正常分页
        resp = requests.get(f"{BASE_URL}/api/orders?page=1&limit=10", headers=auth_headers)
        assert resp.status_code == 200
        data = resp.json()
        assert "items" in data
        assert "total" in data
        assert "page" in data

        # 越界页码
        resp = requests.get(f"{BASE_URL}/api/orders?page=99999&limit=10", headers=auth_headers)
        assert resp.status_code == 200
        assert len(resp.json()["items"]) == 0

        # limit 过大
        resp = requests.get(f"{BASE_URL}/api/orders?page=1&limit=99999", headers=auth_headers)
        assert resp.status_code == 400  # 或强制限制到最大值

import io.restassured.RestAssured;
import io.restassured.http.ContentType;
import io.restassured.module.jsv.JsonSchemaValidator;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

import static io.restassured.RestAssured.given;
import static org.hamcrest.Matchers.*;

class OrderApiTest {

    @BeforeAll
    static void setUp() {
        RestAssured.baseURI = "http://localhost:8080";
    }

    String getAuthToken() {
        return given()
            .contentType(ContentType.JSON)
            .body("{\"email\":\"test@example.com\",\"password\":\"testpass123\"}")
        .when()
            .post("/api/auth/login")
        .then()
            .statusCode(200)
            .extract().path("access_token");
    }

    @Test
    void shouldCreateOrder() {
        String token = getAuthToken();

        given()
            .contentType(ContentType.JSON)
            .header("Authorization", "Bearer " + token)
            .body("""
                {
                    "customerId": "cust-001",
                    "items": [
                        {"productId": "p1", "quantity": 2, "price": 29.99}
                    ]
                }
                """)
        .when()
            .post("/api/orders")
        .then()
            .statusCode(201)
            .body("status", equalTo("pending"))
            .body("totalAmount", closeTo(59.98f, 0.01f))
            .body("items", hasSize(1))
            .header("Location", containsString("/api/orders/"))
            // JSON Schema 校验
            .body(JsonSchemaValidator
                .matchesJsonSchemaInClasspath("schemas/order-response.json"));
    }

    @Test
    void shouldReturnValidationError() {
        given()
            .contentType(ContentType.JSON)
            .header("Authorization", "Bearer " + getAuthToken())
            .body("{\"customerId\": \"\", \"items\": []}")
        .when()
            .post("/api/orders")
        .then()
            .statusCode(400)
            .body("error", equalTo("VALIDATION_ERROR"))
            .body("details", notNullValue());
    }
}

# 安装 Newman
npm install -g newman newman-reporter-htmlextra

# 执行 Collection
newman run api-tests.postman_collection.json \
  -e production.postman_environment.json \
  --reporters cli,htmlextra,junit \
  --reporter-junit-export results/junit.xml \
  --reporter-htmlextra-export results/report.html

# 失败时非零退出码，CI 自动阻断

# .github/workflows/api-test.yml
name: API Tests
on: [push, pull_request]

jobs:
  api-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Start services
        run: docker-compose -f docker-compose.test.yml up -d

      - name: Wait for services
        run: npx wait-on http://localhost:8080/health --timeout 30000

      - name: Run Newman
        run: |
          npx newman run tests/api/collection.json \
            -e tests/api/env.ci.json \
            --reporters cli,junit \
            --reporter-junit-export newman-results.xml

      - name: Upload results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: api-test-results
          path: newman-results.xml

import { MockedProvider } from '@apollo/client/testing';
import { render, screen, waitFor } from '@testing-library/react';
import { GET_USER_ORDERS } from './queries';
import { UserOrders } from './UserOrders';

const mocks = [
  {
    request: {
      query: GET_USER_ORDERS,
      variables: { userId: 'user-123', limit: 10 },
    },
    result: {
      data: {
        user: {
          id: 'user-123',
          orders: [
            { id: 'order-1', total: 99.99, status: 'DELIVERED' },
            { id: 'order-2', total: 49.50, status: 'PENDING' },
          ],
        },
      },
    },
  },
  // 错误场景
  {
    request: {
      query: GET_USER_ORDERS,
      variables: { userId: 'invalid', limit: 10 },
    },
    error: new Error('User not found'),
  },
];

it('renders user orders', async () => {
  render(
    <MockedProvider mocks={mocks} addTypename={false}>
      <UserOrders userId="user-123" />
    </MockedProvider>
  );

  await waitFor(() => {
    expect(screen.getByText('order-1')).toBeInTheDocument();
  });

  expect(screen.getAllByTestId('order-item')).toHaveLength(2);
});

import pytest
import requests

GRAPHQL_URL = "http://localhost:8000/graphql"

@pytest.fixture
def graphql_client(auth_headers):
    def execute(query: str, variables: dict = None):
        response = requests.post(
            GRAPHQL_URL,
            headers=auth_headers,
            json={"query": query, "variables": variables or {}}
        )
        data = response.json()
        # GraphQL 错误在 200 响应中
        if "errors" in data:
            raise AssertionError(f"GraphQL errors: {data['errors']}")
        return data["data"]
    return execute

class TestGraphQL:
    def test_get_user_with_orders(self, graphql_client):
        query = """
        query GetUser($id: ID!) {
            user(id: $id) {
                id
                name
                orders(limit: 5) {
                    id
                    total
                    status
                }
            }
        }
        """
        result = graphql_client(query, {"id": "user-123"})
        assert result["user"]["id"] == "user-123"
        assert len(result["user"]["orders"]) <= 5

    def test_mutation_create_order_idempotent(self, graphql_client):
        """幂等性测试：相同 idempotencyKey 返回相同结果。"""
        mutation = """
        mutation CreateOrder($input: CreateOrderInput!) {
            createOrder(input: $input) {
                id
                status
                totalAmount
            }
        }
        """
        variables = {
            "input": {
                "customerId": "cust-001",
                "items": [{"productId": "p1", "quantity": 1}],
                "idempotencyKey": "unique-key-123"
            }
        }
        result1 = graphql_client(mutation, variables)
        result2 = graphql_client(mutation, variables)
        assert result1 == result2  # 幂等

import pytest
import grpc
from concurrent import futures

# 假设有生成的 proto 代码
from generated import order_pb2, order_pb2_grpc
from server import OrderServicer

@pytest.fixture(scope="module")
def grpc_channel():
    """启动内存 gRPC 服务器用于测试。"""
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
    order_pb2_grpc.add_OrderServiceServicer_to_server(OrderServicer(), server)
    port = server.add_insecure_port('localhost:0')
    server.start()

    channel = grpc.insecure_channel(f'localhost:{port}')
    yield channel
    channel.close()
    server.stop(None)

class TestOrderGrpc:
    def test_create_order(self, grpc_channel):
        stub = order_pb2_grpc.OrderServiceStub(grpc_channel)

        request = order_pb2.CreateOrderRequest(
            customer_id="cust-001",
            items=[
                order_pb2.OrderItem(product_id="p1", quantity=2, price=29.99)
            ]
        )

        response = stub.CreateOrder(request)

        assert response.order_id != ""
        assert response.status == order_pb2.OrderStatus.PENDING
        assert abs(response.total_amount - 59.98) < 0.01

    def test_create_order_validation(self, grpc_channel):
        stub = order_pb2_grpc.OrderServiceStub(grpc_channel)

        # 空 items 应该返回 INVALID_ARGUMENT
        with pytest.raises(grpc.RpcError) as exc_info:
            stub.CreateOrder(order_pb2.CreateOrderRequest(
                customer_id="cust-001",
                items=[]
            ))

        assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT

@ExtendWith(GrpcTestExtension.class)
class OrderGrpcServiceTest {

    @GrpcClient
    private OrderServiceGrpc.OrderServiceBlockingStub stub;

    @Test
    void shouldCreateOrder() {
        CreateOrderRequest request = CreateOrderRequest.newBuilder()
            .setCustomerId("cust-001")
            .addItems(OrderItem.newBuilder()
                .setProductId("p1")
                .setQuantity(2)
                .setPrice(29.99)
                .build())
            .build();

        CreateOrderResponse response = stub.createOrder(request);

        assertThat(response.getOrderId()).isNotEmpty();
        assertThat(response.getStatus()).isEqualTo(OrderStatus.PENDING);
        assertThat(response.getTotalAmount()).isCloseTo(59.98, within(0.01));
    }

    @Test
    void shouldRejectEmptyItems() {
        CreateOrderRequest request = CreateOrderRequest.newBuilder()
            .setCustomerId("cust-001")
            .build();

        StatusRuntimeException exception = catchThrowableOfType(
            () -> stub.createOrder(request),
            StatusRuntimeException.class
        );

        assertThat(exception.getStatus().getCode())
            .isEqualTo(Status.Code.INVALID_ARGUMENT);
    }
}

# 安装
pip install schemathesis

# 自动基于 OpenAPI Spec 生成测试用例
st run http://localhost:8000/openapi.json \
  --base-url http://localhost:8000 \
  --checks all \
  --hypothesis-max-examples 100

# 输出：
# GET /api/orders/{orderId} .                                [OK]
# POST /api/orders F                                         [500]
#   Hypothesis found 500 when sending {"items": null}
#   → 缺少 null 校验！

npm install -g dredd

# dredd.yml
# endpoint: http://localhost:8000
# openapi: ./openapi.yaml

dredd
# 自动验证所有端点的实际行为与 OpenAPI 定义的一致性

import jwt
from datetime import datetime, timedelta, timezone

@pytest.fixture
def expired_token():
    """生成一个已过期的 JWT 用于测试。"""
    payload = {
        "sub": "user-123",
        "exp": datetime.now(timezone.utc) - timedelta(hours=1),
        "scope": "read"
    }
    return jwt.encode(payload, "secret", algorithm="HS256")

class TestAuth:
    def test_expired_token_rejected(self, expired_token):
        resp = requests.get(
            f"{BASE_URL}/api/orders",
            headers={"Authorization": f"Bearer {expired_token}"}
        )
        assert resp.status_code == 401
        assert "expired" in resp.json()["message"].lower()

    def test_insufficient_scope(self, auth_headers):
        # scope=read 的 token 尝试写入
        read_only_token = generate_token(scope="read")
        resp = requests.post(
            f"{BASE_URL}/api/orders",
            headers={"Authorization": f"Bearer {read_only_token}"},
            json={"customer_id": "x", "items": []}
        )
        assert resp.status_code == 403

import pytest
import time
import statistics

@pytest.mark.benchmark
class TestAPIPerformance:
    def test_order_list_response_time(self, auth_headers):
        """验证 95% 的请求延迟 < 200ms。"""
        times = []
        for _ in range(20):
            start = time.perf_counter()
            resp = requests.get(f"{BASE_URL}/api/orders?limit=10", headers=auth_headers)
            elapsed = (time.perf_counter() - start) * 1000
            times.append(elapsed)
            assert resp.status_code == 200

        p95 = statistics.quantiles(times, n=20)[18]  # 近似 P95
        avg = statistics.mean(times)

        assert p95 < 200, f"P95 latency {p95:.1f}ms exceeds 200ms"
        assert avg < 100, f"Average latency {avg:.1f}ms exceeds 100ms"