113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""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)
|