186 lines
7.9 KiB
Python
186 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
import requests
|
|
|
|
from mobilede_scraper.mobile_de.client import (
|
|
MOBILEDE_HTTP_MAX_RETRIES,
|
|
MobileDeClient,
|
|
MobileDeDetailError,
|
|
classify_detail_error,
|
|
)
|
|
from mobilede_scraper.mobile_de.flight import extract_detail_listing, extract_search_card_enrichment
|
|
from mobilede_scraper.mobile_de.mapper import MobileDeMapper
|
|
|
|
|
|
class TestMobileDeDetailTransports(unittest.TestCase):
|
|
def test_extract_search_ssr_card_enrichment_keeps_listing_id_relation(self) -> None:
|
|
html = """
|
|
<div data-testid="base-result-listing-1">
|
|
<a href="/ru/transport/details.html?id=454495868&ref=srp">
|
|
<div data-testid="base-result-listing-1-title"><h2>
|
|
<span data-testid="listing-title-card-view" title="BMW M135">BMW M135</span>
|
|
<span title="i xDrive * Leder">i xDrive * Leder</span>
|
|
</h2></div>
|
|
</a>
|
|
</div>
|
|
<div data-testid="base-result-listing-2">
|
|
<a href="/ru/transport/details.html?id=454495869">
|
|
<div data-testid="base-result-listing-2-title"><h2>
|
|
<span data-testid="listing-title-card-view" title="BMW 320">BMW 320</span>
|
|
<span title="sDrive">sDrive</span>
|
|
</h2></div>
|
|
</a>
|
|
</div>
|
|
"""
|
|
|
|
enrichment = extract_search_card_enrichment(html)
|
|
|
|
self.assertEqual(enrichment["454495868"]["searchSsrTitle"], "BMW M135")
|
|
self.assertEqual(enrichment["454495868"]["searchSsrSubtitle"], "i xDrive * Leder")
|
|
self.assertEqual(enrichment["454495869"]["searchSsrDriveText"], "BMW 320 sDrive")
|
|
|
|
def test_fetch_search_page_retry_override_is_request_local(self) -> None:
|
|
client = MobileDeClient()
|
|
original_retries = MOBILEDE_HTTP_MAX_RETRIES
|
|
response = Mock(status_code=200, text='<script id="__NEXT_DATA__" type="application/json">{}</script>')
|
|
response.raise_for_status.return_value = None
|
|
client.session.get = Mock(return_value=response)
|
|
|
|
client.fetch_search_page(search_url="https://www.mobile.de/ru/test", max_retries=0)
|
|
|
|
self.assertEqual(MOBILEDE_HTTP_MAX_RETRIES, original_retries)
|
|
client.session.get.assert_called_once()
|
|
|
|
def test_fetch_html_retries_transient_transport_error(self) -> None:
|
|
client = MobileDeClient()
|
|
response = Mock(status_code=200, text="ok")
|
|
response.raise_for_status.return_value = None
|
|
client.session.get = Mock(
|
|
side_effect=[requests.exceptions.SSLError("temporary EOF"), response]
|
|
)
|
|
|
|
with patch.object(client, "_compute_backoff", return_value=0):
|
|
result = client.fetch_html("https://www.mobile.de/test", max_retries=2)
|
|
|
|
self.assertEqual(result, "ok")
|
|
self.assertEqual(client.session.get.call_count, 2)
|
|
|
|
def test_fetch_html_does_not_retry_non_retryable_http_error(self) -> None:
|
|
client = MobileDeClient()
|
|
response = Mock(status_code=404)
|
|
response.raise_for_status.side_effect = requests.HTTPError("not found", response=response)
|
|
client.session.get = Mock(return_value=response)
|
|
|
|
with self.assertRaises(requests.HTTPError):
|
|
client.fetch_html("https://www.mobile.de/missing", max_retries=2)
|
|
|
|
client.session.get.assert_called_once()
|
|
|
|
def test_extracts_and_normalizes_legacy_initial_state(self) -> None:
|
|
state = {
|
|
"search": {
|
|
"vip": {
|
|
"ads": {
|
|
"461936419": {
|
|
"data": {
|
|
"ad": {
|
|
"shortTitle": "BMW 118",
|
|
"subTitle": "i LiveCockpit",
|
|
"make": "BMW",
|
|
"model": "118",
|
|
"price": {"grossAmount": 21908, "grossCurrency": "EUR"},
|
|
"contactInfo": {"country": "DE"},
|
|
"attributes": [
|
|
{"tag": "mileage", "value": "18.285 km"},
|
|
{"tag": "firstRegistration", "value": "07/2022"},
|
|
],
|
|
"galleryImages": [
|
|
{
|
|
"src": (
|
|
"https://img.classistatic.de/api/v1/mo-prod/images/aa/"
|
|
"a1111111-1111-4111-8111-111111111111?rule=mo-360"
|
|
)
|
|
}
|
|
],
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
html = f"<script>window.__INITIAL_STATE__ = {json.dumps(state)};</script>"
|
|
|
|
detail = extract_detail_listing(html)
|
|
record = MobileDeMapper().detail_to_car_record("461936419", detail)
|
|
|
|
self.assertEqual(detail["make"], {"localized": "BMW"})
|
|
self.assertEqual(detail["model"], {"localized": "118"})
|
|
self.assertEqual(detail["contact"], {"country": "DE"})
|
|
self.assertEqual(record.price, 21908)
|
|
self.assertEqual(record.currency, "EUR")
|
|
self.assertEqual(record.year, 2022)
|
|
self.assertEqual(record.mileage, 18285)
|
|
self.assertEqual(len(record.images), 1)
|
|
self.assertEqual(
|
|
record.origin_url,
|
|
MobileDeClient.build_transport_detail_url("461936419"),
|
|
)
|
|
|
|
def test_search_listing_uses_russian_transport_detail_url(self) -> None:
|
|
listing = MobileDeClient._map_listing({"id": "461936419"})
|
|
|
|
self.assertEqual(
|
|
listing.url,
|
|
MobileDeClient.build_transport_detail_url("461936419"),
|
|
)
|
|
|
|
def test_legacy_transport_is_explicitly_opt_in(self) -> None:
|
|
client = MobileDeClient()
|
|
html = (
|
|
'<script>window.__INITIAL_STATE__ = '
|
|
'{"search":{"vip":{"ads":{"461936419":{"data":{"ad":{"make":"BMW","model":"118"}}}}}}};'
|
|
'</script>'
|
|
)
|
|
|
|
with (
|
|
patch("mobilede_scraper.mobile_de.client.MOBILEDE_DETAIL_TRANSPORT", "legacy"),
|
|
patch.object(client, "fetch_html", return_value=html) as fetch_html,
|
|
):
|
|
client.fetch_detail("461936419")
|
|
|
|
self.assertEqual(fetch_html.call_args.args[0], client.build_legacy_detail_url("461936419"))
|
|
|
|
def test_empty_detail_payload_is_rejected_before_mapping(self) -> None:
|
|
client = MobileDeClient()
|
|
with (
|
|
patch.object(client, "fetch_html", return_value="<html>antibot or parser drift</html>"),
|
|
self.assertRaisesRegex(ValueError, "detail payload is empty"),
|
|
):
|
|
client.fetch_detail("461936419")
|
|
|
|
def test_detail_http_errors_are_classified_by_outcome(self) -> None:
|
|
for status_code, expected_kind in ((404, "unavailable"), (410, "unavailable"), (403, "blocked"), (429, "throttled"), (503, "transient")):
|
|
response = Mock(status_code=status_code)
|
|
exc = requests.HTTPError(f"status {status_code}", response=response)
|
|
self.assertEqual(classify_detail_error(exc), (expected_kind, status_code))
|
|
|
|
def test_fetch_detail_preserves_typed_http_outcome(self) -> None:
|
|
client = MobileDeClient()
|
|
response = Mock(status_code=404)
|
|
with (
|
|
patch.object(client, "fetch_html", side_effect=requests.HTTPError("not found", response=response)),
|
|
self.assertRaises(MobileDeDetailError) as raised,
|
|
):
|
|
client.fetch_detail("missing")
|
|
|
|
self.assertEqual(raised.exception.error_kind, "unavailable")
|
|
self.assertEqual(raised.exception.status_code, 404)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |