Dmitry
3b3ee4d682
POST /metrics/refresh ставит задачу в sync_job (source=METRICS_JOB) и отвечает 202, пересчёт делает worker; GET /metrics/status отдаёт refreshing и consistent. metric_refresh_log хранит failed_step и step_timings. Источники с needs="tinvest_token" не попадают в расписание без токена, tinvest/moex добавлены в default_schedule. Воркер трогает heartbeat-файл для healthcheck.
179 lines
7.4 KiB
Python
179 lines
7.4 KiB
Python
from datetime import timedelta
|
|
|
|
from sqlalchemy import select
|
|
|
|
from factories import make_account, make_category, make_txn, month_back
|
|
from fintracker.analytics import today_local
|
|
from fintracker.models import METRICS_JOB, AccountRole, JobStatus, SyncJob
|
|
from fintracker.worker.scheduler import _process_queued_jobs
|
|
|
|
|
|
async def _queue_and_run(client, auth_headers):
|
|
"""POST /metrics/refresh, then let the worker take the job, as it does every 5 s."""
|
|
r = await client.post("/api/v1/metrics/refresh", headers=auth_headers)
|
|
assert r.status_code == 202, r.text
|
|
await _process_queued_jobs()
|
|
return r.json()
|
|
|
|
|
|
async def test_refresh_populates_every_metric_endpoint(client, auth_headers):
|
|
card = await make_account(name="Карта", balance="200000")
|
|
await make_account(name="Вклад", balance="100000", role=AccountRole.savings)
|
|
await make_account(name="Кредитка", balance="-5000", role=AccountRole.debt)
|
|
food = await make_category("Еда")
|
|
m = month_back(1)
|
|
await make_txn(m + timedelta(days=1), income="150000", income_account_id=card)
|
|
await make_txn(
|
|
m + timedelta(days=2),
|
|
outcome="30000",
|
|
outcome_account_id=card,
|
|
primary_category_id=food,
|
|
)
|
|
await make_txn(today_local() - timedelta(days=1), outcome="500", outcome_account_id=card)
|
|
|
|
before = (await client.get("/api/v1/metrics/status", headers=auth_headers)).json()
|
|
assert before == {"last_refresh": None, "consistent": True, "refreshing": False}
|
|
|
|
job = await _queue_and_run(client, auth_headers)
|
|
assert job["status"] == "queued"
|
|
|
|
status = (await client.get("/api/v1/metrics/status", headers=auth_headers)).json()
|
|
assert status["last_refresh"]["trigger"] == "manual"
|
|
assert status["last_refresh"]["error"] is None
|
|
assert status["last_refresh"]["finished_at"] is not None
|
|
assert status["last_refresh"]["failed_step"] is None
|
|
assert status["last_refresh"]["step_timings"] # every step is timed
|
|
assert status["consistent"] is True
|
|
assert status["refreshing"] is False
|
|
|
|
series = (await client.get("/api/v1/networth/series", headers=auth_headers)).json()
|
|
assert series
|
|
assert series[-1]["d"] == str(today_local())
|
|
assert series[-1]["total_rub"].startswith("295000")
|
|
assert isinstance(series[-1]["by_currency"]["RUB"], str)
|
|
|
|
breakdown = (await client.get("/api/v1/networth/breakdown", headers=auth_headers)).json()
|
|
assert breakdown["d"] == str(today_local())
|
|
assert breakdown["debt_rub"].startswith("-5000")
|
|
assert {a["name"] for a in breakdown["accounts"]} == {"Карта", "Вклад", "Кредитка"}
|
|
assert all(isinstance(a["balance_rub"], str) for a in breakdown["accounts"])
|
|
|
|
monthly = (await client.get("/api/v1/cashflow/monthly", headers=auth_headers)).json()
|
|
last_month = next(row for row in monthly if row["month"] == str(m))
|
|
assert last_month["income_rub"].startswith("150000")
|
|
assert last_month["expense_rub"].startswith("30000")
|
|
assert last_month["savings_rate"].startswith("0.8")
|
|
|
|
spending = (
|
|
await client.get(
|
|
"/api/v1/spending/categories",
|
|
headers=auth_headers,
|
|
params={"month": m.strftime("%Y-%m")},
|
|
)
|
|
).json()
|
|
assert spending[0]["category_name"] == "Еда"
|
|
assert spending[0]["root_category_name"] == "Еда"
|
|
assert spending[0]["amount_rub"].startswith("30000")
|
|
|
|
runway = (await client.get("/api/v1/runway", headers=auth_headers)).json()
|
|
assert runway["liquid_reserve_rub"].startswith("300000")
|
|
# 30000 baseline in the previous month, 0 in the two before it -> 10000 average
|
|
assert runway["avg_baseline_3m_rub"].startswith("10000")
|
|
assert runway["runway_months"].startswith("30")
|
|
|
|
quality = (await client.get("/api/v1/data-quality", headers=auth_headers)).json()
|
|
assert isinstance(quality, list)
|
|
|
|
|
|
async def test_spending_rejects_a_bad_month(client, auth_headers):
|
|
r = await client.get(
|
|
"/api/v1/spending/categories", headers=auth_headers, params={"month": "2026/01"}
|
|
)
|
|
assert r.status_code == 400
|
|
r = await client.get(
|
|
"/api/v1/spending/categories", headers=auth_headers, params={"month": "2026-13"}
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
async def test_empty_database_reports_no_transactions(client, auth_headers):
|
|
await _queue_and_run(client, auth_headers)
|
|
quality = (await client.get("/api/v1/data-quality", headers=auth_headers)).json()
|
|
assert [row["check_name"] for row in quality] == ["no_transactions"]
|
|
assert (await client.get("/api/v1/runway", headers=auth_headers)).json()[
|
|
"runway_months"
|
|
] is None
|
|
assert (await client.get("/api/v1/networth/series", headers=auth_headers)).json() == []
|
|
assert (await client.get("/api/v1/spending/categories", headers=auth_headers)).json() == []
|
|
|
|
|
|
async def test_refresh_returns_before_the_work_is_done(client, auth_headers):
|
|
"""The 202 is honest: nothing has been rebuilt until the worker picks the job up."""
|
|
r = await client.post("/api/v1/metrics/refresh", headers=auth_headers)
|
|
assert r.status_code == 202
|
|
assert r.json()["source"] == METRICS_JOB
|
|
|
|
status = (await client.get("/api/v1/metrics/status", headers=auth_headers)).json()
|
|
assert status["last_refresh"] is None
|
|
assert status["refreshing"] is True
|
|
|
|
await _process_queued_jobs()
|
|
|
|
status = (await client.get("/api/v1/metrics/status", headers=auth_headers)).json()
|
|
assert status["refreshing"] is False
|
|
assert status["last_refresh"] is not None
|
|
|
|
|
|
async def test_a_waiting_refresh_is_shared_not_duplicated(client, auth_headers):
|
|
first = (await client.post("/api/v1/metrics/refresh", headers=auth_headers)).json()
|
|
second = (await client.post("/api/v1/metrics/refresh", headers=auth_headers)).json()
|
|
assert first["id"] == second["id"]
|
|
|
|
from fintracker.db import get_sessionmaker
|
|
|
|
async with get_sessionmaker()() as session:
|
|
jobs = (
|
|
(await session.execute(select(SyncJob).where(SyncJob.source == METRICS_JOB)))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
assert len(jobs) == 1
|
|
|
|
|
|
async def test_a_refresh_that_fails_part_way_is_reported_as_inconsistent(
|
|
client, auth_headers, monkeypatch
|
|
):
|
|
from fintracker.db import get_sessionmaker
|
|
from fintracker.metrics import refresh as refresh_module
|
|
|
|
async def fine(session):
|
|
pass
|
|
|
|
async def boom(session):
|
|
raise RuntimeError("step blew up")
|
|
|
|
monkeypatch.setattr("fintracker.analytics.register_steps", lambda: None)
|
|
monkeypatch.setattr(refresh_module, "STEPS", [("fine", fine), ("boom", boom), ("never", fine)])
|
|
|
|
await _queue_and_run(client, auth_headers)
|
|
|
|
status = (await client.get("/api/v1/metrics/status", headers=auth_headers)).json()
|
|
last = status["last_refresh"]
|
|
assert last["failed_step"] == "boom"
|
|
assert "step blew up" in last["error"]
|
|
assert list(last["step_timings"]) == ["fine"] # 'never' did not run, 'boom' did not finish
|
|
assert status["consistent"] is False
|
|
|
|
async with get_sessionmaker()() as session:
|
|
job = (
|
|
await session.execute(select(SyncJob).where(SyncJob.source == METRICS_JOB))
|
|
).scalar_one()
|
|
assert job.status == JobStatus.error
|
|
|
|
# a later clean run makes the tables one snapshot again
|
|
monkeypatch.setattr(refresh_module, "STEPS", [("fine", fine)])
|
|
await _queue_and_run(client, auth_headers)
|
|
status = (await client.get("/api/v1/metrics/status", headers=auth_headers)).json()
|
|
assert status["consistent"] is True
|
|
assert status["last_refresh"]["failed_step"] is None
|