270 lines
12 KiB
Python
270 lines
12 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 delete, func # For remove_user_from_group and get_group_member_count
|
|
|
|
from app.crud.group import (
|
|
create_group,
|
|
get_user_groups,
|
|
get_group_by_id,
|
|
is_user_member,
|
|
get_user_role_in_group,
|
|
add_user_to_group,
|
|
remove_user_from_group,
|
|
get_group_member_count,
|
|
check_group_membership,
|
|
check_user_role_in_group
|
|
)
|
|
from app.schemas.group import GroupCreate
|
|
from app.models import User as UserModel, Group as GroupModel, UserGroup as UserGroupModel, UserRoleEnum
|
|
from app.core.exceptions import (
|
|
GroupOperationError,
|
|
GroupNotFoundError,
|
|
DatabaseConnectionError,
|
|
DatabaseIntegrityError,
|
|
DatabaseQueryError,
|
|
DatabaseTransactionError,
|
|
GroupMembershipError,
|
|
GroupPermissionError
|
|
)
|
|
|
|
# Fixtures
|
|
@pytest.fixture
|
|
def mock_db_session():
|
|
session = AsyncMock()
|
|
# Patch begin_nested for SQLAlchemy 1.4+ if used, or just begin() if that's the pattern
|
|
# For simplicity, assuming `async with db.begin():` translates to db.begin() and db.commit()/rollback()
|
|
session.begin = AsyncMock() # Mock the begin call used in async with db.begin()
|
|
session.commit = AsyncMock()
|
|
session.rollback = AsyncMock()
|
|
session.refresh = AsyncMock()
|
|
session.add = MagicMock()
|
|
session.delete = MagicMock() # For remove_user_from_group (if it uses session.delete)
|
|
session.execute = AsyncMock()
|
|
session.get = AsyncMock()
|
|
session.flush = AsyncMock()
|
|
return session
|
|
|
|
@pytest.fixture
|
|
def group_create_data():
|
|
return GroupCreate(name="Test Group")
|
|
|
|
@pytest.fixture
|
|
def creator_user_model():
|
|
return UserModel(id=1, name="Creator User", email="creator@example.com")
|
|
|
|
@pytest.fixture
|
|
def member_user_model():
|
|
return UserModel(id=2, name="Member User", email="member@example.com")
|
|
|
|
@pytest.fixture
|
|
def db_group_model(creator_user_model):
|
|
return GroupModel(id=1, name="Test Group", created_by_id=creator_user_model.id, creator=creator_user_model)
|
|
|
|
@pytest.fixture
|
|
def db_user_group_owner_assoc(db_group_model, creator_user_model):
|
|
return UserGroupModel(user_id=creator_user_model.id, group_id=db_group_model.id, role=UserRoleEnum.owner, user=creator_user_model, group=db_group_model)
|
|
|
|
@pytest.fixture
|
|
def db_user_group_member_assoc(db_group_model, member_user_model):
|
|
return UserGroupModel(user_id=member_user_model.id, group_id=db_group_model.id, role=UserRoleEnum.member, user=member_user_model, group=db_group_model)
|
|
|
|
# --- create_group Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_create_group_success(mock_db_session, group_create_data, creator_user_model):
|
|
async def mock_refresh(instance):
|
|
instance.id = 1 # Simulate ID assignment by DB
|
|
return None
|
|
mock_db_session.refresh = AsyncMock(side_effect=mock_refresh)
|
|
|
|
created_group = await create_group(mock_db_session, group_create_data, creator_user_model.id)
|
|
|
|
assert mock_db_session.add.call_count == 2 # Group and UserGroup
|
|
mock_db_session.flush.assert_called() # Called multiple times
|
|
mock_db_session.refresh.assert_called_once_with(created_group)
|
|
assert created_group is not None
|
|
assert created_group.name == group_create_data.name
|
|
assert created_group.created_by_id == creator_user_model.id
|
|
# Further check if UserGroup was created correctly by inspecting mock_db_session.add calls or by fetching
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_group_integrity_error(mock_db_session, group_create_data, creator_user_model):
|
|
mock_db_session.flush.side_effect = IntegrityError("mock integrity error", "params", "orig")
|
|
with pytest.raises(DatabaseIntegrityError):
|
|
await create_group(mock_db_session, group_create_data, creator_user_model.id)
|
|
mock_db_session.rollback.assert_called_once() # Assuming rollback within the except block of create_group
|
|
|
|
# --- get_user_groups Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_groups_success(mock_db_session, db_group_model, creator_user_model):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalars.return_value.all.return_value = [db_group_model]
|
|
mock_db_session.execute.return_value = mock_result
|
|
|
|
groups = await get_user_groups(mock_db_session, creator_user_model.id)
|
|
assert len(groups) == 1
|
|
assert groups[0].name == db_group_model.name
|
|
mock_db_session.execute.assert_called_once()
|
|
|
|
# --- get_group_by_id Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_get_group_by_id_found(mock_db_session, db_group_model):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalars.return_value.first.return_value = db_group_model
|
|
mock_db_session.execute.return_value = mock_result
|
|
|
|
group = await get_group_by_id(mock_db_session, db_group_model.id)
|
|
assert group is not None
|
|
assert group.id == db_group_model.id
|
|
# Add assertions for eager loaded members if applicable and mocked
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_group_by_id_not_found(mock_db_session):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalars.return_value.first.return_value = None
|
|
mock_db_session.execute.return_value = mock_result
|
|
group = await get_group_by_id(mock_db_session, 999)
|
|
assert group is None
|
|
|
|
# --- is_user_member Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_is_user_member_true(mock_db_session, db_group_model, creator_user_model):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalar_one_or_none.return_value = 1 # Simulate UserGroup.id found
|
|
mock_db_session.execute.return_value = mock_result
|
|
is_member = await is_user_member(mock_db_session, db_group_model.id, creator_user_model.id)
|
|
assert is_member is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_user_member_false(mock_db_session, db_group_model, member_user_model):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalar_one_or_none.return_value = None # Simulate no UserGroup.id found
|
|
mock_db_session.execute.return_value = mock_result
|
|
is_member = await is_user_member(mock_db_session, db_group_model.id, member_user_model.id + 1) # Non-member
|
|
assert is_member is False
|
|
|
|
# --- get_user_role_in_group Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_role_in_group_owner(mock_db_session, db_group_model, creator_user_model):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalar_one_or_none.return_value = UserRoleEnum.owner
|
|
mock_db_session.execute.return_value = mock_result
|
|
role = await get_user_role_in_group(mock_db_session, db_group_model.id, creator_user_model.id)
|
|
assert role == UserRoleEnum.owner
|
|
|
|
# --- add_user_to_group Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_add_user_to_group_new_member(mock_db_session, db_group_model, member_user_model):
|
|
# First execute call for checking existing membership returns None
|
|
mock_existing_check_result = AsyncMock()
|
|
mock_existing_check_result.scalar_one_or_none.return_value = None
|
|
mock_db_session.execute.return_value = mock_existing_check_result
|
|
|
|
async def mock_refresh_user_group(instance):
|
|
instance.id = 100 # Simulate ID for UserGroupModel
|
|
return None
|
|
mock_db_session.refresh = AsyncMock(side_effect=mock_refresh_user_group)
|
|
|
|
user_group_assoc = await add_user_to_group(mock_db_session, db_group_model.id, member_user_model.id, UserRoleEnum.member)
|
|
|
|
mock_db_session.add.assert_called_once()
|
|
mock_db_session.flush.assert_called_once()
|
|
mock_db_session.refresh.assert_called_once()
|
|
assert user_group_assoc is not None
|
|
assert user_group_assoc.user_id == member_user_model.id
|
|
assert user_group_assoc.group_id == db_group_model.id
|
|
assert user_group_assoc.role == UserRoleEnum.member
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_user_to_group_already_member(mock_db_session, db_group_model, creator_user_model, db_user_group_owner_assoc):
|
|
mock_existing_check_result = AsyncMock()
|
|
mock_existing_check_result.scalar_one_or_none.return_value = db_user_group_owner_assoc # User is already a member
|
|
mock_db_session.execute.return_value = mock_existing_check_result
|
|
|
|
user_group_assoc = await add_user_to_group(mock_db_session, db_group_model.id, creator_user_model.id)
|
|
assert user_group_assoc is None
|
|
mock_db_session.add.assert_not_called()
|
|
|
|
# --- remove_user_from_group Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_remove_user_from_group_success(mock_db_session, db_group_model, member_user_model):
|
|
mock_delete_result = AsyncMock()
|
|
mock_delete_result.scalar_one_or_none.return_value = 1 # Simulate a row was deleted (returning ID)
|
|
mock_db_session.execute.return_value = mock_delete_result
|
|
|
|
removed = await remove_user_from_group(mock_db_session, db_group_model.id, member_user_model.id)
|
|
assert removed is True
|
|
# Assert that db.execute was called with a delete statement
|
|
# This requires inspecting the call args of mock_db_session.execute
|
|
# For simplicity, we check it was called. A deeper check would validate the SQL query itself.
|
|
mock_db_session.execute.assert_called_once()
|
|
|
|
# --- get_group_member_count Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_get_group_member_count_success(mock_db_session, db_group_model):
|
|
mock_count_result = AsyncMock()
|
|
mock_count_result.scalar_one.return_value = 5
|
|
mock_db_session.execute.return_value = mock_count_result
|
|
count = await get_group_member_count(mock_db_session, db_group_model.id)
|
|
assert count == 5
|
|
|
|
# --- check_group_membership Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_check_group_membership_is_member(mock_db_session, db_group_model, creator_user_model):
|
|
mock_db_session.get.return_value = db_group_model # Group exists
|
|
mock_membership_result = AsyncMock()
|
|
mock_membership_result.scalar_one_or_none.return_value = 1 # User is a member
|
|
mock_db_session.execute.return_value = mock_membership_result
|
|
|
|
await check_group_membership(mock_db_session, db_group_model.id, creator_user_model.id)
|
|
# No exception means success
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_group_membership_group_not_found(mock_db_session, creator_user_model):
|
|
mock_db_session.get.return_value = None # Group does not exist
|
|
with pytest.raises(GroupNotFoundError):
|
|
await check_group_membership(mock_db_session, 999, creator_user_model.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_group_membership_not_member(mock_db_session, db_group_model, member_user_model):
|
|
mock_db_session.get.return_value = db_group_model # Group exists
|
|
mock_membership_result = AsyncMock()
|
|
mock_membership_result.scalar_one_or_none.return_value = None # User is not a member
|
|
mock_db_session.execute.return_value = mock_membership_result
|
|
with pytest.raises(GroupMembershipError):
|
|
await check_group_membership(mock_db_session, db_group_model.id, member_user_model.id)
|
|
|
|
# --- check_user_role_in_group Tests ---
|
|
@pytest.mark.asyncio
|
|
async def test_check_user_role_in_group_sufficient_role(mock_db_session, db_group_model, creator_user_model):
|
|
# Mock check_group_membership (implicitly called)
|
|
mock_db_session.get.return_value = db_group_model
|
|
mock_membership_check = AsyncMock()
|
|
mock_membership_check.scalar_one_or_none.return_value = 1 # User is member
|
|
|
|
# Mock get_user_role_in_group
|
|
mock_role_check = AsyncMock()
|
|
mock_role_check.scalar_one_or_none.return_value = UserRoleEnum.owner
|
|
|
|
mock_db_session.execute.side_effect = [mock_membership_check, mock_role_check]
|
|
|
|
await check_user_role_in_group(mock_db_session, db_group_model.id, creator_user_model.id, UserRoleEnum.member)
|
|
# No exception means success
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_user_role_in_group_insufficient_role(mock_db_session, db_group_model, member_user_model):
|
|
mock_db_session.get.return_value = db_group_model # Group exists
|
|
mock_membership_check = AsyncMock()
|
|
mock_membership_check.scalar_one_or_none.return_value = 1 # User is member (for check_group_membership call)
|
|
|
|
mock_role_check = AsyncMock()
|
|
mock_role_check.scalar_one_or_none.return_value = UserRoleEnum.member # User's actual role
|
|
|
|
mock_db_session.execute.side_effect = [mock_membership_check, mock_role_check]
|
|
|
|
with pytest.raises(GroupPermissionError):
|
|
await check_user_role_in_group(mock_db_session, db_group_model.id, member_user_model.id, UserRoleEnum.owner)
|
|
|
|
# TODO: Add tests for DB operational/SQLAlchemy errors for each function similar to create_group_integrity_error
|
|
# TODO: Test edge cases like trying to add user to non-existent group (should be caught by FK constraints or prior checks) |