mirror of https://github.com/scrapy/scrapy.git
143 lines
4.4 KiB
Python
143 lines
4.4 KiB
Python
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,
|
|
},
|
|
}
|