298 lines
12 KiB
Python
298 lines
12 KiB
Python
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 _StubState:
|
|
value = bot.BotMode.subscription.state
|
|
|
|
async def get_state(self):
|
|
return self.value
|
|
|
|
|
|
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] = []
|
|
self.answer_kwargs: list[dict] = []
|
|
|
|
async def answer(self, text: str, **kwargs):
|
|
self.answers.append(text)
|
|
self.answer_kwargs.append(kwargs)
|
|
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", return_value=True),
|
|
patch.object(payments, "save_payment") as save,
|
|
patch.object(payments.asyncio, "to_thread", new=_direct_thread_mock()),
|
|
):
|
|
await payments.handle_successful_payment(message)
|
|
save.assert_not_called()
|
|
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_called()
|
|
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, "DEPLOY_ENV", "production"),
|
|
patch.object(bot.asyncio, "to_thread", new=AsyncMock()),
|
|
patch.object(bot, "send_subscription_invoice", new=AsyncMock()) as invoice,
|
|
):
|
|
await bot.subscribe_command(message, _StubState())
|
|
invoice.assert_awaited_once_with(message.bot, 555, None)
|
|
self.assertIn("Счёт на подписку", message.answers[0])
|
|
self.assertIn("ID пользователя: <code>42</code>", message.answers[0])
|
|
self.assertEqual(len(message.answers), 2)
|
|
self.assertNotIn("dev", " ".join(message.answers).lower())
|
|
self.assertNotIn("reply_markup", message.answer_kwargs[-1])
|
|
|
|
async def test_dev_subscribe_shows_manual_confirmation_button(self):
|
|
message = self._message()
|
|
with (
|
|
patch.object(bot, "DEPLOY_ENV", "development"),
|
|
patch.object(bot.asyncio, "to_thread", new=AsyncMock()),
|
|
patch.object(bot, "send_subscription_invoice", new=AsyncMock()),
|
|
):
|
|
await bot.subscribe_command(message, _StubState())
|
|
self.assertIn("без реального платежа", message.answers[-1])
|
|
self.assertEqual(len(message.answers), 2)
|
|
self.assertIn("Счёт", message.answers[0])
|
|
markup = message.answer_kwargs[-1]["reply_markup"]
|
|
self.assertEqual(markup.inline_keyboard[0][0].text, "🧪 Оплатил (dev)")
|
|
self.assertIn("ID пользователя: <code>42</code>", message.answers[0])
|
|
|
|
async def test_dev_confirmation_activates_only_for_invoice_user(self):
|
|
callback = AsyncMock()
|
|
callback.data = "dev_paid:42"
|
|
callback.from_user = SimpleNamespace(id=42)
|
|
callback.message = None
|
|
with (
|
|
patch.object(bot, "DEPLOY_ENV", "development"),
|
|
patch.object(bot.asyncio, "to_thread", new=AsyncMock(return_value=30)) as to_thread,
|
|
):
|
|
await bot.confirm_development_payment(callback)
|
|
self.assertEqual(to_thread.await_args.args[0], user_store.activate_subscription)
|
|
callback.answer.assert_awaited_once_with()
|
|
|
|
async def test_dev_confirmation_rejects_other_user(self):
|
|
callback = AsyncMock()
|
|
callback.data = "dev_paid:42"
|
|
callback.from_user = SimpleNamespace(id=99)
|
|
with (
|
|
patch.object(bot, "DEPLOY_ENV", "development"),
|
|
patch.object(bot.asyncio, "to_thread", new=AsyncMock()) as to_thread,
|
|
):
|
|
await bot.confirm_development_payment(callback)
|
|
to_thread.assert_not_awaited()
|
|
callback.answer.assert_awaited_once()
|
|
|
|
async def test_prod_confirmation_does_not_grant_subscription(self):
|
|
callback = AsyncMock()
|
|
callback.data = "dev_paid:42"
|
|
callback.from_user = SimpleNamespace(id=42)
|
|
with (
|
|
patch.object(bot, "DEPLOY_ENV", "production"),
|
|
patch.object(bot.asyncio, "to_thread", new=AsyncMock()) as to_thread,
|
|
):
|
|
await bot.confirm_development_payment(callback)
|
|
to_thread.assert_not_awaited()
|
|
callback.answer.assert_awaited_once_with()
|
|
|
|
async def test_unknown_environment_disables_dev_payment(self):
|
|
callback = AsyncMock()
|
|
callback.data = "dev_paid:42"
|
|
callback.from_user = SimpleNamespace(id=42)
|
|
with (
|
|
patch.object(bot, "DEPLOY_ENV", "staging"),
|
|
patch.object(bot.asyncio, "to_thread", new=AsyncMock()) as to_thread,
|
|
):
|
|
await bot.confirm_development_payment(callback)
|
|
to_thread.assert_not_awaited()
|
|
callback.answer.assert_awaited_once_with()
|
|
|
|
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, _StubState())
|
|
self.assertIn("поддержке", message.answers[0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |