from __future__ import annotations from datetime import datetime from uuid import uuid4 import pytest from sqlalchemy import select from app.config import Settings from app.db.models import PaymentOrder, ReferralBonus, ReferralInvite, TelegramUser from app.services.payment_service import ( FULFILLED_PAYMENT_STATUS, PENDING_PAYMENT_STATUS, REFERRAL_BONUS_APPLIED_STATUS, REFERRAL_BONUS_PENDING_STATUS, REVIEW_PAYMENT_STATUS, PaymentService, ) from app.services.remnawave_client import RemnawaveApiError from tests.helpers import make_remote_user class _StubSyncService: def __init__(self, *, cached_users_by_telegram: dict[int, list] | None = None) -> None: self.refreshed: list[int] = [] self.cached_users_by_telegram = cached_users_by_telegram or {} async def get_referral_summary(self, telegram_id: int): return type( "ReferralSummaryStub", (), { "applied_referral_code": "", "referral_code": "", "total_invited": 0, "recent_names": [], }, )() async def refresh_cached_users_for_telegram(self, *, telegram_id: int): self.refreshed.append(telegram_id) return [] async def get_cached_users_for_telegram(self, telegram_id: int): return list(self.cached_users_by_telegram.get(telegram_id, [])) class _StubRemnawaveClient: def __init__(self, *, created_user, users_by_telegram_id: dict[int, list] | None = None) -> None: self.created_user = created_user self.created_bodies: list[dict[str, object]] = [] self.updated_bodies: list[dict[str, object]] = [] self.users_by_telegram_id = users_by_telegram_id or {} async def get_user_by_uuid(self, user_uuid: str): for users in self.users_by_telegram_id.values(): for user in users: if str(user.uuid) == user_uuid: return user raise RemnawaveApiError(status_code=404, message="not found") async def get_user_by_username(self, username: str): raise RemnawaveApiError(status_code=404, message="not found") async def get_users_by_telegram_id(self, telegram_id: int): return list(self.users_by_telegram_id.get(telegram_id, [])) async def create_user(self, body: dict[str, object]): self.created_bodies.append(body) return self.created_user async def update_user(self, body: dict[str, object]): self.updated_bodies.append(body) raw_uuid = str(body.get("uuid") or "") raw_expire_at = str(body.get("expireAt") or "") if raw_expire_at.endswith("Z"): raw_expire_at = raw_expire_at.replace("Z", "+00:00") next_expire_at = datetime.fromisoformat(raw_expire_at) if raw_expire_at else None for telegram_id, users in self.users_by_telegram_id.items(): for index, user in enumerate(users): if str(user.uuid) != raw_uuid: continue updated_user = make_remote_user( user_uuid=str(user.uuid), user_id=user.id, short_uuid=user.short_uuid, username=user.username, telegram_id=telegram_id, expire_at=next_expire_at or user.expire_at, traffic_limit_bytes=user.traffic_limit_bytes, ) users[index] = updated_user return updated_user raise AssertionError("update_user should not be called in this scenario") async def _seed_order( session_factory, *, telegram_id: int, status: str, order_uuid: str | None = None, ) -> str: resolved_order_uuid = order_uuid or str(uuid4()) async with session_factory() as session: user = TelegramUser( telegram_id=telegram_id, username="tester", first_name="Test", language_code="ru", ) session.add(user) await session.flush() session.add( PaymentOrder( order_uuid=resolved_order_uuid, telegram_user_id=user.id, telegram_id=telegram_id, plan_code="30d", plan_title="OREOL VPN на 30 дней", plan_duration_days=30, traffic_limit_bytes=0, traffic_limit_strategy="NO_RESET", amount_stars=250, currency="RUB", status=status, invoice_payload=f"payload-{resolved_order_uuid}", provision_username="Oreol-1001-tester", error_message=None, ) ) await session.commit() return resolved_order_uuid async def _seed_referral_invite( session_factory, *, inviter_telegram_id: int, invited_telegram_id: int, ) -> None: async with session_factory() as session: session.add( ReferralInvite( inviter_telegram_id=inviter_telegram_id, invited_telegram_id=invited_telegram_id, invited_username="buyer", invited_display_name="Buyer", ) ) await session.commit() def _make_settings() -> Settings: return Settings.model_construct( bot_admin_ids_raw="", payment_internal_squad_uuids_raw="11111111-1111-1111-1111-111111111111", payment_user_tag="BOT", payment_external_squad_uuid="", payment_username_prefix="Oreol", payment_plan_traffic_limit_gb=0, payment_plan_traffic_reset_period="NO_RESET", referral_discount_percent=5, referral_bonus_days=7, ) async def test_mark_order_under_review_only_transitions_once(session_factory) -> None: service = PaymentService( settings=_make_settings(), session_factory=session_factory, remnawave_client=_StubRemnawaveClient( created_user=make_remote_user( user_uuid="11111111-1111-1111-1111-111111111111", user_id=1, short_uuid="abc123", username="oreol-user", telegram_id=1001, ) ), sync_service=_StubSyncService(), ) order_uuid = await _seed_order( session_factory, telegram_id=1001, status=PENDING_PAYMENT_STATUS, ) order = await service.mark_order_under_review(order_uuid=order_uuid, telegram_id=1001) assert order.status == REVIEW_PAYMENT_STATUS with pytest.raises(ValueError): await service.mark_order_under_review(order_uuid=order_uuid, telegram_id=1001) async def test_approve_order_requires_review_status(session_factory) -> None: remote_user = make_remote_user( user_uuid="22222222-2222-2222-2222-222222222222", user_id=2, short_uuid="def456", username="oreol-user", telegram_id=1001, ) service = PaymentService( settings=_make_settings(), session_factory=session_factory, remnawave_client=_StubRemnawaveClient(created_user=remote_user), sync_service=_StubSyncService(), ) order_uuid = await _seed_order( session_factory, telegram_id=1001, status=PENDING_PAYMENT_STATUS, ) with pytest.raises(ValueError): await service.approve_order(order_uuid=order_uuid) async def test_approve_order_fulfills_and_refreshes_cache(session_factory) -> None: remote_user = make_remote_user( user_uuid="33333333-3333-3333-3333-333333333333", user_id=3, short_uuid="ghi789", username="oreol-user", telegram_id=1001, ) sync_service = _StubSyncService() remnawave_client = _StubRemnawaveClient(created_user=remote_user) service = PaymentService( settings=_make_settings(), session_factory=session_factory, remnawave_client=remnawave_client, sync_service=sync_service, ) order_uuid = await _seed_order( session_factory, telegram_id=1001, status=REVIEW_PAYMENT_STATUS, ) access = await service.approve_order(order_uuid=order_uuid) assert access.remnawave_user_uuid == str(remote_user.uuid) assert access.is_renewal is False assert sync_service.refreshed == [1001] assert remnawave_client.created_bodies async with session_factory() as session: order = await session.scalar( select(PaymentOrder).where(PaymentOrder.order_uuid == order_uuid) ) assert order is not None assert order.status == FULFILLED_PAYMENT_STATUS assert order.remnawave_user_uuid == str(remote_user.uuid) async def test_approve_order_grants_and_applies_referral_bonus(session_factory) -> None: buyer_remote_user = make_remote_user( user_uuid="44444444-4444-4444-4444-444444444444", user_id=4, short_uuid="buyer444", username="oreol-buyer", telegram_id=1001, ) inviter_remote_user = make_remote_user( user_uuid="55555555-5555-5555-5555-555555555555", user_id=5, short_uuid="inviter555", username="oreol-inviter", telegram_id=2002, ) sync_service = _StubSyncService() remnawave_client = _StubRemnawaveClient( created_user=buyer_remote_user, users_by_telegram_id={2002: [inviter_remote_user]}, ) service = PaymentService( settings=_make_settings(), session_factory=session_factory, remnawave_client=remnawave_client, sync_service=sync_service, ) order_uuid = await _seed_order( session_factory, telegram_id=1001, status=REVIEW_PAYMENT_STATUS, ) await _seed_referral_invite( session_factory, inviter_telegram_id=2002, invited_telegram_id=1001, ) access = await service.approve_order(order_uuid=order_uuid) assert access.referral_bonus is not None assert access.referral_bonus.inviter_telegram_id == 2002 assert access.referral_bonus.bonus_days == 7 assert access.referral_bonus.status == REFERRAL_BONUS_APPLIED_STATUS assert sync_service.refreshed == [1001, 2002] assert any( body.get("uuid") == str(inviter_remote_user.uuid) for body in remnawave_client.updated_bodies ) async with session_factory() as session: bonus = await session.scalar( select(ReferralBonus).where(ReferralBonus.source_order_uuid == order_uuid) ) assert bonus is not None assert bonus.status == REFERRAL_BONUS_APPLIED_STATUS async def test_approve_order_keeps_referral_bonus_pending_without_inviter_access( session_factory, ) -> None: buyer_remote_user = make_remote_user( user_uuid="66666666-6666-6666-6666-666666666666", user_id=6, short_uuid="buyer666", username="oreol-buyer-2", telegram_id=1001, ) sync_service = _StubSyncService() remnawave_client = _StubRemnawaveClient(created_user=buyer_remote_user) service = PaymentService( settings=_make_settings(), session_factory=session_factory, remnawave_client=remnawave_client, sync_service=sync_service, ) order_uuid = await _seed_order( session_factory, telegram_id=1001, status=REVIEW_PAYMENT_STATUS, ) await _seed_referral_invite( session_factory, inviter_telegram_id=2002, invited_telegram_id=1001, ) access = await service.approve_order(order_uuid=order_uuid) assert access.referral_bonus is not None assert access.referral_bonus.status == REFERRAL_BONUS_PENDING_STATUS assert sync_service.refreshed == [1001] async with session_factory() as session: bonus = await session.scalar( select(ReferralBonus).where(ReferralBonus.source_order_uuid == order_uuid) ) assert bonus is not None assert bonus.status == REFERRAL_BONUS_PENDING_STATUS async def test_create_order_generates_unique_provision_username(session_factory) -> None: service = PaymentService( settings=Settings.model_construct( bot_admin_ids_raw="", payment_plans_raw="30:250", payment_transfer_text="Реквизиты", payment_internal_squad_uuids_raw="11111111-1111-1111-1111-111111111111", payment_user_tag="BOT", payment_external_squad_uuid="", payment_username_prefix="Oreol", payment_plan_traffic_limit_gb=0, payment_plan_traffic_reset_period="NO_RESET", referral_discount_percent=5, referral_bonus_days=7, ), session_factory=session_factory, remnawave_client=_StubRemnawaveClient( created_user=make_remote_user( user_uuid="77777777-7777-7777-7777-777777777777", user_id=7, short_uuid="buyer777", username="oreol-user", telegram_id=1001, ) ), sync_service=_StubSyncService(), ) first_order = await service.create_order( telegram_id=1001, username="tester", first_name="Test", last_name=None, language_code="ru", plan_code="30d", ) second_order = await service.create_order( telegram_id=1001, username="tester", first_name="Test", last_name=None, language_code="ru", plan_code="30d", ) assert first_order.provision_username != second_order.provision_username assert first_order.extends_existing_access is False assert second_order.extends_existing_access is False async def test_create_order_transliterates_russian_first_name_without_username( session_factory, ) -> None: service = PaymentService( settings=Settings.model_construct( bot_admin_ids_raw="", payment_plans_raw="30:250", payment_transfer_text="Реквизиты", payment_internal_squad_uuids_raw="11111111-1111-1111-1111-111111111111", payment_user_tag="BOT", payment_external_squad_uuid="", payment_username_prefix="Oreol", payment_plan_traffic_limit_gb=0, payment_plan_traffic_reset_period="NO_RESET", referral_discount_percent=5, referral_bonus_days=7, ), session_factory=session_factory, remnawave_client=_StubRemnawaveClient( created_user=make_remote_user( user_uuid="99999999-9999-9999-9999-999999999999", user_id=9, short_uuid="buyer999", username="oreol-user", telegram_id=1001, ) ), sync_service=_StubSyncService(), ) order = await service.create_order( telegram_id=1001, username=None, first_name="\u0410\u043b\u0435\u043a\u0441\u0435\u0439", last_name=None, language_code="ru", plan_code="30d", ) assert order.provision_username.startswith("Oreol-1001-") assert order.provision_username.endswith("-Aleksey") assert order.provision_username.isascii() async def test_create_order_marks_existing_access_as_renewal(session_factory) -> None: service = PaymentService( settings=Settings.model_construct( bot_admin_ids_raw="", payment_plans_raw="30:250", payment_transfer_text="Реквизиты", payment_internal_squad_uuids_raw="11111111-1111-1111-1111-111111111111", payment_user_tag="BOT", payment_external_squad_uuid="", payment_username_prefix="Oreol", payment_plan_traffic_limit_gb=0, payment_plan_traffic_reset_period="NO_RESET", referral_discount_percent=5, referral_bonus_days=7, ), session_factory=session_factory, remnawave_client=_StubRemnawaveClient( created_user=make_remote_user( user_uuid="77777777-7777-7777-7777-777777777777", user_id=7, short_uuid="buyer777", username="oreol-user", telegram_id=1001, ) ), sync_service=_StubSyncService(cached_users_by_telegram={1001: [object()]}), ) order = await service.create_order( telegram_id=1001, username="tester", first_name="Test", last_name=None, language_code="ru", plan_code="30d", ) assert order.extends_existing_access is True assert "продлит текущую подписку" in service.build_transfer_text(order=order) async def test_create_order_targets_external_subscription_linked_by_telegram_id( session_factory, ) -> None: existing_remote_user = make_remote_user( user_uuid="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", user_id=10, short_uuid="external10", username="custom-external", telegram_id=1001, ) remnawave_client = _StubRemnawaveClient( created_user=existing_remote_user, users_by_telegram_id={1001: [existing_remote_user]}, ) service = PaymentService( settings=Settings.model_construct( bot_admin_ids_raw="", payment_plans_raw="30:250", payment_transfer_text="Реквизиты", payment_internal_squad_uuids_raw="11111111-1111-1111-1111-111111111111", payment_user_tag="BOT", payment_external_squad_uuid="", payment_username_prefix="Oreol", payment_plan_traffic_limit_gb=0, payment_plan_traffic_reset_period="NO_RESET", referral_discount_percent=5, referral_bonus_days=7, ), session_factory=session_factory, remnawave_client=remnawave_client, sync_service=_StubSyncService(), ) order = await service.create_order( telegram_id=1001, username="buyer", first_name="Buyer", last_name=None, language_code="ru", plan_code="30d", ) assert order.extends_existing_access is True assert order.provision_username == "custom-external" async with session_factory() as session: stored_order = await session.scalar( select(PaymentOrder).where(PaymentOrder.order_uuid == order.order_uuid) ) assert stored_order is not None assert stored_order.remnawave_user_uuid == str(existing_remote_user.uuid) await service.mark_order_under_review(order_uuid=order.order_uuid, telegram_id=1001) access = await service.approve_order(order_uuid=order.order_uuid) assert access.is_renewal is True assert access.username == "custom-external" assert remnawave_client.created_bodies == [] assert remnawave_client.updated_bodies[-1]["username"] == "custom-external" async def test_approve_order_extends_existing_user_found_by_telegram_id(session_factory) -> None: existing_remote_user = make_remote_user( user_uuid="88888888-8888-8888-8888-888888888888", user_id=8, short_uuid="existing888", username="oreol-existing", telegram_id=1001, ) sync_service = _StubSyncService() remnawave_client = _StubRemnawaveClient( created_user=existing_remote_user, users_by_telegram_id={1001: [existing_remote_user]}, ) service = PaymentService( settings=_make_settings(), session_factory=session_factory, remnawave_client=remnawave_client, sync_service=sync_service, ) order_uuid = await _seed_order( session_factory, telegram_id=1001, status=REVIEW_PAYMENT_STATUS, ) access = await service.approve_order(order_uuid=order_uuid) assert access.remnawave_user_uuid == str(existing_remote_user.uuid) assert access.is_renewal is True assert remnawave_client.created_bodies == [] assert remnawave_client.updated_bodies