"""Tests for the CrawlerEngine class.""" import tempfile from pathlib import Path import anyio import pytest from scrapling.spiders.engine import CrawlerEngine, _dump from scrapling.spiders.request import Request from scrapling.spiders.robotstxt import RobotsTxtManager from scrapling.spiders.session import SessionManager from scrapling.spiders.result import CrawlStats, ItemList from scrapling.spiders.checkpoint import CheckpointData from scrapling.core._types import Any, Dict, Set, AsyncGenerator # --------------------------------------------------------------------------- # Mock helpers # --------------------------------------------------------------------------- class MockResponse: """Minimal Response stand-in.""" def __init__(self, status: int = 200, body: bytes = b"ok", url: str = "https://example.com", encoding: str = "utf-8"): self.status = status self.body = body self.url = url self.encoding = encoding self.request: Any = None self.meta: Dict[str, Any] = {} def __str__(self) -> str: return self.url class MockSession: """Mock session that returns a canned response.""" def __init__(self, name: str = "mock", response: MockResponse | None = None): self.name = name self._is_alive = False self._response = response or MockResponse() self.fetch_calls: list[dict] = [] async def __aenter__(self): self._is_alive = True return self async def __aexit__(self, *args): self._is_alive = False async def fetch(self, url: str, **kwargs): self.fetch_calls.append({"url": url, **kwargs}) resp = MockResponse(status=self._response.status, body=self._response.body, url=url) return resp class ErrorSession(MockSession): """Session that raises on fetch.""" def __init__(self, error: Exception | None = None): super().__init__("error") self._error = error or RuntimeError("fetch failed") async def fetch(self, url: str, **kwargs): raise self._error class MockSpider: """Lightweight spider stub for engine tests.""" def __init__( self, *, concurrent_requests: int = 4, concurrent_requests_per_domain: int = 0, download_delay: float = 0.0, max_blocked_retries: int = 3, allowed_domains: Set[str] | None = None, fp_include_kwargs: bool = False, fp_include_headers: bool = False, fp_keep_fragments: bool = False, is_blocked_fn=None, on_scraped_item_fn=None, retry_blocked_request_fn=None, robots_txt_obey: bool = False, start_urls: list[str] | None = None, ): self.concurrent_requests = concurrent_requests self.concurrent_requests_per_domain = concurrent_requests_per_domain self.download_delay = download_delay self.max_blocked_retries = max_blocked_retries self.allowed_domains = allowed_domains or set() self.fp_include_kwargs = fp_include_kwargs self.fp_include_headers = fp_include_headers self.fp_keep_fragments = fp_keep_fragments self.name = "test_spider" self.robots_txt_obey = robots_txt_obey self.development_mode = False self.development_cache_dir = None self.start_urls = start_urls or [] # Tracking lists self.on_start_calls: list[dict] = [] self.on_close_calls: int = 0 self.on_error_calls: list[tuple[Request, Exception]] = [] self.scraped_items: list[dict] = [] self.blocked_responses: list = [] self.retry_requests: list = [] # Pluggable behaviour self._is_blocked_fn = is_blocked_fn self._on_scraped_item_fn = on_scraped_item_fn self._retry_blocked_request_fn = retry_blocked_request_fn # Log counter stub self._log_counter = _LogCounterStub() async def parse(self, response) -> AsyncGenerator[Dict[str, Any] | Request | None, None]: yield {"url": str(response)} async def on_start(self, resuming: bool = False) -> None: self.on_start_calls.append({"resuming": resuming}) async def on_close(self) -> None: self.on_close_calls += 1 async def on_error(self, request: Request, error: Exception) -> None: self.on_error_calls.append((request, error)) async def on_scraped_item(self, item: Dict[str, Any]) -> Dict[str, Any] | None: if self._on_scraped_item_fn: return self._on_scraped_item_fn(item) self.scraped_items.append(item) return item async def is_blocked(self, response) -> bool: if self._is_blocked_fn: return self._is_blocked_fn(response) return False async def retry_blocked_request(self, request: Request, response) -> Request: self.retry_requests.append(request) if self._retry_blocked_request_fn: return self._retry_blocked_request_fn(request, response) return request async def start_requests(self) -> AsyncGenerator[Request, None]: yield Request("https://example.com", sid="default") class _LogCounterStub: """Stub for LogCounterHandler.""" def get_counts(self) -> Dict[str, int]: return {"debug": 0, "info": 0, "warning": 0, "error": 0, "critical": 0} def _make_engine( spider: MockSpider | None = None, session: MockSession | None = None, crawldir: str | None = None, interval: float = 300.0, ) -> CrawlerEngine: """Create a CrawlerEngine wired to mock objects.""" spider = spider or MockSpider() sm = SessionManager() sm.add("default", session or MockSession()) return CrawlerEngine(spider, sm, crawldir=crawldir, interval=interval) # --------------------------------------------------------------------------- # Tests: _dump helper # --------------------------------------------------------------------------- class TestDumpHelper: def test_dump_returns_json_string(self): result = _dump({"key": "value"}) assert '"key": "value"' in result def test_dump_handles_nested(self): result = _dump({"a": {"b": 1}}) assert '"a"' in result assert '"b"' in result # --------------------------------------------------------------------------- # Tests: __init__ # --------------------------------------------------------------------------- class TestCrawlerEngineInit: def test_default_initialisation(self): engine = _make_engine() assert engine._running is False assert engine._active_tasks == 0 assert engine._pause_requested is False assert engine._force_stop is False assert engine.paused is False assert isinstance(engine.stats, CrawlStats) assert isinstance(engine.items, ItemList) def test_checkpoint_system_disabled_by_default(self): engine = _make_engine() assert engine._checkpoint_system_enabled is False def test_checkpoint_system_enabled_with_crawldir(self): with tempfile.TemporaryDirectory() as tmpdir: engine = _make_engine(crawldir=tmpdir) assert engine._checkpoint_system_enabled is True def test_global_limiter_uses_concurrent_requests(self): spider = MockSpider(concurrent_requests=8) engine = _make_engine(spider=spider) assert engine._global_limiter.total_tokens == 8 def test_allowed_domains_from_spider(self): spider = MockSpider(allowed_domains={"example.com", "test.org"}) engine = _make_engine(spider=spider) assert engine._allowed_domains == {"example.com", "test.org"} # --------------------------------------------------------------------------- # Tests: _is_domain_allowed # --------------------------------------------------------------------------- class TestIsDomainAllowed: def test_all_allowed_when_empty(self): engine = _make_engine() request = Request("https://anything.com/page") assert engine._is_domain_allowed(request) is True def test_exact_domain_match(self): spider = MockSpider(allowed_domains={"example.com"}) engine = _make_engine(spider=spider) assert engine._is_domain_allowed(Request("https://example.com/page")) is True assert engine._is_domain_allowed(Request("https://other.com/page")) is False def test_subdomain_match(self): spider = MockSpider(allowed_domains={"example.com"}) engine = _make_engine(spider=spider) assert engine._is_domain_allowed(Request("https://sub.example.com/page")) is True assert engine._is_domain_allowed(Request("https://deep.sub.example.com/x")) is True def test_partial_name_not_matched(self): spider = MockSpider(allowed_domains={"example.com"}) engine = _make_engine(spider=spider) # "notexample.com" should NOT match "example.com" assert engine._is_domain_allowed(Request("https://notexample.com/x")) is False def test_multiple_allowed_domains(self): spider = MockSpider(allowed_domains={"a.com", "b.org"}) engine = _make_engine(spider=spider) assert engine._is_domain_allowed(Request("https://a.com/")) is True assert engine._is_domain_allowed(Request("https://b.org/")) is True assert engine._is_domain_allowed(Request("https://c.net/")) is False # --------------------------------------------------------------------------- # Tests: _rate_limiter # --------------------------------------------------------------------------- class TestRateLimiter: def test_returns_global_limiter_when_per_domain_disabled(self): engine = _make_engine() # concurrent_requests_per_domain=0 limiter = engine._rate_limiter("example.com") assert limiter is engine._global_limiter def test_returns_per_domain_limiter_when_enabled(self): spider = MockSpider(concurrent_requests_per_domain=2) engine = _make_engine(spider=spider) limiter = engine._rate_limiter("example.com") assert limiter is not engine._global_limiter assert limiter.total_tokens == 2 def test_same_domain_returns_same_limiter(self): spider = MockSpider(concurrent_requests_per_domain=2) engine = _make_engine(spider=spider) l1 = engine._rate_limiter("example.com") l2 = engine._rate_limiter("example.com") assert l1 is l2 def test_different_domains_get_different_limiters(self): spider = MockSpider(concurrent_requests_per_domain=2) engine = _make_engine(spider=spider) l1 = engine._rate_limiter("a.com") l2 = engine._rate_limiter("b.com") assert l1 is not l2 # --------------------------------------------------------------------------- # Tests: _normalize_request # --------------------------------------------------------------------------- class TestNormalizeRequest: def test_sets_default_sid_when_empty(self): engine = _make_engine() request = Request("https://example.com") assert request.sid == "" engine._normalize_request(request) assert request.sid == "default" def test_preserves_existing_sid(self): engine = _make_engine() request = Request("https://example.com", sid="custom") engine._normalize_request(request) assert request.sid == "custom" # --------------------------------------------------------------------------- # Tests: _process_request # --------------------------------------------------------------------------- class TestProcessRequest: @pytest.mark.asyncio async def test_successful_fetch_updates_stats(self): spider = MockSpider() session = MockSession(response=MockResponse(status=200, body=b"hello")) engine = _make_engine(spider=spider, session=session) request = Request("https://example.com", sid="default") await engine._process_request(request) assert engine.stats.requests_count == 1 assert engine.stats.response_bytes == 5 # len(b"hello") from MockSession assert "status_200" in engine.stats.response_status_count @pytest.mark.asyncio async def test_failed_fetch_increments_failed_count(self): spider = MockSpider() sm = SessionManager() sm.add("default", ErrorSession()) engine = CrawlerEngine(spider, sm) request = Request("https://example.com", sid="default") await engine._process_request(request) assert engine.stats.failed_requests_count == 1 assert len(spider.on_error_calls) == 1 @pytest.mark.asyncio async def test_failed_fetch_does_not_increment_requests_count(self): spider = MockSpider() sm = SessionManager() sm.add("default", ErrorSession()) engine = CrawlerEngine(spider, sm) request = Request("https://example.com", sid="default") await engine._process_request(request) assert engine.stats.requests_count == 0 @pytest.mark.asyncio async def test_blocked_response_triggers_retry(self): spider = MockSpider(is_blocked_fn=lambda r: True, max_blocked_retries=2) engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default") await engine._process_request(request) assert engine.stats.blocked_requests_count == 1 # A retry request should be enqueued assert not engine.scheduler.is_empty @pytest.mark.asyncio async def test_blocked_response_max_retries_exceeded(self): spider = MockSpider(is_blocked_fn=lambda r: True, max_blocked_retries=2) engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default") request._retry_count = 2 # Already at max await engine._process_request(request) assert engine.stats.blocked_requests_count == 1 # No retry enqueued assert engine.scheduler.is_empty @pytest.mark.asyncio async def test_retry_request_has_dont_filter(self): spider = MockSpider(is_blocked_fn=lambda r: True, max_blocked_retries=3) engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default") await engine._process_request(request) retry = await engine.scheduler.dequeue() assert retry.dont_filter is True assert retry._retry_count == 1 @pytest.mark.asyncio async def test_retry_clears_proxy_kwargs(self): spider = MockSpider(is_blocked_fn=lambda r: True, max_blocked_retries=3) engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default", proxy="http://proxy:8080") await engine._process_request(request) retry = await engine.scheduler.dequeue() assert "proxy" not in retry._session_kwargs assert "proxies" not in retry._session_kwargs @pytest.mark.asyncio async def test_callback_yielding_dict_increments_items(self): spider = MockSpider() engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default") await engine._process_request(request) assert engine.stats.items_scraped == 1 assert len(engine.items) == 1 @pytest.mark.asyncio async def test_callback_yielding_request_enqueues(self): async def callback(response) -> AsyncGenerator: yield Request("https://example.com/page2", sid="default") spider = MockSpider() engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default", callback=callback) await engine._process_request(request) assert not engine.scheduler.is_empty @pytest.mark.asyncio async def test_callback_yielding_offsite_request_filtered(self): async def callback(response) -> AsyncGenerator: yield Request("https://other.com/page", sid="default") spider = MockSpider(allowed_domains={"example.com"}) engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default", callback=callback) await engine._process_request(request) assert engine.stats.offsite_requests_count == 1 assert engine.scheduler.is_empty @pytest.mark.asyncio async def test_dropped_item_when_on_scraped_item_returns_none(self): spider = MockSpider(on_scraped_item_fn=lambda item: None) engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default") await engine._process_request(request) assert engine.stats.items_dropped == 1 assert engine.stats.items_scraped == 0 assert len(engine.items) == 0 @pytest.mark.asyncio async def test_callback_exception_calls_on_error(self): async def bad_callback(response) -> AsyncGenerator: raise ValueError("callback boom") yield # noqa: unreachable spider = MockSpider() engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default", callback=bad_callback) await engine._process_request(request) assert len(spider.on_error_calls) == 1 assert isinstance(spider.on_error_calls[0][1], ValueError) @pytest.mark.asyncio async def test_proxy_tracked_in_stats(self): spider = MockSpider() engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default", proxy="http://p:8080") await engine._process_request(request) assert "http://p:8080" in engine.stats.proxies @pytest.mark.asyncio async def test_proxies_dict_tracked_in_stats(self): spider = MockSpider() engine = _make_engine(spider=spider) proxies = {"http": "http://p:8080", "https": "https://p:8443"} request = Request("https://example.com", sid="default", proxies=proxies) await engine._process_request(request) assert len(engine.stats.proxies) == 1 assert engine.stats.proxies[0] == proxies @pytest.mark.asyncio async def test_uses_parse_when_no_callback(self): items_seen = [] async def custom_parse(response) -> AsyncGenerator: yield {"from": "custom_parse"} spider = MockSpider() spider.parse = custom_parse # type: ignore[assignment] engine = _make_engine(spider=spider) request = Request("https://example.com", sid="default") # No callback set → should use spider.parse await engine._process_request(request) assert engine.stats.items_scraped == 1 # --------------------------------------------------------------------------- # Tests: _task_wrapper # --------------------------------------------------------------------------- class TestTaskWrapper: @pytest.mark.asyncio async def test_decrements_active_tasks(self): engine = _make_engine() engine._active_tasks = 1 request = Request("https://example.com", sid="default") await engine._task_wrapper(request) assert engine._active_tasks == 0 @pytest.mark.asyncio async def test_decrements_even_on_error(self): spider = MockSpider() sm = SessionManager() sm.add("default", ErrorSession()) engine = CrawlerEngine(spider, sm) engine._active_tasks = 1 request = Request("https://example.com", sid="default") await engine._task_wrapper(request) assert engine._active_tasks == 0 # --------------------------------------------------------------------------- # Tests: request_pause # --------------------------------------------------------------------------- class TestRequestPause: def test_first_call_sets_pause_requested(self): engine = _make_engine() engine.request_pause() assert engine._pause_requested is True assert engine._force_stop is False def test_second_call_sets_force_stop(self): engine = _make_engine() engine.request_pause() # first engine.request_pause() # second assert engine._pause_requested is True assert engine._force_stop is True def test_third_call_after_force_stop_is_noop(self): engine = _make_engine() engine.request_pause() engine.request_pause() engine.request_pause() # should not raise assert engine._force_stop is True # --------------------------------------------------------------------------- # Tests: checkpoint methods # --------------------------------------------------------------------------- class TestCheckpointMethods: def test_is_checkpoint_time_false_when_disabled(self): engine = _make_engine() # no crawldir assert engine._is_checkpoint_time() is False @pytest.mark.asyncio async def test_save_and_restore_checkpoint(self): with tempfile.TemporaryDirectory() as tmpdir: spider = MockSpider() engine = _make_engine(spider=spider, crawldir=tmpdir) # Enqueue a request so snapshot has data req = Request("https://example.com", sid="default") engine._normalize_request(req) await engine.scheduler.enqueue(req) await engine._save_checkpoint() # Verify checkpoint file exists checkpoint_path = Path(tmpdir) / "checkpoint.pkl" assert checkpoint_path.exists() @pytest.mark.asyncio async def test_restore_when_no_checkpoint_returns_false(self): with tempfile.TemporaryDirectory() as tmpdir: engine = _make_engine(crawldir=tmpdir) result = await engine._restore_from_checkpoint() assert result is False @pytest.mark.asyncio async def test_restore_from_checkpoint_raises_when_disabled(self): engine = _make_engine() # no crawldir → checkpoint disabled assert (await engine._restore_from_checkpoint()) is False # --------------------------------------------------------------------------- # Tests: crawl # --------------------------------------------------------------------------- class TestCrawl: @pytest.mark.asyncio async def test_basic_crawl_returns_stats(self): spider = MockSpider() engine = _make_engine(spider=spider) stats = await engine.crawl() assert isinstance(stats, CrawlStats) assert stats.requests_count >= 1 assert stats.items_scraped >= 1 @pytest.mark.asyncio async def test_crawl_calls_on_start_and_on_close(self): spider = MockSpider() engine = _make_engine(spider=spider) await engine.crawl() assert len(spider.on_start_calls) == 1 assert spider.on_start_calls[0]["resuming"] is False assert spider.on_close_calls == 1 @pytest.mark.asyncio async def test_crawl_sets_stats_timing(self): spider = MockSpider() engine = _make_engine(spider=spider) stats = await engine.crawl() assert stats.start_time > 0 assert stats.end_time > 0 assert stats.end_time >= stats.start_time @pytest.mark.asyncio async def test_crawl_sets_concurrency_stats(self): spider = MockSpider(concurrent_requests=16, concurrent_requests_per_domain=4) engine = _make_engine(spider=spider) stats = await engine.crawl() assert stats.concurrent_requests == 16 assert stats.concurrent_requests_per_domain == 4 @pytest.mark.asyncio async def test_crawl_processes_multiple_start_urls(self): spider = MockSpider() urls = ["https://example.com/1", "https://example.com/2", "https://example.com/3"] async def multi_start_requests() -> AsyncGenerator[Request, None]: for url in urls: yield Request(url, sid="default") spider.start_requests = multi_start_requests # type: ignore[assignment] engine = _make_engine(spider=spider) stats = await engine.crawl() assert stats.requests_count == 3 assert stats.items_scraped == 3 @pytest.mark.asyncio async def test_crawl_follows_yielded_requests(self): """Test that requests yielded from callbacks are processed.""" call_count = 0 async def parse_with_follow(response) -> AsyncGenerator: nonlocal call_count call_count += 1 if call_count == 1: yield Request("https://example.com/page2", sid="default") yield {"page": str(response)} spider = MockSpider() spider.parse = parse_with_follow # type: ignore[assignment] engine = _make_engine(spider=spider) stats = await engine.crawl() assert stats.requests_count == 2 assert stats.items_scraped == 2 @pytest.mark.asyncio async def test_crawl_with_download_delay(self): spider = MockSpider(download_delay=0.01) engine = _make_engine(spider=spider) stats = await engine.crawl() assert stats.download_delay == 0.01 assert stats.requests_count >= 1 @pytest.mark.asyncio async def test_crawl_filters_offsite_requests(self): async def parse_offsite(response) -> AsyncGenerator: yield Request("https://other-domain.com/page", sid="default") yield {"url": str(response)} spider = MockSpider(allowed_domains={"example.com"}) spider.parse = parse_offsite # type: ignore[assignment] engine = _make_engine(spider=spider) stats = await engine.crawl() assert stats.offsite_requests_count == 1 assert stats.requests_count == 1 # Only the initial request @pytest.mark.asyncio async def test_crawl_cleans_up_checkpoint_on_completion(self): with tempfile.TemporaryDirectory() as tmpdir: spider = MockSpider() engine = _make_engine(spider=spider, crawldir=tmpdir) await engine.crawl() checkpoint_path = Path(tmpdir) / "checkpoint.pkl" assert not checkpoint_path.exists() # Cleaned up @pytest.mark.asyncio async def test_crawl_handles_fetch_error_gracefully(self): spider = MockSpider() sm = SessionManager() sm.add("default", ErrorSession()) engine = CrawlerEngine(spider, sm) stats = await engine.crawl() assert stats.failed_requests_count == 1 assert len(spider.on_error_calls) == 1 @pytest.mark.asyncio async def test_crawl_log_levels_populated(self): spider = MockSpider() engine = _make_engine(spider=spider) stats = await engine.crawl() assert isinstance(stats.log_levels_counter, dict) @pytest.mark.asyncio async def test_crawl_resets_state_on_each_run(self): spider = MockSpider() engine = _make_engine(spider=spider) # Run first crawl await engine.crawl() assert engine.stats.requests_count >= 1 # Run second crawl - stats should reset stats = await engine.crawl() # Items are cleared on each crawl assert engine.paused is False # --------------------------------------------------------------------------- # Tests: items property # --------------------------------------------------------------------------- class TestItemsProperty: def test_items_returns_item_list(self): engine = _make_engine() assert isinstance(engine.items, ItemList) def test_items_initially_empty(self): engine = _make_engine() assert len(engine.items) == 0 @pytest.mark.asyncio async def test_items_populated_after_crawl(self): engine = _make_engine() await engine.crawl() assert len(engine.items) >= 1 # --------------------------------------------------------------------------- # Tests: streaming (__aiter__ / _stream) # --------------------------------------------------------------------------- class TestStreaming: @pytest.mark.asyncio async def test_stream_yields_items(self): spider = MockSpider() engine = _make_engine(spider=spider) items = [] async for item in engine: items.append(item) assert len(items) >= 1 assert isinstance(items[0], dict) @pytest.mark.asyncio async def test_stream_processes_follow_up_requests(self): call_count = 0 async def parse_with_follow(response) -> AsyncGenerator: nonlocal call_count call_count += 1 if call_count == 1: yield Request("https://example.com/page2", sid="default") yield {"page": call_count} spider = MockSpider() spider.parse = parse_with_follow # type: ignore[assignment] engine = _make_engine(spider=spider) items = [] async for item in engine: items.append(item) assert len(items) == 2 @pytest.mark.asyncio async def test_stream_items_not_stored_in_items_list(self): """When streaming, items go to the stream, not to engine._items.""" spider = MockSpider() engine = _make_engine(spider=spider) items = [] async for item in engine: items.append(item) # Items were sent through stream, not appended to _items assert len(items) >= 1 assert len(engine.items) == 0 # --------------------------------------------------------------------------- # Tests: pause during crawl # --------------------------------------------------------------------------- class TestPauseDuringCrawl: @pytest.mark.asyncio async def test_pause_stops_crawl_gracefully(self): processed = 0 async def slow_parse(response) -> AsyncGenerator: nonlocal processed processed += 1 # Yield more requests to keep the crawl going if processed <= 2: yield Request(f"https://example.com/p{processed + 1}", sid="default") yield {"n": processed} spider = MockSpider() spider.parse = slow_parse # type: ignore[assignment] engine = _make_engine(spider=spider) # Request pause immediately - the engine will stop as soon as active tasks complete engine._pause_requested = True stats = await engine.crawl() # Should stop without processing everything assert engine._running is False @pytest.mark.asyncio async def test_pause_with_checkpoint_sets_paused(self): with tempfile.TemporaryDirectory() as tmpdir: parse_count = 0 async def parse_and_pause(response) -> AsyncGenerator: nonlocal parse_count parse_count += 1 # Request pause after first request, but yield follow-ups if parse_count == 1: engine.request_pause() yield Request("https://example.com/p2", sid="default") yield {"n": parse_count} spider = MockSpider() spider.parse = parse_and_pause # type: ignore[assignment] engine = _make_engine(spider=spider, crawldir=tmpdir) await engine.crawl() assert engine.paused is True @pytest.mark.asyncio async def test_pause_without_checkpoint_does_not_set_paused(self): spider = MockSpider() engine = _make_engine(spider=spider) engine._pause_requested = True await engine.crawl() assert engine.paused is False # --------------------------------------------------------------------------- # Tests: _prefetch_robots_txt # --------------------------------------------------------------------------- class TestPrefetchRobotsTxt: """_prefetch_robots_txt warms the robots.txt cache before the crawl loop.""" @staticmethod def _make_counting_fetch(): """Return (fetch_fn, calls_list) where calls_list records every (url, sid) pair.""" calls: list[tuple[str, str]] = [] async def _fetch(url: str, sid: str): calls.append((url, sid)) return MockResponse(status=200, body=b"", url=url) return _fetch, calls @pytest.mark.asyncio async def test_prefetch_uses_start_urls(self): fetch_fn, calls = self._make_counting_fetch() spider = MockSpider(robots_txt_obey=True, start_urls=["https://example.com/page1"]) engine = _make_engine(spider=spider) engine._robots_manager = RobotsTxtManager(fetch_fn) await engine._prefetch_robots_txt() assert len(calls) == 1 assert calls[0][0] == "https://example.com/robots.txt" @pytest.mark.asyncio async def test_prefetch_noop_when_robots_disabled(self): fetch_fn, calls = self._make_counting_fetch() spider = MockSpider(robots_txt_obey=False) engine = _make_engine(spider=spider) assert engine._robots_manager is None await engine._prefetch_robots_txt() assert calls == [] @pytest.mark.asyncio async def test_prefetch_noop_when_start_urls_empty(self): fetch_fn, calls = self._make_counting_fetch() spider = MockSpider(robots_txt_obey=True, start_urls=[]) engine = _make_engine(spider=spider) engine._robots_manager = RobotsTxtManager(fetch_fn) await engine._prefetch_robots_txt() assert calls == [] @pytest.mark.asyncio async def test_prefetch_deduplicates_same_domain_in_start_urls(self): fetch_fn, calls = self._make_counting_fetch() spider = MockSpider(robots_txt_obey=True, start_urls=["https://example.com/a", "https://example.com/b"]) engine = _make_engine(spider=spider) engine._robots_manager = RobotsTxtManager(fetch_fn) await engine._prefetch_robots_txt() # set of Request.domain values deduplicates to one task per domain assert len(calls) == 1 assert calls[0][0] == "https://example.com/robots.txt"