diff --git a/tradein-mvp/backend/tests/services/test_sber_index.py b/tradein-mvp/backend/tests/services/test_sber_index.py deleted file mode 100644 index 804f92b3..00000000 --- a/tradein-mvp/backend/tests/services/test_sber_index.py +++ /dev/null @@ -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=&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) diff --git a/tradein-mvp/backend/tests/test_estimator_imv_blend.py b/tradein-mvp/backend/tests/test_estimator_imv_blend.py index c79f3383..e80d16dc 100644 --- a/tradein-mvp/backend/tests/test_estimator_imv_blend.py +++ b/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( - 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"] == 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", - new=AsyncMock(return_value=None)), + patch("app.services.estimator.dadata_clean_address", 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", - return_value=(list(_BLEND_ANALOGS), False, "S")), + patch( + "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", - new=AsyncMock(return_value=None)), - patch("app.services.estimator._get_or_fetch_yandex_valuation_cached", - new=AsyncMock(return_value=None)), - patch("app.services.estimator.estimate_via_cian_valuation", - new=AsyncMock(return_value=None)), + patch( + "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( + "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), ): diff --git a/tradein-mvp/backend/tests/services/test_estimator_sber_time_adjust.py b/tradein-mvp/backend/tests/test_estimator_sber_time_adjust.py similarity index 100% rename from tradein-mvp/backend/tests/services/test_estimator_sber_time_adjust.py rename to tradein-mvp/backend/tests/test_estimator_sber_time_adjust.py diff --git a/tradein-mvp/backend/tests/test_sber_index.py b/tradein-mvp/backend/tests/test_sber_index.py index f658ca40..936c6918 100644 --- a/tradein-mvp/backend/tests/test_sber_index.py +++ b/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=&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] - yield ("Россия", date(2024, 6, 1), "Вторичный", dashboard, 115000.0) + async def _mock_fetch( # type: ignore[override] + 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(