qananasikq/iaai-parser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tune vps config and update tests

Browse Source
This commit is contained in:
qananasikq
2026-04-15 21:58:10 +03:00
parent 37d01c4ba1
commit 3c3aea8eb3
8 changed files with 282 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files

238
tests/test_scraper.py
View File

@@ -28,104 +28,129 @@ class TestScraperSync(unittest.TestCase):
s.database.url = "sqlite://"
return IAAIScraper(s)
def test_sync_vehicle_uses_db_record_without_remapping(self) -> None:
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")})
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)
scraper.persistence.upsert_car.assert_called_once()
def test_sync_listing_uses_db_record_without_remapping(self) -> None:
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=(set(), set()))
scraper.collect_listing = MagicMock(return_value={"vehicle_urls": ["https://www.iaai.com/VehicleDetail/222~US"]})
scraper.sync_batch = MagicMock(return_value={
"cars_upserted": 1, "cars_failed": 0, "images_upserted": 1, "failures": [],
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",
]
})
result = scraper.sync_listing()
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)
self.assertEqual(result["cars_failed"], 0)
self.assertIn("trace_id", result)
self.assertIn("elapsed_seconds", result)
scraper.sync_batch.assert_called_once()
def test_sync_listing_respects_limit(self) -> None:
# 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()
# Проверка пути only_new с limit.
scraper._collect_listing_iterative = MagicMock(return_value=(
["https://www.iaai.com/VehicleDetail/222~US"],
["https://www.iaai.com/VehicleDetail/222~US",
"https://www.iaai.com/VehicleDetail/333~US"],
{"vehicle_urls": [], "pages_collected": 1, "early_stopped": False},
0,
))
scraper.sync_batch = MagicMock(return_value={
"cars_upserted": 1, "cars_failed": 0, "images_upserted": 1, "failures": [],
})
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(limit=1)
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},
]
# Должен уйти только один URL.
scraper.sync_batch.assert_called_once()
batch_urls = scraper.sync_batch.call_args[0][0]
self.assertEqual(len(batch_urls), 1)
# Resume: пропускаем TOYOTA, начинаем с FORD на стр. 5.
result = scraper.sync_listing_segmented(
segments=segments, start_segment=1, start_page=5,
)
def test_sync_listing_only_new_filters_existing_by_url_and_origin_id(self) -> None:
scraper = self._make_scraper()
self.assertEqual(result["segments_completed"], 2) # FORD + HONDA
self.assertEqual(result["cars_upserted"], 10)
# FORD: start_page=5, HONDA: start_page=1.
self.assertEqual(call_args_log[0]["start_page"], 5)
self.assertEqual(call_args_log[1]["start_page"], 1)
# URL содержит бренд.
self.assertIn("FORD", call_args_log[0]["listing_url"])
self.assertIn("HONDA", call_args_log[1]["listing_url"])
scraper.persistence.create_tables = MagicMock()
scraper.persistence.start_sync_run = MagicMock(return_value=5)
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"},
))
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)
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
]
})
# Возвращаем результат для одного нового авто.
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)
# В batch должен попасть только новый URL.
scraper.sync_batch.assert_called_once()
batch_urls = scraper.sync_batch.call_args[0][0]
self.assertEqual(len(batch_urls), 1)
self.assertIn("333", batch_urls[0])
scraper.persistence.get_existing_urls_and_ids.assert_called_once()
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()
@@ -134,85 +159,34 @@ class TestScraperSync(unittest.TestCase):
scraper.context = MagicMock()
scraper.browser = MagicMock()
scraper.playwright = MagicMock()
scraper.close()
http_pool.clear.assert_called_once()
self.assertIsNone(scraper._http_pool)
self.assertIsNone(scraper.context)
self.assertIsNone(scraper.browser)
self.assertIsNone(scraper.playwright)
def test_guard_raises_on_antibot_signals(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",
)
def test_guard_raises_on_empty_vehicle_page(self) -> None:
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",
)
def test_is_protection_or_network_error_detects_known_signals(self) -> None:
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_recover_empty_listing_page_reload_recovers_links(self) -> None:
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)
all_raw_urls: list[str] = []
seen_urls: set[str] = set()
recovered_result, recovered_urls = scraper._recover_empty_listing_page(
page,
page_number=5,
all_raw_urls=all_raw_urls,
seen_urls=seen_urls,
)
page.reload.assert_called_once()
self.assertIs(recovered_result, page_result)
self.assertEqual(recovered_urls, ["https://www.iaai.com/VehicleDetail/123~US"])
self.assertEqual(all_raw_urls, ["https://www.iaai.com/VehicleDetail/123~US"])
def test_recover_empty_listing_page_reopens_listing_for_first_page(self) -> None:
scraper = self._make_scraper()
page = MagicMock()
page_result = SimpleNamespace(
vehicle_links=[SimpleNamespace(href="https://www.iaai.com/VehicleDetail/456~US")],
)
scraper.listing_collector.open_cars_listing = MagicMock()
scraper.listing_collector.collect_current_page = MagicMock(return_value=page_result)
recovered_result, recovered_urls = scraper._recover_empty_listing_page(
page,
page_number=1,
all_raw_urls=[],
seen_urls=set(),
# Страница > 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)
scraper.listing_collector.open_cars_listing.assert_called_once_with(page)
# Страница 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()
self.assertIs(recovered_result, page_result)
self.assertEqual(recovered_urls, ["https://www.iaai.com/VehicleDetail/456~US"])
if __name__ == "__main__":