import gzip import json import tempfile import unittest from pathlib import Path from unittest.mock import patch from car_tracker.collect import SourceAccessError, collect, minimize, parse_detail, parse_search, search_url def payload(ids=(123,), total=1, pages=1): return json.dumps({"props": {"pageProps": {"searchData": { "ads": [{"list_id": i, "price": ["12000"], "body": ""} for i in ids], "total": total, "max_pages": pages, }}}}) def detail_payload(identifier=123): return json.dumps({"props": {"pageProps": {"ad": { "list_id": identifier, "subject": "Synthetic Saab 900 Turbo", "body": "Synthetic description: serviced in 2025.", "price": ["12500"], "attributes": [{"key": "vehicle_new_option", "value": "original"}], "location": {"lat": 43.6, "lng": 1.4, "city": "Test", "zipcode": "31000"}, "owner": {"name": "Synthetic Seller"}, }}}}) class CollectorTests(unittest.TestCase): def test_minimization_preserves_values_without_seller_data(self): attribute = {"key": "mileage", "value": "unknown", "value_label": "Unknown"} result = minimize({ "list_id": 123, "price": ["12000"], "subject": "free text", "body": "Synthetic description", "owner": {"name": "Synthetic Seller"}, "images": {"urls": ["image"]}, "attributes": [attribute, {"key": "profile_picture_url", "value": "private"}], "location": {"department_id": "31", "lat": 43, "city": "Test"}, }) self.assertEqual(result, { "list_id": 123, "price": ["12000"], "attributes": [attribute], "subject": "free text", "body": "Synthetic description", "location": {"department_id": "31", "lat": 43, "city": "Test"}, }) def test_new_attributes_and_location_metadata_are_retained(self): ad = parse_detail(detail_payload(), 123) ad["location"]["origin_type"] = "city" result = minimize(ad) self.assertEqual(result["attributes"], ad["attributes"]) self.assertEqual(result["location"], ad["location"]) self.assertNotIn("owner", result) def test_missing_fields_stay_missing(self): self.assertEqual(minimize({"list_id": 123}), {"list_id": 123}) self.assertEqual(minimize({"list_id": 123, "attributes": [], "location": {}}), {"list_id": 123, "attributes": [], "location": {}}) def test_known_sensitive_attributes_are_excluded(self): keys = ["profile_picture_url", "seller_name", "contact_phone", "email", "rating_score", "VIN", "vehicle_history_report_public_url"] result = minimize({"attributes": [{"key": k, "value": "private"} for k in keys]}) self.assertEqual(result["attributes"], []) def test_detail_payload_must_match_requested_listing(self): for text in ("blocked", payload(), detail_payload(456), detail_payload("../123")): with self.subTest(text=text), self.assertRaises(ValueError): parse_detail(text, 123) def test_invalid_payload_is_not_an_empty_market(self): for text in ("blocked", "{}", payload([], total=10), payload(pages=-1)): with self.subTest(text=text), self.assertRaises(ValueError): parse_search(text) def test_search_is_nationwide_and_encoded(self): self.assertEqual(search_url("saab 900", 2), "https://www.leboncoin.fr/recherche?category=2&text=saab+900&page=2") def run_collection(self, responses, **kwargs): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) iterator = iter(responses) def fetch(url): result = next(iterator) if isinstance(result, BaseException): raise result return result with patch("car_tracker.collect.time.sleep"): kwargs.setdefault("search_only", True) path, status = collect(fetch, [{"family": "synthetic", "text": "test"}], Path(temporary.name), **kwargs) manifest = json.loads(path.read_text()) self.assertEqual(status, manifest["status"]) return path, manifest def test_complete_pagination_and_replay(self): path, manifest = self.run_collection([ payload([123], total=2, pages=2), payload([456], total=2, pages=2), ]) self.assertEqual(manifest["status"], "complete") self.assertEqual(manifest["searches"][0]["records"], 2) for page in manifest["searches"][0]["pages"]: with gzip.open(path.parent / page["file"], "rt") as stream: record = json.loads(stream.readline()) self.assertEqual(record["run_id"], manifest["run_id"]) self.assertEqual(record["payload"]["price"], ["12000"]) self.assertEqual(record["payload"]["body"], "") self.assertEqual(record["record_type"], "search") self.assertEqual(record["schema_version"], 2) def test_page_limit_is_partial(self): _, manifest = self.run_collection([payload(pages=2)], max_pages=1) self.assertEqual(manifest["status"], "partial") self.assertEqual(manifest["searches"][0]["reason"], "page_limit") def test_failure_preserves_previous_page(self): path, manifest = self.run_collection([payload(pages=2), RuntimeError("private body")]) self.assertEqual(manifest["status"], "partial") self.assertEqual(len(list(path.parent.glob("*.jsonl.gz"))), 1) self.assertNotIn("private body", path.read_text()) def test_block_is_failed_not_complete(self): _, manifest = self.run_collection([SourceAccessError(403)]) self.assertEqual(manifest["status"], "failed") self.assertEqual(manifest["searches"][0]["http_status"], 403) def test_403_capture_is_linked_without_exposing_private_contents(self): from unittest.mock import Mock capture = Mock(return_value={"captured": ["response-body.html"], "errors": {}}) path, manifest = self.run_collection([payload(pages=2), SourceAccessError(403, capture=capture)]) capture.assert_called_once_with(path.parent / "diagnostics/http-403") self.assertEqual(manifest["diagnostics"]["capture_status"], "complete") self.assertEqual(manifest["error_type"], "SourceAccessError") self.assertEqual(len(list(path.parent.glob("*.jsonl.gz"))), 1) def test_capture_failure_preserves_original_error_and_records(self): from unittest.mock import Mock capture = Mock(side_effect=OSError("private filesystem details")) path, manifest = self.run_collection([payload(pages=2), SourceAccessError(403, capture=capture)]) self.assertEqual(manifest["diagnostics"]["capture_status"], "failed") self.assertEqual(manifest["searches"][0]["http_status"], 403) self.assertNotIn("private filesystem", path.read_text()) def test_capture_does_not_run_for_other_statuses(self): from unittest.mock import Mock capture = Mock() self.run_collection([SourceAccessError(429, capture=capture)]) capture.assert_not_called() def test_discovery_failure_does_not_claim_detail_completion(self): _, manifest = self.run_collection([SourceAccessError(403)], search_only=False) self.assertEqual(manifest["details"]["status"], "not_attempted") def test_discovered_empty_market_has_complete_detail_coverage(self): _, manifest = self.run_collection([payload([], total=0, pages=0)], search_only=False) self.assertEqual(manifest["status"], "complete") self.assertEqual(manifest["details"]["status"], "complete") def test_failure_stops_remaining_searches(self): with tempfile.TemporaryDirectory() as directory: searches = [{"family": "synthetic", "text": "test"}, {"family": "synthetic-2", "text": "test-2"}] path, status = collect(lambda url: "blocked", searches, Path(directory)) manifest = json.loads(path.read_text()) self.assertEqual(status, "failed") self.assertEqual(manifest["planned_searches"], searches) self.assertEqual(len(manifest["searches"]), 1) def test_repeated_page_is_partial(self): _, manifest = self.run_collection([payload(pages=2), payload(pages=2)]) self.assertEqual(manifest["status"], "partial") self.assertEqual(manifest["searches"][0]["records"], 1) def test_empty_search_is_complete(self): _, manifest = self.run_collection([payload([], total=0, pages=0)]) self.assertEqual(manifest["status"], "complete") self.assertEqual(manifest["searches"][0]["records"], 0) def test_interrupt_is_recorded(self): _, manifest = self.run_collection([KeyboardInterrupt()]) self.assertEqual(manifest["status"], "failed") self.assertEqual(manifest["searches"][0]["status"], "failed") def test_detail_observation_does_not_overwrite_search(self): path, manifest = self.run_collection([payload(), detail_payload()], search_only=False) self.assertEqual(manifest["status"], "complete") self.assertEqual(manifest["details"]["status"], "complete") item = manifest["details"]["listings"]["123"] with gzip.open(path.parent / item["file"], "rt") as stream: detail = json.loads(stream.readline()) with gzip.open(path.parent / manifest["searches"][0]["pages"][0]["file"], "rt") as stream: search = json.loads(stream.readline()) self.assertEqual(detail["record_type"], "detail") self.assertEqual(detail["payload"]["body"], "Synthetic description: serviced in 2025.") self.assertEqual(detail["payload"]["location"]["lng"], 1.4) self.assertEqual(detail["payload"]["price"], ["12500"]) self.assertEqual(search["payload"]["price"], ["12000"]) self.assertEqual(search["payload"]["body"], "") self.assertNotIn("owner", detail["payload"]) self.assertGreaterEqual(detail["observed_at"], search["observed_at"]) def test_detail_block_preserves_discovery_and_stops(self): path, manifest = self.run_collection([ payload([123, 456], total=2), SourceAccessError(403), ], search_only=False) self.assertEqual(manifest["status"], "partial") self.assertEqual(manifest["searches"][0]["status"], "complete") self.assertEqual(len(list(path.parent.glob("*.jsonl.gz"))), 1) self.assertEqual(manifest["details"]["listings"]["123"]["http_status"], 403) self.assertEqual(manifest["details"]["listings"]["456"]["status"], "not_attempted") self.assertEqual(manifest["details"]["attempted"], 1) def test_unavailable_detail_does_not_stop_other_details(self): for status in (404, 410): with self.subTest(status=status): _, manifest = self.run_collection([ payload([123, 456], total=2), SourceAccessError(status), detail_payload(456), ], search_only=False) self.assertEqual(manifest["status"], "partial") items = manifest["details"]["listings"] self.assertEqual(items["123"]["status"], "unavailable") self.assertEqual(items["456"]["status"], "complete") def test_malformed_detail_is_not_success(self): _, manifest = self.run_collection([payload(), detail_payload(456)], search_only=False) self.assertEqual(manifest["status"], "partial") self.assertEqual(manifest["details"]["listings"]["123"]["reason"], "ValueError") def test_detail_limit_does_not_limit_search_pagination(self): _, manifest = self.run_collection([ payload([123], total=2, pages=2), detail_payload(), payload([456], total=2, pages=2), ], search_only=False, max_details=1) self.assertEqual(manifest["status"], "partial") self.assertEqual(manifest["searches"][0]["status"], "complete") self.assertEqual(manifest["details"]["attempted"], 1) self.assertEqual(manifest["details"]["listings"]["456"]["reason"], "detail_limit") def test_detail_interrupt_preserves_search(self): _, manifest = self.run_collection([payload(), KeyboardInterrupt()], search_only=False) self.assertEqual(manifest["status"], "partial") self.assertEqual(manifest["searches"][0]["status"], "complete") self.assertEqual(manifest["details"]["listings"]["123"]["status"], "failed") def test_interrupt_during_delay_does_not_count_as_detail_attempt(self): with tempfile.TemporaryDirectory() as directory: with patch("car_tracker.collect.time.sleep", side_effect=KeyboardInterrupt()): path, status = collect(lambda url: payload(), [{"family": "synthetic", "text": "test"}], Path(directory)) manifest = json.loads(path.read_text()) self.assertEqual(status, "partial") self.assertEqual(manifest["details"]["attempted"], 0) self.assertEqual(manifest["details"]["listings"]["123"]["status"], "not_attempted") def test_overlapping_searches_fetch_detail_once_and_pace_all_requests(self): with tempfile.TemporaryDirectory() as directory: searches = [{"family": "synthetic", "text": "test"}, {"family": "synthetic", "text": "test-2"}] with patch("car_tracker.collect.time.sleep") as sleep: responses = iter([payload(), detail_payload(), payload()]) urls = [] def fetch(url): urls.append(url) return next(responses) path, status = collect(fetch, searches, Path(directory), delay=7) manifest = json.loads(path.read_text()) self.assertEqual(status, "complete") self.assertEqual(manifest["details"]["attempted"], 1) self.assertEqual(len(urls), 3) self.assertEqual(urls[1], "https://www.leboncoin.fr/ad/voitures/123") self.assertEqual(sleep.call_count, 2) sleep.assert_called_with(7) def test_search_only_explicitly_disables_detail_coverage(self): _, manifest = self.run_collection([payload()]) self.assertEqual(manifest["details"]["status"], "disabled") self.assertEqual(manifest["details"]["attempted"], 0) if __name__ == "__main__": unittest.main()