From 0a424636e9de7e97a21a965fe71da52640fc5d6c Mon Sep 17 00:00:00 2001 From: Gil Desmarais Date: Fri, 28 Aug 2026 23:29:33 +0200 Subject: [PATCH 1/2] fix(engine): prefer challenge_block over timeout false-hope When a nav/timeout exception leaves an inspectable challenge page, classify as challenge_block (no timeout_phase) instead of burning the budget into timeout/work. Hard HTTP blocks abort remaining strategies; soft challenge markers may still retry. --- app/engine/browser_tier.py | 162 ++++++++++++----- tests/engine/test_browser_tier_challenge.py | 192 ++++++++++++++++++++ 2 files changed, 313 insertions(+), 41 deletions(-) create mode 100644 tests/engine/test_browser_tier_challenge.py diff --git a/app/engine/browser_tier.py b/app/engine/browser_tier.py index f87629b..b8fb392 100644 --- a/app/engine/browser_tier.py +++ b/app/engine/browser_tier.py @@ -30,7 +30,7 @@ from app.infra.scrape_progress import ScrapeProgress from app.infra.xhr_collector import XhrCollector from app.logging_config import get_logger -from app.schemas.enums import ErrorCategory, ExecutionTier, TimeoutPhase +from app.schemas.enums import ErrorCategory, ExecutionTier, NavigationMode, TimeoutPhase from app.schemas.request import ScrapeRequest from app.schemas.response import ScrapeError, ScrapeSuccess, XhrResponse @@ -66,16 +66,58 @@ def settle_page_state( return html, meta, assessment, xhr_responses +def _inspect_assessment( + driver: DriverProtocol, target_url: str +) -> ChallengeAssessment | None: + """Best-effort challenge assessment after a nav/timeout exception.""" + try: + return _page_state(driver, target_url)[2] + except Exception: + return None + + +def _challenge_block_error( + target_url: str, + *, + request_id: str, + attempts: int, + strategy: NavigationMode | None, + render_ms: int, + assessment: ChallengeAssessment, +) -> ScrapeError: + marker = assessment.detected_marker or "unknown" + return build_error( + target_url, + f"Bot challenge detected ({marker})", + request_id=request_id, + error_category=ErrorCategory.CHALLENGE_BLOCK, + attempts=attempts, + strategy_used=strategy, + render_ms=render_ms, + execution_tier=ExecutionTier.BROWSER_DRIVER, + assessment=assessment, + timeout_phase=None, + ) + + +def _should_retry_challenge( + assessment: ChallengeAssessment, *, has_more_strategies: bool +) -> bool: + """Soft challenge markers may retry strategies; hard HTTP blocks do not.""" + return assessment.challenge_detected and has_more_strategies + + def _boot_storage_error( target_url: str, request_id: str, started_monotonic: float, exc: OSError, ) -> ScrapeError: - if exc.errno == errno.ENOSPC: - detail = "Scrape runtime storage full" - else: - detail = "Scrape runtime storage unavailable" + detail = ( + "Scrape runtime storage full" + if exc.errno == errno.ENOSPC + else "Scrape runtime storage unavailable" + ) return build_error( target_url, f"{detail}: {exc}", @@ -88,6 +130,52 @@ def _boot_storage_error( ) +def _boot_driver_error( + target_url: str, + request_id: str, + started_monotonic: float, + exc: Exception, +) -> ScrapeError: + is_timeout = is_timeout_exception(exc) + return build_error( + target_url, + str(exc), + request_id=request_id, + attempts=0, + render_ms=elapsed_ms(started_monotonic), + error_category=( + ErrorCategory.TIMEOUT if is_timeout else ErrorCategory.NAVIGATION_ERROR + ), + execution_tier=ExecutionTier.BROWSER_DRIVER, + timeout_phase=TimeoutPhase.BOOT, + ) + + +def _work_exception_error( + target_url: str, + *, + request_id: str, + attempts: int, + strategy: NavigationMode, + started_monotonic: float, + exc: Exception, +) -> ScrapeError: + is_timeout = is_timeout_exception(exc) + return build_error( + target_url, + str(exc), + request_id=request_id, + attempts=attempts, + strategy_used=strategy, + render_ms=elapsed_ms(started_monotonic), + error_category=( + ErrorCategory.TIMEOUT if is_timeout else ErrorCategory.NAVIGATION_ERROR + ), + execution_tier=ExecutionTier.BROWSER_DRIVER, + timeout_phase=TimeoutPhase.WORK if is_timeout else None, + ) + + def run_browser_tier( payload: ScrapeRequest, session: ScrapeSession, @@ -153,21 +241,7 @@ def run_browser_tier( ), ) except Exception as exc: - is_timeout = is_timeout_exception(exc) - return build_error( - target_url, - str(exc), - request_id=request_id, - attempts=0, - render_ms=elapsed_ms(started_monotonic), - error_category=( - ErrorCategory.TIMEOUT - if is_timeout - else ErrorCategory.NAVIGATION_ERROR - ), - execution_tier=ExecutionTier.BROWSER_DRIVER, - timeout_phase=TimeoutPhase.BOOT, - ) + return _boot_driver_error(target_url, request_id, started_monotonic, exc) session.driver = driver @@ -196,6 +270,7 @@ def run_browser_tier( for attempt_index, strategy in enumerate(strategies, start=1): attempts = attempt_index + has_more = attempt_index < len(strategies) progress.mark( TimeoutPhase.WORK, attempts=attempts, @@ -222,26 +297,24 @@ def run_browser_tier( if not assessment.is_clean: logger.warning( - "scrape_challenge_detected request_id=%s host=%s strategy=%s attempt=%d marker=%s", + "scrape_challenge_detected request_id=%s host=%s strategy=%s " + "attempt=%d marker=%s", request_id, urlparse(target_url).hostname, strategy.value, attempt_index, assessment.detected_marker, ) - if attempt_index < len(strategies): + if _should_retry_challenge(assessment, has_more_strategies=has_more): collector.reset() continue - return build_error( + return _challenge_block_error( target_url, - f"Bot challenge detected ({assessment.detected_marker or 'unknown'})", request_id=request_id, - error_category=ErrorCategory.CHALLENGE_BLOCK, attempts=attempts, - strategy_used=strategy, + strategy=strategy, render_ms=elapsed_ms(started_monotonic), - execution_tier=ExecutionTier.BROWSER_DRIVER, assessment=assessment, ) @@ -262,7 +335,8 @@ def run_browser_tier( ) except Exception as exc: logger.warning( - "scrape_attempt_failed request_id=%s host=%s mode=%s strategy=%s attempt=%d error=%s", + "scrape_attempt_failed request_id=%s host=%s mode=%s strategy=%s " + "attempt=%d error=%s", request_id, urlparse(target_url).hostname, payload.navigation_mode, @@ -270,25 +344,31 @@ def run_browser_tier( attempt_index, str(exc), ) - if attempt_index < len(strategies): + # Prefer knowable challenge over timeout when the page is inspectable. + assessment = _inspect_assessment(driver, target_url) + if assessment is not None and not assessment.is_clean: + if _should_retry_challenge(assessment, has_more_strategies=has_more): + collector.reset() + continue + return _challenge_block_error( + target_url, + request_id=request_id, + attempts=attempts, + strategy=strategy, + render_ms=elapsed_ms(started_monotonic), + assessment=assessment, + ) + if has_more: collector.reset() continue - is_timeout = is_timeout_exception(exc) - return build_error( + return _work_exception_error( target_url, - str(exc), request_id=request_id, attempts=attempts, - strategy_used=strategy, - render_ms=elapsed_ms(started_monotonic), - error_category=( - ErrorCategory.TIMEOUT - if is_timeout - else ErrorCategory.NAVIGATION_ERROR - ), - execution_tier=ExecutionTier.BROWSER_DRIVER, - timeout_phase=TimeoutPhase.WORK if is_timeout else None, + strategy=strategy, + started_monotonic=started_monotonic, + exc=exc, ) return build_error( diff --git a/tests/engine/test_browser_tier_challenge.py b/tests/engine/test_browser_tier_challenge.py new file mode 100644 index 0000000..4e2e6a8 --- /dev/null +++ b/tests/engine/test_browser_tier_challenge.py @@ -0,0 +1,192 @@ +"""Browser-tier challenge-before-timeout and hard-block abort.""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from typing import Any, ClassVar +from unittest.mock import patch + +from app.config import get_settings +from app.engine import ScraperEngine +from app.schemas.enums import ErrorCategory, ExecutionMode, NavigationMode, TimeoutPhase +from app.schemas.response import ScrapeError +from tests.support.factories import scrape_request +from tests.support.fakes import FakeDriver + + +class _PassiveRequest: + def __init__(self, status: int, url: str) -> None: + self.url = url + self.response = type( + "Resp", + (), + {"status_code": status, "headers": {"content-type": "text/html"}}, + )() + + +class _ChallengeHtmlDriver(FakeDriver): + """Timeout on navigate while page already shows a challenge interstitial.""" + + navigate_calls: ClassVar[int] = 0 + + def __init__(self, *args: object, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.page_html = "Just a moment..." + self.current_url = "https://example.com/" + + @classmethod + def reset(cls) -> None: + cls.navigate_calls = 0 + + def get(self, *_args: object, **_kwargs: Any) -> None: + type(self).navigate_calls += 1 + raise TimeoutError("Navigation timeout after 15s") + + def google_get(self, *_args: object, **_kwargs: Any) -> None: + type(self).navigate_calls += 1 + raise TimeoutError("Navigation timeout after 15s") + + +class _CleanTimeoutDriver(FakeDriver): + """Timeout on navigate with a clean page body.""" + + def __init__(self, *args: object, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.page_html = "

Example Domain

" + self.current_url = "https://example.com/" + + def get(self, *_args: object, **_kwargs: Any) -> None: + raise TimeoutError("Navigation timeout after 15s") + + def google_get(self, *_args: object, **_kwargs: Any) -> None: + raise TimeoutError("Navigation timeout after 15s") + + +class _HardBlockDriver(FakeDriver): + """HTTP hard block without challenge marker — must not burn more strategies.""" + + navigate_calls: ClassVar[int] = 0 + + def __init__(self, *args: object, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.page_html = "

Forbidden

" + self.current_url = "https://example.com/" + self.requests = [_PassiveRequest(403, "https://example.com/")] + + @classmethod + def reset(cls) -> None: + cls.navigate_calls = 0 + + def get(self, *_args: object, **_kwargs: Any) -> None: + type(self).navigate_calls += 1 + + def google_get(self, *_args: object, **_kwargs: Any) -> None: + type(self).navigate_calls += 1 + + +class _SoftChallengeDriver(FakeDriver): + """Soft challenge marker — may retry remaining strategies.""" + + navigate_calls: ClassVar[int] = 0 + + def __init__(self, *args: object, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.page_html = "Just a moment..." + self.current_url = "https://example.com/" + + @classmethod + def reset(cls) -> None: + cls.navigate_calls = 0 + + def get(self, *_args: object, **_kwargs: Any) -> None: + type(self).navigate_calls += 1 + + def google_get(self, *_args: object, **_kwargs: Any) -> None: + type(self).navigate_calls += 1 + + +class BrowserTierChallengeTests(unittest.TestCase): + def test_timeout_with_challenge_html_returns_challenge_block(self) -> None: + _ChallengeHtmlDriver.reset() + payload = scrape_request( + execution_mode=ExecutionMode.BROWSER, + navigation_mode=NavigationMode.GET, + max_retries=0, + ) + + with tempfile.TemporaryDirectory() as tmp: + engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) + with patch("botasaurus.browser.Driver", _ChallengeHtmlDriver): + result = engine.execute(payload, request_id="req-timeout-challenge") + + self.assertIsInstance(result, ScrapeError) + assert isinstance(result, ScrapeError) + self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) + self.assertIsNone(result.diagnostics.timeout_phase) + self.assertIsNotNone(result.diagnostics.challenge) + assert result.diagnostics.challenge is not None + self.assertTrue(result.diagnostics.challenge.detected) + self.assertEqual(_ChallengeHtmlDriver.navigate_calls, 1) + + def test_timeout_with_clean_page_returns_timeout_work(self) -> None: + payload = scrape_request( + execution_mode=ExecutionMode.BROWSER, + navigation_mode=NavigationMode.GET, + max_retries=0, + ) + + with tempfile.TemporaryDirectory() as tmp: + engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) + with patch("botasaurus.browser.Driver", _CleanTimeoutDriver): + result = engine.execute(payload, request_id="req-timeout-clean") + + self.assertIsInstance(result, ScrapeError) + assert isinstance(result, ScrapeError) + self.assertEqual(result.error_category, ErrorCategory.TIMEOUT) + self.assertEqual(result.diagnostics.timeout_phase, TimeoutPhase.WORK) + + def test_hard_block_aborts_remaining_strategies(self) -> None: + _HardBlockDriver.reset() + payload = scrape_request( + execution_mode=ExecutionMode.BROWSER, + navigation_mode=NavigationMode.AUTO, + max_retries=2, + ) + + with tempfile.TemporaryDirectory() as tmp: + engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) + with patch("botasaurus.browser.Driver", _HardBlockDriver): + result = engine.execute(payload, request_id="req-hard-block") + + self.assertIsInstance(result, ScrapeError) + assert isinstance(result, ScrapeError) + self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) + self.assertIsNone(result.diagnostics.timeout_phase) + self.assertEqual(result.diagnostics.attempts, 1) + self.assertEqual(_HardBlockDriver.navigate_calls, 1) + assert result.diagnostics.challenge is not None + self.assertTrue(result.diagnostics.challenge.blocked) + self.assertFalse(result.diagnostics.challenge.detected) + + def test_soft_challenge_retries_strategies_then_blocks(self) -> None: + _SoftChallengeDriver.reset() + payload = scrape_request( + execution_mode=ExecutionMode.BROWSER, + navigation_mode=NavigationMode.AUTO, + max_retries=2, + ) + + with tempfile.TemporaryDirectory() as tmp: + engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) + with patch("botasaurus.browser.Driver", _SoftChallengeDriver): + result = engine.execute(payload, request_id="req-soft-challenge") + + self.assertIsInstance(result, ScrapeError) + assert isinstance(result, ScrapeError) + self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) + self.assertEqual(result.diagnostics.attempts, 3) + self.assertEqual(_SoftChallengeDriver.navigate_calls, 3) + assert result.diagnostics.challenge is not None + self.assertTrue(result.diagnostics.challenge.detected) From 3dde3aa8956087aa8fc7e0136316221ec32c11b4 Mon Sep 17 00:00:00 2001 From: Gil Desmarais Date: Fri, 28 Aug 2026 23:44:18 +0200 Subject: [PATCH 2/2] refactor(engine): harden challenge-before-timeout inspect path Keep ChallengeAssessment as soft/hard retry owner; lean post-exception inspect skips CDP so hung pages still classify without discarding HTML signals. Shared unclean disposition + thinner challenge fixtures. --- app/engine/browser_tier.py | 115 +++++++++------ app/infra/detector.py | 4 + tests/engine/test_browser_tier_challenge.py | 151 ++++++++------------ tests/infra/test_challenge_detector.py | 10 ++ 4 files changed, 141 insertions(+), 139 deletions(-) diff --git a/app/engine/browser_tier.py b/app/engine/browser_tier.py index b8fb392..fcc6100 100644 --- a/app/engine/browser_tier.py +++ b/app/engine/browser_tier.py @@ -67,11 +67,24 @@ def settle_page_state( def _inspect_assessment( - driver: DriverProtocol, target_url: str + driver: DriverProtocol, _target_url: str ) -> ChallengeAssessment | None: - """Best-effort challenge assessment after a nav/timeout exception.""" + """Best-effort challenge assessment after a nav/timeout exception. + + Lean path only: html + passive request status + driver signals via + ChallengeDetector. Skips CDP log walks that can stall on a hung page. + Metadata failures must not discard HTML/driver challenge signals. + """ + try: + html = driver.page_html or "" + except Exception: + return None + try: + status_code, _, _ = MetadataExtractor.extract_from_requests(driver) + except Exception: + status_code = None try: - return _page_state(driver, target_url)[2] + return ChallengeDetector.detect(html, status_code, driver=driver) except Exception: return None @@ -100,11 +113,29 @@ def _challenge_block_error( ) -def _should_retry_challenge( - assessment: ChallengeAssessment, *, has_more_strategies: bool -) -> bool: - """Soft challenge markers may retry strategies; hard HTTP blocks do not.""" - return assessment.challenge_detected and has_more_strategies +def _unclean_retry_or_block( + target_url: str, + *, + request_id: str, + attempts: int, + strategy: NavigationMode, + started_monotonic: float, + assessment: ChallengeAssessment, + has_more_strategies: bool, + collector: XhrCollector, +) -> ScrapeError | None: + """Retry soft challenges; otherwise return challenge_block. None ⇒ continue.""" + if assessment.may_retry_strategies(has_more=has_more_strategies): + collector.reset() + return None + return _challenge_block_error( + target_url, + request_id=request_id, + attempts=attempts, + strategy=strategy, + render_ms=elapsed_ms(started_monotonic), + assessment=assessment, + ) def _boot_storage_error( @@ -130,35 +161,15 @@ def _boot_storage_error( ) -def _boot_driver_error( - target_url: str, - request_id: str, - started_monotonic: float, - exc: Exception, -) -> ScrapeError: - is_timeout = is_timeout_exception(exc) - return build_error( - target_url, - str(exc), - request_id=request_id, - attempts=0, - render_ms=elapsed_ms(started_monotonic), - error_category=( - ErrorCategory.TIMEOUT if is_timeout else ErrorCategory.NAVIGATION_ERROR - ), - execution_tier=ExecutionTier.BROWSER_DRIVER, - timeout_phase=TimeoutPhase.BOOT, - ) - - -def _work_exception_error( +def _tier_exception_error( target_url: str, *, request_id: str, attempts: int, - strategy: NavigationMode, + strategy: NavigationMode | None, started_monotonic: float, exc: Exception, + timeout_phase: TimeoutPhase | None, ) -> ScrapeError: is_timeout = is_timeout_exception(exc) return build_error( @@ -172,7 +183,7 @@ def _work_exception_error( ErrorCategory.TIMEOUT if is_timeout else ErrorCategory.NAVIGATION_ERROR ), execution_tier=ExecutionTier.BROWSER_DRIVER, - timeout_phase=TimeoutPhase.WORK if is_timeout else None, + timeout_phase=timeout_phase, ) @@ -241,7 +252,15 @@ def run_browser_tier( ), ) except Exception as exc: - return _boot_driver_error(target_url, request_id, started_monotonic, exc) + return _tier_exception_error( + target_url, + request_id=request_id, + attempts=0, + strategy=None, + started_monotonic=started_monotonic, + exc=exc, + timeout_phase=TimeoutPhase.BOOT, + ) session.driver = driver @@ -305,18 +324,19 @@ def run_browser_tier( attempt_index, assessment.detected_marker, ) - if _should_retry_challenge(assessment, has_more_strategies=has_more): - collector.reset() - continue - - return _challenge_block_error( + blocked = _unclean_retry_or_block( target_url, request_id=request_id, attempts=attempts, strategy=strategy, - render_ms=elapsed_ms(started_monotonic), + started_monotonic=started_monotonic, assessment=assessment, + has_more_strategies=has_more, + collector=collector, ) + if blocked is None: + continue + return blocked return build_success( target_url, @@ -347,28 +367,33 @@ def run_browser_tier( # Prefer knowable challenge over timeout when the page is inspectable. assessment = _inspect_assessment(driver, target_url) if assessment is not None and not assessment.is_clean: - if _should_retry_challenge(assessment, has_more_strategies=has_more): - collector.reset() - continue - return _challenge_block_error( + blocked = _unclean_retry_or_block( target_url, request_id=request_id, attempts=attempts, strategy=strategy, - render_ms=elapsed_ms(started_monotonic), + started_monotonic=started_monotonic, assessment=assessment, + has_more_strategies=has_more, + collector=collector, ) + if blocked is None: + continue + return blocked if has_more: collector.reset() continue - return _work_exception_error( + return _tier_exception_error( target_url, request_id=request_id, attempts=attempts, strategy=strategy, started_monotonic=started_monotonic, exc=exc, + timeout_phase=( + TimeoutPhase.WORK if is_timeout_exception(exc) else None + ), ) return build_error( diff --git a/app/infra/detector.py b/app/infra/detector.py index 9d78247..923fdfd 100644 --- a/app/infra/detector.py +++ b/app/infra/detector.py @@ -37,6 +37,10 @@ class ChallengeAssessment: def is_clean(self) -> bool: return not self.blocked_detected and not self.challenge_detected + def may_retry_strategies(self, *, has_more: bool) -> bool: + """Soft challenge markers may retry strategies; hard HTTP blocks do not.""" + return self.challenge_detected and has_more + def to_signal(self) -> ChallengeSignal: """Convert domain assessment to wire ChallengeSignal DTO.""" return ChallengeSignal( diff --git a/tests/engine/test_browser_tier_challenge.py b/tests/engine/test_browser_tier_challenge.py index 4e2e6a8..83cdf11 100644 --- a/tests/engine/test_browser_tier_challenge.py +++ b/tests/engine/test_browser_tier_challenge.py @@ -26,142 +26,113 @@ def __init__(self, status: int, url: str) -> None: )() -class _ChallengeHtmlDriver(FakeDriver): - """Timeout on navigate while page already shows a challenge interstitial.""" +class _ScenarioDriver(FakeDriver): + """Shared navigate counter + optional timeout / HTTP status fixtures.""" navigate_calls: ClassVar[int] = 0 + page_body: ClassVar[str] = "ok" + timeout_on_nav: ClassVar[bool] = False + http_status: ClassVar[int | None] = None def __init__(self, *args: object, **kwargs: Any) -> None: super().__init__(*args, **kwargs) - self.page_html = "Just a moment..." + self.page_html = type(self).page_body self.current_url = "https://example.com/" + status = type(self).http_status + if status is not None: + self.requests = [_PassiveRequest(status, "https://example.com/")] @classmethod def reset(cls) -> None: cls.navigate_calls = 0 - def get(self, *_args: object, **_kwargs: Any) -> None: + def _navigate(self) -> None: type(self).navigate_calls += 1 - raise TimeoutError("Navigation timeout after 15s") - - def google_get(self, *_args: object, **_kwargs: Any) -> None: - type(self).navigate_calls += 1 - raise TimeoutError("Navigation timeout after 15s") - - -class _CleanTimeoutDriver(FakeDriver): - """Timeout on navigate with a clean page body.""" - - def __init__(self, *args: object, **kwargs: Any) -> None: - super().__init__(*args, **kwargs) - self.page_html = "

Example Domain

" - self.current_url = "https://example.com/" + if type(self).timeout_on_nav: + raise TimeoutError("Navigation timeout after 15s") def get(self, *_args: object, **_kwargs: Any) -> None: - raise TimeoutError("Navigation timeout after 15s") + self._navigate() def google_get(self, *_args: object, **_kwargs: Any) -> None: - raise TimeoutError("Navigation timeout after 15s") - + self._navigate() -class _HardBlockDriver(FakeDriver): - """HTTP hard block without challenge marker — must not burn more strategies.""" +class _ChallengeHtmlDriver(_ScenarioDriver): navigate_calls: ClassVar[int] = 0 + page_body = "Just a moment..." + timeout_on_nav = True - def __init__(self, *args: object, **kwargs: Any) -> None: - super().__init__(*args, **kwargs) - self.page_html = "

Forbidden

" - self.current_url = "https://example.com/" - self.requests = [_PassiveRequest(403, "https://example.com/")] - - @classmethod - def reset(cls) -> None: - cls.navigate_calls = 0 - - def get(self, *_args: object, **_kwargs: Any) -> None: - type(self).navigate_calls += 1 - def google_get(self, *_args: object, **_kwargs: Any) -> None: - type(self).navigate_calls += 1 +class _CleanTimeoutDriver(_ScenarioDriver): + page_body = "

Example Domain

" + timeout_on_nav = True -class _SoftChallengeDriver(FakeDriver): - """Soft challenge marker — may retry remaining strategies.""" - +class _HardBlockDriver(_ScenarioDriver): navigate_calls: ClassVar[int] = 0 + page_body = "

Forbidden

" + http_status = 403 - def __init__(self, *args: object, **kwargs: Any) -> None: - super().__init__(*args, **kwargs) - self.page_html = "Just a moment..." - self.current_url = "https://example.com/" - - @classmethod - def reset(cls) -> None: - cls.navigate_calls = 0 - def get(self, *_args: object, **_kwargs: Any) -> None: - type(self).navigate_calls += 1 - - def google_get(self, *_args: object, **_kwargs: Any) -> None: - type(self).navigate_calls += 1 +class _SoftChallengeDriver(_ScenarioDriver): + navigate_calls: ClassVar[int] = 0 + page_body = "Just a moment..." class BrowserTierChallengeTests(unittest.TestCase): - def test_timeout_with_challenge_html_returns_challenge_block(self) -> None: - _ChallengeHtmlDriver.reset() + def _execute( + self, + driver_cls: type[_ScenarioDriver], + *, + navigation_mode: NavigationMode, + max_retries: int, + request_id: str, + ) -> ScrapeError: + driver_cls.reset() payload = scrape_request( execution_mode=ExecutionMode.BROWSER, - navigation_mode=NavigationMode.GET, - max_retries=0, + navigation_mode=navigation_mode, + max_retries=max_retries, ) - with tempfile.TemporaryDirectory() as tmp: engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) - with patch("botasaurus.browser.Driver", _ChallengeHtmlDriver): - result = engine.execute(payload, request_id="req-timeout-challenge") - + with patch("botasaurus.browser.Driver", driver_cls): + result = engine.execute(payload, request_id=request_id) self.assertIsInstance(result, ScrapeError) assert isinstance(result, ScrapeError) + return result + + def test_timeout_with_challenge_html_returns_challenge_block(self) -> None: + result = self._execute( + _ChallengeHtmlDriver, + navigation_mode=NavigationMode.GET, + max_retries=0, + request_id="req-timeout-challenge", + ) self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) self.assertIsNone(result.diagnostics.timeout_phase) - self.assertIsNotNone(result.diagnostics.challenge) assert result.diagnostics.challenge is not None self.assertTrue(result.diagnostics.challenge.detected) self.assertEqual(_ChallengeHtmlDriver.navigate_calls, 1) def test_timeout_with_clean_page_returns_timeout_work(self) -> None: - payload = scrape_request( - execution_mode=ExecutionMode.BROWSER, + result = self._execute( + _CleanTimeoutDriver, navigation_mode=NavigationMode.GET, max_retries=0, + request_id="req-timeout-clean", ) - - with tempfile.TemporaryDirectory() as tmp: - engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) - with patch("botasaurus.browser.Driver", _CleanTimeoutDriver): - result = engine.execute(payload, request_id="req-timeout-clean") - - self.assertIsInstance(result, ScrapeError) - assert isinstance(result, ScrapeError) self.assertEqual(result.error_category, ErrorCategory.TIMEOUT) self.assertEqual(result.diagnostics.timeout_phase, TimeoutPhase.WORK) def test_hard_block_aborts_remaining_strategies(self) -> None: - _HardBlockDriver.reset() - payload = scrape_request( - execution_mode=ExecutionMode.BROWSER, + result = self._execute( + _HardBlockDriver, navigation_mode=NavigationMode.AUTO, max_retries=2, + request_id="req-hard-block", ) - - with tempfile.TemporaryDirectory() as tmp: - engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) - with patch("botasaurus.browser.Driver", _HardBlockDriver): - result = engine.execute(payload, request_id="req-hard-block") - - self.assertIsInstance(result, ScrapeError) - assert isinstance(result, ScrapeError) self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) self.assertIsNone(result.diagnostics.timeout_phase) self.assertEqual(result.diagnostics.attempts, 1) @@ -171,20 +142,12 @@ def test_hard_block_aborts_remaining_strategies(self) -> None: self.assertFalse(result.diagnostics.challenge.detected) def test_soft_challenge_retries_strategies_then_blocks(self) -> None: - _SoftChallengeDriver.reset() - payload = scrape_request( - execution_mode=ExecutionMode.BROWSER, + result = self._execute( + _SoftChallengeDriver, navigation_mode=NavigationMode.AUTO, max_retries=2, + request_id="req-soft-challenge", ) - - with tempfile.TemporaryDirectory() as tmp: - engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) - with patch("botasaurus.browser.Driver", _SoftChallengeDriver): - result = engine.execute(payload, request_id="req-soft-challenge") - - self.assertIsInstance(result, ScrapeError) - assert isinstance(result, ScrapeError) self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) self.assertEqual(result.diagnostics.attempts, 3) self.assertEqual(_SoftChallengeDriver.navigate_calls, 3) diff --git a/tests/infra/test_challenge_detector.py b/tests/infra/test_challenge_detector.py index feaf49e..be5bb02 100644 --- a/tests/infra/test_challenge_detector.py +++ b/tests/infra/test_challenge_detector.py @@ -25,6 +25,16 @@ def test_clean_response(self): self.assertFalse(res.blocked_detected) self.assertFalse(res.challenge_detected) + def test_soft_challenge_may_retry_strategies(self): + soft = ChallengeDetector.detect("Just a moment...", 200) + self.assertTrue(soft.may_retry_strategies(has_more=True)) + self.assertFalse(soft.may_retry_strategies(has_more=False)) + + def test_hard_block_does_not_retry_strategies(self): + hard = ChallengeDetector.detect("Forbidden", 403) + self.assertFalse(hard.may_retry_strategies(has_more=True)) + self.assertFalse(hard.may_retry_strategies(has_more=False)) + def test_driver_bot_detection_integration(self): mock_driver = MagicMock() mock_driver.is_bot_detected.return_value = True