"""Robots evaluation: the one policy every collector applies. Three shipped incidents live here. Python's standard parser predates wildcard patterns, so it read 'Disallow: /*/search/' as permission and a whole retailer family was collected under that misreading. The matcher that replaced it keyed our group on the first User-Agent token ("mozilla"), so a group addressed to DutyFreeProfessorBot by name was ignored. And collectors disagreed on what an unreadable robots.txt meant, one crawling on where another stopped. """ import urllib.error import pytest from app.services.collectors import robots from app.services.collectors.fetch import USER_AGENT, SourceBlocked from app.services.collectors.robots import BOT_NAME, RobotsUnavailable, check_allowed, parse REAL_WORLD = """User-agent: * Disallow: /*/search/ Allow: /en/global/ User-agent: PriceSpiderBot Disallow: / """ def allows(text: str, path: str) -> bool: return parse(text).allows(path) class TestIdentity: def test_the_identity_we_honour_is_the_identity_we_send(self): """The whole policy rests on the UA string naming BOT_NAME: a retailer can only address us by the name we present.""" assert BOT_NAME in USER_AGENT def test_a_group_addressed_to_us_by_name_applies(self): """Reproduced 2026-09-04: the old matcher keyed on the UA's first token ("mozilla") and let a run proceed past this exact group.""" assert not allows("User-agent: DutyFreeProfessorBot\nDisallow: /", "/en/x") @pytest.mark.parametrize("agent", ["dutyfreeprofessorbot", "DutyFreeProfessorBot/0.1", "DUTYFREEPROFESSOR"]) def test_case_and_version_suffix_do_not_matter(self, agent): assert not allows(f"User-agent: {agent}\nDisallow: /private/", "/private/x") def test_the_first_ua_token_is_not_our_name(self): assert allows("User-agent: Mozilla\nDisallow: /", "/anything") def test_another_agents_ban_does_not_apply_to_us(self): assert allows(REAL_WORLD, "/en/global/anything") class TestWildcards: def test_a_wildcard_disallow_refuses_the_search_endpoint(self): """The exact rule that stopped collection across the Heinemann family.""" assert not allows(REAL_WORLD, "/en/global/search/results") def test_the_query_string_wildcard_forbids_paginated_category_pages(self): """Four Avolta storefronts publish 'Disallow: /*?'. Page 2 of a category is '?p=2', so those stores are page-1-only; the stdlib parser let us through and we walked pages 2 to 6 for weeks.""" text = "User-agent: *\nDisallow: /*?" assert allows(text, "/en/1/liquor/whisky") assert not allows(text, "/en/1/liquor/whisky?p=2") assert not allows(text, "https://london-heathrow.worlddutyfree.com/en/1/liquor?p=2") def test_paths_outside_the_rule_are_still_allowed(self): assert allows(REAL_WORLD, "/en/global/products/x") def test_a_star_spans_slashes(self): assert not allows("User-agent: *\nDisallow: /a/*/c", "/a/b/deep/c") def test_a_dollar_anchors_the_end(self): text = "User-agent: *\nDisallow: /exact$" assert not allows(text, "/exact") assert allows(text, "/exact/child") def test_a_longer_allow_does_not_out_lawyer_a_disallow(self): """Deliberately stricter than RFC 9309's longest-match rule.""" assert not allows("User-agent: *\nDisallow: /a/\nAllow: /a/b/c", "/a/b/c") def test_a_full_url_is_matched_on_its_path(self): assert not allows("User-agent: *\nDisallow: /checkout", "https://x.example/checkout?step=1") class TestGroups: def test_a_blanket_ban_is_honoured(self): assert not allows("User-agent: *\nDisallow: /", "/anything") def test_either_our_group_or_star_refusing_is_a_no(self): text = "User-agent: *\nDisallow: /a/\n\nUser-agent: DutyFreeProfessorBot\nDisallow: /b/" assert not allows(text, "/a/x") assert not allows(text, "/b/x") assert allows(text, "/c/x") def test_consecutive_user_agent_lines_form_one_group(self): text = "User-agent: Foo\nUser-agent: DutyFreeProfessorBot\nDisallow: /x" assert not allows(text, "/x") def test_rules_before_any_group_are_ignored(self): assert allows("Disallow: /\nUser-agent: *\nDisallow: /y", "/x") class TestCrawlDelay: def test_our_groups_delay_beats_star(self): text = "User-agent: *\nCrawl-delay: 30\n\nUser-agent: DutyFreeProfessorBot\nCrawl-delay: 60" assert parse(text).crawl_delay == 60 def test_star_delay_applies_when_ours_names_none(self): text = "User-agent: DutyFreeProfessorBot\nDisallow: /x\n\nUser-agent: *\nCrawl-delay: 30" assert parse(text).crawl_delay == 30 def test_the_slower_of_theirs_and_ours_wins(self): r = parse("User-agent: *\nCrawl-delay: 30") assert r.delay_for(1.0) == 30 assert r.delay_for(45.0) == 45 assert parse("User-agent: *\nDisallow:").delay_for(2.5) == 2.5 def test_an_unparseable_delay_is_ignored(self): assert parse("User-agent: *\nCrawl-delay: soon").crawl_delay is None class TestPermissiveCases: @pytest.mark.parametrize( "text", [ "User-agent: *\nDisallow:", # empty value means allow all "", # no robots content "# just a comment\n", "Sitemap: https://x/sitemap.xml\n", ], ) def test_nothing_to_refuse_means_allowed(self, text): assert allows(text, "/anything") def test_noindex_is_not_a_crawl_rule(self): """One host writes 'Noindex: /*/search/'. That governs indexing, not fetching, and must not be read as a Disallow.""" assert allows("User-agent: *\nNoindex: /*/search/", "/en/global/search/x") def test_comments_are_stripped(self): assert not allows("User-agent: *\nDisallow: /x # why", "/x") class TestUnreadablePolicy: """The three outcomes for a robots.txt we cannot read, decided 2026-09-04.""" HOST = "https://shop.example" @staticmethod def fetcher_returning(status: int, body: bytes = b""): return lambda url: (status, body) @pytest.mark.parametrize("status", [404, 410]) def test_absent_means_unrestricted(self, status): r = check_allowed(self.HOST, ["/anything"], fetcher=self.fetcher_returning(status)) assert r.allows("/anything") and r.crawl_delay is None and r.status == status @pytest.mark.parametrize("status", [401, 403]) def test_forbidden_is_a_refusal(self, status): with pytest.raises(SourceBlocked): check_allowed(self.HOST, ["/x"], fetcher=self.fetcher_returning(status)) @pytest.mark.parametrize("status", [500, 502, 503, 429]) def test_server_trouble_stops_the_run_without_calling_it_a_block(self, status): with pytest.raises(RobotsUnavailable): check_allowed(self.HOST, ["/x"], fetcher=self.fetcher_returning(status)) def test_a_timeout_stops_the_run_without_calling_it_a_block(self): def timing_out(url): raise TimeoutError("timed out") with pytest.raises(RobotsUnavailable): check_allowed(self.HOST, ["/x"], fetcher=timing_out) def test_a_dns_failure_stops_the_run_without_calling_it_a_block(self): def failing(url): raise urllib.error.URLError("name or service not known") with pytest.raises(RobotsUnavailable): check_allowed(self.HOST, ["/x"], fetcher=failing) def test_an_empty_robots_is_no_rules_not_a_block(self): """fetch.fetch() treats an empty body as a challenge; robots must not.""" r = check_allowed(self.HOST, ["/x"], fetcher=self.fetcher_returning(200, b"")) assert r.allows("/x") def test_a_disallowed_entry_path_refuses_the_run_before_its_first_request(self): body = b"User-agent: *\nDisallow: /en/global/search/" with pytest.raises(SourceBlocked): check_allowed(self.HOST, ["/en/global/search/results"], fetcher=self.fetcher_returning(200, body)) def test_the_run_gets_the_hosts_delay_back(self): body = b"User-agent: *\nCrawl-delay: 30\nDisallow: /cart" r = check_allowed(self.HOST, ["/en/"], fetcher=self.fetcher_returning(200, body)) assert r.delay_for(1.0) == 30 and not r.allows("/cart") def test_a_cached_read_is_reused_and_a_fresh_one_is_not(self): calls = [] def counting(url): calls.append(url) return 200, b"User-agent: *\nDisallow:" robots._cache.clear() check_allowed(self.HOST, ["/x"], fetcher=counting, fresh=False) check_allowed(self.HOST, ["/x"], fetcher=counting, fresh=False) assert len(calls) == 1 check_allowed(self.HOST, ["/x"], fetcher=counting, fresh=True) assert len(calls) == 2 class TestPatternRegex: """The one translation the fetcher's route filter may import instead of mirroring.""" def test_wildcard_and_anchor(self): from app.services.collectors.robots import pattern_regex import re assert re.match(pattern_regex("/*?"), "/en/64/liquor?p=2") assert not re.match(pattern_regex("/*?"), "/en/64/liquor") assert re.match(pattern_regex("/medias/*.jpg$"), "/medias/a/b.jpg") assert not re.match(pattern_regex("/medias/*.jpg$"), "/medias/a/b.jpg?x=1") def test_the_fetcher_agrees_with_it(self): from app.services.collectors.fetch import disallow_regexes from app.services.collectors.robots import parse, pattern_regex robots = parse("User-agent: *\nDisallow: /*?\nDisallow: /estore/_ui/\nDisallow: /x$\n") assert disallow_regexes(robots) == [pattern_regex(p) for p in robots.disallows]