mirror of https://github.com/scrapy/scrapy.git
58 lines
1.4 KiB
Python
58 lines
1.4 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.test import get_crawler
|
|
|
|
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
|