351 lines
13 KiB
Python
351 lines
13 KiB
Python
"""Тесты обработки подписок пользователей через invite_link."""
|
||
|
||
import datetime
|
||
|
||
import pytest
|
||
|
||
from src import domain, usecase
|
||
from tests.conftest import MockDatabase
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handle_subscription_success(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
|
||
"""Первая подписка пользователя по invite_link успешно создаётся."""
|
||
user = domain.User(telegram_id=123456, username='testuser')
|
||
await mock_database.create_user(user)
|
||
|
||
target_channel = domain.TargetChannel(
|
||
telegram_id=-1001234567890,
|
||
title='Target Channel',
|
||
username='targetchannel',
|
||
user_id=user.id,
|
||
is_active=True,
|
||
)
|
||
await mock_database.upsert_target_channel(target_channel)
|
||
|
||
external_channel = domain.ExternalChannel(
|
||
telegram_id=-1009876543210,
|
||
title='External Channel',
|
||
username='externalchannel',
|
||
description=None,
|
||
subscribers_count=50000,
|
||
user_id=user.id,
|
||
)
|
||
await mock_database.create_external_channel(external_channel)
|
||
|
||
creative = domain.Creative(
|
||
name='Test Creative',
|
||
text='Test text',
|
||
target_channel_id=target_channel.id,
|
||
user_id=user.id,
|
||
status=domain.CreativeStatus.ACTIVE,
|
||
)
|
||
await mock_database.create_creative(creative)
|
||
|
||
invite_link = 'https://t.me/+unique_link_123'
|
||
placement = domain.Placement(
|
||
target_channel_id=target_channel.id,
|
||
external_channel_id=external_channel.id,
|
||
creative_id=creative.id,
|
||
user_id=user.id,
|
||
placement_date=datetime.datetime.now(datetime.UTC),
|
||
invite_link=invite_link,
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PlacementStatus.ACTIVE,
|
||
subscriptions_count=0,
|
||
)
|
||
await mock_database.create_placement(placement)
|
||
|
||
# Симулируем подписку пользователя
|
||
subscriber_id = 999888777
|
||
subscriber_username = 'new_subscriber'
|
||
await usecases.handle_subscription(
|
||
user_telegram_id=subscriber_id,
|
||
username=subscriber_username,
|
||
invite_link=invite_link,
|
||
)
|
||
|
||
# Проверяем, что подписка создана
|
||
subscriber = await mock_database.get_subscriber(subscriber_id)
|
||
assert subscriber is not None
|
||
assert subscriber.telegram_id == subscriber_id
|
||
assert subscriber.username == subscriber_username
|
||
|
||
subscription = await mock_database.get_subscription_by_subscriber_and_placement(subscriber.id, placement.id)
|
||
assert subscription is not None
|
||
assert subscription.subscriber_id == subscriber.id
|
||
assert subscription.invite_link == invite_link
|
||
|
||
# Проверяем, что счётчик подписок у закупа увеличился
|
||
updated_placement = await mock_database.get_placement(user.id, placement.id)
|
||
assert updated_placement is not None
|
||
assert updated_placement.subscriptions_count == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handle_subscription_duplicate_ignored(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
|
||
"""Повторная подписка того же пользователя игнорируется - не создаёт дубликат."""
|
||
user = domain.User(telegram_id=123456, username='testuser')
|
||
await mock_database.create_user(user)
|
||
|
||
target_channel = domain.TargetChannel(
|
||
telegram_id=-1001234567890,
|
||
title='Target Channel',
|
||
username='targetchannel',
|
||
user_id=user.id,
|
||
is_active=True,
|
||
)
|
||
await mock_database.upsert_target_channel(target_channel)
|
||
|
||
external_channel = domain.ExternalChannel(
|
||
telegram_id=-1009876543210,
|
||
title='External Channel',
|
||
username='externalchannel',
|
||
description=None,
|
||
subscribers_count=30000,
|
||
user_id=user.id,
|
||
)
|
||
await mock_database.create_external_channel(external_channel)
|
||
|
||
creative = domain.Creative(
|
||
name='Test Creative',
|
||
text='Test text',
|
||
target_channel_id=target_channel.id,
|
||
user_id=user.id,
|
||
status=domain.CreativeStatus.ACTIVE,
|
||
)
|
||
await mock_database.create_creative(creative)
|
||
|
||
invite_link = 'https://t.me/+unique_link_456'
|
||
placement = domain.Placement(
|
||
target_channel_id=target_channel.id,
|
||
external_channel_id=external_channel.id,
|
||
creative_id=creative.id,
|
||
user_id=user.id,
|
||
placement_date=datetime.datetime.now(datetime.UTC),
|
||
invite_link=invite_link,
|
||
invite_link_type=domain.InviteLinkType.APPROVAL,
|
||
status=domain.PlacementStatus.ACTIVE,
|
||
subscriptions_count=0,
|
||
)
|
||
await mock_database.create_placement(placement)
|
||
|
||
subscriber_id = 888777666
|
||
subscriber_username = 'duplicate_user'
|
||
|
||
# Первая подписка
|
||
await usecases.handle_subscription(
|
||
user_telegram_id=subscriber_id,
|
||
username=subscriber_username,
|
||
invite_link=invite_link,
|
||
)
|
||
|
||
# Проверяем счётчик
|
||
updated_placement = await mock_database.get_placement(user.id, placement.id)
|
||
assert updated_placement is not None
|
||
assert updated_placement.subscriptions_count == 1
|
||
|
||
# Повторная попытка подписки того же пользователя
|
||
await usecases.handle_subscription(
|
||
user_telegram_id=subscriber_id,
|
||
username=subscriber_username,
|
||
invite_link=invite_link,
|
||
)
|
||
|
||
# Проверяем, что счётчик не увеличился
|
||
updated_placement_again = await mock_database.get_placement(user.id, placement.id)
|
||
assert updated_placement_again is not None
|
||
assert updated_placement_again.subscriptions_count == 1 # Остался 1, не увеличился
|
||
|
||
# Проверяем, что в БД только одна подписка этого подписчика
|
||
subscriber = await mock_database.get_subscriber(subscriber_id)
|
||
assert subscriber is not None
|
||
all_subscriptions = [s for s in mock_database.subscriptions.values() if s.subscriber_id == subscriber.id]
|
||
assert len(all_subscriptions) == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handle_subscription_invalid_invite_link(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
|
||
"""Подписка по несуществующему invite_link не создаётся."""
|
||
fake_invite_link = 'https://t.me/+fake_link_999'
|
||
|
||
# Пытаемся обработать подписку по несуществующей ссылке
|
||
await usecases.handle_subscription(
|
||
user_telegram_id=777666555,
|
||
username='unknown_user',
|
||
invite_link=fake_invite_link,
|
||
)
|
||
|
||
# Проверяем, что подписки не создано
|
||
assert len(mock_database.subscriptions) == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_multiple_users_subscription_same_placement(
|
||
usecases: usecase.Usecase, mock_database: MockDatabase
|
||
) -> None:
|
||
"""Несколько пользователей подписываются через один закуп - каждый учитывается."""
|
||
user = domain.User(telegram_id=123456, username='testuser')
|
||
await mock_database.create_user(user)
|
||
|
||
target_channel = domain.TargetChannel(
|
||
telegram_id=-1001234567890,
|
||
title='Target Channel',
|
||
username='targetchannel',
|
||
user_id=user.id,
|
||
is_active=True,
|
||
)
|
||
await mock_database.upsert_target_channel(target_channel)
|
||
|
||
external_channel = domain.ExternalChannel(
|
||
telegram_id=-1009876543210,
|
||
title='External Channel',
|
||
username='externalchannel',
|
||
description=None,
|
||
subscribers_count=40000,
|
||
user_id=user.id,
|
||
)
|
||
await mock_database.create_external_channel(external_channel)
|
||
|
||
creative = domain.Creative(
|
||
name='Viral Creative',
|
||
text='Viral content',
|
||
target_channel_id=target_channel.id,
|
||
user_id=user.id,
|
||
status=domain.CreativeStatus.ACTIVE,
|
||
)
|
||
await mock_database.create_creative(creative)
|
||
|
||
invite_link = 'https://t.me/+popular_link_789'
|
||
placement = domain.Placement(
|
||
target_channel_id=target_channel.id,
|
||
external_channel_id=external_channel.id,
|
||
creative_id=creative.id,
|
||
user_id=user.id,
|
||
placement_date=datetime.datetime.now(datetime.UTC),
|
||
invite_link=invite_link,
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PlacementStatus.ACTIVE,
|
||
subscriptions_count=0,
|
||
)
|
||
await mock_database.create_placement(placement)
|
||
|
||
# Подписываются три разных пользователя
|
||
users_data = [
|
||
(111222333, 'user_one'),
|
||
(444555666, 'user_two'),
|
||
(777888999, 'user_three'),
|
||
]
|
||
|
||
for telegram_id, username in users_data:
|
||
await usecases.handle_subscription(
|
||
user_telegram_id=telegram_id,
|
||
username=username,
|
||
invite_link=invite_link,
|
||
)
|
||
|
||
# Проверяем, что создано 3 подписки
|
||
placement_subscriptions = [s for s in mock_database.subscriptions.values() if s.placement_id == placement.id]
|
||
assert len(placement_subscriptions) == 3
|
||
|
||
# Проверяем, что счётчик правильный
|
||
updated_placement = await mock_database.get_placement(user.id, placement.id)
|
||
assert updated_placement is not None
|
||
assert updated_placement.subscriptions_count == 3
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_subscription_tracking_across_different_placements(
|
||
usecases: usecase.Usecase, mock_database: MockDatabase
|
||
) -> None:
|
||
"""Один пользователь может подписаться через разные закупы - каждая подписка учитывается отдельно."""
|
||
user = domain.User(telegram_id=123456, username='testuser')
|
||
await mock_database.create_user(user)
|
||
|
||
target_channel = domain.TargetChannel(
|
||
telegram_id=-1001234567890,
|
||
title='Target Channel',
|
||
username='targetchannel',
|
||
user_id=user.id,
|
||
is_active=True,
|
||
)
|
||
await mock_database.upsert_target_channel(target_channel)
|
||
|
||
external_channel = domain.ExternalChannel(
|
||
telegram_id=-1009876543210,
|
||
title='External Channel',
|
||
username='externalchannel',
|
||
description=None,
|
||
subscribers_count=25000,
|
||
user_id=user.id,
|
||
)
|
||
await mock_database.create_external_channel(external_channel)
|
||
|
||
creative = domain.Creative(
|
||
name='Test Creative',
|
||
text='Test text',
|
||
target_channel_id=target_channel.id,
|
||
user_id=user.id,
|
||
status=domain.CreativeStatus.ACTIVE,
|
||
)
|
||
await mock_database.create_creative(creative)
|
||
|
||
# Создаём два разных закупа с разными invite_link
|
||
placement1 = domain.Placement(
|
||
target_channel_id=target_channel.id,
|
||
external_channel_id=external_channel.id,
|
||
creative_id=creative.id,
|
||
user_id=user.id,
|
||
placement_date=datetime.datetime.now(datetime.UTC),
|
||
invite_link='https://t.me/+link_a',
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PlacementStatus.ACTIVE,
|
||
subscriptions_count=0,
|
||
)
|
||
placement2 = domain.Placement(
|
||
target_channel_id=target_channel.id,
|
||
external_channel_id=external_channel.id,
|
||
creative_id=creative.id,
|
||
user_id=user.id,
|
||
placement_date=datetime.datetime.now(datetime.UTC),
|
||
invite_link='https://t.me/+link_b',
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PlacementStatus.ACTIVE,
|
||
subscriptions_count=0,
|
||
)
|
||
await mock_database.create_placement(placement1)
|
||
await mock_database.create_placement(placement2)
|
||
|
||
subscriber_id = 555444333
|
||
|
||
# Пользователь подписывается через первый закуп
|
||
await usecases.handle_subscription(
|
||
user_telegram_id=subscriber_id,
|
||
username='multi_subscriber',
|
||
invite_link=placement1.invite_link,
|
||
)
|
||
|
||
# Тот же пользователь "подписывается" через второй закуп
|
||
# (технически невозможно, но тестируем логику)
|
||
await usecases.handle_subscription(
|
||
user_telegram_id=subscriber_id,
|
||
username='multi_subscriber',
|
||
invite_link=placement2.invite_link,
|
||
)
|
||
|
||
# Проверяем, что у каждого закупа свой счётчик
|
||
updated_placement1 = await mock_database.get_placement(user.id, placement1.id)
|
||
updated_placement2 = await mock_database.get_placement(user.id, placement2.id)
|
||
|
||
assert updated_placement1 is not None
|
||
assert updated_placement1.subscriptions_count == 1
|
||
|
||
assert updated_placement2 is not None
|
||
assert updated_placement2.subscriptions_count == 1
|
||
|
||
# Проверяем, что создано 2 подписки
|
||
subscriber = await mock_database.get_subscriber(subscriber_id)
|
||
assert subscriber is not None
|
||
user_subscriptions = [s for s in mock_database.subscriptions.values() if s.subscriber_id == subscriber.id]
|
||
assert len(user_subscriptions) == 2
|