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 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, update_group_member_role # Assuming this will be added ) from app.schemas.group import GroupCreate, GroupUpdate # Added GroupUpdate 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, UserNotFoundError, # For adding user to group ConflictError # For updates ) # Fixtures @pytest.fixture def mock_db_session(): session = AsyncMock() mock_transaction_context = AsyncMock() session.begin = MagicMock(return_value=mock_transaction_context) session.commit = AsyncMock() session.rollback = AsyncMock() session.refresh = AsyncMock() session.add = MagicMock() session.delete = MagicMock() 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 group_update_data(): return GroupUpdate(name="Updated Test Group", version=1) @pytest.fixture def creator_user_model(): return UserModel(id=1, name="Creator User", email="creator@example.com", version=1) @pytest.fixture def member_user_model(): return UserModel(id=2, name="Member User", email="member@example.com", version=1) @pytest.fixture def non_member_user_model(): return UserModel(id=3, name="Non Member User", email="nonmember@example.com", version=1) @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, version=1) @pytest.fixture def db_user_group_owner_assoc(db_group_model, creator_user_model): return UserGroupModel(id=1, user_id=creator_user_model.id, group_id=db_group_model.id, role=UserRoleEnum.owner, user=creator_user_model, group=db_group_model, version=1) @pytest.fixture def db_user_group_member_assoc(db_group_model, member_user_model): return UserGroupModel(id=2, user_id=member_user_model.id, group_id=db_group_model.id, role=UserRoleEnum.member, user=member_user_model, group=db_group_model, version=1) # --- 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, attribute_names=None, with_for_update=None): if isinstance(instance, GroupModel): instance.id = 1 # Simulate ID assignment by DB instance.version = 1 # Simulate the UserGroup association being added and refreshed if done via relationship back_populates instance.members = [UserGroupModel(user_id=creator_user_model.id, group_id=instance.id, role=UserRoleEnum.owner, version=1)] elif isinstance(instance, UserGroupModel): instance.id = 1 # Simulate ID for UserGroupModel instance.version = 1 return None mock_db_session.refresh.side_effect = mock_refresh # Mock the user get for the creator mock_db_session.get.return_value = creator_user_model 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() assert mock_db_session.refresh.call_count >= 1 # Called for group, maybe for UserGroup too assert created_group is not None assert created_group.name == group_create_data.name assert created_group.created_by_id == creator_user_model.id assert len(created_group.members) == 1 assert created_group.members[0].role == UserRoleEnum.owner @pytest.mark.asyncio async def test_create_group_integrity_error(mock_db_session, group_create_data, creator_user_model): mock_db_session.get.return_value = creator_user_model # Creator user found 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() # --- get_user_groups Tests --- @pytest.mark.asyncio async def test_get_user_groups_success(mock_db_session, db_group_model, creator_user_model): # Mock the execute call that fetches groups for a user mock_result_groups = AsyncMock() mock_result_groups.scalars.return_value.all.return_value = [db_group_model] mock_db_session.execute.return_value = mock_result_groups 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_db_session.get.return_value = db_group_model 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 mock_db_session.get.assert_called_once_with(GroupModel, db_group_model.id, options=ANY) # options for eager loading @pytest.mark.asyncio async def test_get_group_by_id_not_found(mock_db_session): mock_db_session.get.return_value = None group = await get_group_by_id(mock_db_session, 999) assert group is None # --- is_user_member Tests --- from unittest.mock import ANY # For checking options in get @pytest.mark.asyncio async def test_is_user_member_true(mock_db_session, db_group_model, creator_user_model, db_user_group_owner_assoc): mock_result = AsyncMock() mock_result.scalar_one_or_none.return_value = db_user_group_owner_assoc.id 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, non_member_user_model): mock_result = AsyncMock() mock_result.scalar_one_or_none.return_value = None mock_db_session.execute.return_value = mock_result is_member = await is_user_member(mock_db_session, db_group_model.id, non_member_user_model.id) 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, non_member_user_model): # Mock is_user_member to return False initially with patch('app.crud.group.is_user_member', new_callable=AsyncMock) as mock_is_member: mock_is_member.return_value = False # Mock get for the user to be added mock_db_session.get.return_value = non_member_user_model async def mock_refresh_user_group(instance, attribute_names=None, with_for_update=None): instance.id = 100 instance.version = 1 return None mock_db_session.refresh.side_effect = mock_refresh_user_group user_group_assoc = await add_user_to_group(mock_db_session, db_group_model, non_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 == non_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): with patch('app.crud.group.is_user_member', new_callable=AsyncMock) as mock_is_member: mock_is_member.return_value = True # User is already a member # No need to mock session.get for the user if is_user_member is true first user_group_assoc = await add_user_to_group(mock_db_session, db_group_model, creator_user_model.id) assert user_group_assoc is None # Should return None if user already member mock_db_session.add.assert_not_called() @pytest.mark.asyncio async def test_add_user_to_group_user_not_found(mock_db_session, db_group_model): with patch('app.crud.group.is_user_member', new_callable=AsyncMock) as mock_is_member: mock_is_member.return_value = False # User not member initially mock_db_session.get.return_value = None # User to be added not found with pytest.raises(UserNotFoundError): await add_user_to_group(mock_db_session, db_group_model, 999, UserRoleEnum.member) 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, db_user_group_member_assoc): # Mock get_user_role_in_group to confirm user is not owner with patch('app.crud.group.get_user_role_in_group', new_callable=AsyncMock) as mock_get_role: mock_get_role.return_value = UserRoleEnum.member # Mock the execute call for the delete statement mock_delete_result = AsyncMock() mock_delete_result.rowcount = 1 # Simulate one row was affected/deleted mock_db_session.execute.return_value = mock_delete_result removed = await remove_user_from_group(mock_db_session, db_group_model, member_user_model.id) assert removed is True mock_db_session.execute.assert_called_once() # Check that the delete statement was indeed called, e.g., by checking the structure of the query passed to execute # This is a bit more involved if you want to match the exact SQLAlchemy delete object. # For now, assert_called_once() confirms it was called. @pytest.mark.asyncio async def test_remove_user_from_group_owner_last_member(mock_db_session, db_group_model, creator_user_model): with patch('app.crud.group.get_user_role_in_group', new_callable=AsyncMock) as mock_get_role, \ patch('app.crud.group.get_group_member_count', new_callable=AsyncMock) as mock_member_count: mock_get_role.return_value = UserRoleEnum.owner mock_member_count.return_value = 1 # This user is the last member with pytest.raises(GroupOperationError, match="Cannot remove the sole owner of a group. Delete the group instead."): await remove_user_from_group(mock_db_session, db_group_model, creator_user_model.id) mock_db_session.execute.assert_not_called() # Delete should not be called @pytest.mark.asyncio async def test_remove_user_from_group_not_member(mock_db_session, db_group_model, non_member_user_model): # Mock get_user_role_in_group to return None, indicating not a member or role not found (effectively not a member for removal purposes) with patch('app.crud.group.get_user_role_in_group', new_callable=AsyncMock) as mock_get_role: mock_get_role.return_value = None # For this specific test, we might not even need to mock `execute` if `get_user_role_in_group` returning None # already causes the function to exit or raise an error handled by `GroupMembershipError`. # However, if the function proceeds to attempt a delete that affects 0 rows, then `rowcount = 0` is the correct mock. mock_delete_result = AsyncMock() mock_delete_result.rowcount = 0 mock_db_session.execute.return_value = mock_delete_result with pytest.raises(GroupMembershipError, match="User is not a member of the group or cannot be removed."): await remove_user_from_group(mock_db_session, db_group_model, non_member_user_model.id) # Depending on the implementation: execute might be called or not. # If there's a check before executing delete, it might not be called. # If it tries to delete and finds nothing, it would be called. # For now, let's assume it could be called. If your function logic prevents it, adjust this. # mock_db_session.execute.assert_called_once() <--- This might fail if not called # --- get_group_member_count Tests --- @pytest.mark.asyncio async def test_get_group_member_count_success(mock_db_session, db_group_model): mock_result_count = AsyncMock() mock_result_count.scalar_one.return_value = 5 # Example count mock_db_session.execute.return_value = mock_result_count 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 get_group_by_id with patch('app.crud.group.get_group_by_id', new_callable=AsyncMock) as mock_get_group, \ patch('app.crud.group.is_user_member', new_callable=AsyncMock) as mock_is_member: mock_get_group.return_value = db_group_model mock_is_member.return_value = True group = await check_group_membership(mock_db_session, db_group_model.id, creator_user_model.id) assert group is db_group_model @pytest.mark.asyncio async def test_check_group_membership_group_not_found(mock_db_session, creator_user_model): with patch('app.crud.group.get_group_by_id', new_callable=AsyncMock) as mock_get_group: mock_get_group.return_value = None 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, non_member_user_model): with patch('app.crud.group.get_group_by_id', new_callable=AsyncMock) as mock_get_group, \ patch('app.crud.group.is_user_member', new_callable=AsyncMock) as mock_is_member: mock_get_group.return_value = db_group_model mock_is_member.return_value = False with pytest.raises(GroupMembershipError, match="User is not a member of the specified group"): await check_group_membership(mock_db_session, db_group_model.id, non_member_user_model.id) # --- check_user_role_in_group (standalone check, not just membership) --- @pytest.mark.asyncio async def test_check_user_role_in_group_sufficient_role(mock_db_session, db_group_model, creator_user_model): # This test assumes check_group_membership is called internally first, or similar logic applies with patch('app.crud.group.check_group_membership', new_callable=AsyncMock) as mock_check_membership, \ patch('app.crud.group.get_user_role_in_group', new_callable=AsyncMock) as mock_get_role: mock_check_membership.return_value = db_group_model # Group exists and user is member mock_get_role.return_value = UserRoleEnum.owner # Check if owner has owner role (should pass) await check_user_role_in_group(mock_db_session, db_group_model.id, creator_user_model.id, UserRoleEnum.owner) # Check if owner has member role (should pass, as owner is implicitly a member with higher privileges) await check_user_role_in_group(mock_db_session, db_group_model.id, creator_user_model.id, UserRoleEnum.member) @pytest.mark.asyncio async def test_check_user_role_in_group_insufficient_role(mock_db_session, db_group_model, member_user_model): with patch('app.crud.group.check_group_membership', new_callable=AsyncMock) as mock_check_membership, \ patch('app.crud.group.get_user_role_in_group', new_callable=AsyncMock) as mock_get_role: mock_check_membership.return_value = db_group_model mock_get_role.return_value = UserRoleEnum.member with pytest.raises(GroupPermissionError, match="User does not have the required role in the group."): await check_user_role_in_group(mock_db_session, db_group_model.id, member_user_model.id, UserRoleEnum.owner) # Future test ideas, to be moved to a proper test planning tool or issue tracker. # Consider these during major refactors or when expanding test coverage. # Example of a DB operational error test (can be adapted for other functions) # @pytest.mark.asyncio # async def test_create_group_operational_error(mock_db_session, group_create_data, creator_user_model): # mock_db_session.get.return_value = creator_user_model # mock_db_session.flush.side_effect = OperationalError("mock operational error", "params", "orig") # with pytest.raises(DatabaseConnectionError): # Assuming OperationalError maps to this # await create_group(mock_db_session, group_create_data, creator_user_model.id) # mock_db_session.rollback.assert_called_once()