qzl
376 lines
12 KiB
Python
376 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any, cast
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from core.http.errors import ApiProblemError
|
|
from core.auth.models import CurrentUser
|
|
from models.inbox_messages import InboxMessage, InboxMessageType
|
|
from models.schedule_items import ScheduleItem
|
|
from schemas.enums import FriendshipStatus
|
|
from v1.auth.schemas import UserByPhoneResponse
|
|
from v1.schedule_items.repository import ScheduleItemRepository
|
|
from v1.schedule_items.schemas import ScheduleItemShareRequest
|
|
from v1.schedule_items.service import ScheduleItemService
|
|
|
|
|
|
def test_share_request_schema() -> None:
|
|
request = ScheduleItemShareRequest(
|
|
phone="+8613810000000",
|
|
permission_view=True,
|
|
permission_edit=True,
|
|
permission_invite=False,
|
|
)
|
|
assert request.phone == "+8613810000000"
|
|
assert request.permission_view is True
|
|
|
|
|
|
def test_permission_bits_calculation() -> None:
|
|
request = ScheduleItemShareRequest(
|
|
phone="+8613810000000",
|
|
permission_view=True,
|
|
permission_edit=True,
|
|
permission_invite=False,
|
|
)
|
|
assert request._permission_value() == 5
|
|
|
|
|
|
def _build_item(item_id: UUID, owner_id: UUID) -> ScheduleItem:
|
|
item = MagicMock(spec=ScheduleItem)
|
|
item.id = item_id
|
|
item.owner_id = owner_id
|
|
item.title = "test"
|
|
item.description = None
|
|
item.start_at = datetime(2026, 2, 28, 16, 0, 0, tzinfo=timezone.utc)
|
|
item.end_at = None
|
|
item.timezone = "UTC"
|
|
item.extra_metadata = {}
|
|
item.created_at = datetime(2026, 2, 28, 10, 0, 0, tzinfo=timezone.utc)
|
|
item.updated_at = datetime(2026, 2, 28, 10, 0, 0, tzinfo=timezone.utc)
|
|
return item
|
|
|
|
|
|
class ShareRepo:
|
|
def __init__(self, item: ScheduleItem | None) -> None:
|
|
self._item = item
|
|
|
|
async def get_by_id(self, item_id: UUID) -> ScheduleItem | None:
|
|
if self._item and self._item.id == item_id:
|
|
return self._item
|
|
return None
|
|
|
|
async def get_subscription(self, item_id: UUID, subscriber_id: UUID) -> None:
|
|
return None
|
|
|
|
async def create_subscription(self, data: dict[str, object]) -> None:
|
|
return None
|
|
|
|
|
|
class AuthGatewayStub:
|
|
async def get_user_by_phone(self, phone: str) -> UserByPhoneResponse:
|
|
return UserByPhoneResponse(
|
|
id="00000000-0000-0000-0000-000000000222",
|
|
phone=phone,
|
|
created_at="2026-02-28T10:00:00Z",
|
|
phone_confirmed_at=None,
|
|
)
|
|
|
|
async def get_user_by_id(self, user_id: str) -> UserByPhoneResponse:
|
|
return UserByPhoneResponse(
|
|
id=user_id,
|
|
phone="+8613810000000",
|
|
created_at="2026-02-28T10:00:00Z",
|
|
phone_confirmed_at=None,
|
|
)
|
|
|
|
|
|
class InboxRepoStub:
|
|
async def create(self, data: dict[str, object]) -> InboxMessage:
|
|
return InboxMessage(
|
|
id=uuid4(),
|
|
recipient_id=UUID("00000000-0000-0000-0000-000000000222"),
|
|
sender_id=UUID("00000000-0000-0000-0000-000000000001"),
|
|
message_type=InboxMessageType.CALENDAR,
|
|
schedule_item_id=uuid4(),
|
|
content='{"type": "invite", "permission": 1, "action": "pending"}',
|
|
created_by=UUID("00000000-0000-0000-0000-000000000001"),
|
|
)
|
|
|
|
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
|
|
|
|
async def get_pending_calendar_invite(
|
|
self, schedule_item_id: UUID, recipient_id: UUID
|
|
) -> InboxMessage | None:
|
|
return None
|
|
|
|
async def get_calendar_invite(
|
|
self, schedule_item_id: UUID, recipient_id: UUID
|
|
) -> InboxMessage | None:
|
|
return None
|
|
|
|
|
|
class AuthGatewayInvalidIdStub:
|
|
async def get_user_by_phone(self, phone: str) -> UserByPhoneResponse:
|
|
return UserByPhoneResponse(
|
|
id="not-a-uuid",
|
|
phone=phone,
|
|
created_at="2026-02-28T10:00:00Z",
|
|
phone_confirmed_at=None,
|
|
)
|
|
|
|
async def get_user_by_id(self, user_id: str) -> UserByPhoneResponse:
|
|
return UserByPhoneResponse(
|
|
id=user_id,
|
|
phone="+8613810000000",
|
|
created_at="2026-02-28T10:00:00Z",
|
|
phone_confirmed_at=None,
|
|
)
|
|
|
|
|
|
class FriendshipRepoStub:
|
|
def __init__(self, accepted: bool = True) -> None:
|
|
self._accepted = accepted
|
|
|
|
async def get_friendship_between_users(self, user_id_1: UUID, user_id_2: UUID):
|
|
if not self._accepted:
|
|
return None
|
|
friendship = MagicMock()
|
|
friendship.status = FriendshipStatus.ACCEPTED
|
|
return friendship
|
|
|
|
|
|
class UserRepoStub:
|
|
async def get_by_user_id(self, user_id: UUID):
|
|
profile = MagicMock()
|
|
profile.id = user_id
|
|
profile.username = "owner"
|
|
return profile
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_share_forbidden_when_not_owner() -> None:
|
|
owner_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
requester_id = UUID("00000000-0000-0000-0000-000000000002")
|
|
item_id = uuid4()
|
|
service = ScheduleItemService(
|
|
repository=cast(
|
|
ScheduleItemRepository,
|
|
ShareRepo(_build_item(item_id=item_id, owner_id=owner_id)),
|
|
),
|
|
session=AsyncMock(),
|
|
current_user=CurrentUser(id=requester_id),
|
|
auth_gateway=cast(Any, AuthGatewayStub()),
|
|
inbox_repository=InboxRepoStub(),
|
|
friendship_repository=cast(Any, FriendshipRepoStub()),
|
|
user_repository=cast(Any, UserRepoStub()),
|
|
)
|
|
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
|
await service.share(
|
|
item_id,
|
|
ScheduleItemShareRequest(
|
|
phone="+8613810000000",
|
|
permission_view=True,
|
|
permission_edit=False,
|
|
permission_invite=False,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_share_success_creates_calendar_invitation_message() -> None:
|
|
owner_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
item_id = uuid4()
|
|
session = AsyncMock()
|
|
session.add = MagicMock()
|
|
service = ScheduleItemService(
|
|
repository=cast(
|
|
ScheduleItemRepository,
|
|
ShareRepo(_build_item(item_id=item_id, owner_id=owner_id)),
|
|
),
|
|
session=session,
|
|
current_user=CurrentUser(id=owner_id),
|
|
auth_gateway=cast(Any, AuthGatewayStub()),
|
|
inbox_repository=InboxRepoStub(),
|
|
friendship_repository=cast(Any, FriendshipRepoStub()),
|
|
user_repository=cast(Any, UserRepoStub()),
|
|
)
|
|
|
|
result = await service.share(
|
|
item_id,
|
|
ScheduleItemShareRequest(
|
|
phone="+8613810000000",
|
|
permission_view=True,
|
|
permission_edit=True,
|
|
permission_invite=False,
|
|
),
|
|
)
|
|
|
|
assert result.message == "Calendar invitation sent"
|
|
session.add.assert_called_once()
|
|
message = session.add.call_args.args[0]
|
|
assert isinstance(message, InboxMessage)
|
|
assert message.sender_id == owner_id
|
|
assert message.schedule_item_id == item_id
|
|
assert message.message_type == InboxMessageType.CALENDAR
|
|
assert message.content is not None
|
|
assert message.content["type"] == "invite"
|
|
assert message.content["schema_version"] == 2
|
|
assert message.content["permission"] == 5
|
|
assert message.content["item"]["id"] == str(item_id)
|
|
assert message.content["item"]["title"] == "test"
|
|
assert message.content["item"]["start_at"] == "2026-02-28T16:00:00+00:00"
|
|
assert message.content["item"]["end_at"] is None
|
|
assert message.content["item"]["timezone"] == "UTC"
|
|
assert message.content["actor"]["username"] == "owner"
|
|
assert message.content["actor"]["phone"] == "+8613810000000"
|
|
session.commit.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_share_returns_not_found_when_item_missing() -> None:
|
|
requester_id = UUID("00000000-0000-0000-0000-000000000002")
|
|
service = ScheduleItemService(
|
|
repository=cast(ScheduleItemRepository, ShareRepo(None)),
|
|
session=AsyncMock(),
|
|
current_user=CurrentUser(id=requester_id),
|
|
auth_gateway=cast(Any, AuthGatewayStub()),
|
|
inbox_repository=InboxRepoStub(),
|
|
friendship_repository=cast(Any, FriendshipRepoStub()),
|
|
user_repository=cast(Any, UserRepoStub()),
|
|
)
|
|
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
|
await service.share(
|
|
uuid4(),
|
|
ScheduleItemShareRequest(
|
|
phone="+8613810000000",
|
|
permission_view=True,
|
|
permission_edit=False,
|
|
permission_invite=False,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_share_invalid_auth_user_id_returns_503() -> None:
|
|
owner_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
item_id = uuid4()
|
|
session = AsyncMock()
|
|
service = ScheduleItemService(
|
|
repository=cast(
|
|
ScheduleItemRepository,
|
|
ShareRepo(_build_item(item_id=item_id, owner_id=owner_id)),
|
|
),
|
|
session=session,
|
|
current_user=CurrentUser(id=owner_id),
|
|
auth_gateway=cast(Any, AuthGatewayInvalidIdStub()),
|
|
inbox_repository=InboxRepoStub(),
|
|
friendship_repository=cast(Any, FriendshipRepoStub()),
|
|
user_repository=cast(Any, UserRepoStub()),
|
|
)
|
|
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
|
await service.share(
|
|
item_id,
|
|
ScheduleItemShareRequest(
|
|
phone="+8613810000000",
|
|
permission_view=True,
|
|
permission_edit=False,
|
|
permission_invite=False,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 503
|
|
assert exc_info.value.detail == "Auth lookup unavailable"
|
|
session.rollback.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_share_sqlalchemy_error_rolls_back() -> None:
|
|
owner_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
item_id = uuid4()
|
|
session = AsyncMock()
|
|
session.add = MagicMock(side_effect=SQLAlchemyError("db error"))
|
|
service = ScheduleItemService(
|
|
repository=cast(
|
|
ScheduleItemRepository,
|
|
ShareRepo(_build_item(item_id=item_id, owner_id=owner_id)),
|
|
),
|
|
session=session,
|
|
current_user=CurrentUser(id=owner_id),
|
|
auth_gateway=cast(Any, AuthGatewayStub()),
|
|
inbox_repository=InboxRepoStub(),
|
|
friendship_repository=cast(Any, FriendshipRepoStub()),
|
|
user_repository=cast(Any, UserRepoStub()),
|
|
)
|
|
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
|
await service.share(
|
|
item_id,
|
|
ScheduleItemShareRequest(
|
|
phone="+8613810000000",
|
|
permission_view=True,
|
|
permission_edit=False,
|
|
permission_invite=False,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 503
|
|
assert exc_info.value.detail == "Schedule item store unavailable"
|
|
session.rollback.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_share_returns_forbidden_when_target_is_not_friend() -> None:
|
|
owner_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
item_id = uuid4()
|
|
service = ScheduleItemService(
|
|
repository=cast(
|
|
ScheduleItemRepository,
|
|
ShareRepo(_build_item(item_id=item_id, owner_id=owner_id)),
|
|
),
|
|
session=AsyncMock(),
|
|
current_user=CurrentUser(id=owner_id),
|
|
auth_gateway=cast(Any, AuthGatewayStub()),
|
|
inbox_repository=InboxRepoStub(),
|
|
friendship_repository=cast(Any, FriendshipRepoStub(accepted=False)),
|
|
user_repository=cast(Any, UserRepoStub()),
|
|
)
|
|
|
|
with pytest.raises(ApiProblemError) as exc_info:
|
|
await service.share(
|
|
item_id,
|
|
ScheduleItemShareRequest(
|
|
phone="+8613810000000",
|
|
permission_view=True,
|
|
permission_edit=False,
|
|
permission_invite=False,
|
|
),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
assert exc_info.value.detail == "You can only share calendar with accepted friends"
|