如何为涉及大量网络请求和数据库操作的 Scrapy 项目编写 unit test？

这个项目的大部分解析逻辑都可以用快速、离线的 unit test 验证：构造 HtmlResponse，调用 spider 回调并消费返回结果，再检查生成的 Request 或 item。网络下载、数据库写入和文件导出分别测试，最后用少量完整抓取测试检查它们能否配合工作。

使用 Scrapy 的 Response、XPath 和 Request，不会自动让测试变成 integration test。判断时应看测试覆盖的行为、跨越的边界，以及失败后是否容易定位问题。

当前测试没有执行预期的行为

parse() 是生成器，调用后还没有执行解析

由于函数中包含 yield，下面的调用只会返回一个生成器对象：

spider.parse(response=response)

消费生成器时，函数体才会执行，例如：

outputs = list(spider.parse(response))

生成的 Request 对象也不会自动发送请求。正常抓取时，Scrapy 引擎负责调度请求，再把下载结果交给对应的回调。直接调用 parse() 不会启动这条流程。

文件写入超出了 parse() 的职责

给出的 parse() 负责生成详情页请求和下一页请求，parseObjectDownloadLink() 负责生成 item，两者都不直接写文件。

如果项目通过 Feed Exports 或 item pipeline 保存数据，写入工作由相应组件完成。因此，parse() 测试应检查生成的请求，文件导出则单独测试。

文件断言与提示文字相反

原来的代码：

fileEmptyOrNot = os.stat("../../../artworkData.jsonl").st_size == 0
self.assertEqual(fileEmptyOrNot, True, ...)

实际上要求文件为空。即使改成要求文件非空，也只能证明文件里有字节，无法证明本次解析正确，也无法排除上一次运行留下的数据。

另外，下面的写法没有构造出预期的 offset=0 查询参数：

spider.url + "0"

测试中的响应 URL 也应与待测试页面一致，因为相对链接解析和分页逻辑可能依赖它。

unit test、integration test 与测试大小如何区分

测试范围与运行资源是两个不同的维度。

这些名称的边界并不完全统一。有些团队会把使用真实 Scrapy 对象的解析测试称为组件测试，这不影响测试本身的价值。

读取本地 HTML fixture，也不意味着测试就属于 integration test。即便某套规范把磁盘访问归到 medium，它描述的也是资源限制。测试可以只覆盖很小的范围，同时需要读取文件。

同样，不能仅凭“只调用一个方法”就把测试归为 unit test。这个方法可能访问多个外部服务；一个测试也可能调用几个协作函数，共同验证一项独立行为。

直接测试回调的输入和输出

通常可以保留真实的 HtmlResponse、XPath 解析和 Request。如果全部 mock 掉，测试就会绕过容易出错的选择器和链接处理。

原代码中的 params 是类级可变字典，多个 spider 实例会共享它，影响测试隔离。应将它移到实例初始化中，并删除原来的类级 params：

class MetSpider(scrapy.Spider):
    name = "metQueryScraper"

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.params = {"offset": "0"}

    # 保留其他方法

以下测试使用 pytest 直接调用原有回调。示例 HTML 按题目中的选择器构造，不代表博物馆网站当前的页面结构。

from urllib.parse import parse_qs, urlsplit

import pytest
from scrapy import Request
from scrapy.http import HtmlResponse

from spiders.metQueryScraper import MetSpider


def make_response(html, url, meta=None):
    request = Request(url=url, meta=meta)
    return HtmlResponse(
        url=url,
        request=request,
        body=html.encode("utf-8"),
        encoding="utf-8",
    )


@pytest.fixture
def spider():
    return MetSpider()


def test_parse_creates_artwork_request(spider):
    html = """
    <figure class="collection-object">
      <div class="collection-object-module">
        <a href="/art/collection/search/123" title="树木习作">
          查看作品
        </a>
      </div>
    </figure>
    """
    response = make_response(
        html,
        url=spider.url + "offset=0",
    )

    outputs = list(spider.parse(response))

    assert len(outputs) == 1
    request = outputs[0]
    assert isinstance(request, Request)
    assert request.url == (
        "https://www.metmuseum.org/art/collection/search/123"
    )
    assert request.callback == spider.parseObjectDownloadLink
    assert request.meta["artObject"]["title"] == "树木习作"


@pytest.mark.parametrize(
    "offset, has_next, expected_offsets",
    [
        ("0", False, []),
        ("0", True, ["40"]),
        ("40", True, ["80"]),
        ("80", True, []),
    ],
)
def test_parse_pagination(
    spider, offset, has_next, expected_offsets
):
    spider.params["offset"] = offset
    html = """
    <div class="pagination-controls">
      <button aria-label="Next Page">Next</button>
    </div>
    """ if has_next else "<html></html>"

    response = make_response(
        html,
        url=spider.url + f"offset={offset}",
    )

    outputs = list(spider.parse(response))

    assert len(outputs) == len(expected_offsets)
    for request, expected_offset in zip(outputs, expected_offsets):
        assert isinstance(request, Request)
        assert request.callback == spider.parse
        query = parse_qs(urlsplit(request.url).query)
        assert query["offset"] == [expected_offset]


def test_parse_object_adds_download_link(spider):
    art_object = {"title": "树木习作"}
    response = make_response(
        """
        <figure itemtype="https://schema.org/ImageObject">
          <img fetchpriority="high"
               src="https://images.example.org/artwork.jpg">
        </figure>
        """,
        url="https://www.metmuseum.org/art/collection/search/123",
        meta={"artObject": art_object},
    )

    outputs = list(spider.parseObjectDownloadLink(response))

    assert outputs == [
        {
            "title": "树木习作",
            "link": "https://images.example.org/artwork.jpg",
        }
    ]

这些测试不会发送网络请求。详情页测试构造了带有 meta 的 Request，因为原回调通过 response.meta 读取上一阶段传来的作品信息。

检查请求 URL、回调和传递的数据，就能验证解析结果是否符合约定。无需断言 xpath() 的调用次数或内部变量名。

还可以逐步补充这些情况：

• 搜索页没有作品，也没有下一页。
• 一页包含多件作品。
• 标题或图片地址缺失。
• 同一作品重复出现。

遇到缺失字段时，应先约定处理方式：输出 None、丢弃 item，还是记录错误，再用测试固定规则。

录制响应可以使用，但要分清保存的内容

可以保存真实响应作为 fixture，但需要区分两类文件：

• HTML 正文，可以直接作为 HtmlResponse.body。
• HTTP 录制文件，可能包含请求、状态码、响应头和编码后的正文。

如果 artwork-recorded-session.json 是 Betamax cassette，把整个文件读成字节后传给 HtmlResponse，并不等于重放响应。应使用录制工具重放，或按文件的实际格式提取并解码响应正文。如果接口返回的是 JSON，就应使用 JSON 数据测试相应的解析逻辑。

Betamax 用于录制 requests.Session 的 HTTP 交互。Scrapy 默认下载链路不经过 requests.Session，因此仅继承 BetamaxTestCase，无法自动录制或重放 Scrapy 请求。使用前需要确认录制工具是否支持项目实际使用的 HTTP 客户端。

这类 spider 通常可以搭配使用两种 fixture：

1. 少量手写 HTML，分别覆盖分页、空结果、缺失字段等情况。
2. 少量从真实页面保存的 HTML，检查选择器能否适配有代表性的页面结构。

文件可以放在 tests/fixtures/，根据测试文件的位置定位：

from pathlib import Path

fixture_path = (
    Path(__file__).parent / "fixtures" / "met_search.html"
)
html = fixture_path.read_text(encoding="utf-8")

无需为预计抓取的每一个页面保存样本。测试数据应覆盖页面类型和处理分支，数据量、并发和吞吐另做性能测试。

固定 fixture 不会随外部网站更新。可以保留少量定期运行的在线冒烟测试，并限制请求量。外部站点故障与本地解析测试失败应分开处理。

数据库和文件导出分别怎么测

题目没有提供数据库代码，暂时无法给出针对具体 pipeline 的可运行测试，但可以先确定测试边界。

负责清洗、校验和决定是否保存的逻辑，可以通过一个小的 repository 接口访问存储，再在 unit test 中用内存实现或 mock 替代它。例如，检查有效 item 是否交给 repository，缺少必填字段的 item 是否按约定被拒绝。

SQL、唯一约束和事务行为需要连接实际测试数据库验证。应尽量使用与生产相同的数据库引擎，重点检查：

• 字段是否正确写入，读取后的值是否符合预期。
• 重复作品是否按约定更新、忽略或报错。
• 写入失败时，事务是否按预期回滚。

用 mock 断言 save() 被调用，只能证明调用发生过，无法证明数据库实际保存了数据。

文件导出测试应使用测试专属的临时目录，例如 pytest 的 tmp_path。等受控抓取或导出流程完成、文件关闭后，再逐行解析 JSON Lines，比较记录数量和字段值。这样既能检查实际内容，也能避免固定路径中历史文件的干扰。

要验证整条流程，可以启动一个只提供少量 fixture 页面的本地 HTTP 服务，让 Scrapy 完成“搜索页 → 详情页 → item → 存储”。这种测试能覆盖直接调用回调时未经过的调度、下载和 pipeline 行为。

是否需要为了 unit test 重构代码

大量使用框架很正常，不能只看各类测试的数量来判断代码结构。可以检查一个具体问题：修改标题清洗规则后，是否必须联网、启动整个 crawler 并连接数据库才能验证。

如果是，可以逐步拆分职责：

• 将标题清洗、字段校验、作品标识生成等规则写成输入输出明确的函数。
• 由 spider 从响应中提取数据，并生成后续请求。
• 由 pipeline 或 repository 负责持久化。

当前分页逻辑也值得调整。共享的 self.params["offset"] 会随回调执行而变化，加入多个入口或并发处理后，它可能无法准确表示当前响应属于哪一页。可以从 response.url 读取当前 offset，再计算下一页，使分页决策依赖当前响应。

这些拆分可以逐步完成。先写好 parse() 输出、详情页字段和分页边界的测试，再补充数据库与完整抓取测试，方便在失败时定位问题。

备注：内容仅供参考。