"""Тесты для parcel_meta в POST /api/v1/parcels/{cad_num}/analyze (#29 G2). Покрывает: 1. analyze когда cad_parcels has row → parcel_meta содержит permitted_use/land_category/cad_cost 2. analyze когда cad_parcels row отсутствует → parcel_meta == None Стратегия mock: аналогична test_analyze_market_price.py — DB mock через dependency_overrides, тяжёлые сервисы патчим через unittest.mock.patch. Диспетчеризация мока — ПО СИГНАТУРЕ SQL, а не по порядковому номеру вызова. Позиционный вариант ломался при каждом добавлении запроса в analyze_parcel (следы в истории: #29 G2, #2464 cluster B, затем #2464-G с запросом покрытия шумовой карты). Такое падение говорит не о parcel_meta, ради которого написан файл, а о том, что кто-то добавил блок выше по коду — красный CI без отношения к предмету теста. Матчинг по сигнатуре к порядку нечувствителен; всё, что не распознано, отдаёт пустой результат (обработчик это переживает — прочие блоки обёрнуты в try/except SAVEPOINT). """ from __future__ import annotations from typing import Any from unittest.mock import MagicMock, patch from fastapi.testclient import TestClient from app.main import app _CAD = "66:41:0204016:10" _WKT = "POLYGON((60.6 56.838, 60.61 56.838, 60.61 56.845, 60.6 56.845, 60.6 56.838))" _GEOJSON = '{"type":"Polygon","coordinates":[[[60.6,56.838],[60.61,56.838]]]}' def _make_mapping(data: dict[str, Any]) -> MagicMock: m = MagicMock() m.__getitem__ = lambda self, k: data[k] m.get = lambda k, default=None: data.get(k, default) return m def _make_db_for_analyze( parcel_meta_row: dict[str, Any] | None = None, ) -> MagicMock: """Сконструировать mock DB Session для analyze_parcel с фокусом на parcel_meta. parcel_meta_row=None → имитирует "строки нет в cad_parcels" → parcel_meta=None. parcel_meta_row={...} → имитирует найденную строку. """ db = MagicMock() geom_row = _make_mapping({"geom_geojson": _GEOJSON, "geom_wkb": None, "source": "cad_quarter"}) wkt_row = _make_mapping({"wkt": _WKT}) district_row = _make_mapping( { "district_name": "Октябрьский", "ekb_reference_median": 120000, "quarter_median_12m": None, "quarter_deals_count": 0, "dist_to_center": 1500.0, } ) centroid_row = _make_mapping({"lat": 56.84, "lon": 60.605}) pm_mock = _make_mapping(parcel_meta_row) if parcel_meta_row is not None else None def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock: sql = " ".join(str(args[0]).split()) if args else "" first_val: Any = None all_val: list[Any] = [] if "AS geom_geojson" in sql: first_val = geom_row elif "AS wkt" in sql: first_val = wkt_row elif "AS ekb_reference_median" in sql and "district_name" in sql: first_val = district_row elif "AS lon" in sql and "AS lat" in sql: first_val = centroid_row # Предмет файла: parcel_meta из cad_parcels (#29 G2). elif "AS permitted_use" in sql and "FROM cad_parcels" in sql: first_val = pm_mock # _neighbors_summary — ОДИН statement (neighbors + overlap + total через # json_agg + neighbors_total CTE, PR #1130 / #2464 cluster B). elif "neighbors_total_count" in sql or "json_agg" in sql: first_val = {"neighbors": [], "overlap_rows": [], "neighbors_total_count": 0} r = MagicMock() r.mappings.return_value.first.return_value = first_val r.mappings.return_value.all.return_value = all_val r.scalar.return_value = 0 return r db.execute.side_effect = _execute_side_effect ctx = MagicMock() ctx.__enter__ = MagicMock(return_value=ctx) ctx.__exit__ = MagicMock(return_value=False) db.begin_nested.return_value = ctx return db def _override_db(db: MagicMock): def _get_db_override(): yield db return _get_db_override _PATCHES = [ patch("app.api.v1.parcels.get_air_quality_cached", return_value=None), patch("app.api.v1.parcels.get_weather_cached", return_value=None), patch("app.api.v1.parcels.get_seasonal_weather_cached", return_value=None), patch( "app.api.v1.parcels.get_quarter_dump_data", return_value={ "nspd_zoning": None, "nspd_zouit_overlaps": [], "nspd_engineering_nearby": [], "nspd_risk_zones": [], "nspd_opportunity_parcels": [], "nspd_red_lines": [], "nspd_dump": {"available": False, "stale": False, "harvest_triggered": False}, }, ), patch("app.api.v1.parcels.compute_velocity", return_value=None), patch("app.api.v1.parcels.compute_gate_verdict", return_value={"verdict": "unknown"}), # enable_ird_analyze=True by default — мокаем чтобы не делать сетевые вызовы в тестах patch( "app.api.v1.parcels.build_ird_analyze_block", return_value={ "ird_overlaps": [], "ird_by_kind": {}, "opportunity_overlaps": [], "planning_projects": [], "functional_zone": None, "krt": [], "zone_regulation": None, }, ), ] def _start_patches() -> None: for p in _PATCHES: p.start() def _stop_patches() -> None: for p in _PATCHES: p.stop() def test_parcel_meta_found_in_cad_parcels() -> None: """analyze → cad_parcels has row → parcel_meta содержит permitted_use/land_category/cad_cost.""" from app.core.db import get_db pm_data: dict[str, Any] = { "permitted_use": "многоквартирный дом", "land_category": "Земли населённых пунктов", "land_subtype": "жилая застройка", "cad_cost": 5_000_000.0, } db = _make_db_for_analyze(parcel_meta_row=pm_data) app.dependency_overrides[get_db] = _override_db(db) _start_patches() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() assert "parcel_meta" in body, "parcel_meta отсутствует в ответе" pm = body["parcel_meta"] assert pm is not None, "parcel_meta должен быть не None при наличии строки" assert pm["permitted_use"] == "многоквартирный дом" assert pm["land_category"] == "Земли населённых пунктов" assert pm["land_subtype"] == "жилая застройка" assert pm["cad_cost"] == 5_000_000.0 assert pm["source"] == "cad_parcels" finally: app.dependency_overrides.clear() _stop_patches() def test_parcel_meta_none_when_cad_parcels_missing() -> None: """analyze → cad_parcels row отсутствует → parcel_meta == None.""" from app.core.db import get_db db = _make_db_for_analyze(parcel_meta_row=None) app.dependency_overrides[get_db] = _override_db(db) _start_patches() try: client = TestClient(app) resp = client.post(f"/api/v1/parcels/{_CAD}/analyze") assert resp.status_code == 200, resp.text body = resp.json() assert "parcel_meta" in body, "parcel_meta ключ должен присутствовать в ответе" assert body["parcel_meta"] is None, "parcel_meta должен быть None при отсутствии строки" finally: app.dependency_overrides.clear() _stop_patches()