"""Tests for cbr_macro service — pure unit tests, no network/DB. Coverage: (a) parse_ru_period: русские имена месяцев → date(y, m, 1), включая мусор. (b) parse_mortgage_sheet: unpivot мини-листа (реальные значения Свердловской обл., замерены вживую 08.09.2026: 11, 10.64, 10.98, 10.84, 9.2, 11.49). (c) parse_key_rate_response: разбор реального SOAP-ответа KeyRate (namespace-агностично). (d) build_key_rate_envelope: SOAP-конверт содержит нужные теги/значения. (e) upsert SQL — статические asserts: ON CONFLICT, CAST(:x AS type), нет `::`-кастов, fetched_at отсутствует в DO UPDATE SET. (f) load_cbr_mortgage_series / load_key_rate — httpx.MockTransport round-trip. (g) db.commit() отсутствует в исходниках сервисных функций (коммитит caller). """ from __future__ import annotations import inspect import os import re from datetime import date, datetime from unittest.mock import MagicMock # DATABASE_URL required by config before any app import. os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") import httpx import openpyxl import pytest from app.services import cbr_macro # --------------------------------------------------------------------------- # parse_ru_period # --------------------------------------------------------------------------- class TestParseRuPeriod: def test_january(self) -> None: assert cbr_macro.parse_ru_period("Январь 2019") == date(2019, 1, 1) def test_december(self) -> None: assert cbr_macro.parse_ru_period("Декабрь 2026") == date(2026, 12, 1) def test_july_2026(self) -> None: assert cbr_macro.parse_ru_period("Июль 2026") == date(2026, 7, 1) def test_unknown_month(self) -> None: with pytest.raises(ValueError, match="месяца"): cbr_macro.parse_ru_period("Смогтябрь 2019") def test_garbage_no_year(self) -> None: with pytest.raises(ValueError, match="период"): cbr_macro.parse_ru_period("Январь") def test_bad_year(self) -> None: with pytest.raises(ValueError, match="период"): cbr_macro.parse_ru_period("Январь двадцать") # --------------------------------------------------------------------------- # parse_mortgage_sheet — unpivot # --------------------------------------------------------------------------- # Мини-лист по форме реального 02_13_Rates_mortgage.xlsx («ставка в рублях»): # строка 1 пустая, строка 2 заголовок, строка 3 — шапка периодов, строки 4+ территории. # Последние 6 значений Свердловской обл. — реальные, замерены 08.09.2026. _RATE_SHEET_ROWS: list[tuple] = [ (None, None, None, None, None, None, None, None), ( "Средневзвешенная ставка по ипотечным жилищным кредитам, %", None, None, None, None, None, None, None, ), (None, "Февраль 2026", "Март 2026", "Апрель 2026", "Май 2026", "Июнь 2026", "Июль 2026", None), ("РОССИЙСКАЯ ФЕДЕРАЦИЯ", 21.5, 21.1, 20.8, 19.9, 18.6, 17.9, None), ("Свердловская область", 11, 10.64, 10.98, 10.84, 9.2, 11.49, None), ] class TestParseMortgageSheet: def test_too_short_returns_empty(self) -> None: assert cbr_macro.parse_mortgage_sheet([(1,), (2,)], series_slug="rate_rub") == [] def test_unpivot_counts(self) -> None: rows = cbr_macro.parse_mortgage_sheet(_RATE_SHEET_ROWS, series_slug="rate_rub") # 2 территории × 6 месяцев (последний столбец None — trailing empty col пропущен) assert len(rows) == 12 def test_sverdlovsk_values(self) -> None: rows = cbr_macro.parse_mortgage_sheet(_RATE_SHEET_ROWS, series_slug="rate_rub") sverdlovsk = [r for r in rows if r.region == "Свердловская область"] sverdlovsk.sort(key=lambda r: r.period_month) values = [r.value for r in sverdlovsk] assert values == [11.0, 10.64, 10.98, 10.84, 9.2, 11.49] assert sverdlovsk[0].period_month == date(2026, 2, 1) assert sverdlovsk[-1].period_month == date(2026, 7, 1) assert all(r.series == "rate_rub" for r in sverdlovsk) def test_skips_empty_region_rows(self) -> None: rows_in = [*_RATE_SHEET_ROWS, (None, 1, 2, 3, 4, 5, 6, None), ("", 1, 2, 3, 4, 5, 6, None)] rows = cbr_macro.parse_mortgage_sheet(rows_in, series_slug="rate_rub") assert len(rows) == 12 # мусорные строки без региона не добавляют записей def test_skips_non_numeric_cell(self) -> None: bad_rows = [*_RATE_SHEET_ROWS[:-1], ("Москва", "n/a", 10.0, None, None, None, None, None)] rows = cbr_macro.parse_mortgage_sheet(bad_rows, series_slug="rate_rub") moscow = [r for r in rows if r.region == "Москва"] assert len(moscow) == 1 # только числовая ячейка попала в результат assert moscow[0].value == 10.0 # --------------------------------------------------------------------------- # SOAP key rate # --------------------------------------------------------------------------- _KEY_RATE_XML = """
2024-01-01T00:00:00+03:00
16
2024-07-29T00:00:00+03:00
18 """ class TestKeyRateSoap: def test_parse_real_response_shape(self) -> None: rows = cbr_macro.parse_key_rate_response(_KEY_RATE_XML) assert rows == [(date(2024, 1, 1), 16.0), (date(2024, 7, 29), 18.0)] def test_parse_empty_response(self) -> None: empty = ( "" "" ) assert cbr_macro.parse_key_rate_response(empty) == [] def test_build_envelope_contains_dates_and_soapaction_target(self) -> None: envelope = cbr_macro.build_key_rate_envelope(date(2024, 1, 1), date(2024, 12, 31)) assert "2024-01-01" in envelope assert "2024-12-31" in envelope assert "http://web.cbr.ru/" in envelope assert "" in envelope # --------------------------------------------------------------------------- # Upsert SQL — статические asserts (без БД) # --------------------------------------------------------------------------- class TestUpsertSql: def test_mortgage_sql_no_double_colon_cast(self) -> None: sql = re.sub(r"\s+", " ", str(cbr_macro._UPSERT_MORTGAGE_SQL.text)) assert not re.search(r":\w+::", sql) def test_mortgage_sql_has_cast_and_on_conflict(self) -> None: sql = re.sub(r"\s+", " ", str(cbr_macro._UPSERT_MORTGAGE_SQL.text)) assert "CAST(:region AS text)" in sql assert "CAST(:period_month AS date)" in sql assert "CAST(:value AS double precision)" in sql assert "ON CONFLICT (region, period_month, series)" in sql assert "DO UPDATE SET" in sql def test_mortgage_sql_does_not_touch_fetched_at_on_conflict(self) -> None: sql = str(cbr_macro._UPSERT_MORTGAGE_SQL.text) do_update_part = sql.split("DO UPDATE SET", 1)[1] # fetched_at упоминается только в SQL-комментарии (объясняет почему НЕ обновляем), # но не как цель присваивания. assert not re.search(r"fetched_at\s*=", do_update_part) def test_key_rate_sql_no_double_colon_cast(self) -> None: sql = re.sub(r"\s+", " ", str(cbr_macro._UPSERT_KEY_RATE_SQL.text)) assert not re.search(r":\w+::", sql) def test_key_rate_sql_shape(self) -> None: sql = re.sub(r"\s+", " ", str(cbr_macro._UPSERT_KEY_RATE_SQL.text)) assert "CAST(:rate_date AS date)" in sql assert "CAST(:rate AS double precision)" in sql assert "ON CONFLICT (rate_date)" in sql def test_key_rate_sql_does_not_touch_fetched_at_on_conflict(self) -> None: sql = str(cbr_macro._UPSERT_KEY_RATE_SQL.text) do_update_part = sql.split("DO UPDATE SET", 1)[1] assert not re.search(r"fetched_at\s*=", do_update_part) # --------------------------------------------------------------------------- # Services do not commit — caller commits # --------------------------------------------------------------------------- class TestNoCommitInService: @pytest.mark.parametrize( "fn", [ cbr_macro.load_cbr_mortgage_series, cbr_macro.pull_cbr_mortgage, cbr_macro.load_key_rate, cbr_macro._upsert_mortgage_rows, cbr_macro._upsert_key_rate_rows, ], ) def test_no_commit_call(self, fn) -> None: src = inspect.getsource(fn) assert "db.commit()" not in src # --------------------------------------------------------------------------- # Network round-trip via httpx.MockTransport # --------------------------------------------------------------------------- def _xlsx_bytes_for(rows: list[tuple], *, sheet_name: str) -> bytes: import io wb = openpyxl.Workbook() ws = wb.active ws.title = sheet_name for row in rows: ws.append(row) buf = io.BytesIO() wb.save(buf) return buf.getvalue() class TestMortgageNetworkRoundtrip: def test_load_series_upserts_expected_rows(self) -> None: series = cbr_macro.CbrMortgageSeries( "rate_rub", "https://www.cbr.ru/fake.xlsx", "ставка в рублях" ) xlsx_bytes = _xlsx_bytes_for(_RATE_SHEET_ROWS, sheet_name="ставка в рублях") def handler(request: httpx.Request) -> httpx.Response: assert str(request.url) == series.url return httpx.Response(200, content=xlsx_bytes) client = httpx.Client(transport=httpx.MockTransport(handler)) db = MagicMock() db.execute.return_value.rowcount = 1 result = cbr_macro.load_cbr_mortgage_series(db, series, client=client) assert result["rows"] == 12 assert result["upserted"] == 12 assert db.execute.call_count == 12 def test_load_series_dry_run_skips_upsert(self) -> None: series = cbr_macro.CbrMortgageSeries( "rate_rub", "https://www.cbr.ru/fake.xlsx", "ставка в рублях" ) xlsx_bytes = _xlsx_bytes_for(_RATE_SHEET_ROWS, sheet_name="ставка в рублях") def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, content=xlsx_bytes) client = httpx.Client(transport=httpx.MockTransport(handler)) db = MagicMock() result = cbr_macro.load_cbr_mortgage_series(db, series, client=client, dry_run=True) assert result["rows"] == 12 assert result["upserted"] == 0 db.execute.assert_not_called() def test_missing_sheet_raises(self) -> None: series = cbr_macro.CbrMortgageSeries( "rate_rub", "https://www.cbr.ru/fake.xlsx", "нет такого листа" ) xlsx_bytes = _xlsx_bytes_for(_RATE_SHEET_ROWS, sheet_name="ставка в рублях") def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, content=xlsx_bytes) client = httpx.Client(transport=httpx.MockTransport(handler)) db = MagicMock() with pytest.raises(ValueError, match="лист"): cbr_macro.load_cbr_mortgage_series(db, series, client=client) def test_pull_isolates_series_failure(self) -> None: good = cbr_macro.CbrMortgageSeries( "rate_rub", "https://www.cbr.ru/good.xlsx", "ставка в рублях" ) bad = cbr_macro.CbrMortgageSeries("debt_rub", "https://www.cbr.ru/bad.xlsx", "в рублях") xlsx_bytes = _xlsx_bytes_for(_RATE_SHEET_ROWS, sheet_name="ставка в рублях") def handler(request: httpx.Request) -> httpx.Response: if str(request.url) == bad.url: return httpx.Response(500) return httpx.Response(200, content=xlsx_bytes) client = httpx.Client(transport=httpx.MockTransport(handler)) db = MagicMock() db.execute.return_value.rowcount = 1 counters = cbr_macro.pull_cbr_mortgage(db, series_list=[good, bad], client=client) assert counters["upserted"] == 12 assert counters["errors"] == 1 db.rollback.assert_called_once() class TestKeyRateNetworkRoundtrip: def test_load_key_rate_upserts_parsed_rows(self) -> None: def handler(request: httpx.Request) -> httpx.Response: assert request.method == "POST" assert request.headers["soapaction"] == "http://web.cbr.ru/KeyRate" return httpx.Response(200, content=_KEY_RATE_XML.encode("utf-8")) client = httpx.Client(transport=httpx.MockTransport(handler)) db = MagicMock() db.execute.return_value.rowcount = 1 result = cbr_macro.load_key_rate( db, from_date=date(2024, 1, 1), to_date=date(2024, 12, 31), client=client ) assert result["rows"] == 2 assert result["upserted"] == 2 assert db.execute.call_count == 2 def test_parse_mortgage_sheet_debt_layout_datetime_header() -> None: """02_14_Debt_mortgage.xlsx: шапка в строке 2 и периоды — datetime, не «Январь 2019». Реальная раскладка (проверено на живом файле 08.09.2026): в 02_11/02_13 строка 1 пустая, описание в строке 2, шапка в строке 3 текстом; в 02_14 описание в строке 1, а шапка уже в строке 2 и хранит настоящие даты. Захардкоженный индекс строки давал по серии задолженности ноль строк — молча, только с warning'ами. """ months = [ datetime(2019, 2, 1), datetime(2019, 3, 1), datetime(2019, 4, 1), datetime(2019, 5, 1), datetime(2019, 6, 1), datetime(2019, 7, 1), ] rows: list[tuple[object, ...]] = [ ("Задолженность по ипотечным жилищным кредитам", None, None, None, None, None, None), (None, *months), ("РОССИЙСКАЯ ФЕДЕРАЦИЯ", 6605685, 6719335, 6823490, 6937855, 7018162, 7095485), ("Свердловская область", 100, 101, 102, 103, 104, 105), ] out = cbr_macro.parse_mortgage_sheet(rows, series_slug="debt_rub") sverdlovsk = [r for r in out if r.region == "Свердловская область"] assert len(sverdlovsk) == 6 assert sverdlovsk[0].period_month == date(2019, 2, 1) assert sverdlovsk[0].value == 100.0 assert {r.region for r in out} == {"РОССИЙСКАЯ ФЕДЕРАЦИЯ", "Свердловская область"} def test_parse_mortgage_sheet_returns_empty_when_no_header_found() -> None: """Нет распознаваемой шапки — пустой результат и предупреждение, а не мусор.""" rows: list[tuple[object, ...]] = [ ("Какой-то заголовок", None, None), ("РОССИЙСКАЯ ФЕДЕРАЦИЯ", 1, 2), ] assert cbr_macro.parse_mortgage_sheet(rows, series_slug="debt_rub") == []