social-post-writer-bot/tests/test_payments.py

import os
import tempfile
import unittest
from unittest.mock import AsyncMock, patch

from types import SimpleNamespace

from aiogram.types import LabeledPrice, SuccessfulPayment

import bot
import payments
import user_store


class PaymentStoreTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.NamedTemporaryFile(suffix=".sqlite3", delete=False)
        self.tmp.close()
        self.db = self.tmp.name
        user_store.init_db(self.db)
        payments.init_payments_table(self.db)

    def tearDown(self):
        os.unlink(self.db)

    def test_payment_is_recorded_once(self):
        payments.save_payment("charge1", 42, "sub30:1", "XTR", 100, self.db)
        self.assertTrue(payments.payment_recorded("charge1", self.db))
        self.assertFalse(payments.payment_recorded("charge2", self.db))

    def test_save_payment_is_idempotent(self):
        payments.save_payment("charge1", 42, "sub30:1", "XTR", 100, self.db)
        payments.save_payment("charge1", 42, "sub30:1", "XTR", 100, self.db)
        with payments.closing(payments.connect(self.db)) as connection:
            rows = connection.execute("SELECT COUNT(*) FROM payments").fetchone()
        self.assertEqual(rows[0], 1)


class InvoiceTests(unittest.IsolatedAsyncioTestCase):
    async def test_invoice_uses_stars_and_correct_price(self):
        fake_bot = AsyncMock()
        await payments.send_subscription_invoice(fake_bot, 555)
        kwargs = fake_bot.send_invoice.await_args.kwargs
        self.assertEqual(kwargs["currency"], "XTR")
        self.assertEqual(kwargs["provider_token"], "")
        self.assertEqual(kwargs["chat_id"], 555)
        self.assertTrue(kwargs["payload"].startswith(payments.INVOICE_PREFIX))
        prices = kwargs["prices"]
        self.assertIsInstance(prices[0], LabeledPrice)
        self.assertEqual(prices[0].amount, user_store.SUBSCRIPTION_PRICE_STARS)

    async def test_invoice_includes_thread_when_given(self):
        fake_bot = AsyncMock()
        await payments.send_subscription_invoice(fake_bot, 555, thread_id=77)
        self.assertEqual(fake_bot.send_invoice.await_args.kwargs["message_thread_id"], 77)

    async def test_invoice_omits_thread_when_absent(self):
        fake_bot = AsyncMock()
        await payments.send_subscription_invoice(fake_bot, 555)
        self.assertNotIn("message_thread_id", fake_bot.send_invoice.await_args.kwargs)


class PreCheckoutTests(unittest.IsolatedAsyncioTestCase):
    def _query(self, currency="XTR", amount=100, payload="sub30:1"):
        query = AsyncMock()
        query.currency = currency
        query.total_amount = amount
        query.invoice_payload = payload
        return query

    async def test_accepts_valid_subscription_invoice(self):
        query = self._query()
        await payments.handle_pre_checkout(query)
        query.answer.assert_awaited_once_with(ok=True)

    async def test_rejects_wrong_amount(self):
        query = self._query(amount=1)
        await payments.handle_pre_checkout(query)
        self.assertFalse(query.answer.await_args.kwargs["ok"])

    async def test_rejects_wrong_currency(self):
        query = self._query(currency="USD")
        await payments.handle_pre_checkout(query)
        self.assertFalse(query.answer.await_args.kwargs["ok"])

    async def test_rejects_foreign_payload(self):
        query = self._query(payload="someone-elses-invoice")
        await payments.handle_pre_checkout(query)
        self.assertFalse(query.answer.await_args.kwargs["ok"])


def _direct_thread(*args, **kwargs):
    """Вызывает синхронную функцию прямо, без пула потоков."""
    func = args[0]
    return func(*args[1:], **kwargs)


def _direct_thread_mock():
    return AsyncMock(side_effect=_direct_thread)

class _StubMessage:
    """Минимальный Message: нам нужны только successful_payment, from_user и answer."""

    def __init__(self, payment=None, user_id=42, chat_id=555):
        self.successful_payment = payment
        self.from_user = SimpleNamespace(id=user_id)
        self.chat = SimpleNamespace(id=chat_id)
        self.answers: list[str] = []

    async def answer(self, text: str, **kwargs):
        self.answers.append(text)
        return self


def _paid_message(charge_id="charge1", amount=100, currency="XTR", payload="sub30:1"):
    payment = SuccessfulPayment(
        currency=currency,
        total_amount=amount,
        invoice_payload=payload,
        telegram_payment_charge_id=charge_id,
        provider_payment_charge_id="",
    )
    return _StubMessage(payment=payment)


def _db() -> str:
    handle = tempfile.NamedTemporaryFile(suffix=".sqlite3", delete=False)
    handle.close()
    user_store.init_db(handle.name)
    payments.init_payments_table(handle.name)
    return handle.name


class SuccessfulPaymentTests(unittest.IsolatedAsyncioTestCase):
    def tearDown(self):
        if getattr(self, "db", None):
            os.unlink(self.db)

    async def test_payment_activates_subscription(self):
        message = _paid_message()
        self.db = _db()
        user_store.ensure_subscription(42, self.db)
        with patch.dict(os.environ, {"BOT_DB_PATH": self.db}):
            await payments.handle_successful_payment(message)
        self.assertEqual(len(message.answers), 1)
        self.assertIn("активирована", message.answers[0])
        self.assertTrue(payments.payment_recorded("charge1", self.db))
        row = user_store.subscription_row(42, self.db)
        self.assertIsNotNone(row["last_paid_at"])

    async def test_duplicate_payment_does_not_extend_twice(self):
        message = _paid_message(charge_id="charge-dup")
        with (
            patch.object(payments, "payment_recorded", new=AsyncMock(return_value=True)),
            patch.object(payments, "save_payment", new=AsyncMock()) as save,
            patch.object(payments.asyncio, "to_thread", new=_direct_thread_mock()),
        ):
            await payments.handle_successful_payment(message)
        save.assert_not_awaited()
        self.assertIn("уже учтён", message.answers[0])

    async def test_wrong_amount_is_rejected_without_activation(self):
        message = _paid_message(amount=5)
        with (
            patch.object(payments, "save_payment", new=AsyncMock()) as save,
            patch.object(payments, "activate_subscription") as activate,
        ):
            await payments.handle_successful_payment(message)
        activate.assert_not_called()
        save.assert_not_awaited()
        self.assertIn("поддержке", message.answers[0])

    async def test_wrong_currency_is_rejected(self):
        message = _paid_message(currency="USD")
        with patch.object(payments, "activate_subscription") as activate:
            await payments.handle_successful_payment(message)
        activate.assert_not_called()

    async def test_foreign_payload_is_rejected_without_activation(self):
        message = _paid_message(payload="other")
        with patch.object(payments, "activate_subscription") as activate:
            await payments.handle_successful_payment(message)
        activate.assert_not_called()


class SubscribeCommandTests(unittest.IsolatedAsyncioTestCase):
    def _message(self):
        message = _StubMessage()
        message.bot = AsyncMock()
        message.message_thread_id = None
        return message

    async def test_subscribe_sends_invoice_and_never_activates_directly(self):
        from pathlib import Path

        source = Path(bot.__file__).read_text(encoding="utf-8")
        body = source.split("async def subscribe_command")[1].split("\nasync def ")[0]
        self.assertNotIn("activate_subscription", body)

        message = self._message()
        with (
            patch.object(bot.asyncio, "to_thread", new=AsyncMock()),
            patch.object(bot, "send_subscription_invoice", new=AsyncMock()) as invoice,
        ):
            await bot.subscribe_command(message)
        invoice.assert_awaited_once_with(message.bot, 555, None)
        self.assertIn("Счёт", message.answers[0])

    async def test_subscribe_reports_failure_when_invoice_fails(self):
        message = self._message()
        with (
            patch.object(bot.asyncio, "to_thread", new=AsyncMock()),
            patch.object(bot, "send_subscription_invoice", new=AsyncMock(side_effect=RuntimeError("boom"))),
        ):
            await bot.subscribe_command(message)
        self.assertIn("поддержке", message.answers[0])


if __name__ == "__main__":
    unittest.main()