Files
fin-tracker/backend/tests/sources/test_zenmoney.py
T
Dmitry c55fe19e48 feat(sources): контракт источников, worker и синк ZenMoney + ЦБ
Source.sync(ctx) -> SyncResult пишет только raw_* и возвращает курсор; локи,
журнал, ошибки и продвижение курсора берёт на себя worker/runner.

ZenMoney читается единственным доступным способом — POST /v8/diff/ по
serverTimestamp; токен живёт сутки, поэтому worker ротирует refresh_token через
source_credential. Маппер всегда пересобирает core из полных raw_*, так что
удаление в ZenMoney исчезает и у нас.

ЦБ ходит мимо прокси (trust_env=False) и отдаёт cp1251 с делением на Nominal.
Курсы только по рабочим дням — протяжку по календарю делает аналитика.

Планировщик — APScheduler в отдельном процессе, на источник advisory-лок
sync:<name>, чтобы ручной запуск не пересёкся с плановым.
2026-09-18 13:43:49 +03:00

518 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""ZenMoney source: the diff loop, the raw tier, the core mapping and both auth modes."""
from __future__ import annotations
import json
from datetime import UTC, datetime, timedelta
from decimal import Decimal
import httpx
import pytest
from sqlalchemy import func, select
from fintracker.config import Settings
from fintracker.db import get_sessionmaker
from fintracker.models import (
Account,
AccountKind,
AccountRole,
CashTxn,
CashTxnTag,
Category,
FlowType,
Merchant,
RawZenmoneyDeletion,
RawZenmoneyEntity,
SourceCredential,
SyncState,
)
from fintracker.sources.zenmoney.client import (
DIFF_URL,
TOKEN_URL,
ZenmoneyAuthError,
)
from fintracker.sources.zenmoney.sync import ZenmoneySource
CARD = "aaaaaaaa-0000-0000-0000-000000000001"
USD_CASH = "aaaaaaaa-0000-0000-0000-000000000002"
DEPOSIT = "aaaaaaaa-0000-0000-0000-000000000003"
DEBT = "aaaaaaaa-0000-0000-0000-000000000009"
TAG_FOOD = "bbbbbbbb-0000-0000-0000-000000000001"
TAG_CAFE = "bbbbbbbb-0000-0000-0000-000000000002"
TXN_EXPENSE = "dddddddd-0000-0000-0000-000000000001"
TXN_INCOME = "dddddddd-0000-0000-0000-000000000002"
TXN_TRANSFER = "dddddddd-0000-0000-0000-000000000003"
TXN_USD = "dddddddd-0000-0000-0000-000000000004"
TXN_DELETED = "dddddddd-0000-0000-0000-000000000005"
TXN_HOLD = "dddddddd-0000-0000-0000-000000000006"
def static_settings() -> Settings:
return Settings(
zenmoney_token="static-access-token",
zenmoney_client_id=None,
zenmoney_client_secret=None,
zenmoney_refresh_token=None,
)
def oauth_settings() -> Settings:
return Settings(
zenmoney_token=None,
zenmoney_client_id="client-id",
zenmoney_client_secret="client-secret",
zenmoney_refresh_token="seed-refresh-token",
)
async def counts() -> dict[str, int]:
async with get_sessionmaker()() as session:
out = {}
for name, model in (
("raw", RawZenmoneyEntity),
("accounts", Account),
("categories", Category),
("merchants", Merchant),
("transactions", CashTxn),
("tags", CashTxnTag),
):
out[name] = (
await session.execute(select(func.count()).select_from(model))
).scalar_one()
return out
async def account_by_source_id(source_id: str) -> Account:
async with get_sessionmaker()() as session:
return (
await session.execute(select(Account).where(Account.source_id == source_id))
).scalar_one()
async def txn_by_source_id(source_id: str) -> CashTxn:
async with get_sessionmaker()() as session:
return (
await session.execute(select(CashTxn).where(CashTxn.source_id == source_id))
).scalar_one()
async def test_first_sync_asks_for_everything_and_maps_core_rows(
app, mock_http, fixture_json, run_sync
):
diff = fixture_json("zenmoney", "diff_full.json")
route = mock_http.post(DIFF_URL).mock(return_value=httpx.Response(200, json=diff))
result = await run_sync(ZenmoneySource(), settings=static_settings())
body = json.loads(route.calls.last.request.content)
assert body["serverTimestamp"] == 0
assert "transaction" in body["forceFetch"] and "account" in body["forceFetch"]
assert body["currentClientTimestamp"] > 1_700_000_000
assert route.calls.last.request.headers["authorization"] == "Bearer static-access-token"
assert result.cursor_after == "1750000000"
assert result.changed is True
assert (
result.counts["raw_upserted"] == 18
) # 2 instruments + company + user + 4 acc + 3 tags + 1 merchant + 6 txns
assert result.counts["transactions"] == 6
assert result.counts["accounts"] == 4
assert result.counts["categories"] == 3
rows = await counts()
assert rows["accounts"] == 4
assert rows["categories"] == 3
assert rows["merchants"] == 1
assert rows["transactions"] == 6
assert rows["tags"] == 3 # 2 on the expense, 1 on the income
async def test_account_mapping(app, mock_http, fixture_json, run_sync):
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json"))
)
await run_sync(ZenmoneySource(), settings=static_settings())
card = await account_by_source_id(CARD)
assert card.kind is AccountKind.zm_card # ZenMoney says "ccard"
assert card.role is AccountRole.liquid
assert card.currency == "RUB"
assert card.include_in_net_worth is True
assert card.balance == Decimal("25000.5")
assert card.credit_limit == Decimal("100000")
assert card.balance_as_of is not None
assert card.deposit_terms is None
usd = await account_by_source_id(USD_CASH)
assert usd.currency == "USD"
assert usd.kind is AccountKind.zm_cash
deposit = await account_by_source_id(DEPOSIT)
assert deposit.role is AccountRole.savings
assert deposit.deposit_terms is not None
assert deposit.deposit_terms["percent"] == 16.5
assert deposit.opened_at is not None and deposit.opened_at.isoformat() == "2025-02-01"
debt = await account_by_source_id(DEBT)
assert debt.kind is AccountKind.zm_debt
assert debt.role is AccountRole.debt
assert debt.include_in_net_worth is False # inBalance = false on the system debt account
async def test_category_tree_and_transaction_mapping(app, mock_http, fixture_json, run_sync):
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json"))
)
await run_sync(ZenmoneySource(), settings=static_settings())
async with get_sessionmaker()() as session:
food = (
await session.execute(select(Category).where(Category.source_id == TAG_FOOD))
).scalar_one()
cafe = (
await session.execute(select(Category).where(Category.source_id == TAG_CAFE))
).scalar_one()
assert food.parent_id is None
assert cafe.parent_id == food.id
assert food.color == 4294198070
expense = (
await session.execute(select(CashTxn).where(CashTxn.source_id == TXN_EXPENSE))
).scalar_one()
tags = (
(
await session.execute(
select(CashTxnTag)
.where(CashTxnTag.txn_id == expense.id)
.order_by(CashTxnTag.ord)
)
)
.scalars()
.all()
)
assert expense.flow_type is FlowType.expense
assert expense.outcome == Decimal("350.4")
assert expense.income == Decimal(0)
assert expense.outcome_currency == "RUB"
assert expense.mcc == 5812
assert expense.payee == "COFFEE HOUSE"
assert expense.original_payee == "COFFEE HOUSE MOSCOW"
assert expense.merchant_id is not None
assert expense.ts == datetime.fromtimestamp(1749500000, UTC)
assert expense.date.isoformat() == "2025-09-01"
assert expense.meta == {"latitude": 55.75, "longitude": 37.61}
# primary category is the FIRST tag as ZenMoney ordered it, not the parent
assert [t.category_id for t in tags] == [cafe.id, food.id]
assert expense.primary_category_id == cafe.id
assert expense.category_id == cafe.id
income = await txn_by_source_id(TXN_INCOME)
assert income.flow_type is FlowType.income
assert income.income == Decimal("180000")
transfer = await txn_by_source_id(TXN_TRANSFER)
assert transfer.flow_type is FlowType.internal_transfer
assert transfer.income_account_id != transfer.outcome_account_id
assert transfer.income == Decimal("50000") and transfer.outcome == Decimal("50000")
usd = await txn_by_source_id(TXN_USD)
assert usd.outcome_currency == "USD"
assert usd.outcome == Decimal("19.99")
assert usd.op_outcome == Decimal("1850.5")
assert usd.op_outcome_currency == "RUB"
deleted = await txn_by_source_id(TXN_DELETED)
assert deleted.deleted is True
assert deleted.flow_type is FlowType.deleted
hold = await txn_by_source_id(TXN_HOLD)
assert hold.hold is True
assert hold.flow_type is FlowType.expense
async def test_second_sync_with_empty_diff_changes_nothing(app, mock_http, fixture_json, run_sync):
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json"))
)
first = await run_sync(ZenmoneySource(), settings=static_settings())
before = await counts()
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_empty.json"))
)
second = await run_sync(ZenmoneySource(), settings=static_settings(), cursor=first.cursor_after)
assert second.changed is False
assert second.cursor_after == "1750000500"
assert second.counts == {"raw_upserted": 0, "deleted": 0}
assert await counts() == before
async def test_rerunning_the_same_diff_is_idempotent(app, mock_http, fixture_json, run_sync):
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json"))
)
await run_sync(ZenmoneySource(), settings=static_settings())
before = await counts()
await run_sync(ZenmoneySource(), settings=static_settings())
assert await counts() == before
async def test_deletion_flags_the_core_row_and_keeps_the_fact(
app, mock_http, fixture_json, run_sync
):
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json"))
)
first = await run_sync(ZenmoneySource(), settings=static_settings())
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_deletion.json"))
)
third = await run_sync(ZenmoneySource(), settings=static_settings(), cursor=first.cursor_after)
assert third.changed is True
assert third.counts["deleted"] == 1
assert third.cursor_after == "1750001000"
gone = await txn_by_source_id(TXN_HOLD)
assert gone.deleted is True
assert gone.flow_type is FlowType.deleted
card = await account_by_source_id(CARD)
# the upstream rename to "Карта Т-Банк" is NOT applied: `name` is user-owned after the
# first insert (see test_sync_does_not_overwrite_user_edits_of_an_account)
assert card.name == "Карта Тинькофф"
assert card.balance == Decimal("24000.5")
async with get_sessionmaker()() as session:
raw = await session.get(RawZenmoneyEntity, {"entity_type": "transaction", "id": TXN_HOLD})
assert raw is None
deletion = await session.get(
RawZenmoneyDeletion, {"entity_type": "transaction", "id": TXN_HOLD}
)
assert deletion is not None and deletion.stamp == 1750000950
# the transaction count is unchanged: a deleted row is flagged, never dropped
assert (await counts())["transactions"] == 6
async def test_cursor_is_stored_by_the_worker(app, mock_http, fixture_json, monkeypatch):
from fintracker.worker import runner
monkeypatch.setattr(runner, "get_settings", static_settings)
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json"))
)
run = await runner.run_source("zenmoney", triggered_by="cli")
assert run.status.value == "ok", run.error
assert run.cursor_after == "1750000000"
async with get_sessionmaker()() as session:
state = await session.get(SyncState, "zenmoney")
assert state is not None and state.cursor == "1750000000"
# --- auth -----------------------------------------------------------------------------
async def test_static_token_401_tells_the_user_to_renew_it(app, mock_http, fixture_json, run_sync):
mock_http.post(DIFF_URL).mock(return_value=httpx.Response(401, json={"error": "nope"}))
with pytest.raises(ZenmoneyAuthError) as exc:
await run_sync(ZenmoneySource(), settings=static_settings())
assert "ZENMONEY_TOKEN" in str(exc.value)
async def test_missing_static_token_is_a_clear_error(app, run_sync):
settings = Settings(
zenmoney_token=None,
zenmoney_client_id=None,
zenmoney_client_secret=None,
zenmoney_refresh_token=None,
)
with pytest.raises(ZenmoneyAuthError) as exc:
await run_sync(ZenmoneySource(), settings=settings)
assert "ZENMONEY_TOKEN is not set" in str(exc.value)
async def test_oauth_seeds_from_settings_and_stores_the_new_pair(
app, mock_http, fixture_json, run_sync
):
token_route = mock_http.post(TOKEN_URL).mock(
return_value=httpx.Response(
200,
json={
"access_token": "fresh-access",
"refresh_token": "fresh-refresh",
"expires_in": 86400,
"token_type": "bearer",
},
)
)
diff_route = mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_empty.json"))
)
await run_sync(ZenmoneySource(), settings=oauth_settings())
form = dict(httpx.QueryParams(token_route.calls.last.request.content.decode()))
assert form["grant_type"] == "refresh_token"
assert form["refresh_token"] == "seed-refresh-token"
assert form["client_id"] == "client-id"
assert form["client_secret"] == "client-secret"
assert diff_route.calls.last.request.headers["authorization"] == "Bearer fresh-access"
async with get_sessionmaker()() as session:
cred = await session.get(SourceCredential, "zenmoney")
assert cred is not None
assert cred.payload["access_token"] == "fresh-access"
assert cred.payload["refresh_token"] == "fresh-refresh"
assert cred.payload["expires_at"] > datetime.now(UTC).isoformat()
async def test_oauth_refreshes_only_when_expired(app, mock_http, fixture_json, run_sync):
async with get_sessionmaker()() as session:
session.add(
SourceCredential(
source="zenmoney",
payload={
"access_token": "still-good",
"refresh_token": "stored-refresh",
"expires_at": (datetime.now(UTC) + timedelta(hours=5)).isoformat(),
},
)
)
await session.commit()
token_route = mock_http.post(TOKEN_URL).mock(return_value=httpx.Response(200, json={}))
diff_route = mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_empty.json"))
)
await run_sync(ZenmoneySource(), settings=oauth_settings())
assert not token_route.called
assert diff_route.calls.last.request.headers["authorization"] == "Bearer still-good"
async def test_oauth_refreshes_an_expired_token_and_retries_a_401(
app, mock_http, fixture_json, run_sync
):
async with get_sessionmaker()() as session:
session.add(
SourceCredential(
source="zenmoney",
payload={
"access_token": "expired-access",
"refresh_token": "stored-refresh",
"expires_at": (datetime.now(UTC) - timedelta(minutes=1)).isoformat(),
},
)
)
await session.commit()
token_route = mock_http.post(TOKEN_URL).mock(
return_value=httpx.Response(
200,
json={
"access_token": "rotated-access",
"refresh_token": "rotated-refresh",
"expires_in": 86400,
},
)
)
diff_route = mock_http.post(DIFF_URL).mock(
side_effect=[
httpx.Response(401, json={"error": "expired"}),
httpx.Response(200, json=fixture_json("zenmoney", "diff_empty.json")),
]
)
result = await run_sync(ZenmoneySource(), settings=oauth_settings())
assert result.cursor_after == "1750000500"
assert token_route.call_count == 2 # once because expired, once after the 401
assert diff_route.call_count == 2
assert diff_route.calls.last.request.headers["authorization"] == "Bearer rotated-access"
async with get_sessionmaker()() as session:
cred = await session.get(SourceCredential, "zenmoney")
assert cred is not None and cred.payload["refresh_token"] == "rotated-refresh"
# --- user-owned account fields --------------------------------------------------------
async def test_sync_does_not_overwrite_user_edits_of_an_account(
app, client, auth_headers, mock_http, fixture_json, run_sync
):
"""`name`, `role` and `include_in_net_worth` belong to the user (PATCH /accounts/{id});
a later sync may only refresh what the source owns, such as the balance."""
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json"))
)
first = await run_sync(ZenmoneySource(), settings=static_settings())
card = await account_by_source_id(CARD)
assert card.name == "Карта Тинькофф"
patched = await client.patch(
f"/api/v1/accounts/{card.id}",
headers=auth_headers,
json={"name": "Основная карта", "role": "savings", "include_in_net_worth": False},
)
assert patched.status_code == 200, patched.text
# the same account comes back renamed and re-balanced upstream
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_deletion.json"))
)
await run_sync(ZenmoneySource(), settings=static_settings(), cursor=first.cursor_after)
card = await account_by_source_id(CARD)
assert card.name == "Основная карта"
assert card.role is AccountRole.savings
assert card.include_in_net_worth is False
assert card.balance == Decimal("24000.5") # source-owned: still updated
assert card.archived is False
async def test_archived_upstream_still_leaves_net_worth(
app, client, auth_headers, mock_http, fixture_json, run_sync
):
"""The one exception to user ownership: an account the source archived cannot stay in
net worth, the same way a deleted one does not."""
mock_http.post(DIFF_URL).mock(
return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json"))
)
first = await run_sync(ZenmoneySource(), settings=static_settings())
card = await account_by_source_id(CARD)
assert card.include_in_net_worth is True
patched = await client.patch(
f"/api/v1/accounts/{card.id}", headers=auth_headers, json={"role": "savings"}
)
assert patched.status_code == 200, patched.text
archived_diff = fixture_json("zenmoney", "diff_deletion.json")
archived_diff["account"][0]["archive"] = True
mock_http.post(DIFF_URL).mock(return_value=httpx.Response(200, json=archived_diff))
await run_sync(ZenmoneySource(), settings=static_settings(), cursor=first.cursor_after)
card = await account_by_source_id(CARD)
assert card.archived is True
assert card.include_in_net_worth is False
assert card.role is AccountRole.savings # still the user's choice
async def test_instrument_id_zero_is_a_valid_currency(app):
from fintracker.sources.zenmoney.mapper import _ccy, instrument_codes
codes = instrument_codes([{"id": 0, "shortTitle": "RUB"}, {"id": 2, "shortTitle": "USD"}])
assert _ccy(codes, 0) == "RUB"
assert _ccy(codes, 2) == "USD"
assert _ccy(codes, None) is None
assert _ccy(codes, 99) is None