如何让Scrapy与Pytest适配AsyncioSelectorReactor?
问题复现环境
- 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

