mirror of https://github.com/scrapy/scrapy.git
244 lines
8.5 KiB
Python
244 lines
8.5 KiB
Python
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"
|