From a90325e8fef62e1477c6346cd0408a7aaa062742 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/3] 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 78f8a867e9909921737d4545dfec139e43d2e7d6 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/3] 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 9c870a55a9998deb21a42a54f5c1dcc99daa9ed0 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/3] 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.