Spiders Contracts

测试 spider 可能会特别令人烦恼,虽然没有什么能阻止你编写单元测试,但这任务很快就会变得麻烦。Scrapy 提供了一种集成的方式来测试你的 spider,即通过 contracts。

这允许你通过硬编码一个示例 URL 来测试 spider 的每个回调函数,并检查回调函数处理响应时是否满足各种约束。每个 contract 都以一个 @ 为前缀并包含在 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
    """

你可以使用以下 contracts

class scrapy.contracts.default.UrlContract[源码]

此 contract (@url) 设置用于检查此 spider 的其他 contract 条件时的示例 URL。此 contract 是必需的。运行检查时,所有缺少此 contract 的回调函数都会被忽略。

@url url
class scrapy.contracts.default.CallbackKeywordArgumentsContract[源码]

此 contract (@cb_kwargs) 为示例请求设置 cb_kwargs 属性。它必须是一个有效的 JSON 字典。

@cb_kwargs {"arg1": "value1", "arg2": "value2", ...}
class scrapy.contracts.default.MetadataContract[源码]

此 contract (@meta) 为示例请求设置 meta 属性。它必须是一个有效的 JSON 字典。

@meta {"arg1": "value1", "arg2": "value2", ...}
class scrapy.contracts.default.ReturnsContract[源码]

此 contract (@returns) 设置 spider 返回的 item 和请求的下限和上限。上限是可选的。

@returns item(s)|request(s) [min [max]]
class scrapy.contracts.default.ScrapesContract[源码]

此 contract (@scrapes) 检查回调函数返回的所有 item 是否具有指定的字段。

@scrapes field_1 field_2 ...

使用 check 命令来运行 contract 检查。

自定义 Contracts

如果你觉得内置的 Scrapy contract 不够强大,可以通过 SPIDER_CONTRACTS 设置在项目中创建和加载你自己的 contracts。

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

每个 contract 必须继承自 Contract,并且可以重写三个方法。

class scrapy.contracts.Contract(method, *args)[源码]
参数:
  • method (collections.abc.Callable) – 与 contract 关联的回调函数

  • args (list) – docstring 中传入的参数列表(空格分隔)

adjust_request_args(args)[源码]

这接收一个 dict 作为参数,其中包含 request 对象的默认参数。默认使用 Request,但这可以通过 request_cls 属性更改。如果链中多个 contract 定义了此属性,则使用最后一个定义。

必须返回相同或修改后的版本。

pre_process(response)

这允许在接收到示例请求的响应后、将其传递给回调函数之前,进行各种检查。

post_process(output)

这允许处理回调函数的输出。迭代器在传递到此钩子之前会转换为列表。

如果期望不满足,则从 pre_processpost_process 中引发 ContractFail

class scrapy.exceptions.ContractFail[源码]

contract 检查失败时引发的错误。

这里是一个示例 contract,它检查接收到的响应中是否存在自定义头部。

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")

检测检查运行

scrapy check 正在运行时,SCRAPY_CHECK 环境变量会被设置为 true 字符串。你可以使用 os.environ 在使用 scrapy check 时对你的 spider 或设置进行任何更改。

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