tests/test_app.py

"""End-to-end webhook tests with a stubbed Telegram transport.

Environment is configured at import time, before the app's lifespan calls
``get_settings()`` (which is lru_cached, so the first call wins).
"""

import os
import tempfile

os.environ.update(
    {
        "DATA_DIR": tempfile.mkdtemp(prefix="finbot-test-"),
        "TELEGRAM_BOT_TOKEN": "",  # keeps lifespan from calling Telegram
        "TELEGRAM_WEBHOOK_SECRET": "test-secret",
        "ALLOWED_TELEGRAM_USER_IDS": "42",
        "HF_TOKEN": "",  # deterministic paths only; no network
        "HF_DATASET_REPO": "",
        "DEFAULT_CURRENCY": "INR",
        "TIMEZONE": "Asia/Kolkata",
    }
)

import pytest  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402

from app import app  # noqa: E402

SECRET = {"X-Telegram-Bot-Api-Secret-Token": "test-secret"}


class FakeTelegram:
    """Records what would have been sent instead of calling the Bot API."""

    def __init__(self):
        self.username = "FinBotTest"
        self.messages: list[tuple[int, str]] = []
        self.documents: list[tuple[int, str, bytes]] = []
        self.enabled = True

    async def send_message(self, chat_id, text, parse_mode="HTML"):
        self.messages.append((chat_id, text))

    async def send_document(self, chat_id, filename, data, caption=""):
        self.documents.append((chat_id, filename, data))

    async def send_typing(self, chat_id):
        pass

    async def download(self, file_id):
        return b"", "file.bin"

    async def close(self):
        pass

    @property
    def last(self) -> str:
        return self.messages[-1][1] if self.messages else ""


@pytest.fixture
def client():
    with TestClient(app) as test_client:
        # Swap the Telegram transport on the live runtime so nothing hits the
        # network; background tasks hold this same Runtime instance.
        rt = app.state.rt
        fake = FakeTelegram()
        rt.telegram = fake
        test_client.fake = fake
        # Isolate each test: empty the ledger and any pending imports.
        store = rt.services.store
        for expense in store.list_expenses(42):
            store.delete_expense(42, expense.id)
        rt.services.pending.clear()
        yield test_client


def send(client, text, user_id=42, update_id=None):
    import random

    payload = {
        "update_id": update_id if update_id is not None else random.randint(1, 10**9),
        "message": {
            "from": {"id": user_id},
            "chat": {"id": user_id},
            "text": text,
        },
    }
    return client.post("/telegram/webhook", json=payload, headers=SECRET)


class TestEndpoints:
    def test_health(self, client):
        response = client.get("/health")
        assert response.status_code == 200
        assert response.json()["status"] == "ok"

    def test_health_leaks_no_ledger_detail(self, client):
        # The Space may be public; liveness must not publish personal metrics.
        body = client.get("/health").json()
        assert set(body) == {"status", "version"}

    def test_index_serves_the_gradio_ui(self, client):
        response = client.get("/")
        assert response.status_code == 200
        assert "Oracle" in response.text

    def test_index_leaks_no_ledger_detail(self, client):
        send(client, "coffee 250")
        text = client.get("/").text.lower()
        for leak in ("mirroring", "not configured", "hf_dataset_repo", "telegram_bot_token"):
            assert leak not in text

    def test_public_page_shows_only_the_landing_copy(self, client):
        # No token: a visitor must see the explainer, never anyone's numbers.
        send(client, "coffee 250")
        text = client.get("/").text
        assert "sign-in link" in text
        assert "₹250.00" not in text

    def test_diagnostics_requires_the_secret(self, client):
        assert client.get("/admin/diagnostics").status_code == 403
        assert client.get(
            "/admin/diagnostics", headers={"X-Admin-Secret": "wrong"}
        ).status_code == 403

    def test_diagnostics_with_the_secret(self, client):
        response = client.get("/admin/diagnostics", headers={"X-Admin-Secret": "test-secret"})
        assert response.status_code == 200
        assert response.json()["ok"] is True
        assert "rows" in response.json()


class TestWebhookSecurity:
    def test_rejects_a_missing_secret(self, client):
        response = client.post("/telegram/webhook", json={})
        assert response.status_code == 403

    def test_rejects_a_wrong_secret(self, client):
        response = client.post(
            "/telegram/webhook",
            json={},
            headers={"X-Telegram-Bot-Api-Secret-Token": "wrong"},
        )
        assert response.status_code == 403

    def test_rejects_malformed_json(self, client):
        response = client.post("/telegram/webhook", content=b"not json", headers=SECRET)
        assert response.status_code == 400

    def test_unauthorised_user_is_refused_and_told_their_id(self, client):
        send(client, "coffee 250", user_id=9999)
        assert "9999" in client.fake.last
        assert "allowlist" in client.fake.last

    def test_unauthorised_user_writes_nothing(self, client):
        send(client, "coffee 250", user_id=9999)
        assert app.state.rt.services.store.count(9999) == 0

    def test_duplicate_update_is_processed_once(self, client):
        send(client, "coffee 250", update_id=555)
        send(client, "coffee 250", update_id=555)
        assert app.state.rt.services.store.count(42) == 1


class TestLoggingFlow:
    def test_plain_text_is_logged(self, client):
        response = send(client, "coffee 250")
        assert response.status_code == 200
        assert "Logged" in client.fake.last
        assert "₹250.00" in client.fake.last
        assert app.state.rt.services.store.count(42) == 1

    def test_text_without_an_amount_becomes_conversation(self, client):
        # Previously replied "I couldn't find an amount in that", which made the
        # bot feel like a form. Plain text now reaches the coach instead.
        send(client, "hello there")
        assert "couldn't find an amount" not in client.fake.last
        assert client.fake.last  # the coach said something
        assert app.state.rt.services.store.count(42) == 0

    def test_spend_command(self, client):
        send(client, "/spend 450 swiggy dinner")
        assert "Logged" in client.fake.last
        assert app.state.rt.services.store.count(42) == 1

    def test_spend_without_args_explains(self, client):
        send(client, "/spend")
        assert "/spend 250 coffee" in client.fake.last

    def test_undo_removes_the_last_entry(self, client):
        send(client, "coffee 250")
        send(client, "/undo")
        assert "Removed" in client.fake.last
        assert app.state.rt.services.store.count(42) == 0

    def test_undo_with_nothing_to_undo(self, client):
        send(client, "/undo")
        assert "Nothing to undo" in client.fake.last


class TestSummaries:
    def test_today(self, client):
        send(client, "coffee 250")
        send(client, "300 uber")
        send(client, "/today")
        assert "₹550.00" in client.fake.last

    def test_month(self, client):
        send(client, "1000 rent")
        send(client, "/month")
        assert "₹1,000.00" in client.fake.last

    def test_empty_period(self, client):
        send(client, "/today")
        assert "Nothing logged" in client.fake.last

    def test_multi_currency_is_reported_separately(self, client):
        send(client, "coffee 250")
        send(client, "€40 dinner")
        send(client, "/today")
        # Two currencies must never be summed into one figure.
        assert "Also:" in client.fake.last

    def test_list(self, client):
        send(client, "coffee 250")
        send(client, "/list")
        assert "coffee" in client.fake.last

    def test_report_accepts_a_period(self, client):
        send(client, "/report last_month")
        assert client.fake.last


class TestBudgets:
    def test_set_and_view(self, client):
        send(client, "/budget food_dining 8000")
        assert "Budget set" in client.fake.last
        send(client, "/budgets")
        assert "Food & Dining" in client.fake.last

    def test_budget_tracks_spending(self, client):
        send(client, "/budget food_dining 1000")
        send(client, "500 lunch")
        send(client, "/budgets")
        assert "50%" in client.fake.last

    def test_unknown_category_is_rejected(self, client):
        send(client, "/budget nonsense 500")
        assert "don't recognise" in client.fake.last

    def test_budget_without_amount(self, client):
        send(client, "/budget groceries")
        assert "need an amount" in client.fake.last

    def test_categories_listing(self, client):
        send(client, "/categories")
        assert "food_dining" in client.fake.last


class TestCoreCommands:
    def test_start(self, client):
        send(client, "/start")
        assert "Hello" in client.fake.last

    def test_help_lists_registered_features(self, client):
        send(client, "/help")
        text = client.fake.last
        assert "/spend" in text
        assert "/budget" in text
        assert "/advice" in text

    def test_unknown_command(self, client):
        send(client, "/nonsense")
        assert "don't know" in client.fake.last

    def test_status_warns_about_ephemeral_storage(self, client):
        send(client, "/status")
        # HF_DATASET_REPO is unset in this environment, so it must say so.
        assert "ephemeral" in client.fake.last.lower()

    def test_whoami(self, client):
        send(client, "/whoami")
        assert "42" in client.fake.last

    def test_advice_works_without_inference(self, client):
        send(client, "500 lunch")
        send(client, "/advice")
        assert client.fake.last  # deterministic observations still fire

    def test_export_sends_a_csv(self, client):
        send(client, "coffee 250")
        send(client, "/export")
        assert len(client.fake.documents) == 1
        _chat, filename, payload = client.fake.documents[0]
        assert filename.endswith(".csv")
        assert b"coffee" in payload

    def test_export_with_nothing_logged(self, client):
        send(client, "/export")
        assert "Nothing to export" in client.fake.last