lekss361
2bcccdf619
Some checks failed
Deploy Trade-In / deploy (push) Blocked by required conditions
Deploy Trade-In / changes (push) Successful in 6s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 31s
Deploy Trade-In / build-backend (push) Has been cancelled
Co-authored-by: lekss361 <lekss361@gendsgn.local> Co-committed-by: lekss361 <lekss361@gendsgn.local>
828 lines
30 KiB
Python
828 lines
30 KiB
Python
"""Unit tests for YandexRealtyScraper DOM-based SERP parser."""
|
||
|
||
import json
|
||
from datetime import date
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from app.services.scrapers.base import ScrapedLot
|
||
from app.services.scrapers.yandex_realty import (
|
||
DEFAULT_CITY,
|
||
DEFAULT_PRICE_RANGES,
|
||
MAX_PAGES,
|
||
ROOM_PATH,
|
||
YandexRealtyScraper,
|
||
_combo_label,
|
||
_extract_offer_areas_from_state,
|
||
_is_captcha,
|
||
_load_cookies_from_file,
|
||
)
|
||
|
||
# Realistic single-card fixture (trimmed but selector-faithful).
|
||
# Note: selectolax.text(strip=True) concatenates all text nodes without
|
||
# added separators — so floor/price data lives in a single line div with
|
||
# middot (·) separators, matching real Yandex SERP card layout.
|
||
#
|
||
# The address container mirrors the live DOM shape (2026-05):
|
||
# <div class="AddressWithGeoLinks__addressContainer--XXXX">
|
||
# <a href=".../kupit/kvartira/st-...">улица Малышева</a>, 125
|
||
# </div>
|
||
# selectolax sees the trailing text node ", 125" as a sibling of <a>, so
|
||
# text(strip=False) yields "улица Малышева, 125" while text(strip=True)
|
||
# collapses whitespace to "улица Малышева,125".
|
||
SINGLE_CARD_HTML = """
|
||
<html><body>
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/7567094292504417257/">Открыть</a>
|
||
<div class="AddressWithGeoLinks__addressContainer--abc12">
|
||
<a href="/ekaterinburg/kupit/kvartira/st-frezerovshchikov-12345/">улица Фрезеровщиков</a>, 84
|
||
</div>
|
||
<a href="/ekaterinburg/kupit/novostrojka/tatlin-1592987/">ЖК Татлин</a>
|
||
<img src="https://avatars.mds.yandex.net/get-realty/12345/app_snippet_small/test.jpg">
|
||
<img src="https://avatars.mds.yandex.net/get-realty/67890/app_snippet_small/test2.jpg">
|
||
<img src="https://other-domain.example/photo.jpg">
|
||
<div>36,8 м² · 1-комнатная квартира · 8 этаж из 9 · 4 399 000 ₽ · 119 566 ₽ за м²</div>
|
||
<div>торг возможен · опубликовано 17 февраля 2026</div>
|
||
</div>
|
||
</body></html>
|
||
"""
|
||
|
||
STUDIO_CARD_HTML = """
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/9999999/">x</a>
|
||
<div>Студия · 28 м² · 2 этаж из 9 · 3 200 000 ₽</div>
|
||
</div>
|
||
"""
|
||
|
||
NO_PRICE_CARD_HTML = """
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/1111/">x</a>
|
||
<div>40 м² · 1-комнатная · 3 этаж из 5</div>
|
||
</div>
|
||
"""
|
||
|
||
# House number with letter suffix (e.g. "2В") — real-world variant
|
||
LETTER_HOUSE_CARD_HTML = """
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/2222/">x</a>
|
||
<div class="AddressWithGeoLinks__addressContainer--xyz">
|
||
<a href="/ekaterinburg/kupit/kvartira/st-ulica-pekhotintsev-1/">улица Пехотинцев</a>, 2В
|
||
</div>
|
||
<div>40 м² · 1-комнатная · 3 этаж из 5 · 5 000 000 ₽</div>
|
||
</div>
|
||
"""
|
||
|
||
# House number with slash (e.g. "14/3") — corner building
|
||
SLASH_HOUSE_CARD_HTML = """
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/3333/">x</a>
|
||
<div class="AddressWithGeoLinks__addressContainer--xyz">
|
||
<a href="/ekaterinburg/kupit/kvartira/st-akademika-landau-2/">улица Академика Ландау</a>, 14/3
|
||
</div>
|
||
<div>50 м² · 2-комнатная · 4 этаж из 9 · 7 000 000 ₽</div>
|
||
</div>
|
||
"""
|
||
|
||
# District / city prefix before the street link (cards in suburbs or
|
||
# named micro-districts) — must keep the full prefixed address.
|
||
DISTRICT_PREFIX_CARD_HTML = """
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/4444/">x</a>
|
||
<div class="AddressWithGeoLinks__addressContainer--xyz">
|
||
<a href="/berezovskiy/kupit/kvartira/">Берёзовский</a>,
|
||
<a href="/berezovskiy/kupit/kvartira/st-aprospekt-1/">Александровский проспект</a>, 5А
|
||
</div>
|
||
<div>35 м² · 1-комнатная · 2 этаж из 12 · 4 500 000 ₽</div>
|
||
</div>
|
||
"""
|
||
|
||
# Legacy fallback case: no AddressWithGeoLinks container — must still
|
||
# capture street name via the aggregator link (street-only) to avoid NULL.
|
||
NO_ADDR_CONTAINER_CARD_HTML = """
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/5555/">x</a>
|
||
<a href="/ekaterinburg/kupit/kvartira/st-ulica-malysheva-3/">улица Малышева</a>
|
||
<div>30 м² · 1-комнатная · 1 этаж из 5 · 3 500 000 ₽</div>
|
||
</div>
|
||
"""
|
||
|
||
EMPTY_PAGE_HTML = "<html><body><div>Нет объявлений</div></body></html>"
|
||
|
||
|
||
def test_max_pages_default():
|
||
assert MAX_PAGES == 3
|
||
|
||
|
||
def test_default_city():
|
||
assert DEFAULT_CITY == "ekaterinburg"
|
||
|
||
|
||
def test_build_url_first_page():
|
||
s = YandexRealtyScraper()
|
||
assert s._build_url(page=0) == (
|
||
"https://realty.yandex.ru/ekaterinburg/kupit/kvartira/vtorichniy-rynok/"
|
||
)
|
||
|
||
|
||
def test_build_url_paginated():
|
||
s = YandexRealtyScraper()
|
||
assert s._build_url(page=2) == (
|
||
"https://realty.yandex.ru/ekaterinburg/kupit/kvartira/vtorichniy-rynok/?page=2"
|
||
)
|
||
|
||
|
||
def test_build_url_other_city():
|
||
s = YandexRealtyScraper(city="moscow")
|
||
assert s._build_url(page=0) == (
|
||
"https://realty.yandex.ru/moscow/kupit/kvartira/vtorichniy-rynok/"
|
||
)
|
||
|
||
|
||
def test_parse_html_extracts_card():
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(SINGLE_CARD_HTML, page=0)
|
||
assert len(lots) == 1
|
||
lot = lots[0]
|
||
assert lot.source == "yandex"
|
||
assert lot.source_id == "7567094292504417257"
|
||
assert lot.source_url == "https://realty.yandex.ru/offer/7567094292504417257/"
|
||
# Address must include the house number (was "улица Фрезеровщиков" only — buggy)
|
||
assert lot.address == "улица Фрезеровщиков, 84"
|
||
assert any(ch.isdigit() for ch in lot.address)
|
||
assert lot.area_m2 == pytest.approx(36.8)
|
||
assert lot.rooms == 1
|
||
assert lot.floor == 8
|
||
assert lot.total_floors == 9
|
||
assert lot.price_rub == 4_399_000
|
||
assert lot.price_per_m2 == 119_566
|
||
assert lot.bargain_allowed is True
|
||
assert lot.listing_date == date(2026, 2, 17)
|
||
assert lot.house_source == "yandex_realty_nb"
|
||
assert lot.house_ext_id == "1592987"
|
||
assert lot.house_url == (
|
||
"https://realty.yandex.ru/ekaterinburg/kupit/novostrojka/tatlin-1592987/"
|
||
)
|
||
assert lot.listing_segment == "vtorichka"
|
||
assert lot.lat is None and lot.lon is None # SERP has no coords
|
||
# Photos: 2 from yandex CDN, the third (other-domain) skipped
|
||
assert len(lot.photo_urls) == 2
|
||
assert all("avatars.mds.yandex" in u for u in lot.photo_urls)
|
||
# Size upgrade applied: app_snippet_small → main
|
||
assert all("app_snippet_small" not in u for u in lot.photo_urls)
|
||
assert all("main" in u for u in lot.photo_urls)
|
||
|
||
|
||
def test_parse_html_studio_rooms_zero():
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(STUDIO_CARD_HTML, page=0)
|
||
assert len(lots) == 1
|
||
assert lots[0].rooms == 0 # studio convention
|
||
assert lots[0].area_m2 == pytest.approx(28.0)
|
||
assert lots[0].floor == 2 and lots[0].total_floors == 9
|
||
|
||
|
||
def test_parse_html_no_price_skipped():
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(NO_PRICE_CARD_HTML, page=0)
|
||
assert lots == []
|
||
|
||
|
||
def test_parse_html_empty_page():
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(EMPTY_PAGE_HTML, page=0)
|
||
assert lots == []
|
||
|
||
|
||
def test_parse_html_multi_card():
|
||
multi = SINGLE_CARD_HTML + STUDIO_CARD_HTML
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(multi, page=1)
|
||
assert len(lots) == 2
|
||
assert {lot.source_id for lot in lots} == {"7567094292504417257", "9999999"}
|
||
assert all(lot.raw_payload["page"] == 1 for lot in lots)
|
||
|
||
|
||
def test_address_with_letter_house_number():
|
||
"""House numbers with letter suffix (e.g. '2В') must round-trip intact."""
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(LETTER_HOUSE_CARD_HTML, page=0)
|
||
assert len(lots) == 1
|
||
assert lots[0].address == "улица Пехотинцев, 2В"
|
||
|
||
|
||
def test_address_with_slash_house_number():
|
||
"""Corner-building numbers like '14/3' must be preserved verbatim."""
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(SLASH_HOUSE_CARD_HTML, page=0)
|
||
assert len(lots) == 1
|
||
assert lots[0].address == "улица Академика Ландау, 14/3"
|
||
|
||
|
||
def test_address_keeps_district_prefix():
|
||
"""Cards with district / city prefix must keep the full prefixed
|
||
address — the geocoder benefits from disambiguation."""
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(DISTRICT_PREFIX_CARD_HTML, page=0)
|
||
assert len(lots) == 1
|
||
assert lots[0].address == "Берёзовский, Александровский проспект, 5А"
|
||
|
||
|
||
def test_address_fallback_to_street_link():
|
||
"""When the AddressWithGeoLinks container is missing, fall back to the
|
||
street aggregator link (street-only) — better than NULL."""
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(NO_ADDR_CONTAINER_CARD_HTML, page=0)
|
||
assert len(lots) == 1
|
||
assert lots[0].address == "улица Малышева"
|
||
|
||
|
||
# ── PART A: area from embedded JSON state ────────────────────────────────────
|
||
|
||
|
||
def _make_yandex_state_html(offer_id: str, area: float, price_rub: int = 5_000_000) -> str:
|
||
"""Build minimal Yandex SERP HTML with:
|
||
- <script id="initial_state_script"> containing offer area in JSON state
|
||
- A DOM card where the text div has NO area (simulating pre-JS-hydration).
|
||
"""
|
||
state = {
|
||
"offers": [
|
||
{
|
||
"offerId": offer_id,
|
||
"totalArea": area,
|
||
}
|
||
]
|
||
}
|
||
state_json = json.dumps(state, ensure_ascii=False)
|
||
price_fmt = f"{price_rub:,}".replace(",", " ")
|
||
return f"""<html><head>
|
||
<script id="initial_state_script">{state_json}</script>
|
||
</head><body>
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/{offer_id}/">Открыть</a>
|
||
<div class="AddressWithGeoLinks__addressContainer--xyz">
|
||
<a href="/ekaterinburg/kupit/kvartira/st-test-1/">улица Тестовая</a>, 10
|
||
</div>
|
||
<div>{price_fmt} ₽</div>
|
||
</div>
|
||
</body></html>"""
|
||
|
||
|
||
def test_area_from_state_json_no_dom_text():
|
||
"""area_m2 extracted from embedded JSON state when DOM text has no area."""
|
||
html = _make_yandex_state_html(offer_id="1234567890", area=63.5)
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(html, page=0)
|
||
assert len(lots) == 1
|
||
assert lots[0].area_m2 == pytest.approx(63.5)
|
||
|
||
|
||
def test_area_from_state_json_multiple_offers():
|
||
"""State with multiple offers — each card gets its own area from state."""
|
||
state = {
|
||
"offers": [
|
||
{"offerId": "111", "totalArea": 42.0},
|
||
{"offerId": "222", "totalArea": 78.3},
|
||
]
|
||
}
|
||
state_json = json.dumps(state)
|
||
html = f"""<html><head>
|
||
<script id="initial_state_script">{state_json}</script>
|
||
</head><body>
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/111/">x</a>
|
||
<div>3 000 000 ₽</div>
|
||
</div>
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/222/">x</a>
|
||
<div>5 000 000 ₽</div>
|
||
</div>
|
||
</body></html>"""
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(html, page=0)
|
||
assert len(lots) == 2
|
||
areas = {lot.source_id: lot.area_m2 for lot in lots}
|
||
assert areas["111"] == pytest.approx(42.0)
|
||
assert areas["222"] == pytest.approx(78.3)
|
||
|
||
|
||
def test_area_dom_fallback_when_no_state():
|
||
"""When state script absent, area falls back to DOM regex (existing behavior)."""
|
||
# SINGLE_CARD_HTML has area in DOM text — state script absent
|
||
s = YandexRealtyScraper()
|
||
lots = s._parse_html(SINGLE_CARD_HTML, page=0)
|
||
assert len(lots) == 1
|
||
assert lots[0].area_m2 == pytest.approx(36.8)
|
||
|
||
|
||
def test_extract_offer_areas_from_state_nested_path():
|
||
"""State via pageData.offers path (SERP v2 shape)."""
|
||
state = {
|
||
"pageData": {
|
||
"offers": [
|
||
{"offerId": "999", "spaceTotal": 55.0},
|
||
]
|
||
}
|
||
}
|
||
html = '<script id="initial_state_script">' + json.dumps(state) + "</script>"
|
||
areas = _extract_offer_areas_from_state(html)
|
||
assert areas == {"999": 55.0}
|
||
|
||
|
||
def test_extract_offer_areas_area_dict_value():
|
||
"""area field as nested dict {'value': 48.0, 'unit': 'SQM'} (some Yandex responses)."""
|
||
state = {
|
||
"offers": [
|
||
{"offerId": "777", "area": {"value": 48.0, "unit": "SQM"}},
|
||
]
|
||
}
|
||
html = '<script id="initial_state_script">' + json.dumps(state) + "</script>"
|
||
areas = _extract_offer_areas_from_state(html)
|
||
assert areas == {"777": 48.0}
|
||
|
||
|
||
def test_extract_offer_areas_empty_when_no_script():
|
||
"""No state script → empty dict returned, no crash."""
|
||
html = "<html><body><p>no state here</p></body></html>"
|
||
areas = _extract_offer_areas_from_state(html)
|
||
assert areas == {}
|
||
|
||
|
||
# ── PART B: T4 transport + NBSP fix ─────────────────────────────────────────
|
||
|
||
|
||
def test_nbsp_price_fix():
|
||
"""NBSP (\xa0) in price tokens must not break parse_rub → price_rub must be non-None.
|
||
|
||
Yandex uses \xa0 as thousands separator in prices. Without the replace, parse_rub
|
||
returns None because regex can't capture across NBSP — the card is then skipped.
|
||
"""
|
||
# Price with \xa0 thousands separator (e.g. "4\xa0399\xa0000 ₽")
|
||
nbsp_card_html = """
|
||
<html><body>
|
||
<div data-test="OffersSerpItem">
|
||
<a href="/offer/8888888/">x</a>
|
||
<div>36,8\xa0м²\xa0·\xa01-комнатная\xa0·\xa08\xa0этаж\xa0из\xa09\xa0·\xa04\xa0399\xa0000\xa0₽</div>
|
||
</div>
|
||
</body></html>"""
|
||
s = YandexRealtyScraper()
|
||
# _parse_html applies NBSP-fix before parsing — price must be extracted
|
||
lots = s._parse_html(nbsp_card_html, page=0)
|
||
assert len(lots) == 1
|
||
assert lots[0].price_rub is not None
|
||
assert lots[0].price_rub > 0
|
||
|
||
|
||
def test_is_captcha_detects_showcaptcha():
|
||
"""HTML containing /showcaptcha must be flagged as captcha."""
|
||
html = '<html><head><meta http-equiv="refresh" content="0;url=/showcaptcha?..."></head></html>'
|
||
assert _is_captcha(html) is True
|
||
|
||
|
||
def test_is_captcha_detects_smartcaptcha():
|
||
"""HTML with smartcaptcha JS is flagged."""
|
||
html = "<html><body><script src='smartcaptcha.js'></script></body></html>"
|
||
assert _is_captcha(html) is True
|
||
|
||
|
||
def test_is_captcha_clean_page():
|
||
"""Normal SERP page is not flagged as captcha."""
|
||
assert _is_captcha(SINGLE_CARD_HTML) is False
|
||
|
||
|
||
def test_is_captcha_robot_text():
|
||
"""Russian 'докажите, что вы не робот' message is detected."""
|
||
html = "<html><body>Докажите, что вы не робот — введите код.</body></html>"
|
||
assert _is_captcha(html) is True
|
||
|
||
|
||
def test_load_cookies_from_file_missing_path():
|
||
"""None or missing path → empty dict, no exception."""
|
||
assert _load_cookies_from_file(None) == {}
|
||
assert _load_cookies_from_file("/nonexistent/path.json") == {}
|
||
|
||
|
||
def test_load_cookies_from_file_valid(tmp_path):
|
||
"""Valid Netscape-format cookie JSON → dict with name→value."""
|
||
data = [
|
||
{"name": "yandex_login", "value": "user123", "domain": ".yandex.ru"},
|
||
{"name": "Session_id", "value": "abc456", "domain": ".yandex.ru"},
|
||
{"missing_name_key": "ignored"},
|
||
]
|
||
f = tmp_path / "cookies.json"
|
||
f.write_text(json.dumps(data), encoding="utf-8")
|
||
result = _load_cookies_from_file(str(f))
|
||
assert result == {"yandex_login": "user123", "Session_id": "abc456"}
|
||
|
||
|
||
# ── PART C: T5 combos — _build_url with rooms + price filters ───────────────
|
||
|
||
|
||
def test_build_url_with_rooms():
|
||
"""Room slug mapped correctly to Yandex path segment."""
|
||
s = YandexRealtyScraper()
|
||
url = s._build_url(page=0, rooms="1")
|
||
assert "/odnokomnatnaya/vtorichniy-rynok/" in url
|
||
assert "page" not in url
|
||
|
||
|
||
def test_build_url_with_rooms_2():
|
||
s = YandexRealtyScraper()
|
||
url = s._build_url(page=0, rooms="2")
|
||
assert "/dvuhkomnatnaya/vtorichniy-rynok/" in url
|
||
|
||
|
||
def test_build_url_with_rooms_studio():
|
||
s = YandexRealtyScraper()
|
||
url = s._build_url(page=0, rooms="studio")
|
||
assert "/studiya/vtorichniy-rynok/" in url
|
||
|
||
|
||
def test_build_url_with_price_range():
|
||
"""Price range adds priceMin/priceMax + newFlat=NO_DEAL."""
|
||
s = YandexRealtyScraper()
|
||
url = s._build_url(page=0, price_min=5_000_000, price_max=7_000_000)
|
||
assert "priceMin=5000000" in url
|
||
assert "priceMax=7000000" in url
|
||
assert "newFlat=NO_DEAL" in url
|
||
# No rooms → base citywide path
|
||
assert "/vtorichniy-rynok/" in url
|
||
assert "odnokomnatnaya" not in url
|
||
|
||
|
||
def test_build_url_rooms_and_price_and_page():
|
||
"""All parameters combined: rooms + price + paginated."""
|
||
s = YandexRealtyScraper()
|
||
url = s._build_url(page=3, rooms="3", price_min=10_000_000, price_max=15_000_000)
|
||
assert "/trehkomnatnaya/vtorichniy-rynok/" in url
|
||
assert "priceMin=10000000" in url
|
||
assert "priceMax=15000000" in url
|
||
assert "newFlat=NO_DEAL" in url
|
||
assert "page=3" in url
|
||
|
||
|
||
def test_build_url_open_ended_price_min_only():
|
||
"""Only priceMin set → no priceMax param but newFlat present."""
|
||
s = YandexRealtyScraper()
|
||
url = s._build_url(price_min=25_000_000)
|
||
assert "priceMin=25000000" in url
|
||
assert "priceMax" not in url
|
||
assert "newFlat=NO_DEAL" in url
|
||
|
||
|
||
def test_build_url_open_ended_price_max_only():
|
||
"""Only priceMax → no priceMin param."""
|
||
s = YandexRealtyScraper()
|
||
url = s._build_url(price_max=5_000_000)
|
||
assert "priceMax=5000000" in url
|
||
assert "priceMin" not in url
|
||
assert "newFlat=NO_DEAL" in url
|
||
|
||
|
||
def test_build_url_unknown_rooms_falls_back_to_citywide():
|
||
"""Unknown/invalid rooms key falls back to citywide (no room path segment)."""
|
||
s = YandexRealtyScraper()
|
||
url = s._build_url(rooms="99-room-penthouse")
|
||
assert "/kupit/kvartira/vtorichniy-rynok/" in url
|
||
assert "studiya" not in url
|
||
assert "odnokomnatnaya" not in url
|
||
|
||
|
||
def test_combos_produce_different_urls():
|
||
"""T5: rooms × price-ranges produce distinct URLs (no URL collisions)."""
|
||
s = YandexRealtyScraper()
|
||
urls = set()
|
||
rooms_list = list(ROOM_PATH.keys())
|
||
for rooms in rooms_list:
|
||
for lo, hi in DEFAULT_PRICE_RANGES:
|
||
urls.add(s._build_url(page=0, rooms=rooms, price_min=lo, price_max=hi))
|
||
expected = len(rooms_list) * len(DEFAULT_PRICE_RANGES)
|
||
assert len(urls) == expected, f"URL collisions: {expected} combos but {len(urls)} unique URLs"
|
||
|
||
|
||
def test_combo_label_all():
|
||
assert _combo_label(None, None, None) == "all-rooms/any-price"
|
||
|
||
|
||
def test_combo_label_room_and_price():
|
||
assert _combo_label("1", 5_000_000, 7_000_000) == "1/5M-7M"
|
||
|
||
|
||
def test_combo_label_open_ended():
|
||
assert _combo_label("4+", 25_000_000, None) == "4+/25M-inf"
|
||
assert _combo_label("studio", None, 5_000_000) == "studio/0-5M"
|
||
|
||
|
||
# ── PART D: T4 transport — async context manager + _http_get mock ───────────
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_aenter_creates_cffi_session():
|
||
"""__aenter__ creates a _CurlCffiSession (Chrome120 impersonation)."""
|
||
mock_settings = MagicMock()
|
||
mock_settings.yandex_cookies_file = None
|
||
|
||
with patch("app.services.scrapers.yandex_realty._CurlCffiSession") as mock_cls:
|
||
mock_session = AsyncMock()
|
||
mock_cls.return_value = mock_session
|
||
with patch("app.services.scrapers.yandex_realty._load_cookies_from_file", return_value={}):
|
||
# Patch settings import inside __aenter__ without triggering Settings()
|
||
with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}):
|
||
s = YandexRealtyScraper()
|
||
result = await s.__aenter__()
|
||
assert result is s
|
||
mock_cls.assert_called_once()
|
||
call_kwargs = mock_cls.call_args
|
||
assert call_kwargs.kwargs.get("impersonate") == "chrome120"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_aexit_closes_session():
|
||
"""__aexit__ closes the curl_cffi session and sets it to None."""
|
||
s = YandexRealtyScraper()
|
||
mock_session = AsyncMock()
|
||
s._cffi_session = mock_session
|
||
await s.__aexit__(None, None, None)
|
||
mock_session.close.assert_awaited_once()
|
||
assert s._cffi_session is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_http_get_raises_when_no_context():
|
||
"""_http_get must raise RuntimeError when called outside async context manager."""
|
||
s = YandexRealtyScraper()
|
||
with pytest.raises(RuntimeError, match="async context manager"):
|
||
await s._http_get("https://realty.yandex.ru/test/")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_around_returns_empty_on_captcha():
|
||
"""fetch_around returns [] and logs warning when captcha detected."""
|
||
s = YandexRealtyScraper()
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 200
|
||
mock_resp.text = (
|
||
"<html><body>Докажите, что вы не робот. "
|
||
'<a href="/showcaptcha">пройдите проверку</a></body></html>'
|
||
)
|
||
s._cffi_session = AsyncMock()
|
||
s._cffi_session.get = AsyncMock(return_value=mock_resp)
|
||
|
||
with patch.object(s, "sleep_between_requests", new_callable=AsyncMock):
|
||
lots = await s.fetch_around(lat=0.0, lon=0.0)
|
||
assert lots == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_around_multi_room_dedup():
|
||
"""fetch_around_multi_room deduplicates offers seen in multiple combos."""
|
||
s = YandexRealtyScraper()
|
||
|
||
# Two combos both return the same offer_id "1234" plus unique ids
|
||
combo_results: list[list[ScrapedLot]] = []
|
||
for i in range(2):
|
||
lot_shared = MagicMock(spec=ScrapedLot)
|
||
lot_shared.source_id = "shared_offer_1234"
|
||
lot_shared.source_url = "https://realty.yandex.ru/offer/shared_offer_1234/"
|
||
|
||
lot_unique = MagicMock(spec=ScrapedLot)
|
||
lot_unique.source_id = f"unique_offer_{i}"
|
||
lot_unique.source_url = f"https://realty.yandex.ru/offer/unique_offer_{i}/"
|
||
|
||
combo_results.append([lot_shared, lot_unique])
|
||
|
||
call_count = 0
|
||
|
||
async def fake_fetch_around(*args, **kwargs):
|
||
nonlocal call_count
|
||
idx = call_count
|
||
call_count += 1
|
||
if idx < len(combo_results):
|
||
return combo_results[idx]
|
||
return []
|
||
|
||
with patch.object(s, "fetch_around", side_effect=fake_fetch_around):
|
||
lots = await s.fetch_around_multi_room(
|
||
lat=0.0,
|
||
lon=0.0,
|
||
rooms_list=["1"],
|
||
price_ranges=[(None, 5_000_000), (5_000_000, 7_000_000)],
|
||
max_pages=1,
|
||
)
|
||
|
||
# shared_offer_1234 appears once; 2 unique offers
|
||
ids = {lot.source_id for lot in lots}
|
||
assert "shared_offer_1234" in ids
|
||
assert len(ids) == 3 # shared + 2 unique
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_around_multi_room_legacy_no_combos():
|
||
"""Legacy mode (no rooms_list/price_ranges) uses single citywide sweep."""
|
||
s = YandexRealtyScraper()
|
||
calls: list[dict] = []
|
||
|
||
async def fake_fetch_around(
|
||
lat, lon, radius_m=1000, page=0, rooms=None, price_min=None, price_max=None
|
||
):
|
||
calls.append({"page": page, "rooms": rooms, "price_min": price_min})
|
||
return [] # empty → stops pagination
|
||
|
||
with patch.object(s, "fetch_around", side_effect=fake_fetch_around):
|
||
await s.fetch_around_multi_room(lat=0.0, lon=0.0, max_pages=3)
|
||
|
||
# Should call with rooms=None, price_min=None (citywide)
|
||
assert all(c["rooms"] is None for c in calls)
|
||
assert all(c["price_min"] is None for c in calls)
|
||
|
||
|
||
# ── PART E: exhaustive load — fetch_all_secondary + _walk_price_range ────────
|
||
|
||
|
||
def _make_lots(n: int, prefix: str = "lot") -> list[MagicMock]:
|
||
"""Create n mock ScrapedLot objects with unique source_id."""
|
||
lots = []
|
||
for i in range(n):
|
||
lot = MagicMock(spec=ScrapedLot)
|
||
lot.source_id = f"{prefix}_{i}"
|
||
lot.source_url = f"https://realty.yandex.ru/offer/{prefix}_{i}/"
|
||
lots.append(lot)
|
||
return lots
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_walk_price_range_splits_when_total_exceeds_cap():
|
||
"""When totalItems > price_cap_per_bucket and bracket >= MIN_BRACKET,
|
||
_walk_price_range recurses and subdivides — each leaf has ≤ cap items."""
|
||
s = YandexRealtyScraper()
|
||
s._cffi_session = AsyncMock()
|
||
|
||
call_ranges: list[tuple[int | None, int]] = []
|
||
|
||
async def fake_fetch_page_html(
|
||
rooms: str | None, page: int, price_min: int | None, price_max: int
|
||
) -> str:
|
||
call_ranges.append((price_min, price_max))
|
||
# Return captcha-free HTML with 0 cards (parsing gives no lots, which is fine —
|
||
# we only care that the range is split, not about actual lot count in this test)
|
||
return "<html><body>empty</body></html>"
|
||
|
||
# total_count returns 600 for the first (wide) probe, then 0 for sub-ranges
|
||
# so recursion terminates after the first split.
|
||
call_counts: list[int] = [0]
|
||
|
||
def fake_extract_total(html: str) -> int | None:
|
||
call_counts[0] += 1
|
||
# First call (wide range) returns 600 > cap=500, triggering a split
|
||
# Subsequent calls (sub-ranges) return 0 → no further recursion
|
||
if call_counts[0] == 1:
|
||
return 600
|
||
return 0
|
||
|
||
seen: dict = {}
|
||
with (
|
||
patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
|
||
patch.object(s, "_extract_total_count", side_effect=fake_extract_total),
|
||
patch.object(s, "_is_captcha_html", return_value=False, create=True),
|
||
):
|
||
# Patch asyncio.sleep to be instant
|
||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||
await s._walk_price_range(
|
||
rooms="1",
|
||
lo=0,
|
||
hi=10_000_000,
|
||
seen=seen,
|
||
price_cap_per_bucket=500,
|
||
max_pages_per_bucket=100,
|
||
concurrency=2,
|
||
)
|
||
|
||
# Should have probed at least the initial range + 2 sub-ranges
|
||
assert len(call_ranges) >= 3, f"Expected >= 3 probes (wide + 2 sub-ranges), got {call_ranges}"
|
||
# Initial probe covers the full range
|
||
assert call_ranges[0] == (None, 10_000_000) # lo=0 → _lo_param=None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_walk_price_range_fires_on_bucket_per_leaf():
|
||
"""on_bucket callback is called once per leaf bucket with the bucket's lots."""
|
||
s = YandexRealtyScraper()
|
||
s._cffi_session = AsyncMock()
|
||
|
||
bucket_calls: list[list] = []
|
||
|
||
def fake_on_bucket(lots: list) -> None:
|
||
bucket_calls.append(list(lots))
|
||
|
||
# total = 10 (< cap=500) → paginate leaf immediately, 1 page (page=0)
|
||
async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str:
|
||
# Return HTML with 1 parseable card for page=0, empty for others
|
||
if page == 0:
|
||
return SINGLE_CARD_HTML
|
||
return EMPTY_PAGE_HTML
|
||
|
||
seen: dict = {}
|
||
with (
|
||
patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
|
||
patch.object(s, "_extract_total_count", return_value=10),
|
||
patch("asyncio.sleep", new_callable=AsyncMock),
|
||
):
|
||
await s._walk_price_range(
|
||
rooms="1",
|
||
lo=0,
|
||
hi=20_000_000,
|
||
seen=seen,
|
||
price_cap_per_bucket=500,
|
||
max_pages_per_bucket=5,
|
||
concurrency=2,
|
||
on_bucket=fake_on_bucket,
|
||
)
|
||
|
||
# on_bucket must have been called at least once
|
||
assert len(bucket_calls) >= 1
|
||
# All calls contain ScrapedLot-like objects (the actual card from SINGLE_CARD_HTML)
|
||
for call_lots in bucket_calls:
|
||
assert len(call_lots) >= 0 # may be 0 if card parse yields empty — still fires
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_walk_price_range_fallback_when_total_none():
|
||
"""When totalItems=None (state unavailable), falls back to paginate-until-empty
|
||
and still fires on_bucket with whatever was collected."""
|
||
s = YandexRealtyScraper()
|
||
s._cffi_session = AsyncMock()
|
||
|
||
bucket_calls: list[list] = []
|
||
|
||
def fake_on_bucket(lots: list) -> None:
|
||
bucket_calls.append(list(lots))
|
||
|
||
page_htmls = {
|
||
0: SINGLE_CARD_HTML, # page=0: 1 card
|
||
1: EMPTY_PAGE_HTML, # page=1: empty → stop
|
||
}
|
||
|
||
async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str | None:
|
||
return page_htmls.get(page, EMPTY_PAGE_HTML)
|
||
|
||
seen: dict = {}
|
||
with (
|
||
patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
|
||
patch.object(s, "_extract_total_count", return_value=None),
|
||
patch("asyncio.sleep", new_callable=AsyncMock),
|
||
):
|
||
await s._walk_price_range(
|
||
rooms="2",
|
||
lo=0,
|
||
hi=15_000_000,
|
||
seen=seen,
|
||
price_cap_per_bucket=500,
|
||
max_pages_per_bucket=10,
|
||
concurrency=2,
|
||
on_bucket=fake_on_bucket,
|
||
)
|
||
|
||
# Fallback path must fire on_bucket with the collected lots
|
||
assert len(bucket_calls) >= 1, "on_bucket must be called in fallback path"
|
||
# The card from SINGLE_CARD_HTML should have been collected
|
||
all_lots = [lot for call in bucket_calls for lot in call]
|
||
assert len(all_lots) >= 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_all_secondary_deduplicates_across_rooms():
|
||
"""fetch_all_secondary deduplicates offers that appear in multiple room buckets."""
|
||
s = YandexRealtyScraper()
|
||
|
||
shared_lot = MagicMock(spec=ScrapedLot)
|
||
shared_lot.source_id = "shared_999"
|
||
shared_lot.source_url = "https://realty.yandex.ru/offer/shared_999/"
|
||
|
||
unique_lot_1 = MagicMock(spec=ScrapedLot)
|
||
unique_lot_1.source_id = "unique_001"
|
||
unique_lot_1.source_url = "https://realty.yandex.ru/offer/unique_001/"
|
||
|
||
unique_lot_2 = MagicMock(spec=ScrapedLot)
|
||
unique_lot_2.source_id = "unique_002"
|
||
unique_lot_2.source_url = "https://realty.yandex.ru/offer/unique_002/"
|
||
|
||
call_count = 0
|
||
|
||
async def fake_walk(*, rooms, lo, hi, seen, **kwargs): # type: ignore[override]
|
||
nonlocal call_count
|
||
# Both room buckets yield the shared lot + one unique each
|
||
seen["shared_999"] = shared_lot
|
||
if call_count == 0:
|
||
seen["unique_001"] = unique_lot_1
|
||
else:
|
||
seen["unique_002"] = unique_lot_2
|
||
call_count += 1
|
||
|
||
with patch.object(s, "_walk_price_range", side_effect=fake_walk):
|
||
result = await s.fetch_all_secondary(rooms_buckets=["1", "2"])
|
||
|
||
ids = {lot.source_id for lot in result}
|
||
assert "shared_999" in ids
|
||
assert "unique_001" in ids
|
||
assert "unique_002" in ids
|
||
assert len(ids) == 3 # shared deduplicated — only 3 unique
|