feat: Add endpoint to remove all users blong to an admin

Author: ImMohammad20000

app/operation/admin.py

@@ -14,11 +14,12 @@
     update_admin,
 )
 from app.db.crud.bulk import activate_all_disabled_users, disable_all_active_users
-from app.db.crud.user import get_users
+from app.db.crud.user import get_users, remove_users
 from app.db.models import Admin as DBAdmin
 from app.models.admin import AdminCreate, AdminDetails, AdminModify
 from app.node import node_manager
 from app.operation import BaseOperation, OperatorType
+from app.operation.user import UserOperation
 from app.utils.logger import get_logger
 
 logger = get_logger("admin-operation")
@@ -126,6 +127,32 @@ async def activate_all_disabled_users(self, db: AsyncSession, username: str, adm
 
         logger.info(f'Admin "{username}" users has been activated by admin "{admin.username}"')
 
+    async def remove_all_users(self, db: AsyncSession, username: str, admin: AdminDetails) -> int:
+        """Delete all users that belong to the specified admin."""
+        db_admin = await self.get_validated_admin(db, username=username)
+        target_username = db_admin.username
+
+        if self.operator_type != OperatorType.CLI and db_admin.is_sudo:
+            await self.raise_error(message="You're not allowed to delete sudo admin users.", code=403)
+
+        users = await get_users(db, admin=db_admin)
+        if not users:
+            return 0
+
+        user_operation = UserOperation(self.operator_type)
+        serialized_users = [await user_operation.validate_user(user) for user in users]
+
+        await remove_users(db, users)
+
+        for user in serialized_users:
+            await node_manager.remove_user(user)
+            asyncio.create_task(notification.remove_user(user, admin))
+
+        logger.info(
+            f'Admin "{admin.username}" deleted {len(serialized_users)} users belonging to admin "{target_username}"'
+        )
+        return len(serialized_users)
+
     async def reset_admin_usage(self, db: AsyncSession, username: str, admin: AdminDetails) -> AdminDetails:
         db_admin = await self.get_validated_admin(db, username=username)
 

app/routers/admin.py

@@ -1,13 +1,16 @@
 import asyncio
-from fastapi import APIRouter, Depends, HTTPException, Request, status, Header
+
+from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
 from fastapi.security import OAuth2PasswordRequestForm
+
 from app import notification
 from app.db import AsyncSession, get_db
 from app.models.admin import AdminCreate, AdminDetails, AdminModify, Token
 from app.operation import OperatorType
 from app.operation.admin import AdminOperation
 from app.utils import responses
 from app.utils.jwt import create_admin_token
+
 from .authentication import check_sudo_admin, get_current, validate_admin, validate_mini_app_admin
 
 router = APIRouter(tags=["Admin"], prefix="/api/admin", responses={401: responses._401, 403: responses._403})
@@ -147,6 +150,15 @@ async def activate_all_disabled_users(
     return {}
 
 
+@router.delete("/{username}/users", responses={403: responses._403, 404: responses._404})
+async def remove_all_users(
+    username: str, db: AsyncSession = Depends(get_db), admin: AdminDetails = Depends(check_sudo_admin)
+):
+    """Remove all users under a specific admin."""
+    deleted = await admin_operator.remove_all_users(db, username=username, admin=admin)
+    return {"detail": f"operation has been successfuly done {deleted} users deleted"}
+
+
 @router.post("/{username}/reset", response_model=AdminDetails, responses={404: responses._404})
 async def reset_admin_usage(
     username: str, db: AsyncSession = Depends(get_db), admin: AdminDetails = Depends(check_sudo_admin)

tests/api/test_a_admin.py

@@ -1,6 +1,26 @@
+import asyncio
+
 from fastapi import status
 
-from tests.api import client
+from app.db.models import Group
+from tests.api import TestSession, client
+
+
+async def _create_group_record(name: str) -> int:
+    async with TestSession() as session:
+        group = Group(name=name, inbounds=[], is_disabled=False)
+        session.add(group)
+        await session.commit()
+        await session.refresh(group)
+        return group.id
+
+
+async def _delete_group_record(group_id: int):
+    async with TestSession() as session:
+        group = await session.get(Group, group_id)
+        if group:
+            await session.delete(group)
+            await session.commit()
 
 
 def test_admin_login():
@@ -101,6 +121,72 @@ def test_disable_admin():
     assert response.json()["detail"] == "your account has been disabled"
 
 
+def test_admin_delete_all_users_endpoint(access_token):
+    """Test deleting all users belonging to an admin."""
+
+    admin_username = "testadminbulkdelete"
+    admin_password = "TestAdminBulkdelete#11"
+
+    response = client.post(
+        url="/api/admin",
+        json={"username": admin_username, "password": admin_password, "is_sudo": False},
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+
+    group_id = asyncio.run(_create_group_record(f"{admin_username}_group"))
+
+    created_users = []
+    for idx in range(2):
+        user_name = f"{admin_username}_user_{idx}"
+        user_response = client.post(
+            "/api/user",
+            headers={"Authorization": f"Bearer {access_token}"},
+            json={
+                "username": user_name,
+                "proxy_settings": {},
+                "group_ids": [group_id],
+                "data_limit": 1024,
+                "data_limit_reset_strategy": "no_reset",
+                "status": "active",
+            },
+        )
+        assert user_response.status_code == status.HTTP_201_CREATED
+        created_users.append(user_name)
+
+        ownership_response = client.put(
+            f"/api/user/{user_name}/set_owner",
+            headers={"Authorization": f"Bearer {access_token}"},
+            params={"admin_username": admin_username},
+        )
+        assert ownership_response.status_code == status.HTTP_200_OK
+        assert ownership_response.json()["admin"]["username"] == admin_username
+
+    response = client.delete(
+        url=f"/api/admin/{admin_username}/users",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["deleted"] == len(created_users)
+
+    for username in created_users:
+        user_check = client.get(
+            "/api/users",
+            params={"username": username},
+            headers={"Authorization": f"Bearer {access_token}"},
+        )
+        assert user_check.status_code == status.HTTP_200_OK
+        assert user_check.json()["users"] == []
+
+    cleanup = client.delete(
+        url=f"/api/admin/{admin_username}",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert cleanup.status_code == status.HTTP_204_NO_CONTENT
+
+    asyncio.run(_delete_group_record(group_id))
+
+
 def test_admin_delete(access_token):
     """Test that the admin delete route is accessible."""