from datetime import date import pytest from finbot.config import Settings from finbot.models import Budget, Expense from finbot.store import Store TODAY = date(2026, 8, 15) @pytest.fixture def store(tmp_path): settings = Settings( data_dir=str(tmp_path), hf_dataset_repo="", # sync off: no network in tests hf_token="", telegram_bot_token="", telegram_webhook_secret="", allowed_telegram_user_ids="", ) store = Store(settings) yield store store.close() def make(amount=25000, category="food_dining", day=TODAY, merchant="", batch=""): return Expense( user_id=1, amount_minor=amount, currency="INR", description="coffee", category=category, occurred_on=day, merchant=merchant, batch_id=batch, ) class TestExpenses: def test_add_and_read_back(self, store): expense = store.add_expense(make()) rows = store.list_expenses(1) assert len(rows) == 1 assert rows[0].id == expense.id assert rows[0].amount_minor == 25000 assert rows[0].occurred_on == TODAY def test_isolated_per_user(self, store): store.add_expense(make()) other = make() other.user_id = 2 store.add_expense(other) assert len(store.list_expenses(1)) == 1 assert len(store.list_expenses(2)) == 1 def test_date_filtering(self, store): store.add_expense(make(day=date(2026, 8, 1))) store.add_expense(make(day=date(2026, 7, 1))) rows = store.list_expenses(1, date(2026, 8, 1), date(2026, 8, 31)) assert len(rows) == 1 def test_category_filtering(self, store): store.add_expense(make(category="rent")) store.add_expense(make(category="food_dining")) assert len(store.list_expenses(1, category="rent")) == 1 def test_limit(self, store): for _ in range(5): store.add_expense(make()) assert len(store.list_expenses(1, limit=3)) == 3 def test_last_expense(self, store): store.add_expense(make(amount=100)) latest = store.add_expense(make(amount=999)) assert store.last_expense(1).id == latest.id def test_last_expense_none_when_empty(self, store): assert store.last_expense(1) is None def test_delete(self, store): expense = store.add_expense(make()) assert store.delete_expense(1, expense.id) is True assert store.list_expenses(1) == [] def test_delete_wrong_user_is_a_no_op(self, store): expense = store.add_expense(make()) assert store.delete_expense(999, expense.id) is False assert len(store.list_expenses(1)) == 1 def test_delete_batch(self, store): for _ in range(3): store.add_expense(make(batch="batch-1")) store.add_expense(make(batch="")) assert store.delete_batch(1, "batch-1") == 3 assert len(store.list_expenses(1)) == 1 def test_delete_batch_ignores_empty_id(self, store): store.add_expense(make(batch="")) assert store.delete_batch(1, "") == 0 def test_count(self, store): store.add_expense(make()) store.add_expense(make()) assert store.count(1) == 2 assert store.count() == 2 def test_currencies_used(self, store): store.add_expense(make()) usd = make() usd.currency = "USD" store.add_expense(usd) assert set(store.currencies_used(1)) == {"INR", "USD"} class TestDedupe: def test_matching_transactions_share_a_key(self, store): a = make(merchant="Swiggy") b = make(merchant="swiggy!") # different case and punctuation assert a.dedupe_key == b.dedupe_key def test_key_changes_with_amount(self, store): assert make(amount=100).dedupe_key != make(amount=200).dedupe_key def test_key_changes_with_date(self, store): assert make(day=date(2026, 8, 1)).dedupe_key != make(day=date(2026, 8, 2)).dedupe_key def test_existing_keys_lookup(self, store): expense = store.add_expense(make(merchant="Swiggy")) found = store.existing_dedupe_keys(1, [expense.dedupe_key, "nope"]) assert found == {expense.dedupe_key} def test_existing_keys_empty_input(self, store): assert store.existing_dedupe_keys(1, []) == set() def test_duplicates_are_still_insertable(self, store): # Two identical coffees on one day are both real -- dedupe is advisory # at import time, not a database constraint. store.add_expense(make()) store.add_expense(make()) assert store.count(1) == 2 def test_chunks_large_key_lists(self, store): expense = store.add_expense(make()) keys = [f"k{i}" for i in range(1200)] + [expense.dedupe_key] assert store.existing_dedupe_keys(1, keys) == {expense.dedupe_key} class TestBudgets: def test_set_and_get(self, store): store.set_budget(Budget(1, "food_dining", 100000, "INR")) budgets = store.get_budgets(1) assert len(budgets) == 1 assert budgets[0].amount_minor == 100000 def test_upsert_replaces(self, store): store.set_budget(Budget(1, "food_dining", 100000, "INR")) store.set_budget(Budget(1, "food_dining", 250000, "INR")) budgets = store.get_budgets(1) assert len(budgets) == 1 assert budgets[0].amount_minor == 250000 def test_delete(self, store): store.set_budget(Budget(1, "rent", 100, "INR")) assert store.delete_budget(1, "rent") is True assert store.get_budgets(1) == [] def test_delete_missing(self, store): assert store.delete_budget(1, "rent") is False class TestPersistence: def test_survives_reopen(self, tmp_path): settings = Settings( data_dir=str(tmp_path), hf_dataset_repo="", hf_token="", telegram_bot_token="", telegram_webhook_secret="", ) first = Store(settings) first.add_expense(make()) first.close() second = Store(settings) assert second.count(1) == 1 second.close() def test_sync_disabled_without_config(self, store): assert store.sync_status == "disabled" class TestSyncMisconfiguration: """A dataset repo with no token looks configured and persists nothing.""" def _store(self, tmp_path, **overrides): base = dict( data_dir=str(tmp_path), hf_dataset_repo="", hf_token="", telegram_bot_token="", telegram_webhook_secret="", ) base.update(overrides) return Store(Settings(**base)) def test_repo_without_token_is_flagged(self, tmp_path): store = self._store(tmp_path, hf_dataset_repo="me/data") assert "MISCONFIGURED" in store.sync_status store.close() def test_missing_required_names_the_token(self, tmp_path): gaps = Settings( data_dir=str(tmp_path), hf_dataset_repo="me/data", hf_token="" ).missing_required() assert any("HF_TOKEN" in g and "STILL lost" in g for g in gaps) def test_no_repo_reports_the_repo_not_the_token(self, tmp_path): gaps = Settings( data_dir=str(tmp_path), hf_dataset_repo="", hf_token="" ).missing_required() assert any("HF_DATASET_REPO" in g for g in gaps) assert not any("HF_TOKEN" in g for g in gaps) def test_neither_set_is_plain_local(self, tmp_path): store = self._store(tmp_path) assert store.sync_status == "disabled" store.close()