"""Тесты для recent_permits_in_quarter в POST /api/v1/parcels/{cad_num}/analyze (#105 Phase 5). Покрывает: 1. analyze с quarter где есть РНС/РВЭ → recent_permits non-empty + summary calculated 2. analyze с quarter без permits → recent_permits=[], summary={rns_count=0, ...} 3. analyze 404 на invalid cad (no regression) Порядок db.execute в analyze_parcel меняется при добавлении блоков (SF-B5 вклинил red_lines / district_price / risk_zones), поэтому mock диспетчеризует строки по СИГНАТУРЕ SQL, а не по позиционному индексу — устойчиво к реордерингу: • geom UNION ALL / WKT / district / centroid — по содержимому SELECT • recent permits (#105) — по уникальной сигнатуре "permit_type, permit_number ... developer_inn" Прочие запросы обёрнуты в try/except SAVEPOINT в коде → дефолтный пустой mock (first=None, all=[], scalar=0) их не валит. """ from __future__ import annotations import datetime as dt from typing import Any from unittest.mock import MagicMock, patch import pytest from fastapi.testclient import TestClient from app.main import app # ── Константы ───────────────────────────────────────────────────────────────── _CAD = "66:41:0204016:10" # cad_num с 4 частями — quarter prefix = "66:41:0204016" _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]]]}' # ── Mock factories ───────────────────────────────────────────────────────────── def _make_mapping(data: dict[str, Any]) -> MagicMock: """Создать mock-строку (mapping) с __getitem__ + .get() для dict-like доступа.""" 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( geom_found: bool = True, permits_rows: list[dict[str, Any]] | None = None, ) -> MagicMock: """Сконструировать mock DB Session для analyze_parcel. permits_rows=None → пустой список (нет разрешений в квартале). permits_rows=[...] → список разрешений для агрегации. """ db = MagicMock() geom_row = ( _make_mapping({"geom_geojson": _GEOJSON, "geom_wkb": None, "source": "cad_quarter"}) if geom_found else None ) wkt_row = _make_mapping({"wkt": _WKT}) if geom_found else None district_row = _make_mapping( { "district_name": "Октябрьский", "median_price_per_m2": 120000, "dist_to_center": 1500.0, } ) centroid_row = _make_mapping({"lat": 56.84, "lon": 60.605}) raw_permits = [_make_mapping(p) for p in (permits_rows or [])] 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] = [] # Структурные запросы (нужны чтобы обработчик дошёл до permits без 404/202). if "AS geom_geojson" in sql: first_val = geom_row elif "AS wkt" in sql: first_val = wkt_row elif "AS median_price_per_m2" 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 # recent permits (#105) — уникальная сигнатура permit_type+permit_number+developer_inn. elif "permit_type" in sql and "permit_number" in sql and "developer_inn" in sql: all_val = raw_permits 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._fetch_air_quality_sync", 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_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_recent_permits_found_in_quarter() -> None: """analyze с квартала где есть РНС/РВЭ → recent_permits non-empty + summary calculated.""" from app.core.db import get_db rns_permit: dict[str, Any] = { "permit_type": "RNS", "permit_number": "66-RNS-2024-001", "issue_date": dt.date(2024, 6, 15), "developer_name": "Атомстрой", "developer_inn": "6671234567", "object_name": "МКД ЖК Тест", "object_type": "МКД", "construction_address": "г. Екатеринбург, ул. Ленина, 1", "total_area_sqm": 45000.0, } rve_permit: dict[str, Any] = { "permit_type": "RVE", "permit_number": "66-RVE-2025-002", "issue_date": dt.date(2025, 3, 10), "developer_name": "Атомстрой", "developer_inn": "6671234567", "object_name": "МКД ЖК Тест (1 оч.)", "object_type": "МКД", "construction_address": "г. Екатеринбург, ул. Ленина, 1", "total_area_sqm": None, } db = _make_db_for_analyze(permits_rows=[rns_permit, rve_permit]) 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 "recent_permits_in_quarter" in body assert "permits_summary" in body permits = body["recent_permits_in_quarter"] assert len(permits) == 2 rns = next(p for p in permits if p["permit_type"] == "RNS") assert rns["permit_number"] == "66-RNS-2024-001" assert rns["developer_name"] == "Атомстрой" assert rns["issue_date"] == "2024-06-15" assert rns["total_area_sqm"] == pytest.approx(45000.0) rve = next(p for p in permits if p["permit_type"] == "RVE") assert rve["issue_date"] == "2025-03-10" assert "units_count" not in rve summary = body["permits_summary"] assert summary["rns_count"] == 1 assert summary["rve_count"] == 1 assert summary["rns_total_area_sqm"] == pytest.approx(45000.0) assert "rve_total_units" not in summary assert len(summary["by_developer"]) == 1 assert summary["by_developer"][0]["name"] == "Атомстрой" assert summary["by_developer"][0]["permits_count"] == 2 finally: app.dependency_overrides.clear() _stop_patches() def test_recent_permits_empty_quarter() -> None: """analyze с quarter без permits → recent_permits=[], summary с нулями.""" from app.core.db import get_db db = _make_db_for_analyze(permits_rows=[]) 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 "recent_permits_in_quarter" in body assert body["recent_permits_in_quarter"] == [] assert "permits_summary" in body summary = body["permits_summary"] assert summary["rns_count"] == 0 assert summary["rve_count"] == 0 assert summary["rns_total_area_sqm"] == pytest.approx(0.0) assert "rve_total_units" not in summary assert summary["by_developer"] == [] finally: app.dependency_overrides.clear() _stop_patches() def test_recent_permits_invalid_cad_no_regression() -> None: """analyze с несуществующим cad → 202 или 404 (no regression от Phase 5).""" from app.core.db import get_db db = _make_db_for_analyze(geom_found=False) app.dependency_overrides[get_db] = _override_db(db) _start_patches() try: client = TestClient(app) resp = client.post("/api/v1/parcels/00:00:0000000:0/analyze") assert resp.status_code in ( 202, 404, ), f"Ожидали 202 или 404 для неизвестного cad, получили {resp.status_code}: {resp.text}" finally: app.dependency_overrides.clear() _stop_patches()