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

Initial project version

Browse Source
This commit is contained in:
qananasikq
2026-04-08 23:36:19 +03:00
commit 50d107cfba
54 changed files with 4592 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)

151
tests/test_db.py Normal file
View File

@@ -0,0 +1,151 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from sqlalchemy import select
from iaai_scraper.core.config import Settings
from iaai_scraper.storage.db import PersistenceService
from iaai_scraper.storage.models import Car, Image, SyncRun
from iaai_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, content_hash: str = "hash1") -> CarRecord:
return CarRecord(
parser_id=f"iaai:{origin_id}",
brand="Toyota",
model="Camry",
year=2014,
price=price,
origin_url=f"https://www.iaai.com/VehicleDetail/{origin_id}~US",
origin_id=origin_id,
slug=f"toyota-camry-{origin_id}",
content_hash=content_hash,
images=[
ImageRecord(
fullres_image="https://vis.iaai.com/resizer?imageKeys=1&width=845&height=633",
preview_image="https://vis.iaai.com/resizer?imageKeys=1&width=400&height=300",
order_index=0,
)
],
raw_attributes={"foo": "bar"},
)
def test_insert_update_and_skip_flow(self) -> None:
first = self._record("777", price=1000, content_hash="same")
inserted = self.persistence.upsert_car(first)
self.assertEqual(inserted["action"], "inserted")
self.assertEqual(inserted["images_upserted"], 1)
same = self._record("777", price=1000, content_hash="same")
updated_same = self.persistence.upsert_car(same)
self.assertEqual(updated_same["action"], "skipped")
self.assertEqual(updated_same["images_upserted"], 0)
changed = self._record("777", price=1500, content_hash="changed")
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", content_hash="v1")
self.persistence.upsert_car(first)
second = self._record("888", content_hash="v2")
second.images = [
ImageRecord(
fullres_image="https://vis.iaai.com/resizer?imageKeys=2&width=845&height=633",
preview_image="https://vis.iaai.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_same_content_hash_is_skipped(self) -> None:
first = self._record("999", content_hash="same-hash")
self.persistence.upsert_car(first)
second = self._record("999", content_hash="same-hash")
result = self.persistence.upsert_car(second)
self.assertEqual(result["action"], "skipped")
self.assertEqual(result["images_upserted"], 0)
def test_persistence_ignores_non_db_fields(self) -> None:
record = self._record("1000", content_hash="schema-test")
record.raw_attributes = {"vin": "123"}
record.mapping_notes = ["note"]
result = self.persistence.upsert_car(record)
self.assertEqual(result["action"], "inserted")
with self.persistence.session_scope() as session:
car = session.execute(select(Car).where(Car.origin_id == "1000")).scalar_one()
self.assertEqual(car.origin_id, "1000")
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", content_hash="v1")
first.origin_url = "https://www.iaai.com/VehicleDetail/45089484~US"
self.persistence.upsert_car(first)
second = self._record("NEW-ID", content_hash="v2")
second.origin_url = "https://www.iaai.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()

42
tests/test_listing.py Normal file
View File

@@ -0,0 +1,42 @@
from __future__ import annotations
import unittest
from iaai_scraper.browser.pace import HumanPacer
from iaai_scraper.core.config import Settings
from iaai_scraper.browser.listing import ListingCollector
class _FakePage:
def __init__(self, counts: dict[str, int]) -> None:
self._counts = counts
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))
class TestListingUnit(unittest.TestCase):
def test_has_next_page_true_for_known_selector(self) -> None:
page = _FakePage({"a[aria-label*='Next']": 1})
self.assertTrue(ListingCollector._has_next_page(page))
def test_has_next_page_false_when_no_selectors(self) -> None:
page = _FakePage({})
self.assertFalse(ListingCollector._has_next_page(page))
def test_constructor_with_settings_and_pacer(self) -> None:
settings = Settings()
pacer = HumanPacer(settings)
collector = ListingCollector(settings, pacer)
self.assertIsNotNone(collector)
if __name__ == "__main__":
unittest.main()

139
tests/test_mappers.py Normal file
View File

@@ -0,0 +1,139 @@
from __future__ import annotations
import unittest
from iaai_scraper.parsing.mapper import CarMapper
class TestCarMapper(unittest.TestCase):
def setUp(self) -> None:
self.mapper = CarMapper()
def test_content_hash_is_sha256(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://www.iaai.com/VehicleDetail/45089484~US",
vehicle_summary={"make": "Toyota", "model": "Camry", "year": "2014"},
payload_insights={
"vehicle_core": {"odometer": "120,000", "body_type": "sedan"},
"pricing": {"buy_now": "$4,500"},
"damage": {"primary": "normal wear"},
"auction": {},
"images": {"urls": []},
},
)
self.assertEqual(len(record.content_hash), 64)
def test_content_hash_changes_when_image_set_changes(self) -> None:
record_one = self.mapper.map_to_car_record(
vehicle_url="https://www.iaai.com/VehicleDetail/45089484~US",
vehicle_summary={
"make": "Toyota",
"model": "Camry",
"year": "2014",
"image_urls": ["https://vis.iaai.com/resizer?imageKeys=1&width=845&height=633"],
},
payload_insights={"vehicle_core": {}, "pricing": {}, "damage": {}, "auction": {}, "images": {}},
)
record_two = self.mapper.map_to_car_record(
vehicle_url="https://www.iaai.com/VehicleDetail/45089484~US",
vehicle_summary={
"make": "Toyota",
"model": "Camry",
"year": "2014",
"image_urls": ["https://vis.iaai.com/resizer?imageKeys=2&width=845&height=633"],
},
payload_insights={"vehicle_core": {}, "pricing": {}, "damage": {}, "auction": {}, "images": {}},
)
self.assertNotEqual(record_one.content_hash, record_two.content_hash)
def test_deduplicates_images_by_image_key(self) -> None:
urls = [
"https://vis.iaai.com/resizer?imageKeys=1&width=200&height=150",
"https://vis.iaai.com/resizer?imageKeys=1&width=845&height=633",
"https://vis.iaai.com/resizer?imageKeys=2&width=400&height=300",
]
record = self.mapper.map_to_car_record(
vehicle_url="https://www.iaai.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.iaai.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_and_empty_values_fallbacks(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://www.iaai.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.assertIsNone(record.steering_wheel if record.steering_wheel not in {"LEFT", None} else None)
self.assertEqual(record.drive, "NA")
self.assertEqual(record.gearbox, "NA")
def test_mapper_handles_case_and_spaces(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://www.iaai.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(record.drive, "FWD")
self.assertEqual(record.gearbox, "AT")
def test_price_parsing_dirty_formats(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://www.iaai.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)
def test_currency_detection_from_symbol(self) -> None:
record = self.mapper.map_to_car_record(
vehicle_url="https://www.iaai.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(record.currency, "EUR")
self.assertEqual(record.price, 4500)
if __name__ == "__main__":
unittest.main()

62
tests/test_parser.py Normal file
View File

@@ -0,0 +1,62 @@
from __future__ import annotations
import unittest
from iaai_scraper.parsing.parser import VehicleParser
class TestVehicleParserUnit(unittest.TestCase):
def setUp(self) -> None:
self.parser = VehicleParser()
def test_parse_dom_key_value_pairs_extracts_known_fields(self) -> None:
dom_text = """
Stock #:
45089484
VIN (Status):
1HGCM82633A123456 (OK)
Primary Damage:
Front End
"""
result = self.parser._parse_dom_key_value_pairs(dom_text)
self.assertEqual(result.get("lot_number"), "45089484")
self.assertEqual(result.get("vin"), "1HGCM82633A123456 (OK)")
self.assertEqual(result.get("primary_damage"), "Front End")
def test_parse_title_for_year_make_model(self) -> None:
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_other_vehicle(self) -> None:
vehicle_url = "https://www.iaai.com/VehicleDetail/45089484~US"
payloads = [
{
"imageUrls": [
"https://vis.iaai.com/resizer?imageKeys=45089484~SID1&width=845&height=633",
"https://vis.iaai.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_extract_image_urls_deduplicates_same_url(self) -> None:
vehicle_url = "https://www.iaai.com/VehicleDetail/45089484~US"
payloads = [{"imageUrls": [
"https://vis.iaai.com/resizer?imageKeys=45089484~SID1&width=845&height=633",
"https://vis.iaai.com/resizer?imageKeys=45089484~SID1&width=845&height=633",
]}]
urls = self.parser._extract_image_urls(payloads, "", vehicle_url)
self.assertEqual(len(urls), 1)
def test_dom_hints_detect_captcha_and_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"])
if __name__ == "__main__":
unittest.main()

154
tests/test_scraper.py Normal file
View File

@@ -0,0 +1,154 @@
from __future__ import annotations
import unittest
from unittest.mock import MagicMock
from iaai_scraper.core.config import Settings
from iaai_scraper.scraper import IAAIScraper
from iaai_scraper.storage.schemas import CarRecord
def make_db_record(origin_id: str) -> dict[str, object]:
return CarRecord(
parser_id=f"iaai:{origin_id}",
brand="Toyota",
model="Camry",
origin_url=f"https://www.iaai.com/VehicleDetail/{origin_id}~US",
origin_id=origin_id,
slug=f"toyota-camry-{origin_id}",
).model_dump(mode="json")
class TestScraperSync(unittest.TestCase):
def _make_scraper(self) -> IAAIScraper:
s = Settings()
s.log_level = "CRITICAL"
s.database.url = "sqlite://"
return IAAIScraper(s)
def test_sync_vehicle_uses_db_record_without_remapping(self) -> None:
scraper = self._make_scraper()
scraper.persistence.create_tables = MagicMock()
scraper.persistence.start_sync_run = MagicMock(return_value=1)
scraper.persistence.finish_sync_run = MagicMock()
scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 0})
scraper.scrape_vehicle_detail = MagicMock(return_value={"db_record": make_db_record("111")})
scraper.car_mapper.map_to_car_record = MagicMock(side_effect=AssertionError("should not be called"))
result = scraper.sync_vehicle("https://www.iaai.com/VehicleDetail/111~US")
self.assertEqual(result["status"], "success")
self.assertIn("trace_id", result)
self.assertIn("elapsed_seconds", result)
self.assertEqual(scraper.persistence.upsert_car.call_count, 1)
def test_sync_listing_uses_db_record_without_remapping(self) -> None:
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.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 1})
scraper.persistence.get_existing_origin_urls = MagicMock(return_value=set())
scraper.persistence.get_existing_origin_ids = MagicMock(return_value=set())
scraper.collect_listing = MagicMock(return_value={"vehicle_urls": ["https://www.iaai.com/VehicleDetail/222~US"]})
page = MagicMock()
scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("222")})
scraper._get_page = MagicMock(return_value=page)
scraper.car_mapper.map_to_car_record = MagicMock(side_effect=AssertionError("should not be called"))
result = scraper.sync_listing()
self.assertEqual(result["cars_upserted"], 1)
self.assertEqual(result["cars_failed"], 0)
self.assertIn("trace_id", result)
self.assertIn("elapsed_seconds", result)
self.assertEqual(scraper.persistence.upsert_car.call_count, 1)
page.close.assert_called_once()
def test_sync_listing_respects_limit(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()
scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 1})
scraper.persistence.get_existing_origin_urls = MagicMock(return_value=set())
scraper.persistence.get_existing_origin_ids = MagicMock(return_value=set())
scraper.collect_listing = MagicMock(return_value={
"vehicle_urls": [
"https://www.iaai.com/VehicleDetail/222~US",
"https://www.iaai.com/VehicleDetail/333~US",
]
})
scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("222")})
scraper._get_page = MagicMock(return_value=MagicMock())
scraper.sync_listing(limit=1)
self.assertEqual(scraper._scrape_on_page.call_count, 1)
def test_sync_listing_does_not_count_skipped_as_upserted(self) -> None:
scraper = self._make_scraper()
scraper.persistence.create_tables = MagicMock()
scraper.persistence.start_sync_run = MagicMock(return_value=4)
scraper.persistence.finish_sync_run = MagicMock()
scraper.persistence.upsert_car = MagicMock(return_value={"action": "skipped", "images_upserted": 0})
scraper.persistence.get_existing_origin_urls = MagicMock(return_value=set())
scraper.persistence.get_existing_origin_ids = MagicMock(return_value=set())
scraper.collect_listing = MagicMock(return_value={"vehicle_urls": ["https://www.iaai.com/VehicleDetail/444~US"]})
scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("444")})
scraper._get_page = MagicMock(return_value=MagicMock())
result = scraper.sync_listing()
self.assertEqual(result["cars_upserted"], 0)
def test_sync_listing_only_new_filters_existing_by_url_and_origin_id(self) -> None:
scraper = self._make_scraper()
scraper.persistence.create_tables = MagicMock()
scraper.persistence.start_sync_run = MagicMock(return_value=5)
scraper.persistence.finish_sync_run = MagicMock()
scraper.persistence.upsert_car = MagicMock(return_value={"action": "inserted", "images_upserted": 0})
scraper.persistence.get_existing_origin_urls = MagicMock(return_value={"https://www.iaai.com/VehicleDetail/111~US"})
scraper.persistence.get_existing_origin_ids = MagicMock(return_value={"222"})
scraper.collect_listing = MagicMock(return_value={
"vehicle_urls": [
"https://www.iaai.com/VehicleDetail/111~US", # exists by URL
"https://www.iaai.com/VehicleDetail/222~US", # exists by ID
"https://www.iaai.com/VehicleDetail/333~US", # new
]
})
page = MagicMock()
scraper._get_page = MagicMock(return_value=page)
scraper._scrape_on_page = MagicMock(return_value={"db_record": make_db_record("333")})
result = scraper.sync_listing(only_new=True)
self.assertEqual(result["skipped_existing"], 2)
self.assertEqual(result["cars_upserted"], 1)
self.assertEqual(scraper._scrape_on_page.call_count, 1)
scraper.persistence.get_existing_origin_urls.assert_called_once()
scraper.persistence.get_existing_origin_ids.assert_called_once()
def test_close_resets_browser_state(self) -> None:
scraper = self._make_scraper()
scraper.context = MagicMock()
scraper.browser = MagicMock()
scraper.playwright = MagicMock()
scraper.close()
self.assertIsNone(scraper.context)
self.assertIsNone(scraper.browser)
self.assertIsNone(scraper.playwright)
if __name__ == "__main__":
unittest.main()

31
tests/test_utils.py Normal file
View File

@@ -0,0 +1,31 @@
from __future__ import annotations
import unittest
from iaai_scraper.core.utils import deep_find_key
class TestDeepFindKey(unittest.TestCase):
def test_finds_key_in_nested_structure(self) -> None:
payload = {
"root": {
"target": "a",
"nested": [{"target": "b"}, {"x": 1}],
}
}
found = deep_find_key(payload, {"target"})
self.assertEqual(found, ["a", "b"])
def test_respects_max_depth(self) -> None:
payload = {"l1": {"l2": {"l3": {"target": "value"}}}}
found_too_shallow = deep_find_key(payload, {"target"}, max_depth=2)
found_enough_depth = deep_find_key(payload, {"target"}, max_depth=8)
self.assertEqual(found_too_shallow, [])
self.assertEqual(found_enough_depth, ["value"])
if __name__ == "__main__":
unittest.main()