from __future__ import annotations import unittest from types import SimpleNamespace from unittest.mock import MagicMock from iaai_scraper.core.config import Settings from iaai_scraper.core.exceptions import AntiBotDetectedError, SiteStructureChangedError 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" s.database.url = "sqlite://" return IAAIScraper(s) def test_sync_vehicle_uses_db_record(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")}) result = scraper.sync_vehicle("https://www.iaai.com/VehicleDetail/111~US") self.assertEqual(result["status"], "success") self.assertIn("trace_id", result) scraper.persistence.upsert_car.assert_called_once() def test_only_new_routing_legacy_and_streaming(self) -> None: # Legacy path: only_new без listing_url. 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.get_existing_urls_and_ids = MagicMock(return_value=( {"https://www.iaai.com/VehicleDetail/111~US"}, {"iaai:222"}, )) scraper.collect_listing = MagicMock(return_value={ "vehicle_urls": [ "https://www.iaai.com/VehicleDetail/111~US", "https://www.iaai.com/VehicleDetail/222~US", "https://www.iaai.com/VehicleDetail/333~US", ] }) scraper.sync_batch = MagicMock(return_value={ "cars_upserted": 1, "cars_failed": 0, "images_upserted": 0, "failures": [], }) result = scraper.sync_listing(only_new=True) self.assertEqual(result["skipped_existing"], 2) self.assertEqual(result["cars_upserted"], 1) # Streaming path: only_new + listing_url. scraper2 = self._make_scraper() scraper2.persistence.create_tables = MagicMock() scraper2.persistence.start_sync_run = MagicMock(return_value=6) scraper2.persistence.finish_sync_run = MagicMock() scraper2.collect_listing = MagicMock(side_effect=AssertionError("legacy path should not be used")) scraper2._sync_listing_streaming = MagicMock(return_value={ "listing": {"vehicles_collected": 10, "early_stopped": False, "truncated_by_time_budget": False}, "total": 10, "skipped_existing": 0, "cars_upserted": 10, "cars_failed": 0, "images_upserted": 20, "failures": [], "all_listing_origin_urls": set(), }) result2 = scraper2.sync_listing( only_new=True, listing_url="https://www.iaai.com/Vehiclelisting/Cars?Make=TOYOTA", year_min=2020, year_max=2027, ) self.assertEqual(result2["cars_upserted"], 10) scraper2._sync_listing_streaming.assert_called_once() scraper2.collect_listing.assert_not_called() def test_segmented_sync_calls_per_segment_and_resumes(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() call_args_log: list[dict] = [] def _fake_sync_listing(**kwargs): call_args_log.append(kwargs) return { "status": "success", "cars_upserted": 5, "cars_failed": 0, "images_upserted": 10, "skipped_existing": 0, "listing": {"vehicles_collected": 50}, "failures": [], } scraper.sync_listing = MagicMock(side_effect=_fake_sync_listing) segments = [ {"make": "TOYOTA", "year_min": 2020, "year_max": 2027}, {"make": "FORD", "year_min": None, "year_max": None}, {"make": "HONDA", "year_min": None, "year_max": None}, ] # Resume: пропускаем TOYOTA, начинаем сразу с FORD. result = scraper.sync_listing_segmented( segments=segments, start_segment=1, ) self.assertEqual(result["segments_completed"], 2) # FORD + HONDA self.assertEqual(result["cars_upserted"], 10) # URL содержит бренд. self.assertIn("FORD", call_args_log[0]["listing_url"]) self.assertIn("HONDA", call_args_log[1]["listing_url"]) def test_segmented_sync_does_not_mark_full_scan_completed_when_segment_failed(self) -> None: scraper = self._make_scraper() scraper.sync_listing = MagicMock(side_effect=[ { "status": "failed", "full_scan_completed": False, "cars_upserted": 0, "cars_failed": 0, "images_upserted": 0, "skipped_existing": 0, "listing": {"vehicles_collected": 0}, "failures": [{"vehicle_url": "segment", "error": "resume failed"}], }, { "status": "success", "full_scan_completed": True, "cars_upserted": 1, "cars_failed": 0, "images_upserted": 0, "skipped_existing": 0, "listing": {"vehicles_collected": 10}, "failures": [], }, ]) result = scraper.sync_listing_segmented( segments=[ {"make": "EAGLE", "year_min": None, "year_max": None}, {"make": "FORD", "year_min": None, "year_max": None}, ] ) self.assertFalse(result["full_scan_completed"]) self.assertEqual(result["status"], "partial_success") def test_build_segment_listing_url(self) -> None: base = "https://www.iaai.com/Vehiclelisting/Cars" self.assertEqual( IAAIScraper._build_segment_listing_url(base, "TOYOTA"), "https://www.iaai.com/Vehiclelisting/Cars?Make=TOYOTA", ) self.assertEqual( IAAIScraper._build_segment_listing_url(base, "LAND ROVER"), "https://www.iaai.com/Vehiclelisting/Cars?Make=LAND%20ROVER", ) self.assertEqual(IAAIScraper._build_segment_listing_url(base, None), base) self.assertEqual(IAAIScraper._build_segment_listing_url(base, ""), base) def test_guard_and_protection_detection(self) -> None: with self.assertRaises(AntiBotDetectedError): IAAIScraper._raise_if_blocked_or_incomplete( {"dom_hints": {"has_captcha_text": True, "has_antibot_text": False}, "access_notes": {}, "vehicle_summary": {}}, "https://www.iaai.com/VehicleDetail/999~US", ) with self.assertRaises(SiteStructureChangedError): IAAIScraper._raise_if_blocked_or_incomplete( {"dom_hints": {"has_captcha_text": False, "has_antibot_text": False}, "access_notes": {"possible_captcha": False, "possible_antibot": False}, "vehicle_summary": {}}, "https://www.iaai.com/VehicleDetail/999~US", ) self.assertTrue(IAAIScraper._is_protection_or_network_error(RuntimeError("NS_ERROR_NET_INTERRUPT"))) self.assertTrue(IAAIScraper._is_protection_or_network_error(RuntimeError("captcha challenge"))) self.assertFalse(IAAIScraper._is_protection_or_network_error(RuntimeError("plain validation error"))) def test_close_resets_browser_state(self) -> None: scraper = self._make_scraper() http_pool = MagicMock() scraper._http_pool = http_pool scraper.context = MagicMock() scraper.browser = MagicMock() scraper.playwright = MagicMock() scraper.close() http_pool.clear.assert_called_once() self.assertIsNone(scraper.context) self.assertIsNone(scraper.browser) def test_recover_empty_listing_page(self) -> None: scraper = self._make_scraper() page = MagicMock() page_result = SimpleNamespace( vehicle_links=[SimpleNamespace(href="https://www.iaai.com/VehicleDetail/123~US")], ) scraper.listing_collector.collect_current_page = MagicMock(return_value=page_result) scraper.listing_collector.open_cars_listing = MagicMock() # Страница > 1: reload. _, urls = scraper._recover_empty_listing_page( page, page_number=5, all_raw_urls=[], seen_urls=set(), ) page.reload.assert_called_once() self.assertEqual(len(urls), 1) # Страница 1: переоткрытие листинга. page.reset_mock() _, urls = scraper._recover_empty_listing_page( page, page_number=1, all_raw_urls=[], seen_urls=set(), ) scraper.listing_collector.open_cars_listing.assert_called_once() page.reload.assert_not_called() def test_sync_listing_always_starts_from_page_one(self) -> None: scraper = self._make_scraper() scraper.settings.celery.batch_size = 1 page = MagicMock() page_result = SimpleNamespace( page_number=1, vehicle_links=[SimpleNamespace(href="https://www.iaai.com/VehicleDetail/999~US", lot_number="999")], next_page_detected=False, ) scraper._get_page_with_warmup = MagicMock(return_value=page) # Pagination-resume убран: _reopen_listing_and_resume не должен вызываться. scraper._reopen_listing_and_resume = MagicMock( side_effect=AssertionError("pagination resume must not be used") ) scraper.listing_collector.open_cars_listing = MagicMock() scraper.listing_collector.apply_filters = MagicMock(return_value={ "make": None, "model": None, "year_min": None, "year_max": None, }) scraper.listing_collector.collect_current_page = MagicMock(return_value=page_result) scraper._extract_page_urls = MagicMock(return_value=["https://www.iaai.com/VehicleDetail/999~US"]) scraper.sync_batch = MagicMock(return_value={ "cars_upserted": 1, "cars_failed": 0, "images_upserted": 0, "failures": [], }) result = scraper._sync_listing_streaming( make=None, model=None, lane="iaai_cars", limit=None, effective_only_new=False, started_at=0.0, listing_url="https://www.iaai.com/Vehiclelisting/Cars?Make=EAGLE", ) scraper.listing_collector.open_cars_listing.assert_called_once() scraper._reopen_listing_and_resume.assert_not_called() scraper.sync_batch.assert_called_once() self.assertEqual(result["cars_upserted"], 1) self.assertEqual(result["total"], 1) if __name__ == "__main__": unittest.main()