from __future__ import annotations
import json
from pathlib import Path
from iaai_sync_service.client import (
LISTING_MARKER,
HybridSessionAuth,
build_resizer_images_from_keys,
is_challenge_response,
parse_listing_page,
parse_product_details_vm,
resolve_listing_scope_paths,
)
from iaai_sync_service.settings import Settings
def _catalog_html() -> str:
return (Path(__file__).resolve().parents[1] / "html" / "catalog.html").read_text(encoding="utf-8")
def _page_html() -> str:
return (Path(__file__).resolve().parents[1] / "html" / "page.html").read_text(encoding="utf-8")
def test_parse_listing_page_extracts_hidden_payloads() -> None:
parsed = parse_listing_page(_catalog_html())
assert parsed.result_count >= len(parsed.vehicles)
assert parsed.page_size == 100
assert parsed.current_page == 1
assert parsed.vehicles
first = parsed.vehicles[0]
assert first.inventory_id.endswith("~US")
assert first.inventory_status == "RS"
def test_parse_product_details_vm_extracts_json_script() -> None:
payload = parse_product_details_vm(_page_html())
inventory_view = payload["inventoryView"]
attrs = inventory_view["attributes"]
assert attrs["Id"] == "45078011~US"
assert attrs["Make"] == "ACURA"
assert attrs["Model"] == "RSX"
def test_build_resizer_images_from_keys_deduplicates_and_orders() -> None:
keys = [
{"k": "abc~1", "w": 2000, "h": 1500, "i": 2},
{"k": "abc~1", "w": 2000, "h": 1500, "i": 2},
{"k": "abc~2", "w": 1800, "h": 1200, "i": 1},
]
images = build_resizer_images_from_keys(keys)
assert len(images) == 2
assert images[0]["order_index"] == 1
assert "imageKeys=abc~2" in str(images[0]["fullres_image"])
def test_is_challenge_response_detects_missing_marker() -> None:
html = "Blocked
"
assert is_challenge_response(status_code=200, body_text=html, expected_marker='id="GBPSearchQuery"') is True
def test_is_challenge_response_detects_incapsula_marker() -> None:
html = ''
assert is_challenge_response(status_code=200, body_text=html, expected_marker=None) is True
def test_is_challenge_response_accepts_expected_marker_even_with_incapsula_script() -> None:
html = f''
assert is_challenge_response(status_code=200, body_text=html, expected_marker=LISTING_MARKER) is False
def test_catalog_hidden_vehicle_json_is_valid() -> None:
parsed = parse_listing_page(_catalog_html())
# smoke-check we can serialize parsed query back to JSON
dumped = json.dumps(parsed.gbp_search_query)
assert "CurrentPage" in dumped
def test_resolve_listing_scope_paths_prefers_explicit_url() -> None:
paths = resolve_listing_scope_paths(
listing_start_url="https://www.iaai.com/Search?url=abc",
brands={"acura", "bmw"},
)
assert paths == ["https://www.iaai.com/Search?url=abc"]
class _FakeResponse:
def __init__(self, status_code: int, text: str) -> None:
self.status_code = status_code
self.text = text
def close(self) -> None:
return
class _FakeSession:
def __init__(self, responses: list[_FakeResponse], *, cookies: list[object] | None = None) -> None:
self._responses = responses
self.calls = 0
self.cookies = cookies or []
def request(self, *args, **kwargs): # noqa: ANN002, ANN003, ANN202
self.calls += 1
return self._responses.pop(0)
def _settings() -> Settings:
return Settings(
database_url="sqlite+pysqlite:///:memory:",
celery_broker_url="redis://localhost:6379/0",
celery_result_backend="redis://localhost:6379/1",
iaai_base_url="https://www.iaai.com",
iaai_listing_start_url="",
sync_runtime_config_file=Path("sync_runtime_config.json"),
iaai_session_cookies="",
iaai_storage_state_path=Path("iaai_storage_state.json"),
iaai_login="",
iaai_password="",
http_timeout=10,
http_retries=0,
http_retry_backoff_ms=1,
fetch_concurrency=1,
db_commit_batch_size=10,
session_keepalive_enabled=True,
session_keepalive_interval_minutes=15,
schema_bootstrap_enabled=False,
advisory_lock_key=1,
error_summary_max_len=2000,
)
def test_hybrid_auth_retries_after_challenge(monkeypatch) -> None: # noqa: ANN001
auth = HybridSessionAuth(_settings())
cookie = type("Cookie", (), {"domain": ".iaai.com"})()
fake_session = _FakeSession(
[
_FakeResponse(403, "forbidden"),
_FakeResponse(200, f"{LISTING_MARKER}"),
],
cookies=[cookie],
)
auth._thread_local.session = fake_session # type: ignore[attr-defined]
auth._bootstrap_cookies_loaded = True # type: ignore[attr-defined]
refreshed = {"count": 0}
def _fake_refresh(*, expected_marker=None, session=None) -> None: # noqa: ANN001
refreshed["count"] += 1
monkeypatch.setattr(auth, "_refresh_session_via_playwright", _fake_refresh)
response = auth.request(
"GET",
"https://www.iaai.com/Vehiclelisting/Cars",
timeout=5,
retries=0,
retry_backoff_ms=1,
expected_marker=LISTING_MARKER,
)
assert response.status_code == 200
assert refreshed["count"] == 1
assert fake_session.calls == 2
def test_hybrid_auth_bootstraps_anonymous_session_when_no_cookies(monkeypatch) -> None: # noqa: ANN001
auth = HybridSessionAuth(_settings())
fake_session = _FakeSession(
[
_FakeResponse(200, f"{LISTING_MARKER}"),
],
cookies=[],
)
auth._thread_local.session = fake_session # type: ignore[attr-defined]
auth._bootstrap_cookies_loaded = True # type: ignore[attr-defined]
refreshed = {"count": 0}
def _fake_refresh(*, expected_marker=None, session=None) -> None: # noqa: ANN001
refreshed["count"] += 1
monkeypatch.setattr(auth, "_refresh_session_via_playwright", _fake_refresh)
response = auth.request(
"GET",
"https://www.iaai.com/Vehiclelisting/Cars",
timeout=5,
retries=0,
retry_backoff_ms=1,
expected_marker=LISTING_MARKER,
)
assert response.status_code == 200
assert refreshed["count"] == 1
assert fake_session.calls == 1
def test_wait_until_non_challenge_tolerates_transient_navigation_errors() -> None:
class _FakePage:
def __init__(self) -> None:
self.content_calls = 0
self.goto_calls = 0
def wait_for_timeout(self, *_args, **_kwargs) -> None: # noqa: ANN002, ANN003
return
def wait_for_load_state(self, *_args, **_kwargs) -> None: # noqa: ANN002, ANN003
return
def content(self) -> str:
self.content_calls += 1
if self.content_calls == 1:
raise RuntimeError(
"Page.content: Unable to retrieve content because the page is navigating and changing the content."
)
return f"{LISTING_MARKER}"
def goto(self, *_args, **_kwargs) -> None: # noqa: ANN002, ANN003
self.goto_calls += 1
page = _FakePage()
HybridSessionAuth._wait_until_non_challenge(
page=page,
target="https://www.iaai.com/Vehiclelisting/Cars",
timeout_ms=5_000,
expected_marker=LISTING_MARKER,
)
assert page.content_calls >= 2
assert page.goto_calls == 0