tradein-mvp/backend/tests/services/test_sber_index.py

@@ -1,113 +0,0 @@
-"""Tests for app/services/sber_index.py — per-dashboard filter fix (#794)."""
-
-from __future__ import annotations
-
-import base64
-import json
-from datetime import date
-from pathlib import Path
-from urllib.parse import unquote
-
-import pytest
-
-from app.services.sber_index import (
-    SBER_DASHBOARDS,
-    SberDashboard,
-    _parse_period_month,
-    build_sber_route,
-    decode_sber_response,
-)
-
-# ---------------------------------------------------------------------------
-# Fixture path
-# ---------------------------------------------------------------------------
-
-FIXTURE_PATH = Path(__file__).parent.parent / "fixtures" / "sber_real_estate_deals_66.json"
-
-
-# ---------------------------------------------------------------------------
-# test_build_sber_route_per_dashboard
-# LOCKS the bug fix: each dashboard must have the correct filter keys.
-# ---------------------------------------------------------------------------
-
-
-@pytest.mark.parametrize("dashboard", SBER_DASHBOARDS)
-def test_build_sber_route_per_dashboard(dashboard: SberDashboard) -> None:
-    route_b64 = build_sber_route(dashboard.slug, "66", dashboard.extra_filter)
-    path = base64.b64decode(route_b64).decode("utf-8")
-
-    # Extract the filter JSON from the path: ?filter=<urlencoded>&limit=...
-    filter_part = path.split("filter=")[1].split("&")[0]
-    filter_dict = json.loads(unquote(filter_part))
-
-    # All routes must have REF_AREA + FREQ=M
-    assert filter_dict.get("REF_AREA") == "66"
-    assert filter_dict.get("FREQ") == "M"
-
-    if dashboard.slug == "real_estate_deals":
-        assert filter_dict.get("SOURCE") == "SI"
-        assert filter_dict.get("REALTY") == "2"
-
-    elif dashboard.slug == "residential_real_estate_prices":
-        assert filter_dict.get("SOURCE") == "SI"
-        assert filter_dict.get("REAL_ESTATE_NOVELTY") == "3"
-
-    elif dashboard.slug == "dinamika-tsen-obyavlenii":
-        assert filter_dict.get("SOURCE") == "DK"
-        assert filter_dict.get("PRICE_TYPE") == "1"
-
-    else:
-        pytest.fail(f"Unknown dashboard slug: {dashboard.slug!r}")
-
-
-# ---------------------------------------------------------------------------
-# test_decode_sber_response_real_fixture
-# ---------------------------------------------------------------------------
-
-
-def test_decode_sber_response_real_fixture() -> None:
-    with FIXTURE_PATH.open(encoding="utf-8") as f:
-        raw = json.load(f)
-
-    rows = decode_sber_response(raw)
-
-    assert len(rows) >= 1, "Expected at least 1 decoded row from real fixture"
-
-    # All rows should have the expected fields present
-    first = rows[0]
-    assert "ref_area" in first
-    assert "realty" in first
-    assert "value" in first
-    assert "period" in first
-
-    # Values must be typed correctly
-    assert isinstance(first["value"], float)
-    assert isinstance(first["period"], str) and first["period"]
-
-    # Verify region label decoded correctly
-    regions = {r["ref_area"] for r in rows}
-    assert (
-        "Свердловская область" in regions
-    ), f"Expected 'Свердловская область' in ref_area values, got: {regions}"
-
-    # Verify secondary-market segment present
-    realty_vals = {r["realty"] for r in rows}
-    assert any(
-        "тори" in str(v) or "Вторичн" in str(v) for v in realty_vals
-    ), f"Expected secondary-market label in realty field, got: {realty_vals}"
-
-
-# ---------------------------------------------------------------------------
-# test_parse_period_month
-# ---------------------------------------------------------------------------
-
-
-def test_parse_period_month() -> None:
-    result = _parse_period_month("2026-04-29T21:00:00.000Z")
-    assert result == date(2026, 4, 1)
-
-
-def test_parse_period_month_z_suffix() -> None:
-    """Original format from live API."""
-    result = _parse_period_month("2017-01-30T21:00:00Z")
-    assert result == date(2017, 1, 1)

tradein-mvp/backend/tests/test_estimator_imv_blend.py

@@ -184,9 +184,8 @@ def test_dkp_corridor_aggregates_ppm2() -> None:
         {"price_per_m2": 120_000},
         {"price_per_m2": 140_000},
     ]
-    res = _fetch_dkp_corridor(
+    with patch("app.services.estimator._load_sber_index_series", return_value={}):
-        mock_db, address="улица Ленина, 5", rooms=2, area=50.0
+        res = _fetch_dkp_corridor(mock_db, address="улица Ленина, 5", rooms=2, area=50.0)
-    )
     assert res is not None
     assert res["count"] == 3
     assert res["low_ppm2"] == 100_000
@@ -275,7 +274,8 @@ def test_dkp_corridor_count_below_three_still_aggregates() -> None:
         {"price_per_m2": 100_000},
         {"price_per_m2": 130_000},
     ]
-    res = _fetch_dkp_corridor(mock_db, address="улица Ленина, 5", rooms=2, area=50.0)
+    with patch("app.services.estimator._load_sber_index_series", return_value={}):
+        res = _fetch_dkp_corridor(mock_db, address="улица Ленина, 5", rooms=2, area=50.0)
     assert res is not None
     assert res["count"] == 2
     assert res["low_ppm2"] == 100_000
@@ -351,20 +351,26 @@ def _run_estimate_with_anchor(
     async def _run():
         with (
             patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_fake_geo())),
-            patch("app.services.estimator.dadata_clean_address",
+            patch("app.services.estimator.dadata_clean_address", new=AsyncMock(return_value=None)),
-                  new=AsyncMock(return_value=None)),
             patch("app.services.estimator.match_house_readonly", return_value=None),
             patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
-            patch("app.services.estimator._fetch_analogs",
+            patch(
-                  return_value=(list(_BLEND_ANALOGS), False, "S")),
+                "app.services.estimator._fetch_analogs",
+                return_value=(list(_BLEND_ANALOGS), False, "S"),
+            ),
             patch("app.services.estimator._fetch_deals", return_value=[]),
             patch("app.services.estimator._fetch_dkp_corridor", return_value=None),
-            patch("app.services.estimator._get_or_fetch_imv_cached",
+            patch(
-                  new=AsyncMock(return_value=None)),
+                "app.services.estimator._get_or_fetch_imv_cached", new=AsyncMock(return_value=None)
-            patch("app.services.estimator._get_or_fetch_yandex_valuation_cached",
+            ),
-                  new=AsyncMock(return_value=None)),
+            patch(
-            patch("app.services.estimator.estimate_via_cian_valuation",
+                "app.services.estimator._get_or_fetch_yandex_valuation_cached",
-                  new=AsyncMock(return_value=None)),
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator.estimate_via_cian_valuation",
+                new=AsyncMock(return_value=None),
+            ),
             patch("app.services.estimator._get_asking_sold_ratio", return_value=ratio_tuple),
             patch("app.services.estimator._fetch_house_imv_anchor", return_value=anchor),
         ):

tradein-mvp/backend/tests/test_sber_index.py

@@ -1,10 +1,12 @@
 """Tests for sber_index service — pure unit tests, no network/DB.
 
 Coverage:
-  (a) build_sber_route: base64 route builder produces correct encoded string.
+  (a) build_sber_route: base64 route builder produces correct encoded string, per dashboard.
   (b) decode_sber_response: decodes a sample base64 payload into correct rows.
   (c) pull_sber_indices upsert SQL is idempotent (ON CONFLICT shape assertion via
       fake-db roundtrip — verifies upsert is called with correct SQL and params).
+  (d) Real-fixture decode: exercises actual sberindex.ru response format end-to-end.
+  (e) Period parsing edge-cases.
 """
 
 from __future__ import annotations
@@ -14,6 +16,7 @@ import json
 import os
 import sys
 from datetime import date
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 from urllib.parse import unquote
 
@@ -27,19 +30,32 @@ sys.modules.setdefault("weasyprint", _wp_mock)
 import pytest  # noqa: E402
 
 from app.services.sber_index import (  # noqa: E402
+    SBER_DASHBOARDS,
+    SberDashboard,
     _parse_period_month,
     build_sber_route,
     decode_sber_response,
 )
 
+# Lookup helper: slug → SberDashboard
+_DASH: dict[str, SberDashboard] = {d.slug: d for d in SBER_DASHBOARDS}
+
 # ---------------------------------------------------------------------------
-# (a) Route builder
+# Fixture path (real sberindex.ru response)
+# ---------------------------------------------------------------------------
+
+FIXTURE_PATH = Path(__file__).parent / "fixtures" / "sber_real_estate_deals_66.json"
+
+
+# ---------------------------------------------------------------------------
+# (a) Route builder — per-dashboard filter assertions
 # ---------------------------------------------------------------------------
 
 
 def test_build_sber_route_rф_real_estate_deals() -> None:
     """Route for РФ (643) + real_estate_deals should decode to expected path."""
-    encoded = build_sber_route("real_estate_deals", "643")
+    d = _DASH["real_estate_deals"]
+    encoded = build_sber_route(d.slug, "643", d.extra_filter)
     decoded = base64.b64decode(encoded).decode("utf-8")
 
     assert decoded.startswith("/dataset/v1/real_estate_deals?filter=")
@@ -58,14 +74,16 @@ def test_build_sber_route_rф_real_estate_deals() -> None:
 
 def test_build_sber_route_different_dashboard() -> None:
     """Route slug is correctly embedded in the decoded path."""
-    encoded = build_sber_route("dinamika-tsen-obyavlenii", "643")
+    d = _DASH["dinamika-tsen-obyavlenii"]
+    encoded = build_sber_route(d.slug, "643", d.extra_filter)
     decoded = base64.b64decode(encoded).decode("utf-8")
     assert "/dataset/v1/dinamika-tsen-obyavlenii?" in decoded
 
 
 def test_build_sber_route_is_valid_base64() -> None:
     """Encoded route must be decodeable ASCII base64."""
-    encoded = build_sber_route("residential_real_estate_prices", "77")
+    d = _DASH["residential_real_estate_prices"]
+    encoded = build_sber_route(d.slug, "77", d.extra_filter)
     # Should not raise
     decoded_bytes = base64.b64decode(encoded)
     assert len(decoded_bytes) > 0
@@ -73,13 +91,44 @@ def test_build_sber_route_is_valid_base64() -> None:
 
 def test_build_sber_route_ref_area_embedded() -> None:
     """REF_AREA value is preserved exactly in the filter JSON."""
-    encoded = build_sber_route("real_estate_deals", "66")
+    d = _DASH["real_estate_deals"]
+    encoded = build_sber_route(d.slug, "66", d.extra_filter)
     decoded = base64.b64decode(encoded).decode("utf-8")
     filter_part = decoded.split("filter=")[1].split("&")[0]
     filter_json = json.loads(unquote(filter_part))
     assert filter_json["REF_AREA"] == "66"
 
 
+@pytest.mark.parametrize("dashboard", SBER_DASHBOARDS)
+def test_build_sber_route_per_dashboard(dashboard: SberDashboard) -> None:
+    """Each dashboard must produce the correct per-dashboard filter keys (locks #794 fix)."""
+    route_b64 = build_sber_route(dashboard.slug, "66", dashboard.extra_filter)
+    path = base64.b64decode(route_b64).decode("utf-8")
+
+    # Extract the filter JSON from the path: ?filter=<urlencoded>&limit=...
+    filter_part = path.split("filter=")[1].split("&")[0]
+    filter_dict = json.loads(unquote(filter_part))
+
+    # All routes must have REF_AREA + FREQ=M
+    assert filter_dict.get("REF_AREA") == "66"
+    assert filter_dict.get("FREQ") == "M"
+
+    if dashboard.slug == "real_estate_deals":
+        assert filter_dict.get("SOURCE") == "SI"
+        assert filter_dict.get("REALTY") == "2"
+
+    elif dashboard.slug == "residential_real_estate_prices":
+        assert filter_dict.get("SOURCE") == "SI"
+        assert filter_dict.get("REAL_ESTATE_NOVELTY") == "3"
+
+    elif dashboard.slug == "dinamika-tsen-obyavlenii":
+        assert filter_dict.get("SOURCE") == "DK"
+        assert filter_dict.get("PRICE_TYPE") == "1"
+
+    else:
+        pytest.fail(f"Unknown dashboard slug: {dashboard.slug!r}")
+
+
 # ---------------------------------------------------------------------------
 # (b) Response decoder
 # ---------------------------------------------------------------------------
@@ -191,6 +240,59 @@ def test_decode_sber_response_missing_fields_key_raises() -> None:
         decode_sber_response(bad_payload)
 
 
+# ---------------------------------------------------------------------------
+# (d) Real-fixture decode — exercises actual sberindex.ru response format
+# ---------------------------------------------------------------------------
+
+
+def test_decode_sber_response_real_fixture() -> None:
+    with FIXTURE_PATH.open(encoding="utf-8") as f:
+        raw = json.load(f)
+
+    rows = decode_sber_response(raw)
+
+    assert len(rows) >= 1, "Expected at least 1 decoded row from real fixture"
+
+    # All rows should have the expected fields present
+    first = rows[0]
+    assert "ref_area" in first
+    assert "realty" in first
+    assert "value" in first
+    assert "period" in first
+
+    # Values must be typed correctly
+    assert isinstance(first["value"], float)
+    assert isinstance(first["period"], str) and first["period"]
+
+    # Verify region label decoded correctly
+    regions = {r["ref_area"] for r in rows}
+    assert (
+        "Свердловская область" in regions
+    ), f"Expected 'Свердловская область' in ref_area values, got: {regions}"
+
+    # Verify secondary-market segment present
+    realty_vals = {r["realty"] for r in rows}
+    assert any(
+        "тори" in str(v) or "Вторичн" in str(v) for v in realty_vals
+    ), f"Expected secondary-market label in realty field, got: {realty_vals}"
+
+
+# ---------------------------------------------------------------------------
+# (e) Period parsing edge-cases
+# ---------------------------------------------------------------------------
+
+
+def test_parse_period_month() -> None:
+    result = _parse_period_month("2026-04-29T21:00:00.000Z")
+    assert result == date(2026, 4, 1)
+
+
+def test_parse_period_month_z_suffix() -> None:
+    """Original format from live API."""
+    result = _parse_period_month("2017-01-30T21:00:00Z")
+    assert result == date(2017, 1, 1)
+
+
 # ---------------------------------------------------------------------------
 # (c) Upsert SQL idempotency — fake-db roundtrip
 # ---------------------------------------------------------------------------
@@ -220,7 +322,9 @@ async def test_pull_sber_indices_upsert_on_conflict_idempotent() -> None:
     fake_row = ("Россия", date(2024, 1, 1), "Вторичный", "real_estate_deals", 95000.0)
 
     # Patch fetch_sber_index to yield one deterministic row per call
-    async def _mock_fetch(client: object, *, dashboard: str, ref_area: str):  # type: ignore[override]
+    async def _mock_fetch(  # type: ignore[override]
+        client: object, *, dashboard: SberDashboard, ref_area: str
+    ):
         yield fake_row
 
     with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch):
@@ -228,7 +332,7 @@ async def test_pull_sber_indices_upsert_on_conflict_idempotent() -> None:
         result = await pull_sber_indices(
             fake_db,  # type: ignore[arg-type]
             cities={"643": "Россия"},
-            dashboards=["real_estate_deals"],
+            dashboards=[_DASH["real_estate_deals"]],
         )
 
     assert result["upserted"] == 1
@@ -261,18 +365,20 @@ async def test_pull_sber_indices_error_per_series_continues() -> None:
     fake_db = _ExecuteCapture()
     call_count = 0
 
-    async def _mock_fetch_with_error(client: object, *, dashboard: str, ref_area: str):  # type: ignore[override]
+    async def _mock_fetch_with_error(  # type: ignore[override]
+        client: object, *, dashboard: SberDashboard, ref_area: str
+    ):
         nonlocal call_count
         call_count += 1
-        if dashboard == "real_estate_deals":
+        if dashboard.slug == "real_estate_deals":
             raise Exception("Simulated network failure")
-        yield ("Россия", date(2024, 1, 1), "Вторичный", dashboard, 90000.0)
+        yield ("Россия", date(2024, 1, 1), "Вторичный", dashboard.slug, 90000.0)
 
     with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch_with_error):
         result = await pull_sber_indices(
             fake_db,  # type: ignore[arg-type]
             cities={"643": "Россия"},
-            dashboards=["real_estate_deals", "residential_real_estate_prices"],
+            dashboards=[_DASH["real_estate_deals"], _DASH["residential_real_estate_prices"]],
         )
 
     # First series failed, second succeeded
@@ -289,15 +395,17 @@ async def test_pull_sber_indices_asking_benchmark_logged(caplog: pytest.LogCaptu
 
     fake_db = _ExecuteCapture()
 
-    async def _mock_fetch(client: object, *, dashboard: str, ref_area: str):  # type: ignore[override]
+    async def _mock_fetch(  # type: ignore[override]
-        yield ("Россия", date(2024, 6, 1), "Вторичный", dashboard, 115000.0)
+        client: object, *, dashboard: SberDashboard, ref_area: str
+    ):
+        yield ("Россия", date(2024, 6, 1), "Вторичный", dashboard.slug, 115000.0)
 
     with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch):
         with caplog.at_level(logging.INFO, logger="app.services.sber_index"):
             await pull_sber_indices(
                 fake_db,  # type: ignore[arg-type]
                 cities={"643": "Россия"},
-                dashboards=["dinamika-tsen-obyavlenii"],
+                dashboards=[_DASH["dinamika-tsen-obyavlenii"]],
             )
 
     assert any(