test(tradein): fix #794 regressions — update existing sber/dkp tests to new API, dedup test files
This commit is contained in:
parent
468bdb052b
commit
d2722df52f
4 changed files with 143 additions and 142 deletions
|
|
@ -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)
|
||||
|
|
@ -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),
|
||||
):
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
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(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue