qzl
89 lines
2.7 KiB
Python
89 lines
2.7 KiB
Python
from unittest.mock import AsyncMock, MagicMock
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from models.inbox_messages import InboxMessageType
|
|
from v1.inbox_messages.repository import SQLAlchemyInboxMessageRepository
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_adds_message_and_flushes() -> None:
|
|
session = AsyncMock()
|
|
session.add = MagicMock()
|
|
repository = SQLAlchemyInboxMessageRepository(session)
|
|
recipient_id = uuid4()
|
|
|
|
result = await repository.create(
|
|
{
|
|
"recipient_id": recipient_id,
|
|
"sender_id": uuid4(),
|
|
"message_type": InboxMessageType.CALENDAR,
|
|
"schedule_item_id": uuid4(),
|
|
"content": "invite",
|
|
"created_by": uuid4(),
|
|
}
|
|
)
|
|
|
|
session.add.assert_called_once_with(result)
|
|
session.flush.assert_awaited_once()
|
|
assert result.recipient_id == recipient_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_id_returns_message_when_exists() -> None:
|
|
session = AsyncMock()
|
|
repository = SQLAlchemyInboxMessageRepository(session)
|
|
expected = MagicMock()
|
|
execute_result = MagicMock()
|
|
execute_result.scalar_one_or_none.return_value = expected
|
|
session.execute.return_value = execute_result
|
|
|
|
result = await repository.get_by_id(uuid4(), uuid4())
|
|
|
|
assert result is expected
|
|
session.execute.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_by_recipient_returns_messages() -> None:
|
|
session = AsyncMock()
|
|
repository = SQLAlchemyInboxMessageRepository(session)
|
|
message_one = MagicMock()
|
|
message_two = MagicMock()
|
|
execute_result = MagicMock()
|
|
execute_result.scalars.return_value.all.return_value = [message_one, message_two]
|
|
session.execute.return_value = execute_result
|
|
|
|
result = await repository.list_by_recipient(uuid4(), False)
|
|
|
|
assert result == [message_one, message_two]
|
|
session.execute.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_as_read_returns_updated_message_and_flushes() -> None:
|
|
session = AsyncMock()
|
|
repository = SQLAlchemyInboxMessageRepository(session)
|
|
updated = MagicMock()
|
|
execute_result = MagicMock()
|
|
execute_result.scalar_one_or_none.return_value = updated
|
|
session.execute.return_value = execute_result
|
|
|
|
result = await repository.mark_as_read(uuid4(), uuid4())
|
|
|
|
assert result is updated
|
|
session.execute.assert_awaited_once()
|
|
session.flush.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_id_propagates_sqlalchemy_error() -> None:
|
|
session = AsyncMock()
|
|
repository = SQLAlchemyInboxMessageRepository(session)
|
|
session.execute.side_effect = SQLAlchemyError("boom")
|
|
|
|
with pytest.raises(SQLAlchemyError):
|
|
await repository.get_by_id(uuid4(), uuid4())
|