y0zhyck/RemnaWaveBOT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add

Browse Source
This commit is contained in:
Evgeniy Spirin
2026-07-09 16:45:17 +03:00
parent 5c7a9710f3
commit da24ed8ba7
23 changed files with 1542 additions and 358 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
tests/test_bot_config_service.py
View File

@@ -19,14 +19,12 @@ def _make_settings() -> Settings:
referral_discount_percent=5,
referral_bonus_days=7,
payment_plans_raw="30:250,180:600",
payment_product_plans_raw="",
payment_transfer_text="Реквизиты",
payment_support_text="Напишите в поддержку",
payment_success_text="Доступ выдан",
payment_plan_traffic_limit_gb=0,
payment_plan_traffic_reset_period="NO_RESET",
payment_internal_squad_uuids_raw="uuid-1",
payment_vpn_squad_uuids_raw="vpn-uuid",
payment_white_squad_uuids_raw="white-uuid",
payment_external_squad_uuid="",
payment_username_prefix="Oreol",
payment_user_tag="BOT",
@@ -50,13 +48,13 @@ async def test_bot_config_service_updates_runtime_snapshot(session_factory) -> N
updated_by_telegram_id=1,
)
await service.update_setting(
key="payment_product_plans_raw",
raw_value="vpn_white:30:450,white:30:250,vpn:30:200",
key="referral_enabled",
raw_value="off",
updated_by_telegram_id=1,
)
await service.update_setting(
key="referral_enabled",
raw_value="off",
key="payment_success_text",
raw_value="Новый текст после оплаты",
updated_by_telegram_id=1,
)
@@ -64,10 +62,10 @@ async def test_bot_config_service_updates_runtime_snapshot(session_factory) -> N
assert snapshot.bot_brand_name == "NEW BRAND"
assert snapshot.referral_enabled is False
assert [(plan.code, plan.days, plan.amount_rub) for plan in snapshot.payment_plans] == [
("vpn_white", 30, 450),
("white", 30, 250),
("vpn", 30, 200),
assert snapshot.payment_success_text == "Новый текст после оплаты"
assert [(plan.days, plan.amount_rub) for plan in snapshot.payment_plans] == [
(30, 300),
(90, 700),
]
@@ -81,4 +79,3 @@ async def test_bot_config_service_formats_values_for_admin_ui(session_factory) -
assert BotConfigService.format_value(snapshot=snapshot, key="bot_public_username") == "@oreol_vpn_bot"
assert BotConfigService.format_value(snapshot=snapshot, key="referral_enabled") == "on"
assert BotConfigService.format_value(snapshot=snapshot, key="payment_plan_traffic_limit_gb") == "0 GB (unlimited)"
assert BotConfigService.format_value(snapshot=snapshot, key="payment_vpn_squad_uuids_raw") == "vpn-uuid"

19
tests/test_config.py
View File

@@ -79,22 +79,3 @@ def test_payment_plans_are_parsed() -> None:
assert [plan.code for plan in plans] == ["30d", "180d", "365d"]
assert [plan.days for plan in plans] == [30, 180, 365]
assert [plan.amount_rub for plan in plans] == [250, 600, 1000]
def test_payment_product_plans_override_legacy_plans() -> None:
settings = Settings.model_construct(
payment_product_plans_raw="vpn_white:30:450,white:30:250,vpn:30:200",
payment_plans_raw="30:999",
payment_plan_duration_days=30,
)
plans = settings.payment_plans
assert [plan.code for plan in plans] == ["vpn_white", "white", "vpn"]
assert [plan.title for plan in plans] == ["VPN + Белые списки", "Белые списки", "VPN"]
assert [plan.amount_rub for plan in plans] == [450, 250, 200]
assert [plan.squad_groups for plan in plans] == [
("vpn", "white"),
("white",),
("vpn",),
]

13
tests/test_formatters.py
View File

@@ -75,6 +75,19 @@ def test_format_issued_access_contains_subscription_data() -> None:
assert "/me</code>" not in text
def test_format_issued_access_includes_custom_success_text() -> None:
text = format_issued_access(
subscription_url="https://example.com/sub",
expire_at=datetime(2026, 5, 21, 18, 0),
traffic_limit_bytes=0,
brand_name="OREOL VPN",
success_text="Откройте приложение и импортируйте ссылку <важно>.",
)
assert "Откройте приложение и импортируйте ссылку" in text
assert "&lt;важно&gt;" in text
def test_format_issued_access_for_renewal_mentions_extension() -> None:
text = format_issued_access(
subscription_url="https://example.com/sub",

230
tests/test_payment_service.py
View File

@@ -21,15 +21,9 @@ from tests.helpers import make_remote_user
class _StubSyncService:
def __init__(
self,
*,
cached_users_by_telegram: dict[int, list] | None = None,
active_promo_by_telegram: dict[int, object] | None = None,
) -> None:
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 {}
self.active_promo_by_telegram = active_promo_by_telegram or {}
async def get_referral_summary(self, telegram_id: int):
return type(
@@ -50,9 +44,6 @@ class _StubSyncService:
async def get_cached_users_for_telegram(self, telegram_id: int):
return list(self.cached_users_by_telegram.get(telegram_id, []))
async def get_active_promo_for_user(self, telegram_id: int):
return self.active_promo_by_telegram.get(telegram_id)
class _StubRemnawaveClient:
def __init__(self, *, created_user, users_by_telegram_id: dict[int, list] | None = None) -> None:
@@ -62,6 +53,10 @@ class _StubRemnawaveClient:
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):
@@ -107,9 +102,6 @@ async def _seed_order(
telegram_id: int,
status: str,
order_uuid: str | None = None,
plan_code: str = "30d",
plan_title: str = "OREOL VPN на 30 дней",
amount_stars: int = 250,
) -> str:
resolved_order_uuid = order_uuid or str(uuid4())
@@ -128,12 +120,12 @@ async def _seed_order(
order_uuid=resolved_order_uuid,
telegram_user_id=user.id,
telegram_id=telegram_id,
plan_code=plan_code,
plan_title=plan_title,
plan_code="30d",
plan_title="OREOL VPN на 30 дней",
plan_duration_days=30,
traffic_limit_bytes=0,
traffic_limit_strategy="NO_RESET",
amount_stars=amount_stars,
amount_stars=250,
currency="RUB",
status=status,
invoice_payload=f"payload-{resolved_order_uuid}",
@@ -167,29 +159,7 @@ async def _seed_referral_invite(
def _make_settings() -> Settings:
return Settings.model_construct(
bot_admin_ids_raw="",
payment_product_plans_raw="",
payment_internal_squad_uuids_raw="11111111-1111-1111-1111-111111111111",
payment_vpn_squad_uuids_raw="",
payment_white_squad_uuids_raw="",
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,
)
def _make_product_settings() -> Settings:
return Settings.model_construct(
bot_admin_ids_raw="",
payment_plans_raw="30:999",
payment_product_plans_raw="vpn_white:30:450,white:30:250,vpn:30:200",
payment_transfer_text="Реквизиты",
payment_internal_squad_uuids_raw="",
payment_vpn_squad_uuids_raw="11111111-1111-1111-1111-111111111111",
payment_white_squad_uuids_raw="22222222-2222-2222-2222-222222222222",
payment_user_tag="BOT",
payment_external_squad_uuid="",
payment_username_prefix="Oreol",
@@ -292,37 +262,6 @@ async def test_approve_order_fulfills_and_refreshes_cache(session_factory) -> No
assert order.remnawave_user_uuid == str(remote_user.uuid)
async def test_approve_product_order_sends_product_squads_to_remnawave(session_factory) -> None:
remote_user = make_remote_user(
user_uuid="99999999-9999-9999-9999-999999999999",
user_id=9,
short_uuid="product999",
username="oreol-product",
telegram_id=1001,
)
remnawave_client = _StubRemnawaveClient(created_user=remote_user)
service = PaymentService(
settings=_make_product_settings(),
session_factory=session_factory,
remnawave_client=remnawave_client,
sync_service=_StubSyncService(),
)
order_uuid = await _seed_order(
session_factory,
telegram_id=1001,
status=REVIEW_PAYMENT_STATUS,
plan_code="white",
plan_title="OREOL VPN · Белые списки",
amount_stars=250,
)
await service.approve_order(order_uuid=order_uuid)
assert remnawave_client.created_bodies[0]["activeInternalSquads"] == [
"22222222-2222-2222-2222-222222222222",
]
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",
@@ -475,16 +414,30 @@ async def test_create_order_generates_unique_provision_username(session_factory)
assert second_order.extends_existing_access is False
async def test_product_plan_uses_matching_squads_and_price(session_factory) -> None:
async def test_create_order_transliterates_russian_first_name_without_username(
session_factory,
) -> None:
service = PaymentService(
settings=_make_product_settings(),
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-product",
username="oreol-user",
telegram_id=1001,
)
),
@@ -493,75 +446,16 @@ async def test_product_plan_uses_matching_squads_and_price(session_factory) -> N
order = await service.create_order(
telegram_id=1001,
username="tester",
first_name="Test",
username=None,
first_name="\u0410\u043b\u0435\u043a\u0441\u0435\u0439",
last_name=None,
language_code="ru",
plan_code="vpn_white",
plan_code="30d",
)
assert order.plan.label == "VPN + Белые списки"
assert order.plan.amount_rub == 450
assert order.plan.internal_squad_uuids == [
"11111111-1111-1111-1111-111111111111",
"22222222-2222-2222-2222-222222222222",
]
async def test_promo_code_discounts_only_matching_plan(session_factory) -> None:
promo = type(
"PromoStub",
(),
{
"code": "WHITE50",
"discount_percent": 50,
"plan_code": "white",
"expires_at": None,
},
)()
service = PaymentService(
settings=_make_product_settings(),
session_factory=session_factory,
remnawave_client=_StubRemnawaveClient(
created_user=make_remote_user(
user_uuid="10101010-1010-1010-1010-101010101010",
user_id=10,
short_uuid="buyer101",
username="oreol-promo",
telegram_id=1001,
)
),
sync_service=_StubSyncService(active_promo_by_telegram={1001: promo}),
)
plans = await service.get_available_plans(telegram_id=1001)
assert [(plan.code, plan.amount_rub, plan.discount_code) for plan in plans] == [
("vpn_white", 450, ""),
("white", 125, "WHITE50"),
("vpn", 200, ""),
]
order = await service.create_order(
telegram_id=1001,
username="tester",
first_name="Test",
last_name=None,
language_code="ru",
plan_code="white",
)
assert order.plan.amount_rub == 125
assert order.plan.discount_source == "promo"
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.amount_stars == 125
assert stored_order.error_message == "WHITE50"
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:
@@ -605,6 +499,68 @@ async def test_create_order_marks_existing_access_as_renewal(session_factory) ->
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",

62
tests/test_sync_service.py
View File

@@ -1,19 +1,11 @@
from __future__ import annotations
from datetime import datetime, timedelta
from datetime import datetime
from sqlalchemy import select
from app.config import Settings
from app.db.base import utcnow
from app.db.models import (
PromoCode,
PromoCodeApplication,
ReferralCode,
ReferralInvite,
RemnawaveUser,
TelegramUser,
)
from app.db.models import ReferralCode, ReferralInvite, RemnawaveUser, TelegramUser
from app.schemas.remnawave import PaginatedUsers
from app.services.sync_service import SyncService
from tests.helpers import make_remote_user
@@ -221,56 +213,6 @@ async def test_get_admin_telegram_user_profile_returns_referral_and_access_data(
assert profile.accesses[0].record.username == "oreol-profile"
async def test_promo_code_can_be_applied_through_referral_input(session_factory) -> None:
service = SyncService(
settings=Settings.model_construct(
bot_admin_ids_raw="",
sync_subscription_history=False,
),
session_factory=session_factory,
remnawave_client=_StubRemnawaveClient([]),
)
promo = await service.create_promo_code(
code="white50",
plan_code="white",
discount_percent=50,
expires_at=utcnow() + timedelta(days=7),
created_by_telegram_id=1001,
)
applied_code = await service.apply_referral_code(
telegram_id=2001,
username="buyer",
first_name="Buyer",
last_name=None,
language_code="ru",
referral_code="white50",
)
summary = await service.get_referral_summary(2001)
active_promo = await service.get_active_promo_for_user(2001)
assert promo.code == "WHITE50"
assert applied_code == "WHITE50"
assert summary.applied_referral_code == ""
assert summary.applied_promo_code == "WHITE50"
assert summary.applied_promo_discount_percent == 50
assert summary.applied_promo_plan_code == "white"
assert active_promo is not None
assert active_promo.code == "WHITE50"
async with session_factory() as session:
promo_record = await session.scalar(
select(PromoCode).where(PromoCode.code == "WHITE50")
)
application = await session.scalar(
select(PromoCodeApplication).where(PromoCodeApplication.telegram_id == 2001)
)
assert promo_record is not None
assert application is not None
assert application.promo_code_id == promo_record.id
async def test_sync_all_users_does_not_autoflush_half_built_user(
session_factory,
) -> None: