535 lines
20 KiB
Python
535 lines
20 KiB
Python
"""Tests for sber_index service — pure unit tests, no network/DB.
|
||
|
||
Coverage:
|
||
(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
|
||
|
||
import base64
|
||
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
|
||
|
||
# DATABASE_URL required by config before any app import.
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
# WeasyPrint stub — not installed in CI without GTK.
|
||
_wp_mock = MagicMock()
|
||
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}
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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."""
|
||
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=")
|
||
assert "limit=1000" in decoded
|
||
assert "offset=0" in decoded
|
||
|
||
# Extract filter JSON from decoded path
|
||
filter_part = decoded.split("filter=")[1].split("&")[0]
|
||
filter_json = json.loads(unquote(filter_part))
|
||
|
||
assert filter_json["REF_AREA"] == "643"
|
||
assert filter_json["FREQ"] == "M"
|
||
assert filter_json["SOURCE"] == "SI"
|
||
assert filter_json["REALTY"] == "2"
|
||
|
||
|
||
def test_build_sber_route_different_dashboard() -> None:
|
||
"""Route slug is correctly embedded in the decoded path."""
|
||
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."""
|
||
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
|
||
|
||
|
||
def test_build_sber_route_ref_area_embedded() -> None:
|
||
"""REF_AREA value is preserved exactly in the filter JSON."""
|
||
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":
|
||
# #902: вторичка hedonic — re-captured live 2026-06-17 (was TYPE=1/NOVELTY=3 → 404).
|
||
assert filter_dict.get("SOURCE") == "SI"
|
||
assert filter_dict.get("REAL_ESTATE_TYPE") == "2"
|
||
assert filter_dict.get("REAL_ESTATE_NOVELTY") == "1"
|
||
|
||
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}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# (a2) #902 — exact base64 lock for the live-captured /dataset/v1/<slug> routes
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# REF_AREA=643 (РФ) routes captured live 2026-06-17 via Playwright route-interception
|
||
# on the sber dashboards (all 200 OK). Each entry: (slug, filter_dict, expected_base64).
|
||
# The expected base64 is base64(/dataset/v1/<slug>?filter=<urlencode(json)>&limit=1000&offset=0)
|
||
# with json.dumps(separators=(",", ":")) field order = REF_AREA, FREQ, SOURCE, <dims...>.
|
||
_CAPTURED_ROUTES_643: list[tuple[str, dict[str, str], str]] = [
|
||
(
|
||
"residential_real_estate_prices",
|
||
{"SOURCE": "SI", "REAL_ESTATE_TYPE": "1", "REAL_ESTATE_NOVELTY": "1"},
|
||
"L2RhdGFzZXQvdjEvcmVzaWRlbnRpYWxfcmVhbF9lc3RhdGVfcHJpY2VzP2ZpbHRlcj0lN0IlMjJSRUZf"
|
||
"QVJFQSUyMiUzQSUyMjY0MyUyMiUyQyUyMkZSRVElMjIlM0ElMjJNJTIyJTJDJTIyU09VUkNFJTIyJTNB"
|
||
"JTIyU0klMjIlMkMlMjJSRUFMX0VTVEFURV9UWVBFJTIyJTNBJTIyMSUyMiUyQyUyMlJFQUxfRVNUQVRF"
|
||
"X05PVkVMVFklMjIlM0ElMjIxJTIyJTdEJmxpbWl0PTEwMDAmb2Zmc2V0PTA=",
|
||
),
|
||
(
|
||
"residential_real_estate_prices",
|
||
{"SOURCE": "SI", "REAL_ESTATE_TYPE": "2", "REAL_ESTATE_NOVELTY": "1"},
|
||
"L2RhdGFzZXQvdjEvcmVzaWRlbnRpYWxfcmVhbF9lc3RhdGVfcHJpY2VzP2ZpbHRlcj0lN0IlMjJSRUZf"
|
||
"QVJFQSUyMiUzQSUyMjY0MyUyMiUyQyUyMkZSRVElMjIlM0ElMjJNJTIyJTJDJTIyU09VUkNFJTIyJTNB"
|
||
"JTIyU0klMjIlMkMlMjJSRUFMX0VTVEFURV9UWVBFJTIyJTNBJTIyMiUyMiUyQyUyMlJFQUxfRVNUQVRF"
|
||
"X05PVkVMVFklMjIlM0ElMjIxJTIyJTdEJmxpbWl0PTEwMDAmb2Zmc2V0PTA=",
|
||
),
|
||
(
|
||
"dinamika-tsen-obyavlenii",
|
||
{"SOURCE": "DK", "PRICE_TYPE": "1"},
|
||
"L2RhdGFzZXQvdjEvZGluYW1pa2EtdHNlbi1vYnlhdmxlbmlpP2ZpbHRlcj0lN0IlMjJSRUZfQVJFQSUy"
|
||
"MiUzQSUyMjY0MyUyMiUyQyUyMkZSRVElMjIlM0ElMjJNJTIyJTJDJTIyU09VUkNFJTIyJTNBJTIyREsl"
|
||
"MjIlMkMlMjJQUklDRV9UWVBFJTIyJTNBJTIyMSUyMiU3RCZsaW1pdD0xMDAwJm9mZnNldD0w",
|
||
),
|
||
(
|
||
"dinamika-tsen-obyavlenii",
|
||
{"SOURCE": "DK", "PRICE_TYPE": "2"},
|
||
"L2RhdGFzZXQvdjEvZGluYW1pa2EtdHNlbi1vYnlhdmxlbmlpP2ZpbHRlcj0lN0IlMjJSRUZfQVJFQSUy"
|
||
"MiUzQSUyMjY0MyUyMiUyQyUyMkZSRVElMjIlM0ElMjJNJTIyJTJDJTIyU09VUkNFJTIyJTNBJTIyREsl"
|
||
"MjIlMkMlMjJQUklDRV9UWVBFJTIyJTNBJTIyMiUyMiU3RCZsaW1pdD0xMDAwJm9mZnNldD0w",
|
||
),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize(("slug", "extra_filter", "expected_b64"), _CAPTURED_ROUTES_643)
|
||
def test_build_sber_route_matches_captured_b64(
|
||
slug: str, extra_filter: dict[str, str], expected_b64: str
|
||
) -> None:
|
||
"""#902: build_sber_route must reproduce the live-captured base64 route byte-for-byte.
|
||
|
||
Locks the /dataset/v1/<slug> path (NOT /dataset/v1/data/<slug>) and the exact filter
|
||
dimension set + JSON key order. Guards against silent dataset-path regressions.
|
||
"""
|
||
encoded = build_sber_route(slug, "643", extra_filter)
|
||
assert encoded == expected_b64
|
||
|
||
# And the decoded path must use the bare /dataset/v1/<slug> form (not /data/).
|
||
decoded = base64.b64decode(encoded).decode("utf-8")
|
||
assert decoded.startswith(f"/dataset/v1/{slug}?filter=")
|
||
assert "/dataset/v1/data/" not in decoded
|
||
|
||
|
||
def test_active_dashboards_match_captured_secondary_series() -> None:
|
||
"""#902: the two previously-404'ing dashboards now use a captured-200 filter combo.
|
||
|
||
residential_real_estate_prices → вторичка hedonic (TYPE=2, NOVELTY=1).
|
||
dinamika-tsen-obyavlenii → ДомКлик asking (PRICE_TYPE=1).
|
||
Each active SBER_DASHBOARDS entry's base64 must equal one of the captured 643 routes.
|
||
"""
|
||
captured_b64 = {b for _, _, b in _CAPTURED_ROUTES_643}
|
||
for slug in ("residential_real_estate_prices", "dinamika-tsen-obyavlenii"):
|
||
d = _DASH[slug]
|
||
encoded = build_sber_route(d.slug, "643", d.extra_filter)
|
||
assert encoded in captured_b64, (
|
||
f"active dashboard {slug!r} route not in captured-live set — "
|
||
f"filter={d.extra_filter}"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# (b) Response decoder
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_b64_cell(value: str) -> str:
|
||
return "__string__" + base64.b64encode(value.encode("utf-8")).decode("ascii")
|
||
|
||
|
||
def _make_num_cell(value: float) -> str:
|
||
return f"__number__{value}"
|
||
|
||
|
||
def _make_sowa_response(rows: list[list[str]]) -> dict: # type: ignore[type-arg]
|
||
"""Build a minimal SOWA response dict with a fixed field list."""
|
||
fields = [
|
||
"indicator_id",
|
||
"kpi_id",
|
||
"period",
|
||
"value",
|
||
"obs_status",
|
||
"source",
|
||
"ref_area",
|
||
"realty",
|
||
"freq",
|
||
"unit_measure",
|
||
"unit_mult",
|
||
]
|
||
encoded_fields = [_make_b64_cell(f) for f in fields]
|
||
return {
|
||
"SOWA": {
|
||
"method": "GET",
|
||
"route": "test_route",
|
||
"data": {
|
||
"type": "object",
|
||
"value": [
|
||
{"key": "fields", "value": encoded_fields},
|
||
{"key": "data", "value": rows},
|
||
],
|
||
},
|
||
}
|
||
}
|
||
|
||
|
||
def test_decode_sber_response_basic() -> None:
|
||
"""Decode a two-row response and check period + value extraction."""
|
||
row1 = [
|
||
_make_b64_cell("1"), # indicator_id
|
||
_make_num_cell(643032001), # kpi_id
|
||
_make_b64_cell("2017-01-30T21:00:00Z"), # period
|
||
_make_num_cell(48549), # value
|
||
_make_b64_cell("A"), # obs_status
|
||
_make_b64_cell("Данные СберИндекса"), # source
|
||
_make_b64_cell("Россия"), # ref_area
|
||
_make_b64_cell("Вторичный"), # realty
|
||
_make_b64_cell("Месяц"), # freq
|
||
_make_b64_cell("руб"), # unit_measure
|
||
_make_num_cell(0), # unit_mult
|
||
]
|
||
row2 = [
|
||
_make_b64_cell("1"),
|
||
_make_num_cell(643032001),
|
||
_make_b64_cell("2024-06-30T21:00:00Z"),
|
||
_make_num_cell(120000),
|
||
_make_b64_cell("A"),
|
||
_make_b64_cell("Данные СберИндекса"),
|
||
_make_b64_cell("Россия"),
|
||
_make_b64_cell("Вторичный"),
|
||
_make_b64_cell("Месяц"),
|
||
_make_b64_cell("руб"),
|
||
_make_num_cell(0),
|
||
]
|
||
payload = _make_sowa_response([row1, row2])
|
||
decoded = decode_sber_response(payload)
|
||
|
||
assert len(decoded) == 2
|
||
assert decoded[0]["ref_area"] == "Россия"
|
||
assert decoded[0]["value"] == 48549.0
|
||
assert decoded[0]["period"] == "2017-01-30T21:00:00Z"
|
||
assert decoded[1]["value"] == 120000.0
|
||
|
||
|
||
def test_decode_sber_response_period_normalises_to_first_of_month() -> None:
|
||
"""_parse_period_month always returns the first day of the month."""
|
||
result = _parse_period_month("2017-01-30T21:00:00Z")
|
||
assert result == date(2017, 1, 1)
|
||
|
||
result2 = _parse_period_month("2024-06-30T21:00:00Z")
|
||
assert result2 == date(2024, 6, 1)
|
||
|
||
|
||
def test_decode_sber_response_empty_data() -> None:
|
||
"""Empty data section returns empty list."""
|
||
payload = _make_sowa_response([])
|
||
decoded = decode_sber_response(payload)
|
||
assert decoded == []
|
||
|
||
|
||
def test_decode_sber_response_missing_fields_key_raises() -> None:
|
||
"""Response without 'fields' key raises ValueError."""
|
||
bad_payload = {
|
||
"SOWA": {
|
||
"method": "GET",
|
||
"route": "x",
|
||
"data": {"type": "object", "value": [{"key": "data", "value": []}]},
|
||
}
|
||
}
|
||
with pytest.raises(ValueError, match="fields"):
|
||
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
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class _ExecuteCapture:
|
||
"""Captures execute() calls and their params; simulate no-op DB."""
|
||
|
||
def __init__(self) -> None:
|
||
self.calls: list[tuple[str, dict]] = [] # type: ignore[type-arg]
|
||
|
||
def execute(self, stmt: object, params: dict | None = None) -> MagicMock: # type: ignore[return-value]
|
||
sql_str = str(stmt)
|
||
self.calls.append((sql_str, params or {}))
|
||
return MagicMock(fetchone=lambda: None, fetchall=lambda: [])
|
||
|
||
def commit(self) -> None:
|
||
pass
|
||
|
||
def rollback(self) -> None:
|
||
pass
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_pull_sber_indices_upsert_on_conflict_idempotent() -> None:
|
||
"""pull_sber_indices emits ON CONFLICT ... DO UPDATE upsert SQL (idempotent)."""
|
||
from app.services.sber_index import pull_sber_indices
|
||
|
||
fake_db = _ExecuteCapture()
|
||
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( # 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):
|
||
# Only pull one dashboard × one city to keep assertion simple
|
||
result = await pull_sber_indices(
|
||
fake_db, # type: ignore[arg-type]
|
||
cities={"643": "Россия"},
|
||
dashboards=[_DASH["real_estate_deals"]],
|
||
)
|
||
|
||
assert result["upserted"] == 1
|
||
assert result["errors"] == 0
|
||
|
||
# Find the INSERT call
|
||
insert_calls = [sql for sql, _ in fake_db.calls if "INSERT INTO sber_price_index" in sql]
|
||
assert insert_calls, "Expected at least one INSERT INTO sber_price_index call"
|
||
|
||
upsert_sql = insert_calls[0]
|
||
# Must contain ON CONFLICT clause with DO UPDATE SET
|
||
assert "ON CONFLICT" in upsert_sql
|
||
assert "DO UPDATE SET" in upsert_sql
|
||
assert "index_value_rub_m2" in upsert_sql
|
||
assert "fetched_at" in upsert_sql
|
||
|
||
# Must NOT contain :: type casts (psycopg v3 rule)
|
||
import re
|
||
|
||
assert not re.search(
|
||
r":[a-z_]+::[a-z]", upsert_sql
|
||
), "SQL must not contain ::type casts — use CAST(... AS type) instead"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_pull_sber_indices_error_per_series_continues() -> None:
|
||
"""A failing series increments errors but does not abort other series."""
|
||
from app.services.sber_index import pull_sber_indices
|
||
|
||
fake_db = _ExecuteCapture()
|
||
call_count = 0
|
||
|
||
async def _mock_fetch_with_error( # type: ignore[override]
|
||
client: object, *, dashboard: SberDashboard, ref_area: str
|
||
):
|
||
nonlocal call_count
|
||
call_count += 1
|
||
if dashboard.slug == "real_estate_deals":
|
||
raise Exception("Simulated network failure")
|
||
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=[_DASH["real_estate_deals"], _DASH["residential_real_estate_prices"]],
|
||
)
|
||
|
||
# First series failed, second succeeded
|
||
assert result["errors"] == 1
|
||
assert result["upserted"] == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_pull_sber_indices_404_logs_warning_with_path_hint(
|
||
caplog: pytest.LogCaptureFixture,
|
||
) -> None:
|
||
"""#902: a 404 logs a WARNING naming the slug + 'dataset-path invalid?' hint."""
|
||
import logging
|
||
|
||
import httpx
|
||
|
||
from app.services.sber_index import pull_sber_indices
|
||
|
||
fake_db = _ExecuteCapture()
|
||
|
||
async def _mock_fetch_404( # type: ignore[override]
|
||
client: object, *, dashboard: SberDashboard, ref_area: str
|
||
):
|
||
request = httpx.Request("POST", "https://sberindex.ru/api/sowa")
|
||
response = httpx.Response(404, request=request)
|
||
raise httpx.HTTPStatusError("Not Found", request=request, response=response)
|
||
yield # pragma: no cover — makes this an async generator
|
||
|
||
with patch("app.services.sber_index.fetch_sber_index", side_effect=_mock_fetch_404):
|
||
with caplog.at_level(logging.WARNING, logger="app.services.sber_index"):
|
||
result = await pull_sber_indices(
|
||
fake_db, # type: ignore[arg-type]
|
||
cities={"643": "Россия"},
|
||
dashboards=[_DASH["residential_real_estate_prices"]],
|
||
)
|
||
|
||
assert result["errors"] == 1
|
||
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
|
||
assert any(
|
||
"residential_real_estate_prices" in r.message and "dataset-path invalid" in r.message
|
||
for r in warnings
|
||
), f"Expected a 404 dataset-path warning, got: {[r.message for r in warnings]}"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_pull_sber_indices_asking_benchmark_logged(caplog: pytest.LogCaptureFixture) -> None:
|
||
"""After upserting dinamika-tsen-obyavlenii, a benchmark INFO line is logged."""
|
||
import logging
|
||
|
||
from app.services.sber_index import pull_sber_indices
|
||
|
||
fake_db = _ExecuteCapture()
|
||
|
||
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=[_DASH["dinamika-tsen-obyavlenii"]],
|
||
)
|
||
|
||
assert any(
|
||
"benchmark" in record.message for record in caplog.records
|
||
), "Expected a benchmark log line after upserting dinamika-tsen-obyavlenii"
|
||
assert any(
|
||
"115000" in record.message for record in caplog.records
|
||
), "Benchmark log should include the latest asking index value"
|