771 lines
30 KiB
Python
771 lines
30 KiB
Python
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
from app.database import Database
|
|
from app.database import FamilyJoinError
|
|
from app.categories import category_label
|
|
from app.domain import Category, TransactionKind
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transaction_lifecycle_and_totals(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10,
|
|
chat_id=10,
|
|
username="tester",
|
|
full_name="Test User",
|
|
timezone="UTC",
|
|
report_time="21:00",
|
|
now=now,
|
|
)
|
|
await database.upsert_user(
|
|
user_id=20,
|
|
chat_id=20,
|
|
username=None,
|
|
full_name="Other User",
|
|
timezone="UTC",
|
|
report_time="21:00",
|
|
now=now,
|
|
)
|
|
|
|
expense = await database.add_transaction(
|
|
user_id=10,
|
|
amount_kopecks=30_000,
|
|
description="бензин",
|
|
kind=TransactionKind.EXPENSE,
|
|
category=Category.CAR,
|
|
occurred_at=now,
|
|
)
|
|
await database.add_transaction(
|
|
user_id=10,
|
|
amount_kopecks=100_000,
|
|
description="подработка",
|
|
kind=TransactionKind.INCOME,
|
|
category=Category.INCOME,
|
|
occurred_at=now,
|
|
)
|
|
|
|
totals = await database.totals(10, now - timedelta(hours=1), now + timedelta(hours=1))
|
|
assert totals.income == 100_000
|
|
assert totals.expense == 30_000
|
|
assert totals.balance == 70_000
|
|
assert await database.category_totals(
|
|
10, now - timedelta(hours=1), now + timedelta(hours=1)
|
|
) == {category_label(Category.CAR): 30_000}
|
|
assert await database.category_totals(10) == {category_label(Category.CAR): 30_000}
|
|
|
|
assert not await database.update_category(expense.id, 20, Category.OTHER)
|
|
assert not await database.delete_transaction(expense.id, 20)
|
|
assert await database.update_category(expense.id, 10, Category.OTHER)
|
|
assert (await database.get_transaction(expense.id, 10)).category is Category.OTHER
|
|
assert await database.delete_transaction(expense.id, 10)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_daily_report_idempotency(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 21, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10,
|
|
chat_id=10,
|
|
username=None,
|
|
full_name="Test User",
|
|
timezone="UTC",
|
|
report_time="21:00",
|
|
now=now,
|
|
)
|
|
assert not await database.report_was_sent(10, "2026-07-13")
|
|
await database.mark_report_sent(10, "2026-07-13", now)
|
|
await database.mark_report_sent(10, "2026-07-13", now)
|
|
assert await database.report_was_sent(10, "2026-07-13")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_family_budget_shares_history_and_new_transactions(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 20, tzinfo=ZoneInfo("UTC"))
|
|
for user_id, name in ((10, "Алексей"), (20, "Мария"), (30, "Посторонний")):
|
|
await database.upsert_user(
|
|
user_id=user_id,
|
|
chat_id=user_id,
|
|
username=None,
|
|
full_name=name,
|
|
timezone="UTC",
|
|
report_time="21:00",
|
|
now=now,
|
|
)
|
|
|
|
await database.add_transaction(
|
|
user_id=10,
|
|
amount_kopecks=30_000,
|
|
description="бензин",
|
|
kind=TransactionKind.EXPENSE,
|
|
category=Category.CAR,
|
|
occurred_at=now,
|
|
)
|
|
await database.add_transaction(
|
|
user_id=20,
|
|
amount_kopecks=10_000,
|
|
description="ярче",
|
|
kind=TransactionKind.EXPENSE,
|
|
category=Category.PRODUCTS,
|
|
occurred_at=now,
|
|
)
|
|
|
|
family = await database.create_family(10, now)
|
|
request = await database.request_family_join(20, family.invite_code.lower(), now)
|
|
assert request.owner_chat_id == 10
|
|
with pytest.raises(FamilyJoinError, match="обработан"):
|
|
await database.resolve_family_join(
|
|
owner_user_id=30,
|
|
candidate_user_id=20,
|
|
approve=True,
|
|
now=now,
|
|
)
|
|
approved, candidate_chat_id, joined = await database.resolve_family_join(
|
|
owner_user_id=10,
|
|
candidate_user_id=20,
|
|
approve=True,
|
|
now=now,
|
|
)
|
|
assert approved
|
|
assert candidate_chat_id == 20
|
|
assert joined is not None
|
|
assert joined.household_id == family.household_id
|
|
assert joined.members == ("Алексей", "Мария")
|
|
|
|
for user_id in (10, 20):
|
|
totals = await database.totals(user_id)
|
|
assert totals.expense == 40_000
|
|
assert await database.category_totals(user_id) == {
|
|
category_label(Category.CAR): 30_000,
|
|
category_label(Category.PRODUCTS): 10_000,
|
|
}
|
|
recent = await database.recent_transactions(user_id)
|
|
assert {item.author_name for item in recent} == {"Алексей", "Мария"}
|
|
|
|
recipients = await database.family_recipients(10)
|
|
assert [(row["user_id"], row["chat_id"]) for row in recipients] == [(20, 20)]
|
|
with pytest.raises(FamilyJoinError, match="создатель"):
|
|
await database.bind_family_chat(20, -100500, "Наш бюджет")
|
|
bound = await database.bind_family_chat(10, -100500, "Наш бюджет")
|
|
assert bound.notification_chat_id == -100500
|
|
assert await database.family_notification_chats(10) == [20, -100500]
|
|
assert await database.family_report_target(20) == (family.household_id, -100500)
|
|
assert not await database.family_report_was_sent(family.household_id, "2026-07-13")
|
|
await database.mark_family_report_sent(family.household_id, "2026-07-13", now)
|
|
assert await database.family_report_was_sent(family.household_id, "2026-07-13")
|
|
|
|
with pytest.raises(FamilyJoinError, match="создатель"):
|
|
await database.set_family_debt(20, 53_071_300, now)
|
|
await database.set_family_debt(10, 53_071_300, now)
|
|
salary = await database.add_transaction(
|
|
user_id=20,
|
|
amount_kopecks=40_000_000,
|
|
description="зарплата",
|
|
kind=TransactionKind.INCOME,
|
|
category=Category.INCOME,
|
|
occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
assert await database.family_debt(10) == 13_071_300
|
|
purchase = await database.add_transaction(
|
|
user_id=10,
|
|
amount_kopecks=2_000_000,
|
|
description="покупка",
|
|
kind=TransactionKind.EXPENSE,
|
|
category=Category.OTHER,
|
|
occurred_at=now + timedelta(seconds=2),
|
|
)
|
|
assert await database.family_debt(20) == 15_071_300
|
|
assert await database.delete_transaction(purchase.id, 10)
|
|
assert await database.family_debt(10) == 13_071_300
|
|
assert salary.author_name == "Мария"
|
|
|
|
categories = await database.budget_categories(10)
|
|
fuel = next(category for category in categories if category.system_key == "car_fuel")
|
|
assert fuel.parent_name == "Машина"
|
|
health = await database.create_budget_category(
|
|
user_id=20,
|
|
name="Здоровье",
|
|
parent_id=None,
|
|
now=now,
|
|
)
|
|
doctors = await database.create_budget_category(
|
|
user_id=10,
|
|
name="Врачи",
|
|
parent_id=health.id,
|
|
now=now,
|
|
)
|
|
other = await database.add_transaction(
|
|
user_id=10,
|
|
amount_kopecks=50_000,
|
|
description="консультация",
|
|
kind=TransactionKind.EXPENSE,
|
|
category=Category.OTHER,
|
|
occurred_at=now + timedelta(seconds=3),
|
|
)
|
|
assert await database.update_transaction_budget_category(other.id, 10, doctors.id)
|
|
moved = await database.get_transaction(other.id, 10)
|
|
assert moved.parent_category_name == "Здоровье"
|
|
assert moved.budget_category_name == "Врачи"
|
|
analytics = await database.category_totals(20)
|
|
assert analytics["📁 Здоровье → 📁 Врачи"] == 50_000
|
|
with pytest.raises(FamilyJoinError, match="два участника"):
|
|
await database.request_family_join(30, family.invite_code, now)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_database_is_migrated_for_family_budget(tmp_path):
|
|
import aiosqlite
|
|
|
|
path = tmp_path / "old.db"
|
|
async with aiosqlite.connect(path) as db:
|
|
await db.executescript(
|
|
"""
|
|
CREATE TABLE users (
|
|
user_id INTEGER PRIMARY KEY, chat_id INTEGER NOT NULL, username TEXT,
|
|
full_name TEXT NOT NULL, timezone TEXT NOT NULL, report_time TEXT NOT NULL,
|
|
created_at TEXT NOT NULL, updated_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE transactions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL,
|
|
amount_kopecks INTEGER NOT NULL, description TEXT NOT NULL,
|
|
kind TEXT NOT NULL, category TEXT NOT NULL,
|
|
occurred_at TEXT NOT NULL, created_at TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
await db.commit()
|
|
|
|
database = Database(path)
|
|
await database.initialize()
|
|
async with aiosqlite.connect(path) as db:
|
|
cursor = await db.execute("PRAGMA table_info(transactions)")
|
|
columns = {row[1] for row in await cursor.fetchall()}
|
|
assert "household_id" in columns
|
|
async with aiosqlite.connect(path) as db:
|
|
cursor = await db.execute("PRAGMA table_info(households)")
|
|
household_columns = {row[1] for row in await cursor.fetchall()}
|
|
assert {"debt_base_kopecks", "debt_started_at"} <= household_columns
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_migration_adds_debt_effect_and_kind_columns(tmp_path):
|
|
import aiosqlite
|
|
|
|
path = tmp_path / "old.db"
|
|
async with aiosqlite.connect(path) as db:
|
|
await db.executescript(
|
|
"""
|
|
CREATE TABLE users (
|
|
user_id INTEGER PRIMARY KEY, chat_id INTEGER NOT NULL, username TEXT,
|
|
full_name TEXT NOT NULL, timezone TEXT NOT NULL, report_time TEXT NOT NULL,
|
|
created_at TEXT NOT NULL, updated_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE transactions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL,
|
|
amount_kopecks INTEGER NOT NULL, description TEXT NOT NULL,
|
|
kind TEXT NOT NULL, category TEXT NOT NULL,
|
|
occurred_at TEXT NOT NULL, created_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE households (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT, invite_code TEXT NOT NULL UNIQUE,
|
|
created_by INTEGER NOT NULL, created_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE household_members (
|
|
household_id INTEGER NOT NULL, user_id INTEGER NOT NULL UNIQUE,
|
|
joined_at TEXT NOT NULL, PRIMARY KEY (household_id, user_id)
|
|
);
|
|
CREATE TABLE budget_categories (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT, household_id INTEGER NOT NULL,
|
|
name TEXT NOT NULL, icon TEXT NOT NULL DEFAULT '📁',
|
|
parent_id INTEGER, system_key TEXT, created_by INTEGER,
|
|
created_at TEXT NOT NULL, UNIQUE (household_id, system_key)
|
|
);
|
|
"""
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO users VALUES (1, 1, NULL, 'Test', 'UTC', '21:00', '2026-01-01', '2026-01-01')"
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO households(invite_code, created_by, created_at) VALUES ('FIN-T1', 1, '2026-01-01')"
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO household_members VALUES (1, 1, '2026-01-01')"
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO transactions(user_id, amount_kopecks, description, kind, category, occurred_at, created_at) "
|
|
"VALUES (1, 100000, 'бензин', 'expense', 'car', '2026-01-01', '2026-01-01')"
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO transactions(user_id, amount_kopecks, description, kind, category, occurred_at, created_at) "
|
|
"VALUES (1, 500000, 'зарплата', 'income', 'income', '2026-01-02', '2026-01-02')"
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO budget_categories(household_id, name, icon, system_key, created_at) "
|
|
"VALUES (1, 'Доход', '💚', 'income', '2026-01-01')"
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO budget_categories(household_id, name, icon, system_key, created_at) "
|
|
"VALUES (1, 'Машина', '🚗', 'car', '2026-01-01')"
|
|
)
|
|
await db.commit()
|
|
|
|
database = Database(path)
|
|
await database.initialize()
|
|
|
|
async with aiosqlite.connect(path) as db:
|
|
cursor = await db.execute("PRAGMA table_info(transactions)")
|
|
columns = {row[1] for row in await cursor.fetchall()}
|
|
assert "debt_effect" in columns
|
|
|
|
cursor = await db.execute("SELECT kind, debt_effect FROM transactions ORDER BY id")
|
|
rows = await cursor.fetchall()
|
|
assert rows[0] == ("expense", 1)
|
|
assert rows[1] == ("income", -1)
|
|
|
|
cursor = await db.execute("PRAGMA table_info(budget_categories)")
|
|
columns = {row[1] for row in await cursor.fetchall()}
|
|
assert "kind" in columns
|
|
|
|
cursor = await db.execute("SELECT system_key, kind FROM budget_categories ORDER BY id")
|
|
rows = await cursor.fetchall()
|
|
assert rows[0] == ("income", "income")
|
|
assert rows[1] == ("car", "expense")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_migration_is_idempotent(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
await database.initialize()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_salary_categories_seeded(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
categories = await database.budget_categories(10)
|
|
|
|
salary_root = next((c for c in categories if c.system_key == "salary"), None)
|
|
assert salary_root is not None
|
|
assert salary_root.name == "Зарплата"
|
|
assert salary_root.kind == "income"
|
|
assert salary_root.parent_id is None
|
|
|
|
salary_yana = next((c for c in categories if c.system_key == "salary_yana"), None)
|
|
assert salary_yana is not None
|
|
assert salary_yana.name == "Зарплата Яна"
|
|
assert salary_yana.kind == "income"
|
|
assert salary_yana.parent_id == salary_root.id
|
|
|
|
salary_vitya = next((c for c in categories if c.system_key == "salary_vitya"), None)
|
|
assert salary_vitya is not None
|
|
assert salary_vitya.name == "Зарплата Витя"
|
|
assert salary_vitya.kind == "income"
|
|
assert salary_vitya.parent_id == salary_root.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_salary_routing_owner_and_spouse(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.upsert_user(
|
|
user_id=20, chat_id=20, username=None, full_name="Яна",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
family = await database.create_family(10, now)
|
|
await database.request_family_join(20, family.invite_code, now)
|
|
await database.resolve_family_join(
|
|
owner_user_id=10, candidate_user_id=20, approve=True, now=now,
|
|
)
|
|
|
|
owner_cat_id = await database.resolve_salary_category(10)
|
|
assert owner_cat_id is not None
|
|
owner_cat = next(c for c in await database.budget_categories(10) if c.id == owner_cat_id)
|
|
assert owner_cat.system_key == "salary_vitya"
|
|
|
|
spouse_cat_id = await database.resolve_salary_category(20)
|
|
assert spouse_cat_id is not None
|
|
spouse_cat = next(c for c in await database.budget_categories(20) if c.id == spouse_cat_id)
|
|
assert spouse_cat.system_key == "salary_yana"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_salary_routing_no_household_returns_none(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Одинокий",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
assert await database.resolve_salary_category(10) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_negative_expense_decreases_debt(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
await database.set_family_debt(10, 100_000, now)
|
|
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=50_000, description="рабочий закуп",
|
|
kind=TransactionKind.EXPENSE, category=Category.OTHER,
|
|
debt_effect=-1, occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
assert tx.debt_effect == -1
|
|
assert await database.family_debt(10) == 50_000
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_positive_expense_increases_debt(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
await database.set_family_debt(10, 100_000, now)
|
|
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=30_000, description="бензин",
|
|
kind=TransactionKind.EXPENSE, category=Category.CAR,
|
|
debt_effect=1, occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
assert tx.debt_effect == 1
|
|
assert await database.family_debt(10) == 130_000
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_income_decreases_debt(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
await database.set_family_debt(10, 100_000, now)
|
|
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=50_000, description="зарплата",
|
|
kind=TransactionKind.INCOME, category=Category.INCOME,
|
|
debt_effect=-1, occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
assert tx.debt_effect == -1
|
|
assert await database.family_debt(10) == 50_000
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_recomputes_debt(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
await database.set_family_debt(10, 100_000, now)
|
|
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=30_000, description="бензин",
|
|
kind=TransactionKind.EXPENSE, category=Category.CAR,
|
|
debt_effect=1, occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
assert await database.family_debt(10) == 130_000
|
|
|
|
assert await database.delete_transaction(tx.id, 10)
|
|
assert await database.family_debt(10) == 100_000
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_negative_expense_restores_debt(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
await database.set_family_debt(10, 100_000, now)
|
|
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=30_000, description="рабочий закуп",
|
|
kind=TransactionKind.EXPENSE, category=Category.OTHER,
|
|
debt_effect=-1, occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
assert await database.family_debt(10) == 70_000
|
|
|
|
assert await database.delete_transaction(tx.id, 10)
|
|
assert await database.family_debt(10) == 100_000
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_budget_category_syncs_kind(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
await database.set_family_debt(10, 100_000, now)
|
|
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=30_000, description="бензин",
|
|
kind=TransactionKind.EXPENSE, category=Category.CAR,
|
|
debt_effect=1, occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
assert await database.family_debt(10) == 130_000
|
|
|
|
salary_cat = next(
|
|
c for c in await database.budget_categories(10)
|
|
if c.system_key == "salary_vitya"
|
|
)
|
|
changed = await database.update_transaction_budget_category(tx.id, 10, salary_cat.id)
|
|
assert changed
|
|
|
|
updated_tx = await database.get_transaction(tx.id, 10)
|
|
assert updated_tx.kind is TransactionKind.INCOME
|
|
assert updated_tx.budget_category_name == "Зарплата Витя"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_category_kind_and_inherited_subcategory(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
|
|
income_cat = await database.create_budget_category(
|
|
user_id=10, name="Инвестиции", parent_id=None, kind="income", now=now,
|
|
)
|
|
assert income_cat.kind == "income"
|
|
|
|
sub_cat = await database.create_budget_category(
|
|
user_id=10, name="Дивиденды", parent_id=income_cat.id, now=now,
|
|
)
|
|
assert sub_cat.kind == "income"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expense_main_category_defaults_to_expense_kind(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
|
|
cat = await database.create_budget_category(
|
|
user_id=10, name="Здоровье", parent_id=None, kind="expense", now=now,
|
|
)
|
|
assert cat.kind == "expense"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_category_change_income_to_expense_updates_debt(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
await database.set_family_debt(10, 100_000, now)
|
|
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=50_000, description="зарплата",
|
|
kind=TransactionKind.INCOME, category=Category.INCOME,
|
|
debt_effect=-1, occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
assert await database.family_debt(10) == 50_000
|
|
|
|
expense_cat = next(
|
|
c for c in await database.budget_categories(10)
|
|
if c.system_key == "car"
|
|
)
|
|
changed = await database.update_transaction_budget_category(tx.id, 10, expense_cat.id)
|
|
assert changed
|
|
|
|
updated_tx = await database.get_transaction(tx.id, 10)
|
|
assert updated_tx.kind is TransactionKind.EXPENSE
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_category_change_expense_to_income_updates_kind(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=30_000, description="покупка",
|
|
kind=TransactionKind.EXPENSE, category=Category.OTHER,
|
|
debt_effect=1, occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
|
|
salary_cat = next(
|
|
c for c in await database.budget_categories(10)
|
|
if c.system_key == "salary_vitya"
|
|
)
|
|
changed = await database.update_transaction_budget_category(tx.id, 10, salary_cat.id)
|
|
assert changed
|
|
|
|
updated_tx = await database.get_transaction(tx.id, 10)
|
|
assert updated_tx.kind is TransactionKind.INCOME
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_budget_category_kind_visible_in_list(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
|
|
categories = await database.budget_categories(10)
|
|
for cat in categories:
|
|
assert cat.kind in ("expense", "income")
|
|
|
|
salary_root = next(c for c in categories if c.system_key == "salary")
|
|
assert salary_root.kind == "income"
|
|
|
|
car = next(c for c in categories if c.system_key == "car")
|
|
assert car.kind == "expense"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_transaction_default_debt_effect_from_kind(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
|
|
exp = await database.add_transaction(
|
|
user_id=10, amount_kopecks=10_000, description="кофе",
|
|
kind=TransactionKind.EXPENSE, category=Category.PRODUCTS,
|
|
occurred_at=now,
|
|
)
|
|
assert exp.debt_effect == 1
|
|
|
|
inc = await database.add_transaction(
|
|
user_id=10, amount_kopecks=20_000, description="подработка",
|
|
kind=TransactionKind.INCOME, category=Category.INCOME,
|
|
occurred_at=now,
|
|
)
|
|
assert inc.debt_effect == -1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spouse_can_read_and_recategorize_family_transaction(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
for user_id, name in ((10, "Витя"), (20, "Яна"), (30, "Посторонний")):
|
|
await database.upsert_user(
|
|
user_id=user_id, chat_id=user_id, username=None, full_name=name,
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
family = await database.create_family(10, now)
|
|
await database.request_family_join(20, family.invite_code, now)
|
|
await database.resolve_family_join(
|
|
owner_user_id=10, candidate_user_id=20, approve=True, now=now,
|
|
)
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=10_000, description="покупка",
|
|
kind=TransactionKind.EXPENSE, category=Category.OTHER, occurred_at=now,
|
|
)
|
|
salary = next(
|
|
category for category in await database.budget_categories(20)
|
|
if category.system_key == "salary_yana"
|
|
)
|
|
|
|
spouse_view = await database.get_transaction(tx.id, 20)
|
|
assert spouse_view is not None
|
|
assert spouse_view.id == tx.id
|
|
assert await database.get_transaction(tx.id, 30) is None
|
|
assert not await database.update_transaction_budget_category(tx.id, 30, salary.id)
|
|
assert await database.update_transaction_budget_category(tx.id, 20, salary.id)
|
|
updated = await database.get_transaction(tx.id, 10)
|
|
assert updated is not None
|
|
assert updated.kind is TransactionKind.INCOME
|
|
assert updated.debt_effect == -1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_explicit_minus_keeps_reducing_debt_after_recategorization(tmp_path):
|
|
database = Database(tmp_path / "finance.db")
|
|
await database.initialize()
|
|
now = datetime(2026, 7, 13, 12, tzinfo=ZoneInfo("UTC"))
|
|
await database.upsert_user(
|
|
user_id=10, chat_id=10, username=None, full_name="Витя",
|
|
timezone="UTC", report_time="21:00", now=now,
|
|
)
|
|
await database.create_family(10, now)
|
|
await database.set_family_debt(10, 100_000, now)
|
|
tx = await database.add_transaction(
|
|
user_id=10, amount_kopecks=20_000, description="рабочий закуп",
|
|
kind=TransactionKind.EXPENSE, category=Category.OTHER,
|
|
debt_effect=-1, debt_effect_override=-1,
|
|
occurred_at=now + timedelta(seconds=1),
|
|
)
|
|
car = next(
|
|
category for category in await database.budget_categories(10)
|
|
if category.system_key == "car"
|
|
)
|
|
|
|
assert await database.update_transaction_budget_category(tx.id, 10, car.id)
|
|
updated = await database.get_transaction(tx.id, 10)
|
|
assert updated is not None
|
|
assert updated.kind is TransactionKind.EXPENSE
|
|
assert updated.debt_effect == -1
|
|
assert await database.family_debt(10) == 80_000
|