如何为Scrapy爬虫编写测试用例及数据点验证器

你需要为爬虫实现两类验证逻辑：解析函数的单元测试、数据字段合法性校验，你最初想到的unittest单元测试完全适用，以下是具体实现方案：

Scrapy的解析函数仅依赖Response对象，你可以手动构造模拟响应，不需要真实发送网络请求即可完成测试。

1.1 前置准备 #

先抓取3个解析逻辑对应的真实页面HTML，保存到项目的test_samples目录下，作为测试的基准样本。

1.2 测试代码示例 #

import unittest
import scrapy
from scrapy.http import HtmlResponse
from your_spider_file import YourCompanySpider # 替换为你实际的爬虫类名

class TestCompanySpider(unittest.TestCase):
    def setUp(self):
        self.spider = YourCompanySpider()
    
    def test_parse_method(self):
        # 读取一级列表页的测试样本
        with open("test_samples/first_level_page.html", "r", encoding="utf-8") as f:
            html_content = f.read()
        # 构造模拟响应
        response = HtmlResponse(url="https://example.com/first-page", body=html_content, encoding="utf-8")
        # 收集parse方法返回的所有结果
        results = list(self.spider.parse(response))
        # 断言返回的全是Request对象，且回调函数正确
        for res in results:
            self.assertIsInstance(res, scrapy.Request)
            self.assertEqual(res.callback, self.spider.parse_company_index)
        # 可根据样本实际的链接数量，断言返回的Request数量符合预期
        self.assertEqual(len(results), 10) # 替换为你实际的样本链接数
    
    def test_parse_company_index(self):
        with open("test_samples/company_index_page.html", "r", encoding="utf-8") as f:
            html_content = f.read()
        response = HtmlResponse(url="https://example.com/company-index", body=html_content, encoding="utf-8")
        results = list(self.spider.parse_company_index(response))
        
        # 拆分结果：item和请求
        items = [r for r in results if isinstance(r, dict)]
        requests = [r for r in results if isinstance(r, scrapy.Request)]
        
        # 校验item字段
        for item in items:
            self.assertEqual(item["record_type"], "company_index")
            self.assertTrue(item["company_name"].strip()) # 公司名非空
            self.assertTrue(item["source_url"].startswith("http")) # 链接格式合法
        
        # 校验请求：分为详情页请求和下一页请求
        detail_requests = [r for r in requests if r.callback == self.spider.parse_company_profiles]
        next_page_requests = [r for r in requests if r.callback == self.spider.parse_company_index]
        self.assertEqual(len(detail_requests), len(items)) # 每个公司对应一个详情页请求
        self.assertLessEqual(len(next_page_requests), 1) # 最多1个下一页请求
    
    def test_parse_company_profiles(self):
        with open("test_samples/company_profile_page.html", "r", encoding="utf-8") as f:
            html_content = f.read()
        response = HtmlResponse(url="https://example.com/company/123", body=html_content, encoding="utf-8")
        items = list(self.spider.parse_company_profiles(response))
        
        for item in items:
            self.assertEqual(item["record_type"], "company_profiles")
            # 校验基础字段非空
            for field in ["company_name", "company_location", "company_website"]:
                self.assertTrue(item[field].strip())
            # 校验联系人结构
            contact_details = item["contact_details"]
            if contact_details != ["None"]:
                self.assertIsInstance(contact_details[0], list)
                for contact in contact_details[0]:
                    self.assertIn("Contact_name", contact)
                    self.assertIn("Contact_jobtitle", contact)

if __name__ == "__main__":
    unittest.main()

推荐使用Pydantic做数据字段校验，规则配置灵活，还支持自定义校验逻辑。

2.1 定义校验模型 #

from pydantic import BaseModel, HttpUrl, field_validator
from typing import List, Optional, Union

class ContactDetail(BaseModel):
    Contact_name: str
    Contact_jobtitle: str
    Contact_email_domain: str

    @field_validator("Contact_email_domain")
    def check_email_domain(cls, v):
        if "@" not in v:
            raise ValueError(f"非法邮箱域名 {v}")
        return v.strip()

class CompanyIndexItem(BaseModel):
    record_type: str = "company_index"
    company_name: str
    source_url: HttpUrl # 自动校验是否为合法URL

    @field_validator("company_name")
    def name_not_empty(cls, v):
        if not v.strip():
            raise ValueError("公司名不能为空")
        return v.strip()

class CompanyProfileItem(BaseModel):
    record_type: str = "company_profiles"
    company_name: str
    company_location: str
    company_website: HttpUrl
    company_webdomain: str
    company_industry: str
    company_employee_size: str
    company_revenue: str
    contact_details: Union[List[str], List[List[ContactDetail]]]

    @field_validator("company_employee_size")
    def check_employee_size(cls, v):
        if v and not any([k in v for k in ["人", "员工", "employee"]]):
            raise ValueError(f"非法员工数格式 {v}")
        return v.strip()

2.2 接入爬虫 #

在解析函数yield数据之前，先做校验，自动过滤非法数据：

# 在parse_company_index中
for item in data:
    raw_dict = {
        'record_type': 'company_index',
        'company_name': item[0],
        'source_url': item[1],
    }
    # 校验数据
    try:
        valid_item = CompanyIndexItem(**raw_dict)
        yield valid_item.model_dump()
    except Exception as e:
        # 可以打日志记录非法数据，不影响爬虫继续运行
        self.logger.warning(f"无效公司索引数据 {raw_dict}, 错误：{str(e)}")

# parse_company_profiles里也做同样的校验，用CompanyProfileItem即可

• 每次修改XPath规则、解析逻辑后都运行一次单元测试，避免改动引入bug
• 目标网站结构更新后，同步更新测试用的HTML样本，保证测试的准确性
• 可以将数据校验逻辑放到Scrapy Pipeline中统一处理，所有item统一校验，不合格的数据标记后单独存储或触发重爬
• 可新增轻量集成测试，每周发送少量真实请求校验目标网站结构是否发生大范围变更

内容的提问来源于stack exchange，提问作者Shahriar Tasnim