qananasikq/dubizzle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add helper scripts

Browse Source
This commit is contained in:
qananasikq
2026-04-24 20:47:18 +03:00
commit cec1288652
70 changed files with 14249 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
tests/conftest.py Normal file
View File

@@ -0,0 +1,7 @@
from __future__ import annotations
import logging
def pytest_configure() -> None:
logging.disable(logging.CRITICAL)

127
tests/test_db.py Normal file
View File

@@ -0,0 +1,127 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from sqlalchemy import select
from dubizzle_scraper.core.config import Settings
from dubizzle_scraper.storage.db import PersistenceService
from dubizzle_scraper.storage.models import Car, Image, SyncRun
from dubizzle_scraper.storage.schemas import CarRecord, ImageRecord
class TestPersistenceServiceIntegration(unittest.TestCase):
def setUp(self) -> None:
self.tmp_dir = tempfile.TemporaryDirectory()
db_path = Path(self.tmp_dir.name) / "test.sqlite"
self.settings = Settings()
self.settings.database.url = f"sqlite:///{db_path.as_posix()}"
self.settings.database.echo = False
self.persistence = PersistenceService(self.settings)
self.persistence.create_tables()
def tearDown(self) -> None:
self.persistence.engine.dispose()
self.tmp_dir.cleanup()
@staticmethod
def _record(origin_id: str, *, price: int = 1000) -> CarRecord:
return CarRecord(
parser_id=f"dubizzle:{origin_id}",
brand="Toyota",
model="Camry",
year=2014,
price=price,
origin_url=f"https://www.dubizzle.com/VehicleDetail/{origin_id}~US",
origin_id=origin_id,
slug=f"toyota-camry-{origin_id}",
images=[
ImageRecord(
fullres_image="https://vis.dubizzle.com/resizer?imageKeys=1&width=845&height=633",
preview_image="https://vis.dubizzle.com/resizer?imageKeys=1&width=400&height=300",
order_index=0,
)
],
)
def test_insert_update_and_skip_flow(self) -> None:
first = self._record("777", price=1000)
inserted = self.persistence.upsert_car(first)
self.assertEqual(inserted["action"], "inserted")
self.assertEqual(inserted["images_upserted"], 1)
same = self._record("777", price=1000)
updated_same = self.persistence.upsert_car(same)
self.assertEqual(updated_same["action"], "updated")
self.assertEqual(updated_same["images_upserted"], 1)
changed = self._record("777", price=1500)
updated = self.persistence.upsert_car(changed)
self.assertEqual(updated["action"], "updated")
self.assertEqual(updated["images_upserted"], 1)
with self.persistence.session_scope() as session:
cars = session.execute(select(Car)).scalars().all()
images = session.execute(select(Image)).scalars().all()
self.assertEqual(len(cars), 1)
self.assertEqual(cars[0].price, 1500)
self.assertEqual(len(images), 1)
def test_update_replaces_old_images(self) -> None:
first = self._record("888")
self.persistence.upsert_car(first)
second = self._record("888")
second.images = [
ImageRecord(
fullres_image="https://vis.dubizzle.com/resizer?imageKeys=2&width=845&height=633",
preview_image="https://vis.dubizzle.com/resizer?imageKeys=2&width=400&height=300",
order_index=0,
)
]
self.persistence.upsert_car(second)
with self.persistence.session_scope() as session:
images = session.execute(select(Image)).scalars().all()
self.assertEqual(len(images), 1)
self.assertIn("imageKeys=2", images[0].fullres_image)
def test_start_sync_run_marks_stale_running_runs_as_failed(self) -> None:
first_run_id = self.persistence.start_sync_run("lane-a")
second_run_id = self.persistence.start_sync_run("lane-b")
self.assertNotEqual(first_run_id, second_run_id)
with self.persistence.session_scope() as session:
first = session.get(SyncRun, first_run_id)
second = session.get(SyncRun, second_run_id)
self.assertEqual(first.status, "failed")
self.assertIsNotNone(first.finished_at)
self.assertEqual(second.status, "running")
def test_upsert_falls_back_to_origin_url_to_prevent_duplicates(self) -> None:
first = self._record("OLD-ID")
first.origin_url = "https://www.dubizzle.com/VehicleDetail/45089484~US"
self.persistence.upsert_car(first)
second = self._record("NEW-ID")
second.origin_url = "https://www.dubizzle.com/VehicleDetail/45089484~US"
result = self.persistence.upsert_car(second)
self.assertEqual(result["action"], "updated")
with self.persistence.session_scope() as session:
cars = session.execute(select(Car)).scalars().all()
self.assertEqual(len(cars), 1)
self.assertEqual(cars[0].origin_id, "NEW-ID")
if __name__ == "__main__":
unittest.main()

68
tests/test_listing.py Normal file
View File

@@ -0,0 +1,68 @@
from __future__ import annotations
import unittest
from dubizzle_scraper.browser.pace import HumanPacer
from dubizzle_scraper.core.config import Settings
from dubizzle_scraper.browser.listing import ListingCollector
class _FakePage:
def __init__(self, counts: dict[str, int]) -> None:
self._counts = counts
self._evaluate_result = False
self._evaluate_values: list[object] = []
class _Locator:
def __init__(self, count_value: int) -> None:
self._count_value = count_value
def count(self) -> int:
return self._count_value
def locator(self, selector: str) -> "_FakePage._Locator":
return _FakePage._Locator(self._counts.get(selector, 0))
def evaluate(self, _script: str):
if self._evaluate_values:
return self._evaluate_values.pop(0)
return self._evaluate_result
class TestListingUnit(unittest.TestCase):
def test_pagination_detection_and_page_number(self) -> None:
# Есть кнопка Next → True.
self.assertTrue(ListingCollector._has_next_page(_FakePage({"a[aria-label*='Next']": 1})))
# Нет селекторов → False.
self.assertFalse(ListingCollector._has_next_page(_FakePage({})))
# Числовая пагинация через JS → True только если evaluate явно нашёл next.
page = _FakePage({})
page._evaluate_result = True
self.assertTrue(ListingCollector._has_next_page(page))
# Номер текущей страницы.
page2 = _FakePage({})
page2._evaluate_values = [5]
self.assertEqual(ListingCollector._get_current_page_number(page2), 5)
def test_extract_vehicle_links_from_html_finds_detail_urls(self) -> None:
collector = ListingCollector(Settings(), HumanPacer(Settings()))
html = """
<div>
<h4><a href=\"/VehicleDetail/45184893~US\">Car 1</a></h4>
<script>window.__data = {\"href\":\"\\/VehicleDetail\\/45171480~US\"}</script>
</div>
"""
links = collector._extract_vehicle_links_from_html(html)
self.assertEqual(
links,
[
("https://www.dubizzle.com/VehicleDetail/45184893~US", "45184893"),
("https://www.dubizzle.com/VehicleDetail/45171480~US", "45171480"),
],
)
if __name__ == "__main__":
unittest.main()

181
tests/test_mappers.py Normal file
View File

@@ -0,0 +1,181 @@
from __future__ import annotations
import unittest
from dubizzle_scraper.parsing.mapper import CarMapper
class TestCarMapper(unittest.TestCase):
def setUp(self) -> None:
self.mapper = CarMapper()
def test_extracts_make_model_from_algolia_details(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://dubizzle.com/s/DOBI7J3",
vehicle_summary={
"id": 16810399,
"details_v2": {
"tertiary": [
{"slug": "make", "value": {"en": "Toyota", "ar": "تويوتا"}},
{"slug": "model", "value": {"en": "Land Cruiser", "ar": "لاند كروزر"}},
]
},
"category_v2": {
"names_en": ["Motors", "Used Cars", "Toyota", "Land Cruiser"],
"slug_paths": [
"motors",
"motors/used-cars",
"motors/used-cars/toyota",
"motors/used-cars/toyota/land-cruiser",
],
},
},
payload_insights={"vehicle_core": {}, "pricing": {}, "damage": {}, "auction": {}, "images": {}},
)
self.assertEqual(record.brand, "Toyota")
self.assertEqual(record.model, "Land Cruiser")
def test_extracts_make_model_from_legacy_details(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://dubizzle.com/s/DOBIlegacy",
vehicle_summary={
"objectID": "item:legacy:1",
"details": {
"Make": {"en": {"label": "Make", "value": "Honda"}},
"Model": {"en": {"label": "Model", "value": "Civic"}},
},
},
payload_insights={"vehicle_core": {}, "pricing": {}, "damage": {}, "auction": {}, "images": {}},
)
self.assertEqual(record.brand, "Honda")
self.assertEqual(record.model, "Civic")
def test_extracts_extended_algolia_fields(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://dubizzle.com/s/DOBGaGs",
vehicle_summary={
"id": 16345114,
"year": 2012,
"price": 59000.0,
"kilometers": 91800,
"seller_type": "DL",
"site": {"en": "Dubai"},
"category_v2": {
"names_en": ["Motors", "Used Cars", "Porsche", "Cayenne"],
"slug_paths": [
"motors",
"motors/used-cars",
"motors/used-cars/porsche",
"motors/used-cars/porsche/cayenne",
],
},
"details": {
"Body Type": {"en": {"value": "SUV"}},
"Transmission Type": {"en": {"value": "Automatic Transmission"}},
"Exterior Color": {"en": {"value": "Brown"}},
"Steering Side": {"en": {"value": "Left Hand"}},
"Engine Capacity (cc)": {"en": {"value": "4800 cc"}},
"Make": {"en": {"value": "Porsche"}},
"Model": {"en": {"value": "Cayenne"}},
},
"photo_mains": ["https://dbz-images.dubizzle.com/images/example.jpeg?impolicy=dpv"],
},
payload_insights={"vehicle_core": {}, "pricing": {}, "damage": {}, "auction": {}, "images": {}},
)
self.assertEqual(record.brand, "Porsche")
self.assertEqual(record.model, "Cayenne")
self.assertEqual(record.price, 59000)
self.assertEqual(record.mileage, 91800)
self.assertEqual(record.color, "brown")
self.assertEqual(record.body_type, "SUV")
self.assertEqual(record.gearbox, "AT")
self.assertEqual(record.steering_wheel, "LEFT")
self.assertEqual(record.engine_volume, 4800)
self.assertEqual(record.country, "AE")
def test_deduplicates_images_by_image_key(self) -> None:
urls = [
"https://vis.dubizzle.com/resizer?imageKeys=1&width=200&height=150",
"https://vis.dubizzle.com/resizer?imageKeys=1&width=845&height=633",
"https://vis.dubizzle.com/resizer?imageKeys=2&width=400&height=300",
]
record = self.mapper.map_to_car_record(
vehicle_url="https://www.dubizzle.com/VehicleDetail/123~US",
vehicle_summary={"make": "Honda", "model": "Civic", "image_urls": urls},
payload_insights={"vehicle_core": {}, "pricing": {}, "damage": {}, "auction": {}, "images": {}},
)
self.assertEqual(len(record.images), 2)
self.assertIn("width=845", record.images[0].fullres_image)
self.assertIn("height=633", record.images[0].fullres_image)
def test_no_damage_marker_is_not_damaged(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://www.dubizzle.com/VehicleDetail/123~US",
vehicle_summary={"make": "Ford", "model": "Focus"},
payload_insights={
"vehicle_core": {},
"pricing": {},
"damage": {"primary": "normal wear"},
"auction": {},
"images": {},
},
)
self.assertFalse(record.is_damaged)
def test_unknown_empty_values_and_normalization(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://www.dubizzle.com/VehicleDetail/999~US",
vehicle_summary={"make": " ", "model": None, "drive": "???", "gearbox": "unknown"},
payload_insights={"vehicle_core": {}, "pricing": {}, "damage": {}, "auction": {}, "images": {}},
)
self.assertEqual(record.brand, "UNKNOWN")
self.assertEqual(record.model, "UNKNOWN")
self.assertEqual(record.drive, "NA")
self.assertEqual(record.gearbox, "NA")
# Нормализация регистра и пробелов.
record2 = self.mapper.map_to_car_record(
vehicle_url="https://www.dubizzle.com/VehicleDetail/888~US",
vehicle_summary={"make": "Honda", "model": "Civic", "drive": " Front Wheel Drive ", "gearbox": " AUTOMATIC "},
payload_insights={"vehicle_core": {}, "pricing": {}, "damage": {}, "auction": {}, "images": {}},
)
self.assertEqual(record2.drive, "FWD")
self.assertEqual(record2.gearbox, "AT")
def test_price_and_currency_parsing(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://www.dubizzle.com/VehicleDetail/777~US",
vehicle_summary={"make": "Toyota", "model": "Corolla"},
payload_insights={
"vehicle_core": {},
"pricing": {"buy_now": "USD 4,500 - 5,200"},
"damage": {},
"auction": {},
"images": {},
},
)
self.assertEqual(record.price, 5200)
record2 = self.mapper.map_to_car_record(
vehicle_url="https://www.dubizzle.com/VehicleDetail/778~US",
vehicle_summary={"make": "Toyota", "model": "Corolla"},
payload_insights={
"vehicle_core": {},
"pricing": {"buy_now": "€4.500,00"},
"damage": {},
"auction": {},
"images": {},
},
)
self.assertEqual(record2.currency, "EUR")
self.assertEqual(record2.price, 4500)
if __name__ == "__main__":
unittest.main()

54
tests/test_parser.py Normal file
View File

@@ -0,0 +1,54 @@
from __future__ import annotations
import unittest
from dubizzle_scraper.parsing.parser import VehicleParser
class TestVehicleParserUnit(unittest.TestCase):
def setUp(self) -> None:
self.parser = VehicleParser()
def test_parse_dom_pairs_and_title(self) -> None:
dom_text = """
Stock #:
45089484
Primary Damage:
Front End
Odometer:
50,123 mi (Actual)
"""
result = self.parser._parse_dom_key_value_pairs(dom_text)
self.assertEqual(result.get("lot_number"), "45089484")
self.assertEqual(result.get("primary_damage"), "Front End")
self.assertEqual(result.get("odometer"), "50,123 mi (Actual)")
parsed = self.parser._parse_title_for_year_make_model("2014 TOYOTA CAMRY for sale", "")
self.assertEqual(parsed["year"], "2014")
self.assertEqual(parsed["make"], "TOYOTA")
self.assertEqual(parsed["model"], "CAMRY")
def test_extract_image_urls_filters_and_deduplicates(self) -> None:
vehicle_url = "https://www.dubizzle.com/VehicleDetail/45089484~US"
payloads = [{"imageUrls": [
"https://vis.dubizzle.com/resizer?imageKeys=45089484~SID1&width=845&height=633",
"https://vis.dubizzle.com/resizer?imageKeys=45089484~SID1&width=845&height=633",
"https://vis.dubizzle.com/resizer?imageKeys=99999999~SID2&width=845&height=633",
]}]
urls = self.parser._extract_image_urls(payloads, "", vehicle_url)
self.assertEqual(len(urls), 1)
self.assertIn("45089484", urls[0])
def test_dom_hints_and_access_notes_detect_antibot(self) -> None:
hints = self.parser._dom_hints("Please verify you are human. CAPTCHA. Incapsula access denied.")
self.assertTrue(hints["has_captcha_text"])
self.assertTrue(hints["has_antibot_text"])
summary = {"vin": "", "image_urls": [], "note": "Incapsula access denied. Verify you are human."}
notes = self.parser._build_access_notes(summary, [])
self.assertTrue(notes["possible_captcha"])
self.assertTrue(notes["possible_antibot"])
if __name__ == "__main__":
unittest.main()

79
tests/test_resilience.py Normal file
View File

@@ -0,0 +1,79 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from dubizzle_scraper.scraper import DUBIZZLEScraper
from dubizzle_scraper.core.config import Settings
from dubizzle_scraper.worker import tasks
class TestResilience(unittest.TestCase):
def _make_scraper(self) -> DUBIZZLEScraper:
s = Settings()
s.log_level = "CRITICAL"
s.database.url = "sqlite://"
return DUBIZZLEScraper(s)
def test_update_task_progress_updates_global_marker(self) -> None:
redis_client = MagicMock()
pipe = MagicMock()
redis_client.pipeline.return_value = pipe
tasks._update_task_progress(
redis_client,
task_id="task-abc",
stage="batch_upserted",
ttl_seconds=180,
cars_upserted=10,
)
redis_client.pipeline.assert_called_once()
self.assertEqual(pipe.set.call_count, 2)
first_call = pipe.set.call_args_list[0]
second_call = pipe.set.call_args_list[1]
self.assertEqual(first_call.args[0], tasks._task_progress_key("task-abc"))
self.assertEqual(second_call.args[0], tasks.GLOBAL_PROGRESS_TS_KEY)
pipe.execute.assert_called_once()
def test_streaming_sync_stops_cleanly_when_page_stays_empty(self) -> None:
scraper = self._make_scraper()
scraper.settings.celery.batch_size = 50
page = MagicMock()
empty_page_result = SimpleNamespace(
page_number=1,
vehicle_links=[],
next_page_detected=True,
)
scraper._open_listing_for_stream = MagicMock(return_value=(
page,
{"make": None, "model": None, "year_min": None, "year_max": None},
))
scraper.listing_collector.collect_current_page = MagicMock(return_value=empty_page_result)
scraper._recover_empty_listing_page = MagicMock(return_value=(empty_page_result, []))
scraper.sync_batch = MagicMock()
result = scraper._sync_listing_streaming(
make=None,
model=None,
lane="dubizzle_cars",
limit=None,
effective_only_new=False,
started_at=0.0,
listing_url="https://www.dubizzle.com/Vehiclelisting/Cars?Make=TEST",
)
self.assertEqual(result["total"], 0)
self.assertEqual(result["cars_upserted"], 0)
self.assertEqual(result["cars_failed"], 0)
self.assertEqual(result["listing"]["pages_collected"], 1)
scraper._recover_empty_listing_page.assert_called_once()
scraper.sync_batch.assert_not_called()
if __name__ == "__main__":
unittest.main()

319
tests/test_scraper.py Normal file
View File

@@ -0,0 +1,319 @@
from __future__ import annotations
import unittest
from concurrent.futures import TimeoutError as FuturesTimeoutError
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from dubizzle_scraper.core.config import Settings
from dubizzle_scraper.core.exceptions import AntiBotDetectedError, SiteStructureChangedError
from dubizzle_scraper.scraper import DUBIZZLEScraper
from dubizzle_scraper.storage.schemas import CarRecord
def make_db_record(origin_id: str) -> dict[str, object]:
return CarRecord(
parser_id=f"dubizzle:{origin_id}",
brand="Toyota",
model="Camry",
origin_url=f"https://www.dubizzle.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) -> DUBIZZLEScraper:
s = Settings()
s.log_level = "CRITICAL"
s.database.url = "sqlite://"
return DUBIZZLEScraper(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.dubizzle.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.dubizzle.com/VehicleDetail/111~US"}, {"dubizzle:222"},
))
scraper.collect_listing = MagicMock(return_value={
"vehicle_urls": [
"https://www.dubizzle.com/VehicleDetail/111~US",
"https://www.dubizzle.com/VehicleDetail/222~US",
"https://www.dubizzle.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.dubizzle.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.dubizzle.com/Vehiclelisting/Cars"
self.assertEqual(
DUBIZZLEScraper._build_segment_listing_url(base, "TOYOTA"),
"https://www.dubizzle.com/Vehiclelisting/Cars?Make=TOYOTA",
)
self.assertEqual(
DUBIZZLEScraper._build_segment_listing_url(base, "LAND ROVER"),
"https://www.dubizzle.com/Vehiclelisting/Cars?Make=LAND%20ROVER",
)
self.assertEqual(DUBIZZLEScraper._build_segment_listing_url(base, None), base)
self.assertEqual(DUBIZZLEScraper._build_segment_listing_url(base, ""), base)
def test_guard_and_protection_detection(self) -> None:
with self.assertRaises(AntiBotDetectedError):
DUBIZZLEScraper._raise_if_blocked_or_incomplete(
{"dom_hints": {"has_captcha_text": True, "has_antibot_text": False},
"access_notes": {}, "vehicle_summary": {}},
"https://www.dubizzle.com/VehicleDetail/999~US",
)
with self.assertRaises(SiteStructureChangedError):
DUBIZZLEScraper._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.dubizzle.com/VehicleDetail/999~US",
)
self.assertTrue(DUBIZZLEScraper._is_protection_or_network_error(RuntimeError("NS_ERROR_NET_INTERRUPT")))
self.assertTrue(DUBIZZLEScraper._is_protection_or_network_error(RuntimeError("captcha challenge")))
self.assertFalse(DUBIZZLEScraper._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.dubizzle.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.dubizzle.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.dubizzle.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="dubizzle_cars",
limit=None,
effective_only_new=False,
started_at=0.0,
listing_url="https://www.dubizzle.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)
def test_sync_batch_timeout_does_not_duplicate_already_processed_urls(self) -> None:
scraper = self._make_scraper()
scraper.persistence.upsert_cars_batch = MagicMock(return_value={
"inserted": 1,
"updated": 0,
"images_upserted": 0,
})
urls = [
"https://www.dubizzle.com/VehicleDetail/111~US",
"https://www.dubizzle.com/VehicleDetail/222~US",
"https://www.dubizzle.com/VehicleDetail/333~US",
]
first_record = CarRecord.model_validate(make_db_record("111"))
class _FakeExecutor:
def __init__(self, *args, **kwargs):
pass
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def map(self, fn, iterable, timeout=None): # noqa: ARG002
yield 0, first_record
raise FuturesTimeoutError()
with patch("dubizzle_scraper.core.runtime_config.RuntimeFiltersConfig.is_empty", return_value=True), \
patch("dubizzle_scraper.scraper.ThreadPoolExecutor", _FakeExecutor), \
patch.object(scraper, "_browser_fallback_parallel", return_value={
"records": [],
"failures": [],
"cars_failed": 0,
"protection_events": 0,
}) as fallback_mock:
result = scraper.sync_batch(urls)
self.assertEqual(result["cars_upserted"], 1)
fallback_urls = fallback_mock.call_args.args[0]
self.assertEqual(len(fallback_urls), 2)
self.assertEqual({u for u, _ in fallback_urls}, {urls[1], urls[2]})
self.assertNotIn(urls[0], {u for u, _ in fallback_urls})
if __name__ == "__main__":
unittest.main()

81
tests/test_self_heal.py Normal file
View File

@@ -0,0 +1,81 @@
from __future__ import annotations
import unittest
from unittest.mock import MagicMock, patch
from dubizzle_scraper.worker import self_heal
class TestSelfHeal(unittest.TestCase):
def test_read_last_progress_ts_uses_global_key(self) -> None:
redis_client = MagicMock()
redis_client.get.return_value = "1776800000"
ts = self_heal._read_last_progress_ts(redis_client)
self.assertEqual(ts, 1776800000)
redis_client.scan_iter.assert_not_called()
def test_read_last_progress_ts_fallbacks_to_task_progress_keys(self) -> None:
redis_client = MagicMock()
redis_client.get.side_effect = lambda key: {
self_heal.GLOBAL_PROGRESS_TS_KEY: None,
"dubizzle:state:task_progress:a": '{"ts": 100}',
"dubizzle:state:task_progress:b": '{"ts": 250}',
"dubizzle:state:task_progress:c": '{"ts": 150}',
}.get(key)
redis_client.scan_iter.return_value = [
"dubizzle:state:task_progress:a",
"dubizzle:state:task_progress:b",
"dubizzle:state:task_progress:c",
]
ts = self_heal._read_last_progress_ts(redis_client)
self.assertEqual(ts, 250)
def test_read_last_progress_ts_ignores_broken_payloads(self) -> None:
redis_client = MagicMock()
redis_client.get.side_effect = lambda key: {
self_heal.GLOBAL_PROGRESS_TS_KEY: None,
"dubizzle:state:task_progress:a": "{bad-json}",
"dubizzle:state:task_progress:b": '{"foo": "bar"}',
}.get(key)
redis_client.scan_iter.return_value = [
"dubizzle:state:task_progress:a",
"dubizzle:state:task_progress:b",
]
ts = self_heal._read_last_progress_ts(redis_client)
self.assertIsNone(ts)
@patch("dubizzle_scraper.worker.self_heal.time.sleep", return_value=None)
@patch("dubizzle_scraper.worker.self_heal.os.kill")
@patch("builtins.open")
def test_kill_worker_process_sends_term_and_kill(self, open_mock, kill_mock, _sleep_mock) -> None:
open_mock.return_value.__enter__.return_value.read.return_value = "123"
# SIGTERM -> process alive check (pid,0) -> SIGKILL
kill_mock.side_effect = [None, None, None]
self_heal._kill_worker_process()
self.assertEqual(kill_mock.call_args_list[0].args[0], 123)
self.assertEqual(kill_mock.call_args_list[1].args, (123, 0))
self.assertEqual(kill_mock.call_args_list[2].args[0], 123)
@patch("dubizzle_scraper.worker.self_heal.time.sleep", return_value=None)
@patch("dubizzle_scraper.worker.self_heal.os.kill")
@patch("builtins.open")
def test_kill_worker_process_skips_sigkill_when_already_exited(self, open_mock, kill_mock, _sleep_mock) -> None:
open_mock.return_value.__enter__.return_value.read.return_value = "123"
kill_mock.side_effect = [None, ProcessLookupError()]
self_heal._kill_worker_process()
# Только SIGTERM и проверка существования процесса.
self.assertEqual(len(kill_mock.call_args_list), 2)
if __name__ == "__main__":
unittest.main()

62
tests/test_utils.py Normal file
View File

@@ -0,0 +1,62 @@
from __future__ import annotations
import unittest
from dubizzle_scraper.core.utils import deep_find_key
from dubizzle_scraper.core.config import parse_listing_segments, _AUTO_YEAR_SPLITS
class TestDeepFindKey(unittest.TestCase):
def test_finds_nested_and_respects_depth(self) -> None:
payload = {
"root": {
"target": "a",
"nested": [{"target": "b"}, {"x": 1}],
}
}
self.assertEqual(deep_find_key(payload, {"target"}), ["a", "b"])
deep_payload = {"l1": {"l2": {"l3": {"target": "value"}}}}
self.assertEqual(deep_find_key(deep_payload, {"target"}, max_depth=2), [])
self.assertEqual(deep_find_key(deep_payload, {"target"}, max_depth=8), ["value"])
class TestParseListingSegments(unittest.TestCase):
def test_empty_and_invalid_return_empty(self) -> None:
self.assertEqual(parse_listing_segments(""), [])
self.assertEqual(parse_listing_segments(" "), [])
self.assertEqual(parse_listing_segments("invalid"), [])
def test_auto_segments_complete_coverage(self) -> None:
"""auto: все годовые диапазоны покрыты, без дыр и без make-фильтра."""
segs = parse_listing_segments("auto")
self.assertEqual(len(segs), len(_AUTO_YEAR_SPLITS))
self.assertEqual(
[(s["year_min"], s["year_max"]) for s in segs],
list(_AUTO_YEAR_SPLITS),
)
self.assertTrue(all(s["make"] is None for s in segs))
# Годовые диапазоны покрывают 1950-2027.
years = set()
for yr_min, yr_max in _AUTO_YEAR_SPLITS:
years.update(range(yr_min, yr_max + 1))
for year in range(1950, 2027):
self.assertIn(year, years)
def test_json_input_formats(self) -> None:
# Массив строк.
segs = parse_listing_segments('["toyota", "ford"]')
self.assertEqual(len(segs), 2)
self.assertEqual(segs[0]["make"], "TOYOTA")
# Массив объектов с годами.
segs = parse_listing_segments('[{"make":"BMW","year_min":2020,"year_max":2025}]')
self.assertEqual(segs[0]["make"], "BMW")
self.assertEqual(segs[0]["year_min"], 2020)
self.assertEqual(segs[0]["year_max"], 2025)
if __name__ == "__main__":
unittest.main()

723
tests/test_worker_tasks.py Normal file
View File

@@ -0,0 +1,723 @@
from __future__ import annotations
import json
from dataclasses import replace
import unittest
from unittest.mock import MagicMock, patch
from dubizzle_scraper.worker import tasks
from dubizzle_scraper.core.config import settings as base_settings
class TestWorkerTaskLockHelpers(unittest.TestCase):
def test_lock_acquire_refresh_release(self) -> None:
redis_client = MagicMock()
# Acquire.
redis_client.set.return_value = True
self.assertTrue(tasks._acquire_lock(redis_client, "lock:key", "owner-token", 120))
redis_client.set.assert_called_once_with("lock:key", "owner-token", nx=True, ex=120)
# Refresh.
redis_client.eval.return_value = 1
self.assertTrue(tasks._refresh_lock_if_owner(redis_client, "lock:key", "owner-token", 120))
# Release.
redis_client.eval.reset_mock()
tasks._release_lock_if_owner(redis_client, "lock:key", "owner-token")
args = redis_client.eval.call_args[0]
self.assertEqual(args[2], "lock:key")
self.assertEqual(args[3], "owner-token")
def test_sync_listing_task_skips_when_lock_not_acquired(self) -> None:
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(tasks, "_acquire_lock", return_value=False):
persistence = MagicMock()
get_persistence.return_value = persistence
get_redis.return_value = MagicMock()
tasks.sync_listing_task.push_request(id="task-123")
try:
result = tasks.sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
persistence.create_tables.assert_called_once()
self.assertEqual(result["status"], "skipped")
self.assertEqual(result["reason"], "sync_already_running")
def test_sync_listing_task_releases_owned_lock(self) -> None:
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=True), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks.sync_listing_task, "update_state"), \
patch.object(tasks, "_run_browser_job", return_value={
"run_id": 7,
"cars_upserted": 2,
"cars_failed": 0,
"images_upserted": 4,
"skipped_existing": 1,
"elapsed_seconds": 1.25,
}):
persistence = MagicMock()
get_persistence.return_value = persistence
redis_client = MagicMock()
redis_client.get.return_value = None
get_redis.return_value = redis_client
stop_event = MagicMock()
heartbeat_thread = MagicMock()
start_heartbeat.return_value = (stop_event, heartbeat_thread)
tasks.sync_listing_task.push_request(id="task-123")
try:
result = tasks.sync_listing_task.run(make="Toyota")
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "success")
stop_event.set.assert_called_once()
heartbeat_thread.join.assert_called_once()
release_lock.assert_called_once()
def test_clear_orphan_sync_listing_lock(self) -> None:
# Нет запущенных задач → удаляет.
redis_client = MagicMock()
redis_client.get.return_value = "owner-token"
redis_client.ttl.return_value = 120
celery_app = MagicMock()
inspector = MagicMock()
inspector.active.return_value = {"worker@node": []}
inspector.reserved.return_value = {"worker@node": []}
inspector.scheduled.return_value = {"worker@node": []}
celery_app.control.inspect.return_value = inspector
self.assertTrue(tasks._clear_orphan_sync_listing_lock(redis_client, celery_app))
redis_client.delete.assert_called_once_with(tasks.SYNC_LISTING_LOCK_KEY)
# Задача активна → не удаляет.
redis_client2 = MagicMock()
redis_client2.get.return_value = "owner-token"
inspector2 = MagicMock()
inspector2.active.return_value = {"worker@node": [{"name": tasks.SYNC_LISTING_TASK_NAME}]}
inspector2.reserved.return_value = {"worker@node": []}
inspector2.scheduled.return_value = {"worker@node": []}
celery_app2 = MagicMock()
celery_app2.control.inspect.return_value = inspector2
self.assertFalse(tasks._clear_orphan_sync_listing_lock(redis_client2, celery_app2))
redis_client2.delete.assert_not_called()
def test_sync_listing_task_recovers_orphan_lock_and_runs(self) -> None:
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(tasks, "_acquire_lock", side_effect=[False, True]) as acquire_lock, \
patch.object(tasks, "_clear_orphan_sync_listing_lock", return_value=True) as clear_orphan, \
patch.object(tasks, "_is_full_scan_done", return_value=True), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks.sync_listing_task, "update_state"), \
patch.object(tasks, "_run_browser_job", return_value={
"run_id": 9,
"cars_upserted": 3,
"cars_failed": 0,
"images_upserted": 5,
"skipped_existing": 0,
"elapsed_seconds": 2.0,
}):
persistence = MagicMock()
get_persistence.return_value = persistence
redis_client = MagicMock()
redis_client.get.return_value = None
get_redis.return_value = redis_client
stop_event = MagicMock()
heartbeat_thread = MagicMock()
start_heartbeat.return_value = (stop_event, heartbeat_thread)
tasks.sync_listing_task.push_request(id="task-456")
try:
result = tasks.sync_listing_task.run(make="Honda")
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "success")
clear_orphan.assert_called_once()
self.assertEqual(acquire_lock.call_count, 2)
release_lock.assert_called_once()
def test_checkpoint_save_load_roundtrip(self) -> None:
storage: dict[str, str] = {}
def _fake_set(key, value, ex=None):
storage[key] = value
return True
redis_client = MagicMock()
redis_client.set.side_effect = _fake_set
redis_client.get.side_effect = lambda key: storage.get(key)
tasks._save_last_completed_segment(redis_client, 12)
self.assertEqual(tasks._load_last_completed_segment(redis_client), 12)
self.assertEqual(redis_client.set.call_args.kwargs["ex"], tasks.SYNC_LISTING_CHECKPOINT_TTL_SECONDS)
def test_load_checkpoint_clears_invalid_payload(self) -> None:
redis_client = MagicMock()
redis_client.get.return_value = "{not-a-number"
self.assertIsNone(tasks._load_last_completed_segment(redis_client))
redis_client.delete.assert_called_once_with(tasks.SYNC_LISTING_CHECKPOINT_KEY)
def test_load_checkpoint_empty_when_missing(self) -> None:
redis_client = MagicMock()
redis_client.get.return_value = None
self.assertIsNone(tasks._load_last_completed_segment(redis_client))
redis_client.delete.assert_not_called()
def test_sync_listing_resumes_from_next_segment_during_bootstrap(self) -> None:
segments = [
{"make": "ACURA"},
{"make": "AUDI"},
{"make": "BMW"},
{"make": "EAGLE"},
]
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(
tasks,
"Settings",
return_value=replace(
base_settings,
discovery=replace(
base_settings.discovery,
always_full_scan=False,
mode="listing",
),
celery=replace(
base_settings.celery,
parallel_segments=False,
),
),
), \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=False), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks, "_run_browser_job", side_effect=lambda fn: fn()), \
patch.object(tasks.sync_listing_task, "update_state"), \
patch("dubizzle_scraper.worker.tasks.parse_listing_segments", return_value=segments):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
redis_client.get.side_effect = lambda key: (
"1" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
)
get_redis.return_value = redis_client
start_heartbeat.return_value = (MagicMock(), MagicMock())
sync_segmented_mock = MagicMock(return_value={
"run_id": 11, "status": "success", "full_scan_completed": True,
"cars_upserted": 1, "cars_failed": 0, "images_upserted": 0,
"skipped_existing": 0, "elapsed_seconds": 1.0, "failures": [],
})
scraper_ctx = MagicMock()
scraper_ctx.__enter__.return_value.sync_listing_segmented = sync_segmented_mock
scraper_ctx.__exit__.return_value = None
with patch.object(tasks, "DUBIZZLEScraper", return_value=scraper_ctx):
tasks.sync_listing_task.push_request(id="task-resume-seg")
try:
result = tasks.sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "success")
# last_completed=1 → start_segment=2 (AUDI завершён, возобновляем с BMW).
self.assertEqual(sync_segmented_mock.call_args.kwargs["start_segment"], 2)
self.assertEqual(sync_segmented_mock.call_args.kwargs["start_page"], 1)
release_lock.assert_called_once()
def test_sync_listing_ignores_checkpoint_after_full_scan_completed(self) -> None:
segments = [{"make": "ACURA"}, {"make": "AUDI"}]
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(
tasks,
"Settings",
return_value=replace(
base_settings,
discovery=replace(
base_settings.discovery,
always_full_scan=False,
mode="listing",
),
celery=replace(
base_settings.celery,
parallel_segments=False,
),
),
), \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=True), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks, "_clear_sync_checkpoint") as clear_checkpoint, \
patch.object(tasks, "_run_browser_job", side_effect=lambda fn: fn()), \
patch.object(tasks.sync_listing_task, "update_state"), \
patch("dubizzle_scraper.worker.tasks.parse_listing_segments", return_value=segments):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
# Оставшийся чекпоинт не должен использоваться.
redis_client.get.side_effect = lambda key: (
"0" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
)
get_redis.return_value = redis_client
start_heartbeat.return_value = (MagicMock(), MagicMock())
sync_segmented_mock = MagicMock(return_value={
"run_id": 14, "status": "success", "full_scan_completed": True,
"cars_upserted": 1, "cars_failed": 0, "images_upserted": 0,
"skipped_existing": 0, "elapsed_seconds": 1.0, "failures": [],
})
scraper_ctx = MagicMock()
scraper_ctx.__enter__.return_value.sync_listing_segmented = sync_segmented_mock
scraper_ctx.__exit__.return_value = None
with patch.object(tasks, "DUBIZZLEScraper", return_value=scraper_ctx):
tasks.sync_listing_task.push_request(id="task-792")
try:
result = tasks.sync_listing_task.run(limit=1)
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "success")
self.assertEqual(sync_segmented_mock.call_args.kwargs["start_segment"], 0)
self.assertIsNone(sync_segmented_mock.call_args.kwargs["progress_callback"])
clear_checkpoint.assert_called()
release_lock.assert_called_once()
def test_sync_listing_checkpoint_beyond_segments_restarts_from_zero(self) -> None:
segments = [{"make": "ACURA"}, {"make": "AUDI"}]
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(
tasks,
"Settings",
return_value=replace(
base_settings,
discovery=replace(
base_settings.discovery,
always_full_scan=False,
mode="listing",
),
celery=replace(
base_settings.celery,
parallel_segments=False,
),
),
), \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=False), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks, "_run_browser_job", side_effect=lambda fn: fn()), \
patch.object(tasks.sync_listing_task, "update_state"), \
patch("dubizzle_scraper.worker.tasks.parse_listing_segments", return_value=segments):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
redis_client.get.side_effect = lambda key: (
"99" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
)
get_redis.return_value = redis_client
start_heartbeat.return_value = (MagicMock(), MagicMock())
sync_segmented_mock = MagicMock(return_value={
"run_id": 15, "status": "success", "full_scan_completed": True,
"cars_upserted": 0, "cars_failed": 0, "images_upserted": 0,
"skipped_existing": 0, "elapsed_seconds": 1.0, "failures": [],
})
scraper_ctx = MagicMock()
scraper_ctx.__enter__.return_value.sync_listing_segmented = sync_segmented_mock
scraper_ctx.__exit__.return_value = None
with patch.object(tasks, "DUBIZZLEScraper", return_value=scraper_ctx):
tasks.sync_listing_task.push_request(id="task-beyond")
try:
result = tasks.sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "success")
self.assertEqual(sync_segmented_mock.call_args.kwargs["start_segment"], 0)
release_lock.assert_called_once()
def test_sync_listing_task_clears_checkpoint_on_hourly_run(self) -> None:
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=True), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks, "_clear_sync_checkpoint") as clear_checkpoint, \
patch.object(tasks, "_run_browser_job", return_value={
"run_id": 13,
"status": "partial_success",
"full_scan_completed": True,
"cars_upserted": 2,
"cars_failed": 1,
"images_upserted": 3,
"skipped_existing": 0,
"elapsed_seconds": 4.0,
"failures": [{"vehicle_url": "v", "error": "e"}],
}), \
patch.object(tasks.sync_listing_task, "update_state"):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
redis_client.get.return_value = None
get_redis.return_value = redis_client
start_heartbeat.return_value = (MagicMock(), MagicMock())
tasks.sync_listing_task.push_request(id="task-791")
try:
result = tasks.sync_listing_task.run(make="Toyota")
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "partial_success")
# Hourly-ветка (full_scan_done=True) всегда удаляет оставшийся чекпоинт
# до браузерного job + после. Главное — вызов произошёл.
clear_checkpoint.assert_called()
release_lock.assert_called_once()
def test_try_set_followup_pending_deduplicates(self) -> None:
redis_client = MagicMock()
redis_client.set.side_effect = [True, False]
self.assertTrue(tasks._try_set_followup_pending(redis_client, ttl_seconds=120))
self.assertFalse(tasks._try_set_followup_pending(redis_client, ttl_seconds=120))
def test_bump_bootstrap_failure_streak_opens_circuit_breaker(self) -> None:
redis_client = MagicMock()
redis_client.incr.return_value = tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT
streak, should_enqueue = tasks._bump_bootstrap_failure_streak(
redis_client,
reason="max_retries_exceeded",
)
self.assertEqual(streak, tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT)
self.assertFalse(should_enqueue)
redis_client.expire.assert_called_once_with(
tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY,
tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS,
)
def test_bump_bootstrap_failure_streak_allows_retry_before_limit(self) -> None:
redis_client = MagicMock()
redis_client.incr.return_value = 1
streak, should_enqueue = tasks._bump_bootstrap_failure_streak(
redis_client,
reason="bootstrap_not_completed",
)
self.assertEqual(streak, 1)
self.assertTrue(should_enqueue)
def test_sync_listing_task_does_not_enqueue_followup_after_bootstrap_error_limit(self) -> None:
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=False), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks, "_try_set_followup_pending", return_value=True), \
patch.object(tasks, "_bump_bootstrap_failure_streak", return_value=(tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT, False)) as bump_streak, \
patch.object(tasks, "_clear_followup_pending") as clear_pending, \
patch.object(tasks.sync_listing_task, "update_state"), \
patch.object(tasks, "_run_browser_job", return_value={
"run_id": 99,
"status": "failed",
"full_scan_completed": False,
"cars_upserted": 0,
"cars_failed": 0,
"images_upserted": 0,
"skipped_existing": 0,
"elapsed_seconds": 1.0,
"failures": [{"vehicle_url": "listing", "error": "bad resume"}],
"listing": {"vehicles_collected": 0},
}):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
redis_client.get.return_value = None
get_redis.return_value = redis_client
start_heartbeat.return_value = (MagicMock(), MagicMock())
with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app:
tasks.sync_listing_task.push_request(id="task-900")
try:
result = tasks.sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "failed")
bump_streak.assert_called_once()
clear_pending.assert_called()
task_app.send_task.assert_not_called()
def test_sync_listing_task_soft_timeout_deduplicates_continuation(self) -> None:
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=True), \
patch.object(tasks, "_start_lock_heartbeat", return_value=(MagicMock(), MagicMock())), \
patch.object(tasks, "_release_lock_if_owner"), \
patch.object(tasks, "_run_browser_job", side_effect=tasks.SoftTimeLimitExceeded()), \
patch.object(tasks, "_try_set_followup_pending", return_value=False) as set_pending, \
patch.object(tasks.sync_listing_task, "update_state"):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
redis_client.get.return_value = None
get_redis.return_value = redis_client
with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app:
tasks.sync_listing_task.push_request(id="task-soft-timeout")
try:
result = tasks.sync_listing_task.run(make="Toyota")
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "timed_out")
set_pending.assert_called_once()
task_app.send_task.assert_not_called()
def test_sync_listing_task_stops_immediate_bootstrap_continuation_after_limit(self) -> None:
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=False), \
patch.object(tasks, "_start_lock_heartbeat", return_value=(MagicMock(), MagicMock())), \
patch.object(tasks, "_release_lock_if_owner"), \
patch.object(tasks, "_try_set_followup_pending", return_value=True), \
patch.object(tasks, "_bump_bootstrap_continuation_streak", return_value=(999, False)), \
patch.object(tasks, "_set_full_scan_done") as set_full_scan_done, \
patch.object(tasks, "_clear_sync_checkpoint") as clear_checkpoint, \
patch.object(tasks.sync_listing_task, "update_state"), \
patch.object(tasks, "_run_browser_job", return_value={
"run_id": 101,
"status": "partial_success",
"full_scan_completed": False,
"cars_upserted": 2,
"cars_failed": 0,
"images_upserted": 1,
"skipped_existing": 0,
"elapsed_seconds": 1.0,
"failures": [],
"listing": {"vehicles_collected": 2},
}):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
redis_client.get.return_value = None
get_redis.return_value = redis_client
with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app:
tasks.sync_listing_task.push_request(id="task-breaker-stop")
try:
result = tasks.sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "partial_success")
# Сначала bootstrap помечается незавершённым, затем breaker переключает на hourly.
self.assertTrue(any(call.args == (redis_client, False) for call in set_full_scan_done.call_args_list))
self.assertTrue(any(call.args == (redis_client, True) for call in set_full_scan_done.call_args_list))
clear_checkpoint.assert_called_once()
task_app.send_task.assert_not_called()
def test_sync_listing_task_always_full_scan_forces_bootstrap_even_after_done(self) -> None:
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(
tasks,
"Settings",
return_value=replace(
base_settings,
discovery=replace(base_settings.discovery, always_full_scan=True),
),
), \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=True), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks, "_set_full_scan_done") as set_full_scan_done, \
patch.object(tasks.sync_listing_task, "update_state"), \
patch.object(tasks, "_run_browser_job") as run_job, \
patch("dubizzle_scraper.worker.tasks.parse_listing_segments", return_value=[]):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
redis_client.get.return_value = None
get_redis.return_value = redis_client
start_heartbeat.return_value = (MagicMock(), MagicMock())
run_job.return_value = {
"run_id": 17,
"status": "success",
"full_scan_completed": True,
"cars_upserted": 1,
"cars_failed": 0,
"images_upserted": 0,
"skipped_existing": 0,
"elapsed_seconds": 1.0,
"failures": [],
}
tasks.sync_listing_task.push_request(id="task-always-full")
try:
result = tasks.sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "success")
set_full_scan_done.assert_called_with(redis_client, False)
release_lock.assert_called_once()
def test_sync_listing_task_always_full_scan_uses_segmented_resume_path(self) -> None:
segments = [{"make": "TOYOTA"}, {"make": "FORD"}, {"make": "HONDA"}]
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(
tasks,
"Settings",
return_value=replace(
base_settings,
discovery=replace(base_settings.discovery, always_full_scan=True),
),
), \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=True), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks, "_run_browser_job", side_effect=lambda fn: fn()), \
patch.object(tasks.sync_listing_task, "update_state"), \
patch("dubizzle_scraper.worker.tasks.parse_listing_segments", return_value=segments):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
redis_client.get.side_effect = lambda key: (
"0" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
)
get_redis.return_value = redis_client
start_heartbeat.return_value = (MagicMock(), MagicMock())
sync_segmented_mock = MagicMock(return_value={
"run_id": 18,
"status": "success",
"full_scan_completed": True,
"cars_upserted": 1,
"cars_failed": 0,
"images_upserted": 0,
"skipped_existing": 0,
"elapsed_seconds": 1.0,
"failures": [],
})
scraper_ctx = MagicMock()
scraper_ctx.__enter__.return_value.sync_listing_segmented = sync_segmented_mock
scraper_ctx.__enter__.return_value.sync_listing = MagicMock()
scraper_ctx.__exit__.return_value = None
with patch.object(tasks, "DUBIZZLEScraper", return_value=scraper_ctx):
tasks.sync_listing_task.push_request(id="task-always-full-resume")
try:
result = tasks.sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "success")
self.assertEqual(sync_segmented_mock.call_args.kwargs["start_segment"], 1)
self.assertEqual(sync_segmented_mock.call_args.kwargs["start_page"], 1)
release_lock.assert_called_once()
def test_sync_listing_task_always_full_scan_resets_segment_progress_and_dispatches_again(self) -> None:
segments = [{"make": "TOYOTA"}, {"make": "FORD"}]
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(
tasks,
"Settings",
return_value=replace(
base_settings,
discovery=replace(base_settings.discovery, always_full_scan=True),
celery=replace(base_settings.celery, parallel_segments=True),
),
), \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=True), \
patch.object(tasks, "_is_segmented_scan_in_progress", return_value=False), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks, "_reset_segments_progress") as reset_progress, \
patch.object(tasks, "_set_segmented_scan_active") as set_active, \
patch.object(tasks.sync_listing_task, "update_state"):
get_persistence.return_value = MagicMock()
redis_client = MagicMock()
redis_client.smembers.return_value = {"0", "1"}
redis_client.get.return_value = None
get_redis.return_value = redis_client
start_heartbeat.return_value = (MagicMock(), MagicMock())
with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app, \
patch("dubizzle_scraper.worker.tasks.parse_listing_segments", return_value=segments):
tasks.sync_listing_task.push_request(id="task-always-hourly")
try:
result = tasks.sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "success")
self.assertEqual(result["segments_dispatched"], 2)
reset_progress.assert_called_once_with(redis_client, len(segments))
set_active.assert_called_once_with(redis_client, total=len(segments))
self.assertEqual(task_app.send_task.call_count, 2)
release_lock.assert_called_once()
def test_sync_listing_task_skips_duplicate_parallel_dispatch_while_segments_active(self) -> None:
segments = [{"make": "TOYOTA"}, {"make": "FORD"}]
with patch.object(tasks, "_get_persistence") as get_persistence, \
patch.object(tasks, "_get_redis") as get_redis, \
patch.object(
tasks,
"Settings",
return_value=replace(
base_settings,
discovery=replace(base_settings.discovery, always_full_scan=True),
celery=replace(base_settings.celery, parallel_segments=True),
),
), \
patch.object(tasks, "_acquire_lock", return_value=True), \
patch.object(tasks, "_is_full_scan_done", return_value=True), \
patch.object(tasks, "_is_segmented_scan_in_progress", return_value=True), \
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
patch.object(tasks.sync_listing_task, "update_state"):
get_persistence.return_value = MagicMock()
get_redis.return_value = MagicMock()
start_heartbeat.return_value = (MagicMock(), MagicMock())
with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app, \
patch("dubizzle_scraper.worker.tasks.parse_listing_segments", return_value=segments):
tasks.sync_listing_task.push_request(id="task-always-skip-duplicate")
try:
result = tasks.sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
self.assertEqual(result["status"], "skipped")
self.assertEqual(result["reason"], "segmented_scan_in_progress")
task_app.send_task.assert_not_called()
release_lock.assert_called_once()
if __name__ == "__main__":
unittest.main()