425 lines
15 KiB
Python
425 lines
15 KiB
Python
"""Тесты работы с просмотрами рекламных постов."""
|
||
|
||
import datetime
|
||
|
||
import pytest
|
||
|
||
from src import domain, dto, usecase
|
||
from tests.conftest import MockDatabase
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_views_no_ad_post_url(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
|
||
"""Попытка получить просмотры для закупа без ad_post_url возвращает unavailable."""
|
||
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)
|
||
|
||
purchase = domain.Purchase(
|
||
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/+link1',
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PurchaseStatus.ACTIVE,
|
||
ad_post_url=None, # Нет ссылки на пост
|
||
)
|
||
await mock_database.create_purchase(purchase)
|
||
|
||
input_data = dto.FetchViewsInputManually(
|
||
purchase_id=purchase.id,
|
||
user_id=user.id,
|
||
)
|
||
|
||
result = await usecases.fetch_views(input=input_data)
|
||
|
||
assert result.views_count is None
|
||
assert result.views_availability == domain.PostViewsAvailability.UNAVAILABLE
|
||
assert result.error_message == 'No ad_post_url specified'
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_fetch_views_api_unavailable(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
|
||
"""Если Telegram API недоступен (возвращает None), статус = unavailable."""
|
||
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)
|
||
|
||
purchase = domain.Purchase(
|
||
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/+link2',
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PurchaseStatus.ACTIVE,
|
||
ad_post_url='https://t.me/externalchannel/123',
|
||
)
|
||
await mock_database.create_purchase(purchase)
|
||
|
||
input_data = dto.FetchViewsInputManually(
|
||
purchase_id=purchase.id,
|
||
user_id=user.id,
|
||
)
|
||
|
||
# Mock telegram writer returns None (TDLib not integrated)
|
||
result = await usecases.fetch_views(input=input_data)
|
||
|
||
assert result.views_count is None
|
||
assert result.views_availability == domain.PostViewsAvailability.UNAVAILABLE
|
||
assert 'Failed to fetch views' in result.error_message
|
||
|
||
# Проверяем, что purchase обновлён
|
||
updated_purchase = await mock_database.get_purchase(user.id, purchase.id)
|
||
assert updated_purchase is not None
|
||
assert updated_purchase.views_availability == domain.PostViewsAvailability.UNAVAILABLE
|
||
assert updated_purchase.last_views_fetch_at is not None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_views_manually(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='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)
|
||
|
||
purchase = domain.Purchase(
|
||
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/+link3',
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PurchaseStatus.ACTIVE,
|
||
ad_post_url='https://t.me/externalchannel/456',
|
||
)
|
||
await mock_database.create_purchase(purchase)
|
||
|
||
input_data = dto.UpdateViewsManuallyInput(
|
||
purchase_id=purchase.id,
|
||
user_id=user.id,
|
||
views_count=15000,
|
||
)
|
||
|
||
result = await usecases.update_views_manually(input=input_data)
|
||
|
||
# Проверяем возвращённый результат
|
||
assert result.views_count == 15000
|
||
assert result.views_availability == domain.PostViewsAvailability.MANUAL
|
||
assert result.last_views_fetch_at is not None
|
||
|
||
# Проверяем, что создан снимок
|
||
snapshots = [s for s in mock_database.views_snapshots.values() if s.purchase_id == purchase.id]
|
||
assert len(snapshots) == 1
|
||
assert snapshots[0].views_count == 15000
|
||
|
||
# Проверяем, что закуп обновлён
|
||
updated_purchase = await mock_database.get_purchase(user.id, purchase.id)
|
||
assert updated_purchase is not None
|
||
assert updated_purchase.views_count == 15000
|
||
assert updated_purchase.views_availability == domain.PostViewsAvailability.MANUAL
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_views_history_empty(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)
|
||
|
||
purchase = domain.Purchase(
|
||
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/+link4',
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PurchaseStatus.ACTIVE,
|
||
)
|
||
await mock_database.create_purchase(purchase)
|
||
|
||
input_data = dto.GetViewsHistoryInput(
|
||
purchase_id=purchase.id,
|
||
user_id=user.id,
|
||
)
|
||
|
||
result = await usecases.get_views_history(input=input_data)
|
||
|
||
assert len(result.snapshots) == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_views_history_with_snapshots(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=35000,
|
||
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)
|
||
|
||
purchase = domain.Purchase(
|
||
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/+link5',
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PurchaseStatus.ACTIVE,
|
||
)
|
||
await mock_database.create_purchase(purchase)
|
||
|
||
# Создаём несколько снимков
|
||
now = datetime.datetime.now(datetime.UTC)
|
||
snapshot1 = domain.ViewsSnapshot(
|
||
purchase_id=purchase.id,
|
||
views_count=1000,
|
||
fetched_at=now - datetime.timedelta(hours=2),
|
||
)
|
||
snapshot2 = domain.ViewsSnapshot(
|
||
purchase_id=purchase.id,
|
||
views_count=1500,
|
||
fetched_at=now - datetime.timedelta(hours=1),
|
||
)
|
||
snapshot3 = domain.ViewsSnapshot(
|
||
purchase_id=purchase.id,
|
||
views_count=2000,
|
||
fetched_at=now,
|
||
)
|
||
await mock_database.create_views_snapshot(snapshot1)
|
||
await mock_database.create_views_snapshot(snapshot2)
|
||
await mock_database.create_views_snapshot(snapshot3)
|
||
|
||
input_data = dto.GetViewsHistoryInput(
|
||
purchase_id=purchase.id,
|
||
user_id=user.id,
|
||
)
|
||
|
||
result = await usecases.get_views_history(input=input_data)
|
||
|
||
assert len(result.snapshots) == 3
|
||
# Проверяем сортировку по времени (старые первые)
|
||
assert result.snapshots[0].views_count == 1000
|
||
assert result.snapshots[1].views_count == 1500
|
||
assert result.snapshots[2].views_count == 2000
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_views_history_with_date_filter(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=28000,
|
||
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)
|
||
|
||
purchase = domain.Purchase(
|
||
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/+link6',
|
||
invite_link_type=domain.InviteLinkType.PUBLIC,
|
||
status=domain.PurchaseStatus.ACTIVE,
|
||
)
|
||
await mock_database.create_purchase(purchase)
|
||
|
||
# Создаём снимки с разными датами
|
||
now = datetime.datetime.now(datetime.UTC)
|
||
snapshot1 = domain.ViewsSnapshot(
|
||
purchase_id=purchase.id,
|
||
views_count=1000,
|
||
fetched_at=now - datetime.timedelta(days=3),
|
||
)
|
||
snapshot2 = domain.ViewsSnapshot(
|
||
purchase_id=purchase.id,
|
||
views_count=1500,
|
||
fetched_at=now - datetime.timedelta(days=1),
|
||
)
|
||
snapshot3 = domain.ViewsSnapshot(
|
||
purchase_id=purchase.id,
|
||
views_count=2000,
|
||
fetched_at=now,
|
||
)
|
||
await mock_database.create_views_snapshot(snapshot1)
|
||
await mock_database.create_views_snapshot(snapshot2)
|
||
await mock_database.create_views_snapshot(snapshot3)
|
||
|
||
# Запрашиваем историю за последние 2 дня
|
||
input_data = dto.GetViewsHistoryInput(
|
||
purchase_id=purchase.id,
|
||
user_id=user.id,
|
||
from_date=now - datetime.timedelta(days=2),
|
||
)
|
||
|
||
result = await usecases.get_views_history(input=input_data)
|
||
|
||
# Должны вернуться только snapshot2 и snapshot3
|
||
assert len(result.snapshots) == 2
|
||
assert result.snapshots[0].views_count == 1500
|
||
assert result.snapshots[1].views_count == 2000
|