root/tgex-backend
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: purchases added

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2025-11-10 16:43:20 +03:00
parent f8edd81395
commit 2549242b06
20 changed files with 1429 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
migration/versions/6517eda4a92e_add_purchases_table.py Normal file
View File

@@ -0,0 +1,74 @@
"""add purchases table
Revision ID: 6517eda4a92e
Revises: e005b383b5c6
Create Date: 2025-11-10 16:36:15.533520
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = '6517eda4a92e'
down_revision: str | None = 'e005b383b5c6'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
'purchases',
sa.Column('target_channel_id', sa.Uuid(), nullable=False),
sa.Column('external_channel_id', sa.Uuid(), nullable=False),
sa.Column('creative_id', sa.Uuid(), nullable=False),
sa.Column('user_id', sa.Uuid(), nullable=False),
sa.Column('placement_date', sa.DateTime(timezone=True), nullable=False),
sa.Column('cost', sa.Float(), nullable=True),
sa.Column('comment', sa.String(), nullable=True),
sa.Column('ad_post_url', sa.String(), nullable=True),
sa.Column('invite_link_type', sa.Enum('PUBLIC', 'APPROVAL', name='invitelinktype'), nullable=False),
sa.Column('invite_link', sa.String(), nullable=False),
sa.Column('status', sa.Enum('ACTIVE', 'ARCHIVED', name='purchasestatus'), nullable=False),
sa.Column('subscriptions_count', sa.Integer(), server_default='0', nullable=False),
sa.Column('views_count', sa.Integer(), nullable=True),
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(
['creative_id'],
['creatives.id'],
),
sa.ForeignKeyConstraint(
['external_channel_id'],
['externalchannels.id'],
),
sa.ForeignKeyConstraint(
['target_channel_id'],
['targetchannels.id'],
),
sa.ForeignKeyConstraint(
['user_id'],
['users.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_purchases_creative_id'), 'purchases', ['creative_id'], unique=False)
op.create_index(op.f('ix_purchases_external_channel_id'), 'purchases', ['external_channel_id'], unique=False)
op.create_index(op.f('ix_purchases_target_channel_id'), 'purchases', ['target_channel_id'], unique=False)
op.create_index(op.f('ix_purchases_user_id'), 'purchases', ['user_id'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_purchases_user_id'), table_name='purchases')
op.drop_index(op.f('ix_purchases_target_channel_id'), table_name='purchases')
op.drop_index(op.f('ix_purchases_external_channel_id'), table_name='purchases')
op.drop_index(op.f('ix_purchases_creative_id'), table_name='purchases')
op.drop_table('purchases')
# ### end Alembic commands ###

77
src/adapter/postgres.py
View File

@@ -222,3 +222,80 @@ class Postgres(DatabaseBase):
stmt = delete(domain.Creative).where(domain.Creative.id == creative_id)
await self.session.execute(stmt)
await self.session.flush()
async def get_purchase(self, user_id: uuid.UUID, purchase_id: uuid.UUID) -> domain.Purchase | None:
stmt = (
select(domain.Purchase)
.options(
selectinload(domain.Purchase.target_channel),
selectinload(domain.Purchase.external_channel),
selectinload(domain.Purchase.creative),
)
.where(domain.Purchase.id == purchase_id, domain.Purchase.user_id == user_id)
)
result = await self.session.execute(stmt)
return result.scalar_one_or_none()
async def create_purchase(self, purchase: domain.Purchase) -> domain.Purchase:
self.session.add(purchase)
await self.session.flush()
await self.session.refresh(
purchase,
['target_channel', 'external_channel', 'creative'],
)
return purchase
async def update_purchase(self, purchase: domain.Purchase) -> domain.Purchase:
await self.session.flush()
await self.session.refresh(
purchase,
['target_channel', 'external_channel', 'creative'],
)
return purchase
async def get_user_purchases(
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.Purchase]:
stmt = (
select(domain.Purchase)
.options(
selectinload(domain.Purchase.target_channel),
selectinload(domain.Purchase.external_channel),
selectinload(domain.Purchase.creative),
)
.where(domain.Purchase.user_id == user_id)
)
if target_channel_id:
stmt = stmt.where(domain.Purchase.target_channel_id == target_channel_id)
if external_channel_id:
stmt = stmt.where(domain.Purchase.external_channel_id == external_channel_id)
if creative_id:
stmt = stmt.where(domain.Purchase.creative_id == creative_id)
if not include_archived:
stmt = stmt.where(domain.Purchase.status == domain.PurchaseStatus.ACTIVE)
stmt = stmt.order_by(domain.Purchase.placement_date.desc())
result = await self.session.execute(stmt)
return list(result.scalars().all())
async def delete_purchase(self, purchase_id: uuid.UUID) -> None:
stmt = delete(domain.Purchase).where(domain.Purchase.id == purchase_id)
await self.session.execute(stmt)
await self.session.flush()
async def check_creative_in_use(self, creative_id: uuid.UUID) -> bool:
stmt = select(domain.Purchase).where(
domain.Purchase.creative_id == creative_id,
domain.Purchase.status == domain.PurchaseStatus.ACTIVE,
)
result = await self.session.execute(stmt)
return result.scalar_one_or_none() is not None

7
src/adapter/telegram.py
View File

@@ -19,3 +19,10 @@ class Telegram(TelegramBase):
async def get_chat_member(self, chat_id: int, user_id: int) -> dict[str, Any]:
member = await self.bot.get_chat_member(chat_id=chat_id, user_id=user_id)
return member.model_dump()
async def create_chat_invite_link(self, chat_id: int, requires_approval: bool = False) -> str:
invite_link = await self.bot.create_chat_invite_link(
chat_id=chat_id,
creates_join_request=requires_approval,
)
return invite_link.invite_link

2
src/controller/http/__init__.py
View File

@@ -3,6 +3,7 @@ from fastapi import APIRouter
from src.controller.http.auth import auth_router
from src.controller.http.creatives import creatives_router
from src.controller.http.external_channels import external_channels_router
from src.controller.http.purchases import purchases_router
from src.controller.http.target_channels import target_channels_router
api_router = APIRouter()
@@ -15,5 +16,6 @@ api_v1_router = APIRouter(prefix='/api/v1')
api_v1_router.include_router(target_channels_router)
api_v1_router.include_router(external_channels_router)
api_v1_router.include_router(creatives_router)
api_v1_router.include_router(purchases_router)
api_router.include_router(api_v1_router)

87
src/controller/http/purchases.py Normal file
View File

@@ -0,0 +1,87 @@
import uuid
from typing import Annotated
from fastapi import Depends
from fastapi.routing import APIRouter
from src import dependencies, dto
from src.adapter.jwt import JWTPayload
purchases_router = APIRouter(prefix='/purchases', tags=['purchases'])
@purchases_router.get('')
async def list_purchases(
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
target_channel_id: uuid.UUID | None = None,
external_channel_id: uuid.UUID | None = None,
creative_id: uuid.UUID | None = None,
include_archived: bool = False,
) -> dto.GetPurchasesOutput:
input = dto.GetPurchasesInput(
user_id=current_user.user_id,
target_channel_id=target_channel_id,
external_channel_id=external_channel_id,
creative_id=creative_id,
include_archived=include_archived,
)
return await dependencies.get_usecase().get_purchases(input=input)
@purchases_router.get('/{purchase_id}')
async def get_purchase(
purchase_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
) -> dto.PurchaseOutput:
input = dto.GetPurchaseInput(
purchase_id=purchase_id,
user_id=current_user.user_id,
)
return await dependencies.get_usecase().get_purchase(input=input)
@purchases_router.post('')
async def create_purchase(
request: dto.CreatePurchaseInput,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
) -> dto.PurchaseOutput:
return await dependencies.get_usecase().create_purchase(request, current_user.user_id)
@purchases_router.patch('/{purchase_id}')
async def update_purchase(
purchase_id: uuid.UUID,
request: dto.UpdatePurchaseInput,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
) -> dto.PurchaseOutput:
return await dependencies.get_usecase().update_purchase(
purchase_id=purchase_id,
input=request,
user_id=current_user.user_id,
)
@purchases_router.post('/{purchase_id}/archive')
async def archive_purchase(
purchase_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
) -> dto.PurchaseOutput:
return await dependencies.get_usecase().archive_purchase(
purchase_id=purchase_id,
user_id=current_user.user_id,
)
@purchases_router.delete('/{purchase_id}')
async def delete_purchase(
purchase_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
) -> None:
input = dto.DeletePurchaseInput(
purchase_id=purchase_id,
user_id=current_user.user_id,
)
await dependencies.get_usecase().delete_purchase(input)

6
src/domain/__init__.py
View File

@@ -4,8 +4,11 @@ __all__ = (
'TargetChannel',
'ExternalChannel',
'Creative',
'Purchase',
'TargetChannelStatus',
'CreativeStatus',
'PurchaseStatus',
'InviteLinkType',
'LoginToken',
'UserNotFound',
'LoginTokenNotFound',
@@ -18,6 +21,7 @@ __all__ = (
'ExternalChannelAlreadyExists',
'CreativeNotFound',
'CreativeInUse',
'PurchaseNotFound',
)
from .base import Base
@@ -32,10 +36,12 @@ from .error import (
LoginTokenAlreadyUsed,
LoginTokenExpired,
LoginTokenNotFound,
PurchaseNotFound,
TargetChannelNotFound,
UserNotFound,
)
from .external_channel import ExternalChannel
from .login_token import LoginToken
from .purchase import InviteLinkType, Purchase, PurchaseStatus
from .target_channel import TargetChannel, TargetChannelStatus
from .user import User

6
src/domain/error.py
View File

@@ -58,3 +58,9 @@ def CreativeInUse(creative_id: uuid.UUID) -> HTTPException:
status.HTTP_400_BAD_REQUEST,
f'Creative {creative_id} is used in active purchases and cannot be deleted',
)
def PurchaseNotFound(purchase_id: uuid.UUID | None = None) -> HTTPException:
if purchase_id is None:
return HTTPException(status.HTTP_404_NOT_FOUND, 'Purchase not found')
return HTTPException(status.HTTP_404_NOT_FOUND, f'Purchase {purchase_id} not found')

52
src/domain/purchase.py Normal file
View File

@@ -0,0 +1,52 @@
import datetime
import uuid
from enum import StrEnum
from typing import TYPE_CHECKING
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src import domain
if TYPE_CHECKING:
from .creative import Creative
from .external_channel import ExternalChannel
from .target_channel import TargetChannel
class PurchaseStatus(StrEnum):
ACTIVE = 'active'
ARCHIVED = 'archived'
class InviteLinkType(StrEnum):
PUBLIC = 'public' # открытая ссылка
APPROVAL = 'approval' # с одобрением ботом
class Purchase(domain.Base):
# Основные поля
target_channel_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('targetchannels.id'), index=True)
external_channel_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('externalchannels.id'), index=True)
creative_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('creatives.id'), index=True)
user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('users.id'), index=True)
# Данные закупа
placement_date: Mapped[datetime.datetime] # дата размещения (планируемая или фактическая)
cost: Mapped[float | None] # стоимость закупа
comment: Mapped[str | None] # комментарий
ad_post_url: Mapped[str | None] # ссылка на рекламное сообщение во внешнем канале
invite_link_type: Mapped[InviteLinkType] # тип пригласительной ссылки
invite_link: Mapped[str] # уникальная пригласительная ссылка
# Статус
status: Mapped[PurchaseStatus] = mapped_column(default=PurchaseStatus.ACTIVE)
# Метрики (будут обновляться из Telegram API)
subscriptions_count: Mapped[int] = mapped_column(default=0, server_default='0')
views_count: Mapped[int | None] # просмотры рекламного поста (если доступны)
# Relationships
target_channel: Mapped['TargetChannel'] = relationship()
external_channel: Mapped['ExternalChannel'] = relationship()
creative: Mapped['Creative'] = relationship()

18
src/dto/__init__.py
View File

@@ -26,6 +26,14 @@ __all__ = (
'UpdateCreativeInput',
'ArchiveCreativeInput',
'DeleteCreativeInput',
'PurchaseOutput',
'GetPurchasesInput',
'GetPurchasesOutput',
'GetPurchaseInput',
'CreatePurchaseInput',
'UpdatePurchaseInput',
'ArchivePurchaseInput',
'DeletePurchaseInput',
)
from .creative import (
@@ -48,6 +56,16 @@ from .external_channel import (
ImportExternalChannelsOutput,
UpdateExternalChannelLinksInput,
)
from .purchase import (
ArchivePurchaseInput,
CreatePurchaseInput,
DeletePurchaseInput,
GetPurchaseInput,
GetPurchasesInput,
GetPurchasesOutput,
PurchaseOutput,
UpdatePurchaseInput,
)
from .target_channel import (
ChannelBotPermissions,
ConnectTargetChanInput,

71
src/dto/purchase.py Normal file
View File

@@ -0,0 +1,71 @@
import datetime
import uuid
import pydantic
from src import domain
class PurchaseOutput(pydantic.BaseModel):
id: uuid.UUID
target_channel_id: uuid.UUID
target_channel_title: str
external_channel_id: uuid.UUID
external_channel_title: str
creative_id: uuid.UUID
creative_name: str
placement_date: datetime.datetime
cost: float | None
comment: str | None
ad_post_url: str | None
invite_link_type: domain.InviteLinkType
invite_link: str
status: domain.PurchaseStatus
subscriptions_count: int
views_count: int | None
created_at: datetime.datetime
class GetPurchasesInput(pydantic.BaseModel):
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
class GetPurchasesOutput(pydantic.BaseModel):
purchases: list[PurchaseOutput]
class GetPurchaseInput(pydantic.BaseModel):
purchase_id: uuid.UUID
user_id: uuid.UUID
class CreatePurchaseInput(pydantic.BaseModel):
target_channel_id: uuid.UUID
external_channel_id: uuid.UUID
creative_id: uuid.UUID
placement_date: datetime.datetime
cost: float | None = None
comment: str | None = None
ad_post_url: str | None = None
invite_link_type: domain.InviteLinkType = domain.InviteLinkType.PUBLIC
class UpdatePurchaseInput(pydantic.BaseModel):
placement_date: datetime.datetime | None = None
cost: float | None = None
comment: str | None = None
ad_post_url: str | None = None
class ArchivePurchaseInput(pydantic.BaseModel):
purchase_id: uuid.UUID
user_id: uuid.UUID
class DeletePurchaseInput(pydantic.BaseModel):
purchase_id: uuid.UUID
user_id: uuid.UUID

34
src/usecase/__init__.py
View File

@@ -16,6 +16,12 @@ from .external_channel.delete_external_channel import delete_external_channel
from .external_channel.get_external_channels import get_external_channels
from .external_channel.import_external_channels_from_excel import import_external_channels_from_excel
from .external_channel.update_external_channel_links import update_external_channel_links
from .purchase.archive_purchase import archive_purchase
from .purchase.create_purchase import create_purchase
from .purchase.delete_purchase import delete_purchase
from .purchase.get_purchase import get_purchase
from .purchase.get_purchases import get_purchases
from .purchase.update_purchase import update_purchase
from .target_channel.connect_target_chan import connect_target_chan
from .target_channel.disconnect_target_chan import disconnect_target_chan
from .target_channel.disconnect_target_chan_by_tg_id import disconnect_target_chan_by_tg_id
@@ -84,6 +90,26 @@ class Database(typing.Protocol):
async def delete_creative(self, creative_id: UUID) -> None: ...
async def get_purchase(self, user_id: UUID, purchase_id: UUID) -> domain.Purchase | None: ...
async def create_purchase(self, purchase: domain.Purchase) -> domain.Purchase: ...
async def update_purchase(self, purchase: domain.Purchase) -> domain.Purchase: ...
async def get_user_purchases(
self,
user_id: UUID,
*,
target_channel_id: UUID | None = None,
external_channel_id: UUID | None = None,
creative_id: UUID | None = None,
include_archived: bool = False,
) -> list[domain.Purchase]: ...
async def delete_purchase(self, purchase_id: UUID) -> None: ...
async def check_creative_in_use(self, creative_id: UUID) -> bool: ...
class TelegramWriter(typing.Protocol):
async def send_message(self, text: str, chat_id: int) -> None: ...
@@ -92,6 +118,8 @@ class TelegramWriter(typing.Protocol):
async def get_chat_member(self, chat_id: int, user_id: int) -> dict[str, typing.Any]: ...
async def create_chat_invite_link(self, chat_id: int, requires_approval: bool = False) -> str: ...
class JWTEncoder(typing.Protocol):
def encode_access_token(self, user_id: UUID, telegram_id: int, username: str | None = None) -> str: ...
@@ -121,3 +149,9 @@ class Usecase:
update_creative = update_creative
archive_creative = archive_creative
delete_creative = delete_creative
get_purchases = get_purchases
get_purchase = get_purchase
create_purchase = create_purchase
update_purchase = update_purchase
archive_purchase = archive_purchase
delete_purchase = delete_purchase

41
src/usecase/purchase/archive_purchase.py Normal file
View File

@@ -0,0 +1,41 @@
import logging
import uuid
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def archive_purchase(self: 'Usecase', purchase_id: uuid.UUID, user_id: uuid.UUID) -> dto.PurchaseOutput:
async with self.database.transaction():
purchase = await self.database.get_purchase(user_id, purchase_id)
if not purchase:
log.warning('Purchase %s not found for user %s', purchase_id, user_id)
raise domain.PurchaseNotFound(purchase_id)
purchase.status = domain.PurchaseStatus.ARCHIVED
updated = await self.database.update_purchase(purchase)
return dto.PurchaseOutput(
id=updated.id,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
external_channel_id=updated.external_channel_id,
external_channel_title=updated.external_channel.title,
creative_id=updated.creative_id,
creative_name=updated.creative.name,
placement_date=updated.placement_date,
cost=updated.cost,
comment=updated.comment,
ad_post_url=updated.ad_post_url,
invite_link_type=updated.invite_link_type,
invite_link=updated.invite_link,
status=updated.status,
subscriptions_count=updated.subscriptions_count,
views_count=updated.views_count,
created_at=updated.created_at,
)

83
src/usecase/purchase/create_purchase.py Normal file
View File

@@ -0,0 +1,83 @@
import logging
import uuid
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def create_purchase(self: 'Usecase', input: dto.CreatePurchaseInput, user_id: uuid.UUID) -> dto.PurchaseOutput:
async with self.database.transaction():
# Проверяем существование целевого канала
target_channel = await self.database.get_target_channel(user_id, channel_id=input.target_channel_id)
if not target_channel:
log.warning(
'User %s attempted to create purchase for unavailable target %s', user_id, input.target_channel_id
)
raise domain.TargetChannelNotFound(input.target_channel_id)
# Проверяем существование внешнего канала
external_channel = await self.database.get_external_channel(user_id, channel_id=input.external_channel_id)
if not external_channel:
log.warning(
'User %s attempted to create purchase for unavailable external %s', user_id, input.external_channel_id
)
raise domain.ExternalChannelNotFound(input.external_channel_id)
# Проверяем существование креатива
creative = await self.database.get_creative(user_id, input.creative_id)
if not creative:
log.warning('User %s attempted to create purchase for unavailable creative %s', user_id, input.creative_id)
raise domain.CreativeNotFound(input.creative_id)
# Создаём уникальную пригласительную ссылку через Telegram Bot API
requires_approval = input.invite_link_type == domain.InviteLinkType.APPROVAL
invite_link = await self.telegram_writer.create_chat_invite_link(
chat_id=target_channel.telegram_id,
requires_approval=requires_approval,
)
# Создаём закуп
purchase = domain.Purchase(
target_channel_id=input.target_channel_id,
external_channel_id=input.external_channel_id,
creative_id=input.creative_id,
user_id=user_id,
placement_date=input.placement_date,
cost=input.cost,
comment=input.comment,
ad_post_url=input.ad_post_url,
invite_link_type=input.invite_link_type,
invite_link=invite_link,
status=domain.PurchaseStatus.ACTIVE,
)
created = await self.database.create_purchase(purchase)
# Увеличиваем счётчик закупов у креатива
creative.purchases_count += 1
await self.database.update_creative(creative)
return dto.PurchaseOutput(
id=created.id,
target_channel_id=created.target_channel_id,
target_channel_title=created.target_channel.title,
external_channel_id=created.external_channel_id,
external_channel_title=created.external_channel.title,
creative_id=created.creative_id,
creative_name=created.creative.name,
placement_date=created.placement_date,
cost=created.cost,
comment=created.comment,
ad_post_url=created.ad_post_url,
invite_link_type=created.invite_link_type,
invite_link=created.invite_link,
status=created.status,
subscriptions_count=created.subscriptions_count,
views_count=created.views_count,
created_at=created.created_at,
)

26
src/usecase/purchase/delete_purchase.py Normal file
View File

@@ -0,0 +1,26 @@
import logging
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def delete_purchase(self: 'Usecase', input: dto.DeletePurchaseInput) -> None:
async with self.database.transaction():
purchase = await self.database.get_purchase(input.user_id, input.purchase_id)
if not purchase:
log.warning('Purchase %s not found for user %s', input.purchase_id, input.user_id)
raise domain.PurchaseNotFound(input.purchase_id)
# Уменьшаем счётчик закупов у креатива, если закуп был активным
if purchase.status == domain.PurchaseStatus.ACTIVE:
creative = await self.database.get_creative(input.user_id, purchase.creative_id)
if creative and creative.purchases_count > 0:
creative.purchases_count -= 1
await self.database.update_creative(creative)
await self.database.delete_purchase(input.purchase_id)

37
src/usecase/purchase/get_purchase.py Normal file
View File

@@ -0,0 +1,37 @@
import logging
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def get_purchase(self: 'Usecase', input: dto.GetPurchaseInput) -> dto.PurchaseOutput:
async with self.database.transaction():
purchase = await self.database.get_purchase(input.user_id, input.purchase_id)
if not purchase:
log.warning('Purchase %s not found for user %s', input.purchase_id, input.user_id)
raise domain.PurchaseNotFound(input.purchase_id)
return dto.PurchaseOutput(
id=purchase.id,
target_channel_id=purchase.target_channel_id,
target_channel_title=purchase.target_channel.title,
external_channel_id=purchase.external_channel_id,
external_channel_title=purchase.external_channel.title,
creative_id=purchase.creative_id,
creative_name=purchase.creative.name,
placement_date=purchase.placement_date,
cost=purchase.cost,
comment=purchase.comment,
ad_post_url=purchase.ad_post_url,
invite_link_type=purchase.invite_link_type,
invite_link=purchase.invite_link,
status=purchase.status,
subscriptions_count=purchase.subscriptions_count,
views_count=purchase.views_count,
created_at=purchase.created_at,
)

42
src/usecase/purchase/get_purchases.py Normal file
View File

@@ -0,0 +1,42 @@
from typing import TYPE_CHECKING
from src import dto
if TYPE_CHECKING:
from .. import Usecase
async def get_purchases(self: 'Usecase', input: dto.GetPurchasesInput) -> dto.GetPurchasesOutput:
async with self.database.transaction():
purchases = await self.database.get_user_purchases(
input.user_id,
target_channel_id=input.target_channel_id,
external_channel_id=input.external_channel_id,
creative_id=input.creative_id,
include_archived=input.include_archived,
)
return dto.GetPurchasesOutput(
purchases=[
dto.PurchaseOutput(
id=purchase.id,
target_channel_id=purchase.target_channel_id,
target_channel_title=purchase.target_channel.title,
external_channel_id=purchase.external_channel_id,
external_channel_title=purchase.external_channel.title,
creative_id=purchase.creative_id,
creative_name=purchase.creative.name,
placement_date=purchase.placement_date,
cost=purchase.cost,
comment=purchase.comment,
ad_post_url=purchase.ad_post_url,
invite_link_type=purchase.invite_link_type,
invite_link=purchase.invite_link,
status=purchase.status,
subscriptions_count=purchase.subscriptions_count,
views_count=purchase.views_count,
created_at=purchase.created_at,
)
for purchase in purchases
]
)

52
src/usecase/purchase/update_purchase.py Normal file
View File

@@ -0,0 +1,52 @@
import logging
import uuid
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def update_purchase(
self: 'Usecase', purchase_id: uuid.UUID, input: dto.UpdatePurchaseInput, user_id: uuid.UUID
) -> dto.PurchaseOutput:
async with self.database.transaction():
purchase = await self.database.get_purchase(user_id, purchase_id)
if not purchase:
log.warning('Purchase %s not found for user %s', purchase_id, user_id)
raise domain.PurchaseNotFound(purchase_id)
# Обновляем только те поля, которые были переданы
if input.placement_date is not None:
purchase.placement_date = input.placement_date
if input.cost is not None:
purchase.cost = input.cost
if input.comment is not None:
purchase.comment = input.comment
if input.ad_post_url is not None:
purchase.ad_post_url = input.ad_post_url
updated = await self.database.update_purchase(purchase)
return dto.PurchaseOutput(
id=updated.id,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
external_channel_id=updated.external_channel_id,
external_channel_title=updated.external_channel.title,
creative_id=updated.creative_id,
creative_name=updated.creative.name,
placement_date=updated.placement_date,
cost=updated.cost,
comment=updated.comment,
ad_post_url=updated.ad_post_url,
invite_link_type=updated.invite_link_type,
invite_link=updated.invite_link,
status=updated.status,
subscriptions_count=updated.subscriptions_count,
views_count=updated.views_count,
created_at=updated.created_at,
)

75
tests/conftest.py
View File

@@ -1,6 +1,7 @@
import datetime
import uuid
from typing import Any, AsyncContextManager
from contextlib import AbstractAsyncContextManager
from typing import Any
from unittest.mock import AsyncMock
import pytest
@@ -17,9 +18,10 @@ class MockDatabase:
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.purchases: dict[uuid.UUID, domain.Purchase] = {}
self.m2m_target_external: dict[uuid.UUID, list[uuid.UUID]] = {}
def transaction(self) -> AsyncContextManager[None]:
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:
@@ -192,10 +194,74 @@ class MockDatabase:
async def delete_creative(self, creative_id: uuid.UUID) -> None:
self.creatives.pop(creative_id, None)
async def create_purchase(self, purchase: domain.Purchase) -> domain.Purchase:
if not purchase.id:
purchase.id = uuid.uuid4()
if not hasattr(purchase, 'created_at') or purchase.created_at is None:
purchase.created_at = datetime.datetime.now(datetime.UTC)
if not hasattr(purchase, 'updated_at') or purchase.updated_at is None:
purchase.updated_at = datetime.datetime.now(datetime.UTC)
if not hasattr(purchase, 'subscriptions_count') or purchase.subscriptions_count is None:
purchase.subscriptions_count = 0
if not hasattr(purchase, 'status') or purchase.status is None:
purchase.status = domain.PurchaseStatus.ACTIVE
# Загружаем связанные объекты
target_ch = self.target_channels.get(purchase.target_channel_id)
if target_ch:
purchase.target_channel = target_ch
ext_ch = self.external_channels.get(purchase.external_channel_id)
if ext_ch:
purchase.external_channel = ext_ch
creative = self.creatives.get(purchase.creative_id)
if creative:
purchase.creative = creative
self.purchases[purchase.id] = purchase
return purchase
async def get_purchase(self, user_id: uuid.UUID, purchase_id: uuid.UUID) -> domain.Purchase | None:
purchase = self.purchases.get(purchase_id)
if purchase and purchase.user_id == user_id:
return purchase
return None
async def update_purchase(self, purchase: domain.Purchase) -> domain.Purchase:
self.purchases[purchase.id] = purchase
return purchase
async def get_user_purchases(
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.Purchase]:
result = [p for p in self.purchases.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.PurchaseStatus.ACTIVE]
return sorted(result, key=lambda p: p.placement_date, reverse=True)
async def delete_purchase(self, purchase_id: uuid.UUID) -> None:
self.purchases.pop(purchase_id, None)
async def check_creative_in_use(self, creative_id: uuid.UUID) -> bool:
for purchase in self.purchases.values():
if purchase.creative_id == creative_id and purchase.status == domain.PurchaseStatus.ACTIVE:
return True
return False
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})
@@ -208,6 +274,11 @@ class MockTelegramWriter:
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}'
class MockJWTEncoder:
def encode_access_token(self, user_id: uuid.UUID, telegram_id: int, username: str | None = None) -> str:

18
tests/test_creatives.py
View File

@@ -1,4 +1,5 @@
"""Тесты работы с креативами."""
import pytest
from fastapi import HTTPException
@@ -119,10 +120,18 @@ async def test_get_creatives_filtered_by_target_channel(usecases: usecase.Usecas
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
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
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)
@@ -230,8 +239,5 @@ async def test_delete_creative(usecases: usecase.Usecase, mock_database: MockDat
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)
)
creatives_list = await usecases.get_creatives(dto.GetCreativesInput(user_id=user.id, include_archived=True))
assert len(creatives_list.creatives) == 0

629
tests/test_purchases.py Normal file
View File

@@ -0,0 +1,629 @@
"""Тесты работы с закупами."""
import datetime
import pytest
from fastapi import HTTPException
from src import domain, dto, usecase
from tests.conftest import MockDatabase
@pytest.mark.asyncio
async def test_create_purchase_with_public_invite_link(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='Ad channel',
subscribers_count=50000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Summer Sale',
text='Подписывайтесь на наш канал по ссылке: {link}',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
placement_date = datetime.datetime.now(datetime.UTC)
input_data = dto.CreatePurchaseInput(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
placement_date=placement_date,
cost=5000.0,
comment='Утреннее размещение',
ad_post_url='https://t.me/externalchannel/123',
invite_link_type=domain.InviteLinkType.PUBLIC,
)
result = await usecases.create_purchase(input_data, user.id)
assert result.target_channel_id == target_channel.id
assert result.external_channel_id == external_channel.id
assert result.creative_id == creative.id
assert result.cost == 5000.0
assert result.comment == 'Утреннее размещение'
assert result.invite_link_type == domain.InviteLinkType.PUBLIC
assert result.invite_link.startswith('https://t.me/+public_link_')
assert result.status == domain.PurchaseStatus.ACTIVE
assert result.subscriptions_count == 0
# Проверяем, что счётчик закупов у креатива увеличился
updated_creative = await mock_database.get_creative(user.id, creative.id)
assert updated_creative is not None
assert updated_creative.purchases_count == 1
@pytest.mark.asyncio
async def test_create_purchase_with_approval_invite_link(
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='Premium Creative',
text='Эксклюзивный контент: {link}',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
placement_date = datetime.datetime.now(datetime.UTC)
input_data = dto.CreatePurchaseInput(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
placement_date=placement_date,
invite_link_type=domain.InviteLinkType.APPROVAL,
)
result = await usecases.create_purchase(input_data, user.id)
assert result.invite_link_type == domain.InviteLinkType.APPROVAL
assert 'approval_link_' in result.invite_link
@pytest.mark.asyncio
async def test_create_purchase_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()
fake_external_id = uuid.uuid4()
fake_creative_id = uuid.uuid4()
input_data = dto.CreatePurchaseInput(
target_channel_id=fake_target_id,
external_channel_id=fake_external_id,
creative_id=fake_creative_id,
placement_date=datetime.datetime.now(datetime.UTC),
)
with pytest.raises(HTTPException) as exc_info:
await usecases.create_purchase(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_purchases(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=20000,
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)
# Создаём два закупа
purchase1 = 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),
cost=3000.0,
invite_link='https://t.me/+link1',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
)
purchase2 = 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) - datetime.timedelta(days=1),
cost=4000.0,
invite_link='https://t.me/+link2',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
)
await mock_database.create_purchase(purchase1)
await mock_database.create_purchase(purchase2)
result = await usecases.get_purchases(dto.GetPurchasesInput(user_id=user.id))
assert len(result.purchases) == 2
# Проверяем сортировку по дате размещения (новые первые)
assert result.purchases[0].id == purchase1.id
@pytest.mark.asyncio
async def test_get_purchases_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)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=15000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
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)
purchase1 = domain.Purchase(
target_channel_id=target1.id,
external_channel_id=external_channel.id,
creative_id=creative1.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,
)
purchase2 = domain.Purchase(
target_channel_id=target2.id,
external_channel_id=external_channel.id,
creative_id=creative2.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,
)
await mock_database.create_purchase(purchase1)
await mock_database.create_purchase(purchase2)
result = await usecases.get_purchases(dto.GetPurchasesInput(user_id=user.id, target_channel_id=target1.id))
assert len(result.purchases) == 1
assert result.purchases[0].target_channel_id == target1.id
@pytest.mark.asyncio
async def test_get_purchases_filtered_by_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)
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)
creative1 = domain.Creative(
name='Creative A',
text='Вариант A',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
creative2 = domain.Creative(
name='Creative B',
text='Вариант B',
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)
# Создаём закупы с разными креативами
purchase1 = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative1.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,
)
purchase2 = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative2.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,
)
await mock_database.create_purchase(purchase1)
await mock_database.create_purchase(purchase2)
result = await usecases.get_purchases(dto.GetPurchasesInput(user_id=user.id, creative_id=creative1.id))
assert len(result.purchases) == 1
assert result.purchases[0].creative_id == creative1.id
@pytest.mark.asyncio
async def test_archive_purchase(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=18000,
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,
)
await mock_database.create_purchase(purchase)
result = await usecases.archive_purchase(purchase.id, user.id)
assert result.status == domain.PurchaseStatus.ARCHIVED
# Проверяем, что архивированный закуп не возвращается по умолчанию
purchases_list = await usecases.get_purchases(dto.GetPurchasesInput(user_id=user.id))
assert len(purchases_list.purchases) == 0
# Проверяем, что архивированный закуп возвращается с флагом include_archived
purchases_with_archived = await usecases.get_purchases(
dto.GetPurchasesInput(user_id=user.id, include_archived=True)
)
assert len(purchases_with_archived.purchases) == 1
@pytest.mark.asyncio
async def test_update_purchase(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=22000,
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)
old_date = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=1)
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=old_date,
cost=3000.0,
comment='Старый комментарий',
invite_link='https://t.me/+link1',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
)
await mock_database.create_purchase(purchase)
new_date = datetime.datetime.now(datetime.UTC)
update_input = dto.UpdatePurchaseInput(
placement_date=new_date,
cost=3500.0,
comment='Обновлённый комментарий',
ad_post_url='https://t.me/externalchannel/456',
)
result = await usecases.update_purchase(purchase.id, update_input, user.id)
assert result.placement_date == new_date
assert result.cost == 3500.0
assert result.comment == 'Обновлённый комментарий'
assert result.ad_post_url == 'https://t.me/externalchannel/456'
@pytest.mark.asyncio
async def test_delete_purchase(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,
purchases_count=1,
)
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,
)
await mock_database.create_purchase(purchase)
await usecases.delete_purchase(dto.DeletePurchaseInput(purchase_id=purchase.id, user_id=user.id))
# Проверяем, что закуп удалён
purchases_list = await usecases.get_purchases(dto.GetPurchasesInput(user_id=user.id, include_archived=True))
assert len(purchases_list.purchases) == 0
# Проверяем, что счётчик закупов у креатива уменьшился
updated_creative = await mock_database.get_creative(user.id, creative.id)
assert updated_creative is not None
assert updated_creative.purchases_count == 0
@pytest.mark.asyncio
async def test_multiple_purchases_same_creative(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Создание нескольких закупов с одним креативом - для A/B тестирования в разных каналах."""
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)
external1 = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External 1',
username='external1',
description=None,
subscribers_count=10000,
user_id=user.id,
)
external2 = domain.ExternalChannel(
telegram_id=-1009876543211,
title='External 2',
username='external2',
description=None,
subscribers_count=20000,
user_id=user.id,
)
await mock_database.create_external_channel(external1)
await mock_database.create_external_channel(external2)
creative = domain.Creative(
name='Universal Creative',
text='Универсальный текст',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
# Создаём два закупа с одним креативом в разных внешних каналах
input1 = dto.CreatePurchaseInput(
target_channel_id=target_channel.id,
external_channel_id=external1.id,
creative_id=creative.id,
placement_date=datetime.datetime.now(datetime.UTC),
cost=2000.0,
)
input2 = dto.CreatePurchaseInput(
target_channel_id=target_channel.id,
external_channel_id=external2.id,
creative_id=creative.id,
placement_date=datetime.datetime.now(datetime.UTC),
cost=4000.0,
)
result1 = await usecases.create_purchase(input1, user.id)
result2 = await usecases.create_purchase(input2, user.id)
# Проверяем, что созданы два разных закупа
assert result1.id != result2.id
assert result1.invite_link != result2.invite_link
# Проверяем, что счётчик креатива увеличился на 2
updated_creative = await mock_database.get_creative(user.id, creative.id)
assert updated_creative is not None
assert updated_creative.purchases_count == 2
# Проверяем фильтрацию по креативу
purchases_for_creative = await usecases.get_purchases(
dto.GetPurchasesInput(user_id=user.id, creative_id=creative.id)
)
assert len(purchases_for_creative.purchases) == 2