qzl
157 lines
5.0 KiB
Python
157 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Callable
|
|
from uuid import UUID, uuid4
|
|
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import app
|
|
from v1.inbox_messages.dependencies import get_inbox_message_service
|
|
from v1.inbox_messages.schemas import (
|
|
InboxMessageAcceptRequest,
|
|
InboxMessageListRequest,
|
|
InboxMessageResponse,
|
|
InboxMessageStatus,
|
|
InboxMessageType,
|
|
)
|
|
from v1.inbox_messages.service import InboxMessageService
|
|
|
|
|
|
class FakeInboxMessageService:
|
|
def __init__(
|
|
self,
|
|
messages: list[InboxMessageResponse],
|
|
accepted: InboxMessageResponse,
|
|
dismissed: InboxMessageResponse,
|
|
) -> None:
|
|
self._messages = messages
|
|
self._accepted = accepted
|
|
self._dismissed = dismissed
|
|
|
|
async def list_messages(
|
|
self, request: InboxMessageListRequest
|
|
) -> list[InboxMessageResponse]:
|
|
if request.status is None:
|
|
return self._messages
|
|
return [
|
|
message for message in self._messages if message.status == request.status
|
|
]
|
|
|
|
async def accept_invitation(
|
|
self,
|
|
message_id: UUID,
|
|
request: InboxMessageAcceptRequest,
|
|
) -> InboxMessageResponse:
|
|
if message_id != self._accepted.id:
|
|
raise HTTPException(status_code=404, detail="Inbox message not found")
|
|
if not request.permission_view:
|
|
raise HTTPException(status_code=400, detail="permission_view is required")
|
|
return self._accepted
|
|
|
|
async def dismiss_invitation(self, message_id: UUID) -> InboxMessageResponse:
|
|
if message_id != self._dismissed.id:
|
|
raise HTTPException(status_code=404, detail="Inbox message not found")
|
|
return self._dismissed
|
|
|
|
|
|
def _override_inbox_message_service(
|
|
service: FakeInboxMessageService,
|
|
) -> Callable[[], InboxMessageService]:
|
|
def _get_service() -> InboxMessageService:
|
|
return service # type: ignore[return-value]
|
|
|
|
return _get_service
|
|
|
|
|
|
def _build_message(
|
|
message_id: UUID,
|
|
status: InboxMessageStatus,
|
|
) -> InboxMessageResponse:
|
|
return InboxMessageResponse(
|
|
id=message_id,
|
|
recipient_id=uuid4(),
|
|
sender_id=uuid4(),
|
|
message_type=InboxMessageType.CALENDAR,
|
|
schedule_item_id=uuid4(),
|
|
content='{"permission": 1}',
|
|
is_read=False,
|
|
status=status,
|
|
created_at=datetime(2026, 2, 28, 9, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
|
|
|
|
def test_list_inbox_messages_returns_200() -> None:
|
|
pending_message = _build_message(uuid4(), InboxMessageStatus.PENDING)
|
|
accepted_message = _build_message(uuid4(), InboxMessageStatus.ACCEPTED)
|
|
service = FakeInboxMessageService(
|
|
messages=[pending_message, accepted_message],
|
|
accepted=accepted_message,
|
|
dismissed=_build_message(uuid4(), InboxMessageStatus.DISMISSED),
|
|
)
|
|
app.dependency_overrides[get_inbox_message_service] = (
|
|
_override_inbox_message_service(service)
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.get("/api/v1/inbox/messages", params={"status": "pending"})
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert len(body) == 1
|
|
assert body[0]["status"] == "pending"
|
|
finally:
|
|
app.dependency_overrides = {}
|
|
|
|
|
|
def test_accept_inbox_message_returns_200() -> None:
|
|
accepted_message = _build_message(uuid4(), InboxMessageStatus.ACCEPTED)
|
|
service = FakeInboxMessageService(
|
|
messages=[accepted_message],
|
|
accepted=accepted_message,
|
|
dismissed=_build_message(uuid4(), InboxMessageStatus.DISMISSED),
|
|
)
|
|
app.dependency_overrides[get_inbox_message_service] = (
|
|
_override_inbox_message_service(service)
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.post(
|
|
f"/api/v1/inbox/messages/{accepted_message.id}/accept",
|
|
json={
|
|
"permission_view": True,
|
|
"permission_edit": True,
|
|
"permission_invite": False,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["id"] == str(accepted_message.id)
|
|
assert body["status"] == "accepted"
|
|
finally:
|
|
app.dependency_overrides = {}
|
|
|
|
|
|
def test_dismiss_inbox_message_returns_200() -> None:
|
|
dismissed_message = _build_message(uuid4(), InboxMessageStatus.DISMISSED)
|
|
service = FakeInboxMessageService(
|
|
messages=[dismissed_message],
|
|
accepted=_build_message(uuid4(), InboxMessageStatus.ACCEPTED),
|
|
dismissed=dismissed_message,
|
|
)
|
|
app.dependency_overrides[get_inbox_message_service] = (
|
|
_override_inbox_message_service(service)
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.post(f"/api/v1/inbox/messages/{dismissed_message.id}/dismiss")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["id"] == str(dismissed_message.id)
|
|
assert body["status"] == "dismissed"
|
|
finally:
|
|
app.dependency_overrides = {}
|