Scrapy Spiders Contracts：用契约测试（Contract Tests）系统化验证爬虫回调

2026-09-06 21:30:06作者：管翌锬

本文讲解 Scrapy 的 Spiders Contracts 契约测试机制：通过在 Spider 回调函数的 docstring 中写入 @url、@returns、@scrapes 等声明，即可为每个回调自动构建可执行的测试用例，并通过 scrapy check 命令批量运行。读完后你将掌握内置契约的完整用法、自定义契约的编写方式，以及契约检查在源码层面的执行流程（ContractsManager 如何解析 docstring、如何改写回调以执行 pre/post 钩子）。

为什么需要契约测试

编写爬虫时，"某个回调在给定页面上是否产出了预期的 items / requests"这类断言如果用传统单元测试来写，需要手动构造 Crawler、Response、Item 等对象，样板代码很多，维护成本快速上升。Scrapy 内置的契约机制把测试声明直接写进回调的 docstring：你只需硬编码一个示例 URL，并声明若干约束（返回数量、字段是否存在等），scrapy check 命令会真实发起请求、执行回调、收集输出并逐条断言。

从测试模块可以看到，Scrapy 自身对契约机制覆盖了大量场景：同步/异步回调、异步生成器、@cb_kwargs、@meta、边界值、errback 等都有对应用例，说明这是框架内建且经过验证的测试途径。

契约的书写方式：写在 docstring 里

每条契约以 @ 前缀开头，直接混写在回调函数的 docstring 中。官方文档给出的示例如下：

def parse(self, response):
    """
    This function parses a sample response. Some contracts are mingled
    with this docstring.

    @url http://www.example.com/s?field-keywords=selfish+gene
    @returns items 1 16
    @returns requests 0 0
    @scrapes Title Author Year Price
    """

解析规则来自 ContractsManager.extract_contracts：逐行扫描 docstring，对以 @ 开头的行用正则 @(\w+)\s*(.*) 匹配出契约名和参数（按空白拆分），再实例化为对应契约类。两个关键细节值得注意：

方法是否被测试的判定：ContractsManager.tested_methods_from_spidercls 用正则 ^\s*@（多行模式）检查 docstring 是否存在 @ 行——docstring 中至少有一条契约的回调才会进入检查，缺少契约的回调会被整体忽略。
当前版本支持 async def 回调（包括异步生成器回调）。从源码看，add_pre_hook / add_post_hook 会用 _is_async 判断回调是否为协程/异步生成器函数，并分别为同步和异步包装出不同的 wrapper；若同步方法返回了裸协程对象，_collect 会抛出 TypeError 提示"必须用 async def 定义"。

五个内置契约详解

内置契约全部位于 scrapy/contracts/default.py，并在默认设置中通过 SPIDER_CONTRACTS_BASE 以优先级注册：

SPIDER_CONTRACTS: dict[str, int] = {}
SPIDER_CONTRACTS_BASE = {
    "scrapy.contracts.default.UrlContract": 1,
    "scrapy.contracts.default.CallbackKeywordArgumentsContract": 1,
    "scrapy.contracts.default.MetadataContract": 1,
    "scrapy.contracts.default.ReturnsContract": 2,
    "scrapy.contracts.default.ScrapesContract": 3,
}

@url：指定示例 URL（必填）

@url http://www.example.com/s?field-keywords=selfish+gene

UrlContract 在 adjust_request_args 中把 args["url"] 设置为该值，即契约检查会真实请求这个 URL。它被标注为强制性契约：缺少 @url 的回调在检查时会被忽略（因为无法构造样例请求）。

@cb_kwargs 与 @meta：向回调/请求注入参数

@cb_kwargs {"arg1": "value1", "arg2": "value2"}
@meta {"download_timeout": 10}

CallbackKeywordArgumentsContract 设置样例请求的 cb_kwargs 属性，值必须是合法 JSON 字典，最终会作为关键字参数传给回调（见测试用例 test_cb_kwargs）。
MetadataContract 同理设置 Request.meta，可用于指定下载超时、代理等元数据。

两者都是 json.loads(" ".join(self.args)) 解析，因此 docstring 跨行书写时参数会被空白拼接后再解析。

@returns：约束回调产出的数量

语法为 @returns item(s)|request(s) [min [max]]，上界可选：

@returns request          # 至少 1 个 request
@returns request 2        # 至少 2 个
@returns request 2 10     # 2 到 10 个之间
@returns request 2 2      # 恰好 2 个

ReturnsContract 的实现细节：

第一个参数只接受 item/items 或 request/requests（单复数均可），通过 object_type_verifier 映射分别校验 isinstance(x, Request) 和 is_item(x)；参数个数必须是 1、2 或 3 个，否则抛出 ValueError（对应 test_returns_invalid_argument_count）。
省略下界时默认为 1，省略上界时默认为 +inf；断言失败时抛出 ContractFail，错误信息形如 Returned 92 requests, expected 0..4。
只统计与声明类型匹配的输出元素，其他类型被忽略（test_returns_and_scrapes_ignore_other_types 验证了这一点）。

@scrapes：校验 item 字段是否存在

@scrapes Title Author Year Price

ScrapesContract 在 post_process 中遍历回调输出的每一个 item，用 ItemAdapter 检查所有指定字段是否都存在；一旦某个 item 缺字段，立即抛出 ContractFail("Missing fields: ...") 并列出全部缺失字段名。

契约与回调输出的关系：pre/post 钩子

每个契约实例在 init 中创建两个测试用例（@<name> pre-hook 与 @<name> post-hook）。Contract.add_pre_hook / add_post_hook 会把请求的 callback 包装一层：pre 钩子在回调执行前对 Response 做断言，post 钩子在回调执行后对输出做断言。ReturnsContract 和 ScrapesContract 只实现 post_process，因此它们的断言出现在 post-hook 用例中——这正是 scrapy check 输出里 [first_spider] parse (@returns post-hook) 这类用例名的来源。

在 from_method 中还可以看到：钩子按注册顺序执行，pre 钩子链逆序叠加（for contract in reversed(contracts)）、post 钩子链正序叠加，形成洋葱模型；同时每个契约可通过类属性 request_cls 指定样例请求的 Request 子类（多个契约声明时以最后一个为准），并强制 dont_filter=True 以允许对同一 URL 测试不同回调。

用 scrapy check 命令运行契约检查

使用 check 命令运行检查（详见 check 命令文档）：

$ scrapy check -l
first_spider
  * parse
  * parse_item
second_spider
  * parse
  * parse_item

$ scrapy check
F.F.
======================================================================
FAIL: [first_spider] parse (@returns post-hook)
----------------------------------------------------------------------
Traceback (most recent call last):
  ...
scrapy.exceptions.ContractFail: Returned 92 requests, expected 0..4

命令选项：

-l / --list：仅列出各 Spider 中带契约的方法，不执行检查；
-v / --verbose：打印所有 Spider 的契约测试用例（包括没有契约的）。

两个重要的执行语义，均来自 scrapy/commands/check.py 的源码：

契约检查绕过 item 处理管道。process_options 会在优先级介于 Spider 设置和命令行之间的位置强制把 ITEM_PIPELINES 和 FEEDS 置空——契约检查的是回调的原始输出，若管道被触发反而会带来副作用（如创建空输出文件）。如需恢复，可用 -s 命令行选项重新设置（其优先级更高）。
退出码可接入 CI：run 结束时把 exitcode 设为"是否全部成功"的取反值，因此 scrapy check 非零退出即代表有契约失败或错误。

run 方法还会通过 set_environ(SCRAPY_CHECK="true") 设置环境变量，并临时把每个被检查 Spider 的 start 方法替换为一个直接产出契约请求的异步生成器——也就是说，scrapy check 并非从爬虫的正常入口开始爬取，而是直接把带契约的回调请求交给下载器执行。结果汇总由定制版 TextTestResult.printSummary 打印，形如 Ran 4 contracts in 3.217s / FAILED (failures=2, errors=1)。

编写自定义契约

当内置契约不够用时，可以用 SPIDER_CONTRACTS 设置加载项目自己的契约（settings 文档中的组件设置）：

SPIDER_CONTRACTS = {
    "myproject.contracts.ResponseCheck": 10,
    "myproject.contracts.ItemValidate": 10,
}

自定义契约必须继承 scrapy.contracts.Contract 基类，并声明唯一的 name 属性（对应 docstring 中的 @name）。基类允许覆写三个部分：

adjust_request_args(args)：接收样例请求的默认参数字典（dict），可修改后返回，例如替换 URL、指定请求方法；配合 request_cls 还能换成自定义 Request 子类。
pre_process(response)：在回调收到响应之前对 Response 做断言；
post_process(output)：在回调执行之后处理其输出（迭代器会先被转换为列表再传入）。

期望不满足时，在上述钩子中抛出 scrapy.exceptions.ContractFail（它是 AssertionError 的子类，会被记录为 failure 而非 error）。官方文档中的演示契约——检查响应里是否存在自定义头：

from scrapy.contracts import Contract
from scrapy.exceptions import ContractFail


class HasHeaderContract(Contract):
    """
    Demo contract which checks the presence of a custom header
    @has_header X-CustomHeader
    """

    name = "has_header"

    def pre_process(self, response):
        for header in self.args:
            if header not in response.headers:
                raise ContractFail("X-CustomHeader not present")

使用时的完整形态（@has_header 仍需配合 @url 指定样例 URL）：

def parse(self, response):
    """
    @url https://example.com/api
    @has_header X-CustomHeader
    @returns items 1 3
    """

两点源码级补充：

ContractsManager.init 会对覆写了旧式 add_pre_hook / add_post_hook 的契约发出 ScrapyDeprecationWarning，提示改用 pre_process / post_process——旧式覆写不再支持异步回调。
断言结果经 _run_hook 归一化：抛出 AssertionError（含 ContractFail）计入 failures，其他异常计入 errors，否则记为成功；回调本身的异常则由 _clean_req 追加的包装器捕获，以 [spider] method (callback) 或 (errback) 用例名报告。测试文件中的 test_custom_contracts 与 test_custom_tagged_request_contract 分别演示了自定义契约加载和 request_cls 指定自定义 Request 子类（并顺带修改请求方法为 POST）的完整流程。

识别 check 运行：SCRAPY_CHECK 环境变量

当 scrapy check 正在运行时，环境变量 SCRAPY_CHECK 会被设置为字符串 "true"（见 check 命令源码）。可以用 os.environ 在检查期间临时调整 Spider 或设置的行为，例如降低并发、切换到 mock 站点、跳过某些耗时逻辑：

import os
import scrapy


class ExampleSpider(scrapy.Spider):
    name = "example"

    def __init__(self):
        if os.environ.get("SCRAPY_CHECK"):
            pass  # Do some scraper adjustments when a check is running

这个机制适合把契约测试做成 CI 的一部分：检查环境下的行为可与生产爬取区分开，而不必污染常规配置。

小结与适用边界

契约适合验证"回调对某个真实页面的处理是否符合预期"：产出数量（@returns）、字段完整性（@scrapes）、响应特征（自定义契约）；它不替代面向纯解析逻辑的单元测试——契约检查依赖能访问到 @url 站点，测试 URL 失效时检查也会失败。
检查会真实发起网络请求，但会强制跳过 ITEM_PIPELINES 与 FEEDS，可用 -s 恢复；@url 为必填，缺失时回调被忽略。
当前版本支持 async def 回调与异步生成器；旧式覆写 add_pre_hook/add_post_hook 的自定义契约已弃用且不支持异步回调，应改为实现 pre_process/post_process。
关键源码入口：契约基类与管理器 scrapy/contracts/init.py、内置契约 scrapy/contracts/default.py、检查命令 scrapy/commands/check.py、官方测试 tests/test_contracts.py。