diff --git a/pyproject.toml b/pyproject.toml index a805818..4333d7e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -66,7 +66,7 @@ dependencies = [ "orjson>=3.11.8", "tld>=0.13.2", "w3lib>=2.4.1", - "typing_extensions", + "typing_extensions" ] [project.optional-dependencies] @@ -78,7 +78,8 @@ fetchers = [ "browserforge>=1.2.4", "apify-fingerprint-datapoints>=0.12.0", "msgspec>=0.20.0", - "anyio>=4.12.1" + "anyio>=4.12.1", + "protego>=0.4.0", ] ai = [ "mcp>=1.26.0", diff --git a/scrapling/spiders/engine.py b/scrapling/spiders/engine.py index d77f838..163101b 100644 --- a/scrapling/spiders/engine.py +++ b/scrapling/spiders/engine.py @@ -10,6 +10,7 @@ from scrapling.core.utils import log from scrapling.spiders.request import Request from scrapling.spiders.scheduler import Scheduler from scrapling.spiders.session import SessionManager +from scrapling.spiders.robotstxt import RobotsTxtManager from scrapling.spiders.result import CrawlStats, ItemList from scrapling.spiders.checkpoint import CheckpointManager, CheckpointData from scrapling.core._types import Dict, Union, Optional, TYPE_CHECKING, Any, AsyncGenerator @@ -41,8 +42,18 @@ class CrawlerEngine: ) self.stats = CrawlStats() + if self.spider.robots_txt_obey: + + async def _fetch_robots(url: str, sid: str): + return await self.session_manager.fetch(Request(url, sid=sid)) + + self._robots_manager: Optional[RobotsTxtManager] = RobotsTxtManager(_fetch_robots) + else: + self._robots_manager = None + self._global_limiter = CapacityLimiter(spider.concurrent_requests) self._domain_limiters: dict[str, CapacityLimiter] = {} + self._domain_delays: dict[str, float] = {} self._allowed_domains: set[str] = spider.allowed_domains or set() self._active_tasks: int = 0 @@ -68,13 +79,61 @@ class CrawlerEngine: return True return False + async def _get_domain_delay(self, request: Request) -> float: + """Resolve the effective download delay for a domain. + + Takes the max of the spider's configured delay and any robots.txt + directives (Crawl-delay / Request-rate). Result is cached per domain. + Also pre-creates a per-domain concurrency limiter of 1 when robots.txt + enforces any delay, before the caller acquires it via _rate_limiter(). + """ + robots_manager = self._robots_manager + if robots_manager is None: + return self.spider.download_delay + + domain = request.domain + + # Return cached delay if available + if domain in self._domain_delays: + return self._domain_delays[domain] + + # For domains covered by _prefetch_robots_txt this is a local parser read. + # Domains discovered mid-crawl (not in start_urls/allowed_domains) will fetch here. + # Two concurrent callbacks hitting the same new domain can each trigger a fetch; + # the second write is a no-op in effect (same content), but the extra request is accepted. + c_delay, r_rate = await robots_manager._get_delay_directives(request.url, request.sid) + + delay = self.spider.download_delay + robots_enforced_delay = False + + if r_rate: + req_count, period = r_rate + if req_count > 0: + delay = max(delay, period / req_count) + robots_enforced_delay = True + + if c_delay is not None: + delay = max(delay, c_delay) + robots_enforced_delay = True + + self._domain_delays[domain] = delay + + # Enforce 1 concurrent request for this domain when robots.txt adds a delay + if robots_enforced_delay and delay > 0 and domain not in self._domain_limiters: + if self.spider.concurrent_requests_per_domain: + log.warning( + f"robots.txt for {domain} enforces a delay, overriding" + f" concurrent_requests_per_domain={self.spider.concurrent_requests_per_domain} with 1" + ) + self._domain_limiters[domain] = CapacityLimiter(1) + + return delay + def _rate_limiter(self, domain: str) -> CapacityLimiter: """Get or create a per-domain concurrency limiter if enabled, otherwise use the global limiter.""" if self.spider.concurrent_requests_per_domain: - if domain not in self._domain_limiters: - self._domain_limiters[domain] = CapacityLimiter(self.spider.concurrent_requests_per_domain) - return self._domain_limiters[domain] - return self._global_limiter + self._domain_limiters.setdefault(domain, CapacityLimiter(self.spider.concurrent_requests_per_domain)) + return self._domain_limiters.get(domain, self._global_limiter) def _normalize_request(self, request: Request) -> None: """Normalize request fields before enqueueing. @@ -87,9 +146,21 @@ class CrawlerEngine: async def _process_request(self, request: Request) -> None: """Download and process a single request.""" + if self._robots_manager: + can_fetch = await self._robots_manager.can_fetch(request.url, request.sid) + if not can_fetch: + self.stats.robots_disallowed_count += 1 + log.debug(f"Request disallowed by robots.txt: {request.url}") + return + # Must be called before _rate_limiter: may create CapacityLimiter(1) in _domain_limiters + # when robots.txt enforces a delay, which _rate_limiter then picks up. + delay = await self._get_domain_delay(request) + else: + delay = self.spider.download_delay + async with self._rate_limiter(request.domain): - if self.spider.download_delay: - await anyio.sleep(self.spider.download_delay) + if delay: + await anyio.sleep(delay) if request._session_kwargs.get("proxy"): self.stats.proxies.append(request._session_kwargs["proxy"]) @@ -219,6 +290,27 @@ class CrawlerEngine: return True + async def _prefetch_robots_txt(self) -> None: + """Pre-warm the robots.txt cache before the crawl loop starts. + + Uses allowed_domains if configured, otherwise falls back to unique domains + extracted from start_urls via Request.domain. Both paths use https. + """ + if not self._robots_manager: + return + + if self._allowed_domains: + domains = self._allowed_domains + elif self.spider.start_urls: + # Deduplicate by domain so we spawn exactly one task per domain + domains = {Request(url).domain for url in self.spider.start_urls} + else: + return + + seed_urls = [f"https://{domain}/" for domain in domains] + + await self._robots_manager.prefetch(seed_urls, self.session_manager.default_session_id) + async def crawl(self) -> CrawlStats: """Run the spider and return CrawlStats.""" self._running = True @@ -227,17 +319,23 @@ class CrawlerEngine: self._pause_requested = False self._force_stop = False self.stats = CrawlStats(start_time=anyio.current_time()) + self._domain_limiters.clear() + self._domain_delays.clear() # Check for existing checkpoint resuming = (await self._restore_from_checkpoint()) if self._checkpoint_system_enabled else False self._last_checkpoint_time = anyio.current_time() async with self.session_manager: + # Set stats from spider configuration self.stats.concurrent_requests = self.spider.concurrent_requests self.stats.concurrent_requests_per_domain = self.spider.concurrent_requests_per_domain self.stats.download_delay = self.spider.download_delay + await self.spider.on_start(resuming=resuming) + await self._prefetch_robots_txt() + try: if not resuming: async for request in self.spider.start_requests(): diff --git a/scrapling/spiders/result.py b/scrapling/spiders/result.py index 08a7658..b374152 100644 --- a/scrapling/spiders/result.py +++ b/scrapling/spiders/result.py @@ -47,6 +47,7 @@ class CrawlStats: concurrent_requests_per_domain: int = 0 failed_requests_count: int = 0 offsite_requests_count: int = 0 + robots_disallowed_count: int = 0 response_bytes: int = 0 items_scraped: int = 0 items_dropped: int = 0 @@ -95,6 +96,7 @@ class CrawlStats: "sessions_requests_count": self.sessions_requests_count, "failed_requests_count": self.failed_requests_count, "offsite_requests_count": self.offsite_requests_count, + "robots_disallowed_count": self.robots_disallowed_count, "blocked_requests_count": self.blocked_requests_count, "response_status_count": self.response_status_count, "response_bytes": self.response_bytes, diff --git a/scrapling/spiders/robotstxt.py b/scrapling/spiders/robotstxt.py new file mode 100644 index 0000000..c64c66e --- /dev/null +++ b/scrapling/spiders/robotstxt.py @@ -0,0 +1,169 @@ +from urllib.parse import urlparse + +from anyio import create_task_group +from protego import Protego + +from scrapling.core._types import Dict, Optional, Callable, Awaitable +from scrapling.core.utils import log + + +class RobotsTxtManager: + """Manages fetching, parsing, and caching of robots.txt files. + + Accepts a fetch callable ``(url: str, sid: str) -> Awaitable[Response]`` + so it stays decoupled from any specific session or transport layer. + + All public methods accept only ``(url, sid)`` — domain and scheme are + derived internally from the URL so callers don't pass redundant data. + + Handles all standard robots.txt directives including: + - User-agent specific rules + - Allow/Disallow directives (including wildcards and $ anchors) + - Crawl-delay directives + + robots.txt is a domain-level document and does not vary by session, so the + cache is keyed by domain only. The ``sid`` parameter on public methods + controls which session is used for the initial fetch if the domain is not + yet cached, but all sessions share the same parsed result afterwards. + """ + + def __init__(self, fetch_fn: Callable[[str, str], Awaitable]): + self._fetch_fn = fetch_fn + self._cache: Dict[str, Protego] = {} + + async def _get_parser(self, url: str, sid: str) -> Protego: + parsed = urlparse(url) + domain = parsed.netloc + + if domain in self._cache: + return self._cache[domain] + + scheme = parsed.scheme or "https" + robots_url = f"{scheme}://{domain}/robots.txt" + content = "" + try: + response = await self._fetch_fn(robots_url, sid) + if response.status == 200: + content = response.body.decode(response.encoding, errors="replace") + except Exception as e: + log.warning(f"Failed to fetch robots.txt for {domain}: {e}") + + try: + parser = Protego.parse(content) + except Exception as e: + log.warning(f"Failed to parse robots.txt for {domain}: {e}") + parser = Protego.parse("") + + self._cache[domain] = parser + return parser + + async def can_fetch(self, url: str, sid: str) -> bool: + """Check if a URL can be fetched according to the domain's robots.txt. + + Handles: + - User-agent specific rules (e.g., User-agent: SpinarakBot) + - Wildcard user-agent rules (User-agent: *) + - Allow/Disallow directives with wildcards (e.g., /*.pdf$) + - Allow directives that override Disallow (e.g., Allow: /admin/public-docs/) + + Uses the wildcard user-agent (*) which matches standard robots.txt directives + that apply to all bots. This is the conservative approach — if a URL is + disallowed for all bots, we respect that. + + Args: + url: The full URL to check + sid: Session ID for fetching robots.txt if not yet cached + + Returns: + True if the URL can be fetched, False otherwise + """ + parser = await self._get_parser(url, sid) + return parser.can_fetch(url, "*") + + async def get_crawl_delay(self, url: str, sid: str) -> Optional[float]: + """Get the crawl delay for this crawler. + + Uses the wildcard user-agent (*) to get the general crawl delay + that applies to all bots. + + Args: + url: Any URL on the domain to check + sid: Session ID for fetching robots.txt if not yet cached + + Returns: + The crawl delay in seconds, or None if not specified + """ + parser = await self._get_parser(url, sid) + delay = parser.crawl_delay("*") + return float(delay) if delay is not None else None + + async def get_request_rate(self, url: str, sid: str) -> Optional[tuple[int, int]]: + """Get the request rate for this crawler. + + Uses the wildcard user-agent (*) to get the general request rate + that applies to all bots. + + Args: + url: Any URL on the domain to check + sid: Session ID for fetching robots.txt if not yet cached + + Returns: + A tuple of (requests, seconds) if specified, or None if not specified + """ + parser = await self._get_parser(url, sid) + rate = parser.request_rate("*") + if rate is not None: + return (rate.requests, rate.seconds) + return None + + async def _get_delay_directives(self, url: str, sid: str) -> tuple[Optional[float], Optional[tuple[int, int]]]: + """Return both crawl-delay and request-rate in a single parser lookup. + + Args: + url: Any URL on the domain to check + sid: Session ID for fetching robots.txt if not yet cached + + Returns: + A tuple of (crawl_delay, request_rate) where crawl_delay is in seconds + or None, and request_rate is (requests, seconds) or None. + """ + parser = await self._get_parser(url, sid) + c_delay = parser.crawl_delay("*") + rate = parser.request_rate("*") + return ( + float(c_delay) if c_delay is not None else None, + (rate.requests, rate.seconds) if rate is not None else None, + ) + + async def prefetch(self, urls: list[str], sid: str) -> None: + """Pre-warm the robots.txt cache for a list of seed URLs concurrently. + + Callers are responsible for deduplicating URLs by domain before calling + this method — passing multiple URLs for the same domain will trigger + redundant fetches since no inflight deduplication exists here. + + Args: + urls: Seed URLs whose domains should be pre-fetched (one per domain). + sid: Session ID to use for the robots.txt fetch requests. + """ + if not urls: + return + log.debug(f"Pre-fetching robots.txt for {len(urls)} domain(s)") + async with create_task_group() as tg: + for url in urls: + tg.start_soon(self._get_parser, url, sid) + + def clear_cache(self, domain: Optional[str] = None) -> None: + """Clear the robots.txt cache. + + Note: the ``sid`` parameter was removed — the cache is now keyed by + domain only, so clearing a domain evicts all sessions at once. + + Args: + domain: If specified, only clear cache for this domain. + If None, clears the entire cache. + """ + if domain is None: + self._cache.clear() + else: + self._cache.pop(domain, None) diff --git a/scrapling/spiders/spider.py b/scrapling/spiders/spider.py index 4f38912..52afcbd 100644 --- a/scrapling/spiders/spider.py +++ b/scrapling/spiders/spider.py @@ -72,6 +72,9 @@ class Spider(ABC): start_urls: list[str] = [] allowed_domains: Set[str] = set() + # Robots.txt compliance + robots_txt_obey: bool = True + # Concurrency settings concurrent_requests: int = 4 concurrent_requests_per_domain: int = 0 diff --git a/tests/spiders/test_engine.py b/tests/spiders/test_engine.py index b7bfd0f..e7382f8 100644 --- a/tests/spiders/test_engine.py +++ b/tests/spiders/test_engine.py @@ -8,6 +8,7 @@ import pytest from scrapling.spiders.engine import CrawlerEngine, _dump from scrapling.spiders.request import Request +from scrapling.spiders.robotstxt import RobotsTxtManager from scrapling.spiders.session import SessionManager from scrapling.spiders.result import CrawlStats, ItemList from scrapling.spiders.checkpoint import CheckpointData @@ -22,10 +23,11 @@ from scrapling.core._types import Any, Dict, Set, AsyncGenerator class MockResponse: """Minimal Response stand-in.""" - def __init__(self, status: int = 200, body: bytes = b"ok", url: str = "https://example.com"): + def __init__(self, status: int = 200, body: bytes = b"ok", url: str = "https://example.com", encoding: str = "utf-8"): self.status = status self.body = body self.url = url + self.encoding = encoding self.request: Any = None self.meta: Dict[str, Any] = {} @@ -83,6 +85,8 @@ class MockSpider: is_blocked_fn=None, on_scraped_item_fn=None, retry_blocked_request_fn=None, + robots_txt_obey: bool = False, + start_urls: list[str] | None = None, ): self.concurrent_requests = concurrent_requests self.concurrent_requests_per_domain = concurrent_requests_per_domain @@ -93,6 +97,8 @@ class MockSpider: self.fp_include_headers = fp_include_headers self.fp_keep_fragments = fp_keep_fragments self.name = "test_spider" + self.robots_txt_obey = robots_txt_obey + self.start_urls = start_urls or [] # Tracking lists self.on_start_calls: list[dict] = [] @@ -912,3 +918,83 @@ class TestPauseDuringCrawl: await engine.crawl() assert engine.paused is False + + +# --------------------------------------------------------------------------- +# Tests: _prefetch_robots_txt +# --------------------------------------------------------------------------- + + +class TestPrefetchRobotsTxt: + """_prefetch_robots_txt warms the robots.txt cache before the crawl loop.""" + + @staticmethod + def _make_counting_fetch(): + """Return (fetch_fn, calls_list) where calls_list records every (url, sid) pair.""" + calls: list[tuple[str, str]] = [] + + async def _fetch(url: str, sid: str): + calls.append((url, sid)) + return MockResponse(status=200, body=b"", url=url) + + return _fetch, calls + + @pytest.mark.asyncio + async def test_prefetch_uses_allowed_domains_when_set(self): + fetch_fn, calls = self._make_counting_fetch() + spider = MockSpider(allowed_domains={"a.com", "b.com"}, robots_txt_obey=True) + engine = _make_engine(spider=spider) + engine._robots_manager = RobotsTxtManager(fetch_fn) + + await engine._prefetch_robots_txt() + + fetched_domains = {Request(url).domain for url, _ in calls} + assert fetched_domains == {"a.com", "b.com"} + + @pytest.mark.asyncio + async def test_prefetch_falls_back_to_start_urls_when_no_allowed_domains(self): + fetch_fn, calls = self._make_counting_fetch() + spider = MockSpider(robots_txt_obey=True, start_urls=["https://example.com/page1"]) + engine = _make_engine(spider=spider) + engine._robots_manager = RobotsTxtManager(fetch_fn) + + await engine._prefetch_robots_txt() + + assert len(calls) == 1 + assert calls[0][0] == "https://example.com/robots.txt" + + @pytest.mark.asyncio + async def test_prefetch_noop_when_robots_disabled(self): + fetch_fn, calls = self._make_counting_fetch() + spider = MockSpider(robots_txt_obey=False) + engine = _make_engine(spider=spider) + + assert engine._robots_manager is None + + await engine._prefetch_robots_txt() + + assert calls == [] + + @pytest.mark.asyncio + async def test_prefetch_noop_when_start_urls_empty(self): + fetch_fn, calls = self._make_counting_fetch() + spider = MockSpider(robots_txt_obey=True, start_urls=[]) + engine = _make_engine(spider=spider) + engine._robots_manager = RobotsTxtManager(fetch_fn) + + await engine._prefetch_robots_txt() + + assert calls == [] + + @pytest.mark.asyncio + async def test_prefetch_deduplicates_same_domain_in_start_urls(self): + fetch_fn, calls = self._make_counting_fetch() + spider = MockSpider(robots_txt_obey=True, start_urls=["https://example.com/a", "https://example.com/b"]) + engine = _make_engine(spider=spider) + engine._robots_manager = RobotsTxtManager(fetch_fn) + + await engine._prefetch_robots_txt() + + # set of Request.domain values deduplicates to one task per domain + assert len(calls) == 1 + assert calls[0][0] == "https://example.com/robots.txt" diff --git a/tests/spiders/test_robotstxt.py b/tests/spiders/test_robotstxt.py new file mode 100644 index 0000000..efb8328 --- /dev/null +++ b/tests/spiders/test_robotstxt.py @@ -0,0 +1,632 @@ +"""Tests for RobotsTxtManager.""" + +import asyncio + +import pytest + +from scrapling.spiders.robotstxt import RobotsTxtManager + + +# --------------------------------------------------------------------------- +# Fixtures and helpers +# --------------------------------------------------------------------------- + + +class MockResponse: + """Minimal response stub matching the shape _get_parser expects.""" + + def __init__(self, status: int = 200, body: bytes = b"", encoding: str = "utf-8"): + self.status = status + self.body = body + self.encoding = encoding + + +def make_fetch_fn(status: int = 200, content: str = "", encoding: str = "utf-8"): + """Return an async fetch callable that returns a fixed response. + + Attaches a `.calls` list so tests can assert how many times it was invoked + and with which arguments. + """ + calls: list[tuple] = [] + + async def _fetch(url: str, sid: str) -> MockResponse: + calls.append((url, sid)) + return MockResponse(status=status, body=content.encode(encoding), encoding=encoding) + + _fetch.calls = calls # type: ignore[attr-defined] + return _fetch + + +# --------------------------------------------------------------------------- +# Shared robots.txt fixtures +# --------------------------------------------------------------------------- + +ROBOTS_BASIC = """\ +User-agent: * +Disallow: /admin/ +Crawl-delay: 2 +""" + +ROBOTS_WITH_RATE = """\ +User-agent: * +Request-rate: 1/10 +Disallow: /private/ +""" + +ROBOTS_WITH_SITEMAP = """\ +User-agent: * +Disallow: + +Sitemap: https://example.com/sitemap.xml +Sitemap: https://example.com/sitemap2.xml +""" + +ROBOTS_ALLOW_OVERRIDE = """\ +User-agent: * +Disallow: /secret/ +Allow: /secret/public.html +""" + +ROBOTS_DISALLOW_ALL = """\ +User-agent: * +Disallow: / +""" + + +# --------------------------------------------------------------------------- +# Tests: can_fetch +# --------------------------------------------------------------------------- + + +class TestCanFetch: + @pytest.mark.asyncio + async def test_allowed_url_returns_true(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC)) + + assert await mgr.can_fetch("https://example.com/products", "s1") is True + + @pytest.mark.asyncio + async def test_disallowed_url_returns_false(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC)) + + assert await mgr.can_fetch("https://example.com/admin/", "s1") is False + + @pytest.mark.asyncio + async def test_disallowed_subpath_returns_false(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC)) + + assert await mgr.can_fetch("https://example.com/admin/users", "s1") is False + + @pytest.mark.asyncio + async def test_root_url_is_allowed(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC)) + + assert await mgr.can_fetch("https://example.com/", "s1") is True + + @pytest.mark.asyncio + async def test_allow_directive_overrides_disallow(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_ALLOW_OVERRIDE)) + + assert await mgr.can_fetch("https://example.com/secret/public.html", "s1") is True + assert await mgr.can_fetch("https://example.com/secret/private.html", "s1") is False + + @pytest.mark.asyncio + async def test_disallow_all_blocks_every_path(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_DISALLOW_ALL)) + + assert await mgr.can_fetch("https://example.com/", "s1") is False + assert await mgr.can_fetch("https://example.com/page", "s1") is False + assert await mgr.can_fetch("https://example.com/a/b/c", "s1") is False + + @pytest.mark.asyncio + async def test_empty_robots_allows_everything(self): + mgr = RobotsTxtManager(make_fetch_fn(content="")) + + assert await mgr.can_fetch("https://example.com/anything", "s1") is True + assert await mgr.can_fetch("https://example.com/admin/secret", "s1") is True + + @pytest.mark.asyncio + async def test_non_200_response_allows_everything(self): + for status in [403, 404, 500, 503]: + mgr = RobotsTxtManager(make_fetch_fn(status=status)) + result = await mgr.can_fetch("https://example.com/page", "s1") + assert result is True, f"Expected True for HTTP {status}" + + @pytest.mark.asyncio + async def test_fetch_error_allows_everything(self): + async def failing_fetch(url: str, sid: str) -> MockResponse: + raise ConnectionError("network failure") + + mgr = RobotsTxtManager(failing_fetch) + + assert await mgr.can_fetch("https://example.com/page", "s1") is True + + @pytest.mark.asyncio + async def test_wildcard_path_pattern(self): + content = "User-agent: *\nDisallow: /*.pdf$" + mgr = RobotsTxtManager(make_fetch_fn(content=content)) + + assert await mgr.can_fetch("https://example.com/report.pdf", "s1") is False + assert await mgr.can_fetch("https://example.com/report.html", "s1") is True + + @pytest.mark.asyncio + async def test_returns_bool(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC)) + result = await mgr.can_fetch("https://example.com/", "s1") + assert isinstance(result, bool) + + +# --------------------------------------------------------------------------- +# Tests: get_crawl_delay +# --------------------------------------------------------------------------- + + +class TestGetCrawlDelay: + @pytest.mark.asyncio + async def test_returns_float_when_set(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC)) + + delay = await mgr.get_crawl_delay("https://example.com/", "s1") + + assert delay == 2.0 + assert isinstance(delay, float) + + @pytest.mark.asyncio + async def test_returns_none_when_not_set(self): + content = "User-agent: *\nDisallow: /admin/" + mgr = RobotsTxtManager(make_fetch_fn(content=content)) + + assert await mgr.get_crawl_delay("https://example.com/", "s1") is None + + @pytest.mark.asyncio + async def test_returns_none_for_empty_robots(self): + mgr = RobotsTxtManager(make_fetch_fn(content="")) + + assert await mgr.get_crawl_delay("https://example.com/", "s1") is None + + @pytest.mark.asyncio + async def test_returns_none_on_fetch_error(self): + async def failing_fetch(url: str, sid: str) -> MockResponse: + raise ConnectionError("network failure") + + mgr = RobotsTxtManager(failing_fetch) + + assert await mgr.get_crawl_delay("https://example.com/", "s1") is None + + @pytest.mark.asyncio + async def test_returns_none_for_non_200_response(self): + mgr = RobotsTxtManager(make_fetch_fn(status=404)) + + assert await mgr.get_crawl_delay("https://example.com/", "s1") is None + + @pytest.mark.asyncio + async def test_fractional_delay(self): + content = "User-agent: *\nCrawl-delay: 0.5" + mgr = RobotsTxtManager(make_fetch_fn(content=content)) + + delay = await mgr.get_crawl_delay("https://example.com/", "s1") + + assert delay == 0.5 + + @pytest.mark.asyncio + async def test_url_path_does_not_affect_result(self): + """Any URL on the same domain should return the same delay.""" + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC)) + + d1 = await mgr.get_crawl_delay("https://example.com/", "s1") + d2 = await mgr.get_crawl_delay("https://example.com/deep/path/page.html", "s1") + + assert d1 == d2 + + +# --------------------------------------------------------------------------- +# Tests: get_request_rate +# --------------------------------------------------------------------------- + + +class TestGetRequestRate: + @pytest.mark.asyncio + async def test_returns_tuple_when_set(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_WITH_RATE)) + + rate = await mgr.get_request_rate("https://example.com/", "s1") + + assert rate is not None + assert isinstance(rate, tuple) + assert len(rate) == 2 + + @pytest.mark.asyncio + async def test_tuple_contains_integers(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_WITH_RATE)) + + rate = await mgr.get_request_rate("https://example.com/", "s1") + + assert rate is not None + requests, seconds = rate + assert isinstance(requests, int) + assert isinstance(seconds, int) + + @pytest.mark.asyncio + async def test_returns_none_when_not_set(self): + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC)) + + assert await mgr.get_request_rate("https://example.com/", "s1") is None + + @pytest.mark.asyncio + async def test_returns_none_for_empty_robots(self): + mgr = RobotsTxtManager(make_fetch_fn(content="")) + + assert await mgr.get_request_rate("https://example.com/", "s1") is None + + @pytest.mark.asyncio + async def test_returns_none_on_fetch_error(self): + async def failing_fetch(url: str, sid: str) -> MockResponse: + raise ConnectionError("network failure") + + mgr = RobotsTxtManager(failing_fetch) + + assert await mgr.get_request_rate("https://example.com/", "s1") is None + + @pytest.mark.asyncio + async def test_returns_none_for_non_200_response(self): + mgr = RobotsTxtManager(make_fetch_fn(status=404)) + + assert await mgr.get_request_rate("https://example.com/", "s1") is None + + +# --------------------------------------------------------------------------- +# Tests: caching behaviour +# --------------------------------------------------------------------------- + + +class TestCachingBehaviour: + @pytest.mark.asyncio + async def test_second_call_same_domain_uses_cache(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/page1", "s1") + await mgr.can_fetch("https://example.com/page2", "s1") + + assert len(fetch_fn.calls) == 1 + + @pytest.mark.asyncio + async def test_all_methods_share_cache(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/", "s1") + await mgr.get_crawl_delay("https://example.com/", "s1") + await mgr.get_request_rate("https://example.com/", "s1") + + assert len(fetch_fn.calls) == 1 + + @pytest.mark.asyncio + async def test_different_sids_share_cache_entry(self): + """robots.txt is domain-level — different sessions share the same cached parser.""" + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/", "s1") + await mgr.can_fetch("https://example.com/", "s2") + + assert len(fetch_fn.calls) == 1 + + @pytest.mark.asyncio + async def test_different_domains_use_separate_cache_entries(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/", "s1") + await mgr.can_fetch("https://other.com/", "s1") + + assert len(fetch_fn.calls) == 2 + + @pytest.mark.asyncio + async def test_cache_keyed_by_domain_not_path(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/a/b/c", "s1") + await mgr.can_fetch("https://example.com/x/y/z", "s1") + await mgr.can_fetch("https://example.com/admin/", "s1") + + assert len(fetch_fn.calls) == 1 + + @pytest.mark.asyncio + async def test_sid_is_passed_to_fetch_fn(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/", "my_session") + + _, received_sid = fetch_fn.calls[0] + assert received_sid == "my_session" + + +# --------------------------------------------------------------------------- +# Tests: robots.txt URL construction +# --------------------------------------------------------------------------- + + +class TestRobotsTxtUrlConstruction: + @pytest.mark.asyncio + async def test_http_scheme_preserved(self): + fetch_fn = make_fetch_fn(content="") + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("http://example.com/page", "s1") + + fetched_url, _ = fetch_fn.calls[0] + assert fetched_url == "http://example.com/robots.txt" + + @pytest.mark.asyncio + async def test_https_scheme_preserved(self): + fetch_fn = make_fetch_fn(content="") + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/page", "s1") + + fetched_url, _ = fetch_fn.calls[0] + assert fetched_url == "https://example.com/robots.txt" + + @pytest.mark.asyncio + async def test_fetched_at_domain_root_regardless_of_request_path(self): + fetch_fn = make_fetch_fn(content="") + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/deep/nested/path/page.html", "s1") + + fetched_url, _ = fetch_fn.calls[0] + assert fetched_url == "https://example.com/robots.txt" + + @pytest.mark.asyncio + async def test_port_included_in_url(self): + fetch_fn = make_fetch_fn(content="") + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("http://example.com:8080/page", "s1") + + fetched_url, _ = fetch_fn.calls[0] + assert fetched_url == "http://example.com:8080/robots.txt" + + @pytest.mark.asyncio + async def test_different_ports_treated_as_different_domains(self): + fetch_fn = make_fetch_fn(content="") + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("http://example.com:8000/page", "s1") + await mgr.can_fetch("http://example.com:9000/page", "s1") + + assert len(fetch_fn.calls) == 2 + urls = [call[0] for call in fetch_fn.calls] + assert "http://example.com:8000/robots.txt" in urls + assert "http://example.com:9000/robots.txt" in urls + + +# --------------------------------------------------------------------------- +# Tests: encoding +# --------------------------------------------------------------------------- + + +class TestEncoding: + @pytest.mark.asyncio + async def test_non_utf8_body_decoded_with_response_encoding(self): + content = "User-agent: *\nDisallow: /admin/\nCrawl-delay: 3" + body = content.encode("latin-1") + + async def fetch_fn(url: str, sid: str) -> MockResponse: + return MockResponse(status=200, body=body, encoding="latin-1") + + mgr = RobotsTxtManager(fetch_fn) + delay = await mgr.get_crawl_delay("https://example.com/", "s1") + + assert delay == 3.0 + + @pytest.mark.asyncio + async def test_bytes_body_decoded_correctly(self): + content = "User-agent: *\nDisallow: /private/" + body = content.encode("utf-8") + + async def fetch_fn(url: str, sid: str) -> MockResponse: + return MockResponse(status=200, body=body, encoding="utf-8") + + mgr = RobotsTxtManager(fetch_fn) + + assert await mgr.can_fetch("https://example.com/private/", "s1") is False + assert await mgr.can_fetch("https://example.com/public/", "s1") is True + + +# --------------------------------------------------------------------------- +# Tests: clear_cache +# --------------------------------------------------------------------------- + + +class TestClearCache: + @pytest.mark.asyncio + async def test_clear_all_forces_refetch(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/", "s1") + mgr.clear_cache() + await mgr.can_fetch("https://example.com/", "s1") + + assert len(fetch_fn.calls) == 2 + + @pytest.mark.asyncio + async def test_clear_by_domain_only_invalidates_that_domain(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/", "s1") + await mgr.can_fetch("https://other.com/", "s1") + assert len(fetch_fn.calls) == 2 + + mgr.clear_cache(domain="example.com") + + await mgr.can_fetch("https://example.com/", "s1") # refetched + await mgr.can_fetch("https://other.com/", "s1") # still cached + + assert len(fetch_fn.calls) == 3 + + @pytest.mark.asyncio + async def test_clear_by_domain_invalidates_all_sessions(self): + """Clearing a domain evicts the single shared cache entry for all sessions.""" + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://example.com/", "s1") + assert len(fetch_fn.calls) == 1 + + mgr.clear_cache(domain="example.com") + + await mgr.can_fetch("https://example.com/", "s1") # refetched — cache was cleared + await mgr.can_fetch("https://example.com/", "s2") # hits the newly warm cache, no fetch + + assert len(fetch_fn.calls) == 2 + + def test_clear_nonexistent_domain_does_not_raise(self): + mgr = RobotsTxtManager(make_fetch_fn()) + mgr.clear_cache(domain="nevervisited.com") # should not raise + + def test_clear_empty_cache_does_not_raise(self): + mgr = RobotsTxtManager(make_fetch_fn()) + mgr.clear_cache() # should not raise + + @pytest.mark.asyncio + async def test_clear_all_empties_cache_completely(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.can_fetch("https://a.com/", "s1") + await mgr.can_fetch("https://b.com/", "s1") + await mgr.can_fetch("https://c.com/", "s1") + assert len(fetch_fn.calls) == 3 + + mgr.clear_cache() + + await mgr.can_fetch("https://a.com/", "s1") + await mgr.can_fetch("https://b.com/", "s1") + await mgr.can_fetch("https://c.com/", "s1") + + assert len(fetch_fn.calls) == 6 + + +# --------------------------------------------------------------------------- +# Tests: concurrent access +# --------------------------------------------------------------------------- + + +class TestCacheAndConcurrency: + @pytest.mark.asyncio + async def test_cached_domain_not_refetched(self): + """Once a domain is cached, subsequent calls return the cached parser without fetching.""" + fetch_count = 0 + + async def counting_fetch(url: str, sid: str) -> MockResponse: + nonlocal fetch_count + fetch_count += 1 + return MockResponse(status=200, body=ROBOTS_BASIC.encode(), encoding="utf-8") + + mgr = RobotsTxtManager(counting_fetch) + + # First call fetches and caches + await mgr.can_fetch("https://example.com/page1", "s1") + # Subsequent calls hit the cache + for i in range(7): + await mgr.can_fetch(f"https://example.com/page{i + 2}", "s1") + + assert fetch_count == 1 + + @pytest.mark.asyncio + async def test_concurrent_calls_different_domains_fetch_independently(self): + fetch_count = 0 + + async def slow_fetch(url: str, sid: str) -> MockResponse: + nonlocal fetch_count + fetch_count += 1 + await asyncio.sleep(0.01) + return MockResponse(status=200, body=b"", encoding="utf-8") + + mgr = RobotsTxtManager(slow_fetch) + + await asyncio.gather( + mgr.can_fetch("https://alpha.com/", "s1"), + mgr.can_fetch("https://beta.com/", "s1"), + mgr.can_fetch("https://gamma.com/", "s1"), + ) + + assert fetch_count == 3 + + @pytest.mark.asyncio + async def test_concurrent_calls_consistent_results(self): + """All concurrent callers should see the same allow/disallow result.""" + mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC)) + + results = await asyncio.gather(*[ + mgr.can_fetch("https://example.com/admin/", "s1") + for _ in range(6) + ]) + + assert all(r is False for r in results) + + @pytest.mark.asyncio + async def test_different_sids_share_cache_after_first_fetch(self): + """After the first fetch, all sessions share the cached parser regardless of sid.""" + fetch_count = 0 + + async def counting_fetch(url: str, sid: str) -> MockResponse: + nonlocal fetch_count + fetch_count += 1 + return MockResponse(status=200, body=b"", encoding="utf-8") + + mgr = RobotsTxtManager(counting_fetch) + + # First call fetches and caches + await mgr.can_fetch("https://example.com/", "s1") + # s2 and s3 hit the cache — no additional fetches + await mgr.can_fetch("https://example.com/", "s2") + await mgr.can_fetch("https://example.com/", "s3") + + assert fetch_count == 1 + + +# --------------------------------------------------------------------------- +# Tests: prefetch +# --------------------------------------------------------------------------- + + +class TestPrefetch: + @pytest.mark.asyncio + async def test_prefetch_fetches_all_domains(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.prefetch(["https://a.com/", "https://b.com/", "https://c.com/"], "s1") + + assert len(fetch_fn.calls) == 3 + fetched = {url for url, _ in fetch_fn.calls} + assert fetched == {"https://a.com/robots.txt", "https://b.com/robots.txt", "https://c.com/robots.txt"} + + @pytest.mark.asyncio + async def test_prefetch_warms_cache_for_subsequent_calls(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.prefetch(["https://example.com/"], "s1") + assert len(fetch_fn.calls) == 1 + + # Any subsequent call for the same domain hits the cache + await mgr.can_fetch("https://example.com/products", "s1") + await mgr.can_fetch("https://example.com/products", "s2") + assert len(fetch_fn.calls) == 1 + + @pytest.mark.asyncio + async def test_prefetch_empty_list_is_noop(self): + fetch_fn = make_fetch_fn(content=ROBOTS_BASIC) + mgr = RobotsTxtManager(fetch_fn) + + await mgr.prefetch([], "s1") + + assert len(fetch_fn.calls) == 0