From d5f68ecb705bcc9af97784c89b98bc0e26a2cb04 Mon Sep 17 00:00:00 2001 From: Anas Khan <83116240+anxkhn@users.noreply.github.com> Date: Thu, 6 Aug 2026 12:56:19 +0530 Subject: [PATCH 1/5] Sitemap fetching treats HTTP server errors as successful empty sitemaps Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com> --- src/crawlee/_utils/sitemap.py | 18 +++++- .../_sitemap_request_loader.py | 2 + tests/unit/_utils/test_sitemap.py | 61 +++++++++++++++++++ .../test_sitemap_request_loader.py | 53 ++++++++++++++++ 4 files changed, 133 insertions(+), 1 deletion(-) diff --git a/src/crawlee/_utils/sitemap.py b/src/crawlee/_utils/sitemap.py index d110f0225c..a07aa0ac8d 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 @@ -20,7 +21,9 @@ from crawlee._utils.urls import filter_url from crawlee._utils.web import is_status_code_successful -from crawlee.errors import ProxyError +from crawlee.errors import HttpStatusCodeError, ProxyError + +_HTTP_STATUS_UPPER_BOUND = 600 if TYPE_CHECKING: from collections.abc import AsyncGenerator @@ -30,6 +33,12 @@ from crawlee.http_clients import HttpClient from crawlee.proxy_configuration import ProxyInfo + +def _raise_for_sitemap_status(status_code: int) -> None: + if not is_status_code_successful(status_code): + raise HttpStatusCodeError('Error status code returned while fetching sitemap', status_code) + + logger = getLogger(__name__) VALID_CHANGE_FREQS = {'always', 'hourly', 'daily', 'weekly', 'monthly', 'yearly', 'never'} @@ -376,6 +385,8 @@ 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: + _raise_for_sitemap_status(response.status_code) + # Determine content type and compression content_type = response.headers.get('content-type', '') @@ -459,6 +470,11 @@ async def _fetch_and_process_sitemap( break except Exception as e: + if isinstance(e, HttpStatusCodeError) and not ( + e.status_code in (HTTPStatus.REQUEST_TIMEOUT, HTTPStatus.TOO_MANY_REQUESTS) + or HTTPStatus.INTERNAL_SERVER_ERROR <= e.status_code < _HTTP_STATUS_UPPER_BOUND + ): + raise 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 diff --git a/src/crawlee/request_loaders/_sitemap_request_loader.py b/src/crawlee/request_loaders/_sitemap_request_loader.py index 230c50affc..5cb62ec97b 100644 --- a/src/crawlee/request_loaders/_sitemap_request_loader.py +++ b/src/crawlee/request_loaders/_sitemap_request_loader.py @@ -214,6 +214,8 @@ async def is_empty(self) -> bool: async def is_finished(self) -> bool: """Check if all URLs have been processed.""" state = await self._get_state() + if self._loading_task.done() and not self._loading_task.cancelled(): + self._loading_task.result() return not state.url_queue and len(state.in_progress) == 0 and self._loading_task.done() @override diff --git a/tests/unit/_utils/test_sitemap.py b/tests/unit/_utils/test_sitemap.py index e1030844ab..66ddd79ca5 100644 --- a/tests/unit/_utils/test_sitemap.py +++ b/tests/unit/_utils/test_sitemap.py @@ -21,6 +21,7 @@ discover_valid_sitemaps, parse_sitemap, ) +from crawlee.errors import HttpStatusCodeError from crawlee.http_clients._base import HttpClient, HttpResponse from tests.unit.utils import DEFAULT_URL, get_basic_results, get_basic_sitemap @@ -60,6 +61,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 +83,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) @@ -90,6 +93,30 @@ async def read_stream() -> 'AsyncIterator[bytes]': return client, fetched +def _make_status_stream_client(responses: list[tuple[int, bytes]]) -> tuple[AsyncMock, list[int]]: + """Create a mock client returning the provided status and body sequence.""" + attempts: list[int] = [] + + @asynccontextmanager + async def stream(_url: str, **_kwargs: Any) -> 'AsyncIterator[HttpResponse]': + status, body = responses[min(len(attempts), len(responses) - 1)] + 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 + + def compress_gzip(data: str) -> bytes: """Compress a string using gzip.""" return gzip.compress(data.encode()) @@ -357,6 +384,38 @@ async def test_sitemap_fetch_raises_after_retries_exhausted() -> None: assert len(attempts) == 3 +async def test_sitemap_fetch_retries_retryable_http_status() -> None: + """Retryable HTTP errors are retried before parsing a successful response.""" + client, attempts = _make_status_stream_client( + [(503, b''), (503, b''), (200, get_basic_sitemap().encode())] + ) + + items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] + + assert attempts == [503, 503, 200] + assert {item.loc for item in items} == get_basic_results() + + +async def test_sitemap_fetch_rejects_http_error_after_retries_exhausted() -> None: + """A persistent retryable HTTP error is raised once retries are exhausted.""" + client, attempts = _make_status_stream_client([(503, b'')]) + + with pytest.raises(HttpStatusCodeError, match='503'): + _ = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] + + assert attempts == [503, 503, 503] + + +async def test_sitemap_fetch_does_not_retry_terminal_http_status() -> None: + """Terminal HTTP errors are raised without parsing their response body or retrying.""" + client, attempts = _make_status_stream_client([(404, get_basic_sitemap().encode())]) + + with pytest.raises(HttpStatusCodeError, match='404'): + _ = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] + + assert attempts == [404] + + 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 +447,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 +473,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..d8b936d201 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -4,10 +4,12 @@ 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._utils.sitemap import DEFAULT_MAX_DEPTH +from crawlee.errors import HttpStatusCodeError from crawlee.http_clients._base import HttpClient, HttpResponse from crawlee.request_loaders._sitemap_request_loader import SitemapRequestLoader from crawlee.storages import KeyValueStore @@ -29,6 +31,30 @@ def encode_base64(data: bytes) -> str: return base64.b64encode(data).decode('utf-8') +def _make_status_stream_client(responses: list[tuple[int, bytes]]) -> tuple[AsyncMock, list[int]]: + """Create a mock client returning the provided status and body sequence.""" + attempts: list[int] = [] + + @asynccontextmanager + async def stream(_url: str, **_kwargs: Any) -> 'AsyncIterator[HttpResponse]': + status, body = responses[min(len(attempts), len(responses) - 1)] + 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 test_nested_sitemap_chain_bounded_by_max_depth() -> None: """A malicious endless chain of unique nested sitemaps is followed only up to the default max depth.""" fetched: list[str] = [] @@ -77,6 +103,33 @@ 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() -> None: + """The loader retries transient HTTP errors and loads the eventual sitemap response.""" + client, attempts = _make_status_stream_client( + [(503, b''), (503, b''), (200, get_basic_sitemap().encode())] + ) + loader = SitemapRequestLoader([f'{DEFAULT_URL}sitemap.xml'], 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_propagated_after_retries_exhausted() -> None: + """The loader exposes an exhausted sitemap fetch instead of reporting successful completion.""" + client, attempts = _make_status_stream_client([(503, b'')]) + loader = SitemapRequestLoader([f'{DEFAULT_URL}sitemap.xml'], http_client=client) + + with pytest.raises(HttpStatusCodeError, match='503'): + await loader.fetch_next_request() + + assert attempts == [503, 503, 503] + + 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()) From 4534c73cd13ecfda1bde2881853d4592a221d307 Mon Sep 17 00:00:00 2001 From: Anas Khan <83116240+anxkhn@users.noreply.github.com> Date: Thu, 6 Aug 2026 14:21:00 +0530 Subject: [PATCH 2/5] style: format sitemap tests Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com> --- tests/unit/_utils/test_sitemap.py | 4 +--- tests/unit/request_loaders/test_sitemap_request_loader.py | 4 +--- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/tests/unit/_utils/test_sitemap.py b/tests/unit/_utils/test_sitemap.py index 66ddd79ca5..6291ed8389 100644 --- a/tests/unit/_utils/test_sitemap.py +++ b/tests/unit/_utils/test_sitemap.py @@ -386,9 +386,7 @@ async def test_sitemap_fetch_raises_after_retries_exhausted() -> None: async def test_sitemap_fetch_retries_retryable_http_status() -> None: """Retryable HTTP errors are retried before parsing a successful response.""" - client, attempts = _make_status_stream_client( - [(503, b''), (503, b''), (200, get_basic_sitemap().encode())] - ) + client, attempts = _make_status_stream_client([(503, b''), (503, b''), (200, get_basic_sitemap().encode())]) items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] diff --git a/tests/unit/request_loaders/test_sitemap_request_loader.py b/tests/unit/request_loaders/test_sitemap_request_loader.py index d8b936d201..0a1fce2258 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -105,9 +105,7 @@ async def test_sitemap_traversal(server_url: URL, http_client: HttpClient) -> No async def test_sitemap_http_error_is_retried_before_loading_requests() -> None: """The loader retries transient HTTP errors and loads the eventual sitemap response.""" - client, attempts = _make_status_stream_client( - [(503, b''), (503, b''), (200, get_basic_sitemap().encode())] - ) + client, attempts = _make_status_stream_client([(503, b''), (503, b''), (200, get_basic_sitemap().encode())]) loader = SitemapRequestLoader([f'{DEFAULT_URL}sitemap.xml'], http_client=client) while not await loader.is_finished(): From 9f1318ea9e48a0941edc581e0aa20b5688464f42 Mon Sep 17 00:00:00 2001 From: Anas Khan <83116240+anxkhn@users.noreply.github.com> Date: Tue, 11 Aug 2026 20:06:03 +0530 Subject: [PATCH 3/5] fix(sitemap): handle partial fetch failures Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com> --- src/crawlee/_utils/sitemap.py | 72 +++++++----- .../request_loaders/_request_loader.py | 4 +- .../_sitemap_request_loader.py | 4 +- tests/unit/_utils/test_sitemap.py | 104 +++++++++++------- .../test_sitemap_request_loader.py | 88 +++++++++------ tests/unit/utils.py | 38 ++++++- 6 files changed, 206 insertions(+), 104 deletions(-) diff --git a/src/crawlee/_utils/sitemap.py b/src/crawlee/_utils/sitemap.py index a07aa0ac8d..f97fdfb3a8 100644 --- a/src/crawlee/_utils/sitemap.py +++ b/src/crawlee/_utils/sitemap.py @@ -20,11 +20,9 @@ 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 HttpStatusCodeError, ProxyError -_HTTP_STATUS_UPPER_BOUND = 600 - if TYPE_CHECKING: from collections.abc import AsyncGenerator from xml.sax.xmlreader import AttributesImpl @@ -33,12 +31,6 @@ from crawlee.http_clients import HttpClient from crawlee.proxy_configuration import ProxyInfo - -def _raise_for_sitemap_status(status_code: int) -> None: - if not is_status_code_successful(status_code): - raise HttpStatusCodeError('Error status code returned while fetching sitemap', status_code) - - logger = getLogger(__name__) VALID_CHANGE_FREQS = {'always', 'hourly', 'daily', 'weekly', 'monthly', 'yearly', 'never'} @@ -56,6 +48,21 @@ def _raise_for_sitemap_status(status_code: int) -> None: """Default maximum depth of nested sitemaps to follow, guarding against malicious infinite sitemap chains.""" +def _raise_for_sitemap_status(status_code: int) -> None: + """Raise `HttpStatusCodeError` if the sitemap response status is not 2xx.""" + if not HTTPStatus.OK <= status_code < HTTPStatus.MULTIPLE_CHOICES: + raise HttpStatusCodeError('Error status code returned while fetching sitemap', status_code) + + +def _is_retryable_sitemap_status(status_code: int) -> bool: + """Return whether a sitemap response status should be retried.""" + return ( + HTTPStatus.MULTIPLE_CHOICES <= status_code < HTTPStatus.BAD_REQUEST + or status_code in (HTTPStatus.REQUEST_TIMEOUT, HTTPStatus.TOO_MANY_REQUESTS) + or is_status_code_server_error(status_code) + ) + + @dataclass() class SitemapUrl: loc: str @@ -470,14 +477,14 @@ async def _fetch_and_process_sitemap( break except Exception as e: - if isinstance(e, HttpStatusCodeError) and not ( - e.status_code in (HTTPStatus.REQUEST_TIMEOUT, HTTPStatus.TOO_MANY_REQUESTS) - or HTTPStatus.INTERNAL_SERVER_ERROR <= e.status_code < _HTTP_STATUS_UPPER_BOUND - ): - raise + if isinstance(e, HttpStatusCodeError) and not _is_retryable_sitemap_status(e.status_code): + logger.warning(f'Skipping sitemap {sitemap_url} due to HTTP status code {e.status_code}.') + break 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 + elif isinstance(e, HttpStatusCodeError): + logger.warning(f'Failed to fetch sitemap {sitemap_url}, no retries left: {e}') else: logger.exception(f'Failed to fetch sitemap {sitemap_url}, no retries left.') raise @@ -553,6 +560,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: @@ -575,6 +584,7 @@ async def parse_sitemap( enqueue_strategy=enqueue_strategy, ): yield result + successful_sources += 1 elif source['type'] == 'url' and 'url' in source: # Add to visited set before processing to avoid duplicates @@ -583,22 +593,30 @@ async def parse_sitemap( visited_sitemap_urls.add(source['url']) - 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, + ): + yield result + successful_sources += 1 + except Exception as e: + source_errors.append(e) + logger.warning(f'Failed to process sitemap source {source["url"]}: {e}') else: logger.warning(f'Invalid source configuration: {source}') + if source_errors and successful_sources == 0: + raise source_errors[-1] + async def _merge_async_generators(*generators: AsyncGenerator) -> AsyncGenerator: queue: asyncio.Queue = asyncio.Queue() diff --git a/src/crawlee/request_loaders/_request_loader.py b/src/crawlee/request_loaders/_request_loader.py index 200339a46d..b13fc6c1ed 100644 --- a/src/crawlee/request_loaders/_request_loader.py +++ b/src/crawlee/request_loaders/_request_loader.py @@ -39,14 +39,14 @@ async def is_empty(self) -> bool: @abstractmethod async def is_finished(self) -> bool: - """Return True if all requests have been handled.""" + """Return True if all requests have been handled, or raise if loading failed after pending requests drain.""" @abstractmethod async def fetch_next_request(self) -> Request | None: """Return the next request to be processed, or `None` if there are no more pending requests. The method should return `None` if and only if `is_finished` would return `True`. In other cases, the method - should wait until a request appears. + should wait until a request appears. It can raise a loading error after all pending requests have been handled. """ @abstractmethod diff --git a/src/crawlee/request_loaders/_sitemap_request_loader.py b/src/crawlee/request_loaders/_sitemap_request_loader.py index 5cb62ec97b..72f7695e46 100644 --- a/src/crawlee/request_loaders/_sitemap_request_loader.py +++ b/src/crawlee/request_loaders/_sitemap_request_loader.py @@ -214,9 +214,11 @@ async def is_empty(self) -> bool: async def is_finished(self) -> bool: """Check if all URLs have been processed.""" state = await self._get_state() + if state.url_queue or state.in_progress: + return False if self._loading_task.done() and not self._loading_task.cancelled(): self._loading_task.result() - return not state.url_queue and len(state.in_progress) == 0 and self._loading_task.done() + return self._loading_task.done() @override async def fetch_next_request(self) -> Request | None: diff --git a/tests/unit/_utils/test_sitemap.py b/tests/unit/_utils/test_sitemap.py index 6291ed8389..ec5cf696e8 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, @@ -21,9 +22,14 @@ discover_valid_sitemaps, parse_sitemap, ) -from crawlee.errors import HttpStatusCodeError 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, + sleep_without_delay, +) if TYPE_CHECKING: from collections.abc import AsyncIterator, Callable @@ -93,30 +99,6 @@ async def read_stream() -> 'AsyncIterator[bytes]': return client, fetched -def _make_status_stream_client(responses: list[tuple[int, bytes]]) -> tuple[AsyncMock, list[int]]: - """Create a mock client returning the provided status and body sequence.""" - attempts: list[int] = [] - - @asynccontextmanager - async def stream(_url: str, **_kwargs: Any) -> 'AsyncIterator[HttpResponse]': - status, body = responses[min(len(attempts), len(responses) - 1)] - 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 - - def compress_gzip(data: str) -> bytes: """Compress a string using gzip.""" return gzip.compress(data.encode()) @@ -364,8 +346,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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) 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)] @@ -374,8 +357,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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) client, attempts = _make_flaky_stream_client(get_basic_sitemap().encode(), fail_times=10) with pytest.raises(ConnectionError): @@ -384,9 +368,10 @@ async def test_sitemap_fetch_raises_after_retries_exhausted() -> None: assert len(attempts) == 3 -async def test_sitemap_fetch_retries_retryable_http_status() -> None: +async def test_sitemap_fetch_retries_retryable_http_status(monkeypatch: pytest.MonkeyPatch) -> None: """Retryable HTTP errors are retried before parsing a successful response.""" - client, attempts = _make_status_stream_client([(503, b''), (503, b''), (200, get_basic_sitemap().encode())]) + monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + client, attempts = make_status_stream_client([(503, b''), (503, b''), (200, get_basic_sitemap().encode())]) items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] @@ -394,24 +379,65 @@ async def test_sitemap_fetch_retries_retryable_http_status() -> None: assert {item.loc for item in items} == get_basic_results() -async def test_sitemap_fetch_rejects_http_error_after_retries_exhausted() -> None: - """A persistent retryable HTTP error is raised once retries are exhausted.""" - client, attempts = _make_status_stream_client([(503, b'')]) +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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + client, attempts = make_status_stream_client([(503, b'')]) - with pytest.raises(HttpStatusCodeError, match='503'): - _ = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] + items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] assert attempts == [503, 503, 503] + assert items == [] async def test_sitemap_fetch_does_not_retry_terminal_http_status() -> None: - """Terminal HTTP errors are raised without parsing their response body or retrying.""" - client, attempts = _make_status_stream_client([(404, get_basic_sitemap().encode())]) + """Terminal HTTP errors are skipped without parsing their response body or retrying.""" + client, attempts = make_status_stream_client([(404, get_basic_sitemap().encode())]) - with pytest.raises(HttpStatusCodeError, match='404'): - _ = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] + items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] assert attempts == [404] + assert items == [] + + +async def test_sitemap_fetch_retries_redirect_then_skips(monkeypatch: pytest.MonkeyPatch) -> None: + """Redirect responses that reach the parser are retried and skipped after exhaustion.""" + monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + client, attempts = make_status_stream_client([(302, get_basic_sitemap().encode())]) + + items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] + + assert attempts == [302, 302, 302] + 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.""" + client, attempts = make_status_stream_client([(200, get_basic_sitemap().encode()), (404, b'')]) + sources: list[SitemapSource] = [ + {'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}, + {'type': 'url', 'url': f'{DEFAULT_URL}missing.xml'}, + ] + + 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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + 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_gzip_bomb_sitemap_truncated_at_size_cap(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/unit/request_loaders/test_sitemap_request_loader.py b/tests/unit/request_loaders/test_sitemap_request_loader.py index 0a1fce2258..0b8f757649 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -9,11 +9,17 @@ from crawlee import RequestOptions, RequestTransformAction from crawlee._utils.sitemap import DEFAULT_MAX_DEPTH -from crawlee.errors import HttpStatusCodeError 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, + sleep_without_delay, +) if TYPE_CHECKING: from collections.abc import AsyncIterator @@ -31,30 +37,6 @@ def encode_base64(data: bytes) -> str: return base64.b64encode(data).decode('utf-8') -def _make_status_stream_client(responses: list[tuple[int, bytes]]) -> tuple[AsyncMock, list[int]]: - """Create a mock client returning the provided status and body sequence.""" - attempts: list[int] = [] - - @asynccontextmanager - async def stream(_url: str, **_kwargs: Any) -> 'AsyncIterator[HttpResponse]': - status, body = responses[min(len(attempts), len(responses) - 1)] - 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 test_nested_sitemap_chain_bounded_by_max_depth() -> None: """A malicious endless chain of unique nested sitemaps is followed only up to the default max depth.""" fetched: list[str] = [] @@ -103,9 +85,10 @@ 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() -> None: +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.""" - client, attempts = _make_status_stream_client([(503, b''), (503, b''), (200, get_basic_sitemap().encode())]) + monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + client, attempts = make_status_stream_client([(503, b''), (503, b''), (200, get_basic_sitemap().encode())]) loader = SitemapRequestLoader([f'{DEFAULT_URL}sitemap.xml'], http_client=client) while not await loader.is_finished(): @@ -117,17 +100,56 @@ async def test_sitemap_http_error_is_retried_before_loading_requests() -> None: assert await loader.get_total_count() == 5 -async def test_sitemap_http_error_is_propagated_after_retries_exhausted() -> None: - """The loader exposes an exhausted sitemap fetch instead of reporting successful completion.""" - client, attempts = _make_status_stream_client([(503, b'')]) +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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + client, attempts = make_status_stream_client([(503, b'')]) loader = SitemapRequestLoader([f'{DEFAULT_URL}sitemap.xml'], http_client=client) - with pytest.raises(HttpStatusCodeError, match='503'): - await loader.fetch_next_request() + assert await loader.fetch_next_request() is None assert attempts == [503, 503, 503] +async def test_sitemap_loader_drains_requests_before_propagating_failure(monkeypatch: pytest.MonkeyPatch) -> None: + """A later sitemap failure is exposed only after requests from healthy sources drain.""" + monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + + @asynccontextmanager + async def stream(url: str, **_kwargs: Any) -> 'AsyncIterator[HttpResponse]': + if url.endswith('broken.xml'): + raise ConnectionError('Network error') + + async def read_stream() -> 'AsyncIterator[bytes]': + yield get_basic_sitemap().encode() + + 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) + + client = AsyncMock(spec=HttpClient) + client.stream = stream + 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._loading_task.done) + with pytest.raises(ConnectionError, match='Network error'): + await loader.is_finished() + + 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..390b901d55 100644 --- a/tests/unit/utils.py +++ b/tests/unit/utils.py @@ -4,20 +4,54 @@ import inspect import sys import time -from typing import TYPE_CHECKING, TypeVar, cast, overload +from asyncio import sleep as asyncio_sleep +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING, Any, TypeVar, cast, overload +from unittest.mock import AsyncMock, MagicMock import pytest if TYPE_CHECKING: - from collections.abc import Awaitable, Callable + from collections.abc import AsyncIterator, Awaitable, Callable from yarl import URL +from crawlee.http_clients._base import HttpClient, HttpResponse + T = TypeVar('T') run_alone_on_mac = pytest.mark.run_alone if sys.platform == 'darwin' else lambda x: x +async def sleep_without_delay(_delay: float) -> None: + """Yield to the event loop without waiting for a requested test delay.""" + await asyncio_sleep(0) + + +def make_status_stream_client(responses: list[tuple[int, bytes]]) -> tuple[AsyncMock, list[int]]: + """Create a mock client returning the provided status and body sequence.""" + attempts: list[int] = [] + + @asynccontextmanager + async def stream(_url: str, **_kwargs: Any) -> AsyncIterator[HttpResponse]: + status, body = responses[min(len(attempts), len(responses) - 1)] + 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. From e82e12d85d217326179f425678251f678b7a8cb0 Mon Sep 17 00:00:00 2001 From: Anas Khan <83116240+anxkhn@users.noreply.github.com> Date: Thu, 13 Aug 2026 12:51:41 +0530 Subject: [PATCH 4/5] fix(sitemap): skip non-2xx sources and keep is_finished a predicate Validate sitemap HTTP status before parsing. Retry 408 and 5xx, skip other non-2xx without raising, and raise the first real fetch error only when no source parsed successfully. Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com> --- src/crawlee/_utils/sitemap.py | 60 ++++++---- .../request_loaders/_request_loader.py | 4 +- .../_sitemap_request_loader.py | 6 +- tests/unit/_utils/test_sitemap.py | 106 ++++++++++++++---- .../test_sitemap_request_loader.py | 51 ++++++--- tests/unit/utils.py | 28 +++-- 6 files changed, 181 insertions(+), 74 deletions(-) diff --git a/src/crawlee/_utils/sitemap.py b/src/crawlee/_utils/sitemap.py index f97fdfb3a8..0bb6785a70 100644 --- a/src/crawlee/_utils/sitemap.py +++ b/src/crawlee/_utils/sitemap.py @@ -21,7 +21,7 @@ from crawlee._utils.urls import filter_url from crawlee._utils.web import is_status_code_server_error, is_status_code_successful -from crawlee.errors import HttpStatusCodeError, ProxyError +from crawlee.errors import ProxyError if TYPE_CHECKING: from collections.abc import AsyncGenerator @@ -47,20 +47,23 @@ 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 _raise_for_sitemap_status(status_code: int) -> None: - """Raise `HttpStatusCodeError` if the sitemap response status is not 2xx.""" - if not HTTPStatus.OK <= status_code < HTTPStatus.MULTIPLE_CHOICES: - raise HttpStatusCodeError('Error status code returned while fetching sitemap', status_code) + +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 def _is_retryable_sitemap_status(status_code: int) -> bool: """Return whether a sitemap response status should be retried.""" - return ( - HTTPStatus.MULTIPLE_CHOICES <= status_code < HTTPStatus.BAD_REQUEST - or status_code in (HTTPStatus.REQUEST_TIMEOUT, HTTPStatus.TOO_MANY_REQUESTS) - or is_status_code_server_error(status_code) - ) + return status_code == HTTPStatus.REQUEST_TIMEOUT or is_status_code_server_error(status_code) + + +async def _sleep_before_sitemap_retry() -> None: + """Pause briefly before retrying a failed sitemap fetch.""" + await asyncio.sleep(SITEMAP_RETRY_DELAY) @dataclass() @@ -379,6 +382,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: @@ -392,7 +396,22 @@ 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: - _raise_for_sitemap_status(response.status_code) + status_code = response.status_code + if not _is_successful_sitemap_status(status_code): + if not _is_retryable_sitemap_status(status_code): + logger.warning(f'Skipping sitemap {sitemap_url} due to HTTP status code {status_code}.') + return + if retries_left > 0: + logger.warning( + f'Error fetching sitemap {sitemap_url}: HTTP status {status_code}. ' + f'Retries left: {retries_left}' + ) + await _sleep_before_sitemap_retry() + else: + logger.warning( + f'Failed to fetch sitemap {sitemap_url}, no retries left: HTTP status {status_code}' + ) + continue # Determine content type and compression content_type = response.headers.get('content-type', '') @@ -474,20 +493,17 @@ async def _fetch_and_process_sitemap( yield result finally: parser.close() - break + parsed_ok.append(True) except Exception as e: - if isinstance(e, HttpStatusCodeError) and not _is_retryable_sitemap_status(e.status_code): - logger.warning(f'Skipping sitemap {sitemap_url} due to HTTP status code {e.status_code}.') - break 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 - elif isinstance(e, HttpStatusCodeError): - logger.warning(f'Failed to fetch sitemap {sitemap_url}, no retries left: {e}') + await _sleep_before_sitemap_retry() else: logger.exception(f'Failed to fetch sitemap {sitemap_url}, no retries left.') raise + else: + return class Sitemap: @@ -584,7 +600,6 @@ async def parse_sitemap( enqueue_strategy=enqueue_strategy, ): yield result - successful_sources += 1 elif source['type'] == 'url' and 'url' in source: # Add to visited set before processing to avoid duplicates @@ -592,6 +607,7 @@ 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] = [] try: async for result in _fetch_and_process_sitemap( @@ -605,9 +621,11 @@ async def parse_sitemap( enqueue_strategy=enqueue_strategy, proxy_info=proxy_info, timeout=timeout, + parsed_ok=parsed_ok, ): yield result - successful_sources += 1 + if parsed_ok: + successful_sources += 1 except Exception as e: source_errors.append(e) logger.warning(f'Failed to process sitemap source {source["url"]}: {e}') @@ -615,7 +633,7 @@ async def parse_sitemap( logger.warning(f'Invalid source configuration: {source}') if source_errors and successful_sources == 0: - raise source_errors[-1] + raise source_errors[0] async def _merge_async_generators(*generators: AsyncGenerator) -> AsyncGenerator: diff --git a/src/crawlee/request_loaders/_request_loader.py b/src/crawlee/request_loaders/_request_loader.py index b13fc6c1ed..200339a46d 100644 --- a/src/crawlee/request_loaders/_request_loader.py +++ b/src/crawlee/request_loaders/_request_loader.py @@ -39,14 +39,14 @@ async def is_empty(self) -> bool: @abstractmethod async def is_finished(self) -> bool: - """Return True if all requests have been handled, or raise if loading failed after pending requests drain.""" + """Return True if all requests have been handled.""" @abstractmethod async def fetch_next_request(self) -> Request | None: """Return the next request to be processed, or `None` if there are no more pending requests. The method should return `None` if and only if `is_finished` would return `True`. In other cases, the method - should wait until a request appears. It can raise a loading error after all pending requests have been handled. + should wait until a request appears. """ @abstractmethod diff --git a/src/crawlee/request_loaders/_sitemap_request_loader.py b/src/crawlee/request_loaders/_sitemap_request_loader.py index 72f7695e46..230c50affc 100644 --- a/src/crawlee/request_loaders/_sitemap_request_loader.py +++ b/src/crawlee/request_loaders/_sitemap_request_loader.py @@ -214,11 +214,7 @@ async def is_empty(self) -> bool: async def is_finished(self) -> bool: """Check if all URLs have been processed.""" state = await self._get_state() - if state.url_queue or state.in_progress: - return False - if self._loading_task.done() and not self._loading_task.cancelled(): - self._loading_task.result() - return self._loading_task.done() + return not state.url_queue and len(state.in_progress) == 0 and self._loading_task.done() @override async def fetch_next_request(self) -> Request | None: diff --git a/tests/unit/_utils/test_sitemap.py b/tests/unit/_utils/test_sitemap.py index ec5cf696e8..b4db9628f7 100644 --- a/tests/unit/_utils/test_sitemap.py +++ b/tests/unit/_utils/test_sitemap.py @@ -28,7 +28,6 @@ get_basic_results, get_basic_sitemap, make_status_stream_client, - sleep_without_delay, ) if TYPE_CHECKING: @@ -348,7 +347,7 @@ async def test_malformed_sitemap_keeps_urls() -> 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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + monkeypatch.setattr('crawlee._utils.sitemap._sleep_before_sitemap_retry', AsyncMock()) 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)] @@ -359,7 +358,7 @@ async def test_sitemap_fetch_retries_on_transient_error(monkeypatch: pytest.Monk 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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + monkeypatch.setattr('crawlee._utils.sitemap._sleep_before_sitemap_retry', AsyncMock()) client, attempts = _make_flaky_stream_client(get_basic_sitemap().encode(), fail_times=10) with pytest.raises(ConnectionError): @@ -370,10 +369,13 @@ async def test_sitemap_fetch_raises_after_retries_exhausted(monkeypatch: pytest. 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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) - client, attempts = make_status_stream_client([(503, b''), (503, b''), (200, get_basic_sitemap().encode())]) + monkeypatch.setattr('crawlee._utils.sitemap._sleep_before_sitemap_retry', AsyncMock()) + 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': f'{DEFAULT_URL}sitemap.xml'}], client)] + 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() @@ -381,10 +383,11 @@ async def test_sitemap_fetch_retries_retryable_http_status(monkeypatch: pytest.M 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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) - client, attempts = make_status_stream_client([(503, b'')]) + monkeypatch.setattr('crawlee._utils.sitemap._sleep_before_sitemap_retry', AsyncMock()) + 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': f'{DEFAULT_URL}sitemap.xml'}], client)] + items = [item async for item in parse_sitemap([{'type': 'url', 'url': sitemap_url}], client)] assert attempts == [503, 503, 503] assert items == [] @@ -392,31 +395,50 @@ async def test_sitemap_fetch_skips_http_error_after_retries_exhausted(monkeypatc async def test_sitemap_fetch_does_not_retry_terminal_http_status() -> None: """Terminal HTTP errors are skipped without parsing their response body or retrying.""" - client, attempts = make_status_stream_client([(404, get_basic_sitemap().encode())]) + 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': f'{DEFAULT_URL}sitemap.xml'}], client)] + items = [item async for item in parse_sitemap([{'type': 'url', 'url': sitemap_url}], client)] assert attempts == [404] assert items == [] -async def test_sitemap_fetch_retries_redirect_then_skips(monkeypatch: pytest.MonkeyPatch) -> None: - """Redirect responses that reach the parser are retried and skipped after exhaustion.""" - monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) - client, attempts = make_status_stream_client([(302, get_basic_sitemap().encode())]) +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 == [] - items = [item async for item in parse_sitemap([{'type': 'url', 'url': f'{DEFAULT_URL}sitemap.xml'}], client)] - assert attempts == [302, 302, 302] +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.""" - client, attempts = make_status_stream_client([(200, get_basic_sitemap().encode()), (404, b'')]) + 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': f'{DEFAULT_URL}sitemap.xml'}, - {'type': 'url', 'url': f'{DEFAULT_URL}missing.xml'}, + {'type': 'url', 'url': healthy_url}, + {'type': 'url', 'url': missing_url}, ] items = [item async for item in parse_sitemap(sources, client)] @@ -427,7 +449,7 @@ async def test_sitemap_partial_http_failure_keeps_healthy_source() -> None: 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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) + monkeypatch.setattr('crawlee._utils.sitemap._sleep_before_sitemap_retry', AsyncMock()) client, attempts = _make_flaky_stream_client(get_basic_sitemap().encode(), fail_times=3) sources: list[SitemapSource] = [ {'type': 'url', 'url': f'{DEFAULT_URL}broken.xml'}, @@ -440,6 +462,48 @@ async def test_sitemap_partial_fetch_failure_keeps_healthy_source(monkeypatch: p 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._sleep_before_sitemap_retry', AsyncMock()) + 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._sleep_before_sitemap_retry', AsyncMock()) + 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) diff --git a/tests/unit/request_loaders/test_sitemap_request_loader.py b/tests/unit/request_loaders/test_sitemap_request_loader.py index 0b8f757649..e7fd6deec1 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -8,7 +8,9 @@ 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 @@ -18,7 +20,6 @@ get_basic_sitemap, make_status_stream_client, poll_until_condition, - sleep_without_delay, ) if TYPE_CHECKING: @@ -87,9 +88,12 @@ async def test_sitemap_traversal(server_url: URL, http_client: HttpClient) -> No 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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) - client, attempts = make_status_stream_client([(503, b''), (503, b''), (200, get_basic_sitemap().encode())]) - loader = SitemapRequestLoader([f'{DEFAULT_URL}sitemap.xml'], http_client=client) + monkeypatch.setattr('crawlee._utils.sitemap._sleep_before_sitemap_retry', AsyncMock()) + 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() @@ -102,18 +106,19 @@ async def test_sitemap_http_error_is_retried_before_loading_requests(monkeypatch 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.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) - client, attempts = make_status_stream_client([(503, b'')]) - loader = SitemapRequestLoader([f'{DEFAULT_URL}sitemap.xml'], http_client=client) + monkeypatch.setattr('crawlee._utils.sitemap._sleep_before_sitemap_retry', AsyncMock()) + 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_before_propagating_failure(monkeypatch: pytest.MonkeyPatch) -> None: - """A later sitemap failure is exposed only after requests from healthy sources drain.""" - monkeypatch.setattr('crawlee._utils.sitemap.asyncio.sleep', AsyncMock(side_effect=sleep_without_delay)) +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._sleep_before_sitemap_retry', AsyncMock()) @asynccontextmanager async def stream(url: str, **_kwargs: Any) -> 'AsyncIterator[HttpResponse]': @@ -145,9 +150,29 @@ async def read_stream() -> 'AsyncIterator[bytes]': for request in requests: await loader.mark_request_as_handled(request) - assert await poll_until_condition(loader._loading_task.done) - with pytest.raises(ConnectionError, match='Network error'): - await loader.is_finished() + assert await poll_until_condition(loader.is_finished) + + +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._sleep_before_sitemap_retry', AsyncMock()) + 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: diff --git a/tests/unit/utils.py b/tests/unit/utils.py index 390b901d55..83c0563333 100644 --- a/tests/unit/utils.py +++ b/tests/unit/utils.py @@ -4,37 +4,41 @@ import inspect import sys import time -from asyncio import sleep as asyncio_sleep 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 AsyncIterator, Awaitable, Callable from yarl import URL -from crawlee.http_clients._base import HttpClient, HttpResponse - T = TypeVar('T') run_alone_on_mac = pytest.mark.run_alone if sys.platform == 'darwin' else lambda x: x -async def sleep_without_delay(_delay: float) -> None: - """Yield to the event loop without waiting for a requested test delay.""" - await asyncio_sleep(0) - - -def make_status_stream_client(responses: list[tuple[int, bytes]]) -> tuple[AsyncMock, list[int]]: - """Create a mock client returning the provided status and body sequence.""" +def make_status_stream_client( + responses: dict[str, list[tuple[int, bytes] | BaseException]], +) -> tuple[AsyncMock, list[int]]: + """Create a mock client returning the provided per-URL status and body sequence.""" attempts: list[int] = [] + indexes: dict[str, int] = {} @asynccontextmanager - async def stream(_url: str, **_kwargs: Any) -> AsyncIterator[HttpResponse]: - status, body = responses[min(len(attempts), len(responses) - 1)] + 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, BaseException): + raise spec + + status, body = spec attempts.append(status) async def read_stream() -> AsyncIterator[bytes]: From 086e878e22d7184a23fdc7c7d2461a9a0fae8677 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 14 Aug 2026 11:04:55 +0200 Subject: [PATCH 5/5] fix(sitemap): contain per-source failures during parsing and loading --- src/crawlee/_utils/sitemap.py | 55 ++++++---- .../_sitemap_request_loader.py | 101 ++++++++++-------- tests/unit/_utils/test_sitemap.py | 29 +++-- .../test_sitemap_request_loader.py | 54 ++++++---- tests/unit/utils.py | 10 +- 5 files changed, 149 insertions(+), 100 deletions(-) diff --git a/src/crawlee/_utils/sitemap.py b/src/crawlee/_utils/sitemap.py index 0bb6785a70..7e0d339bb3 100644 --- a/src/crawlee/_utils/sitemap.py +++ b/src/crawlee/_utils/sitemap.py @@ -56,14 +56,21 @@ def _is_successful_sitemap_status(status_code: int) -> bool: return HTTPStatus.OK <= status_code < HTTPStatus.MULTIPLE_CHOICES -def _is_retryable_sitemap_status(status_code: int) -> bool: - """Return whether a sitemap response status should be retried.""" - return status_code == HTTPStatus.REQUEST_TIMEOUT or is_status_code_server_error(status_code) +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 -async def _sleep_before_sitemap_retry() -> None: - """Pause briefly before retrying a failed sitemap fetch.""" - await asyncio.sleep(SITEMAP_RETRY_DELAY) +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() @@ -398,20 +405,10 @@ async def _fetch_and_process_sitemap( ) as response: status_code = response.status_code if not _is_successful_sitemap_status(status_code): - if not _is_retryable_sitemap_status(status_code): - logger.warning(f'Skipping sitemap {sitemap_url} due to HTTP status code {status_code}.') - return - if retries_left > 0: - logger.warning( - f'Error fetching sitemap {sitemap_url}: HTTP status {status_code}. ' - f'Retries left: {retries_left}' - ) - await _sleep_before_sitemap_retry() - else: - logger.warning( - f'Failed to fetch sitemap {sitemap_url}, no retries left: HTTP status {status_code}' - ) - continue + # 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', '') @@ -495,10 +492,18 @@ async def _fetch_and_process_sitemap( parser.close() 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 _sleep_before_sitemap_retry() + await asyncio.sleep(SITEMAP_RETRY_DELAY) else: logger.exception(f'Failed to fetch sitemap {sitemap_url}, no retries left.') raise @@ -564,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 {} @@ -600,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 @@ -627,8 +638,8 @@ async def parse_sitemap( 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) - logger.warning(f'Failed to process sitemap source {source["url"]}: {e}') else: logger.warning(f'Invalid source configuration: {source}') 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 b4db9628f7..e88cca32d4 100644 --- a/tests/unit/_utils/test_sitemap.py +++ b/tests/unit/_utils/test_sitemap.py @@ -347,7 +347,7 @@ async def test_malformed_sitemap_keeps_urls() -> 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._sleep_before_sitemap_retry', AsyncMock()) + 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)] @@ -358,7 +358,7 @@ async def test_sitemap_fetch_retries_on_transient_error(monkeypatch: pytest.Monk 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._sleep_before_sitemap_retry', AsyncMock()) + 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): @@ -369,7 +369,7 @@ async def test_sitemap_fetch_raises_after_retries_exhausted(monkeypatch: pytest. 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._sleep_before_sitemap_retry', AsyncMock()) + 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())]} @@ -383,7 +383,7 @@ async def test_sitemap_fetch_retries_retryable_http_status(monkeypatch: pytest.M 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._sleep_before_sitemap_retry', AsyncMock()) + 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'')]}) @@ -449,7 +449,7 @@ async def test_sitemap_partial_http_failure_keeps_healthy_source() -> None: 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._sleep_before_sitemap_retry', AsyncMock()) + 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'}, @@ -462,9 +462,24 @@ async def test_sitemap_partial_fetch_failure_keeps_healthy_source(monkeypatch: p 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._sleep_before_sitemap_retry', AsyncMock()) + 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( @@ -486,7 +501,7 @@ async def test_sitemap_skipped_source_does_not_hide_fetch_error(monkeypatch: pyt 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._sleep_before_sitemap_retry', AsyncMock()) + 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( diff --git a/tests/unit/request_loaders/test_sitemap_request_loader.py b/tests/unit/request_loaders/test_sitemap_request_loader.py index e7fd6deec1..e39e2a04fb 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -88,7 +88,7 @@ async def test_sitemap_traversal(server_url: URL, http_client: HttpClient) -> No 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._sleep_before_sitemap_retry', AsyncMock()) + 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())]} @@ -106,7 +106,7 @@ async def test_sitemap_http_error_is_retried_before_loading_requests(monkeypatch 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._sleep_before_sitemap_retry', AsyncMock()) + 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) @@ -118,24 +118,13 @@ async def test_sitemap_http_error_is_skipped_after_retries_exhausted(monkeypatch 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._sleep_before_sitemap_retry', AsyncMock()) - - @asynccontextmanager - async def stream(url: str, **_kwargs: Any) -> 'AsyncIterator[HttpResponse]': - if url.endswith('broken.xml'): - raise ConnectionError('Network error') - - async def read_stream() -> 'AsyncIterator[bytes]': - yield get_basic_sitemap().encode() - - 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) - - client = AsyncMock(spec=HttpClient) - client.stream = stream + 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 ) @@ -153,9 +142,32 @@ async def read_stream() -> 'AsyncIterator[bytes]': 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._sleep_before_sitemap_retry', AsyncMock()) + 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) diff --git a/tests/unit/utils.py b/tests/unit/utils.py index 83c0563333..e1875482d2 100644 --- a/tests/unit/utils.py +++ b/tests/unit/utils.py @@ -23,9 +23,13 @@ def make_status_stream_client( - responses: dict[str, list[tuple[int, bytes] | BaseException]], + responses: dict[str, list[tuple[int, bytes] | Exception]], ) -> tuple[AsyncMock, list[int]]: - """Create a mock client returning the provided per-URL status and body sequence.""" + """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] = {} @@ -35,7 +39,7 @@ async def stream(url: str, **_kwargs: Any) -> AsyncIterator[HttpResponse]: index = indexes.get(url, 0) indexes[url] = index + 1 spec = sequence[min(index, len(sequence) - 1)] - if isinstance(spec, BaseException): + if isinstance(spec, Exception): raise spec status, body = spec