Spiders Contracts

0.15版中的新功能.

测试蜘蛛程序可能会特别烦人,而没有什么可以阻止您编写单元测试,因此任务很快就会变得繁琐. Scrapy提供了一种通过合同测试蜘蛛的综合方法.

这允许您通过对示例URL进行硬编码来测试Spider的每个回调,并检查各种约束以了解回调如何处理响应. 每个合同都以@开头,并包含在文档字符串中. 请参见以下示例:

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

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

使用三个内置合同测试了此回调:

class scrapy.contracts.default.UrlContract

该合同( @url )设置了在检查此蜘蛛的其他合同条件时使用的示例URL. 该合同是强制性的. 运行检查时,将忽略所有缺少此协定的回调:

@url url
class scrapy.contracts.default.CallbackKeywordArgumentsContract

该合同( @cb_kwargs )为示例请求设置cb_kwargs属性. 它必须是有效的JSON字典.

@cb_kwargs {"arg1": "value1", "arg2": "value2", ...}
class scrapy.contracts.default.ReturnsContract

该合约( @returns )为蜘蛛返回的项目和请求设置上下限. 上限是可选的:

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

该合同( @scrapes )检查回调返回的所有项目是否具有指定的字段:

@scrapes field_1 field_2 ...

使用check命令运行合同检查.

Custom Contracts

如果发现您需要的能量比内置的scrapy合同更多,则可以使用SPIDER_CONTRACTS设置在项目中创建和加载自己的合同:

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

每个合同必须继承自Contract并且可以覆盖以下三种方法:

class scrapy.contracts.Contract(method, *args)
Parameters:
  • 方法函数 )–与合同关联的回调函数
  • args (list) – list of arguments passed into the docstring (whitespace separated)
adjust_request_args(args)

这将收到一个dict作为参数,其中包含请求对象的默认参数. 默认情况下使用Request ,但是可以使用request_cls属性进行更改. 如果链中有多个合同定义了此属性,则使用最后一个.

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

pre_process(response)

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

post_process(output)

这允许处理回调的输出. 迭代器在传递到此挂钩之前先进行转换列表化.

如果未达到期望,则从pre_processpost_process提高ContractFail

class scrapy.exceptions.ContractFail

如果合同失败,则会引发错误

这是一个演示合同,用于检查收到的响应中是否存在自定义标头:

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

Detecting check runs

运行scrapy check时,将SCRAPY_CHECK环境变量设置为true字符串. 使用scrapy check时,可以使用os.environ对蜘蛛或设置进行任何更改:

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