108 lines
5.4 KiB
Python
108 lines
5.4 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from urllib.parse import parse_qs, urlsplit
|
||
|
||
from sqlalchemy import select
|
||
|
||
from mobilede_scraper.core.config import Settings
|
||
from mobilede_scraper.core.runtime_config import RuntimeFiltersConfig
|
||
from mobilede_scraper.mobile_de.client import DEFAULT_HEADERS, MobileDeClient
|
||
from mobilede_scraper.mobile_de.mapper import MobileDeMapper
|
||
from mobilede_scraper.mobile_de.models import MobileDeListing, MobileDeSearchPage
|
||
from mobilede_scraper.mobile_de.scraper import MobileDeScraper
|
||
from mobilede_scraper.storage.db import PersistenceService
|
||
from mobilede_scraper.storage.models import Car
|
||
from mobilede_scraper.storage.schemas import CarRecord
|
||
|
||
|
||
class _Client:
|
||
def __init__(self, pages: list[MobileDeSearchPage]) -> None:
|
||
self.pages = pages
|
||
|
||
def iter_search_pages(self, **kwargs):
|
||
del kwargs
|
||
yield from self.pages
|
||
|
||
|
||
class _Persistence:
|
||
def __init__(self) -> None:
|
||
self.records: list[CarRecord] = []
|
||
|
||
def create_tables(self) -> None: pass
|
||
def start_sync_run(self, lane: str) -> int:
|
||
del lane
|
||
return 1
|
||
def get_existing_origin_ids(self, origin_ids: list[str]) -> set[str]:
|
||
del origin_ids
|
||
return set()
|
||
def upsert_cars_batch(self, records: list[CarRecord]) -> dict[str, int]:
|
||
self.records.extend(records)
|
||
return {"inserted": len(records), "updated": 0, "images_upserted": 0}
|
||
def finish_sync_run(self, run_id: int, **kwargs) -> None:
|
||
del run_id, kwargs
|
||
|
||
|
||
class TestMobileDeRequirements(unittest.TestCase):
|
||
def test_english_urls_and_header_preserve_existing_query(self) -> None:
|
||
search = MobileDeClient.build_search_url_from_existing(
|
||
"https://www.mobile.de/ru/search.html?ms=3500&lang=de&custom=keep",
|
||
page_number=3,
|
||
)
|
||
detail = MobileDeClient.build_detail_url("123")
|
||
self.assertEqual(DEFAULT_HEADERS["accept-language"], "en-US,en;q=0.9")
|
||
self.assertEqual(urlsplit(search).scheme, "https")
|
||
self.assertEqual(urlsplit(search).netloc, "suchen.mobile.de")
|
||
self.assertEqual(urlsplit(search).path, "/fahrzeuge/search.html")
|
||
self.assertEqual(urlsplit(detail).path, "/fahrzeuge/details.html")
|
||
self.assertEqual(parse_qs(urlsplit(search).query)["lang"], ["en"])
|
||
self.assertEqual(parse_qs(urlsplit(search).query)["custom"], ["keep"])
|
||
self.assertEqual(parse_qs(urlsplit(detail).query)["lang"], ["en"])
|
||
|
||
def test_parser_id_is_deterministic_and_matches_project_format(self) -> None:
|
||
origin_id = "mobile.de:123"
|
||
parser_id = MobileDeMapper._parser_id(origin_id)
|
||
self.assertRegex(parser_id, r"^car-[A-Za-z0-9]{22}$")
|
||
self.assertEqual(parser_id, MobileDeMapper._parser_id(origin_id))
|
||
self.assertNotEqual(parser_id, origin_id)
|
||
|
||
def test_mapper_contains_no_cyrillic_literals(self) -> None:
|
||
mapper_path = Path(__file__).parents[1] / "mobilede_scraper" / "mobile_de" / "mapper.py"
|
||
self.assertIsNone(re.search(r"[А-Яа-яЁё]", mapper_path.read_text(encoding="utf-8")))
|
||
|
||
def test_upsert_uses_origin_id_and_migrates_legacy_parser_id(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
settings = Settings()
|
||
settings.database.url = f"sqlite:///{Path(directory, 'db.sqlite').as_posix()}"
|
||
service = PersistenceService(settings)
|
||
service.create_tables()
|
||
origin_id = "mobile.de:1"
|
||
legacy = CarRecord(parser_id="car-0123456789abcdef", brand="BMW", model="X1", origin_id=origin_id, origin_url="https://example.test/1", slug="bmw-x1")
|
||
current = legacy.model_copy(update={"parser_id": MobileDeMapper._parser_id(origin_id), "price": 20})
|
||
service.upsert_car(legacy)
|
||
service.upsert_car(current)
|
||
service.upsert_car(current)
|
||
with service.session_scope() as session:
|
||
cars = session.execute(select(Car)).scalars().all()
|
||
self.assertEqual(len(cars), 1)
|
||
self.assertEqual(cars[0].parser_id, current.parser_id)
|
||
service.engine.dispose()
|
||
|
||
def test_runtime_filters_and_limit_apply_after_mapping(self) -> None:
|
||
pages = [MobileDeSearchPage(url="https://example.test", page_number=1, total_results=3, listings=[
|
||
MobileDeListing(id="1", url="https://example.test/1", title="BMW X1", raw={"attr": {"yc": "2020", "ml": "10000", "c": "OffRoad", "ecol": "Black", "tr": "Automatic"}, "price": {"grs": {"amount": "20000"}}}),
|
||
MobileDeListing(id="2", url="https://example.test/2", title="BMW X3", raw={"attr": {"yc": "2020", "ml": "10000", "c": "OffRoad", "ecol": "Black", "tr": "Automatic"}, "price": {"grs": {"amount": "21000"}}}),
|
||
MobileDeListing(id="3", url="https://example.test/3", title="Audi Q5", raw={"attr": {"yc": "2020", "ml": "10000", "c": "OffRoad", "ecol": "Black", "tr": "Automatic"}, "price": {"grs": {"amount": "22000"}}}),
|
||
])]
|
||
persistence = _Persistence()
|
||
filters = RuntimeFiltersConfig.from_dict({"brands": ["BMW"], "price": {"min": 15000, "max": 25000}, "exclude_models": ["X3"]})
|
||
MobileDeScraper(client=_Client(pages), persistence=persistence).sync_search(runtime_filters=filters, limit=1)
|
||
self.assertEqual([record.origin_id for record in persistence.records], ["mobile.de:1"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|