You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何让Scrapy与Pytest适配AsyncioSelectorReactor?

Scrapy爬虫测试中的Reactor不匹配问题及解决方案

问题复现环境

  • python 3.12.1
  • scrapy 2.11.2
  • pytest 8.2.1

我的bookspider.py代码如下:

from typing import Iterable

import scrapy
from scrapy.http import Request


class BookSpider(scrapy.Spider):
    name = None

    def start_requests(self) -> Iterable[Request]:
        yield scrapy.Request("https://books.toscrape.com/")

    def parse(self, response):
        books = response.css("article.product_pod")
        for book in books:
            yield {
                "name": self.name,
                "title": book.css("h3 a::text").get().strip(),
            }

test_bookspider.py代码如下:

import json
import os

from pytest_twisted import inlineCallbacks
from scrapy.crawler import CrawlerRunner
from twisted.internet import defer

from bookspider import BookSpider


@inlineCallbacks
def test_bookspider():
    runner = CrawlerRunner(
        settings={
            "REQUEST_FINGERPRINTER_IMPLEMENTATION": "2.7",
            "FEEDS": {"books.json": {"format": "json"}},
            "TWISTED_REACTOR": "twisted.internet.asyncioreactor.AsyncioSelectorReactor",
            # "TWISTED_REACTOR": "twisted.internet.selectreactor.SelectReactor",
        }
    )
    yield runner.crawl(BookSpider, name="books")

    with open("books.json", "r") as f:
        books = json.load(f)
    assert len(books) >= 1
    assert books[0]["name"] == "books"
    assert books[0]["title"] == "A Light in the ..."

    os.remove("books.json")

    defer.returnValue(None)

当启用"TWISTED_REACTOR": "twisted.internet.asyncioreactor.AsyncioSelectorReactor"时,出现如下错误:

Exception: The installed reactor (twisted.internet.selectreactor.SelectReactor) does not match the requested one (twisted.internet.asyncioreactor.AsyncioSelectorReactor)

而启用"TWISTED_REACTOR": "twisted.internet.selectreactor.SelectReactor"时测试可正常通过。


现象解释

这个错误的核心是Twisted的Reactor只能初始化一次,且初始化后无法动态切换。

使用pytest-twisted运行测试时,它会在测试启动前自动初始化默认的SelectReactor。当你在CrawlerRunner的配置里指定AsyncioSelectorReactor时,Scrapy会检查当前已安装的Reactor是否与指定的一致,发现不匹配就抛出错误。反之,当指定的Reactor和pytest-twisted默认初始化的一致时,就不会有冲突,测试正常运行。

用pytest测试CrawlerRunner/CrawlerProcess的正确方法

方法1:提前指定Reactor(推荐)

在任何Reactor相关代码执行前,提前初始化目标Reactor,避免后续冲突。可以在测试文件顶部或者conftest.py中添加:

# 在测试文件最顶部或者conftest.py中添加
from twisted.internet import asyncioreactor

# 提前初始化AsyncioSelectorReactor
asyncioreactor.install()

这样后续CrawlerRunner使用指定的AsyncioSelectorReactor时,就不会出现不匹配问题。

方法2:用Scrapy官方测试工具做单元测试

Scrapy提供了更适合测试的工具,无需启动完整Reactor或生成临时文件,直接模拟请求和解析逻辑:

from scrapy.http import HtmlResponse
from bookspider import BookSpider

def test_bookspider_parse():
    # 构造爬虫实例
    spider = BookSpider(name="books")
    # 模拟响应内容
    sample_html = """
    <article class="product_pod">
        <h3><a href="#" title="A Light in the ...">A Light in the ...</a></h3>
    </article>
    """
    response = HtmlResponse(url="https://books.toscrape.com/", body=sample_html.encode())
    # 调用解析方法获取结果
    result = list(spider.parse(response))
    
    assert len(result) == 1
    assert result[0]["name"] == "books"
    assert result[0]["title"] == "A Light in the ..."

这种方式测试速度更快,无额外文件生成,适合单元测试场景。

方法3:用pytest fixture做Reactor隔离

如果需要在多个测试中使用不同Reactor,可借助pytest fixture结合进程隔离实现:

import pytest
from twisted.internet import asyncioreactor

@pytest.fixture(scope="function")
def asyncio_reactor():
    # 提前安装AsyncioSelectorReactor
    asyncioreactor.install()
    yield

注意:Twisted本身不支持卸载Reactor,若要彻底隔离环境,推荐使用pytest-xdist的进程隔离模式(添加--numprocesses参数),避免不同测试间的Reactor冲突。


内容的提问来源于stack exchange,提问作者Henry Dashwood

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.23 01:10:56