"""Golden-parity: `scraper_kit.providers.cian.*` парсинг ≡ `app.services.scrapers.cian*`.
Strangler-инвариант (#2133 cian): новая scraper_kit-копия cian-провайдера должна давать
БАЙТ-ИДЕНТИЧНЫЙ результат парсинга старому боевому коду на одинаковом входе. Развязка
(settings→ScraperConfig, get_scraper_delay→delay_provider, cian_session→локальная копия)
НЕ меняет распарсенные данные.
Гоняем ОБА модуля на одних фикстурах и сравниваем результат:
- SERP `_parse_serp_html` (Redux state → list[ScrapedLot]) на синтетическом _cianConfig
push-HTML (вторичка + новостройка + студия), сравнение model_dump();
- SERP `_offer_to_lot` пооферно + `_format_address` (pure, восстановление дома ЖК);
- `_extract_total_offers` (totalOffers из state);
- `extract_state` — app-версия ≡ kit-версия на одном HTML (общий парсер state);
- detail `fetch_detail` (defaultState + bti sister → DetailEnrichment, dataclasses.asdict);
- valuation `_parse_valuation_state` (estimation/chart/houseInfo → CianValuationResult);
- newbuilding pure-экстракторы (_extract_chart / _extract_reliability_checks /
_extract_nested_offers / _extract_zhk_url_from_serp).
Идентичность результата = kit-парсер верно скопирован (развязка не задела логику).
"""
from __future__ import annotations
import asyncio
import dataclasses
import json
import os
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
# Старый app.services.scrapers.cian* импортирует app.core.config.settings=Settings(),
# которому нужен DATABASE_URL. Офлайн-парсинг БД не трогает — фиктивный DSN достаточен.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
# НОВЫЙ (scraper_kit) cian-провайдер.
from scraper_kit.cian_state_parser import extract_state as new_extract_state
from scraper_kit.providers.cian.detail import fetch_detail as new_fetch_detail
from scraper_kit.providers.cian.newbuilding import _extract_chart as new_extract_chart
from scraper_kit.providers.cian.newbuilding import (
_extract_nested_offers as new_extract_nested_offers,
)
from scraper_kit.providers.cian.newbuilding import (
_extract_reliability_checks as new_extract_reliability,
)
from scraper_kit.providers.cian.newbuilding import (
_extract_zhk_url_from_serp as new_extract_zhk_url,
)
from scraper_kit.providers.cian.serp import CianScraper as NewCianScraper
from scraper_kit.providers.cian.serp import _format_address as new_format_address
from scraper_kit.providers.cian.valuation import (
_parse_valuation_state as new_parse_valuation,
)
from scraper_kit.providers.cian.valuation import (
estimate_via_cian_valuation as new_estimate_via_cian_valuation,
)
# #2335 (Group E2): estimate_via_cian_valuation full-pipeline parity + mandatory-config
# regression guard. RealScraperConfig — тот же read-only адаптер над settings, что и
# #2306's cian_price_history.py (см. app/services/scraper_adapters.py).
from app.services.scraper_adapters import RealScraperConfig
# СТАРЫЙ (app.services.scrapers) cian-провайдер.
from app.services.scrapers.cian import CianScraper as OldCianScraper
from app.services.scrapers.cian import _format_address as old_format_address
from app.services.scrapers.cian_detail import fetch_detail as old_fetch_detail
from app.services.scrapers.cian_newbuilding import _extract_chart as old_extract_chart
from app.services.scrapers.cian_newbuilding import (
_extract_nested_offers as old_extract_nested_offers,
)
from app.services.scrapers.cian_newbuilding import (
_extract_reliability_checks as old_extract_reliability,
)
from app.services.scrapers.cian_newbuilding import (
_extract_zhk_url_from_serp as old_extract_zhk_url,
)
from app.services.scrapers.cian_state_parser import extract_state as old_extract_state
from app.services.scrapers.cian_valuation import (
_parse_valuation_state as old_parse_valuation,
)
from app.services.scrapers.cian_valuation import (
estimate_via_cian_valuation as old_estimate_via_cian_valuation,
)
from tests.support.parity import assert_parity
# ── Fixtures: SERP offers → _cianConfig push HTML ────────────────────────────
_CONFIG = SimpleNamespace(glitchtip_dsn=None)
def _serp_html(offers: list[dict[str, Any]]) -> str:
"""Обернуть offers в валидный _cianConfig['frontend-serp'].push() HTML."""
state = {"results": {"totalOffers": len(offers), "offers": offers}}
state_json = json.dumps(state, ensure_ascii=False)
return (
""
)
def _vtorichka_offer() -> dict[str, Any]:
return {
"cianId": 328739308,
"id": 328739308,
"fullUrl": "https://ekb.cian.ru/sale/flat/328739308/",
"offerType": "flat",
"flatType": "rooms",
"category": "flatSale",
"isApartments": False,
"roomsCount": 2,
"totalArea": "52.0",
"livingArea": "35.0",
"kitchenArea": "11.5",
"balconiesCount": 1,
"loggiasCount": 0,
"bedroomsCount": None,
"hasFurniture": False,
"isSoldFurnished": False,
"decoration": "euro",
"floorNumber": 7,
"building": {
"floorsCount": 16,
"materialType": "monolithBrick",
"buildYear": 2010,
"parking": {"type": "underground"},
"totalArea": "3400",
},
"cadastralNumber": "66:41:0204016:1234",
"buildingCadastralNumber": "66:41:0204016:100",
"bargainTerms": {
"priceRur": 6500000,
"price": 6500000,
"saleType": "free",
"mortgageAllowed": True,
"bargainAllowed": True,
},
"geo": {
"coordinates": {"lat": 56.8385, "lng": 60.5940},
"address": [
{"type": "location", "fullName": "Свердловская область"},
{"type": "raion", "fullName": "Кировский"},
{"type": "street", "fullName": "улица Постовского"},
{"type": "house", "fullName": "17А"},
{"type": "metro", "fullName": "Геологическая"},
],
"undergrounds": [
{
"name": "Геологическая",
"time": 10,
"transportType": "walk",
"lineId": 10,
"lineColor": "FF0000",
"isDefault": True,
}
],
},
"newbuilding": {"id": 0, "name": ""},
"phones": [{"number": "9012345678", "countryCode": "7"}],
"userId": 13332612,
"publishedUserId": 13332612,
"isByHomeowner": True,
"isPro": False,
"isRosreestrChecked": True,
"description": "Продаётся отличная квартира с евроремонтом",
"descriptionMinhash": "abc123def456abc1",
"addedTimestamp": 1716451200,
"photos": [{"fullUrl": "https://images.cdn-cian.ru/images/328739308-1.jpg"}],
}
def _newbuilding_offer() -> dict[str, Any]:
"""Новостройка без части house в geo.address[] — жмёт _recover_newbuilding_house."""
return {
"cianId": 411112222,
"id": 411112222,
"fullUrl": "https://ekb.cian.ru/sale/flat/411112222/",
"offerType": "flat",
"flatType": "rooms",
"isApartments": False,
"roomsCount": 1,
"totalArea": "38.0",
"floorNumber": 12,
"building": {
"floorsCount": 25,
"materialType": "monolith",
"buildYear": None,
"deadline": {"year": 2026},
},
"buildingCadastralNumber": "66:41:0204016:11396",
"bargainTerms": {"priceRur": 5200000, "price": 5200000, "mortgageAllowed": True},
"geo": {
"coordinates": {"lat": 56.8000, "lng": 60.6000},
"address": [
{"type": "location", "fullName": "Свердловская область"},
{"type": "street", "fullName": "улица Новая"},
],
},
"newbuilding": {"id": 48853, "name": "ЖК Парковый квартал 29А"},
"isFromDeveloper": True,
"photos": [],
}
def _studio_offer() -> dict[str, Any]:
return {
"cianId": 500011111,
"id": 500011111,
"offerType": "flat",
"flatType": "studio",
"roomsCount": None,
"totalArea": "24.5",
"floorNumber": 3,
"building": {"floorsCount": 9, "materialType": "panel", "buildYear": 1980},
"bargainTerms": {"priceRur": 3100000, "price": 3100000},
"geo": {
"coordinates": {"lat": 56.81, "lng": 60.61},
"address": [
{"type": "street", "fullName": "улица Ленина"},
{"type": "house", "fullName": "5"},
],
},
"newbuilding": {"id": 0},
"photos": [],
}
_OFFERS = [_vtorichka_offer(), _newbuilding_offer(), _studio_offer()]
def _make_old() -> OldCianScraper:
with patch("app.services.scrapers.cian.get_scraper_delay", return_value=5.0):
return OldCianScraper()
def _make_new() -> NewCianScraper:
return NewCianScraper(config=_CONFIG)
def _dump(lots: list[Any]) -> list[dict[str, Any]]:
return [lot.model_dump() for lot in lots]
# ── SERP _parse_serp_html parity ─────────────────────────────────────────────
def test_serp_parse_serp_html_parity() -> None:
"""Redux offers → list[ScrapedLot] идентичны (вторичка + новостройка + студия)."""
html = _serp_html(_OFFERS)
old_lots = _make_old()._parse_serp_html(html)
new_lots = _make_new()._parse_serp_html(html)
assert len(old_lots) == len(new_lots) == 3
assert _dump(old_lots) == _dump(new_lots)
def test_serp_offer_to_lot_parity_per_offer() -> None:
"""Пооферный _offer_to_lot: model_dump идентичен для каждого offer."""
old, new = _make_old(), _make_new()
for offer in _OFFERS:
old_lot = old._offer_to_lot(offer)
new_lot = new._offer_to_lot(offer)
assert (old_lot is None) == (new_lot is None)
if old_lot is not None and new_lot is not None:
assert old_lot.model_dump() == new_lot.model_dump()
def test_serp_extract_total_offers_parity() -> None:
html = _serp_html(_OFFERS)
assert _make_old()._extract_total_offers(html) == _make_new()._extract_total_offers(html)
def test_format_address_parity() -> None:
"""_format_address (pure) — вкл. восстановление дома для ЖК без части house."""
building = {"houseNumber": None}
nb = {"name": "ЖК Парковый квартал 29А к1"}
cases = [
# (address_parts, is_newbuilding kwargs)
([], {}),
(
[
{"type": "location", "fullName": "Свердловская область"},
{"type": "street", "fullName": "улица Мира"},
{"type": "house", "fullName": "12"},
],
{},
),
(
[
{"type": "location", "fullName": "Свердловская область"},
{"type": "street", "fullName": "улица Новая"},
],
{
"building": building,
"building_cadastral_number": "66:41:0204016:11396",
"newbuilding": nb,
"is_newbuilding": True,
},
),
]
for parts, kwargs in cases:
assert old_format_address(parts, **kwargs) == new_format_address(parts, **kwargs)
def test_extract_state_parity() -> None:
"""app.extract_state ≡ scraper_kit.extract_state на SERP-фикстуре."""
html = _serp_html(_OFFERS)
old_state = old_extract_state(html, mfe="frontend-serp", key="initialState")
new_state = new_extract_state(html, mfe="frontend-serp", key="initialState")
assert old_state == new_state
assert new_state is not None
# ── detail fetch_detail parity ───────────────────────────────────────────────
def _detail_html() -> str:
"""defaultState + bti sister → _cianConfig['frontend-offer-card'] push HTML."""
default_state = {
"offerData": {
"offer": {
"cianId": 327830237,
"windowsViewType": "yardAndStreet",
"separateWcsCount": 1,
"combinedWcsCount": 0,
"building": {"ceilingHeight": 2.65},
"repairType": "cosmetic",
"kitchenArea": "11.5",
"description": "Отличная квартира",
"priceChanges": [
{"changeTime": "2026-04-01T00:00:00Z", "price": 5500000, "diffPercent": -1.8}
],
"agent": {
"id": 7654321,
"companyName": "Первичка Плюс",
"isPro": True,
"skills": [],
"accountType": "specialist",
},
"newbuilding": {"id": 0},
},
"stats": {
"totalViewsFormattedString": "1 042",
"todayViewsFormattedString": "7",
},
}
}
bti_state = {"houseData": {"series": "1-464", "wallMaterial": "brick"}}
ds_json = json.dumps(default_state, ensure_ascii=False)
bti_json = json.dumps(bti_state, ensure_ascii=False)
return (
""
)
def _mock_session(html: str) -> MagicMock:
response = MagicMock()
response.status_code = 200
response.text = html
session = MagicMock()
session.get = AsyncMock(return_value=response)
session.close = AsyncMock()
return session
@pytest.mark.asyncio
async def test_detail_fetch_parity() -> None:
"""fetch_detail (defaultState + bti) → DetailEnrichment идентичен (все поля)."""
html = _detail_html()
url = "https://ekb.cian.ru/sale/flat/327830237/"
old_res = await old_fetch_detail(url, session=_mock_session(html))
new_res = await new_fetch_detail(url, session=_mock_session(html))
assert old_res is not None
assert new_res is not None
assert dataclasses.asdict(old_res) == dataclasses.asdict(new_res)
# ── valuation _parse_valuation_state parity ──────────────────────────────────
def _valuation_state() -> dict[str, Any]:
return {
"user": {"isAuthenticated": True, "userId": 42},
"estimation": {
"sale": {
"data": {
"price": 6800000,
"accuracy": 87,
"priceFrom": 6400000,
"priceTo": 7200000,
"priceSqm": 130000,
"filtersHash": "hash-abc",
}
},
"rent": {
"data": {
"price": 42000,
"accuracy": 80,
"priceFrom": 39000,
"priceTo": 45000,
"taxPrice": 5460,
}
},
},
"estimationChart": {
"data": {
"title": {"change": "increase", "changeValue": "8,3%"},
"chartData": {
"data": [
{"date": 1714521600000, "price": 6700000, "priceFormatted": "6,7 млн"},
{"date": 1717200000000, "price": 6800000, "priceFormatted": "6,8 млн"},
]
},
}
},
"houseInfo": {
"data": {
"items": [
{"title": "Год постройки", "value": "2010"},
{"title": "Тип дома", "value": "монолит"},
]
}
},
"filters": {"houseId": 998877},
"managementCompany": {"data": {"name": "УК Ромашка", "phones": ["+73430000000"]}},
}
def test_valuation_parse_state_parity() -> None:
"""_parse_valuation_state → CianValuationResult идентичен (все поля)."""
state = _valuation_state()
old_res = old_parse_valuation(state)
new_res = new_parse_valuation(state)
assert dataclasses.asdict(old_res) == dataclasses.asdict(new_res)
# ── estimate_via_cian_valuation full-pipeline parity (#2335, Group E2) ───────
#
# test_valuation_parse_state_parity above only covers the pure parsing helper.
# These tests exercise the FULL async call path (cache lookup → load_session →
# curl_cffi fetch → extract_state → auth-check → parse → sanity-check → cache
# write), with config=RealScraperConfig() wired per the #2306 convention
# (app/services/cian_price_history.py), proving the kit path is safe to switch
# to whenever #2337 wires estimator.py's call site.
def _valuation_mfe_html(state: dict[str, Any]) -> str:
"""Wrap a valuation state dict as a valid _cianConfig push() HTML page."""
state_json = json.dumps(state, ensure_ascii=False)
return (
""
)
def _make_fake_async_session_cls(html: str) -> type:
"""Stand-in for curl_cffi.requests.AsyncSession — returns canned HTML on .get()."""
class _FakeAsyncSession:
def __init__(self, *args: Any, **kwargs: Any) -> None:
pass
async def __aenter__(self) -> _FakeAsyncSession:
return self
async def __aexit__(self, *exc: object) -> bool:
return False
async def get(self, url: str, allow_redirects: bool = True) -> SimpleNamespace:
return SimpleNamespace(status_code=200, text=html)
return _FakeAsyncSession
def test_estimate_via_cian_valuation_full_pipeline_parity(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Full estimate_via_cian_valuation() pipeline parity — not just the pure parser.
Mocks curl_cffi.AsyncSession (canned MFE HTML) + load_session (fixed cookies) on
BOTH legacy and kit modules, cache-miss on a MagicMock db, and drives both async
call paths with identical inputs (kit wired with config=RealScraperConfig()).
"""
html = _valuation_mfe_html(_valuation_state())
fake_session_cls = _make_fake_async_session_cls(html)
def _run_legacy() -> Any:
monkeypatch.setattr("app.services.scrapers.cian_valuation.AsyncSession", fake_session_cls)
monkeypatch.setattr(
"app.services.scrapers.cian_valuation.load_session",
lambda db: {"DMIR_AUTH": "token"},
)
db = MagicMock()
db.execute.return_value.mappings.return_value.first.return_value = None # cache miss
return asyncio.run(
old_estimate_via_cian_valuation(
db,
address="Свердловская обл, Екатеринбург",
total_area=50.0,
rooms_count=2,
floor=5,
total_floors=16,
use_cache=False,
)
)
def _run_kit() -> Any:
monkeypatch.setattr("scraper_kit.providers.cian.valuation.AsyncSession", fake_session_cls)
monkeypatch.setattr(
"scraper_kit.providers.cian.valuation.load_session",
lambda db, *, config: {"DMIR_AUTH": "token"},
)
db = MagicMock()
db.execute.return_value.mappings.return_value.first.return_value = None # cache miss
return asyncio.run(
new_estimate_via_cian_valuation(
db,
config=RealScraperConfig(),
address="Свердловская обл, Екатеринбург",
total_area=50.0,
rooms_count=2,
floor=5,
total_floors=16,
use_cache=False,
)
)
assert_parity(
legacy_fn=_run_legacy,
kit_fn=_run_kit,
fixtures=[()],
)
def test_kit_estimate_via_cian_valuation_requires_config() -> None:
"""Kit's estimate_via_cian_valuation MUST require config= (no default) — mandatory.
Regression guard (#2335): config: ScraperConfig has NO default in the kit
signature, so calling it without config= raises TypeError immediately (at
argument-binding time, before any coroutine runs). This is intentional and
must stay this way: estimator.py's call site (#2337) sits inside a blanket
`except Exception` used for graceful multi-source degradation — a future
change that gave config= a default would let a wiring mistake silently
degrade into "source unavailable" instead of raising loudly at call-time.
See #2306 for the precedent this guards against (silent-degradation footgun
from an optional dependency parameter).
"""
db = MagicMock()
with pytest.raises(TypeError, match="config"):
new_estimate_via_cian_valuation( # not awaited — TypeError raises at call time
db,
address="Свердловская обл, Екатеринбург",
total_area=50.0,
rooms_count=2,
floor=5,
)
def test_kit_cian_session_config_requirement_asymmetry() -> None:
"""scraper_kit.providers.cian.session: load_session requires config=, but
mark_session_invalid does NOT.
Confirmed by direct inspection for #2335 — the two helpers estimate_via_cian_valuation
calls internally are NOT symmetric re: config, unlike the assumption in the original
task brief ("the kit version of those also requires config mandatorily"). Documented
here as a regression guard so a future refactor that changes either signature is a
deliberate, visible decision rather than an accidental drift.
"""
from scraper_kit.providers.cian.session import load_session, mark_session_invalid
db = MagicMock()
with pytest.raises(TypeError, match="config"):
load_session(db) # missing mandatory config=
db2 = MagicMock()
mark_session_invalid(db2, 42) # no config parameter at all — must NOT raise
db2.execute.assert_called_once()
# ── newbuilding pure-экстракторы parity ──────────────────────────────────────
def test_newbuilding_extractors_parity() -> None:
rv_state = {
"data": {
"priceDynamics": {
"chart": {
"data": {"labels": ["2026-01-01", "2026-02-01"], "values": [120000, 125000]}
}
}
}
}
assert old_extract_chart(rv_state) == new_extract_chart(rv_state)
rel_state = {
"checkStatus": {"status": "reliable", "title": "Надёжный застройщик"},
"details": [{"title": "Проектная декларация", "type": "doc"}],
}
assert old_extract_reliability(rel_state) == new_extract_reliability(rel_state)
offers_state = {
"data": {
"total": {
"fromDeveloperRooms": [
{"roomType": "1", "layouts": [{"id": 1, "area": 38}]},
{"roomType": "2", "layouts": [{"id": 2, "area": 55}]},
]
}
}
}
assert old_extract_nested_offers(offers_state) == new_extract_nested_offers(offers_state)
serp_html = (
''
)
assert old_extract_zhk_url(serp_html) == new_extract_zhk_url(serp_html)
# ── strangler-guard: kit-провайдер не импортирует app.* ──────────────────────
def test_kit_cian_has_no_app_imports() -> None:
"""Ни один модуль scraper_kit.providers.cian.* не импортирует app.* (развязка)."""
import inspect
from scraper_kit.providers.cian import detail, newbuilding, serp, session, valuation
for mod in (serp, detail, newbuilding, valuation, session):
source = inspect.getsource(mod)
for line in source.splitlines():
stripped = line.strip()
assert not stripped.startswith("from app"), f"{mod.__name__}: {line!r}"
assert not stripped.startswith("import app"), f"{mod.__name__}: {line!r}"