From b43042c745806799bb9f442242f1c39870f0ed8d Mon Sep 17 00:00:00 2001 From: Artem Tsyrulnikov Date: Mon, 10 Nov 2025 16:21:14 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20=D1=82=D0=B5=D1=81=D1=82=D1=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pyproject.toml | 2 + src/adapter/postgres.py | 2 +- src/controller/http/creatives.py | 2 - .../telegram_callback/my_chat_member.py | 1 + src/dto/creative.py | 5 + src/dto/target_channel.py | 2 +- .../disconnect_target_chan_by_tg_id.py | 8 +- .../update_target_chan_permissions.py | 8 +- tests/conftest.py | 236 +++++++++++++++++ tests/test_auth.py | 107 ++++++++ tests/test_creatives.py | 237 ++++++++++++++++++ tests/test_external_channels.py | 177 +++++++++++++ tests/test_target_channels.py | 168 +++++++++++++ 13 files changed, 949 insertions(+), 6 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/test_auth.py create mode 100644 tests/test_creatives.py create mode 100644 tests/test_external_channels.py create mode 100644 tests/test_target_channels.py diff --git a/pyproject.toml b/pyproject.toml index 7eb6f2a..bd16cce 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,8 @@ select = [ "UP", # pyupgrade ] ignore = ["B008"] +fixable = ["ALL"] +unfixable = [] [tool.ruff.format] quote-style = "single" diff --git a/src/adapter/postgres.py b/src/adapter/postgres.py index a1e0779..617a3bc 100644 --- a/src/adapter/postgres.py +++ b/src/adapter/postgres.py @@ -164,7 +164,7 @@ class Postgres(DatabaseBase): await self.session.execute(stmt) await self.session.flush() - async def get_creative(self, creative_id: uuid.UUID, user_id: uuid.UUID) -> domain.Creative | None: + async def get_creative(self, user_id: uuid.UUID, creative_id: uuid.UUID) -> domain.Creative | None: stmt = ( select(domain.Creative) .options(selectinload(domain.Creative.target_channel)) diff --git a/src/controller/http/creatives.py b/src/controller/http/creatives.py index ec5e267..aa48942 100644 --- a/src/controller/http/creatives.py +++ b/src/controller/http/creatives.py @@ -62,12 +62,10 @@ async def update_creative( @creatives_router.post('/{creative_id}/archive') async def archive_creative( creative_id: uuid.UUID, - request: dto.ArchiveCreativeInput, current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)], ) -> dto.CreativeOutput: return await dependencies.get_usecase().archive_creative( creative_id=creative_id, - input=request, user_id=current_user.user_id, ) diff --git a/src/controller/telegram_callback/my_chat_member.py b/src/controller/telegram_callback/my_chat_member.py index 426f402..b5296d5 100644 --- a/src/controller/telegram_callback/my_chat_member.py +++ b/src/controller/telegram_callback/my_chat_member.py @@ -31,6 +31,7 @@ async def on_bot_added_to_chat(event: ChatMemberUpdated) -> None: if bot_removed: disconnect_input = dto.DisconnectTargetChanByTgIdInput( telegram_id=event.chat.id, + user_telegram_id=event.from_user.id, ) try: await usecase.disconnect_target_chan_by_tg_id(input=disconnect_input) diff --git a/src/dto/creative.py b/src/dto/creative.py index 49bd1e1..8e332b6 100644 --- a/src/dto/creative.py +++ b/src/dto/creative.py @@ -43,6 +43,11 @@ class UpdateCreativeInput(pydantic.BaseModel): text: str | None = None +class ArchiveCreativeInput(pydantic.BaseModel): + creative_id: uuid.UUID + user_id: uuid.UUID + + class DeleteCreativeInput(pydantic.BaseModel): creative_id: uuid.UUID user_id: uuid.UUID diff --git a/src/dto/target_channel.py b/src/dto/target_channel.py index 3d3395c..0e4d9b8 100644 --- a/src/dto/target_channel.py +++ b/src/dto/target_channel.py @@ -46,7 +46,7 @@ class DisconnectTargetChanInput(pydantic.BaseModel): class DisconnectTargetChanByTgIdInput(pydantic.BaseModel): telegram_id: int - user_id: uuid.UUID + user_telegram_id: int class UpdateTargetChanPermissionsInput(pydantic.BaseModel): diff --git a/src/usecase/target_channel/disconnect_target_chan_by_tg_id.py b/src/usecase/target_channel/disconnect_target_chan_by_tg_id.py index 6be718e..3eea08a 100644 --- a/src/usecase/target_channel/disconnect_target_chan_by_tg_id.py +++ b/src/usecase/target_channel/disconnect_target_chan_by_tg_id.py @@ -10,8 +10,14 @@ log = logging.getLogger(__name__) async def disconnect_target_chan_by_tg_id(self: 'Usecase', input: dto.DisconnectTargetChanByTgIdInput) -> None: + # Получаем user_id по telegram_id + user = await self.database.get_user(telegram_id=input.user_telegram_id) + if not user: + log.warning(f'User with telegram_id {input.user_telegram_id} not found when disconnecting channel') + return + async with self.database.transaction(): - channel = await self.database.get_target_channel(input.user_id, telegram_id=input.telegram_id) + channel = await self.database.get_target_channel(user.id, telegram_id=input.telegram_id) if not channel: log.warning(f'Target channel {input.telegram_id} not found') raise domain.TargetChannelNotFound() diff --git a/src/usecase/target_channel/update_target_chan_permissions.py b/src/usecase/target_channel/update_target_chan_permissions.py index c21ae7b..8906bd7 100644 --- a/src/usecase/target_channel/update_target_chan_permissions.py +++ b/src/usecase/target_channel/update_target_chan_permissions.py @@ -16,8 +16,14 @@ async def update_target_chan_permissions(self: 'Usecase', input: dto.UpdateTarge if not input.permissions.can_restrict_members: missing_permissions.append('Управление пользователями') + # Получаем user_id по telegram_id + user = await self.database.get_user(telegram_id=input.user_telegram_id) + if not user: + log.warning(f'User with telegram_id {input.user_telegram_id} not found when updating channel permissions') + return + async with self.database.transaction(): - channel = await self.database.get_target_channel(user_id, telegram_id=input.telegram_id) + channel = await self.database.get_target_channel(user.id, telegram_id=input.telegram_id) if not channel: log.warning(f'Target channel {input.telegram_id} not found when permissions changed') raise domain.TargetChannelNotFound() diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..225f349 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,236 @@ +import datetime +import uuid +from typing import Any, AsyncContextManager +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.m2m_target_external: dict[uuid.UUID, list[uuid.UUID]] = {} + + def transaction(self) -> AsyncContextManager[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, 'purchases_count') or creative.purchases_count is None: + creative.purchases_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) + + +class MockTelegramWriter: + def __init__(self) -> None: + self.sent_messages: list[dict[str, Any]] = [] + + 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 {} + + +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_writer=mock_telegram, jwt_encoder=mock_jwt) diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..f668f25 --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,107 @@ +"""Тесты авторизации через Telegram.""" + +import datetime + +import pytest +from fastapi import HTTPException + +from src import domain, dto, usecase +from tests.conftest import MockDatabase, MockTelegramWriter + + +@pytest.mark.asyncio +async def test_new_user_registration(usecases: usecase.Usecase, mock_telegram: MockTelegramWriter) -> None: + """Новый пользователь регистрируется через Telegram.""" + input_data = dto.TelegramLoginInput(telegram_id=123456, username='testuser', chat_id=123456) + + await usecases.telegram_login(input_data) + + # Проверяем, что пользователь создан + user = await usecases.database.get_user(telegram_id=123456) + assert user is not None + assert user.telegram_id == 123456 + assert user.username == 'testuser' + + # Проверяем, что бот отправил сообщение с кнопкой + assert len(mock_telegram.sent_messages) == 1 + msg = mock_telegram.sent_messages[0] + assert msg['button'] is not None + assert 'token=' in msg['button']['url'] + + +@pytest.mark.asyncio +async def test_existing_user_login(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Существующий пользователь получает токен для входа.""" + # Создаём пользователя + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + input_data = dto.TelegramLoginInput(telegram_id=123456, username='testuser', chat_id=123456) + await usecases.telegram_login(input_data) + + # Проверяем, что не создался второй пользователь + users_count = len(mock_database.users) + assert users_count == 1 + + +@pytest.mark.asyncio +async def test_validate_login_token_success(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Валидный токен обменивается на JWT.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + token = 'test_token_123' + expires_at = datetime.datetime.now(datetime.UTC) + datetime.timedelta(minutes=10) + await mock_database.create_login_token(user.id, token, expires_at) + + result = await usecases.validate_login_token(dto.ValidateLoginTokenInput(token=token)) + + assert result.access_token.startswith('jwt_token_') + assert str(user.id) in result.access_token + + # Проверяем, что токен помечен как использованный + login_token = await mock_database.get_login_token(token) + assert login_token is not None + assert login_token.used_at is not None + + +@pytest.mark.asyncio +async def test_validate_login_token_not_found(usecases: usecase.Usecase) -> None: + """Несуществующий токен вызывает ошибку.""" + with pytest.raises(HTTPException) as exc_info: + await usecases.validate_login_token(dto.ValidateLoginTokenInput(token='nonexistent')) + assert exc_info.value.status_code == 404 + assert 'Login token not found' in str(exc_info.value.detail) + + +@pytest.mark.asyncio +async def test_validate_login_token_already_used(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Использованный токен вызывает ошибку.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + token = 'test_token_123' + expires_at = datetime.datetime.now(datetime.UTC) + datetime.timedelta(minutes=10) + await mock_database.create_login_token(user.id, token, expires_at) + await mock_database.mark_token_as_used(token) + + with pytest.raises(HTTPException) as exc_info: + await usecases.validate_login_token(dto.ValidateLoginTokenInput(token=token)) + assert exc_info.value.status_code == 400 + assert 'already been used' in str(exc_info.value.detail) + + +@pytest.mark.asyncio +async def test_validate_login_token_expired(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Истекший токен вызывает ошибку.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + token = 'test_token_123' + expires_at = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=1) + await mock_database.create_login_token(user.id, token, expires_at) + + with pytest.raises(HTTPException) as exc_info: + await usecases.validate_login_token(dto.ValidateLoginTokenInput(token=token)) + assert exc_info.value.status_code == 400 + assert 'expired' in str(exc_info.value.detail) diff --git a/tests/test_creatives.py b/tests/test_creatives.py new file mode 100644 index 0000000..783820d --- /dev/null +++ b/tests/test_creatives.py @@ -0,0 +1,237 @@ +"""Тесты работы с креативами.""" +import pytest +from fastapi import HTTPException + +from src import domain, dto, usecase +from tests.conftest import MockDatabase + + +@pytest.mark.asyncio +async def test_create_creative_success(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) + + input_data = dto.CreateCreativeInput( + name='Test Creative', + text='Приглашаю подписаться на канал: {link}', + target_channel_id=target_channel.id, + ) + + result = await usecases.create_creative(input_data, user.id) + + assert result.name == 'Test Creative' + assert result.text == 'Приглашаю подписаться на канал: {link}' + assert result.target_channel_id == target_channel.id + assert result.status == domain.CreativeStatus.ACTIVE + assert result.purchases_count == 0 + + +@pytest.mark.asyncio +async def test_create_creative_invalid_target_channel(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Создание креатива для несуществующего целевого канала вызывает ошибку.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + import uuid + + fake_target_id = uuid.uuid4() + + input_data = dto.CreateCreativeInput( + name='Test Creative', + text='Test text', + target_channel_id=fake_target_id, + ) + + with pytest.raises(HTTPException) as exc_info: + await usecases.create_creative(input_data, user.id) + assert exc_info.value.status_code == 404 + assert 'Target channel' in str(exc_info.value.detail) + + +@pytest.mark.asyncio +async def test_get_user_creatives(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) + + creative1 = domain.Creative( + name='Creative 1', + text='Text 1', + target_channel_id=target_channel.id, + user_id=user.id, + status=domain.CreativeStatus.ACTIVE, + ) + creative2 = domain.Creative( + name='Creative 2', + text='Text 2', + target_channel_id=target_channel.id, + user_id=user.id, + status=domain.CreativeStatus.ACTIVE, + ) + await mock_database.create_creative(creative1) + await mock_database.create_creative(creative2) + + result = await usecases.get_creatives(dto.GetCreativesInput(user_id=user.id)) + + assert len(result.creatives) == 2 + + +@pytest.mark.asyncio +async def test_get_creatives_filtered_by_target_channel(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Получение креативов для конкретного целевого канала.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + target1 = domain.TargetChannel( + telegram_id=-1001234567890, + title='Target 1', + username='target1', + user_id=user.id, + is_active=True, + ) + target2 = domain.TargetChannel( + telegram_id=-1001234567891, + title='Target 2', + username='target2', + user_id=user.id, + is_active=True, + ) + await mock_database.upsert_target_channel(target1) + await mock_database.upsert_target_channel(target2) + + creative1 = domain.Creative( + name='Creative 1', text='Text 1', target_channel_id=target1.id, user_id=user.id, status=domain.CreativeStatus.ACTIVE + ) + creative2 = domain.Creative( + name='Creative 2', text='Text 2', target_channel_id=target2.id, user_id=user.id, status=domain.CreativeStatus.ACTIVE + ) + await mock_database.create_creative(creative1) + await mock_database.create_creative(creative2) + + result = await usecases.get_creatives(dto.GetCreativesInput(user_id=user.id, target_channel_id=target1.id)) + + assert len(result.creatives) == 1 + assert result.creatives[0].target_channel_id == target1.id + + +@pytest.mark.asyncio +async def test_archive_creative(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) + + 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) + + result = await usecases.archive_creative(creative.id, user.id) + + assert result.status == domain.CreativeStatus.ARCHIVED + + # Проверяем, что архивированный креатив не возвращается по умолчанию + creatives_list = await usecases.get_creatives(dto.GetCreativesInput(user_id=user.id)) + assert len(creatives_list.creatives) == 0 + + # Проверяем, что архивированный креатив возвращается с флагом include_archived + creatives_with_archived = await usecases.get_creatives( + dto.GetCreativesInput(user_id=user.id, include_archived=True) + ) + assert len(creatives_with_archived.creatives) == 1 + + +@pytest.mark.asyncio +async def test_update_creative(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) + + creative = domain.Creative( + name='Old Name', + text='Old text', + target_channel_id=target_channel.id, + user_id=user.id, + status=domain.CreativeStatus.ACTIVE, + ) + await mock_database.create_creative(creative) + + update_input = dto.UpdateCreativeInput(name='New Name', text='New text') + result = await usecases.update_creative(creative.id, update_input, user.id) + + assert result.name == 'New Name' + assert result.text == 'New text' + + +@pytest.mark.asyncio +async def test_delete_creative(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) + + 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) + + await usecases.delete_creative(dto.DeleteCreativeInput(creative_id=creative.id, user_id=user.id)) + + # Проверяем, что креатив удалён + creatives_list = await usecases.get_creatives( + dto.GetCreativesInput(user_id=user.id, include_archived=True) + ) + assert len(creatives_list.creatives) == 0 + diff --git a/tests/test_external_channels.py b/tests/test_external_channels.py new file mode 100644 index 0000000..c95b9dc --- /dev/null +++ b/tests/test_external_channels.py @@ -0,0 +1,177 @@ +"""Тесты каталога внешних каналов.""" + +import pytest +from fastapi import HTTPException + +from src import domain, dto, usecase +from tests.conftest import MockDatabase + + +@pytest.mark.asyncio +async def test_create_external_channel_with_targets(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) + + input_data = dto.CreateExternalChannelInput( + telegram_id=-1009876543210, + title='External Channel', + username='externalchannel', + description='Test external channel', + subscribers_count=10000, + target_channel_ids=[target_channel.id], + ) + + result = await usecases.create_external_channel(input_data, user.id) + + assert result.telegram_id == -1009876543210 + assert result.title == 'External Channel' + assert result.subscribers_count == 10000 + + +@pytest.mark.asyncio +async def test_create_external_channel_invalid_target(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Создание внешнего канала с несуществующим целевым каналом вызывает ошибку.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + import uuid + + fake_target_id = uuid.uuid4() + + input_data = dto.CreateExternalChannelInput( + telegram_id=-1009876543210, + title='External Channel', + username='externalchannel', + target_channel_ids=[fake_target_id], + ) + + with pytest.raises(HTTPException) as exc_info: + await usecases.create_external_channel(input_data, user.id) + assert exc_info.value.status_code == 404 + assert 'Target channel' in str(exc_info.value.detail) + + +@pytest.mark.asyncio +async def test_get_external_channels_for_target(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) + + ext_channel = domain.ExternalChannel( + telegram_id=-1009876543210, + title='External Channel', + username='externalchannel', + description=None, + subscribers_count=10000, + user_id=user.id, + ) + await mock_database.create_external_channel(ext_channel) + await mock_database.add_external_channel_to_targets(ext_channel.id, [target_channel.id]) + + result = await usecases.get_external_channels( + dto.GetExternalChannelsInput(target_channel_id=target_channel.id, user_id=user.id) + ) + + assert len(result.external_channels) == 1 + assert result.external_channels[0].telegram_id == -1009876543210 + + +@pytest.mark.asyncio +async def test_delete_external_channel(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) + + ext_channel = domain.ExternalChannel( + telegram_id=-1009876543210, + title='External Channel', + username='externalchannel', + description=None, + subscribers_count=10000, + user_id=user.id, + ) + await mock_database.create_external_channel(ext_channel) + await mock_database.add_external_channel_to_targets(ext_channel.id, [target_channel.id]) + + await usecases.delete_external_channel(dto.DeleteExternalChannelInput(channel_id=ext_channel.id, user_id=user.id)) + + # Проверяем, что канал удалён + channels = await mock_database.get_external_channels_for_target(target_channel.id) + assert len(channels) == 0 + + +@pytest.mark.asyncio +async def test_update_external_channel_links(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Обновление привязок внешнего канала к целевым каналам.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + target1 = domain.TargetChannel( + telegram_id=-1001234567890, + title='Target 1', + username='target1', + user_id=user.id, + is_active=True, + ) + target2 = domain.TargetChannel( + telegram_id=-1001234567891, + title='Target 2', + username='target2', + user_id=user.id, + is_active=True, + ) + await mock_database.upsert_target_channel(target1) + await mock_database.upsert_target_channel(target2) + + ext_channel = domain.ExternalChannel( + telegram_id=-1009876543210, + title='External Channel', + username='externalchannel', + description=None, + subscribers_count=10000, + user_id=user.id, + ) + await mock_database.create_external_channel(ext_channel) + await mock_database.add_external_channel_to_targets(ext_channel.id, [target1.id]) + + # Добавляем связь с target2 + await usecases.update_external_channel_links( + ext_channel.id, + dto.UpdateExternalChannelLinksInput( + add_target_channel_ids=[target2.id], + remove_target_channel_ids=[], + ), + user.id, + ) + + # Проверяем, что канал теперь привязан к обоим целевым каналам + channels_for_target2 = await mock_database.get_external_channels_for_target(target2.id) + assert len(channels_for_target2) == 1 diff --git a/tests/test_target_channels.py b/tests/test_target_channels.py new file mode 100644 index 0000000..dc49536 --- /dev/null +++ b/tests/test_target_channels.py @@ -0,0 +1,168 @@ +"""Тесты подключения целевых каналов.""" + +import pytest +from fastapi import HTTPException + +from src import domain, dto, usecase +from tests.conftest import MockDatabase, MockTelegramWriter + + +@pytest.mark.asyncio +async def test_connect_new_channel_success(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Подключение нового канала с правами администратора.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + input_data = dto.ConnectTargetChanInput( + telegram_id=-1001234567890, + title='Test Channel', + username='testchannel', + user_telegram_id=123456, + bot_permissions=dto.ChannelBotPermissions( + is_admin=True, + can_invite_users=True, + can_restrict_members=True, + ), + ) + + result = await usecases.connect_target_chan(input_data) + + assert result.telegram_id == -1001234567890 + assert result.title == 'Test Channel' + assert result.is_active is True + + # Проверяем, что канал сохранён + channel = await mock_database.get_target_channel(user.id, telegram_id=-1001234567890) + assert channel is not None + + +@pytest.mark.asyncio +async def test_connect_channel_without_admin_rights( + usecases: usecase.Usecase, mock_database: MockDatabase, mock_telegram: MockTelegramWriter +) -> None: + """Канал без прав администратора не подключается.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + input_data = dto.ConnectTargetChanInput( + telegram_id=-1001234567890, + title='Test Channel', + username='testchannel', + user_telegram_id=123456, + bot_permissions=dto.ChannelBotPermissions( + is_admin=False, + can_invite_users=False, + can_restrict_members=False, + ), + ) + + with pytest.raises(HTTPException) as exc_info: + await usecases.connect_target_chan(input_data) + + assert exc_info.value.status_code == 403 + # Проверяем, что бот отправил предупреждение + assert len(mock_telegram.sent_messages) == 1 + assert 'не является админом' in mock_telegram.sent_messages[0]['text'] + + +@pytest.mark.asyncio +async def test_connect_channel_without_required_permissions( + usecases: usecase.Usecase, mock_database: MockDatabase, mock_telegram: MockTelegramWriter +) -> None: + """Канал без необходимых прав не подключается.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + input_data = dto.ConnectTargetChanInput( + telegram_id=-1001234567890, + title='Test Channel', + username='testchannel', + user_telegram_id=123456, + bot_permissions=dto.ChannelBotPermissions( + is_admin=True, + can_invite_users=False, # Отсутствует необходимое право + can_restrict_members=True, + ), + ) + + with pytest.raises(HTTPException) as exc_info: + await usecases.connect_target_chan(input_data) + + assert exc_info.value.status_code == 403 + assert len(mock_telegram.sent_messages) == 1 + assert 'не имеет необходимых прав' in mock_telegram.sent_messages[0]['text'] + + +@pytest.mark.asyncio +async def test_connect_channel_user_not_found(usecases: usecase.Usecase, mock_telegram: MockTelegramWriter) -> None: + """Подключение канала неавторизованным пользователем вызывает ошибку.""" + input_data = dto.ConnectTargetChanInput( + telegram_id=-1001234567890, + title='Test Channel', + username='testchannel', + user_telegram_id=999999, # Несуществующий пользователь + bot_permissions=dto.ChannelBotPermissions( + is_admin=True, + can_invite_users=True, + can_restrict_members=True, + ), + ) + + with pytest.raises(HTTPException) as exc_info: + await usecases.connect_target_chan(input_data) + + assert exc_info.value.status_code == 404 + assert 'not found' in str(exc_info.value.detail).lower() + assert len(mock_telegram.sent_messages) == 1 + assert 'авторизоваться' in mock_telegram.sent_messages[0]['text'] + + +@pytest.mark.asyncio +async def test_get_user_target_channels(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Получение списка каналов пользователя.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + channel1 = domain.TargetChannel( + telegram_id=-1001234567890, + title='Channel 1', + username='channel1', + user_id=user.id, + is_active=True, + ) + channel2 = domain.TargetChannel( + telegram_id=-1001234567891, + title='Channel 2', + username='channel2', + user_id=user.id, + is_active=True, + ) + await mock_database.upsert_target_channel(channel1) + await mock_database.upsert_target_channel(channel2) + + result = await usecases.get_user_target_chans(dto.GetUserTargetChansInput(user_id=user.id)) + + assert len(result.target_channels) == 2 + + +@pytest.mark.asyncio +async def test_disconnect_target_channel(usecases: usecase.Usecase, mock_database: MockDatabase) -> None: + """Отключение целевого канала.""" + user = domain.User(telegram_id=123456, username='testuser') + await mock_database.create_user(user) + + channel = domain.TargetChannel( + telegram_id=-1001234567890, + title='Test Channel', + username='testchannel', + user_id=user.id, + is_active=True, + ) + await mock_database.upsert_target_channel(channel) + + await usecases.disconnect_target_chan(dto.DisconnectTargetChanInput(channel_id=channel.id, user_id=user.id)) + + # Проверяем, что канал стал неактивным + updated_channel = await mock_database.get_target_channel(user.id, channel_id=channel.id) + assert updated_channel is not None + assert updated_channel.is_active is False