mirror of
https://github.com/escalante29/WealthySmart.git
synced 2026-07-17 10:08:46 +02:00
4ceb67564a
The n8n flows fail invisibly from the app's perspective (review UX-17, the top trust gap). /api/v1/sync-status derives last-received + cadence thresholds for BAC, salary, municipal, pensions and exchange rate from existing data — no n8n integration needed. New Sincronización page shows per-source freshness and a hint to check n8n; the sidebar item gets an amber dot when any source is stale. Verified live against the dev DB (correctly flags its stale BAC data at 43 days). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
113 lines
3.6 KiB
Python
113 lines
3.6 KiB
Python
"""Sync-status derivation (UX-17) and CSV export (ARCH-18)."""
|
|
|
|
import csv
|
|
import io
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
|
|
from app.models.models import (
|
|
Bank,
|
|
Category,
|
|
Currency,
|
|
ExchangeRate,
|
|
Transaction,
|
|
TransactionSource,
|
|
TransactionType,
|
|
)
|
|
from app.services.csv_export import build_transactions_csv
|
|
from app.services.sync_status import compute_sync_status
|
|
from app.timeutil import utcnow
|
|
|
|
|
|
def by_key(statuses: list[dict]) -> dict[str, dict]:
|
|
return {s["key"]: s for s in statuses}
|
|
|
|
|
|
class TestSyncStatus:
|
|
def test_empty_db_reports_never(self, session):
|
|
statuses = by_key(compute_sync_status(session))
|
|
assert len(statuses) == 5
|
|
assert all(s["status"] == "never" for s in statuses.values())
|
|
assert all(s["last_received"] is None for s in statuses.values())
|
|
|
|
def test_fresh_data_is_ok_and_stale_warns(self, session):
|
|
fresh = Transaction(
|
|
amount=Decimal("1"),
|
|
merchant="X",
|
|
date=utcnow(),
|
|
source=TransactionSource.CREDIT_CARD,
|
|
bank=Bank.BAC,
|
|
)
|
|
fresh.created_at = utcnow() - timedelta(days=1)
|
|
stale_salary = Transaction(
|
|
amount=Decimal("1"),
|
|
merchant="EMPLOYER",
|
|
date=utcnow(),
|
|
transaction_type=TransactionType.SALARY,
|
|
source=TransactionSource.TRANSFER,
|
|
)
|
|
stale_salary.created_at = utcnow() - timedelta(days=60)
|
|
session.add(fresh)
|
|
session.add(stale_salary)
|
|
session.add(
|
|
ExchangeRate(
|
|
date=utcnow(),
|
|
buy_rate=Decimal("500"),
|
|
sell_rate=Decimal("510"),
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
statuses = by_key(compute_sync_status(session))
|
|
assert statuses["bac_credit_card"]["status"] == "ok"
|
|
assert statuses["bac_credit_card"]["age_days"] == 1.0
|
|
assert statuses["salary"]["status"] == "warning"
|
|
assert statuses["exchange_rate"]["status"] == "ok"
|
|
assert statuses["municipal"]["status"] == "never"
|
|
|
|
|
|
class TestCsvExport:
|
|
def _seed(self, session):
|
|
cat = Category(name="Food")
|
|
session.add(cat)
|
|
session.commit()
|
|
session.refresh(cat)
|
|
session.add(
|
|
Transaction(
|
|
amount=Decimal("1500.50"),
|
|
merchant="SODA, LA \"BUENA\"", # exercises CSV quoting
|
|
date=datetime(2026, 4, 2),
|
|
category_id=cat.id,
|
|
currency=Currency.CRC,
|
|
)
|
|
)
|
|
session.add(
|
|
Transaction(
|
|
amount=Decimal("10"),
|
|
merchant="CASHTHING",
|
|
date=datetime(2026, 4, 3),
|
|
source=TransactionSource.CASH,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
def test_full_export_with_quoting_and_category(self, session):
|
|
self._seed(session)
|
|
text = build_transactions_csv(session)
|
|
rows = list(csv.reader(io.StringIO(text)))
|
|
assert rows[0][:3] == ["id", "date", "merchant"]
|
|
assert len(rows) == 3 # header + 2
|
|
newest_first = rows[1]
|
|
assert newest_first[2] == "CASHTHING"
|
|
soda = rows[2]
|
|
assert soda[2] == 'SODA, LA "BUENA"' # round-trips through quoting
|
|
assert soda[4] == "1500.50"
|
|
assert soda[9] == "Food"
|
|
|
|
def test_source_filter(self, session):
|
|
self._seed(session)
|
|
text = build_transactions_csv(session, source=TransactionSource.CASH)
|
|
rows = list(csv.reader(io.StringIO(text)))
|
|
assert len(rows) == 2
|
|
assert rows[1][2] == "CASHTHING"
|