qzl
230 lines
7.1 KiB
Python
230 lines
7.1 KiB
Python
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from core.auth.models import CurrentUser
|
|
from models.inbox_messages import InboxMessage, InboxMessageStatus
|
|
from models.schedule_items import (
|
|
ScheduleItem,
|
|
ScheduleItemSourceType,
|
|
ScheduleItemStatus,
|
|
)
|
|
from models.schedule_subscriptions import ScheduleSubscription
|
|
from v1.schedule_items.service import ScheduleItemService
|
|
|
|
|
|
def _create_mock_schedule_item(
|
|
item_id: UUID = uuid4(),
|
|
owner_id: UUID = UUID("00000000-0000-0000-0000-000000000001"),
|
|
title: str = "Test Event",
|
|
) -> ScheduleItem:
|
|
item = MagicMock(spec=ScheduleItem)
|
|
item.id = item_id
|
|
item.owner_id = owner_id
|
|
item.title = title
|
|
item.description = None
|
|
item.start_at = datetime(2026, 2, 28, 16, 0, 0, tzinfo=timezone.utc)
|
|
item.end_at = datetime(2026, 2, 28, 17, 0, 0, tzinfo=timezone.utc)
|
|
item.timezone = "UTC"
|
|
item.extra_metadata = {}
|
|
item.source_type = ScheduleItemSourceType.MANUAL
|
|
item.status = ScheduleItemStatus.ACTIVE
|
|
item.created_at = datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc)
|
|
item.updated_at = datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc)
|
|
item.deleted_at = None
|
|
return item
|
|
|
|
|
|
class FakeInboxRepo:
|
|
def __init__(self, inbox_message: InboxMessage | None = None) -> None:
|
|
self._inbox = inbox_message
|
|
|
|
async def get_pending_calendar_invite(
|
|
self, schedule_item_id: UUID, recipient_id: UUID
|
|
) -> InboxMessage | None:
|
|
if self._inbox:
|
|
return self._inbox
|
|
return None
|
|
|
|
async def get_calendar_invite(
|
|
self, schedule_item_id: UUID, recipient_id: UUID
|
|
) -> InboxMessage | None:
|
|
return await self.get_pending_calendar_invite(schedule_item_id, recipient_id)
|
|
|
|
async def create(self, data: dict) -> InboxMessage:
|
|
return MagicMock()
|
|
|
|
async def get_by_id(
|
|
self, message_id: UUID, recipient_id: UUID
|
|
) -> InboxMessage | None:
|
|
return None
|
|
|
|
async def list_by_recipient(
|
|
self, recipient_id: UUID, is_read: bool | None = None
|
|
) -> list[InboxMessage]:
|
|
return []
|
|
|
|
async def mark_as_read(
|
|
self, message_id: UUID, recipient_id: UUID
|
|
) -> InboxMessage | None:
|
|
return None
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_session() -> AsyncMock:
|
|
session = AsyncMock()
|
|
session.commit = AsyncMock()
|
|
session.rollback = AsyncMock()
|
|
return session
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_repo() -> MagicMock:
|
|
repo = MagicMock()
|
|
repo.get_subscription = AsyncMock(return_value=None)
|
|
repo.update_subscription_status = AsyncMock(return_value=None)
|
|
repo.archive_expired_subscribed_items = AsyncMock(return_value=0)
|
|
repo.create_subscription = AsyncMock(return_value=MagicMock())
|
|
return repo
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_accept_subscription_success(
|
|
mock_session: AsyncMock, mock_repo: MagicMock
|
|
) -> None:
|
|
user_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
sender_id = UUID("00000000-0000-0000-0000-000000000002")
|
|
item_id = uuid4()
|
|
|
|
inbox_message = MagicMock(spec=InboxMessage)
|
|
inbox_message.id = uuid4()
|
|
inbox_message.sender_id = sender_id
|
|
inbox_message.content = {"type": "invite", "permission": 1}
|
|
inbox_message.status = InboxMessageStatus.PENDING
|
|
|
|
service = ScheduleItemService(
|
|
repository=mock_repo,
|
|
session=mock_session,
|
|
current_user=CurrentUser(id=user_id),
|
|
inbox_repository=FakeInboxRepo(inbox_message),
|
|
)
|
|
|
|
result = await service.accept_subscription(item_id)
|
|
|
|
assert result == {"message": "Subscription accepted"}
|
|
mock_session.commit.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_accept_subscription_not_found(
|
|
mock_session: AsyncMock, mock_repo: MagicMock
|
|
) -> None:
|
|
user_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
item_id = uuid4()
|
|
|
|
service = ScheduleItemService(
|
|
repository=mock_repo,
|
|
session=mock_session,
|
|
current_user=CurrentUser(id=user_id),
|
|
inbox_repository=FakeInboxRepo(None),
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await service.accept_subscription(item_id)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "No pending invitation found" in exc_info.value.detail
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_subscription_success(
|
|
mock_session: AsyncMock, mock_repo: MagicMock
|
|
) -> None:
|
|
user_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
item_id = uuid4()
|
|
|
|
inbox_message = MagicMock(spec=InboxMessage)
|
|
inbox_message.id = uuid4()
|
|
inbox_message.status = InboxMessageStatus.PENDING
|
|
|
|
service = ScheduleItemService(
|
|
repository=mock_repo,
|
|
session=mock_session,
|
|
current_user=CurrentUser(id=user_id),
|
|
inbox_repository=FakeInboxRepo(inbox_message),
|
|
)
|
|
|
|
result = await service.reject_subscription(item_id)
|
|
|
|
assert result == {"message": "Subscription rejected"}
|
|
mock_session.commit.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_subscription_not_found(
|
|
mock_session: AsyncMock, mock_repo: MagicMock
|
|
) -> None:
|
|
user_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
item_id = uuid4()
|
|
|
|
service = ScheduleItemService(
|
|
repository=mock_repo,
|
|
session=mock_session,
|
|
current_user=CurrentUser(id=user_id),
|
|
inbox_repository=FakeInboxRepo(None),
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await service.reject_subscription(item_id)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "No pending invitation found" in exc_info.value.detail
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_by_date_range_with_subscriptions(
|
|
mock_session: AsyncMock, mock_repo: MagicMock
|
|
) -> None:
|
|
user_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
owner_id = UUID("00000000-0000-0000-0000-000000000002")
|
|
item_id = uuid4()
|
|
|
|
owned_item = _create_mock_schedule_item(item_id=item_id, owner_id=user_id)
|
|
subscribed_item = _create_mock_schedule_item(
|
|
item_id=uuid4(), owner_id=owner_id, title="Subscribed Event"
|
|
)
|
|
subscription = MagicMock(spec=ScheduleSubscription)
|
|
subscription.item_id = subscribed_item.id
|
|
subscription.permission = 1
|
|
subscription.subscriber_id = user_id
|
|
|
|
mock_repo.list_by_date_range = AsyncMock(return_value=[owned_item])
|
|
mock_repo.get_user_subscriptions = AsyncMock(return_value=[subscription])
|
|
mock_repo.list_subscribed_items_by_date_range = AsyncMock(
|
|
return_value=[(subscribed_item, subscription)]
|
|
)
|
|
mock_repo.get_by_id = AsyncMock(return_value=subscribed_item)
|
|
|
|
service = ScheduleItemService(
|
|
repository=mock_repo,
|
|
session=mock_session,
|
|
current_user=CurrentUser(id=user_id),
|
|
inbox_repository=FakeInboxRepo(),
|
|
)
|
|
|
|
from v1.schedule_items.schemas import ScheduleItemListRequest
|
|
|
|
request = ScheduleItemListRequest(
|
|
start_at=datetime(2026, 2, 1, 0, 0, 0, tzinfo=timezone.utc),
|
|
end_at=datetime(2026, 3, 1, 0, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
|
|
result = await service.list_by_date_range(request)
|
|
|
|
assert len(result) == 1
|
|
assert result[0].is_owner is False
|
|
assert result[0].permission == 1
|