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)