mirror of https://github.com/scrapy/scrapy.git
140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pytest
|
|
|
|
from scrapy.http import Request, Response
|
|
from scrapy.spidermiddlewares.depth import DepthMiddleware
|
|
from scrapy.spiders import Spider
|
|
from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
|
|
from scrapy.utils.misc import build_from_crawler
|
|
from scrapy.utils.test import get_crawler
|
|
from tests.utils.decorators import coroutine_test
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Generator
|
|
|
|
from scrapy.crawler import Crawler
|
|
from scrapy.statscollectors import StatsCollector
|
|
|
|
|
|
@pytest.fixture
|
|
def crawler() -> Crawler:
|
|
return get_crawler(Spider, {"DEPTH_LIMIT": 1, "DEPTH_STATS_VERBOSE": True})
|
|
|
|
|
|
@pytest.fixture
|
|
def stats(crawler: Crawler) -> Generator[StatsCollector]:
|
|
assert crawler.stats is not None
|
|
crawler.stats.open_spider()
|
|
|
|
yield crawler.stats
|
|
|
|
crawler.stats.close_spider()
|
|
|
|
|
|
@pytest.fixture
|
|
def mw(crawler: Crawler) -> DepthMiddleware:
|
|
return DepthMiddleware.from_crawler(crawler)
|
|
|
|
|
|
def test_process_spider_output(mw: DepthMiddleware, stats: StatsCollector) -> None:
|
|
req = Request("http://scrapytest.org")
|
|
resp = Response("http://scrapytest.org")
|
|
resp.request = req
|
|
result = [Request("http://scrapytest.org")]
|
|
|
|
out = list(mw.process_spider_output(resp, result))
|
|
assert out == result
|
|
|
|
rdc = stats.get_value("request_depth_count/1")
|
|
assert rdc == 1
|
|
|
|
req.meta["depth"] = 1
|
|
|
|
out2 = list(mw.process_spider_output(resp, result))
|
|
assert not out2
|
|
|
|
rdm = stats.get_value("request_depth_max")
|
|
assert rdm == 1
|
|
|
|
|
|
def test_depth_reset(mw: DepthMiddleware, stats: StatsCollector) -> None:
|
|
resp = Response("https://example.com")
|
|
resp.request = Request("https://example.com", meta={"depth": 5})
|
|
result = [Request("https://example.com", meta={"depth_reset": True})]
|
|
|
|
out = list(mw.process_spider_output(resp, result))
|
|
|
|
assert out == result
|
|
assert out[0].meta["depth"] == 0
|
|
assert "depth_reset" not in out[0].meta
|
|
assert stats.get_value("request_depth_count/0") == 1
|
|
|
|
|
|
def test_process_spider_output_no_response(
|
|
mw: DepthMiddleware, stats: StatsCollector
|
|
) -> None:
|
|
result = [Request("https://example.com")]
|
|
|
|
out = list(mw.process_spider_output(None, result))
|
|
assert out == result
|
|
assert "depth" not in out[0].meta
|
|
assert stats.get_value("request_depth_count/0") is None
|
|
|
|
|
|
@coroutine_test
|
|
async def test_process_spider_output_async_no_response(
|
|
mw: DepthMiddleware, stats: StatsCollector
|
|
) -> None:
|
|
result = [Request("https://example.com")]
|
|
|
|
out = await collect_asyncgen(
|
|
mw.process_spider_output_async(None, as_async_generator(result))
|
|
)
|
|
assert out == result
|
|
assert "depth" not in out[0].meta
|
|
assert stats.get_value("request_depth_count/0") is None
|
|
|
|
|
|
def test_ignored_logged_once(
|
|
mw: DepthMiddleware, stats: StatsCollector, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
resp = Response("http://example.com")
|
|
resp.request = Request("http://example.com")
|
|
resp.meta["depth"] = 1
|
|
result = [Request(f"http://example.com/{i}") for i in range(3)]
|
|
|
|
with caplog.at_level("DEBUG", logger="scrapy.spidermiddlewares.depth"):
|
|
assert not list(mw.process_spider_output(resp, result))
|
|
|
|
messages = [r.getMessage() for r in caplog.records]
|
|
assert len(messages) == 1
|
|
assert "http://example.com/0" in messages[0]
|
|
assert stats.get_value("depth/request_ignored_count") == 3
|
|
|
|
|
|
def test_priority_and_non_verbose_stats() -> None:
|
|
crawler = get_crawler(
|
|
Spider,
|
|
{"DEPTH_LIMIT": 0, "DEPTH_STATS_VERBOSE": False, "DEPTH_PRIORITY": 10},
|
|
)
|
|
assert crawler.stats is not None
|
|
crawler.stats.open_spider()
|
|
try:
|
|
mw = build_from_crawler(DepthMiddleware, crawler)
|
|
resp = Response("http://toscrape.com")
|
|
resp.request = Request("http://toscrape.com")
|
|
resp.request.meta["depth"] = 2
|
|
out = list(mw.process_spider_output(resp, [Request("http://toscrape.com")]))
|
|
assert len(out) == 1
|
|
# priority is decremented by depth * DEPTH_PRIORITY
|
|
assert out[0].priority == -30
|
|
assert out[0].meta["depth"] == 3
|
|
# non-verbose stats don't track per-depth counts but still track the max
|
|
assert crawler.stats.get_value("request_depth_count/3") is None
|
|
assert crawler.stats.get_value("request_depth_max") == 3
|
|
finally:
|
|
crawler.stats.close_spider()
|