218 lines
9.5 KiB
Python
218 lines
9.5 KiB
Python
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"""
|
|
<sitemapindex xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
|
|
<sitemap><loc>https://www.iaai.com/sitemap-a.xml</loc></sitemap>
|
|
<sitemap><loc>https://www.iaai.com/sitemap-b.xml</loc></sitemap>
|
|
<sitemap><loc>https://www.iaai.com/sitemapauctions1.xml</loc></sitemap>
|
|
</sitemapindex>
|
|
"""
|
|
sitemap_a = b"""
|
|
<urlset xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
|
|
<url><loc>https://www.iaai.com/VehicleDetail/111~US</loc></url>
|
|
<url><loc>https://www.iaai.com/VehicleDetail/222~US?x=1</loc></url>
|
|
</urlset>
|
|
"""
|
|
sitemap_b = b"""
|
|
<urlset xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
|
|
<url><loc>https://www.iaai.com/VehicleDetail/222~US?y=2</loc></url>
|
|
<url><loc>https://www.iaai.com/VehicleDetail/333~US</loc></url>
|
|
</urlset>
|
|
"""
|
|
|
|
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"<sitemapindex xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\"></sitemapindex>"
|
|
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"""
|
|
<urlset xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
|
|
<url><loc>https://www.iaai.com/VehicleDetail/111~US</loc></url>
|
|
</urlset>
|
|
"""
|
|
sitemap_non_vehicle = b"""
|
|
<urlset xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
|
|
<url><loc>https://www.iaai.com/SalesList/111~US/04222026</loc></url>
|
|
</urlset>
|
|
"""
|
|
index_xml = b"""
|
|
<sitemapindex xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
|
|
<sitemap><loc>https://www.iaai.com/sitemap1.xml</loc></sitemap>
|
|
<sitemap><loc>https://www.iaai.com/sitemapbranches1.xml</loc></sitemap>
|
|
<sitemap><loc>https://www.iaai.com/sitemap2.xml</loc></sitemap>
|
|
<sitemap><loc>https://www.iaai.com/sitemap3.xml</loc></sitemap>
|
|
</sitemapindex>
|
|
"""
|
|
|
|
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"<sitemapindex>"
|
|
b"<sitemap><loc>https://www.iaai.com/sitemap1.xml</loc></sitemap>"
|
|
b"<broken"
|
|
)
|
|
sitemap_vehicle = b"""
|
|
<urlset xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
|
|
<url><loc>https://www.iaai.com/VehicleDetail/111~US</loc></url>
|
|
</urlset>
|
|
"""
|
|
|
|
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"<sitemapindex xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\"></sitemapindex>"
|
|
sitemap_one = b"""
|
|
<urlset xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
|
|
<url><loc>https://www.iaai.com/VehicleDetail/111~US</loc></url>
|
|
</urlset>
|
|
"""
|
|
|
|
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"<sitemapindex xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\"></sitemapindex>"
|
|
sitemap1 = b"""
|
|
<urlset xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
|
|
<url><loc>https://www.iaai.com/VehicleDetail/111~US</loc></url>
|
|
</urlset>
|
|
"""
|
|
|
|
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()
|