GraphQL 接口测试：Schema 校验、查询与变更用例、Mock Resolver 与 N+1 防护

层次一：Schema 层    —— schema lint、breaking change、指令校验
层次二：解析器层    —— 单个 resolver 的单元测试（mock 数据源）
层次三：执行层      —— 完整 query/mutation 的集成测试（mock 或真实数据源）
层次四：契约层      —— 消费者契约 + 响应快照
层次五：非功能层    —— N+1、深度限制、复杂度限流、超时

type Order {
  id: ID!
  status: OrderStatus!
  amount: Decimal!
  items: [OrderItem!]!
}

enum OrderStatus {
  PENDING
  PAID
  SHIPPED
  CANCELLED
}

type Query {
  order(id: ID!): Order
  orders(first: Int = 20, after: String): OrderConnection!
}

# 安装并运行
npx graphql-schema-linter schema.graphql

# 典型规则
# · 类型名必须 PascalCase，字段名必须 camelCase
# · 枚举值必须 SCREAMING_SNAKE_CASE
# · 所有 mutation 必须有描述
# · 字段必须避免复数/单数歧义

# .graphql-schema-linterrc
rules:
  - fields-have-descriptions
  - types-have-descriptions
  - enum-values-sorted-alphabetically
  - input-object-values-are-camel-cased
  - relay-connection-types-spec

# 与主分支的 schema 对比
npx graphql-inspector diff \
  git:origin/main:schema.graphql \
  schema.graphql

# 输出示例：
# ✖ Field 'Order.legacyStatus' was removed
# ⚠ Enum value 'OrderStatus.REFUNDING' was added
# ✔ No breaking changes detected

# 在 CI 中阻断破坏性变更
npx graphql-inspector diff \
  git:origin/main:schema.graphql \
  schema.graphql \
  --rule suppressRemovalOfDeprecatedField
# 退出码非 0 即阻断合并

# test_schema_snapshot.py
from graphql import build_schema, print_schema

def test_schema_snapshot():
    with open("schema.graphql") as f:
        current = print_schema(build_schema(f.read()))
    with open("tests/snapshots/schema.snapshot.graphql") as f:
        expected = f.read()
    assert current == expected, "Schema 已变更，请审阅并更新快照"

# 用例：空结果与空列表的语义差异
query EmptyCase {
  single: order(id: "does-not-exist") { id }      # 期望 null
  many: orders(first: 0) { edges { node { id } } } # 期望 edges: []
}

mutation PlaceOrder($input: PlaceOrderInput!) {
  placeOrder(input: $input) {
    order { id status }
    userErrors { field message }
  }
}

def test_place_order_creates_side_effect(client, db):
    resp = client.post("/graphql", json={
        "query": PLACE_ORDER,
        "variables": {"input": {"productId": "p-1", "quantity": 1}},
    })
    data = resp.json()["data"]["placeOrder"]
    assert data["userErrors"] == []
    assert data["order"]["status"] == "PENDING"
    # 断言副作用：数据库里确实多了一条订单
    assert db.count_orders(product_id="p-1") == 1

def test_place_order_is_idempotent_with_key(client, db):
    key = "idem-key-123"
    for _ in range(3):
        client.post("/graphql", json={
            "query": PLACE_ORDER,
            "variables": {"input": {"productId": "p-1", "quantity": 1,
                                     "idempotencyKey": key}},
        })
    # 三次相同幂等键只应产生一条订单
    assert db.count_orders(product_id="p-1") == 1

def test_partial_success(client):
    # 一个查询里混合了成功字段和失败字段
    query = """
    query {
      goodOrder: order(id: "1") { id status }
      badOrder: order(id: "boom") { id status }   # resolver 抛异常
    }
    """
    resp = client.post("/graphql", json={"query": query})
    body = resp.json()
    # 协议层：HTTP 200
    assert resp.status_code == 200
    # 部分成功：goodOrder 有值
    assert body["data"]["goodOrder"]["status"] == "PAID"
    # badOrder 为 null，且 errors 里有对应条目
    assert body["data"]["badOrder"] is None
    assert body["errors"][0]["path"] == ["badOrder"]

测试栈（自下而上）：
  真实 GraphQL 执行引擎（保留）
  真实 Schema 与校验（保留）
  ← 在这里替换 → Resolver 依赖的 DataSource（mock/fake）
  真实数据库 / 下游 HTTP（替换为内存 fake）

import { ApolloServer } from "@apollo/server";
import { addMocksToSchema } from "@graphql-tools/mock";
import { makeExecutableSchema } from "@graphql-tools/schema";

const schema = makeExecutableSchema({ typeDefs, resolvers });

// 为关键类型提供确定性 mock
const mocks = {
  Order: () => ({
    id: "order-1",
    status: "PAID",
    amount: 199.0,
  }),
  Query: () => ({
    order: () => ({ id: "order-1", status: "PAID", amount: 199.0 }),
  }),
};

const server = new ApolloServer({
  schema: addMocksToSchema({ schema, mocks, preserveResolvers: false }),
});

# 用依赖注入替换数据源
class FakeOrderRepository:
    def __init__(self):
        self._data = {"1": {"id": "1", "status": "PAID", "amount": 199.0}}
    def get(self, order_id):
        return self._data.get(order_id)
    def count_by_product(self, product_id):
        return sum(1 for o in self._data.values())

def build_test_server():
    return build_schema(repository=FakeOrderRepository())

query {
  orders(first: 50) {
    edges { node { id items { product { name } } } }
  }
}

def test_no_n_plus_one(client, query_counter):
    query = """
    query {
      orders(first: 50) { edges { node { id items { product { name } } } } }
    }
    """
    query_counter.reset()
    client.post("/graphql", json={"query": query})
    # 借助 DataLoader 批处理后，SQL 次数应远小于订单数
    assert query_counter.count <= 5, f"N+1 嫌疑：{query_counter.count} 次查询"

import DataLoader from "dataloader";

const productLoader = new DataLoader(async (ids) => {
  // 一次批量查询所有 product，而非逐个查询
  const rows = await db.query("SELECT * FROM products WHERE id = ANY($1)", [ids]);
  const byId = new Map(rows.map((r) => [r.id, r]));
  return ids.map((id) => byId.get(id));
});

const resolvers = {
  OrderItem: {
    product: (item) => productLoader.load(item.productId),
  },
};

def test_dataloader_batches_calls():
    calls = []
    loader = DataLoader(lambda keys: calls.append(keys) or [f"p-{k}" for k in keys])
    # 触发三次 load，应在同一 tick 内合并为一次 batch
    loader.load("1"); loader.load("2"); loader.load("3")
    assert len(calls) == 1
    assert calls[0] == ["1", "2", "3"]

# 攻击性查询：深度递归
query { user { friends { friends { friends { friends { ... } } } } } }

import depthLimit from "graphql-depth-limit";
import { createComplexityLimitRule } from "graphql-validation-complexity";

const server = new ApolloServer({
  validationRules: [
    depthLimit(7),                                    // 最大深度 7
    createComplexityLimitRule(1000, {                  // 最大复杂度 1000
      onCost: (cost) => console.log("query cost:", cost),
    }),
  ],
});

def test_depth_limit_rejects_deep_query(client):
    deep = "query { " + "a { " * 10 + "id" + " }" * 10 + " }"
    resp = client.post("/graphql", json={"query": deep})
    errors = resp.json().get("errors", [])
    assert any("depth" in e["message"].lower() for e in errors)

def test_introspection_can_be_disabled_in_prod(client_prod):
    resp = client_prod.post("/graphql", json={"query": "{ __schema { types { name } } }"})
    assert resp.json().get("errors"), "生产环境应关闭内省"

// consumer-contract.test.js
const CONSUMER_QUERY = `
  query OrderCard($id: ID!) {
    order(id: $id) {
      id
      status
      amount
    }
  }
`;

test("order-service 满足前端订单卡片契约", async () => {
  const result = await executeOperation(server, {
    query: CONSUMER_QUERY,
    variables: { id: "order-1" },
  });
  expect(result.errors).toBeUndefined();
  expect(result.data.order).toMatchObject({
    id: expect.any(String),
    status: expect.stringMatching(/PENDING|PAID|SHIPPED|CANCELLED/),
    amount: expect.any(Number),
  });
});

def test_order_query_snapshot(client, snapshot):
    resp = client.post("/graphql", json={"query": ORDER_QUERY, "variables": {"id": "1"}})
    # 用正则清洗易变字段（时间戳、id），再做快照
    body = sanitize(resp.json())
    snapshot.assert_match(body, "order_query.json")

jobs:
  graphql-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: npm ci
      - name: Schema lint
        run: npx graphql-schema-linter schema.graphql
      - name: Breaking change check
        run: npx graphql-inspector diff git:origin/main:schema.graphql schema.graphql
      - name: Resolver unit tests
        run: npm test -- --testPathPattern=resolvers
      - name: Query/Mutation integration
        run: npm test -- --testPathPattern=graphql
      - name: N+1 guard
        run: npm test -- --testPathPattern=n_plus_one