tests/test_dashboard.py

from datetime import timedelta
from types import SimpleNamespace

import pytest

from finbot.analytics import today_in
from finbot.config import Settings
from finbot.dashboard import (
    authenticate,
    chat_turn,
    category_frame,
    coach_reply,
    load_dashboard,
    recent_frame,
    render_budgets,
    render_header,
    render_kpis,
    render_recurring,
)
from finbot.ingest import Ingestor
from finbot.models import Budget, Expense
from finbot.store import Store


def make_settings(tmp_path, **overrides):
    base = dict(
        data_dir=str(tmp_path),
        hf_dataset_repo="",
        hf_token="",
        telegram_bot_token="",
        telegram_webhook_secret="sekrit",
        allowed_telegram_user_ids="",
        allowed_telegram_usernames="",
        default_currency="INR",
        timezone="Asia/Kolkata",
    )
    base.update(overrides)
    return Settings(**base)


@pytest.fixture
def services(tmp_path):
    settings = make_settings(tmp_path)
    store = Store(settings)
    advisor = SimpleNamespace(advise=lambda **kw: f"advice about: {kw.get('question')}")
    yield SimpleNamespace(store=store, settings=settings, advisor=advisor)
    store.close()


def add(store, amount, category="food_dining", day=None, merchant="", currency="INR"):
    store.add_expense(
        Expense(
            user_id=1,
            amount_minor=amount,
            currency=currency,
            description=merchant or "thing",
            category=category,
            occurred_on=day or today_in("Asia/Kolkata"),
            merchant=merchant,
        )
    )


class TestUserRegistry:
    def test_records_and_reads_back(self, services):
        services.store.record_user(1, "alice", "Alice")
        assert services.store.username_for(1) == "alice"
        assert services.store.display_name_for(1) == "@alice"

    def test_strips_the_at_sign(self, services):
        services.store.record_user(1, "@bob")
        assert services.store.username_for(1) == "bob"

    def test_rename_updates_in_place(self, services):
        services.store.record_user(1, "alice")
        services.store.record_user(1, "alice2")
        assert services.store.username_for(1) == "alice2"
        # Crucially the id is unchanged, so the ledger is not orphaned.
        assert services.store.user_id_for("alice2") == 1

    def test_lookup_is_case_insensitive(self, services):
        services.store.record_user(1, "Alice")
        assert services.store.user_id_for("@ALICE") == 1

    def test_unknown_handle(self, services):
        assert services.store.user_id_for("nobody") is None
        assert services.store.user_id_for("") is None

    def test_falls_back_to_first_name_then_id(self, services):
        services.store.record_user(2, "", "Bob")
        assert services.store.display_name_for(2) == "Bob"
        assert services.store.display_name_for(999) == "999"


class TestUsernameAuthorisation:
    def test_username_allowlist(self, tmp_path):
        s = make_settings(tmp_path, allowed_telegram_usernames="alice, @bob")
        assert s.is_authorised(111, "alice")
        assert s.is_authorised(222, "@bob")
        assert not s.is_authorised(333, "carol")

    def test_case_insensitive(self, tmp_path):
        s = make_settings(tmp_path, allowed_telegram_usernames="Alice")
        assert s.is_authorised(1, "ALICE")

    def test_numeric_ids_still_work(self, tmp_path):
        s = make_settings(tmp_path, allowed_telegram_user_ids="42")
        assert s.is_authorised(42, "")

    def test_no_username_and_not_in_id_list(self, tmp_path):
        s = make_settings(tmp_path, allowed_telegram_usernames="alice")
        assert not s.is_authorised(1, "")

    def test_open_access_overrides(self, tmp_path):
        s = make_settings(tmp_path, open_access=True)
        assert s.is_authorised(999, "stranger")

    def test_missing_required_mentions_usernames(self, tmp_path):
        gaps = make_settings(tmp_path).missing_required()
        assert any("ALLOWED_TELEGRAM_USERNAMES" in g for g in gaps)

    def test_username_allowlist_satisfies_the_check(self, tmp_path):
        gaps = make_settings(
            tmp_path, allowed_telegram_usernames="alice"
        ).missing_required()
        assert not any("ALLOWED" in g for g in gaps)


class TestLoadDashboard:
    def test_empty_ledger(self, services):
        data = load_dashboard(services, 1)
        assert data.month_total == 0
        assert data.month_count == 0
        assert data.recent == []

    def test_totals_this_month(self, services):
        add(services.store, 25000)
        add(services.store, 15000, "transport")
        data = load_dashboard(services, 1)
        assert data.month_total == 40000
        assert data.month_count == 2

    def test_uses_the_display_name(self, services):
        services.store.record_user(1, "alice")
        assert load_dashboard(services, 1).display_name == "@alice"

    def test_category_breakdown_is_sorted(self, services):
        add(services.store, 10000, "food_dining")
        add(services.store, 50000, "rent")
        data = load_dashboard(services, 1)
        assert data.by_category[0][0] == "rent"

    def test_only_counts_the_dominant_currency(self, services):
        add(services.store, 10000, currency="INR")
        add(services.store, 20000, currency="INR")
        add(services.store, 999999, currency="USD")
        data = load_dashboard(services, 1)
        assert data.currency == "INR"
        assert data.month_total == 30000  # never sums across currencies

    def test_is_scoped_to_one_user(self, services):
        add(services.store, 25000)
        other = Expense(
            user_id=2, amount_minor=99999, currency="INR", description="x",
            category="other", occurred_on=today_in("Asia/Kolkata"),
        )
        services.store.add_expense(other)
        assert load_dashboard(services, 1).month_total == 25000

    def test_budgets_included(self, services):
        services.store.set_budget(Budget(1, "food_dining", 100000, "INR"))
        add(services.store, 25000)
        data = load_dashboard(services, 1)
        assert len(data.budgets) == 1
        assert data.budgets[0].spent_minor == 25000


class TestRendering:
    def test_header(self, services):
        services.store.record_user(1, "alice")
        assert "@alice" in render_header(load_dashboard(services, 1))

    def test_kpis_empty_state_is_helpful(self, services):
        assert "Nothing logged" in render_kpis(load_dashboard(services, 1))

    def test_kpis_show_the_total(self, services):
        add(services.store, 25000)
        assert "₹250.00" in render_kpis(load_dashboard(services, 1))

    def test_kpis_show_the_top_category(self, services):
        add(services.store, 25000, "groceries")
        assert "Groceries" in render_kpis(load_dashboard(services, 1))

    def test_category_frame_has_expected_columns(self, services):
        add(services.store, 25000)
        frame = category_frame(load_dashboard(services, 1))
        assert list(frame.columns) == ["Category", "Amount"]

    def test_category_frame_survives_an_empty_ledger(self, services):
        assert len(category_frame(load_dashboard(services, 1))) == 1

    def test_recent_frame(self, services):
        add(services.store, 25000, merchant="Blue Tokai")
        frame = recent_frame(load_dashboard(services, 1))
        assert "Blue Tokai" in frame["Description"].tolist()

    def test_recent_frame_empty(self, services):
        frame = recent_frame(load_dashboard(services, 1))
        assert "Nothing yet" in frame["Description"].tolist()

    def test_budgets_empty_state(self, services):
        assert "None set" in render_budgets(load_dashboard(services, 1))

    def test_budgets_rendered(self, services):
        services.store.set_budget(Budget(1, "rent", 100000, "INR"))
        assert "Rent" in render_budgets(load_dashboard(services, 1))

    def test_recurring_empty_state(self, services):
        assert "Nothing detected" in render_recurring(load_dashboard(services, 1))

    def test_recurring_rendered(self, services):
        today = today_in("Asia/Kolkata")
        for months_back in (0, 1, 2):
            day = (today.replace(day=1) - timedelta(days=31 * months_back)).replace(day=5)
            add(services.store, 64900, "subscriptions", day=day, merchant="Netflix")
        assert "Netflix" in render_recurring(load_dashboard(services, 1))


class TestAuthenticate:
    """The gate on the whole web view. Everything else trusts its answer."""

    def test_valid_token(self, services):
        from finbot.auth import issue

        token = issue(1, services.settings.telegram_webhook_secret)
        assert authenticate(services, token) == 1

    def test_no_token(self, services):
        assert authenticate(services, "") is None

    def test_garbage_token(self, services):
        assert authenticate(services, "not-a-token") is None

    def test_token_signed_with_another_secret(self, services):
        from finbot.auth import issue

        assert authenticate(services, issue(1, "someone-elses-secret")) is None

    def test_expired_token(self, services):
        import time

        from finbot.auth import issue

        stale = issue(
            1,
            services.settings.telegram_webhook_secret,
            ttl_seconds=1,
            now=int(time.time()) - 60,
        )
        assert authenticate(services, stale) is None

    def test_no_services_yet(self):
        # Before the runtime exists the page must not authenticate anyone.
        assert authenticate(None, "anything") is None

    def test_token_identifies_the_right_user(self, services):
        from finbot.auth import issue

        token = issue(4242, services.settings.telegram_webhook_secret)
        assert authenticate(services, token) == 4242


class TestCoach:
    def test_passes_the_question_through(self, services):
        add(services.store, 25000)
        assert "where can I cut back" in coach_reply(services, 1, "where can I cut back")

    def test_works_with_no_data(self, services):
        assert coach_reply(services, 1, "hello")


class TestChatTurn:
    """The chat box must be a peer of Telegram, not an answer-only sidebar.

    Regression: typing "food 800" into the dashboard chat silently did nothing
    and replied with generic advice, because every message went to the coach.
    """

    def _services(self, services):
        from finbot.features import default_features
        from finbot.registry import FeatureRegistry

        reg = FeatureRegistry()
        reg.register_all(default_features())
        services.registry = reg
        services.pending = {}
        services.llm = SimpleNamespace(enabled=False)
        services.ingestor = Ingestor(services.llm, "INR")
        return services

    def test_plain_text_logs_an_expense(self, services):
        s = self._services(services)
        answer = chat_turn(s, 1, "food 800")
        assert "800" in answer
        assert s.store.count(1) == 1

    def test_logged_expense_has_the_right_amount(self, services):
        s = self._services(services)
        chat_turn(s, 1, "food 800")
        assert s.store.list_expenses(1)[0].amount_minor == 80000

    def test_commands_work(self, services):
        s = self._services(services)
        chat_turn(s, 1, "coffee 250")
        assert "250" in chat_turn(s, 1, "/month")

    def test_unknown_command_is_reported_not_coached(self, services):
        s = self._services(services)
        assert "don't know" in chat_turn(s, 1, "/save good food 800")

    def test_html_is_stripped_for_the_chat_bubble(self, services):
        s = self._services(services)
        answer = chat_turn(s, 1, "food 800")
        assert "<" not in answer and "&amp;" not in answer

    def test_a_real_question_still_reaches_the_coach(self, services):
        s = self._services(services)
        answer = chat_turn(s, 1, "where can I cut back")
        assert answer  # falls through to the advisor

    def test_empty_message(self, services):
        assert chat_turn(self._services(services), 1, "   ") == ""

    def test_undo_works_from_chat(self, services):
        s = self._services(services)
        chat_turn(s, 1, "food 800")
        chat_turn(s, 1, "/undo")
        assert s.store.count(1) == 0


class TestConversation:
    """Plain text that is not an expense should be a conversation, with memory."""

    def _svc(self, services, answer="I hear you."):
        from finbot.features import default_features
        from finbot.registry import FeatureRegistry

        seen = {}

        def advise(**kw):
            seen["question"] = kw.get("question")
            seen["history"] = kw.get("history")
            return answer

        reg = FeatureRegistry()
        reg.register_all(default_features())
        services.registry = reg
        services.pending = {}
        services.conversations = {}
        # chat_json is the ingestor's no-amount fallback; returning a zero
        # amount makes it decline, which is what lets the message reach coaching.
        services.llm = SimpleNamespace(
            enabled=True,
            chat_json=lambda *a, **k: {"amount": 0},
            vision_json=lambda *a, **k: {"amount": 0},
        )
        services.ingestor = Ingestor(services.llm, "INR")
        services.advisor = SimpleNamespace(advise=advise)
        return services, seen

    def test_plain_question_is_answered_not_rejected(self, services):
        s, seen = self._svc(services)
        answer = chat_turn(s, 1, "what is my current status")
        assert answer == "I hear you."
        assert seen["question"] == "what is my current status"

    def test_expense_still_wins_over_conversation(self, services):
        s, seen = self._svc(services)
        chat_turn(s, 1, "food 800")
        assert s.store.count(1) == 1
        assert "question" not in seen  # coach never invoked

    def test_history_accumulates(self, services):
        s, _ = self._svc(services)
        chat_turn(s, 1, "hello")
        chat_turn(s, 1, "and then?")
        assert len(s.conversations[1]) == 4  # 2 user + 2 assistant

    def test_history_is_passed_to_the_advisor(self, services):
        s, seen = self._svc(services)
        chat_turn(s, 1, "hello")
        chat_turn(s, 1, "and then?")
        assert seen["history"][0]["content"] == "hello"

    def test_history_is_bounded(self, services):
        from finbot.features.coaching import MAX_HISTORY

        s, _ = self._svc(services)
        for i in range(20):
            chat_turn(s, 1, f"tell me more, round {chr(97 + i % 26)}")
        assert len(s.conversations[1]) == MAX_HISTORY

    def test_conversations_are_per_user(self, services):
        s, _ = self._svc(services)
        chat_turn(s, 1, "hello")
        chat_turn(s, 2, "hi")
        assert len(s.conversations[1]) == 2
        assert len(s.conversations[2]) == 2

    def test_forget_clears_only_the_conversation(self, services):
        s, _ = self._svc(services)
        chat_turn(s, 1, "coffee 250")
        chat_turn(s, 1, "hello")
        chat_turn(s, 1, "/forget")
        assert s.conversations.get(1) in (None, [])
        assert s.store.count(1) == 1  # expenses untouched