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-keywordsselfishgene 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-keywordsselfishgeneUrlContract 在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 PriceScrapesContract 在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_filterTrue以允许对同一 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命令行选项重新设置其优先级更高。退出码可接入 CIrun 结束时把exitcode设为是否全部成功的取反值因此scrapy check非零退出即代表有契约失败或错误。run方法还会通过set_environ(SCRAPY_CHECKtrue)设置环境变量并临时把每个被检查 Spider 的start方法替换为一个直接产出契约请求的异步生成器——也就是说scrapy check并非从爬虫的正常入口开始爬取而是直接把带契约的回调请求交给下载器执行。结果汇总由定制版 TextTestResult.printSummary 打印形如Ran 4 contracts in 3.217s / FAILED (failures2, errors1)。编写自定义契约当内置契约不够用时可以用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指定样例 URLdef 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考