import itertools from typing import Any from unittest.mock import patch from scrapy import Spider from scrapy.crawler import AsyncCrawlerRunner, Crawler, CrawlerRunner from scrapy.exceptions import NotConfigured from scrapy.settings import BaseSettings, Settings from scrapy.utils.test import get_crawler, get_reactor_settings from tests.utils.decorators import inline_callbacks_test class SimpleAddon: def update_settings(self, settings): pass def get_addon_cls(config: dict[str, Any]) -> type: class AddonWithConfig: def update_settings(self, settings: BaseSettings): settings.update(config, priority="addon") return AddonWithConfig class CreateInstanceAddon: def __init__(self, crawler: Crawler) -> None: super().__init__() self.crawler = crawler self.config = crawler.settings.getdict("MYADDON") @classmethod def from_crawler(cls, crawler: Crawler): return cls(crawler) def update_settings(self, settings): settings.update(self.config, "addon") class TestAddon: def test_update_settings(self): settings = BaseSettings() settings.set("KEY1", "default", priority="default") settings.set("KEY2", "project", priority="project") addon_config = {"KEY1": "addon", "KEY2": "addon", "KEY3": "addon"} testaddon = get_addon_cls(addon_config)() testaddon.update_settings(settings) assert settings["KEY1"] == "addon" assert settings["KEY2"] == "project" assert settings["KEY3"] == "addon" class TestAddonManager: def test_load_settings(self): settings_dict = { "ADDONS": {"tests.test_addons.SimpleAddon": 0}, } crawler = get_crawler(settings_dict=settings_dict) manager = crawler.addons assert isinstance(manager.addons[0], SimpleAddon) def test_notconfigured(self): class NotConfiguredAddon: def update_settings(self, settings): raise NotConfigured settings_dict = { "ADDONS": {NotConfiguredAddon: 0}, } crawler = get_crawler(settings_dict=settings_dict) manager = crawler.addons assert not manager.addons def test_notconfigured_with_args(self): class NotConfiguredAddon: def update_settings(self, settings): raise NotConfigured("addon disabled reason") settings_dict = { "ADDONS": {NotConfiguredAddon: 0}, } with patch("scrapy.addons.logger") as logger_mock: crawler = get_crawler(settings_dict=settings_dict) assert not crawler.addons.addons logger_mock.warning.assert_called_once_with( "Disabled %(clspath)s: %(eargs)s", {"clspath": NotConfiguredAddon, "eargs": "addon disabled reason"}, extra={"crawler": crawler}, ) def test_no_update_settings(self): class PreCrawlerOnlyAddon: @classmethod def update_pre_crawler_settings(cls, settings): settings.set("PRE_CRAWLER_KEY", "value", priority="addon") settings_dict = { "ADDONS": {PreCrawlerOnlyAddon: 0}, } crawler = get_crawler(settings_dict=settings_dict) manager = crawler.addons assert len(manager.addons) == 1 assert isinstance(manager.addons[0], PreCrawlerOnlyAddon) def test_load_settings_order(self): # Get three addons with different settings addonlist = [] for i in range(3): addon = get_addon_cls({"KEY1": i}) addon.number = i # type: ignore[attr-defined] addonlist.append(addon) # Test for every possible ordering for ordered_addons in itertools.permutations(addonlist): expected_order = [a.number for a in ordered_addons] # type: ignore[attr-defined] settings = {"ADDONS": {a: i for i, a in enumerate(ordered_addons)}} crawler = get_crawler(settings_dict=settings) manager = crawler.addons assert [a.number for a in manager.addons] == expected_order assert crawler.settings.getint("KEY1") == expected_order[-1] def test_build_from_crawler(self): settings_dict = { "ADDONS": {"tests.test_addons.CreateInstanceAddon": 0}, "MYADDON": {"MYADDON_KEY": "val"}, } crawler = get_crawler(settings_dict=settings_dict) manager = crawler.addons assert isinstance(manager.addons[0], CreateInstanceAddon) assert crawler.settings.get("MYADDON_KEY") == "val" def test_settings_priority(self): config = { "KEY": 15, # priority=addon } settings_dict = { "ADDONS": {get_addon_cls(config): 1}, **get_reactor_settings(), } crawler = get_crawler(settings_dict=settings_dict) assert crawler.settings.getint("KEY") == 15 runner_cls = ( CrawlerRunner if settings_dict.get("TWISTED_REACTOR_ENABLED", True) else AsyncCrawlerRunner ) settings = Settings(settings_dict) settings.set("KEY", 0, priority="default") runner = runner_cls(settings) crawler = runner.create_crawler(Spider) crawler._apply_settings() assert crawler.settings.getint("KEY") == 15 settings_dict = { "KEY": 20, # priority=project "ADDONS": {get_addon_cls(config): 1}, **get_reactor_settings(), } settings = Settings(settings_dict) settings.set("KEY", 0, priority="default") runner = runner_cls(settings) crawler = runner.create_crawler(Spider) crawler._apply_settings() assert crawler.settings.getint("KEY") == 20 def test_fallback_workflow(self): FALLBACK_SETTING = "MY_FALLBACK_SCHEDULER" class AddonWithFallback: def update_settings(self, settings): if not settings.get(FALLBACK_SETTING): settings.set( FALLBACK_SETTING, settings.get("SCHEDULER"), "addon", ) settings["SCHEDULER"] = "AddonScheduler" settings_dict: dict[str, Any] = { "ADDONS": {AddonWithFallback: 1}, } crawler = get_crawler(settings_dict=settings_dict) assert crawler.settings.get("SCHEDULER") == "AddonScheduler" assert ( crawler.settings.get(FALLBACK_SETTING) == "scrapy.core.scheduler.Scheduler" ) settings_dict = { "ADDONS": {AddonWithFallback: 1}, "SCHEDULER": "UserScheduler", } crawler = get_crawler(settings_dict=settings_dict) assert crawler.settings.get("SCHEDULER") == "AddonScheduler" assert crawler.settings.get(FALLBACK_SETTING) == "UserScheduler" def test_logging_message(self): class LoggedAddon: def update_settings(self, settings): pass with ( patch("scrapy.addons.logger") as logger_mock, patch("scrapy.addons.build_from_crawler") as build_from_crawler_mock, ): settings_dict = { "ADDONS": {LoggedAddon: 1}, } addon = LoggedAddon() build_from_crawler_mock.return_value = addon crawler = get_crawler(settings_dict=settings_dict) logger_mock.info.assert_called_once_with( "Enabled addons:\n%(addons)s", {"addons": [addon]}, extra={"crawler": crawler}, ) @inline_callbacks_test def test_enable_addon_in_spider(self): class MySpider(Spider): name = "myspider" @classmethod def from_crawler(cls, crawler, *args, **kwargs): spider = super().from_crawler(crawler, *args, **kwargs) addon_config = {"KEY": "addon"} addon_cls = get_addon_cls(addon_config) spider.settings.set("ADDONS", {addon_cls: 1}, priority="spider") return spider settings = Settings() settings.setdict(get_reactor_settings()) settings.set("KEY", "default", priority="default") runner_cls = ( CrawlerRunner if settings.getbool("TWISTED_REACTOR_ENABLED", True) else AsyncCrawlerRunner ) runner = runner_cls(settings) crawler = runner.create_crawler(MySpider) assert crawler.settings.get("KEY") == "default" yield crawler.crawl() assert crawler.settings.get("KEY") == "addon"