admin
9400050ade
- handle_successful_payment: after status, reset FSM to lobby and send main menu keyboard (state injected by dispatcher) - confirm_development_payment (dev path): same — edit status then reply with main menu; add state param - tests: assert second message carries main-menu markup; _StubState clear/set_state
354 lines
15 KiB
Python
354 lines
15 KiB
Python
import asyncio
|
|
from datetime import datetime
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from types import SimpleNamespace
|
|
|
|
from aiogram.types import Chat, LabeledPrice, Message, SuccessfulPayment
|
|
|
|
import bot
|
|
import payments
|
|
import user_store
|
|
|
|
|
|
class _StubState:
|
|
value = bot.BotMode.subscription.state
|
|
|
|
async def get_state(self):
|
|
return self.value
|
|
|
|
async def clear(self):
|
|
self.value = None
|
|
|
|
async def set_state(self, state):
|
|
self.value = getattr(state, "state", state)
|
|
|
|
|
|
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):
|
|
for days, price in user_store.SUBSCRIPTION_PLANS.items():
|
|
with self.subTest(days=days):
|
|
fake_bot = AsyncMock()
|
|
await payments.send_subscription_invoice(fake_bot, 555, days=days)
|
|
kwargs = fake_bot.send_invoice.await_args.kwargs
|
|
self.assertEqual(kwargs["title"], f"Подписка на {days} дней")
|
|
self.assertEqual(kwargs["description"], "Доступ к боту по генерации текста постов")
|
|
self.assertEqual(kwargs["prices"][0].amount, price)
|
|
self.assertEqual(payments.plan_for_payload(kwargs["payload"]), (days, price))
|
|
self.assertEqual(kwargs["currency"], "XTR")
|
|
self.assertEqual(kwargs["provider_token"], "")
|
|
|
|
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=450, payload="sub:30:20261001120000"):
|
|
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(payload="sub:30:20261001120000")
|
|
await payments.handle_pre_checkout(query)
|
|
query.answer.assert_awaited_once_with(ok=True)
|
|
|
|
async def test_accepts_valid_zero_day_one_star_invoice(self):
|
|
query = self._query(amount=1, payload="sub:0:20261001120000")
|
|
await payments.handle_pre_checkout(query)
|
|
query.answer.assert_awaited_once_with(ok=True)
|
|
|
|
async def test_rejects_zero_day_invoice_with_wrong_amount(self):
|
|
query = self._query(amount=0, payload="sub:0:20261001120000")
|
|
await payments.handle_pre_checkout(query)
|
|
self.assertFalse(query.answer.await_args.kwargs["ok"])
|
|
|
|
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.message_thread_id = None
|
|
self.bot = AsyncMock()
|
|
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=450, currency="XTR", payload="sub:30:20261001120000"):
|
|
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), 2)
|
|
self.assertIn("Подписка активна", message.answers[0])
|
|
# Второе сообщение — возврат в главное меню с его клавиатурой.
|
|
self.assertIsNotNone(message.answer_kwargs[1].get("reply_markup"))
|
|
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_days") 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_days") 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_days") 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):
|
|
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._send_subscription_invoice(message, 42, days=30)
|
|
invoice.assert_awaited_once_with(message.bot, 555, None, days=30)
|
|
self.assertEqual(len(message.answers), 0)
|
|
|
|
async def test_dev_subscribe_shows_selected_days_and_explains_test_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()) as invoice,
|
|
):
|
|
await bot._send_subscription_invoice(message, 42, days=10)
|
|
invoice.assert_awaited_once_with(message.bot, 555, None, days=10)
|
|
self.assertEqual(len(message.answers), 1)
|
|
self.assertIn("имитирует оплату", message.answers[0])
|
|
markup = message.answer_kwargs[-1]["reply_markup"]
|
|
self.assertEqual(markup.inline_keyboard[0][0].text, "🧪 Оплатил (dev, 10 дн.)")
|
|
self.assertEqual(markup.inline_keyboard[0][0].callback_data, "dev_paid:42:10")
|
|
|
|
async def test_zero_day_dev_confirmation_is_accepted_without_activation(self):
|
|
callback = AsyncMock()
|
|
callback.data = "dev_paid:42:0"
|
|
callback.from_user = SimpleNamespace(id=42)
|
|
callback.message = SimpleNamespace(
|
|
**Message(
|
|
message_id=1,
|
|
date=datetime.now(),
|
|
chat=Chat(id=555, type="private"),
|
|
).model_dump(),
|
|
edit_reply_markup=AsyncMock(),
|
|
edit_text=AsyncMock(),
|
|
)
|
|
with (
|
|
patch.object(bot, "DEPLOY_ENV", "development"),
|
|
patch.object(bot.asyncio, "to_thread", new=AsyncMock()) as to_thread,
|
|
):
|
|
await bot.confirm_development_payment(callback, _StubState())
|
|
to_thread.assert_not_awaited()
|
|
callback.answer.assert_awaited_once_with("Dev-оплата подтверждена. Срок подписки не изменён.")
|
|
|
|
async def test_zero_day_plan_is_a_one_star_invoice_and_grants_one_day(self):
|
|
self.assertEqual(payments.plan_for_payload("sub:0:20261001120000"), (0, 1))
|
|
|
|
message = _paid_message(amount=1, payload="sub:0:20261001120000")
|
|
self.db = _db()
|
|
user_store.ensure_subscription(42, self.db)
|
|
with (
|
|
patch.dict(os.environ, {"BOT_DB_PATH": self.db}),
|
|
patch.object(payments.asyncio, "to_thread", new=_direct_thread_mock()),
|
|
patch.object(payments, "activate_subscription_days", wraps=user_store.activate_subscription_days) as activate,
|
|
):
|
|
await payments.handle_successful_payment(message)
|
|
activate.assert_called_once_with(42, 1)
|
|
self.assertTrue(payments.payment_recorded("charge1", self.db))
|
|
self.assertIn("Подписка активна", message.answers[0])
|
|
|
|
async def test_dev_confirmation_activates_selected_term_and_replies_with_status(self):
|
|
callback = AsyncMock()
|
|
callback.data = "dev_paid:42:10"
|
|
callback.from_user = SimpleNamespace(id=42)
|
|
callback.message = Message(message_id=1, date=datetime.now(), chat=Chat(id=555, type="private")).as_(AsyncMock())
|
|
db = _db()
|
|
user_store.ensure_subscription(42, db)
|
|
with (
|
|
patch.object(bot, "DEPLOY_ENV", "development"),
|
|
patch.dict(os.environ, {"BOT_DB_PATH": db}),
|
|
patch.object(bot.asyncio, "to_thread", new=_direct_thread_mock()) as to_thread,
|
|
):
|
|
await bot.confirm_development_payment(callback, _StubState())
|
|
db_row = user_store.subscription_row(42, db)
|
|
patch_calls = [call for call in to_thread.await_args_list if call.args[0] is user_store.activate_subscription_days]
|
|
self.assertEqual(patch_calls[0].args[1:], (42, 10))
|
|
self.assertIsNotNone(db_row)
|
|
callback.answer.assert_awaited_once_with("Dev-оплата подтверждена.")
|
|
|
|
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, _StubState())
|
|
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, _StubState())
|
|
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, _StubState())
|
|
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() |