232 lines
7.5 KiB
Python
232 lines
7.5 KiB
Python
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 = "<html><body><h1>Blocked</h1></body></html>"
|
|
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 = '<html><script>var a="_Incapsula_Resource";</script></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'<html><input id="GBPSearchQuery" value="{{}}" /><script src="/_Incapsula_Resource"></script></html>'
|
|
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, "<html>forbidden</html>"),
|
|
_FakeResponse(200, f"<html>{LISTING_MARKER}</html>"),
|
|
],
|
|
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"<html>{LISTING_MARKER}</html>"),
|
|
],
|
|
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"<html>{LISTING_MARKER}</html>"
|
|
|
|
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
|