diff --git a/src/crawlee/_utils/sitemap.py b/src/crawlee/_utils/sitemap.py index d110f0225c..7e0d339bb3 100644 --- a/src/crawlee/_utils/sitemap.py +++ b/src/crawlee/_utils/sitemap.py @@ -9,6 +9,7 @@ from dataclasses import dataclass from datetime import datetime, timedelta from hashlib import sha256 +from http import HTTPStatus from logging import getLogger from typing import TYPE_CHECKING, Literal, TypedDict from xml.sax import SAXParseException @@ -19,7 +20,7 @@ from yarl import URL from crawlee._utils.urls import filter_url -from crawlee._utils.web import is_status_code_successful +from crawlee._utils.web import is_status_code_server_error, is_status_code_successful from crawlee.errors import ProxyError if TYPE_CHECKING: @@ -46,6 +47,31 @@ DEFAULT_MAX_DEPTH = 10 """Default maximum depth of nested sitemaps to follow, guarding against malicious infinite sitemap chains.""" +SITEMAP_RETRY_DELAY = 1 +"""Seconds to wait before retrying a failed sitemap fetch.""" + + +def _is_successful_sitemap_status(status_code: int) -> bool: + """Return whether a sitemap response status is a successful 2xx.""" + return HTTPStatus.OK <= status_code < HTTPStatus.MULTIPLE_CHOICES + + +class _RetryableSitemapStatusError(Exception): + """Internal signal that a sitemap fetch returned a retryable HTTP status. + + Raised inside the response context manager so the stream is closed before the retry delay. + """ + + def __init__(self, status_code: int) -> None: + super().__init__(f'HTTP status {status_code}') + self.status_code = status_code + + +def _raise_for_retryable_sitemap_status(status_code: int) -> None: + """Raise `_RetryableSitemapStatusError` for HTTP statuses that warrant a fetch retry (408 and 5xx).""" + if status_code == HTTPStatus.REQUEST_TIMEOUT or is_status_code_server_error(status_code): + raise _RetryableSitemapStatusError(status_code) + @dataclass() class SitemapUrl: @@ -363,6 +389,7 @@ async def _fetch_and_process_sitemap( timeout: timedelta | None = None, emit_nested_sitemaps: bool, enqueue_strategy: EnqueueStrategy, + parsed_ok: list[bool], ) -> AsyncGenerator[SitemapUrl | NestedSitemap, None]: """Fetch a sitemap from a URL and process its content.""" if 'url' not in source: @@ -376,6 +403,13 @@ async def _fetch_and_process_sitemap( async with http_client.stream( sitemap_url, method='GET', headers=SITEMAP_HEADERS, proxy_info=proxy_info, timeout=timeout ) as response: + status_code = response.status_code + if not _is_successful_sitemap_status(status_code): + # Retryable statuses raise and route to the retry handler below; the rest skip this source. + _raise_for_retryable_sitemap_status(status_code) + logger.warning(f'Skipping sitemap {sitemap_url} due to HTTP status code {status_code}.') + return + # Determine content type and compression content_type = response.headers.get('content-type', '') @@ -456,15 +490,25 @@ async def _fetch_and_process_sitemap( yield result finally: parser.close() - break + parsed_ok.append(True) + except _RetryableSitemapStatusError as e: + if retries_left > 0: + logger.warning( + f'Error fetching sitemap {sitemap_url}: HTTP status {e.status_code}. Retries left: {retries_left}' + ) + await asyncio.sleep(SITEMAP_RETRY_DELAY) + else: + logger.warning(f'Failed to fetch sitemap {sitemap_url}, no retries left: HTTP status {e.status_code}.') except Exception as e: if retries_left > 0: logger.warning(f'Error fetching sitemap {sitemap_url}: {e}. Retries left: {retries_left}') - await asyncio.sleep(1) # Brief pause before retry + await asyncio.sleep(SITEMAP_RETRY_DELAY) else: logger.exception(f'Failed to fetch sitemap {sitemap_url}, no retries left.') raise + else: + return class Sitemap: @@ -525,6 +569,10 @@ async def parse_sitemap( Default `ParseSitemapOptions.enqueue_strategy` is `same-hostname` which will skip cross-host URLs. Use strategy `all` to process all links. + + URL sources answering with a non-2xx status are skipped: retryable statuses (408 and 5xx) are retried first, + others are skipped immediately. If at least one source fails with a fetch error after all retries and no source + succeeds, the first such error is raised once all sources have been processed. """ # Set default options options = options or {} @@ -537,6 +585,8 @@ async def parse_sitemap( # Setup working state sources = list(initial_sources) visited_sitemap_urls: set[str] = set() + successful_sources = 0 + source_errors: list[Exception] = [] # Process sources until the queue is empty while sources: @@ -559,6 +609,8 @@ async def parse_sitemap( enqueue_strategy=enqueue_strategy, ): yield result + # Raw sources cannot fetch-fail, so they always count as successful for the all-sources-failed check. + successful_sources += 1 elif source['type'] == 'url' and 'url' in source: # Add to visited set before processing to avoid duplicates @@ -566,23 +618,34 @@ async def parse_sitemap( raise RuntimeError('HttpClient must be provided for URL-based sitemap sources.') visited_sitemap_urls.add(source['url']) + parsed_ok: list[bool] = [] - async for result in _fetch_and_process_sitemap( - http_client=http_client, - source=source, - depth=depth, - visited_sitemap_urls=visited_sitemap_urls, - sources=sources, - retries_left=sitemap_retries, - emit_nested_sitemaps=emit_nested_sitemaps, - enqueue_strategy=enqueue_strategy, - proxy_info=proxy_info, - timeout=timeout, - ): - yield result + try: + async for result in _fetch_and_process_sitemap( + http_client=http_client, + source=source, + depth=depth, + visited_sitemap_urls=visited_sitemap_urls, + sources=sources, + retries_left=sitemap_retries, + emit_nested_sitemaps=emit_nested_sitemaps, + enqueue_strategy=enqueue_strategy, + proxy_info=proxy_info, + timeout=timeout, + parsed_ok=parsed_ok, + ): + yield result + if parsed_ok: + successful_sources += 1 + except Exception as e: + # Already logged by `_fetch_and_process_sitemap`; raised below only if no source succeeds. + source_errors.append(e) else: logger.warning(f'Invalid source configuration: {source}') + if source_errors and successful_sources == 0: + raise source_errors[0] + async def _merge_async_generators(*generators: AsyncGenerator) -> AsyncGenerator: queue: asyncio.Queue = asyncio.Queue() diff --git a/src/crawlee/request_loaders/_sitemap_request_loader.py b/src/crawlee/request_loaders/_sitemap_request_loader.py index 230c50affc..a528753e6d 100644 --- a/src/crawlee/request_loaders/_sitemap_request_loader.py +++ b/src/crawlee/request_loaders/_sitemap_request_loader.py @@ -376,58 +376,64 @@ async def _load_sitemaps(self) -> None: ) parsed_sitemap_url = URL(sitemap_url) - async with aclosing( - parse_sitemap( - [SitemapSource(type='url', url=sitemap_url)], - self._http_client, - proxy_info=self._proxy_info, - options=parse_options, - ) - ) as sitemap_items: - async for item in sitemap_items: - if isinstance(item, NestedSitemap): - # Add nested sitemap to queue - if ( - item.loc not in state.pending_sitemap_urls - and item.loc not in state.processed_sitemap_urls - ): - if current_depth >= DEFAULT_MAX_DEPTH: - logger.warning( - f'Skipping nested sitemap {item.loc!r}: max depth {DEFAULT_MAX_DEPTH} reached.' - ) - continue - if not self._passes_filters(item.loc, parsed_sitemap_url, 'nested sitemap'): - continue - state.pending_sitemap_urls.append(item.loc) - state.sitemap_depths[item.loc] = current_depth + 1 - continue - - if isinstance(item, SitemapUrl): - url = item.loc + try: + async with aclosing( + parse_sitemap( + [SitemapSource(type='url', url=sitemap_url)], + self._http_client, + proxy_info=self._proxy_info, + options=parse_options, + ) + ) as sitemap_items: + async for item in sitemap_items: + if isinstance(item, NestedSitemap): + # Add nested sitemap to queue + if ( + item.loc not in state.pending_sitemap_urls + and item.loc not in state.processed_sitemap_urls + ): + if current_depth >= DEFAULT_MAX_DEPTH: + logger.warning( + f'Skipping nested sitemap {item.loc!r}: ' + f'max depth {DEFAULT_MAX_DEPTH} reached.' + ) + continue + if not self._passes_filters(item.loc, parsed_sitemap_url, 'nested sitemap'): + continue + state.pending_sitemap_urls.append(item.loc) + state.sitemap_depths[item.loc] = current_depth + 1 + continue - state = await self._get_state() + if isinstance(item, SitemapUrl): + url = item.loc - # Skip if already processed - if url in state.current_sitemap_processed_urls: - continue + state = await self._get_state() - # Check if URL should be included - if not self._check_url_patterns(url, self._include, self._exclude): - continue + # Skip if already processed + if url in state.current_sitemap_processed_urls: + continue - if not self._passes_filters(url, parsed_sitemap_url, 'sitemap URL'): - continue + # Check if URL should be included + if not self._check_url_patterns(url, self._include, self._exclude): + continue - # Check if we have capacity in the queue - await self._queue_has_capacity.wait() + if not self._passes_filters(url, parsed_sitemap_url, 'sitemap URL'): + continue - async with self._queue_lock: - state.url_queue.append(url) - state.current_sitemap_processed_urls.add(url) - state.total_count += 1 - if len(state.url_queue) >= self._max_buffer_size: - # Notify that the queue is full - self._queue_has_capacity.clear() + # Check if we have capacity in the queue + await self._queue_has_capacity.wait() + + async with self._queue_lock: + state.url_queue.append(url) + state.current_sitemap_processed_urls.add(url) + state.total_count += 1 + if len(state.url_queue) >= self._max_buffer_size: + # Notify that the queue is full + self._queue_has_capacity.clear() + except Exception: + # A sitemap that fails to load is marked processed and skipped, so one dead sitemap + # does not abandon the remaining pending sitemaps. + logger.warning(f'Failed to load sitemap {sitemap_url}, skipping it.') # Clear current sitemap after processing state = await self._get_state() @@ -442,5 +448,6 @@ async def _load_sitemaps(self) -> None: state.completed = True except Exception: + # Not re-raised: nothing retrieves the task's exception (`is_finished` only checks doneness), so it + # would only resurface as an "exception was never retrieved" asyncio error when the task is collected. logger.exception('Error loading sitemaps') - raise diff --git a/tests/unit/_utils/test_sitemap.py b/tests/unit/_utils/test_sitemap.py index e1030844ab..e88cca32d4 100644 --- a/tests/unit/_utils/test_sitemap.py +++ b/tests/unit/_utils/test_sitemap.py @@ -14,6 +14,7 @@ DEFAULT_MAX_DEPTH, ParseSitemapOptions, Sitemap, + SitemapSource, SitemapUrl, _TxtSitemapParser, _XMLSaxSitemapHandler, @@ -22,7 +23,12 @@ parse_sitemap, ) from crawlee.http_clients._base import HttpClient, HttpResponse -from tests.unit.utils import DEFAULT_URL, get_basic_results, get_basic_sitemap +from tests.unit.utils import ( + DEFAULT_URL, + get_basic_results, + get_basic_sitemap, + make_status_stream_client, +) if TYPE_CHECKING: from collections.abc import AsyncIterator, Callable @@ -60,6 +66,7 @@ async def read_stream() -> 'AsyncIterator[bytes]': yield body response = MagicMock(spec=HttpResponse) + response.status_code = 200 response.headers = {'content-type': 'application/xml; charset=utf-8'} response.read_stream = read_stream yield cast('HttpResponse', response) @@ -81,6 +88,7 @@ async def read_stream() -> 'AsyncIterator[bytes]': yield body_for_url(url) response = MagicMock(spec=HttpResponse) + response.status_code = 200 response.headers = {'content-type': 'application/xml; charset=utf-8'} response.read_stream = read_stream yield cast('HttpResponse', response) @@ -337,8 +345,9 @@ async def test_malformed_sitemap_keeps_urls() -> None: assert sitemap.urls == [f'{DEFAULT_URL}first', f'{DEFAULT_URL}second'] -async def test_sitemap_fetch_retries_on_transient_error() -> None: +async def test_sitemap_fetch_retries_on_transient_error(monkeypatch: pytest.MonkeyPatch) -> None: """Transient fetch errors are retried up to `sitemap_retries` times before giving up.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) client, attempts = _make_flaky_stream_client(get_basic_sitemap().encode(), fail_times=2) items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] @@ -347,8 +356,9 @@ async def test_sitemap_fetch_retries_on_transient_error() -> None: assert {item.loc for item in items} == get_basic_results() -async def test_sitemap_fetch_raises_after_retries_exhausted() -> None: +async def test_sitemap_fetch_raises_after_retries_exhausted(monkeypatch: pytest.MonkeyPatch) -> None: """A persistent fetch error is raised to the caller once all retries are exhausted.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) client, attempts = _make_flaky_stream_client(get_basic_sitemap().encode(), fail_times=10) with pytest.raises(ConnectionError): @@ -357,6 +367,158 @@ async def test_sitemap_fetch_raises_after_retries_exhausted() -> None: assert len(attempts) == 3 +async def test_sitemap_fetch_retries_retryable_http_status(monkeypatch: pytest.MonkeyPatch) -> None: + """Retryable HTTP errors are retried before parsing a successful response.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + sitemap_url = f'{DEFAULT_URL}sitemap.xml' + client, attempts = make_status_stream_client( + {sitemap_url: [(503, b''), (503, b''), (200, get_basic_sitemap().encode())]} + ) + + items = [item async for item in parse_sitemap([{'type': 'url', 'url': sitemap_url}], client)] + + assert attempts == [503, 503, 200] + assert {item.loc for item in items} == get_basic_results() + + +async def test_sitemap_fetch_skips_http_error_after_retries_exhausted(monkeypatch: pytest.MonkeyPatch) -> None: + """A persistent retryable HTTP error is skipped once retries are exhausted.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + sitemap_url = f'{DEFAULT_URL}sitemap.xml' + client, attempts = make_status_stream_client({sitemap_url: [(503, b'')]}) + + items = [item async for item in parse_sitemap([{'type': 'url', 'url': sitemap_url}], client)] + + assert attempts == [503, 503, 503] + assert items == [] + + +async def test_sitemap_fetch_does_not_retry_terminal_http_status() -> None: + """Terminal HTTP errors are skipped without parsing their response body or retrying.""" + sitemap_url = f'{DEFAULT_URL}sitemap.xml' + client, attempts = make_status_stream_client({sitemap_url: [(404, get_basic_sitemap().encode())]}) + + items = [item async for item in parse_sitemap([{'type': 'url', 'url': sitemap_url}], client)] + + assert attempts == [404] + assert items == [] + + +async def test_sitemap_fetch_skips_redirect_without_retry() -> None: + """Redirect responses that reach the parser are skipped without retrying.""" + sitemap_url = f'{DEFAULT_URL}sitemap.xml' + client, attempts = make_status_stream_client({sitemap_url: [(302, get_basic_sitemap().encode())]}) + + items = [item async for item in parse_sitemap([{'type': 'url', 'url': sitemap_url}], client)] + + assert attempts == [302] + assert items == [] + + +async def test_sitemap_fetch_does_not_retry_too_many_requests() -> None: + """429 is not retried on the session-less sitemap fetch path.""" + sitemap_url = f'{DEFAULT_URL}sitemap.xml' + client, attempts = make_status_stream_client({sitemap_url: [(429, b'')]}) + + items = [item async for item in parse_sitemap([{'type': 'url', 'url': sitemap_url}], client)] + + assert attempts == [429] + assert items == [] + + +async def test_sitemap_partial_http_failure_keeps_healthy_source() -> None: + """An HTTP failure in one source does not discard URLs from another source.""" + healthy_url = f'{DEFAULT_URL}sitemap.xml' + missing_url = f'{DEFAULT_URL}missing.xml' + client, attempts = make_status_stream_client( + { + healthy_url: [(200, get_basic_sitemap().encode())], + missing_url: [(404, b'')], + } + ) + sources: list[SitemapSource] = [ + {'type': 'url', 'url': healthy_url}, + {'type': 'url', 'url': missing_url}, + ] + + items = [item async for item in parse_sitemap(sources, client)] + + assert attempts == [200, 404] + assert {item.loc for item in items} == get_basic_results() + + +async def test_sitemap_partial_fetch_failure_keeps_healthy_source(monkeypatch: pytest.MonkeyPatch) -> None: + """A fetch exception in one source is suppressed when another source succeeds.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + client, attempts = _make_flaky_stream_client(get_basic_sitemap().encode(), fail_times=3) + sources: list[SitemapSource] = [ + {'type': 'url', 'url': f'{DEFAULT_URL}broken.xml'}, + {'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}, + ] + + items = [item async for item in parse_sitemap(sources, client)] + + assert len(attempts) == 4 + assert {item.loc for item in items} == get_basic_results() + + +async def test_sitemap_raw_source_success_suppresses_url_fetch_error(monkeypatch: pytest.MonkeyPatch) -> None: + """A failing URL source does not discard the URLs from a successful raw sibling source.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + broken_url = f'{DEFAULT_URL}broken.xml' + client, _ = make_status_stream_client({broken_url: [ConnectionError('Network error')]}) + sources: list[SitemapSource] = [ + {'type': 'raw', 'content': get_basic_sitemap()}, + {'type': 'url', 'url': broken_url}, + ] + + items = [item async for item in parse_sitemap(sources, client)] + + assert {item.loc for item in items} == get_basic_results() + + +async def test_sitemap_skipped_source_does_not_hide_fetch_error(monkeypatch: pytest.MonkeyPatch) -> None: + """A skipped HTTP source does not count as success, so a sibling fetch error still raises.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + missing_url = f'{DEFAULT_URL}missing.xml' + broken_url = f'{DEFAULT_URL}broken.xml' + client, attempts = make_status_stream_client( + { + missing_url: [(404, b'')], + broken_url: [ConnectionError('Network error')], + } + ) + sources: list[SitemapSource] = [ + {'type': 'url', 'url': missing_url}, + {'type': 'url', 'url': broken_url}, + ] + + with pytest.raises(ConnectionError, match='Network error'): + _ = [item async for item in parse_sitemap(sources, client)] + + assert attempts == [404] + + +async def test_sitemap_raises_first_source_error_when_all_fail(monkeypatch: pytest.MonkeyPatch) -> None: + """If every source fails with a real error, the first error is raised.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + first_url = f'{DEFAULT_URL}first.xml' + second_url = f'{DEFAULT_URL}second.xml' + client, _ = make_status_stream_client( + { + first_url: [ConnectionError('first')], + second_url: [OSError('second')], + } + ) + sources: list[SitemapSource] = [ + {'type': 'url', 'url': first_url}, + {'type': 'url', 'url': second_url}, + ] + + with pytest.raises(ConnectionError, match='first'): + _ = [item async for item in parse_sitemap(sources, client)] + + async def test_gzip_bomb_sitemap_truncated_at_size_cap(monkeypatch: pytest.MonkeyPatch) -> None: """A gzip sitemap inflating past the size cap is truncated instead of being decompressed without bound.""" monkeypatch.setattr('crawlee._utils.sitemap.MAX_SITEMAP_SIZE', 64 * 1024) @@ -388,6 +550,7 @@ async def read_stream() -> 'AsyncIterator[bytes]': yield b'\x00' * 65536 response = MagicMock(spec=HttpResponse) + response.status_code = 200 response.headers = {'content-type': 'application/gzip'} response.read_stream = read_stream yield cast('HttpResponse', response) @@ -413,6 +576,7 @@ async def read_stream() -> 'AsyncIterator[bytes]': yield body response = MagicMock(spec=HttpResponse) + response.status_code = 200 response.headers = {'content-type': 'application/gzip'} response.read_stream = read_stream yield cast('HttpResponse', response) diff --git a/tests/unit/request_loaders/test_sitemap_request_loader.py b/tests/unit/request_loaders/test_sitemap_request_loader.py index a70c133afa..e39e2a04fb 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -4,14 +4,23 @@ from typing import TYPE_CHECKING, Any, cast from unittest.mock import AsyncMock, MagicMock, patch +import pytest from yarl import URL from crawlee import RequestOptions, RequestTransformAction +from crawlee._types import BasicCrawlingContext from crawlee._utils.sitemap import DEFAULT_MAX_DEPTH +from crawlee.crawlers import BasicCrawler from crawlee.http_clients._base import HttpClient, HttpResponse from crawlee.request_loaders._sitemap_request_loader import SitemapRequestLoader from crawlee.storages import KeyValueStore -from tests.unit.utils import DEFAULT_URL, get_basic_results, get_basic_sitemap, poll_until_condition +from tests.unit.utils import ( + DEFAULT_URL, + get_basic_results, + get_basic_sitemap, + make_status_stream_client, + poll_until_condition, +) if TYPE_CHECKING: from collections.abc import AsyncIterator @@ -77,6 +86,107 @@ async def test_sitemap_traversal(server_url: URL, http_client: HttpClient) -> No assert await sitemap_loader.get_handled_count() == 5 +async def test_sitemap_http_error_is_retried_before_loading_requests(monkeypatch: pytest.MonkeyPatch) -> None: + """The loader retries transient HTTP errors and loads the eventual sitemap response.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + sitemap_url = f'{DEFAULT_URL}sitemap.xml' + client, attempts = make_status_stream_client( + {sitemap_url: [(503, b''), (503, b''), (200, get_basic_sitemap().encode())]} + ) + loader = SitemapRequestLoader([sitemap_url], http_client=client) + + while not await loader.is_finished(): + request = await loader.fetch_next_request() + if request: + await loader.mark_request_as_handled(request) + + assert attempts == [503, 503, 200] + assert await loader.get_total_count() == 5 + + +async def test_sitemap_http_error_is_skipped_after_retries_exhausted(monkeypatch: pytest.MonkeyPatch) -> None: + """The loader finishes empty after an exhausted sitemap HTTP error.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + sitemap_url = f'{DEFAULT_URL}sitemap.xml' + client, attempts = make_status_stream_client({sitemap_url: [(503, b'')]}) + loader = SitemapRequestLoader([sitemap_url], http_client=client) + + assert await loader.fetch_next_request() is None + + assert attempts == [503, 503, 503] + + +async def test_sitemap_loader_drains_requests_after_later_source_fails(monkeypatch: pytest.MonkeyPatch) -> None: + """Already loaded URLs are still handed out after a later sitemap fails.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + client, _ = make_status_stream_client( + { + f'{DEFAULT_URL}sitemap.xml': [(200, get_basic_sitemap().encode())], + f'{DEFAULT_URL}broken.xml': [ConnectionError('Network error')], + } + ) + loader = SitemapRequestLoader( + [f'{DEFAULT_URL}sitemap.xml', f'{DEFAULT_URL}broken.xml'], http_client=client, max_buffer_size=10 + ) + + requests = [] + for _ in range(5): + request = await loader.fetch_next_request() + assert request is not None + requests.append(request) + + assert not await loader.is_finished() + for request in requests: + await loader.mark_request_as_handled(request) + + assert await poll_until_condition(loader.is_finished) + + +async def test_sitemap_loader_continues_after_earlier_source_fails(monkeypatch: pytest.MonkeyPatch) -> None: + """A sitemap that fails to load is skipped and the remaining pending sitemaps are still loaded.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + client, attempts = make_status_stream_client( + { + f'{DEFAULT_URL}broken.xml': [ConnectionError('Network error')], + f'{DEFAULT_URL}sitemap.xml': [(200, get_basic_sitemap().encode())], + } + ) + loader = SitemapRequestLoader([f'{DEFAULT_URL}broken.xml', f'{DEFAULT_URL}sitemap.xml'], http_client=client) + + urls = [] + while not await loader.is_finished(): + request = await loader.fetch_next_request() + if request: + urls.append(request.url) + await loader.mark_request_as_handled(request) + + assert attempts == [200] + assert set(urls) == get_basic_results() + assert await loader.get_total_count() == 5 + + +async def test_crawler_tandem_continues_when_sitemap_is_dead(monkeypatch: pytest.MonkeyPatch) -> None: + """A dead sitemap plus seeded requests still crawls the seeded work.""" + monkeypatch.setattr('crawlee._utils.sitemap.SITEMAP_RETRY_DELAY', 0) + broken_url = f'{DEFAULT_URL}broken.xml' + client, _ = make_status_stream_client({broken_url: [ConnectionError('Network error')]}) + loader = SitemapRequestLoader([broken_url], http_client=client) + request_manager = await loader.to_tandem() + + crawler = BasicCrawler(request_manager=request_manager) + visited: list[str] = [] + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + visited.append(context.request.url) + + stats = await crawler.run([f'{DEFAULT_URL}seeded-a', f'{DEFAULT_URL}seeded-b']) + + assert set(visited) == {f'{DEFAULT_URL}seeded-a', f'{DEFAULT_URL}seeded-b'} + assert stats.requests_finished == 2 + assert stats.requests_failed == 0 + + async def test_is_empty_does_not_depend_on_fetch_next_request(server_url: URL, http_client: HttpClient) -> None: sitemap_url = (server_url / 'sitemap.xml').with_query( base64=encode_base64(get_basic_sitemap(url=server_url).encode()) diff --git a/tests/unit/utils.py b/tests/unit/utils.py index 02f3ece24b..e1875482d2 100644 --- a/tests/unit/utils.py +++ b/tests/unit/utils.py @@ -4,12 +4,16 @@ import inspect import sys import time -from typing import TYPE_CHECKING, TypeVar, cast, overload +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING, Any, TypeVar, cast, overload +from unittest.mock import AsyncMock, MagicMock import pytest +from crawlee.http_clients._base import HttpClient, HttpResponse + if TYPE_CHECKING: - from collections.abc import Awaitable, Callable + from collections.abc import AsyncIterator, Awaitable, Callable from yarl import URL @@ -18,6 +22,44 @@ run_alone_on_mac = pytest.mark.run_alone if sys.platform == 'darwin' else lambda x: x +def make_status_stream_client( + responses: dict[str, list[tuple[int, bytes] | Exception]], +) -> tuple[AsyncMock, list[int]]: + """Create a mock client that answers each URL with its next `(status, body)` response or raised exception. + + The last entry of a URL's sequence repeats for any further requests. The returned list records the status of + every response served (exception entries are not recorded). + """ + attempts: list[int] = [] + indexes: dict[str, int] = {} + + @asynccontextmanager + async def stream(url: str, **_kwargs: Any) -> AsyncIterator[HttpResponse]: + sequence = responses[url] + index = indexes.get(url, 0) + indexes[url] = index + 1 + spec = sequence[min(index, len(sequence) - 1)] + if isinstance(spec, Exception): + raise spec + + status, body = spec + attempts.append(status) + + async def read_stream() -> AsyncIterator[bytes]: + if body: + yield body + + response = MagicMock(spec=HttpResponse) + response.status_code = status + response.headers = {'content-type': 'application/xml; charset=utf-8'} + response.read_stream = read_stream + yield cast('HttpResponse', response) + + client = AsyncMock(spec=HttpClient) + client.stream = stream + return client, attempts + + async def maybe_await(value: Awaitable[T] | T) -> T: """Await `value` if it is awaitable, otherwise return it unchanged.