154 lines
6.8 KiB
Python
154 lines
6.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock
|
|
|
|
from iaai_scraper.core.config import Settings
|
|
from iaai_scraper.scraper import IAAIScraper
|
|
from iaai_scraper.storage.schemas import CarRecord
|
|
|
|
|
|
def make_db_record(origin_id: str) -> dict[str, object]:
|
|
return CarRecord(
|
|
parser_id=f"iaai:{origin_id}",
|
|
brand="Toyota",
|
|
model="Camry",
|
|
origin_url=f"https://www.iaai.com/VehicleDetail/{origin_id}~US",
|
|
origin_id=origin_id,
|
|
slug=f"toyota-camry-{origin_id}",
|
|
).model_dump(mode="json")
|
|
|
|
|
|
class TestScraperSync(unittest.TestCase):
|
|
def _make_scraper(self) -> IAAIScraper:
|
|
s = Settings()
|
|
s.log_level = "CRITICAL"
|
|
return IAAIScraper(s)
|
|
|
|
def test_sync_vehicle_uses_db_record_without_remapping(self) -> None:
|
|
scraper = self._make_scraper()
|
|
|
|
scraper.persistence.create_tables = MagicMock()
|
|
scraper.persistence.start_sync_run = MagicMock(return_value=1)
|
|
scraper.persistence.finish_sync_run = MagicMock()
|
|
scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 0})
|
|
|
|
scraper.scrape_vehicle_detail = MagicMock(return_value={"db_record": make_db_record("111")})
|
|
scraper.car_mapper.map_to_car_record = MagicMock(side_effect=AssertionError("should not be called"))
|
|
|
|
result = scraper.sync_vehicle("https://www.iaai.com/VehicleDetail/111~US")
|
|
|
|
self.assertEqual(result["status"], "success")
|
|
self.assertIn("trace_id", result)
|
|
self.assertIn("elapsed_seconds", result)
|
|
self.assertEqual(scraper.persistence.upsert_car.call_count, 1)
|
|
|
|
def test_sync_listing_uses_db_record_without_remapping(self) -> None:
|
|
scraper = self._make_scraper()
|
|
|
|
scraper.persistence.create_tables = MagicMock()
|
|
scraper.persistence.start_sync_run = MagicMock(return_value=2)
|
|
scraper.persistence.finish_sync_run = MagicMock()
|
|
scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 1})
|
|
scraper.persistence.get_existing_origin_urls = MagicMock(return_value=set())
|
|
scraper.persistence.get_existing_origin_ids = MagicMock(return_value=set())
|
|
|
|
scraper.collect_listing = MagicMock(return_value={"vehicle_urls": ["https://www.iaai.com/VehicleDetail/222~US"]})
|
|
page = MagicMock()
|
|
scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("222")})
|
|
scraper._get_page = MagicMock(return_value=page)
|
|
scraper.car_mapper.map_to_car_record = MagicMock(side_effect=AssertionError("should not be called"))
|
|
|
|
result = scraper.sync_listing()
|
|
|
|
self.assertEqual(result["cars_upserted"], 1)
|
|
self.assertEqual(result["cars_failed"], 0)
|
|
self.assertIn("trace_id", result)
|
|
self.assertIn("elapsed_seconds", result)
|
|
self.assertEqual(scraper.persistence.upsert_car.call_count, 1)
|
|
page.close.assert_called_once()
|
|
|
|
def test_sync_listing_respects_limit(self) -> None:
|
|
scraper = self._make_scraper()
|
|
|
|
scraper.persistence.create_tables = MagicMock()
|
|
scraper.persistence.start_sync_run = MagicMock(return_value=3)
|
|
scraper.persistence.finish_sync_run = MagicMock()
|
|
scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 1})
|
|
scraper.persistence.get_existing_origin_urls = MagicMock(return_value=set())
|
|
scraper.persistence.get_existing_origin_ids = MagicMock(return_value=set())
|
|
scraper.collect_listing = MagicMock(return_value={
|
|
"vehicle_urls": [
|
|
"https://www.iaai.com/VehicleDetail/222~US",
|
|
"https://www.iaai.com/VehicleDetail/333~US",
|
|
]
|
|
})
|
|
scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("222")})
|
|
scraper._get_page = MagicMock(return_value=MagicMock())
|
|
|
|
scraper.sync_listing(limit=1)
|
|
|
|
self.assertEqual(scraper._scrape_on_page.call_count, 1)
|
|
|
|
def test_sync_listing_does_not_count_skipped_as_upserted(self) -> None:
|
|
scraper = self._make_scraper()
|
|
|
|
scraper.persistence.create_tables = MagicMock()
|
|
scraper.persistence.start_sync_run = MagicMock(return_value=4)
|
|
scraper.persistence.finish_sync_run = MagicMock()
|
|
scraper.persistence.upsert_car = MagicMock(return_value={"action": "skipped", "images_upserted": 0})
|
|
scraper.persistence.get_existing_origin_urls = MagicMock(return_value=set())
|
|
scraper.persistence.get_existing_origin_ids = MagicMock(return_value=set())
|
|
scraper.collect_listing = MagicMock(return_value={"vehicle_urls": ["https://www.iaai.com/VehicleDetail/444~US"]})
|
|
scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("444")})
|
|
scraper._get_page = MagicMock(return_value=MagicMock())
|
|
|
|
result = scraper.sync_listing()
|
|
|
|
self.assertEqual(result["cars_upserted"], 0)
|
|
|
|
def test_sync_listing_only_new_filters_existing_by_url_and_origin_id(self) -> None:
|
|
scraper = self._make_scraper()
|
|
|
|
scraper.persistence.create_tables = MagicMock()
|
|
scraper.persistence.start_sync_run = MagicMock(return_value=5)
|
|
scraper.persistence.finish_sync_run = MagicMock()
|
|
scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 0})
|
|
scraper.persistence.get_existing_origin_urls = MagicMock(return_value={"https://www.iaai.com/VehicleDetail/111~US"})
|
|
scraper.persistence.get_existing_origin_ids = MagicMock(return_value={"222"})
|
|
|
|
scraper.collect_listing = MagicMock(return_value={
|
|
"vehicle_urls": [
|
|
"https://www.iaai.com/VehicleDetail/111~US", # exists by URL
|
|
"https://www.iaai.com/VehicleDetail/222~US", # exists by ID
|
|
"https://www.iaai.com/VehicleDetail/333~US", # new
|
|
]
|
|
})
|
|
page = MagicMock()
|
|
scraper._get_page = MagicMock(return_value=page)
|
|
scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("333")})
|
|
|
|
result = scraper.sync_listing(only_new=True)
|
|
|
|
self.assertEqual(result["skipped_existing"], 2)
|
|
self.assertEqual(result["cars_upserted"], 1)
|
|
self.assertEqual(scraper._scrape_on_page.call_count, 1)
|
|
scraper.persistence.get_existing_origin_urls.assert_called_once()
|
|
scraper.persistence.get_existing_origin_ids.assert_called_once()
|
|
|
|
def test_close_resets_browser_state(self) -> None:
|
|
scraper = self._make_scraper()
|
|
scraper.context = MagicMock()
|
|
scraper.browser = MagicMock()
|
|
scraper.playwright = MagicMock()
|
|
|
|
scraper.close()
|
|
|
|
self.assertIsNone(scraper.context)
|
|
self.assertIsNone(scraper.browser)
|
|
self.assertIsNone(scraper.playwright)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|