"""Tests for sber_index service — pure unit tests, no network/DB. Coverage: (a) build_sber_route: base64 route builder produces correct encoded string. (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). """ from __future__ import annotations import base64 import json import os import sys from datetime import date 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 _parse_period_month, build_sber_route, decode_sber_response, ) # --------------------------------------------------------------------------- # (a) Route builder # --------------------------------------------------------------------------- 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") 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.""" encoded = build_sber_route("dinamika-tsen-obyavlenii", "643") 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") # 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.""" encoded = build_sber_route("real_estate_deals", "66") 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" # --------------------------------------------------------------------------- # (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) # --------------------------------------------------------------------------- # (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 @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(client: object, *, dashboard: str, ref_area: str): # type: ignore[override] 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=["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(client: object, *, dashboard: str, ref_area: str): # type: ignore[override] nonlocal call_count call_count += 1 if dashboard == "real_estate_deals": raise Exception("Simulated network failure") yield ("Россия", date(2024, 1, 1), "Вторичный", dashboard, 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"], ) # First series failed, second succeeded assert result["errors"] == 1 assert result["upserted"] == 1 @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(client: object, *, dashboard: str, ref_area: str): # type: ignore[override] yield ("Россия", date(2024, 6, 1), "Вторичный", dashboard, 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"], ) 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"