from __future__ import annotations import unittest from unittest.mock import patch from iaai_scraper.core.config import Settings from iaai_scraper.discovery.sitemap import ( SitemapBlockedError, SitemapDiscoveryError, SitemapFetchResult, _filter_vehicle_urls, discover_vehicle_urls_from_sitemap, discover_vehicle_urls_from_sitemap_with_stats, ) class TestSitemapDiscovery(unittest.TestCase): @staticmethod def _fetch_result(url: str, payload: bytes, *, source: str = "curl_cffi") -> SitemapFetchResult: return SitemapFetchResult(url=url, payload=payload, source=source) def test_filter_vehicle_urls_dedupes_and_normalizes(self) -> None: urls = _filter_vehicle_urls([ "https://www.iaai.com/VehicleDetail/111~US?foo=1", "https://www.iaai.com/VehicleDetail/111~US?bar=2", "https://www.iaai.com/VehicleDetail/222~US", "https://www.iaai.com/about", ]) self.assertEqual( urls, [ "https://www.iaai.com/VehicleDetail/111~US", "https://www.iaai.com/VehicleDetail/222~US", ], ) def test_discover_vehicle_urls_from_sitemap(self) -> None: index_xml = b""" https://www.iaai.com/sitemap-a.xml https://www.iaai.com/sitemap-b.xml https://www.iaai.com/sitemapauctions1.xml """ sitemap_a = b""" https://www.iaai.com/VehicleDetail/111~US https://www.iaai.com/VehicleDetail/222~US?x=1 """ sitemap_b = b""" https://www.iaai.com/VehicleDetail/222~US?y=2 https://www.iaai.com/VehicleDetail/333~US """ def _fake_fetch(url: str) -> SitemapFetchResult: if url.endswith("sitemap_index.xml"): return self._fetch_result(url, index_xml) if url.endswith("sitemap-a.xml"): return self._fetch_result(url, sitemap_a) if url.endswith("sitemap-b.xml"): return self._fetch_result(url, sitemap_b) raise AssertionError(f"unexpected url: {url}") with patch("iaai_scraper.discovery.sitemap._SitemapDownloader.fetch", side_effect=_fake_fetch): result = discover_vehicle_urls_from_sitemap("https://www.iaai.com/sitemap_index.xml") self.assertEqual( result, [ "https://www.iaai.com/VehicleDetail/111~US", "https://www.iaai.com/VehicleDetail/222~US", "https://www.iaai.com/VehicleDetail/333~US", ], ) def test_discover_vehicle_urls_raises_on_empty_index(self) -> None: empty_index = b"" with patch( "iaai_scraper.discovery.sitemap._SitemapDownloader.fetch", return_value=self._fetch_result("https://www.iaai.com/sitemap_index.xml", empty_index), ): with self.assertRaises(SitemapDiscoveryError): discover_vehicle_urls_from_sitemap("https://www.iaai.com/sitemap_index.xml") def test_discover_vehicle_urls_skips_malformed_and_non_vehicle_sitemaps(self) -> None: sitemap_vehicle = b""" https://www.iaai.com/VehicleDetail/111~US """ sitemap_non_vehicle = b""" https://www.iaai.com/SalesList/111~US/04222026 """ index_xml = b""" https://www.iaai.com/sitemap1.xml https://www.iaai.com/sitemapbranches1.xml https://www.iaai.com/sitemap2.xml https://www.iaai.com/sitemap3.xml """ def _fake_fetch(url: str) -> SitemapFetchResult: if url.endswith("sitemap_index.xml"): return self._fetch_result(url, index_xml) if url.endswith("sitemap1.xml"): return self._fetch_result(url, sitemap_vehicle) if url.endswith("sitemap2.xml"): return self._fetch_result(url, sitemap_non_vehicle) if url.endswith("sitemap3.xml"): raise SitemapDiscoveryError("broken sitemap") raise AssertionError(f"unexpected url: {url}") with patch("iaai_scraper.discovery.sitemap._SitemapDownloader.fetch", side_effect=_fake_fetch): result = discover_vehicle_urls_from_sitemap("https://www.iaai.com/sitemap_index.xml") self.assertEqual(result, ["https://www.iaai.com/VehicleDetail/111~US"]) def test_discover_vehicle_urls_falls_back_to_regex_loc_extraction(self) -> None: malformed_index = ( b"" b"https://www.iaai.com/sitemap1.xml" b" https://www.iaai.com/VehicleDetail/111~US """ def _fake_fetch(url: str) -> SitemapFetchResult: if url.endswith("sitemap_index.xml"): return self._fetch_result(url, malformed_index) if url.endswith("sitemap1.xml"): return self._fetch_result(url, sitemap_vehicle) raise AssertionError(f"unexpected url: {url}") with patch("iaai_scraper.discovery.sitemap._SitemapDownloader.fetch", side_effect=_fake_fetch): result = discover_vehicle_urls_from_sitemap("https://www.iaai.com/sitemap_index.xml") self.assertEqual(result, ["https://www.iaai.com/VehicleDetail/111~US"]) def test_discovery_uses_direct_probe_when_index_has_no_urls(self) -> None: index_xml = b"" sitemap_one = b""" https://www.iaai.com/VehicleDetail/111~US """ settings = Settings() settings.discovery.sitemap_direct_probe_limit = 2 settings.discovery.sitemap_direct_probe_stop_after_misses = 1 def _fake_fetch(url: str) -> SitemapFetchResult: if url.endswith("sitemap_index.xml"): return self._fetch_result(url, index_xml) if url.endswith("sitemap1.xml"): return self._fetch_result(url, sitemap_one) raise SitemapDiscoveryError("not found") with patch("iaai_scraper.discovery.sitemap._SitemapDownloader.fetch", side_effect=_fake_fetch): result = discover_vehicle_urls_from_sitemap_with_stats( "https://www.iaai.com/sitemap_index.xml", settings=settings, ) self.assertEqual(result.vehicle_urls, ["https://www.iaai.com/VehicleDetail/111~US"]) self.assertEqual(result.stats.direct_probe_hits, 1) def test_discovery_stats_report_direct_probe_hits(self) -> None: index_xml = b"" sitemap1 = b""" https://www.iaai.com/VehicleDetail/111~US """ def _fake_fetch(url: str): if url.endswith("sitemap_index.xml"): return self._fetch_result(url, index_xml) if url.endswith("sitemap1.xml"): return self._fetch_result(url, sitemap1) raise SitemapDiscoveryError("not found") with patch("iaai_scraper.discovery.sitemap._SitemapDownloader.fetch", side_effect=_fake_fetch): result = discover_vehicle_urls_from_sitemap_with_stats("https://www.iaai.com/sitemap_index.xml") self.assertEqual(result.vehicle_urls, ["https://www.iaai.com/VehicleDetail/111~US"]) self.assertEqual(result.stats.transport, "curl_cffi") self.assertEqual(result.stats.direct_probe_hits, 1) self.assertGreaterEqual(result.stats.direct_probe_misses, 1) def test_block_page_raises_discovery_error(self) -> None: settings = Settings() settings.discovery.sitemap_direct_probe_limit = 1 settings.discovery.sitemap_direct_probe_stop_after_misses = 1 with patch( "iaai_scraper.discovery.sitemap._SitemapDownloader.fetch", side_effect=SitemapBlockedError("Anti-bot page returned for https://www.iaai.com/sitemap_index.xml"), ): with self.assertRaises(SitemapDiscoveryError): discover_vehicle_urls_from_sitemap( "https://www.iaai.com/sitemap_index.xml", settings=settings, ) if __name__ == "__main__": unittest.main()