Scrapy-Pytest执行时Pipeline未运行问题求助

你遇到的问题很典型——本地直接运行爬虫能正常触发Pipeline，但在pytest测试环境下却没执行。下面是几个实用的排查思路：

1. 验证测试环境下的Settings是否正确加载 #

get_project_settings() 依赖项目路径，测试环境下可能因为工作目录或导入路径问题，没有正确读取到你的 ITEM_PIPELINES 配置。可以在测试代码里加个打印，确认配置是否生效：

def test_crawler_execution(crawler):
    # 打印当前加载的Pipeline配置
    print("Loaded Item Pipelines:", crawler.settings.get('ITEM_PIPELINES'))
    crawler.crawl(SubmissionsSpider, n=2)
    crawler.start()

如果输出里没有 crawler.pipelines.MongoWriterPipeline，说明配置没加载对。这时候可以手动指定settings路径，或者在fixture里显式设置：

@pytest.fixture
def crawler():
    from scrapy.settings import Settings
    settings = Settings()
    # 手动加载项目settings
    settings.setmodule('crawler.settings')
    return CrawlerProcess(settings)

2. 确保测试等待异步任务完成 #

Scrapy基于Twisted异步框架，crawler.start() 启动后会异步执行，但pytest可能在爬虫和Pipeline还没处理完就结束了进程，导致Pipeline看起来没执行。可以改用 CrawlerRunner 替代 CrawlerProcess，通过Twisted的Deferred等待任务完成：

from scrapy.crawler import CrawlerRunner
from twisted.internet import reactor

@pytest.fixture
def crawler_runner():
    return CrawlerRunner(get_project_settings())

def test_crawler_execution(crawler_runner):
    # 启动爬虫并获取Deferred对象
    d = crawler_runner.crawl(SubmissionsSpider, n=2)
    # 爬虫完成后停止reactor
    d.addCallback(lambda _: reactor.stop())
    # 启动reactor直到停止
    reactor.run()
    # 这里可以添加断言，比如验证Pipeline处理的数据是否写入数据库

3. 检查Pipeline的初始化与异常情况 #

你的 MongoWriterPipeline 可能在测试环境下初始化失败（比如MongoDB连接失败），Scrapy默认会静默跳过异常的Pipeline。可以在Pipeline的 open_spider 方法里添加日志或抛出异常，排查初始化问题：

# 在pipelines.py里修改
import logging

class MongoWriterPipeline:
    def open_spider(self, spider):
        try:
            # 你的MongoDB连接代码
            self.client = pymongo.MongoClient(...)
            self.db = self.client['test_db']
            logging.info("MongoDB连接成功")
        except Exception as e:
            logging.error(f"Pipeline初始化失败: {str(e)}")
            # 抛出异常让测试能捕获到
            raise e

然后在测试里运行，查看日志或异常信息，确认Pipeline是否正常启动。

4. 使用Scrapy官方测试工具 #

Scrapy提供了专门的测试工具，能更好地适配测试场景。比如用测试专用的Settings和CrawlerProcess：

from scrapy.test import CrawlerProcess
from scrapy.utils.test import get_test_settings

def test_crawler_pipeline():
    # 获取测试用settings并显式设置Pipeline
    settings = get_test_settings()
    settings.set('ITEM_PIPELINES', {'crawler.pipelines.MongoWriterPipeline': 1})
    process = CrawlerProcess(settings)
    
    # 运行爬虫并等待完成
    process.crawl(SubmissionsSpider, n=2)
    process.start()
    
    # 这里添加断言，比如检查数据库中是否有爬取的数据

内容的提问来源于stack exchange，提问作者Gabriel Menezes