from __future__ import annotations from datetime import datetime, timezone import json from types import SimpleNamespace import unittest from unittest.mock import MagicMock, patch from mobilede_scraper.core.runtime_config import RuntimeConfig from mobilede_scraper.worker import planner, refresh_cycle, search_sync, tasks class TestWorkerRuntimeTaskHelpers(unittest.TestCase): class _FakeRedis: def __init__(self) -> None: self.store: dict[str, str] = {} self.sets: dict[str, set[str]] = {} self.lists: dict[str, list[str]] = {} self.zsets: dict[str, dict[str, float]] = {} def get(self, key: str): return self.store.get(key) def pttl(self, key: str): return 1_000 if key in self.store else -2 def set( self, key: str, value: str, nx: bool = False, ex: int | None = None, px: int | None = None, ): # noqa: ARG002 if nx and key in self.store: return False self.store[key] = str(value) return True def incr(self, key: str): current = int(self.store.get(key, "0")) current += 1 self.store[key] = str(current) return current def sadd(self, key: str, member: str): bucket = self.sets.setdefault(key, set()) before = len(bucket) bucket.add(member) return 1 if len(bucket) > before else 0 def srem(self, key: str, member: str): bucket = self.sets.setdefault(key, set()) if member not in bucket: return 0 bucket.remove(member) return 1 def scard(self, key: str): return len(self.sets.get(key, set())) def sismember(self, key: str, member: str): return member in self.sets.get(key, set()) def smembers(self, key: str): return self.sets.get(key, set()) def llen(self, key: str): return len(self.lists.get(key, [])) def zcard(self, key: str): return len(self.zsets.get(key, {})) def zrem(self, key: str, member: str): bucket = self.zsets.setdefault(key, {}) return 1 if bucket.pop(str(member), None) is not None else 0 def zremrangebyscore(self, key: str, minimum, maximum): # noqa: ANN001 bucket = self.zsets.setdefault(key, {}) upper = float(maximum) removed = [member for member, score in bucket.items() if score <= upper] for member in removed: del bucket[member] return len(removed) def eval(self, script: str, numkeys: int, *args): # noqa: ARG002 if "ZCARD" in script and "queued:" in script: active_key, claim_key, now, max_active, token, ttl, listing_id = args self.zremrangebyscore(active_key, "-inf", now) if claim_key in self.store: return 0 if self.zcard(active_key) >= int(max_active): return -1 self.store[claim_key] = f"queued:{token}" self.zsets.setdefault(active_key, {})[str(listing_id)] = float(now) + float(ttl) return 1 if "~= 'queued:'" in script: active_key, claim_key, claim_token, owner, ttl, now, listing_id = args if self.store.get(claim_key) != f"queued:{claim_token}": return 0 self.store[claim_key] = f"running:{owner}" self.zsets.setdefault(active_key, {})[str(listing_id)] = float(now) + float(ttl) return 1 if numkeys == 3 and "return 0" in script: active_key, claim_key, retry_key, owner, listing_id = args if self.store.get(claim_key) != f"running:{owner}": return 0 self.store.pop(claim_key, None) self.zrem(active_key, str(listing_id)) self.store.pop(retry_key, None) return 1 if numkeys == 2 and "ARGV[2]" in script: active_key, claim_key, owner, status, _cooldown, listing_id = args if self.store.get(claim_key) != f"running:{owner}": return 0 self.store[claim_key] = str(status) self.zrem(active_key, str(listing_id)) return 1 raise AssertionError("Unsupported fake Redis Lua script") def delete(self, *keys: str): removed = 0 for key in keys: if key in self.store: del self.store[key] removed += 1 if key in self.sets: del self.sets[key] removed += 1 if key in self.lists: del self.lists[key] removed += 1 if key in self.zsets: del self.zsets[key] removed += 1 return removed def expire(self, key: str, ttl: int): # noqa: ARG002 return True def test_lock_acquire_refresh_release(self) -> None: redis_client = MagicMock() 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) redis_client.eval.return_value = 1 self.assertTrue(tasks._refresh_lock_if_owner(redis_client, "lock:key", "owner-token", 120)) 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_refresh_finalization_claim_is_released_after_db_failure(self) -> None: redis_client = self._FakeRedis() cycle_id = "cycle-1" redis_client.set(tasks.MOBILEDE_REFRESH_CYCLE_ID_KEY, cycle_id) redis_client.set(tasks.MOBILEDE_REFRESH_CYCLE_DONE_KEY, "1") redis_client.set(tasks.MOBILEDE_REFRESH_CYCLE_TOTAL_KEY, "1") redis_client.set(tasks.MOBILEDE_REFRESH_CYCLE_STARTED_AT_KEY, datetime.now(timezone.utc).isoformat()) self.assertTrue(refresh_cycle._mobilede_claim_refresh_cycle_finalization(redis_client, cycle_id=cycle_id)) persistence = MagicMock() persistence.mark_sold_not_seen_since.side_effect = RuntimeError("db unavailable") with self.assertRaisesRegex(RuntimeError, "db unavailable"): refresh_cycle._mobilede_finalize_refresh_cycle_sold_marking( redis_client, cycle_id=cycle_id, logger=MagicMock(), get_persistence=lambda: persistence, origin_prefixes=("mobile.de:",), post_refresh_probe_enabled=False, post_refresh_probe_batch_size=0, schedule_post_refresh_probe=lambda: None, ) claim_key = refresh_cycle._mobilede_refresh_cycle_finalized_key(cycle_id) self.assertIsNone(redis_client.get(claim_key)) self.assertTrue(refresh_cycle._mobilede_claim_refresh_cycle_finalization(redis_client, cycle_id=cycle_id)) def test_mobilede_segment_key_is_stable(self) -> None: segment = {"make_id": "BMW", "price_min": 1000, "price_max": 5000} key1 = tasks._mobilede_task_segment_key( segment=segment, search_url=None, make_id=None, model_id=None, price_min=None, price_max=None, year_min=None, year_max=None, mileage_min=None, mileage_max=None, ) key2 = tasks._mobilede_task_segment_key( segment={"price_max": 5000, "make_id": "BMW", "price_min": 1000}, search_url=None, make_id=None, model_id=None, price_min=None, price_max=None, year_min=None, year_max=None, mileage_min=None, mileage_max=None, ) self.assertEqual(key1, key2) self.assertEqual(len(key1), 16) def test_detail_rate_limit_keys_rotate_across_parallel_slots(self) -> None: redis_client = self._FakeRedis() keys = [tasks._mobilede_detail_rate_limit_key(redis_client, 2) for _ in range(4)] self.assertEqual( keys, [ f"{tasks.MOBILEDE_DETAIL_RATE_LIMIT_KEY}:0", f"{tasks.MOBILEDE_DETAIL_RATE_LIMIT_KEY}:1", f"{tasks.MOBILEDE_DETAIL_RATE_LIMIT_KEY}:0", f"{tasks.MOBILEDE_DETAIL_RATE_LIMIT_KEY}:1", ], ) self.assertEqual( tasks._mobilede_detail_rate_limit_key(redis_client, 1), tasks.MOBILEDE_DETAIL_RATE_LIMIT_KEY, ) def test_sold_probe_circuit_breaker_opens_after_configured_403_429_streak(self) -> None: redis_client = self._FakeRedis() with patch.object(tasks, "MOBILEDE_SOLD_PROBE_ANTIBOT_STREAK_LIMIT", 3): self.assertFalse(tasks._mobilede_record_sold_probe_antibot_result( redis_client, verdict="blocked", status_code=403, )) self.assertFalse(tasks._mobilede_record_sold_probe_antibot_result( redis_client, verdict="blocked", status_code=429, )) self.assertTrue(tasks._mobilede_record_sold_probe_antibot_result( redis_client, verdict="blocked", status_code=403, )) self.assertEqual( redis_client.store[tasks.MOBILEDE_SOLD_PROBE_ANTIBOT_BLOCK_KEY], "403", ) self.assertNotIn(tasks.MOBILEDE_SOLD_PROBE_ANTIBOT_STREAK_KEY, redis_client.store) def test_sold_probe_circuit_breaker_resets_streak_after_definitive_result(self) -> None: redis_client = self._FakeRedis() tasks._mobilede_record_sold_probe_antibot_result( redis_client, verdict="blocked", status_code=403, ) tasks._mobilede_record_sold_probe_antibot_result( redis_client, verdict="available", status_code=200, ) self.assertNotIn(tasks.MOBILEDE_SOLD_PROBE_ANTIBOT_STREAK_KEY, redis_client.store) def test_sold_probe_stops_batch_when_antibot_circuit_breaker_opens(self) -> None: redis_client = self._FakeRedis() persistence = MagicMock() persistence.get_active_cars_batch_for_sold_probe.return_value = [ (index, f"https://example.test/{index}", datetime.now(timezone.utc)) for index in range(1, 6) ] client = MagicMock() with ( patch.object(tasks, "_get_redis", return_value=redis_client), patch.object(tasks, "_get_persistence", return_value=persistence), patch.object(tasks.MobileDeClient, "for_worker", return_value=client), patch.object( tasks, "_mobilede_probe_active_listing_status", return_value=("blocked", 403), ) as probe, patch.object(tasks, "MOBILEDE_SOLD_PROBE_ANTIBOT_STREAK_LIMIT", 3), ): result = tasks.mobilede_verify_active_sold_batch_task.run(limit=5) self.assertEqual(result["status"], "completed") self.assertEqual(result["checked"], 3) self.assertEqual(result["blocked"], 3) self.assertEqual(result["skipped"], 2) self.assertEqual(probe.call_count, 3) persistence.mark_cars_sold_by_ids.assert_not_called() self.assertIn(tasks.MOBILEDE_SOLD_PROBE_ANTIBOT_BLOCK_KEY, redis_client.store) def test_runtime_numeric_filters_intersect_planner_segment(self) -> None: self.assertEqual( search_sync._intersect_numeric_bounds("10000", "30000", 15000, 25000), ("15000", "25000", True), ) self.assertEqual( search_sync._intersect_numeric_bounds("10000", "12000", 15000, 25000), ("10000", "12000", False), ) def test_depth_six_overflow_leaf_can_split_further(self) -> None: segment = { "search_url": "https://www.mobile.de/ru/транспортные-средства/поиск.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&ml=%3A4687&p=15003%3A", "listing_url": "https://www.mobile.de/ru/транспортные-средства/поиск.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&ml=%3A4687&p=15003%3A", "label": "BMW hot leaf", "make_id": "3500", "mileage_max": "4687", "price_min": "15003", "overflow_depth": 6, "overflow_split": "mileage", "max_pages": 50, } with patch.object(planner, "MOBILEDE_OVERFLOW_MAX_SPLIT_DEPTH", 8): groups = planner._mobilede_build_overflow_candidate_groups(segment=segment, max_pages=50) self.assertTrue(groups) self.assertTrue(all(int(child["overflow_depth"]) == 7 for _kind, children in groups for child in children)) self.assertTrue(any(kind == "price" for kind, _children in groups)) def test_detail_queue_prioritizes_new_unique_listings(self) -> None: redis_client = self._FakeRedis() with ( patch.dict("os.environ", {"MOBILEDE_DETAIL_QUEUE_ENABLED": "true", "MOBILEDE_DETAIL_QUEUE_MAX_PENDING": "3"}), patch.object(tasks.mobilede_sync_detail_task, "apply_async") as apply_async, ): result = tasks._queue_mobilede_detail_listing_ids( redis_client, ["mobile.de:100", "mobile.de:100", "mobile.de:101"], priority=9, ) self.assertEqual(result, {"queued": 2, "duplicate": 0, "full": 0}) self.assertEqual(redis_client.sets, {}) self.assertEqual(apply_async.call_count, 2) self.assertTrue(all(call.kwargs["queue"] == tasks.MOBILEDE_IMAGES_QUEUE for call in apply_async.call_args_list)) self.assertTrue(all(call.kwargs["priority"] == 9 for call in apply_async.call_args_list)) payloads = [call.kwargs["kwargs"] for call in apply_async.call_args_list] self.assertEqual([payload["listing_id"] for payload in payloads], ["100", "101"]) self.assertTrue(all(payload["claim_token"] for payload in payloads)) self.assertTrue(all(payload["force_refresh"] is False for payload in payloads)) def test_detail_queue_respects_pending_limit(self) -> None: redis_client = self._FakeRedis() redis_client.zsets[tasks.MOBILEDE_DETAIL_ACTIVE_KEY] = {"100": 9999999999, "101": 9999999999} with ( patch.dict("os.environ", {"MOBILEDE_DETAIL_QUEUE_ENABLED": "true", "MOBILEDE_DETAIL_QUEUE_MAX_PENDING": "2"}), patch.object(tasks.mobilede_sync_detail_task, "apply_async") as apply_async, ): result = tasks._queue_mobilede_detail_listing_ids( redis_client, ["mobile.de:102", "mobile.de:103"], ) self.assertEqual(result, {"queued": 0, "duplicate": 0, "full": 2}) apply_async.assert_not_called() def test_detail_queue_skips_existing_redis_claim(self) -> None: redis_client = self._FakeRedis() redis_client.store[tasks.MOBILEDE_DETAIL_CLAIM_KEY_FMT.format(listing_id="100")] = "running:task-a" with ( patch.dict("os.environ", {"MOBILEDE_DETAIL_QUEUE_ENABLED": "true"}), patch.object(tasks.mobilede_sync_detail_task, "apply_async") as apply_async, ): result = tasks._queue_mobilede_detail_listing_ids( redis_client, ["mobile.de:100", "mobile.de:101"], ) self.assertEqual(result, {"queued": 1, "duplicate": 1, "full": 0}) apply_async.assert_called_once() def test_explicit_gallery_refresh_invalidates_completed_listing(self) -> None: redis_client = self._FakeRedis() claim_key = tasks.MOBILEDE_DETAIL_CLAIM_KEY_FMT.format(listing_id="100") redis_client.store[claim_key] = "exhausted" with ( patch.dict("os.environ", {"MOBILEDE_DETAIL_QUEUE_ENABLED": "true"}), patch.object(tasks.mobilede_sync_detail_task, "apply_async") as apply_async, ): result = tasks._queue_mobilede_detail_listing_ids( redis_client, ["mobile.de:100"], invalidate_completed=True, ) self.assertEqual(result, {"queued": 1, "duplicate": 0, "full": 0}) apply_async.assert_called_once() self.assertTrue(redis_client.store[claim_key].startswith("queued:")) def test_detail_queue_preserves_redis_claim_when_dispatch_is_ambiguous(self) -> None: redis_client = self._FakeRedis() with ( patch.dict("os.environ", {"MOBILEDE_DETAIL_QUEUE_ENABLED": "true"}), patch.object(tasks.mobilede_sync_detail_task, "apply_async", side_effect=RuntimeError("broker down")), self.assertRaisesRegex(RuntimeError, "broker down"), ): tasks._queue_mobilede_detail_listing_ids(redis_client, ["mobile.de:100"]) self.assertIn(tasks.MOBILEDE_DETAIL_CLAIM_KEY_FMT.format(listing_id="100"), redis_client.store) self.assertEqual(redis_client.zcard(tasks.MOBILEDE_DETAIL_ACTIVE_KEY), 1) def test_detail_claim_fences_duplicate_delivery_and_finishes_by_owner(self) -> None: redis_client = self._FakeRedis() claim_token = tasks._mobilede_claim_detail_dispatch(redis_client, "100", max_active=10) self.assertTrue(claim_token) self.assertTrue(tasks._mobilede_start_detail_execution( redis_client, "100", claim_token=str(claim_token), execution_owner="task-a", )) self.assertFalse(tasks._mobilede_start_detail_execution( redis_client, "100", claim_token=str(claim_token), execution_owner="task-b", )) self.assertFalse(tasks._mobilede_finish_detail_claim( redis_client, "100", execution_owner="task-b", )) self.assertTrue(tasks._mobilede_finish_detail_claim( redis_client, "100", execution_owner="task-a", )) self.assertEqual(redis_client.zcard(tasks.MOBILEDE_DETAIL_ACTIVE_KEY), 0) def test_stale_detail_owner_cannot_clear_current_cooldown_claim(self) -> None: redis_client = self._FakeRedis() claim_token = tasks._mobilede_claim_detail_dispatch(redis_client, "100", max_active=10) self.assertTrue(claim_token) self.assertTrue(tasks._mobilede_start_detail_execution( redis_client, "100", claim_token=str(claim_token), execution_owner="current-task", )) tasks._mobilede_cooldown_detail_claim( redis_client, "100", execution_owner="stale-task", status="retry", cooldown_seconds=60, ) claim_key = tasks.MOBILEDE_DETAIL_CLAIM_KEY_FMT.format(listing_id="100") self.assertEqual(redis_client.store[claim_key], "running:current-task") self.assertEqual(redis_client.zcard(tasks.MOBILEDE_DETAIL_ACTIVE_KEY), 1) def test_detail_claim_ttl_is_short_and_configurable(self) -> None: with patch.dict("os.environ", {"MOBILEDE_DETAIL_TIMEOUT_SECONDS": "10"}, clear=False): self.assertEqual(tasks._mobilede_detail_claim_ttl_seconds(), 300) with patch.dict("os.environ", {"MOBILEDE_DETAIL_CLAIM_TTL_SECONDS": "420"}, clear=False): self.assertEqual(tasks._mobilede_detail_claim_ttl_seconds(), 420) def test_only_new_segment_with_updates_only_enters_cooldown(self) -> None: redis_client = self._FakeRedis() segment = {"make_id": "BMW", "label": "BMW"} tasks._mobilede_update_segment_freshness_state( redis_client, segment=segment, only_new=True, inserted=0, updated=20, listings=20, ) cooldown_key = tasks._mobilede_segment_cooldown_key(segment) hot_key = tasks._mobilede_segment_hot_key(segment) self.assertEqual(redis_client.store[cooldown_key], "1") self.assertNotIn(hot_key, redis_client.store) def test_detail_scraper_is_reused_within_worker_process(self) -> None: original = tasks._detail_scraper_instance tasks._detail_scraper_instance = None try: with ( patch.object(tasks.MobileDeClient, "for_worker") as client_factory, patch.object(tasks, "_get_persistence") as persistence_factory, patch.object(tasks, "MobileDeScraper") as scraper_factory, ): expected = scraper_factory.return_value first = tasks._get_detail_scraper() second = tasks._get_detail_scraper() self.assertIs(first, expected) self.assertIs(second, expected) client_factory.assert_called_once_with(delay_seconds=0) persistence_factory.assert_called_once_with() scraper_factory.assert_called_once() finally: tasks._detail_scraper_instance = original def test_prune_overflow_parent_segments_keeps_only_leaf_segments(self) -> None: parent = { "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=1:5000&fr=:2004&ml=:100000", "make_id": "3500", "price_min": "1", "price_max": "5000", "year_max": "2004", "mileage_max": "100000", } parent_key = tasks._mobilede_segment_fingerprint(parent) child_a = { **parent, "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=1:5000&fr=:2004&ml=:50000", "mileage_max": "50000", "overflow_parent": parent_key, } child_b = { **parent, "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=1:5000&fr=:2004&ml=50001:100000", "mileage_min": "50001", "mileage_max": "100000", "overflow_parent": parent_key, } pruned = tasks._mobilede_prune_overflow_parent_segments([parent, child_a, child_b]) self.assertEqual(pruned, [child_a, child_b]) def test_compact_runtime_segments_prunes_observed_empty_and_merges_adjacent_ranges(self) -> None: base_url = "https://www.mobile.de/ru/search.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&fr=2018:2020" segments = [ { "label": "BMW | price=1-100 | year=2018-2020 | total=400", "search_url": f"{base_url}&p=1:100", "source_search_url": base_url, "make_id": "3500", "price_min": "1", "price_max": "100", "year_min": "2018", "year_max": "2020", "total_results": 400, "observed_complete": True, "max_pages": 50, }, { "label": "BMW | price=101-200 | year=2018-2020 | total=500", "search_url": f"{base_url}&p=101:200", "source_search_url": base_url, "make_id": "3500", "price_min": "101", "price_max": "200", "year_min": "2018", "year_max": "2020", "total_results": 500, "observed_complete": True, "max_pages": 50, }, { "label": "BMW | price=201-300 | year=2018-2020 | total=0", "search_url": f"{base_url}&p=201:300", "source_search_url": base_url, "make_id": "3500", "price_min": "201", "price_max": "300", "year_min": "2018", "year_max": "2020", "total_results": 0, "observed_complete": True, "max_pages": 50, }, ] compacted = tasks._mobilede_compact_runtime_segments(segments) self.assertEqual(len(compacted), 1) self.assertEqual(compacted[0]["price_min"], "1") self.assertEqual(compacted[0]["price_max"], "200") self.assertEqual(compacted[0]["total_results"], 900) self.assertEqual(compacted[0]["max_pages"], 45) self.assertEqual(compacted[0]["compacted_from"], 2) def test_finalize_runtime_plan_compacts_before_interleaving_makes(self) -> None: base_url = "https://www.mobile.de/ru/search.html?isSearchRequest=true&s=Car&vc=Car" segments = [ { "label": "BMW | price=1-100 | total=400", "search_url": f"{base_url}&ms=3500&p=1:100", "source_search_url": base_url, "make_id": "3500", "price_min": "1", "price_max": "100", "total_results": 400, "observed_complete": True, "max_pages": 20, }, { "label": "BMW | price=101-200 | total=500", "search_url": f"{base_url}&ms=3500&p=101:200", "source_search_url": base_url, "make_id": "3500", "price_min": "101", "price_max": "200", "total_results": 500, "observed_complete": True, "max_pages": 25, }, { "label": "Porsche | price=1-5000 | total=100", "search_url": f"{base_url}&ms=20100&p=1:5000", "source_search_url": base_url, "make_id": "20100", "price_min": "1", "price_max": "5000", "total_results": 100, "observed_complete": True, "max_pages": 5, }, ] finalized = planner._mobilede_finalize_runtime_plan(segments) self.assertEqual(len(finalized), 2) self.assertEqual(finalized[0]["make_id"], "3500") self.assertEqual(finalized[0]["total_results"], 900) self.assertEqual(finalized[1]["make_id"], "20100") def test_complete_runtime_plan_rejects_unknown_dense_and_under_capacity_leaves(self) -> None: with self.assertRaisesRegex(RuntimeError, "unknown=1 dense=1 under_capacity=2"): planner._mobilede_assert_complete_runtime_plan( [ {"label": "unknown", "total_results": None, "max_pages": 50}, {"label": "dense", "total_results": 1001, "max_pages": 50}, {"label": "under capacity", "total_results": 950, "max_pages": 12}, ] ) def test_segment_pages_cover_known_total_even_when_parent_was_shorter(self) -> None: self.assertEqual(planner._mobilede_segment_pages_for_total(611, 12), 31) self.assertEqual(planner._mobilede_segment_pages_for_total(944, 12), 48) def test_complete_dense_refine_ignores_soft_probe_limit(self) -> None: parent = { "label": "BMW dense", "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=1:3000", "make_id": "3500", "price_min": "1", "price_max": "3000", "total_results": 2400, "max_pages": 50, } parent_key = tasks._mobilede_segment_fingerprint(parent) children = [ { **parent, "label": f"BMW child {index}", "search_url": f"https://www.mobile.de/ru/search.html?ms=3500&p={start}:{end}", "price_min": str(start), "price_max": str(end), "total_results": None, "overflow_parent": parent_key, "overflow_depth": 1, } for index, (start, end) in enumerate(((1, 1000), (1001, 2000), (2001, 3000)), start=1) ] with ( patch.object(planner, "_mobilede_build_overflow_child_segments", return_value=children), patch.object(planner, "_mobilede_probe_segment_total", side_effect=[800, 800, 800]) as probe, ): refined = planner._mobilede_refine_dense_planned_segments([parent]) self.assertEqual(probe.call_count, 3) self.assertEqual([item["total_results"] for item in refined], [800, 800, 800]) planner._mobilede_assert_complete_runtime_plan(refined) def test_partial_transport_failure_does_not_reject_overflow_split(self) -> None: parent = { "label": "Lexus dense", "search_url": "https://www.mobile.de/ru/search.html?ms=13200&p=40001:50000", "make_id": "13200", "total_results": 2452, "max_pages": 50, } children = [ {**parent, "label": "unresolved child", "total_results": None}, {**parent, "label": "tiny child", "total_results": 1}, ] self.assertTrue( planner._mobilede_overflow_candidate_group_is_useful(parent, "mileage", children) ) def test_dense_refine_recursively_completes_tree_in_one_call(self) -> None: parent = { "label": "BMW parent", "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=1:3000", "make_id": "3500", "price_min": "1", "price_max": "3000", "total_results": 2400, "max_pages": 50, } child_dense = { **parent, "label": "BMW dense child", "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=1:2000", "price_max": "2000", "total_results": None, "overflow_depth": 1, } child_complete = { **parent, "label": "BMW complete child", "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=2001:3000", "price_min": "2001", "total_results": None, "overflow_depth": 1, } grandchildren = [ { **child_dense, "label": f"BMW grandchild {index}", "search_url": f"https://www.mobile.de/ru/search.html?ms=3500&p={start}:{end}", "price_min": str(start), "price_max": str(end), "total_results": None, "overflow_depth": 2, } for index, (start, end) in enumerate(((1, 1000), (1001, 2000)), start=1) ] with ( patch.object( planner, "_mobilede_build_overflow_child_segments", side_effect=[[child_dense, child_complete], grandchildren], ) as split, patch.object( planner, "_mobilede_probe_segment_total", side_effect=[1500, 900, 700, 800], ), ): refined = planner._mobilede_refine_dense_planned_segments([parent]) self.assertEqual(split.call_count, 2) self.assertEqual(sorted(item["total_results"] for item in refined), [700, 800, 900]) planner._mobilede_assert_complete_runtime_plan(refined) def test_record_observation_keeps_active_bootstrap_cache_shape(self) -> None: redis_client = MagicMock() settings = MagicMock() first = { "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=1:100", "source_search_url": "https://www.mobile.de/ru/search.html?ms=3500", "make_id": "3500", "price_min": "1", "price_max": "100", "total_results": None, "max_pages": 50, } second = { "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=101:200", "source_search_url": "https://www.mobile.de/ru/search.html?ms=3500", "make_id": "3500", "price_min": "101", "price_max": "200", "total_results": 500, "observed_complete": True, "max_pages": 25, } with ( patch.object(tasks, "_acquire_lock", return_value=True), patch.object(tasks, "_release_lock_if_owner"), patch.object(tasks, "_get_cached_mobilede_runtime_segments", return_value=[first, second]), patch.object(planner, "_mobilede_source_segments_from_settings", return_value=[first]), patch.object(planner, "_mobilede_save_learned_runtime_segments") as save_plan, ): removed = tasks._mobilede_record_observed_segment_total( redis_client, settings, segment=first, total_results=400, listing_count=400, unique_count=400, bootstrap_run=True, ) cached_payload = redis_client.set.call_args.args[1] cached_segments = json.loads(cached_payload) self.assertEqual(removed, 1) self.assertEqual(len(cached_segments), 2) self.assertEqual(cached_segments[0]["total_results"], 400) self.assertTrue(cached_segments[0]["observed_complete"]) learned_segments = save_plan.call_args.args[1] self.assertEqual(len(learned_segments), 1) self.assertEqual(learned_segments[0]["total_results"], 900) def test_runtime_segment_reservation_uses_pending_cache(self) -> None: redis_client = MagicMock() settings = MagicMock() redis_client.get.side_effect = lambda key: { tasks.MOBILEDE_RUNTIME_SEGMENTS_CACHE_KEY: b'[{"make_id":"BMW"},{"make_id":"AUDI"}]', tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY: b"2", tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY: b"2", }.get(key) redis_client.incr.return_value = 2 redis_client.ttl.return_value = -2 segment = tasks._reserve_mobilede_runtime_segment( redis_client, settings, only_new=False, ) self.assertEqual(segment, (1, {"make_id": "AUDI"})) def test_preplan_probe_budget_is_global(self) -> None: original_probe = tasks._mobilede_probe_segment_total original_preplan_enabled = tasks.MOBILEDE_PREPLAN_SEGMENT_PROBES original_max_segments = tasks.MOBILEDE_PREPLAN_MAX_SEGMENTS original_max_probes = tasks.MOBILEDE_PREPLAN_MAX_PROBES calls = {"count": 0} segments = [ {"label": f"Cars {index}", "search_url": f"https://suchen.mobile.de/fahrzeuge/search.html?x={index}"} for index in range(5) ] try: tasks.MOBILEDE_PREPLAN_SEGMENT_PROBES = True tasks.MOBILEDE_PREPLAN_MAX_SEGMENTS = 50 tasks.MOBILEDE_PREPLAN_MAX_PROBES = 2 def _probe(_segment): calls["count"] += 1 return 100 tasks._mobilede_probe_segment_total = _probe planned = tasks._mobilede_preplan_runtime_segments(segments) finally: tasks._mobilede_probe_segment_total = original_probe tasks.MOBILEDE_PREPLAN_SEGMENT_PROBES = original_preplan_enabled tasks.MOBILEDE_PREPLAN_MAX_SEGMENTS = original_max_segments tasks.MOBILEDE_PREPLAN_MAX_PROBES = original_max_probes self.assertEqual(calls["count"], 2) self.assertEqual(len(planned), 5) def test_bootstrap_finalization_waits_for_dispatched_segments(self) -> None: redis_client = MagicMock() redis_client.get.side_effect = lambda key: { tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY: b"10", tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY: b"10", tasks.MOBILEDE_BOOTSTRAP_DONE_KEY: None, tasks.MOBILEDE_BOOTSTRAP_LISTINGS_TOTAL_KEY: b"1000", tasks.MOBILEDE_BOOTSTRAP_UNIQUE_TOTAL_KEY: b"995", tasks.MOBILEDE_BOOTSTRAP_INSERTED_TOTAL_KEY: b"990", tasks.MOBILEDE_BOOTSTRAP_UPDATED_TOTAL_KEY: b"5", tasks.MOBILEDE_BOOTSTRAP_IMAGES_TOTAL_KEY: b"7000", }.get(key) redis_client.scard.return_value = 1 finalized = tasks._mobilede_try_finalize_bootstrap(redis_client) self.assertFalse(finalized) redis_client.set.assert_not_called() def test_bootstrap_finalization_marks_done_when_dispatched_is_empty(self) -> None: redis_client = MagicMock() redis_client.get.side_effect = lambda key: { tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY: b"10", tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY: b"10", tasks.MOBILEDE_BOOTSTRAP_DONE_KEY: None, tasks.MOBILEDE_BOOTSTRAP_LISTINGS_TOTAL_KEY: b"1000", tasks.MOBILEDE_BOOTSTRAP_UNIQUE_TOTAL_KEY: b"995", tasks.MOBILEDE_BOOTSTRAP_INSERTED_TOTAL_KEY: b"990", tasks.MOBILEDE_BOOTSTRAP_UPDATED_TOTAL_KEY: b"5", tasks.MOBILEDE_BOOTSTRAP_IMAGES_TOTAL_KEY: b"7000", }.get(key) redis_client.scard.return_value = 0 finalized = tasks._mobilede_try_finalize_bootstrap(redis_client) self.assertTrue(finalized) redis_client.set.assert_any_call(tasks.MOBILEDE_BOOTSTRAP_DONE_KEY, "1") def test_force_full_scan_only_new_stays_full_pass_for_continuous_hourly_cycle(self) -> None: redis_client = self._FakeRedis() redis_client.store[tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY] = "10" redis_client.store[tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY] = "10" effective_only_new = tasks._mobilede_force_full_scan_only_new( True, redis_client=redis_client, continuous=True, ) self.assertFalse(effective_only_new) def test_post_bootstrap_refresh_disabled_for_continuous_hourly_cycle(self) -> None: redis_client = self._FakeRedis() redis_client.store[tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY] = "10" redis_client.store[tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY] = "10" post_bootstrap_refresh = tasks._mobilede_post_bootstrap_full_refresh( redis_client, False, continuous=True, ) self.assertFalse(post_bootstrap_refresh) def test_stalled_bootstrap_recovery_ignores_stale_global_progress_key(self) -> None: redis_client = self._FakeRedis() redis_client.store[tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY] = "13" redis_client.store[tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY] = "261" redis_client.store[tasks.GLOBAL_PROGRESS_TS_KEY] = "1" redis_client.sets[tasks.MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY] = {"seg-a", "seg-b"} with patch("mobilede_scraper.worker.tasks.time.time", return_value=10_000): recovered = tasks._mobilede_try_recover_stalled_bootstrap_queue(redis_client) self.assertTrue(recovered) self.assertNotIn(tasks.MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY, redis_client.sets) def test_stalled_bootstrap_recovery_keeps_recent_progress(self) -> None: redis_client = self._FakeRedis() redis_client.store[tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_DONE_KEY] = "13" redis_client.store[tasks.MOBILEDE_BOOTSTRAP_SEGMENTS_TOTAL_KEY] = "261" redis_client.store[tasks.GLOBAL_PROGRESS_TS_KEY] = "9950" redis_client.sets[tasks.MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY] = {"seg-a", "seg-b"} with patch("mobilede_scraper.worker.tasks.time.time", return_value=10_000): recovered = tasks._mobilede_try_recover_stalled_bootstrap_queue(redis_client) self.assertFalse(recovered) self.assertIn(tasks.MOBILEDE_BOOTSTRAP_DISPATCHED_SEGMENTS_KEY, redis_client.sets) def test_bootstrap_completion_does_not_finalize_incomplete_refresh_cycle(self) -> None: redis_client = self._FakeRedis() started_at = datetime(2026, 5, 6, 10, 0, tzinfo=timezone.utc) redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_ID_KEY] = "cycle-lag" redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_STARTED_AT_KEY] = started_at.isoformat() redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_TOTAL_KEY] = "377" redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_DONE_KEY] = "141" finalized = tasks._mobilede_try_finalize_refresh_cycle_after_bootstrap_completion( redis_client, cycle_id="cycle-lag", total_segments_hint=377, ) self.assertFalse(finalized) self.assertEqual(redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_DONE_KEY], "141") def test_refresh_cycle_unsafe_reason_skips_sold_marking(self) -> None: redis_client = self._FakeRedis() started_at = datetime(2026, 5, 6, 10, 0, tzinfo=timezone.utc) redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_ID_KEY] = "cycle-unsafe" redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_STARTED_AT_KEY] = started_at.isoformat() redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_TOTAL_KEY] = "1" redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_DONE_KEY] = "1" persistence = MagicMock() tasks._mobilede_mark_refresh_cycle_reconciliation_unsafe( redis_client, cycle_id="cycle-unsafe", reason="only_new", ) with patch.object(tasks, "_get_persistence", return_value=persistence): sold_marked = tasks._mobilede_finalize_refresh_cycle_sold_marking( redis_client, cycle_id="cycle-unsafe", ) self.assertEqual(sold_marked, 0) persistence.mark_sold_not_seen_since.assert_not_called() def test_known_complete_filtered_scope_is_safe_for_reconciliation(self) -> None: complete_url = "https://www.mobile.de/ru/search.html?isSearchRequest=true&s=Car&vc=Car&ms=25100&ms=8600&ms=15200&ms=3500&ref=dsp" runtime_config = RuntimeConfig() settings = MagicMock() settings.listing.filtered_search_urls = [complete_url] with patch.dict( "os.environ", {"MOBILEDE_COMPLETE_SCOPE_BRANDS": "BMW,Volvo,Ferrari,Lexus"}, clear=False, ): reasons = tasks._mobilede_complete_scope_unsafe_reasons(runtime_config, settings) self.assertEqual(reasons, []) def test_partial_filtered_scope_remains_unsafe_for_reconciliation(self) -> None: runtime_config = RuntimeConfig() runtime_config.sync.limit = 100 settings = MagicMock() settings.listing.filtered_search_urls = ["https://www.mobile.de/ru/search.html?ms=3500"] with patch.dict( "os.environ", { "MOBILEDE_COMPLETE_SCOPE_BRANDS": "BMW,Volvo,Ferrari,Lexus", }, clear=False, ): reasons = tasks._mobilede_complete_scope_unsafe_reasons(runtime_config, settings) self.assertIn("sync_limit", reasons) def test_effective_full_pass_is_safe_after_only_new_bootstrap(self) -> None: runtime_config = RuntimeConfig() runtime_config.sync.only_new = True settings = MagicMock() settings.listing.filtered_search_urls = ["https://www.mobile.de/ru/search.html?ms=25100&ms=8600&ms=15200&ms=3500"] with patch.dict( "os.environ", {"MOBILEDE_COMPLETE_SCOPE_BRANDS": "BMW,Volvo,Ferrari,Lexus"}, clear=False, ): reasons = tasks._mobilede_complete_scope_unsafe_reasons( runtime_config, settings, effective_only_new=False, ) self.assertEqual(reasons, []) def test_completed_bootstrap_segment_is_skipped_during_active_full_pass(self) -> None: redis_client = MagicMock() redis_client.get.side_effect = lambda key: "1" if str(key).startswith("mobilede:state:bootstrap_segment_done:") else None should_skip = tasks._mobilede_should_skip_completed_bootstrap_segment( redis_client, segment={"make_id": "3500", "price_min": "1", "price_max": "5000"}, only_new=False, bootstrap_run_active=True, ) self.assertTrue(should_skip) def test_queue_mobilede_bootstrap_recovery_can_force_requeue_existing_pending(self) -> None: redis_client = self._FakeRedis() redis_client.store[tasks.MOBILEDE_BOOTSTRAP_RECOVERY_PENDING_KEY] = "1" with patch.object(tasks.mobilede_sync_runtime_segments_task, "apply_async") as apply_async: queued = tasks._queue_mobilede_bootstrap_recovery( redis_client, lane="mobile_de_cars", delay_seconds=0.7, use_cursor=True, continuous=True, segment_label="runtime_segments", reason="waiting_for_active_bootstrap_tasks", countdown=5, force=True, ) self.assertTrue(queued) apply_async.assert_called_once() self.assertEqual(redis_client.store[tasks.MOBILEDE_BOOTSTRAP_RECOVERY_PENDING_KEY], "1") def test_runtime_segments_task_requeues_pending_bootstrap_recovery_while_tasks_are_still_active(self) -> None: redis_client = self._FakeRedis() redis_client.store[tasks.MOBILEDE_BOOTSTRAP_RECOVERY_PENDING_KEY] = "1" runtime_config = SimpleNamespace(sync=SimpleNamespace(only_new=False)) with ( patch.object(tasks, "_get_redis", return_value=redis_client), patch.object(tasks, "_mobilede_try_recover_stalled_bootstrap_queue", return_value=False), patch.object(tasks, "_get_cached_mobilede_runtime_segments", return_value=[{"make_id": "3500"}]), patch.object(tasks, "Settings", return_value=SimpleNamespace(runtime_config_file="runtime_config.json")), patch.object(tasks.RuntimeConfig, "from_file", return_value=runtime_config), patch.object(tasks, "_mobilede_force_full_scan_only_new", return_value=False), patch.object(tasks, "_mobilede_bootstrap_done", return_value=False), patch.object(tasks, "_mobilede_post_bootstrap_full_refresh", return_value=False), patch.object(tasks, "_mobilede_bootstrap_progress", return_value=(13, 261, 248)), patch.object(tasks, "_mobilede_bootstrap_progress_snapshot", return_value=(13, 261, 248, 2)), patch.object(tasks, "_has_recent_global_progress", return_value=False), patch.object(tasks, "_queue_mobilede_bootstrap_recovery", return_value=True) as queue_recovery, ): result = tasks.mobilede_sync_runtime_segments_task.run( lane="mobile_de_cars", delay_seconds=0.7, use_cursor=True, continuous=True, full_pass_repeat=False, ) self.assertEqual(result["status"], "bootstrap_active") queue_recovery.assert_called_once() def test_exhausted_non_cursor_segment_window_is_skipped(self) -> None: exhausted = tasks._mobilede_segment_window_is_exhausted( {"make_id": "3500", "max_pages": 50}, start_page=51, max_pages=50, use_cursor=False, ) not_exhausted = tasks._mobilede_segment_window_is_exhausted( {"make_id": "3500", "max_pages": 50}, start_page=1, max_pages=50, use_cursor=False, ) cursor_window = tasks._mobilede_segment_window_is_exhausted( {"make_id": "3500", "max_pages": 50}, start_page=51, max_pages=50, use_cursor=True, ) self.assertTrue(exhausted) self.assertFalse(not_exhausted) self.assertFalse(cursor_window) def test_explicit_start_page_is_not_overwritten_by_segment_default(self) -> None: segment = {"make_id": "3500", "start_page": 1} requested_start_page = 51 if segment.get("start_page") is not None and requested_start_page <= 1: effective_start_page = int(segment.get("start_page") or requested_start_page) else: effective_start_page = requested_start_page self.assertEqual(effective_start_page, 51) def test_refresh_cycle_tracking_finalizes_on_last_segment(self) -> None: redis_client = self._FakeRedis() redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_ID_KEY] = "cycle-a" redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_TOTAL_KEY] = "2" redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_DONE_KEY] = "1" done, total, should_finalize = tasks._mobilede_track_refresh_cycle_segment( redis_client, cycle_id="cycle-a", segment={"make_id": "3500", "price_min": "1", "price_max": "5000"}, total_segments_hint=2, ) self.assertEqual(done, 2) self.assertEqual(total, 2) self.assertTrue(should_finalize) def test_refresh_cycle_tracking_ignores_stale_cycle_id(self) -> None: redis_client = self._FakeRedis() redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_ID_KEY] = "cycle-active" redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_TOTAL_KEY] = "2" redis_client.store[tasks.MOBILEDE_REFRESH_CYCLE_DONE_KEY] = "0" done, total, should_finalize = tasks._mobilede_track_refresh_cycle_segment( redis_client, cycle_id="cycle-old", segment={"make_id": "3500", "price_min": "1", "price_max": "5000"}, total_segments_hint=2, ) self.assertEqual(done, 0) self.assertEqual(total, 0) self.assertFalse(should_finalize) def test_overflow_skip_when_known_total_fits_page_cap(self) -> None: redis_client = MagicMock() settings = MagicMock() segment = { "label": "Cars", "search_url": "https://www.mobile.de/ru/search.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&p=1:5000", "total_results": 995, "max_pages": tasks.MOBILEDE_MAX_PAGE_NUMBER, } added = tasks._mobilede_try_expand_overflow_segment( redis_client, settings, segment=segment, listing_count=1000, unique_count=995, max_pages=tasks.MOBILEDE_MAX_PAGE_NUMBER, segment_end_page=tasks.MOBILEDE_MAX_PAGE_NUMBER, ) self.assertEqual(added, 0) redis_client.sadd.assert_not_called() def test_late_overflow_children_are_not_queued_during_preplanned_bootstrap(self) -> None: redis_client = MagicMock() settings = MagicMock() segment = { "label": "Cars | ms=3500 | price=15001-20000", "search_url": "https://www.mobile.de/ru/search.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&p=15001:20000", "make_id": "3500", } with patch.object(planner, "os") as os_mock: os_mock.getenv.return_value = "true" queued = tasks._queue_mobilede_overflow_child_segments( redis_client, settings, parent_segment=segment, lane="mobile_de_cars", delay_seconds=0, use_cursor=True, only_new=False, bootstrap_run=True, refresh_cycle_id="cycle-a", ) self.assertEqual(queued, 0) redis_client.get.assert_not_called() def test_late_overflow_split_is_saved_for_next_pass_without_extending_bootstrap(self) -> None: redis_client = MagicMock() redis_client.sadd.return_value = 1 settings = MagicMock() parent = { "label": "Cars | ms=3500 | price=1-5000", "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=1:5000", "make_id": "3500", "price_min": "1", "price_max": "5000", "max_pages": tasks.MOBILEDE_MAX_PAGE_NUMBER, } parent_key = tasks._mobilede_segment_fingerprint(parent) child = { **parent, "search_url": "https://www.mobile.de/ru/search.html?ms=3500&p=1:2500", "price_max": "2500", "overflow_parent": parent_key, "overflow_depth": 1, } with ( patch.object(planner, "_mobilede_build_overflow_child_segments", return_value=[child]), patch.object(tasks, "_acquire_lock", return_value=True), patch.object(tasks, "_release_lock_if_owner"), patch.object(tasks, "_get_cached_mobilede_runtime_segments", return_value=[parent]), patch.object(planner, "_mobilede_source_segments_from_settings", return_value=[parent]), patch.object(planner, "_mobilede_save_learned_runtime_segments") as save_plan, patch.object(planner, "_mobilede_skip_late_overflow_children_during_bootstrap", return_value=True), ): added = tasks._mobilede_try_expand_overflow_segment( redis_client, settings, segment=parent, listing_count=1000, unique_count=1000, max_pages=tasks.MOBILEDE_MAX_PAGE_NUMBER, segment_end_page=tasks.MOBILEDE_MAX_PAGE_NUMBER, bootstrap_run=True, ) self.assertEqual(added, 0) self.assertFalse( any(call.args and call.args[0] == tasks.MOBILEDE_RUNTIME_SEGMENTS_CACHE_KEY for call in redis_client.set.call_args_list) ) save_plan.assert_called_once() learned_segments = save_plan.call_args.args[1] self.assertEqual(learned_segments, [child]) def test_pre_split_mileage_requires_explicit_flag(self) -> None: original_split = tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE try: tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = False self.assertFalse( tasks._mobilede_should_pre_split_mileage( price_min=30001, price_max=50000, year_min=2010, year_max=2014, ) ) finally: tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = original_split def test_pre_split_mileage_respects_explicit_override(self) -> None: original_split = tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE try: tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = True self.assertTrue( tasks._mobilede_should_pre_split_mileage( price_min=30001, price_max=50000, year_min=2010, year_max=2017, ) ) finally: tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = original_split def test_build_runtime_segments_splits_multi_make_filtered_search_url(self) -> None: settings = MagicMock() settings.listing.filtered_search_urls = [ "https://suchen.mobile.de/fahrzeuge/search.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&ms=11000&lang=en" ] with ( patch.object(planner, "_mobilede_try_keep_root_segment_unsplit", return_value=None), patch.object(planner, "_expand_mobilede_search_url_segment_by_probe", return_value=None), patch.object(planner, "_mobilede_probe_total", return_value=None), patch.object( planner, "_mobilede_preplan_runtime_segments", side_effect=lambda items: [{**item, "total_results": 1} for item in items], ), patch.object(planner, "_mobilede_compact_runtime_segments", side_effect=lambda items: items), ): segments = tasks._build_mobilede_runtime_segments(settings) self.assertGreater(len(segments), 2) self.assertEqual({str(segment.get("make_id")) for segment in segments}, {"3500", "11000"}) self.assertTrue(all("pageNumber=1" in str(segment.get("search_url")) for segment in segments)) self.assertTrue(all("ms=3500&ms=11000" not in str(segment.get("search_url")) for segment in segments)) def test_full_link_coverage_can_pre_split_every_segment_by_mileage(self) -> None: settings = MagicMock() settings.listing.filtered_search_urls = [ "https://suchen.mobile.de/fahrzeuge/search.html?isSearchRequest=true&s=Car&vc=Car&ms=111&ms=222&lang=en" ] original_split = tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE try: tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = True with ( patch.object(planner, "_mobilede_try_keep_root_segment_unsplit", return_value=None), patch.object(planner, "_expand_mobilede_search_url_segment_by_probe", return_value=None), patch.object(planner, "_mobilede_probe_total", return_value=None), patch.object( planner, "_mobilede_preplan_runtime_segments", side_effect=lambda items: [{**item, "total_results": 1} for item in items], ), patch.object(planner, "_mobilede_compact_runtime_segments", side_effect=lambda items: items), ): segments = tasks._build_mobilede_runtime_segments(settings) finally: tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = original_split self.assertGreater(len(segments), 2) self.assertEqual({str(segment.get("make_id")) for segment in segments}, {"111", "222"}) self.assertTrue(all("ms=111&ms=222" not in str(segment.get("search_url")) for segment in segments)) if __name__ == "__main__": unittest.main()