bot-backend
5170107405
Some checks failed
CI / backend-tests (push) Has been cancelled
CI / frontend-tests (push) Has been cancelled
CI / openapi-codegen-check (push) Has been cancelled
CI / changes (push) Has been cancelled
CI / changes (pull_request) Successful in 6s
CI / backend-tests (pull_request) Has been cancelled
CI / frontend-tests (pull_request) Has been cancelled
CI / openapi-codegen-check (pull_request) Has been cancelled
Adds period_type TEXT NOT NULL DEFAULT 'unknown' to macro_indicator and
widens the PRIMARY KEY to (indicator_type, region, obs_date, period_type).
Before: yearly aggregate ('год'→obs_date YYYY-01-01) and Q1 ('I квартал'→
same date) shared the same PK slot → ON CONFLICT DO UPDATE made them
overwrite each other despite the in-memory dedup fix in #1687.
After: each EmissRow carries period_type from _emiss_period_granularity
('year'/'quarter'/'month'); DB ON CONFLICT targets include it, so
yearly+Q1 rows genuinely coexist.
Non-EMISS sources (CBR, rosstat open-data, domrf) use period_type='unknown'
(literal in SQL) — they are disambiguated by obs_date already, so the
wider PK is backward-compatible. All ON CONFLICT clauses in
cbr_macro_sync.py and rosstat_macro_sync.py updated accordingly.
Migration: 163_emiss_pk_period_type.sql (idempotent BEGIN/COMMIT).
Tests: 49 passed (emiss + rosstat suite), ruff clean, py_compile OK.
216 lines
8.5 KiB
Python
216 lines
8.5 KiB
Python
"""Тесты Celery-таски cbr_macro_sync (mock session + mock fetch, без живой сети).
|
||
|
||
Покрывают: контракт upsert-SQL обоих рядов (key_rate + inflation_yoy, region='rf',
|
||
source='cbr', ON CONFLICT, CAST not ::), счётчики fetched/upserted (пер-ряд),
|
||
per-series guard (сбой одного ряда не мешает другому, но пробрасывается).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
|
||
def _make_mock_db() -> tuple[MagicMock, list[dict[str, Any]]]:
|
||
"""Mock DB session с поддержкой db.begin_nested() context manager.
|
||
|
||
Возвращает (db, captured_params) — captured содержит params каждого
|
||
db.execute(..., params).
|
||
"""
|
||
captured: list[dict[str, Any]] = []
|
||
|
||
def execute_router(stmt: Any, params: Any = None) -> MagicMock:
|
||
if params is not None:
|
||
captured.append(params)
|
||
return MagicMock()
|
||
|
||
db = MagicMock()
|
||
db.execute = execute_router
|
||
# begin_nested() → context manager (SAVEPOINT)
|
||
db.begin_nested.return_value.__enter__ = MagicMock(return_value=None)
|
||
db.begin_nested.return_value.__exit__ = MagicMock(return_value=False)
|
||
return db, captured
|
||
|
||
|
||
def test_upsert_sql_contract() -> None:
|
||
"""Smoke: UPSERT_KEY_RATE_SQL содержит обязательные литералы контракта
|
||
macro_indicator и НЕ использует :x::type (psycopg v3)."""
|
||
from app.workers.tasks.cbr_macro_sync import UPSERT_KEY_RATE_SQL
|
||
|
||
sql = str(UPSERT_KEY_RATE_SQL)
|
||
upper = sql.upper()
|
||
|
||
assert "INTO MACRO_INDICATOR" in upper
|
||
# литералы контракта — в SQL они lowercase
|
||
assert "'key_rate'" in sql
|
||
assert "'rf'" in sql
|
||
assert "'cbr'" in sql
|
||
assert "'daily'" in sql
|
||
# ON CONFLICT по новому PK включает period_type (migration 163)
|
||
assert "ON CONFLICT (INDICATOR_TYPE, REGION, OBS_DATE, PERIOD_TYPE) DO UPDATE" in upper
|
||
assert "VALUE = EXCLUDED.VALUE" in upper
|
||
assert "UPDATED_AT = NOW()" in upper
|
||
# psycopg v3: CAST(:x AS type), НИКОГДА :x::type
|
||
assert "CAST(:D AS DATE)" in upper
|
||
assert "CAST(:V AS NUMERIC)" in upper
|
||
assert "::" not in sql
|
||
# CBR-строки используют period_type='unknown' (литерал в SQL)
|
||
assert "'unknown'" in sql
|
||
|
||
|
||
def test_task_upserts_each_row_with_correct_params() -> None:
|
||
"""Каждая (date, rate) → один execute с bind-параметрами :d и :v."""
|
||
from app.workers.tasks import cbr_macro_sync as task_mod
|
||
|
||
db, captured = _make_mock_db()
|
||
fake_rows = [
|
||
(date(2023, 8, 14), Decimal("8.5")),
|
||
(date(2023, 8, 15), Decimal("12.0")),
|
||
]
|
||
|
||
with (
|
||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||
patch.object(task_mod, "fetch_key_rate", return_value=fake_rows) as mock_fetch,
|
||
patch.object(task_mod, "fetch_inflation", return_value=[]) as mock_infl,
|
||
):
|
||
result = task_mod.cbr_macro_sync.run()
|
||
|
||
mock_fetch.assert_called_once()
|
||
mock_infl.assert_called_once()
|
||
assert result["fetched"] == 2
|
||
assert result["upserted"] == 2
|
||
assert result["key_rate"] == {"fetched": 2, "upserted": 2}
|
||
assert result["inflation"] == {"fetched": 0, "upserted": 0}
|
||
# параметры каждого upsert
|
||
assert captured == [
|
||
{"d": date(2023, 8, 14), "v": Decimal("8.5")},
|
||
{"d": date(2023, 8, 15), "v": Decimal("12.0")},
|
||
]
|
||
db.commit.assert_called()
|
||
db.close.assert_called_once()
|
||
|
||
|
||
def test_task_passes_date_range_to_fetch() -> None:
|
||
"""from_date / to_date (ISO-строки) парсятся в date и прокидываются в fetch."""
|
||
from app.workers.tasks import cbr_macro_sync as task_mod
|
||
|
||
db, _ = _make_mock_db()
|
||
|
||
with (
|
||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||
patch.object(task_mod, "fetch_key_rate", return_value=[]) as mock_fetch,
|
||
patch.object(task_mod, "fetch_inflation", return_value=[]) as mock_infl,
|
||
):
|
||
result = task_mod.cbr_macro_sync.run(from_date="2020-01-01", to_date="2020-12-31")
|
||
|
||
mock_fetch.assert_called_once_with(from_date=date(2020, 1, 1), to_date=date(2020, 12, 31))
|
||
# период прокидывается и в inflation fetch
|
||
mock_infl.assert_called_once_with(from_date=date(2020, 1, 1), to_date=date(2020, 12, 31))
|
||
assert result["from_date"] == "2020-01-01"
|
||
assert result["to_date"] == "2020-12-31"
|
||
assert result["fetched"] == 0
|
||
assert result["upserted"] == 0
|
||
|
||
|
||
def test_task_reraises_fetch_failure() -> None:
|
||
"""Ошибка fetch'а ставки пробрасывается (surfaces в Celery/GlitchTip), не глотается.
|
||
DB-сессия при этом закрывается."""
|
||
from app.workers.tasks import cbr_macro_sync as task_mod
|
||
|
||
db, _ = _make_mock_db()
|
||
|
||
with (
|
||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||
patch.object(task_mod, "fetch_key_rate", side_effect=RuntimeError("CBR down")),
|
||
patch.object(task_mod, "fetch_inflation", return_value=[]),
|
||
):
|
||
try:
|
||
task_mod.cbr_macro_sync.run()
|
||
raise AssertionError("expected RuntimeError to propagate")
|
||
except RuntimeError as e:
|
||
assert "CBR down" in str(e)
|
||
|
||
db.close.assert_called_once()
|
||
|
||
|
||
def test_inflation_failure_does_not_block_key_rate() -> None:
|
||
"""Per-series guard: сбой инфляции НЕ мешает ставке апсертнуться, но пробрасывается."""
|
||
from app.workers.tasks import cbr_macro_sync as task_mod
|
||
|
||
db, captured = _make_mock_db()
|
||
kr_rows = [(date(2024, 1, 1), Decimal("16.0"))]
|
||
|
||
with (
|
||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||
patch.object(task_mod, "fetch_key_rate", return_value=kr_rows),
|
||
patch.object(task_mod, "fetch_inflation", side_effect=RuntimeError("infl down")),
|
||
):
|
||
try:
|
||
task_mod.cbr_macro_sync.run()
|
||
raise AssertionError("expected RuntimeError to propagate")
|
||
except RuntimeError as e:
|
||
assert "infl down" in str(e)
|
||
|
||
# key_rate всё равно апсертнулся ДО проброса inflation-ошибки
|
||
assert captured == [{"d": date(2024, 1, 1), "v": Decimal("16.0")}]
|
||
db.close.assert_called_once()
|
||
|
||
|
||
def test_task_upserts_inflation_rows_with_correct_params() -> None:
|
||
"""Каждая (month1st, %) инфляции → execute UPSERT_INFLATION_SQL с :d и :v."""
|
||
from app.workers.tasks import cbr_macro_sync as task_mod
|
||
|
||
db, captured = _make_mock_db()
|
||
infl_rows = [
|
||
(date(2026, 3, 1), Decimal("5.86")),
|
||
(date(2026, 4, 1), Decimal("5.58")),
|
||
]
|
||
|
||
with (
|
||
patch.object(task_mod, "SessionLocal", return_value=db),
|
||
patch.object(task_mod, "fetch_key_rate", return_value=[]),
|
||
patch.object(task_mod, "fetch_inflation", return_value=infl_rows),
|
||
):
|
||
result = task_mod.cbr_macro_sync.run()
|
||
|
||
assert result["inflation"] == {"fetched": 2, "upserted": 2}
|
||
assert captured == [
|
||
{"d": date(2026, 3, 1), "v": Decimal("5.86")},
|
||
{"d": date(2026, 4, 1), "v": Decimal("5.58")},
|
||
]
|
||
|
||
|
||
def test_upsert_inflation_sql_contract() -> None:
|
||
"""UPSERT_INFLATION_SQL: indicator_type='inflation_yoy', monthly, CAST not ::."""
|
||
from app.workers.tasks.cbr_macro_sync import UPSERT_INFLATION_SQL
|
||
|
||
sql = str(UPSERT_INFLATION_SQL)
|
||
upper = sql.upper()
|
||
|
||
assert "INTO MACRO_INDICATOR" in upper
|
||
assert "'inflation_yoy'" in sql
|
||
assert "'rf'" in sql
|
||
assert "'cbr'" in sql
|
||
assert "'monthly'" in sql
|
||
assert "ON CONFLICT (INDICATOR_TYPE, REGION, OBS_DATE, PERIOD_TYPE) DO UPDATE" in upper
|
||
assert "VALUE = EXCLUDED.VALUE" in upper
|
||
assert "UPDATED_AT = NOW()" in upper
|
||
# psycopg v3: CAST(:x AS type), НИКОГДА :x::type
|
||
assert "CAST(:D AS DATE)" in upper
|
||
assert "CAST(:V AS NUMERIC)" in upper
|
||
assert "::" not in sql
|
||
assert "'unknown'" in sql
|
||
|
||
|
||
def test_task_registered_in_beat_weekly() -> None:
|
||
"""Задача зарегистрирована в beat как еженедельная (понедельник)."""
|
||
from app.workers.beat_schedule import build_beat_schedule
|
||
|
||
schedule = build_beat_schedule()
|
||
assert "cbr-macro-sync-weekly" in schedule
|
||
entry = schedule["cbr-macro-sync-weekly"]
|
||
assert entry["task"] == "tasks.cbr_macro_sync.cbr_macro_sync"
|
||
# crontab day_of_week = понедельник (1)
|
||
assert 1 in entry["schedule"].day_of_week
|