qzl
111 lines
3.4 KiB
Python
111 lines
3.4 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 (
|
|
InboxMessageResponse,
|
|
InboxMessageStatus,
|
|
InboxMessageType,
|
|
)
|
|
from v1.inbox_messages.service import InboxMessageService
|
|
|
|
|
|
class FakeInboxMessageService:
|
|
def __init__(
|
|
self,
|
|
messages: list[InboxMessageResponse],
|
|
read_message: InboxMessageResponse,
|
|
) -> None:
|
|
self._messages = messages
|
|
self._read_message = read_message
|
|
|
|
async def list_messages(
|
|
self, is_read: bool | None = None
|
|
) -> list[InboxMessageResponse]:
|
|
if is_read is None:
|
|
return self._messages
|
|
return [message for message in self._messages if message.is_read is is_read]
|
|
|
|
async def mark_as_read(self, message_id: UUID) -> InboxMessageResponse:
|
|
if message_id != self._read_message.id:
|
|
raise HTTPException(status_code=404, detail="Inbox message not found")
|
|
return self._read_message
|
|
|
|
|
|
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)
|
|
read_message = _build_message(uuid4(), InboxMessageStatus.ACCEPTED)
|
|
read_message = read_message.model_copy(update={"is_read": True})
|
|
service = FakeInboxMessageService(
|
|
messages=[pending_message, read_message],
|
|
read_message=read_message,
|
|
)
|
|
app.dependency_overrides[get_inbox_message_service] = (
|
|
_override_inbox_message_service(service)
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.get("/api/v1/inbox/messages", params={"is_read": "false"})
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert len(body) == 1
|
|
assert body[0]["is_read"] is False
|
|
finally:
|
|
app.dependency_overrides = {}
|
|
|
|
|
|
def test_mark_as_read_returns_200() -> None:
|
|
read_message = _build_message(uuid4(), InboxMessageStatus.PENDING)
|
|
read_message = read_message.model_copy(update={"is_read": True})
|
|
service = FakeInboxMessageService(
|
|
messages=[read_message],
|
|
read_message=read_message,
|
|
)
|
|
app.dependency_overrides[get_inbox_message_service] = (
|
|
_override_inbox_message_service(service)
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.patch(f"/api/v1/inbox/messages/{read_message.id}/read")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["id"] == str(read_message.id)
|
|
assert body["is_read"] is True
|
|
finally:
|
|
app.dependency_overrides = {}
|