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

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

【免费下载链接】scrapyScrapy, a fast high-level web crawling & scraping framework for Python.项目地址: https://gitcode.com/GitHub_Trending/sc/scrapy

本文讲解 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 的源码：

1. 契约检查绕过 item 处理管道。process_options 会在优先级介于 Spider 设置和命令行之间的位置强制把ITEM_PIPELINES和FEEDS置空——契约检查的是回调的原始输出，若管道被触发反而会带来副作用（如创建空输出文件）。如需恢复，可用-s命令行选项重新设置（其优先级更高）。
2. 退出码可接入 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。

【免费下载链接】scrapyScrapy, a fast high-level web crawling & scraping framework for Python.项目地址: https://gitcode.com/GitHub_Trending/sc/scrapy