from __future__ import annotations import warnings from abc import ABC, abstractmethod import pytest import scrapy from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.extensions.feedexport import FeedExporter from scrapy.utils.test import get_crawler class TestURIParams(ABC): spider_name = "uri_params_spider" deprecated_options = False @abstractmethod def build_settings(self, uri="file:///tmp/foobar", uri_params=None): raise NotImplementedError def _crawler_feed_exporter(self, settings): if self.deprecated_options: with pytest.warns( ScrapyDeprecationWarning, match="The `FEED_URI` and `FEED_FORMAT` settings have been deprecated", ): crawler = get_crawler(settings_dict=settings) else: crawler = get_crawler(settings_dict=settings) feed_exporter = crawler.get_extension(FeedExporter) return crawler, feed_exporter def test_default(self): settings = self.build_settings( uri="file:///tmp/%(name)s", ) crawler, feed_exporter = self._crawler_feed_exporter(settings) spider = scrapy.Spider(self.spider_name) spider.crawler = crawler with warnings.catch_warnings(): warnings.simplefilter("error", ScrapyDeprecationWarning) feed_exporter.open_spider(spider) assert feed_exporter.slots[0].uri == f"file:///tmp/{self.spider_name}" def test_none(self): def uri_params(params, spider): pass settings = self.build_settings( uri="file:///tmp/%(name)s", uri_params=uri_params, ) crawler, feed_exporter = self._crawler_feed_exporter(settings) spider = scrapy.Spider(self.spider_name) spider.crawler = crawler feed_exporter.open_spider(spider) assert feed_exporter.slots[0].uri == f"file:///tmp/{self.spider_name}" def test_empty_dict(self): def uri_params(params, spider): return {} settings = self.build_settings( uri="file:///tmp/%(name)s", uri_params=uri_params, ) crawler, feed_exporter = self._crawler_feed_exporter(settings) spider = scrapy.Spider(self.spider_name) spider.crawler = crawler with warnings.catch_warnings(): warnings.simplefilter("error", ScrapyDeprecationWarning) with pytest.raises(KeyError): feed_exporter.open_spider(spider) def test_params_as_is(self): def uri_params(params, spider): return params settings = self.build_settings( uri="file:///tmp/%(name)s", uri_params=uri_params, ) crawler, feed_exporter = self._crawler_feed_exporter(settings) spider = scrapy.Spider(self.spider_name) spider.crawler = crawler with warnings.catch_warnings(): warnings.simplefilter("error", ScrapyDeprecationWarning) feed_exporter.open_spider(spider) assert feed_exporter.slots[0].uri == f"file:///tmp/{self.spider_name}" def test_custom_param(self): def uri_params(params, spider): return {**params, "foo": self.spider_name} settings = self.build_settings( uri="file:///tmp/%(foo)s", uri_params=uri_params, ) crawler, feed_exporter = self._crawler_feed_exporter(settings) spider = scrapy.Spider(self.spider_name) spider.crawler = crawler with warnings.catch_warnings(): warnings.simplefilter("error", ScrapyDeprecationWarning) feed_exporter.open_spider(spider) assert feed_exporter.slots[0].uri == f"file:///tmp/{self.spider_name}" class TestURIParamsSetting(TestURIParams): deprecated_options = True def build_settings(self, uri="file:///tmp/foobar", uri_params=None): extra_settings = {} if uri_params: extra_settings["FEED_URI_PARAMS"] = uri_params return { "FEED_URI": uri, **extra_settings, } class TestURIParamsFeedOption(TestURIParams): deprecated_options = False def build_settings(self, uri="file:///tmp/foobar", uri_params=None): options = { "format": "jl", } if uri_params: options["uri_params"] = uri_params return { "FEEDS": { uri: options, }, }