259 lines
11 KiB
Python
259 lines
11 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from sqlalchemy.exc import IntegrityError, OperationalError, SQLAlchemyError
|
|
from sqlalchemy.future import select
|
|
from sqlalchemy import func as sql_func # For get_list_status
|
|
from datetime import datetime, timezone
|
|
|
|
from app.crud.list import (
|
|
create_list,
|
|
get_lists_for_user,
|
|
get_list_by_id,
|
|
update_list,
|
|
delete_list,
|
|
check_list_permission,
|
|
get_list_status
|
|
)
|
|
from app.schemas.list import ListCreate, ListUpdate, ListStatus
|
|
from app.models import List as ListModel, User as UserModel, Group as GroupModel, UserGroup as UserGroupModel, Item as ItemModel
|
|
from app.core.exceptions import (
|
|
ListNotFoundError,
|
|
ListPermissionError,
|
|
ListCreatorRequiredError,
|
|
DatabaseConnectionError,
|
|
DatabaseIntegrityError,
|
|
DatabaseQueryError,
|
|
DatabaseTransactionError,
|
|
ConflictError
|
|
)
|
|
|
|
# Fixtures
|
|
@pytest.fixture
|
|
def mock_db_session():
|
|
session = AsyncMock()
|
|
session.begin = AsyncMock()
|
|
session.commit = AsyncMock()
|
|
session.rollback = AsyncMock()
|
|
session.refresh = AsyncMock()
|
|
session.add = MagicMock()
|
|
session.delete = MagicMock()
|
|
session.execute = AsyncMock()
|
|
session.get = AsyncMock() # Used by check_list_permission via get_list_by_id
|
|
session.flush = AsyncMock()
|
|
return session
|
|
|
|
@pytest.fixture
|
|
def list_create_data():
|
|
return ListCreate(name="New Shopping List", description="Groceries for the week")
|
|
|
|
@pytest.fixture
|
|
def list_update_data():
|
|
return ListUpdate(name="Updated Shopping List", description="Weekend Groceries", version=1)
|
|
|
|
@pytest.fixture
|
|
def user_model():
|
|
return UserModel(id=1, name="Test User", email="test@example.com")
|
|
|
|
@pytest.fixture
|
|
def another_user_model():
|
|
return UserModel(id=2, name="Another User", email="another@example.com")
|
|
|
|
@pytest.fixture
|
|
def group_model():
|
|
return GroupModel(id=1, name="Test Group")
|
|
|
|
@pytest.fixture
|
|
def db_list_personal_model(user_model):
|
|
return ListModel(
|
|
id=1, name="Personal List", created_by_id=user_model.id, creator=user_model,
|
|
version=1, updated_at=datetime.now(timezone.utc), items=[]
|
|
)
|
|
|
|
@pytest.fixture
|
|
def db_list_group_model(user_model, group_model):
|
|
return ListModel(
|
|
id=2, name="Group List", created_by_id=user_model.id, creator=user_model,
|
|
group_id=group_model.id, group=group_model, version=1, updated_at=datetime.now(timezone.utc), items=[]
|
|
)
|
|
|
|
# --- create_list Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_create_list_success(mock_db_session, list_create_data, user_model):
|
|
async def mock_refresh(instance):
|
|
instance.id = 100
|
|
instance.version = 1
|
|
instance.updated_at = datetime.now(timezone.utc)
|
|
return None
|
|
mock_db_session.refresh.return_value = None
|
|
mock_db_session.refresh.side_effect = mock_refresh
|
|
|
|
created_list = await create_list(mock_db_session, list_create_data, user_model.id)
|
|
mock_db_session.add.assert_called_once()
|
|
mock_db_session.flush.assert_called_once()
|
|
mock_db_session.refresh.assert_called_once()
|
|
assert created_list.name == list_create_data.name
|
|
assert created_list.created_by_id == user_model.id
|
|
|
|
# --- get_lists_for_user Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_get_lists_for_user_mix(mock_db_session, user_model, db_list_personal_model, db_list_group_model):
|
|
# Simulate user is part of group for db_list_group_model
|
|
mock_group_ids_result = AsyncMock()
|
|
mock_group_ids_result.scalars.return_value.all.return_value = [db_list_group_model.group_id]
|
|
|
|
mock_lists_result = AsyncMock()
|
|
# Order should be personal list (created by user_id) then group list
|
|
mock_lists_result.scalars.return_value.all.return_value = [db_list_personal_model, db_list_group_model]
|
|
|
|
mock_db_session.execute.side_effect = [mock_group_ids_result, mock_lists_result]
|
|
|
|
lists = await get_lists_for_user(mock_db_session, user_model.id)
|
|
assert len(lists) == 2
|
|
assert db_list_personal_model in lists
|
|
assert db_list_group_model in lists
|
|
assert mock_db_session.execute.call_count == 2
|
|
|
|
# --- get_list_by_id Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_get_list_by_id_found_no_items(mock_db_session, db_list_personal_model):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalars.return_value.first.return_value = db_list_personal_model
|
|
mock_db_session.execute.return_value = mock_result
|
|
|
|
found_list = await get_list_by_id(mock_db_session, db_list_personal_model.id, load_items=False)
|
|
assert found_list is not None
|
|
assert found_list.id == db_list_personal_model.id
|
|
# query options should not include selectinload for items
|
|
# (difficult to assert directly without inspecting query object in detail)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_list_by_id_found_with_items(mock_db_session, db_list_personal_model):
|
|
# Simulate items loaded for the list
|
|
db_list_personal_model.items = [ItemModel(id=1, name="Test Item")]
|
|
mock_result = AsyncMock()
|
|
mock_result.scalars.return_value.first.return_value = db_list_personal_model
|
|
mock_db_session.execute.return_value = mock_result
|
|
|
|
found_list = await get_list_by_id(mock_db_session, db_list_personal_model.id, load_items=True)
|
|
assert found_list is not None
|
|
assert len(found_list.items) == 1
|
|
# query options should include selectinload for items
|
|
|
|
# --- update_list Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_update_list_success(mock_db_session, db_list_personal_model, list_update_data):
|
|
list_update_data.version = db_list_personal_model.version # Match version
|
|
|
|
updated_list = await update_list(mock_db_session, db_list_personal_model, list_update_data)
|
|
assert updated_list.name == list_update_data.name
|
|
assert updated_list.version == db_list_personal_model.version # version incremented in db_list_personal_model
|
|
mock_db_session.add.assert_called_once_with(db_list_personal_model)
|
|
mock_db_session.flush.assert_called_once()
|
|
mock_db_session.refresh.assert_called_once_with(db_list_personal_model)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_list_conflict(mock_db_session, db_list_personal_model, list_update_data):
|
|
list_update_data.version = db_list_personal_model.version + 1 # Version mismatch
|
|
with pytest.raises(ConflictError):
|
|
await update_list(mock_db_session, db_list_personal_model, list_update_data)
|
|
mock_db_session.rollback.assert_called_once()
|
|
|
|
# --- delete_list Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_delete_list_success(mock_db_session, db_list_personal_model):
|
|
await delete_list(mock_db_session, db_list_personal_model)
|
|
mock_db_session.delete.assert_called_once_with(db_list_personal_model)
|
|
mock_db_session.commit.assert_called_once() # from async with db.begin()
|
|
|
|
# --- check_list_permission Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_check_list_permission_creator_access_personal_list(mock_db_session, db_list_personal_model, user_model):
|
|
# get_list_by_id (called by check_list_permission) will mock execute
|
|
mock_list_fetch_result = AsyncMock()
|
|
mock_list_fetch_result.scalars.return_value.first.return_value = db_list_personal_model
|
|
mock_db_session.execute.return_value = mock_list_fetch_result
|
|
|
|
ret_list = await check_list_permission(mock_db_session, db_list_personal_model.id, user_model.id)
|
|
assert ret_list.id == db_list_personal_model.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_list_permission_group_member_access_group_list(mock_db_session, db_list_group_model, another_user_model, group_model):
|
|
# User `another_user_model` is not creator but member of the group
|
|
db_list_group_model.creator_id = user_model.id # Original creator is user_model
|
|
db_list_group_model.creator = user_model
|
|
|
|
# Mock get_list_by_id internal call
|
|
mock_list_fetch_result = AsyncMock()
|
|
mock_list_fetch_result.scalars.return_value.first.return_value = db_list_group_model
|
|
|
|
# Mock is_user_member call
|
|
with patch('app.crud.list.is_user_member', new_callable=AsyncMock) as mock_is_member:
|
|
mock_is_member.return_value = True # another_user_model is a member
|
|
mock_db_session.execute.return_value = mock_list_fetch_result
|
|
|
|
ret_list = await check_list_permission(mock_db_session, db_list_group_model.id, another_user_model.id)
|
|
assert ret_list.id == db_list_group_model.id
|
|
mock_is_member.assert_called_once_with(mock_db_session, group_id=group_model.id, user_id=another_user_model.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_list_permission_non_member_no_access_group_list(mock_db_session, db_list_group_model, another_user_model):
|
|
db_list_group_model.creator_id = user_model.id # Creator is not another_user_model
|
|
|
|
mock_list_fetch_result = AsyncMock()
|
|
mock_list_fetch_result.scalars.return_value.first.return_value = db_list_group_model
|
|
|
|
with patch('app.crud.list.is_user_member', new_callable=AsyncMock) as mock_is_member:
|
|
mock_is_member.return_value = False # another_user_model is NOT a member
|
|
mock_db_session.execute.return_value = mock_list_fetch_result
|
|
|
|
with pytest.raises(ListPermissionError):
|
|
await check_list_permission(mock_db_session, db_list_group_model.id, another_user_model.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_list_permission_list_not_found(mock_db_session, user_model):
|
|
mock_list_fetch_result = AsyncMock()
|
|
mock_list_fetch_result.scalars.return_value.first.return_value = None # List not found
|
|
mock_db_session.execute.return_value = mock_list_fetch_result
|
|
|
|
with pytest.raises(ListNotFoundError):
|
|
await check_list_permission(mock_db_session, 999, user_model.id)
|
|
|
|
# --- get_list_status Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_get_list_status_success(mock_db_session, db_list_personal_model):
|
|
list_updated_at = datetime.now(timezone.utc) - timezone.timedelta(hours=1)
|
|
item_updated_at = datetime.now(timezone.utc)
|
|
item_count = 5
|
|
|
|
db_list_personal_model.updated_at = list_updated_at
|
|
|
|
# Mock for ListModel.updated_at query
|
|
mock_list_updated_result = AsyncMock()
|
|
mock_list_updated_result.scalar_one_or_none.return_value = list_updated_at
|
|
|
|
# Mock for ItemModel status query
|
|
mock_item_status_result = AsyncMock()
|
|
# SQLAlchemy query for func.max and func.count returns a Row-like object or None
|
|
mock_item_status_row = MagicMock()
|
|
mock_item_status_row.latest_item_updated_at = item_updated_at
|
|
mock_item_status_row.item_count = item_count
|
|
mock_item_status_result.first.return_value = mock_item_status_row
|
|
|
|
mock_db_session.execute.side_effect = [mock_list_updated_result, mock_item_status_result]
|
|
|
|
status = await get_list_status(mock_db_session, db_list_personal_model.id)
|
|
assert status.list_updated_at == list_updated_at
|
|
assert status.latest_item_updated_at == item_updated_at
|
|
assert status.item_count == item_count
|
|
assert mock_db_session.execute.call_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_list_status_list_not_found(mock_db_session):
|
|
mock_list_updated_result = AsyncMock()
|
|
mock_list_updated_result.scalar_one_or_none.return_value = None # List not found
|
|
mock_db_session.execute.return_value = mock_list_updated_result
|
|
with pytest.raises(ListNotFoundError):
|
|
await get_list_status(mock_db_session, 999)
|
|
|
|
# TODO: Add more specific DB error tests (Operational, SQLAlchemyError, IntegrityError) for each function.
|
|
# TODO: Test check_list_permission with require_creator=True cases. |