401 lines
18 KiB
Python
401 lines
18 KiB
Python
import datetime
|
|
import uuid
|
|
from contextlib import AbstractAsyncContextManager
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from src import domain, usecase
|
|
|
|
|
|
class MockDatabase:
|
|
def __init__(self) -> None:
|
|
self.users: dict[uuid.UUID, domain.User] = {}
|
|
self.users_by_telegram_id: dict[int, domain.User] = {}
|
|
self.login_tokens: dict[str, domain.LoginToken] = {}
|
|
self.target_channels: dict[uuid.UUID, domain.TargetChannel] = {}
|
|
self.target_channels_by_telegram_id: dict[int, domain.TargetChannel] = {}
|
|
self.external_channels: dict[uuid.UUID, domain.ExternalChannel] = {}
|
|
self.creatives: dict[uuid.UUID, domain.Creative] = {}
|
|
self.placements: dict[uuid.UUID, domain.Placement] = {}
|
|
self.subscribers: dict[uuid.UUID, domain.Subscriber] = {}
|
|
self.subscribers_by_telegram_id: dict[int, domain.Subscriber] = {}
|
|
self.subscriptions: dict[uuid.UUID, domain.Subscription] = {}
|
|
self.placement_views_histories: dict[uuid.UUID, domain.PlacementViewsHistory] = {}
|
|
self.m2m_target_external: dict[uuid.UUID, list[uuid.UUID]] = {}
|
|
|
|
def transaction(self) -> AbstractAsyncContextManager[None]:
|
|
return AsyncMock()
|
|
|
|
async def get_user(self, *, user_id: uuid.UUID | None = None, telegram_id: int | None = None) -> domain.User | None:
|
|
if user_id:
|
|
return self.users.get(user_id)
|
|
if telegram_id:
|
|
return self.users_by_telegram_id.get(telegram_id)
|
|
return None
|
|
|
|
async def create_user(self, user: domain.User) -> domain.User:
|
|
if not user.id:
|
|
user.id = uuid.uuid4()
|
|
if not hasattr(user, 'created_at') or user.created_at is None:
|
|
user.created_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(user, 'updated_at') or user.updated_at is None:
|
|
user.updated_at = datetime.datetime.now(datetime.UTC)
|
|
self.users[user.id] = user
|
|
self.users_by_telegram_id[user.telegram_id] = user
|
|
return user
|
|
|
|
async def create_login_token(
|
|
self, user_id: uuid.UUID, token: str, expires_at: datetime.datetime
|
|
) -> domain.LoginToken:
|
|
login_token = domain.LoginToken(user_id=user_id, token=token, expires_at=expires_at, used_at=None)
|
|
if not login_token.id:
|
|
login_token.id = uuid.uuid4()
|
|
if not hasattr(login_token, 'created_at') or login_token.created_at is None:
|
|
login_token.created_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(login_token, 'updated_at') or login_token.updated_at is None:
|
|
login_token.updated_at = datetime.datetime.now(datetime.UTC)
|
|
self.login_tokens[token] = login_token
|
|
return login_token
|
|
|
|
async def get_login_token(self, token: str) -> domain.LoginToken | None:
|
|
return self.login_tokens.get(token)
|
|
|
|
async def mark_token_as_used(self, token: str) -> None:
|
|
if token in self.login_tokens:
|
|
self.login_tokens[token].used_at = datetime.datetime.now(datetime.UTC)
|
|
|
|
async def get_target_channel(
|
|
self, user_id: uuid.UUID, channel_id: uuid.UUID | None = None, telegram_id: int | None = None
|
|
) -> domain.TargetChannel | None:
|
|
if channel_id:
|
|
channel = self.target_channels.get(channel_id)
|
|
if channel and channel.user_id == user_id:
|
|
return channel
|
|
if telegram_id:
|
|
channel = self.target_channels_by_telegram_id.get(telegram_id)
|
|
if channel and channel.user_id == user_id:
|
|
return channel
|
|
return None
|
|
|
|
async def upsert_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel:
|
|
existing = self.target_channels_by_telegram_id.get(channel.telegram_id)
|
|
if existing:
|
|
existing.title = channel.title
|
|
existing.username = channel.username
|
|
existing.status = channel.status
|
|
return existing
|
|
if not channel.id:
|
|
channel.id = uuid.uuid4()
|
|
if not hasattr(channel, 'created_at') or channel.created_at is None:
|
|
channel.created_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(channel, 'updated_at') or channel.updated_at is None:
|
|
channel.updated_at = datetime.datetime.now(datetime.UTC)
|
|
self.target_channels[channel.id] = channel
|
|
self.target_channels_by_telegram_id[channel.telegram_id] = channel
|
|
return channel
|
|
|
|
async def get_user_target_channels(self, user_id: uuid.UUID) -> list[domain.TargetChannel]:
|
|
return [c for c in self.target_channels.values() if c.user_id == user_id]
|
|
|
|
async def update_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel:
|
|
self.target_channels[channel.id] = channel
|
|
return channel
|
|
|
|
async def create_external_channel(self, channel: domain.ExternalChannel) -> domain.ExternalChannel:
|
|
if not channel.id:
|
|
channel.id = uuid.uuid4()
|
|
if not hasattr(channel, 'created_at') or channel.created_at is None:
|
|
channel.created_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(channel, 'updated_at') or channel.updated_at is None:
|
|
channel.updated_at = datetime.datetime.now(datetime.UTC)
|
|
self.external_channels[channel.id] = channel
|
|
return channel
|
|
|
|
async def add_external_channel_to_targets(
|
|
self, external_channel_id: uuid.UUID, target_channel_ids: list[uuid.UUID]
|
|
) -> None:
|
|
if external_channel_id not in self.m2m_target_external:
|
|
self.m2m_target_external[external_channel_id] = []
|
|
self.m2m_target_external[external_channel_id].extend(target_channel_ids)
|
|
|
|
async def get_external_channels_for_target(self, target_channel_id: uuid.UUID) -> list[domain.ExternalChannel]:
|
|
result = []
|
|
for ext_id, targets in self.m2m_target_external.items():
|
|
if target_channel_id in targets:
|
|
ext_ch = self.external_channels.get(ext_id)
|
|
if ext_ch:
|
|
result.append(ext_ch)
|
|
return result
|
|
|
|
async def get_external_channel(
|
|
self, user_id: uuid.UUID, channel_id: uuid.UUID | None = None, telegram_id: int | None = None
|
|
) -> domain.ExternalChannel | None:
|
|
if channel_id:
|
|
channel = self.external_channels.get(channel_id)
|
|
if channel and channel.user_id == user_id:
|
|
return channel
|
|
if telegram_id:
|
|
for channel in self.external_channels.values():
|
|
if channel.telegram_id == telegram_id and channel.user_id == user_id:
|
|
return channel
|
|
return None
|
|
|
|
async def remove_external_channel_from_target(
|
|
self, external_channel_id: uuid.UUID, target_channel_id: uuid.UUID
|
|
) -> None:
|
|
if external_channel_id in self.m2m_target_external:
|
|
targets = self.m2m_target_external[external_channel_id]
|
|
if target_channel_id in targets:
|
|
targets.remove(target_channel_id)
|
|
|
|
async def delete_external_channel(self, channel_id: uuid.UUID) -> None:
|
|
self.external_channels.pop(channel_id, None)
|
|
self.m2m_target_external.pop(channel_id, None)
|
|
|
|
async def create_creative(self, creative: domain.Creative) -> domain.Creative:
|
|
if not creative.id:
|
|
creative.id = uuid.uuid4()
|
|
if not hasattr(creative, 'created_at') or creative.created_at is None:
|
|
creative.created_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(creative, 'updated_at') or creative.updated_at is None:
|
|
creative.updated_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(creative, 'placements_count') or creative.placements_count is None:
|
|
creative.placements_count = 0
|
|
target_ch = self.target_channels.get(creative.target_channel_id)
|
|
if target_ch:
|
|
creative.target_channel = target_ch
|
|
if not hasattr(creative, 'status') or creative.status is None:
|
|
creative.status = domain.CreativeStatus.ACTIVE
|
|
self.creatives[creative.id] = creative
|
|
return creative
|
|
|
|
async def get_creative(self, user_id: uuid.UUID, creative_id: uuid.UUID) -> domain.Creative | None:
|
|
creative = self.creatives.get(creative_id)
|
|
if creative and creative.user_id == user_id:
|
|
return creative
|
|
return None
|
|
|
|
async def update_creative(self, creative: domain.Creative) -> domain.Creative:
|
|
self.creatives[creative.id] = creative
|
|
return creative
|
|
|
|
async def get_user_creatives(
|
|
self,
|
|
user_id: uuid.UUID,
|
|
*,
|
|
target_channel_id: uuid.UUID | None = None,
|
|
include_archived: bool = False,
|
|
) -> list[domain.Creative]:
|
|
result = [c for c in self.creatives.values() if c.user_id == user_id]
|
|
if target_channel_id:
|
|
result = [c for c in result if c.target_channel_id == target_channel_id]
|
|
if not include_archived:
|
|
result = [c for c in result if c.status == domain.CreativeStatus.ACTIVE]
|
|
return result
|
|
|
|
async def delete_creative(self, creative_id: uuid.UUID) -> None:
|
|
self.creatives.pop(creative_id, None)
|
|
|
|
async def create_placement(self, placement: domain.Placement) -> domain.Placement:
|
|
if not placement.id:
|
|
placement.id = uuid.uuid4()
|
|
if not hasattr(placement, 'created_at') or placement.created_at is None:
|
|
placement.created_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(placement, 'updated_at') or placement.updated_at is None:
|
|
placement.updated_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(placement, 'subscriptions_count') or placement.subscriptions_count is None:
|
|
placement.subscriptions_count = 0
|
|
if not hasattr(placement, 'status') or placement.status is None:
|
|
placement.status = domain.PlacementStatus.ACTIVE
|
|
if not hasattr(placement, 'views_availability') or placement.views_availability is None:
|
|
placement.views_availability = domain.PostViewsAvailability.UNKNOWN
|
|
# Загружаем связанные объекты
|
|
target_ch = self.target_channels.get(placement.target_channel_id)
|
|
if target_ch:
|
|
placement.target_channel = target_ch
|
|
ext_ch = self.external_channels.get(placement.external_channel_id)
|
|
if ext_ch:
|
|
placement.external_channel = ext_ch
|
|
creative = self.creatives.get(placement.creative_id)
|
|
if creative:
|
|
placement.creative = creative
|
|
self.placements[placement.id] = placement
|
|
return placement
|
|
|
|
async def get_placement(self, user_id: uuid.UUID, placement_id: uuid.UUID) -> domain.Placement | None:
|
|
placement = self.placements.get(placement_id)
|
|
if placement and placement.user_id == user_id:
|
|
return placement
|
|
return None
|
|
|
|
async def update_placement(self, placement: domain.Placement) -> domain.Placement:
|
|
self.placements[placement.id] = placement
|
|
return placement
|
|
|
|
async def get_user_placements(
|
|
self,
|
|
user_id: uuid.UUID,
|
|
*,
|
|
target_channel_id: uuid.UUID | None = None,
|
|
external_channel_id: uuid.UUID | None = None,
|
|
creative_id: uuid.UUID | None = None,
|
|
include_archived: bool = False,
|
|
) -> list[domain.Placement]:
|
|
result = [p for p in self.placements.values() if p.user_id == user_id]
|
|
if target_channel_id:
|
|
result = [p for p in result if p.target_channel_id == target_channel_id]
|
|
if external_channel_id:
|
|
result = [p for p in result if p.external_channel_id == external_channel_id]
|
|
if creative_id:
|
|
result = [p for p in result if p.creative_id == creative_id]
|
|
if not include_archived:
|
|
result = [p for p in result if p.status == domain.PlacementStatus.ACTIVE]
|
|
return sorted(result, key=lambda p: p.placement_date, reverse=True)
|
|
|
|
async def delete_placement(self, placement_id: uuid.UUID) -> None:
|
|
self.placements.pop(placement_id, None)
|
|
|
|
async def check_creative_in_use(self, creative_id: uuid.UUID) -> bool:
|
|
for placement in self.placements.values():
|
|
if placement.creative_id == creative_id and placement.status == domain.PlacementStatus.ACTIVE:
|
|
return True
|
|
return False
|
|
|
|
async def get_placement_by_invite_link(self, invite_link: str) -> domain.Placement | None:
|
|
for placement in self.placements.values():
|
|
if placement.invite_link == invite_link:
|
|
return placement
|
|
return None
|
|
|
|
async def get_subscriber(self, telegram_id: int) -> domain.Subscriber | None:
|
|
return self.subscribers_by_telegram_id.get(telegram_id)
|
|
|
|
async def create_subscriber(self, subscriber: domain.Subscriber) -> domain.Subscriber:
|
|
if not subscriber.id:
|
|
subscriber.id = uuid.uuid4()
|
|
if not hasattr(subscriber, 'created_at') or subscriber.created_at is None:
|
|
subscriber.created_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(subscriber, 'updated_at') or subscriber.updated_at is None:
|
|
subscriber.updated_at = datetime.datetime.now(datetime.UTC)
|
|
self.subscribers[subscriber.id] = subscriber
|
|
self.subscribers_by_telegram_id[subscriber.telegram_id] = subscriber
|
|
return subscriber
|
|
|
|
async def upsert_subscriber(self, subscriber: domain.Subscriber) -> domain.Subscriber:
|
|
existing = await self.get_subscriber(subscriber.telegram_id)
|
|
if existing:
|
|
existing.username = subscriber.username
|
|
existing.first_name = subscriber.first_name
|
|
existing.last_name = subscriber.last_name
|
|
return existing
|
|
return await self.create_subscriber(subscriber)
|
|
|
|
async def create_subscription(self, subscription: domain.Subscription) -> domain.Subscription:
|
|
if not subscription.id:
|
|
subscription.id = uuid.uuid4()
|
|
if not hasattr(subscription, 'created_at') or subscription.created_at is None:
|
|
subscription.created_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(subscription, 'updated_at') or subscription.updated_at is None:
|
|
subscription.updated_at = datetime.datetime.now(datetime.UTC)
|
|
# Загружаем связанные объекты
|
|
placement = self.placements.get(subscription.placement_id)
|
|
if placement:
|
|
subscription.placement = placement
|
|
subscriber = self.subscribers.get(subscription.subscriber_id)
|
|
if subscriber:
|
|
subscription.subscriber = subscriber
|
|
self.subscriptions[subscription.id] = subscription
|
|
return subscription
|
|
|
|
async def get_subscription_by_subscriber_and_placement(
|
|
self, subscriber_id: uuid.UUID, placement_id: uuid.UUID
|
|
) -> domain.Subscription | None:
|
|
for subscription in self.subscriptions.values():
|
|
if subscription.subscriber_id == subscriber_id and subscription.placement_id == placement_id:
|
|
return subscription
|
|
return None
|
|
|
|
async def create_placement_views_history(
|
|
self, history: domain.PlacementViewsHistory
|
|
) -> domain.PlacementViewsHistory:
|
|
if not history.id:
|
|
history.id = uuid.uuid4()
|
|
if not hasattr(history, 'created_at') or history.created_at is None:
|
|
history.created_at = datetime.datetime.now(datetime.UTC)
|
|
if not hasattr(history, 'updated_at') or history.updated_at is None:
|
|
history.updated_at = datetime.datetime.now(datetime.UTC)
|
|
placement = self.placements.get(history.placement_id)
|
|
if placement:
|
|
history.placement = placement
|
|
self.placement_views_histories[history.id] = history
|
|
return history
|
|
|
|
async def get_views_history(
|
|
self,
|
|
placement_id: uuid.UUID,
|
|
*,
|
|
from_date: datetime.datetime | None = None,
|
|
to_date: datetime.datetime | None = None,
|
|
) -> list[domain.PlacementViewsHistory]:
|
|
result = [h for h in self.placement_views_histories.values() if h.placement_id == placement_id]
|
|
if from_date:
|
|
result = [h for h in result if h.fetched_at >= from_date]
|
|
if to_date:
|
|
result = [h for h in result if h.fetched_at <= to_date]
|
|
return sorted(result, key=lambda h: h.fetched_at)
|
|
|
|
|
|
class MockTelegramWriter:
|
|
def __init__(self) -> None:
|
|
self.sent_messages: list[dict[str, Any]] = []
|
|
self.invite_link_counter = 0
|
|
|
|
async def send_message(self, text: str, chat_id: int) -> None:
|
|
self.sent_messages.append({'text': text, 'chat_id': chat_id, 'button': None})
|
|
|
|
async def send_message_with_button(self, text: str, chat_id: int, button_text: str, button_url: str) -> None:
|
|
self.sent_messages.append(
|
|
{'text': text, 'chat_id': chat_id, 'button': {'text': button_text, 'url': button_url}}
|
|
)
|
|
|
|
async def get_chat_member(self, chat_id: int, user_id: int) -> dict[str, Any]:
|
|
return {}
|
|
|
|
async def create_chat_invite_link(self, chat_id: int, requires_approval: bool = False) -> str:
|
|
self.invite_link_counter += 1
|
|
link_type = 'approval' if requires_approval else 'public'
|
|
return f'https://t.me/+{link_type}_link_{self.invite_link_counter}_{chat_id}'
|
|
|
|
async def get_post_views(self, post_url: str) -> int | None:
|
|
"""Mock implementation returns None (TDLib not integrated)."""
|
|
return None
|
|
|
|
|
|
class MockJWTEncoder:
|
|
def encode_access_token(self, user_id: uuid.UUID, telegram_id: int, username: str | None = None) -> str:
|
|
return f'jwt_token_{user_id}_{telegram_id}'
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_database() -> MockDatabase:
|
|
return MockDatabase()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_telegram() -> MockTelegramWriter:
|
|
return MockTelegramWriter()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_jwt() -> MockJWTEncoder:
|
|
return MockJWTEncoder()
|
|
|
|
|
|
@pytest.fixture
|
|
def usecases(
|
|
mock_database: MockDatabase, mock_telegram: MockTelegramWriter, mock_jwt: MockJWTEncoder
|
|
) -> usecase.Usecase:
|
|
return usecase.Usecase(database=mock_database, telegram_bot=mock_telegram, jwt_encoder=mock_jwt)
|