diff --git a/tests/spiders/test_engine.py b/tests/spiders/test_engine.py new file mode 100644 index 0000000..ca768bf --- /dev/null +++ b/tests/spiders/test_engine.py @@ -0,0 +1,915 @@ +"""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.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"): + self.status = status + self.body = body + self.url = url + 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, + ): + 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" + + # 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 + with pytest.raises(RuntimeError): + await engine._restore_from_checkpoint() + + +# --------------------------------------------------------------------------- +# 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