mirror of https://github.com/scrapy/scrapy.git
554 lines
19 KiB
Python
554 lines
19 KiB
Python
import asyncio
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from twisted.internet.defer import Deferred, fail, succeed
|
|
|
|
from scrapy import Request, Spider, signals
|
|
from scrapy.crawler import Crawler
|
|
from scrapy.exceptions import ScrapyDeprecationWarning
|
|
from scrapy.pipelines import ItemPipelineManager
|
|
from scrapy.utils.asyncio import call_later
|
|
from scrapy.utils.conf import build_component_list
|
|
from scrapy.utils.defer import deferred_to_future, maybe_deferred_to_future
|
|
from scrapy.utils.spider import DefaultSpider
|
|
from scrapy.utils.test import get_crawler, get_from_asyncio_queue
|
|
from tests.mockserver.http import MockServer
|
|
from tests.utils.decorators import coroutine_test
|
|
|
|
|
|
class SimplePipeline:
|
|
def process_item(self, item):
|
|
item["pipeline_passed"] = True
|
|
return item
|
|
|
|
|
|
class DeprecatedSpiderArgPipeline:
|
|
def open_spider(self, spider):
|
|
pass
|
|
|
|
def close_spider(self, spider):
|
|
pass
|
|
|
|
def process_item(self, item, spider):
|
|
item["pipeline_passed"] = True
|
|
return item
|
|
|
|
|
|
class DeferredPipeline:
|
|
def cb(self, item):
|
|
item["pipeline_passed"] = True
|
|
return item
|
|
|
|
def open_spider(self):
|
|
return succeed(None)
|
|
|
|
def close_spider(self):
|
|
return succeed(None)
|
|
|
|
def process_item(self, item):
|
|
d = Deferred()
|
|
d.addCallback(self.cb)
|
|
d.callback(item)
|
|
return d
|
|
|
|
|
|
class AsyncDefPipeline:
|
|
async def process_item(self, item):
|
|
d = Deferred()
|
|
call_later(0, d.callback, None)
|
|
await maybe_deferred_to_future(d)
|
|
item["pipeline_passed"] = True
|
|
return item
|
|
|
|
|
|
class AsyncDefAsyncioPipeline:
|
|
async def process_item(self, item):
|
|
d = Deferred()
|
|
loop = asyncio.get_event_loop()
|
|
loop.call_later(0, d.callback, None)
|
|
await deferred_to_future(d)
|
|
await asyncio.sleep(0.2)
|
|
item["pipeline_passed"] = await get_from_asyncio_queue(True)
|
|
return item
|
|
|
|
|
|
class AsyncDefNotAsyncioPipeline:
|
|
async def process_item(self, item):
|
|
d1 = Deferred()
|
|
from twisted.internet import reactor
|
|
|
|
reactor.callLater(0, d1.callback, None)
|
|
await d1
|
|
d2 = Deferred()
|
|
reactor.callLater(0, d2.callback, None)
|
|
await maybe_deferred_to_future(d2)
|
|
item["pipeline_passed"] = True
|
|
return item
|
|
|
|
|
|
class ProcessItemExceptionPipeline:
|
|
def process_item(self, item):
|
|
raise ValueError("process_item error")
|
|
|
|
|
|
class ProcessItemExceptionDeferredPipeline:
|
|
def process_item(self, item):
|
|
return fail(ValueError("process_item error"))
|
|
|
|
|
|
class ProcessItemExceptionAsyncPipeline:
|
|
async def process_item(self, item):
|
|
raise ValueError("process_item error")
|
|
|
|
|
|
class OpenSpiderExceptionPipeline:
|
|
def open_spider(self):
|
|
raise ValueError("open_spider error")
|
|
|
|
|
|
class OpenSpiderExceptionDeferredPipeline:
|
|
def open_spider(self):
|
|
return fail(ValueError("open_spider error"))
|
|
|
|
|
|
class OpenSpiderExceptionAsyncPipeline:
|
|
async def open_spider(self):
|
|
raise ValueError("open_spider error")
|
|
|
|
|
|
class ItemSpider(Spider):
|
|
name = "itemspider"
|
|
|
|
async def start(self):
|
|
yield Request(self.mockserver.url("/status?n=200"))
|
|
|
|
def parse(self, response):
|
|
return {"field": 42}
|
|
|
|
|
|
class TestPipeline:
|
|
def _on_item_scraped(self, item):
|
|
assert isinstance(item, dict)
|
|
assert item.get("pipeline_passed")
|
|
self.items.append(item)
|
|
|
|
def _create_crawler(self, pipeline_class: type) -> Crawler:
|
|
settings = {
|
|
"ITEM_PIPELINES": {pipeline_class: 1},
|
|
}
|
|
crawler = get_crawler(ItemSpider, settings)
|
|
crawler.signals.connect(self._on_item_scraped, signals.item_scraped)
|
|
self.items: list[Any] = []
|
|
return crawler
|
|
|
|
@pytest.mark.parametrize(
|
|
"pipeline_class",
|
|
[
|
|
SimplePipeline,
|
|
AsyncDefPipeline,
|
|
pytest.param(AsyncDefAsyncioPipeline, marks=pytest.mark.only_asyncio),
|
|
pytest.param(
|
|
AsyncDefNotAsyncioPipeline, marks=pytest.mark.only_not_asyncio
|
|
),
|
|
],
|
|
)
|
|
@coroutine_test
|
|
async def test_pipeline(self, mockserver: MockServer, pipeline_class: type) -> None:
|
|
crawler = self._create_crawler(pipeline_class)
|
|
await crawler.crawl_async(mockserver=mockserver)
|
|
assert len(self.items) == 1
|
|
|
|
@coroutine_test
|
|
async def test_pipeline_deferred(self, mockserver: MockServer) -> None:
|
|
crawler = self._create_crawler(DeferredPipeline)
|
|
with (
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match="DeferredPipeline.open_spider returned a Deferred",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match="DeferredPipeline.close_spider returned a Deferred",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match="DeferredPipeline.process_item returned a Deferred",
|
|
),
|
|
):
|
|
await crawler.crawl_async(mockserver=mockserver)
|
|
assert len(self.items) == 1
|
|
|
|
@coroutine_test
|
|
async def test_deprecated_spider_arg(self, mockserver: MockServer) -> None:
|
|
crawler = self._create_crawler(DeprecatedSpiderArgPipeline)
|
|
with (
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"DeprecatedSpiderArgPipeline.open_spider\(\) requires a spider argument",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"DeprecatedSpiderArgPipeline.close_spider\(\) requires a spider argument",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"DeprecatedSpiderArgPipeline.process_item\(\) requires a spider argument",
|
|
),
|
|
):
|
|
await crawler.crawl_async(mockserver=mockserver)
|
|
|
|
assert len(self.items) == 1
|
|
|
|
@pytest.mark.parametrize(
|
|
"pipeline_class",
|
|
[
|
|
ProcessItemExceptionPipeline,
|
|
pytest.param(
|
|
ProcessItemExceptionDeferredPipeline,
|
|
marks=pytest.mark.filterwarnings(
|
|
"ignore::scrapy.exceptions.ScrapyDeprecationWarning"
|
|
),
|
|
),
|
|
ProcessItemExceptionAsyncPipeline,
|
|
],
|
|
)
|
|
@coroutine_test
|
|
async def test_process_item_exception(
|
|
self,
|
|
caplog: pytest.LogCaptureFixture,
|
|
mockserver: MockServer,
|
|
pipeline_class: type,
|
|
) -> None:
|
|
crawler = self._create_crawler(pipeline_class)
|
|
await crawler.crawl_async(mockserver=mockserver)
|
|
assert "Error processing {'field': 42}" in caplog.text
|
|
assert "process_item error" in caplog.text
|
|
|
|
@pytest.mark.parametrize(
|
|
"pipeline_class",
|
|
[
|
|
OpenSpiderExceptionPipeline,
|
|
pytest.param(
|
|
OpenSpiderExceptionDeferredPipeline,
|
|
marks=pytest.mark.filterwarnings(
|
|
"ignore::scrapy.exceptions.ScrapyDeprecationWarning"
|
|
),
|
|
),
|
|
OpenSpiderExceptionAsyncPipeline,
|
|
],
|
|
)
|
|
@coroutine_test
|
|
async def test_open_spider_exception(
|
|
self, mockserver: MockServer, pipeline_class: type
|
|
) -> None:
|
|
crawler = self._create_crawler(pipeline_class)
|
|
with pytest.raises(ValueError, match="open_spider error"):
|
|
await crawler.crawl_async(mockserver=mockserver)
|
|
|
|
|
|
class TestCustomPipelineManager:
|
|
@coroutine_test
|
|
async def test_deprecated_process_item_spider_arg(self) -> None:
|
|
class CustomPipelineManager(ItemPipelineManager):
|
|
def process_item(self, item: Any, spider: Spider) -> Deferred[Any]: # pylint: disable=useless-parent-delegation
|
|
return super().process_item(item, spider)
|
|
|
|
crawler = get_crawler(DefaultSpider)
|
|
crawler.spider = crawler._create_spider()
|
|
itemproc = CustomPipelineManager.from_crawler(crawler)
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager.process_item\(\) is deprecated, use process_item_async\(\)",
|
|
):
|
|
itemproc.process_item({}, crawler.spider)
|
|
|
|
@coroutine_test
|
|
async def test_integration_recommended(self, mockserver: MockServer) -> None:
|
|
class CustomPipelineManager(ItemPipelineManager):
|
|
async def process_item_async(self, item):
|
|
return await super().process_item_async(item)
|
|
|
|
items = []
|
|
|
|
def _on_item_scraped(item):
|
|
assert isinstance(item, dict)
|
|
assert item.get("pipeline_passed")
|
|
items.append(item)
|
|
|
|
crawler = get_crawler(
|
|
ItemSpider,
|
|
{
|
|
"ITEM_PROCESSOR": CustomPipelineManager,
|
|
"ITEM_PIPELINES": {SimplePipeline: 1},
|
|
},
|
|
)
|
|
crawler.spider = crawler._create_spider()
|
|
crawler.signals.connect(_on_item_scraped, signals.item_scraped)
|
|
await crawler.crawl_async(mockserver=mockserver)
|
|
|
|
assert len(items) == 1
|
|
|
|
@coroutine_test
|
|
async def test_integration_no_async_subclass(self, mockserver: MockServer) -> None:
|
|
class CustomPipelineManager(ItemPipelineManager):
|
|
def open_spider(self, spider):
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager.open_spider\(\) is deprecated, use open_spider_async\(\)",
|
|
):
|
|
return super().open_spider(spider)
|
|
|
|
def close_spider(self, spider):
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager.close_spider\(\) is deprecated, use close_spider_async\(\)",
|
|
):
|
|
return super().close_spider(spider)
|
|
|
|
def process_item(self, item, spider):
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager.process_item\(\) is deprecated, use process_item_async\(\)",
|
|
):
|
|
return super().process_item(item, spider)
|
|
|
|
items = []
|
|
|
|
def _on_item_scraped(item):
|
|
assert isinstance(item, dict)
|
|
assert item.get("pipeline_passed")
|
|
items.append(item)
|
|
|
|
crawler = get_crawler(
|
|
ItemSpider,
|
|
{
|
|
"ITEM_PROCESSOR": CustomPipelineManager,
|
|
"ITEM_PIPELINES": {SimplePipeline: 1},
|
|
},
|
|
)
|
|
crawler.spider = crawler._create_spider()
|
|
crawler.signals.connect(_on_item_scraped, signals.item_scraped)
|
|
with (
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager overrides open_spider\(\) but doesn't override open_spider_async\(\)",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager overrides close_spider\(\) but doesn't override close_spider_async\(\)",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager overrides process_item\(\) but doesn't override process_item_async\(\)",
|
|
),
|
|
):
|
|
await crawler.crawl_async(mockserver=mockserver)
|
|
|
|
assert len(items) == 1
|
|
|
|
@coroutine_test
|
|
async def test_integration_no_async_not_subclass(
|
|
self, mockserver: MockServer
|
|
) -> None:
|
|
class CustomPipelineManager:
|
|
def __init__(self, crawler):
|
|
self.pipelines = [
|
|
p()
|
|
for p in build_component_list(
|
|
crawler.settings.getwithbase("ITEM_PIPELINES")
|
|
)
|
|
]
|
|
|
|
@classmethod
|
|
def from_crawler(cls, crawler):
|
|
return cls(crawler)
|
|
|
|
def open_spider(self, spider):
|
|
return succeed(None)
|
|
|
|
def close_spider(self, spider):
|
|
return succeed(None)
|
|
|
|
def process_item(self, item, spider):
|
|
for pipeline in self.pipelines:
|
|
item = pipeline.process_item(item)
|
|
return succeed(item)
|
|
|
|
items = []
|
|
|
|
def _on_item_scraped(item):
|
|
assert isinstance(item, dict)
|
|
assert item.get("pipeline_passed")
|
|
items.append(item)
|
|
|
|
crawler = get_crawler(
|
|
ItemSpider,
|
|
{
|
|
"ITEM_PROCESSOR": CustomPipelineManager,
|
|
"ITEM_PIPELINES": {SimplePipeline: 1},
|
|
},
|
|
)
|
|
crawler.spider = crawler._create_spider()
|
|
crawler.signals.connect(_on_item_scraped, signals.item_scraped)
|
|
with (
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager doesn't define a open_spider_async\(\) method",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager doesn't define a close_spider_async\(\) method",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"CustomPipelineManager doesn't define a process_item_async\(\) method",
|
|
),
|
|
):
|
|
await crawler.crawl_async(mockserver=mockserver)
|
|
|
|
assert len(items) == 1
|
|
|
|
|
|
class TestMiddlewareManagerSpider:
|
|
"""Tests for the deprecated spider arg handling in MiddlewareManager.
|
|
|
|
Here because MiddlewareManager doesn't have methods that could take a spider arg."""
|
|
|
|
@pytest.fixture
|
|
def crawler(self) -> Crawler:
|
|
return get_crawler(Spider)
|
|
|
|
@coroutine_test
|
|
async def test_deprecated_spider_arg_no_crawler_spider(
|
|
self, crawler: Crawler
|
|
) -> None:
|
|
"""Crawler is provided, but doesn't have a spider, the methods raise an exception.
|
|
The instance passed to a deprecated method is ignored."""
|
|
mwman = ItemPipelineManager(crawler=crawler)
|
|
with (
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"DeprecatedSpiderArgPipeline.open_spider\(\) requires a spider argument",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"DeprecatedSpiderArgPipeline.close_spider\(\) requires a spider argument",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"DeprecatedSpiderArgPipeline.process_item\(\) requires a spider argument",
|
|
),
|
|
):
|
|
mwman._add_middleware(DeprecatedSpiderArgPipeline())
|
|
with (
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"ItemPipelineManager.open_spider\(\) is deprecated, use open_spider_async\(\) instead",
|
|
),
|
|
pytest.raises(
|
|
ValueError,
|
|
match=r"ItemPipelineManager needs to access self\.crawler\.spider but it is None",
|
|
),
|
|
):
|
|
await maybe_deferred_to_future(mwman.open_spider(DefaultSpider()))
|
|
with pytest.raises(
|
|
ValueError,
|
|
match=r"ItemPipelineManager needs to access self\.crawler\.spider but it is None",
|
|
):
|
|
await mwman.open_spider_async()
|
|
with (
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"ItemPipelineManager.close_spider\(\) is deprecated, use close_spider_async\(\) instead",
|
|
),
|
|
pytest.raises(
|
|
ValueError,
|
|
match=r"ItemPipelineManager needs to access self\.crawler\.spider but it is None",
|
|
),
|
|
):
|
|
await maybe_deferred_to_future(mwman.close_spider(DefaultSpider()))
|
|
with pytest.raises(
|
|
ValueError,
|
|
match=r"ItemPipelineManager needs to access self\.crawler\.spider but it is None",
|
|
):
|
|
await mwman.close_spider_async()
|
|
|
|
@coroutine_test
|
|
async def test_deprecated_spider_arg_with_crawler(self, crawler: Crawler) -> None:
|
|
"""Crawler is provided and has a spider, works. The instance passed to a deprecated method
|
|
is ignored, even if mismatched."""
|
|
mwman = ItemPipelineManager(crawler=crawler)
|
|
crawler.spider = crawler._create_spider("foo")
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"ItemPipelineManager.open_spider\(\) is deprecated, use open_spider_async\(\) instead",
|
|
):
|
|
await maybe_deferred_to_future(mwman.open_spider(DefaultSpider()))
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"ItemPipelineManager.close_spider\(\) is deprecated, use close_spider_async\(\) instead",
|
|
):
|
|
await maybe_deferred_to_future(mwman.close_spider(DefaultSpider()))
|
|
|
|
@coroutine_test
|
|
async def test_deprecated_spider_arg_without_crawler(self) -> None:
|
|
"""The first instance passed to a deprecated method is used. Mismatched ones raise an error."""
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match="was called without the crawler argument",
|
|
):
|
|
mwman = ItemPipelineManager()
|
|
spider = DefaultSpider()
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"ItemPipelineManager.open_spider\(\) is deprecated, use open_spider_async\(\) instead",
|
|
):
|
|
await maybe_deferred_to_future(mwman.open_spider(spider))
|
|
with (
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"ItemPipelineManager.close_spider\(\) is deprecated, use close_spider_async\(\) instead",
|
|
),
|
|
pytest.raises(
|
|
RuntimeError, match="Different instances of Spider were passed"
|
|
),
|
|
):
|
|
await maybe_deferred_to_future(mwman.close_spider(DefaultSpider()))
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"ItemPipelineManager.close_spider\(\) is deprecated, use close_spider_async\(\) instead",
|
|
):
|
|
await maybe_deferred_to_future(mwman.close_spider(spider))
|
|
|
|
@coroutine_test
|
|
async def test_no_spider_arg_without_crawler(self) -> None:
|
|
"""If no crawler and no spider arg, raise an error."""
|
|
with pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match="was called without the crawler argument",
|
|
):
|
|
mwman = ItemPipelineManager()
|
|
with (
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"DeprecatedSpiderArgPipeline.open_spider\(\) requires a spider argument",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"DeprecatedSpiderArgPipeline.close_spider\(\) requires a spider argument",
|
|
),
|
|
pytest.warns(
|
|
ScrapyDeprecationWarning,
|
|
match=r"DeprecatedSpiderArgPipeline.process_item\(\) requires a spider argument",
|
|
),
|
|
):
|
|
mwman._add_middleware(DeprecatedSpiderArgPipeline())
|
|
with (
|
|
pytest.raises(
|
|
ValueError,
|
|
match="has no known Spider instance",
|
|
),
|
|
):
|
|
await mwman.open_spider_async()
|