diff --git a/backend/tests/api/test_admin_error_handlers.py b/backend/tests/api/test_admin_error_handlers.py new file mode 100644 index 0000000..f298c27 --- /dev/null +++ b/backend/tests/api/test_admin_error_handlers.py @@ -0,0 +1,546 @@ +# tests/api/test_admin_error_handlers.py +""" +Tests for admin route exception handlers and error paths. +Focus on code coverage of error handling branches. +""" +import pytest +import pytest_asyncio +from unittest.mock import patch +from fastapi import status +from uuid import uuid4 + + +@pytest_asyncio.fixture +async def superuser_token(client, async_test_superuser): + """Get access token for superuser.""" + response = await client.post( + "/api/v1/auth/login", + json={ + "email": "superuser@example.com", + "password": "SuperPassword123!" + } + ) + assert response.status_code == 200 + return response.json()["access_token"] + + +# ===== USER MANAGEMENT ERROR TESTS ===== + +class TestAdminListUsersFilters: + """Test admin list users with various filters.""" + + @pytest.mark.asyncio + async def test_list_users_with_is_superuser_filter(self, client, superuser_token): + """Test listing users with is_superuser filter (covers line 96).""" + response = await client.get( + "/api/v1/admin/users?is_superuser=true", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert "data" in data + + @pytest.mark.asyncio + async def test_list_users_database_error_propagates(self, client, superuser_token): + """Test that database errors propagate correctly (covers line 118-120).""" + with patch('app.api.routes.admin.user_crud.get_multi_with_total', side_effect=Exception("DB error")): + with pytest.raises(Exception): + await client.get( + "/api/v1/admin/users", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + +class TestAdminCreateUserErrors: + """Test admin create user error handling.""" + + @pytest.mark.asyncio + async def test_create_user_duplicate_email(self, client, async_test_user, superuser_token): + """Test creating user with duplicate email (covers line 145-150).""" + response = await client.post( + "/api/v1/admin/users", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={ + "email": async_test_user.email, + "password": "NewPassword123!", + "first_name": "Duplicate", + "last_name": "User" + } + ) + + # Should get error for duplicate email + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_create_user_unexpected_error_propagates(self, client, superuser_token): + """Test unexpected errors during user creation (covers line 151-153).""" + with patch('app.api.routes.admin.user_crud.create', side_effect=RuntimeError("Unexpected error")): + with pytest.raises(RuntimeError): + await client.post( + "/api/v1/admin/users", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={ + "email": "newerror@example.com", + "password": "NewPassword123!", + "first_name": "New", + "last_name": "User" + } + ) + + +class TestAdminGetUserErrors: + """Test admin get user error handling.""" + + @pytest.mark.asyncio + async def test_get_nonexistent_user(self, client, superuser_token): + """Test getting a user that doesn't exist (covers line 170-175).""" + fake_id = uuid4() + response = await client.get( + f"/api/v1/admin/users/{fake_id}", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + +class TestAdminUpdateUserErrors: + """Test admin update user error handling.""" + + @pytest.mark.asyncio + async def test_update_nonexistent_user(self, client, superuser_token): + """Test updating a user that doesn't exist (covers line 194-198).""" + fake_id = uuid4() + response = await client.put( + f"/api/v1/admin/users/{fake_id}", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={"first_name": "Updated"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_update_user_unexpected_error(self, client, async_test_user, superuser_token): + """Test unexpected errors during user update (covers line 206-208).""" + with patch('app.api.routes.admin.user_crud.update', side_effect=RuntimeError("Update failed")): + with pytest.raises(RuntimeError): + await client.put( + f"/api/v1/admin/users/{async_test_user.id}", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={"first_name": "Updated"} + ) + + +class TestAdminDeleteUserErrors: + """Test admin delete user error handling.""" + + @pytest.mark.asyncio + async def test_delete_nonexistent_user(self, client, superuser_token): + """Test deleting a user that doesn't exist (covers line 226-230).""" + fake_id = uuid4() + response = await client.delete( + f"/api/v1/admin/users/{fake_id}", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_delete_user_unexpected_error(self, client, async_test_user, superuser_token): + """Test unexpected errors during user deletion (covers line 238-240).""" + with patch('app.api.routes.admin.user_crud.soft_delete', side_effect=Exception("Delete failed")): + with pytest.raises(Exception): + await client.delete( + f"/api/v1/admin/users/{async_test_user.id}", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + +class TestAdminActivateUserErrors: + """Test admin activate user error handling.""" + + @pytest.mark.asyncio + async def test_activate_nonexistent_user(self, client, superuser_token): + """Test activating a user that doesn't exist (covers line 270-274).""" + fake_id = uuid4() + response = await client.post( + f"/api/v1/admin/users/{fake_id}/activate", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_activate_user_unexpected_error(self, client, async_test_user, superuser_token): + """Test unexpected errors during user activation (covers line 282-284).""" + with patch('app.api.routes.admin.user_crud.update', side_effect=Exception("Activation failed")): + with pytest.raises(Exception): + await client.post( + f"/api/v1/admin/users/{async_test_user.id}/activate", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + +class TestAdminDeactivateUserErrors: + """Test admin deactivate user error handling.""" + + @pytest.mark.asyncio + async def test_deactivate_nonexistent_user(self, client, superuser_token): + """Test deactivating a user that doesn't exist (covers line 306-310).""" + fake_id = uuid4() + response = await client.post( + f"/api/v1/admin/users/{fake_id}/deactivate", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_deactivate_self_forbidden(self, client, async_test_superuser, superuser_token): + """Test that admin cannot deactivate themselves (covers line 319-323).""" + response = await client.post( + f"/api/v1/admin/users/{async_test_superuser.id}/deactivate", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_403_FORBIDDEN + + @pytest.mark.asyncio + async def test_deactivate_user_unexpected_error(self, client, async_test_user, superuser_token): + """Test unexpected errors during user deactivation (covers line 326-328).""" + with patch('app.api.routes.admin.user_crud.update', side_effect=Exception("Deactivation failed")): + with pytest.raises(Exception): + await client.post( + f"/api/v1/admin/users/{async_test_user.id}/deactivate", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + +# ===== ORGANIZATION MANAGEMENT ERROR TESTS ===== + +class TestAdminListOrganizationsErrors: + """Test admin list organizations error handling.""" + + @pytest.mark.asyncio + async def test_list_organizations_database_error(self, client, superuser_token): + """Test list organizations with database error (covers line 427-456).""" + with patch('app.api.routes.admin.organization_crud.get_multi_with_member_counts', side_effect=Exception("DB error")): + with pytest.raises(Exception): + await client.get( + "/api/v1/admin/organizations", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + +class TestAdminCreateOrganizationErrors: + """Test admin create organization error handling.""" + + @pytest.mark.asyncio + async def test_create_organization_duplicate_slug(self, client, async_test_db, superuser_token): + """Test creating organization with duplicate slug (covers line 480-483).""" + test_engine, AsyncTestingSessionLocal = async_test_db + + # Create an organization first + async with AsyncTestingSessionLocal() as session: + from app.models.organization import Organization + org = Organization( + name="Existing Org", + slug="existing-org", + description="Test org" + ) + session.add(org) + await session.commit() + + # Try to create another with same slug + response = await client.post( + "/api/v1/admin/organizations", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={ + "name": "New Org", + "slug": "existing-org", + "description": "Duplicate slug" + } + ) + + # Should get error for duplicate slug + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_create_organization_unexpected_error(self, client, superuser_token): + """Test unexpected errors during organization creation (covers line 484-485).""" + with patch('app.api.routes.admin.organization_crud.create', side_effect=RuntimeError("Creation failed")): + with pytest.raises(RuntimeError): + await client.post( + "/api/v1/admin/organizations", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={ + "name": "New Org", + "slug": "new-org", + "description": "Test" + } + ) + + +class TestAdminGetOrganizationErrors: + """Test admin get organization error handling.""" + + @pytest.mark.asyncio + async def test_get_nonexistent_organization(self, client, superuser_token): + """Test getting an organization that doesn't exist (covers line 516-520).""" + fake_id = uuid4() + response = await client.get( + f"/api/v1/admin/organizations/{fake_id}", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + +class TestAdminUpdateOrganizationErrors: + """Test admin update organization error handling.""" + + @pytest.mark.asyncio + async def test_update_nonexistent_organization(self, client, superuser_token): + """Test updating an organization that doesn't exist (covers line 552-556).""" + fake_id = uuid4() + response = await client.put( + f"/api/v1/admin/organizations/{fake_id}", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={"name": "Updated Org"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_update_organization_unexpected_error(self, client, async_test_db, superuser_token): + """Test unexpected errors during organization update (covers line 573-575).""" + test_engine, AsyncTestingSessionLocal = async_test_db + + # Create an organization + async with AsyncTestingSessionLocal() as session: + from app.models.organization import Organization + org = Organization( + name="Test Org", + slug="test-org-update-error", + description="Test" + ) + session.add(org) + await session.commit() + await session.refresh(org) + org_id = org.id + + with patch('app.api.routes.admin.organization_crud.update', side_effect=Exception("Update failed")): + with pytest.raises(Exception): + await client.put( + f"/api/v1/admin/organizations/{org_id}", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={"name": "Updated"} + ) + + +class TestAdminDeleteOrganizationErrors: + """Test admin delete organization error handling.""" + + @pytest.mark.asyncio + async def test_delete_nonexistent_organization(self, client, superuser_token): + """Test deleting an organization that doesn't exist (covers line 596-600).""" + fake_id = uuid4() + response = await client.delete( + f"/api/v1/admin/organizations/{fake_id}", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_delete_organization_unexpected_error(self, client, async_test_db, superuser_token): + """Test unexpected errors during organization deletion (covers line 611-613).""" + test_engine, AsyncTestingSessionLocal = async_test_db + + # Create organization + async with AsyncTestingSessionLocal() as session: + from app.models.organization import Organization + org = Organization( + name="Error Org", + slug="error-org-delete", + description="Test" + ) + session.add(org) + await session.commit() + await session.refresh(org) + org_id = org.id + + with patch('app.api.routes.admin.organization_crud.remove', side_effect=Exception("Delete failed")): + with pytest.raises(Exception): + await client.delete( + f"/api/v1/admin/organizations/{org_id}", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + +class TestAdminListOrganizationMembersErrors: + """Test admin list organization members error handling.""" + + @pytest.mark.asyncio + async def test_list_members_nonexistent_organization(self, client, superuser_token): + """Test listing members of non-existent organization (covers line 634-638).""" + fake_id = uuid4() + response = await client.get( + f"/api/v1/admin/organizations/{fake_id}/members", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_list_members_database_error(self, client, async_test_db, superuser_token): + """Test database errors during member listing (covers line 660-662).""" + test_engine, AsyncTestingSessionLocal = async_test_db + + # Create organization + async with AsyncTestingSessionLocal() as session: + from app.models.organization import Organization + org = Organization( + name="Members Error Org", + slug="members-error-org", + description="Test" + ) + session.add(org) + await session.commit() + await session.refresh(org) + org_id = org.id + + with patch('app.api.routes.admin.organization_crud.get_organization_members', side_effect=Exception("DB error")): + with pytest.raises(Exception): + await client.get( + f"/api/v1/admin/organizations/{org_id}/members", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + +class TestAdminAddOrganizationMemberErrors: + """Test admin add organization member error handling.""" + + @pytest.mark.asyncio + async def test_add_member_nonexistent_organization(self, client, async_test_user, superuser_token): + """Test adding member to non-existent organization (covers line 689-693).""" + fake_id = uuid4() + response = await client.post( + f"/api/v1/admin/organizations/{fake_id}/members", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={ + "user_id": str(async_test_user.id), + "role": "member" + } + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_add_nonexistent_user_to_organization(self, client, async_test_db, superuser_token): + """Test adding non-existent user to organization (covers line 696-700).""" + test_engine, AsyncTestingSessionLocal = async_test_db + + # Create organization + async with AsyncTestingSessionLocal() as session: + from app.models.organization import Organization + org = Organization( + name="Add Member Org", + slug="add-member-org", + description="Test" + ) + session.add(org) + await session.commit() + await session.refresh(org) + org_id = org.id + + fake_user_id = uuid4() + response = await client.post( + f"/api/v1/admin/organizations/{org_id}/members", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={ + "user_id": str(fake_user_id), + "role": "member" + } + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_add_member_unexpected_error(self, client, async_test_db, async_test_user, superuser_token): + """Test unexpected errors during member addition (covers line 727-729).""" + test_engine, AsyncTestingSessionLocal = async_test_db + + # Create organization + async with AsyncTestingSessionLocal() as session: + from app.models.organization import Organization + org = Organization( + name="Error Add Org", + slug="error-add-org", + description="Test" + ) + session.add(org) + await session.commit() + await session.refresh(org) + org_id = org.id + + with patch('app.api.routes.admin.organization_crud.add_user', side_effect=Exception("Add failed")): + with pytest.raises(Exception): + await client.post( + f"/api/v1/admin/organizations/{org_id}/members", + headers={"Authorization": f"Bearer {superuser_token}"}, + json={ + "user_id": str(async_test_user.id), + "role": "member" + } + ) + + +class TestAdminRemoveOrganizationMemberErrors: + """Test admin remove organization member error handling.""" + + @pytest.mark.asyncio + async def test_remove_member_nonexistent_organization(self, client, async_test_user, superuser_token): + """Test removing member from non-existent organization (covers line 750-754).""" + fake_id = uuid4() + response = await client.delete( + f"/api/v1/admin/organizations/{fake_id}/members/{async_test_user.id}", + headers={"Authorization": f"Bearer {superuser_token}"} + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.asyncio + async def test_remove_member_unexpected_error(self, client, async_test_db, async_test_user, superuser_token): + """Test unexpected errors during member removal (covers line 780-782).""" + test_engine, AsyncTestingSessionLocal = async_test_db + + # Create organization with member + async with AsyncTestingSessionLocal() as session: + from app.models.organization import Organization + from app.models.user_organization import UserOrganization, OrganizationRole + + org = Organization( + name="Remove Member Org", + slug="remove-member-org", + description="Test" + ) + session.add(org) + await session.commit() + await session.refresh(org) + + member = UserOrganization( + user_id=async_test_user.id, + organization_id=org.id, + role=OrganizationRole.MEMBER + ) + session.add(member) + await session.commit() + org_id = org.id + + with patch('app.api.routes.admin.organization_crud.remove_user', side_effect=Exception("Remove failed")): + with pytest.raises(Exception): + await client.delete( + f"/api/v1/admin/organizations/{org_id}/members/{async_test_user.id}", + headers={"Authorization": f"Bearer {superuser_token}"} + ) diff --git a/backend/tests/api/test_auth_error_handlers.py b/backend/tests/api/test_auth_error_handlers.py new file mode 100644 index 0000000..80a0f5f --- /dev/null +++ b/backend/tests/api/test_auth_error_handlers.py @@ -0,0 +1,216 @@ +# tests/api/test_auth_error_handlers.py +""" +Tests for auth route exception handlers and error paths. +""" +import pytest +from unittest.mock import patch, AsyncMock +from fastapi import status + + +class TestLoginSessionCreationFailure: + """Test login when session creation fails.""" + + @pytest.mark.asyncio + async def test_login_succeeds_despite_session_creation_failure(self, client, async_test_user): + """Test that login succeeds even if session creation fails.""" + # Mock session creation to fail + with patch('app.api.routes.auth.session_crud.create_session', side_effect=Exception("Session creation failed")): + response = await client.post( + "/api/v1/auth/login", + json={ + "email": "testuser@example.com", + "password": "TestPassword123!" + } + ) + + # Login should still succeed, just without session record + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert "access_token" in data + assert "refresh_token" in data + + +class TestOAuthLoginSessionCreationFailure: + """Test OAuth login when session creation fails.""" + + @pytest.mark.asyncio + async def test_oauth_login_succeeds_despite_session_failure(self, client, async_test_user): + """Test OAuth login succeeds even if session creation fails.""" + with patch('app.api.routes.auth.session_crud.create_session', side_effect=Exception("Session failed")): + response = await client.post( + "/api/v1/auth/login/oauth", + data={ + "username": "testuser@example.com", + "password": "TestPassword123!" + } + ) + + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert "access_token" in data + + +class TestRefreshTokenSessionUpdateFailure: + """Test refresh token when session update fails.""" + + @pytest.mark.asyncio + async def test_refresh_token_succeeds_despite_session_update_failure(self, client, async_test_user): + """Test that token refresh succeeds even if session update fails.""" + # First login to get tokens + response = await client.post( + "/api/v1/auth/login", + json={ + "email": "testuser@example.com", + "password": "TestPassword123!" + } + ) + tokens = response.json() + + # Mock session update to fail + with patch('app.api.routes.auth.session_crud.update_refresh_token', side_effect=Exception("Update failed")): + response = await client.post( + "/api/v1/auth/refresh", + json={"refresh_token": tokens["refresh_token"]} + ) + + # Should still succeed - tokens are issued before update + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert "access_token" in data + + +class TestLogoutWithExpiredToken: + """Test logout with expired/invalid token.""" + + @pytest.mark.asyncio + async def test_logout_with_invalid_token_still_succeeds(self, client, async_test_user): + """Test logout succeeds even with invalid refresh token.""" + # Login first + response = await client.post( + "/api/v1/auth/login", + json={ + "email": "testuser@example.com", + "password": "TestPassword123!" + } + ) + access_token = response.json()["access_token"] + + # Try logout with invalid refresh token + response = await client.post( + "/api/v1/auth/logout", + headers={"Authorization": f"Bearer {access_token}"}, + json={"refresh_token": "invalid.token.here"} + ) + + # Should succeed (idempotent) + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert data["success"] is True + + +class TestLogoutWithNonExistentSession: + """Test logout when session doesn't exist.""" + + @pytest.mark.asyncio + async def test_logout_with_no_session_succeeds(self, client, async_test_user): + """Test logout succeeds even if session not found.""" + response = await client.post( + "/api/v1/auth/login", + json={ + "email": "testuser@example.com", + "password": "TestPassword123!" + } + ) + tokens = response.json() + + # Mock session lookup to return None + with patch('app.api.routes.auth.session_crud.get_by_jti', return_value=None): + response = await client.post( + "/api/v1/auth/logout", + headers={"Authorization": f"Bearer {tokens['access_token']}"}, + json={"refresh_token": tokens["refresh_token"]} + ) + + # Should succeed (idempotent) + assert response.status_code == status.HTTP_200_OK + + +class TestLogoutUnexpectedError: + """Test logout with unexpected errors.""" + + @pytest.mark.asyncio + async def test_logout_with_unexpected_error_returns_success(self, client, async_test_user): + """Test logout returns success even on unexpected errors.""" + response = await client.post( + "/api/v1/auth/login", + json={ + "email": "testuser@example.com", + "password": "TestPassword123!" + } + ) + tokens = response.json() + + # Mock to raise unexpected error + with patch('app.api.routes.auth.session_crud.get_by_jti', side_effect=Exception("Unexpected error")): + response = await client.post( + "/api/v1/auth/logout", + headers={"Authorization": f"Bearer {tokens['access_token']}"}, + json={"refresh_token": tokens["refresh_token"]} + ) + + # Should still return success (don't expose errors) + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert data["success"] is True + + +class TestLogoutAllUnexpectedError: + """Test logout-all with unexpected errors.""" + + @pytest.mark.asyncio + async def test_logout_all_database_error(self, client, async_test_user): + """Test logout-all handles database errors.""" + response = await client.post( + "/api/v1/auth/login", + json={ + "email": "testuser@example.com", + "password": "TestPassword123!" + } + ) + access_token = response.json()["access_token"] + + # Mock to raise database error + with patch('app.api.routes.auth.session_crud.deactivate_all_user_sessions', side_effect=Exception("DB error")): + response = await client.post( + "/api/v1/auth/logout-all", + headers={"Authorization": f"Bearer {access_token}"} + ) + + assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR + + +class TestPasswordResetConfirmSessionInvalidation: + """Test password reset invalidates sessions.""" + + @pytest.mark.asyncio + async def test_password_reset_continues_despite_session_invalidation_failure(self, client, async_test_user): + """Test password reset succeeds even if session invalidation fails.""" + # Create a valid password reset token + from app.utils.security import create_password_reset_token + + token = create_password_reset_token(async_test_user.email) + + # Mock session invalidation to fail + with patch('app.api.routes.auth.session_crud.deactivate_all_user_sessions', side_effect=Exception("Invalidation failed")): + response = await client.post( + "/api/v1/auth/password-reset/confirm", + json={ + "token": token, + "new_password": "NewPassword123!" + } + ) + + # Should still succeed - password was reset + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert data["success"] is True diff --git a/backend/tests/crud/test_base.py b/backend/tests/crud/test_base.py index 802e2d6..b3425a5 100644 --- a/backend/tests/crud/test_base.py +++ b/backend/tests/crud/test_base.py @@ -757,3 +757,79 @@ class TestCRUDBaseRestore: restored = await user_crud.restore(session, id=user_id) # UUID object assert restored is not None assert restored.deleted_at is None + + +class TestCRUDBasePaginationValidation: + """Tests for pagination parameter validation (covers lines 254-260).""" + + @pytest.mark.asyncio + async def test_get_multi_with_total_negative_skip(self, async_test_db): + """Test that negative skip raises ValueError.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + with pytest.raises(ValueError, match="skip must be non-negative"): + await user_crud.get_multi_with_total(session, skip=-1, limit=10) + + @pytest.mark.asyncio + async def test_get_multi_with_total_negative_limit(self, async_test_db): + """Test that negative limit raises ValueError.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + with pytest.raises(ValueError, match="limit must be non-negative"): + await user_crud.get_multi_with_total(session, skip=0, limit=-1) + + @pytest.mark.asyncio + async def test_get_multi_with_total_limit_too_large(self, async_test_db): + """Test that limit > 1000 raises ValueError.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + with pytest.raises(ValueError, match="Maximum limit is 1000"): + await user_crud.get_multi_with_total(session, skip=0, limit=1001) + + @pytest.mark.asyncio + async def test_get_multi_with_total_with_filters(self, async_test_db, async_test_user): + """Test pagination with filters (covers lines 270-273).""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + users, total = await user_crud.get_multi_with_total( + session, + skip=0, + limit=10, + filters={"is_active": True} + ) + assert isinstance(users, list) + assert total >= 0 + + @pytest.mark.asyncio + async def test_get_multi_with_total_with_sorting_desc(self, async_test_db): + """Test pagination with descending sort (covers lines 283-284).""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + users, total = await user_crud.get_multi_with_total( + session, + skip=0, + limit=10, + sort_by="created_at", + sort_order="desc" + ) + assert isinstance(users, list) + + @pytest.mark.asyncio + async def test_get_multi_with_total_with_sorting_asc(self, async_test_db): + """Test pagination with ascending sort (covers lines 285-286).""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + users, total = await user_crud.get_multi_with_total( + session, + skip=0, + limit=10, + sort_by="created_at", + sort_order="asc" + ) + assert isinstance(users, list) diff --git a/backend/tests/crud/test_base_db_failures.py b/backend/tests/crud/test_base_db_failures.py new file mode 100644 index 0000000..e93bc41 --- /dev/null +++ b/backend/tests/crud/test_base_db_failures.py @@ -0,0 +1,293 @@ +# tests/crud/test_base_db_failures.py +""" +Comprehensive tests for base CRUD database failure scenarios. +Tests exception handling, rollbacks, and error messages. +""" +import pytest +from unittest.mock import AsyncMock, MagicMock, patch +from sqlalchemy.exc import IntegrityError, OperationalError, DataError +from uuid import uuid4 + +from app.crud.user import user as user_crud +from app.schemas.users import UserCreate, UserUpdate + + +class TestBaseCRUDCreateFailures: + """Test base CRUD create method exception handling.""" + + @pytest.mark.asyncio + async def test_create_operational_error_triggers_rollback(self, async_test_db): + """Test that OperationalError triggers rollback (User CRUD catches as Exception).""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise OperationalError("Connection lost", {}, Exception("DB connection failed")) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + user_data = UserCreate( + email="operror@example.com", + password="TestPassword123!", + first_name="Test", + last_name="User" + ) + + # User CRUD catches this as generic Exception and re-raises + with pytest.raises(OperationalError): + await user_crud.create(session, obj_in=user_data) + + # Verify rollback was called + mock_rollback.assert_called_once() + + @pytest.mark.asyncio + async def test_create_data_error_triggers_rollback(self, async_test_db): + """Test that DataError triggers rollback (User CRUD catches as Exception).""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise DataError("Invalid data type", {}, Exception("Data overflow")) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + user_data = UserCreate( + email="dataerror@example.com", + password="TestPassword123!", + first_name="Test", + last_name="User" + ) + + # User CRUD catches this as generic Exception and re-raises + with pytest.raises(DataError): + await user_crud.create(session, obj_in=user_data) + + mock_rollback.assert_called_once() + + @pytest.mark.asyncio + async def test_create_unexpected_exception_triggers_rollback(self, async_test_db): + """Test that unexpected exceptions trigger rollback and re-raise.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise RuntimeError("Unexpected database error") + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + user_data = UserCreate( + email="unexpected@example.com", + password="TestPassword123!", + first_name="Test", + last_name="User" + ) + + with pytest.raises(RuntimeError, match="Unexpected database error"): + await user_crud.create(session, obj_in=user_data) + + mock_rollback.assert_called_once() + + +class TestBaseCRUDUpdateFailures: + """Test base CRUD update method exception handling.""" + + @pytest.mark.asyncio + async def test_update_operational_error(self, async_test_db, async_test_user): + """Test update with OperationalError.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + user = await user_crud.get(session, id=str(async_test_user.id)) + + async def mock_commit(): + raise OperationalError("Connection timeout", {}, Exception("Timeout")) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(ValueError, match="Database operation failed"): + await user_crud.update(session, db_obj=user, obj_in={"first_name": "Updated"}) + + mock_rollback.assert_called_once() + + @pytest.mark.asyncio + async def test_update_data_error(self, async_test_db, async_test_user): + """Test update with DataError.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + user = await user_crud.get(session, id=str(async_test_user.id)) + + async def mock_commit(): + raise DataError("Invalid data", {}, Exception("Data type mismatch")) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(ValueError, match="Database operation failed"): + await user_crud.update(session, db_obj=user, obj_in={"first_name": "Updated"}) + + mock_rollback.assert_called_once() + + @pytest.mark.asyncio + async def test_update_unexpected_error(self, async_test_db, async_test_user): + """Test update with unexpected error.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + user = await user_crud.get(session, id=str(async_test_user.id)) + + async def mock_commit(): + raise KeyError("Unexpected error") + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(KeyError): + await user_crud.update(session, db_obj=user, obj_in={"first_name": "Updated"}) + + mock_rollback.assert_called_once() + + +class TestBaseCRUDRemoveFailures: + """Test base CRUD remove method exception handling.""" + + @pytest.mark.asyncio + async def test_remove_unexpected_error_triggers_rollback(self, async_test_db, async_test_user): + """Test that unexpected errors in remove trigger rollback.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise RuntimeError("Database write failed") + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(RuntimeError, match="Database write failed"): + await user_crud.remove(session, id=str(async_test_user.id)) + + mock_rollback.assert_called_once() + + +class TestBaseCRUDGetMultiWithTotalFailures: + """Test get_multi_with_total exception handling.""" + + @pytest.mark.asyncio + async def test_get_multi_with_total_database_error(self, async_test_db): + """Test get_multi_with_total handles database errors.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + # Mock execute to raise an error + original_execute = session.execute + + async def mock_execute(*args, **kwargs): + raise OperationalError("Query failed", {}, Exception("Database error")) + + with patch.object(session, 'execute', side_effect=mock_execute): + with pytest.raises(OperationalError): + await user_crud.get_multi_with_total(session, skip=0, limit=10) + + +class TestBaseCRUDCountFailures: + """Test count method exception handling.""" + + @pytest.mark.asyncio + async def test_count_database_error_propagates(self, async_test_db): + """Test count propagates database errors.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_execute(*args, **kwargs): + raise OperationalError("Count failed", {}, Exception("DB error")) + + with patch.object(session, 'execute', side_effect=mock_execute): + with pytest.raises(OperationalError): + await user_crud.count(session) + + +class TestBaseCRUDSoftDeleteFailures: + """Test soft_delete method exception handling.""" + + @pytest.mark.asyncio + async def test_soft_delete_unexpected_error_triggers_rollback(self, async_test_db, async_test_user): + """Test soft_delete handles unexpected errors with rollback.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise RuntimeError("Soft delete failed") + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(RuntimeError, match="Soft delete failed"): + await user_crud.soft_delete(session, id=str(async_test_user.id)) + + mock_rollback.assert_called_once() + + +class TestBaseCRUDRestoreFailures: + """Test restore method exception handling.""" + + @pytest.mark.asyncio + async def test_restore_unexpected_error_triggers_rollback(self, async_test_db): + """Test restore handles unexpected errors with rollback.""" + test_engine, SessionLocal = async_test_db + + # First create and soft delete a user + async with SessionLocal() as session: + user_data = UserCreate( + email="restore_test@example.com", + password="TestPassword123!", + first_name="Restore", + last_name="Test" + ) + user = await user_crud.create(session, obj_in=user_data) + user_id = user.id + await session.commit() + + async with SessionLocal() as session: + await user_crud.soft_delete(session, id=str(user_id)) + + # Now test restore failure + async with SessionLocal() as session: + async def mock_commit(): + raise RuntimeError("Restore failed") + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(RuntimeError, match="Restore failed"): + await user_crud.restore(session, id=str(user_id)) + + mock_rollback.assert_called_once() + + +class TestBaseCRUDGetFailures: + """Test get method exception handling.""" + + @pytest.mark.asyncio + async def test_get_database_error_propagates(self, async_test_db): + """Test get propagates database errors.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_execute(*args, **kwargs): + raise OperationalError("Get failed", {}, Exception("DB error")) + + with patch.object(session, 'execute', side_effect=mock_execute): + with pytest.raises(OperationalError): + await user_crud.get(session, id=str(uuid4())) + + +class TestBaseCRUDGetMultiFailures: + """Test get_multi method exception handling.""" + + @pytest.mark.asyncio + async def test_get_multi_database_error_propagates(self, async_test_db): + """Test get_multi propagates database errors.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_execute(*args, **kwargs): + raise OperationalError("Query failed", {}, Exception("DB error")) + + with patch.object(session, 'execute', side_effect=mock_execute): + with pytest.raises(OperationalError): + await user_crud.get_multi(session, skip=0, limit=10) diff --git a/backend/tests/crud/test_session_db_failures.py b/backend/tests/crud/test_session_db_failures.py new file mode 100644 index 0000000..e7dd5d2 --- /dev/null +++ b/backend/tests/crud/test_session_db_failures.py @@ -0,0 +1,336 @@ +# tests/crud/test_session_db_failures.py +""" +Comprehensive tests for session CRUD database failure scenarios. +""" +import pytest +from unittest.mock import AsyncMock, patch +from sqlalchemy.exc import OperationalError, IntegrityError +from datetime import datetime, timedelta, timezone +from uuid import uuid4 + +from app.crud.session import session as session_crud +from app.models.user_session import UserSession +from app.schemas.sessions import SessionCreate + + +class TestSessionCRUDGetByJtiFailures: + """Test get_by_jti exception handling.""" + + @pytest.mark.asyncio + async def test_get_by_jti_database_error(self, async_test_db): + """Test get_by_jti handles database errors.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_execute(*args, **kwargs): + raise OperationalError("DB connection lost", {}, Exception()) + + with patch.object(session, 'execute', side_effect=mock_execute): + with pytest.raises(OperationalError): + await session_crud.get_by_jti(session, jti="test_jti") + + +class TestSessionCRUDGetActiveByJtiFailures: + """Test get_active_by_jti exception handling.""" + + @pytest.mark.asyncio + async def test_get_active_by_jti_database_error(self, async_test_db): + """Test get_active_by_jti handles database errors.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_execute(*args, **kwargs): + raise OperationalError("Query timeout", {}, Exception()) + + with patch.object(session, 'execute', side_effect=mock_execute): + with pytest.raises(OperationalError): + await session_crud.get_active_by_jti(session, jti="test_jti") + + +class TestSessionCRUDGetUserSessionsFailures: + """Test get_user_sessions exception handling.""" + + @pytest.mark.asyncio + async def test_get_user_sessions_database_error(self, async_test_db, async_test_user): + """Test get_user_sessions handles database errors.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_execute(*args, **kwargs): + raise OperationalError("Database error", {}, Exception()) + + with patch.object(session, 'execute', side_effect=mock_execute): + with pytest.raises(OperationalError): + await session_crud.get_user_sessions( + session, + user_id=str(async_test_user.id) + ) + + +class TestSessionCRUDCreateSessionFailures: + """Test create_session exception handling.""" + + @pytest.mark.asyncio + async def test_create_session_commit_failure_triggers_rollback(self, async_test_db, async_test_user): + """Test create_session handles commit failures with rollback.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise OperationalError("Commit failed", {}, Exception()) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + session_data = SessionCreate( + user_id=async_test_user.id, + refresh_token_jti=str(uuid4()), + device_name="Test Device", + ip_address="127.0.0.1", + user_agent="Test Agent", + expires_at=datetime.now(timezone.utc) + timedelta(days=7), + last_used_at=datetime.now(timezone.utc) + ) + + with pytest.raises(ValueError, match="Failed to create session"): + await session_crud.create_session(session, obj_in=session_data) + + mock_rollback.assert_called_once() + + @pytest.mark.asyncio + async def test_create_session_unexpected_error_triggers_rollback(self, async_test_db, async_test_user): + """Test create_session handles unexpected errors.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise RuntimeError("Unexpected error") + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + session_data = SessionCreate( + user_id=async_test_user.id, + refresh_token_jti=str(uuid4()), + device_name="Test Device", + ip_address="127.0.0.1", + user_agent="Test Agent", + expires_at=datetime.now(timezone.utc) + timedelta(days=7), + last_used_at=datetime.now(timezone.utc) + ) + + with pytest.raises(ValueError, match="Failed to create session"): + await session_crud.create_session(session, obj_in=session_data) + + mock_rollback.assert_called_once() + + +class TestSessionCRUDDeactivateFailures: + """Test deactivate exception handling.""" + + @pytest.mark.asyncio + async def test_deactivate_commit_failure_triggers_rollback(self, async_test_db, async_test_user): + """Test deactivate handles commit failures.""" + test_engine, SessionLocal = async_test_db + + # Create a session first + async with SessionLocal() as session: + user_session = UserSession( + user_id=async_test_user.id, + refresh_token_jti=str(uuid4()), + device_name="Test Device", + ip_address="127.0.0.1", + user_agent="Test Agent", + is_active=True, + expires_at=datetime.now(timezone.utc) + timedelta(days=7), + last_used_at=datetime.now(timezone.utc) + ) + session.add(user_session) + await session.commit() + await session.refresh(user_session) + session_id = user_session.id + + # Test deactivate failure + async with SessionLocal() as session: + async def mock_commit(): + raise OperationalError("Deactivate failed", {}, Exception()) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(OperationalError): + await session_crud.deactivate(session, session_id=str(session_id)) + + mock_rollback.assert_called_once() + + +class TestSessionCRUDDeactivateAllFailures: + """Test deactivate_all_user_sessions exception handling.""" + + @pytest.mark.asyncio + async def test_deactivate_all_commit_failure_triggers_rollback(self, async_test_db, async_test_user): + """Test deactivate_all handles commit failures.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise OperationalError("Bulk deactivate failed", {}, Exception()) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(OperationalError): + await session_crud.deactivate_all_user_sessions( + session, + user_id=str(async_test_user.id) + ) + + mock_rollback.assert_called_once() + + +class TestSessionCRUDUpdateLastUsedFailures: + """Test update_last_used exception handling.""" + + @pytest.mark.asyncio + async def test_update_last_used_commit_failure_triggers_rollback(self, async_test_db, async_test_user): + """Test update_last_used handles commit failures.""" + test_engine, SessionLocal = async_test_db + + # Create a session + async with SessionLocal() as session: + user_session = UserSession( + user_id=async_test_user.id, + refresh_token_jti=str(uuid4()), + device_name="Test Device", + ip_address="127.0.0.1", + user_agent="Test Agent", + is_active=True, + expires_at=datetime.now(timezone.utc) + timedelta(days=7), + last_used_at=datetime.now(timezone.utc) - timedelta(hours=1) + ) + session.add(user_session) + await session.commit() + await session.refresh(user_session) + + # Test update failure + async with SessionLocal() as session: + from sqlalchemy import select + from app.models.user_session import UserSession as US + result = await session.execute(select(US).where(US.id == user_session.id)) + sess = result.scalar_one() + + async def mock_commit(): + raise OperationalError("Update failed", {}, Exception()) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(OperationalError): + await session_crud.update_last_used(session, session=sess) + + mock_rollback.assert_called_once() + + +class TestSessionCRUDUpdateRefreshTokenFailures: + """Test update_refresh_token exception handling.""" + + @pytest.mark.asyncio + async def test_update_refresh_token_commit_failure_triggers_rollback(self, async_test_db, async_test_user): + """Test update_refresh_token handles commit failures.""" + test_engine, SessionLocal = async_test_db + + # Create a session + async with SessionLocal() as session: + user_session = UserSession( + user_id=async_test_user.id, + refresh_token_jti=str(uuid4()), + device_name="Test Device", + ip_address="127.0.0.1", + user_agent="Test Agent", + is_active=True, + expires_at=datetime.now(timezone.utc) + timedelta(days=7), + last_used_at=datetime.now(timezone.utc) + ) + session.add(user_session) + await session.commit() + await session.refresh(user_session) + + # Test update failure + async with SessionLocal() as session: + from sqlalchemy import select + from app.models.user_session import UserSession as US + result = await session.execute(select(US).where(US.id == user_session.id)) + sess = result.scalar_one() + + async def mock_commit(): + raise OperationalError("Token update failed", {}, Exception()) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(OperationalError): + await session_crud.update_refresh_token( + session, + session=sess, + new_jti=str(uuid4()), + new_expires_at=datetime.now(timezone.utc) + timedelta(days=14) + ) + + mock_rollback.assert_called_once() + + +class TestSessionCRUDCleanupExpiredFailures: + """Test cleanup_expired exception handling.""" + + @pytest.mark.asyncio + async def test_cleanup_expired_commit_failure_triggers_rollback(self, async_test_db): + """Test cleanup_expired handles commit failures.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise OperationalError("Cleanup failed", {}, Exception()) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(OperationalError): + await session_crud.cleanup_expired(session, keep_days=30) + + mock_rollback.assert_called_once() + + +class TestSessionCRUDCleanupExpiredForUserFailures: + """Test cleanup_expired_for_user exception handling.""" + + @pytest.mark.asyncio + async def test_cleanup_expired_for_user_commit_failure_triggers_rollback(self, async_test_db, async_test_user): + """Test cleanup_expired_for_user handles commit failures.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_commit(): + raise OperationalError("User cleanup failed", {}, Exception()) + + with patch.object(session, 'commit', side_effect=mock_commit): + with patch.object(session, 'rollback', new_callable=AsyncMock) as mock_rollback: + with pytest.raises(OperationalError): + await session_crud.cleanup_expired_for_user( + session, + user_id=str(async_test_user.id) + ) + + mock_rollback.assert_called_once() + + +class TestSessionCRUDGetUserSessionCountFailures: + """Test get_user_session_count exception handling.""" + + @pytest.mark.asyncio + async def test_get_user_session_count_database_error(self, async_test_db, async_test_user): + """Test get_user_session_count handles database errors.""" + test_engine, SessionLocal = async_test_db + + async with SessionLocal() as session: + async def mock_execute(*args, **kwargs): + raise OperationalError("Count query failed", {}, Exception()) + + with patch.object(session, 'execute', side_effect=mock_execute): + with pytest.raises(OperationalError): + await session_crud.get_user_session_count( + session, + user_id=str(async_test_user.id) + )