feat: added superadmin to deactivate user endpoint

api/v1/routes/user.py

@@ -7,11 +7,18 @@
 from api.v1.models.user import User
 from api.v1.schemas.user import (
     AllUsersResponse, UserUpdate,
-    AdminCreateUserResponse, AdminCreateUser
+    AdminCreateUserResponse, AdminCreateUser,AdminDeleteUserSchema,
 )
 from api.db.database import get_db
 from api.v1.services.user import user_service
 
+from api.utils.dependencies import get_current_user,get_super_admin
+
+from fastapi.responses import JSONResponse
+
+from api.core.dependencies.google_email import mail_service
+
+
 
 user_router = APIRouter(prefix="/users", tags=["Users"])
 
@@ -209,4 +216,69 @@ def get_user_by_id(
             user, 
             exclude=['password', 'is_superadmin', 'is_deleted', 'is_verified', 'updated_at', 'created_at', 'is_active']
         )
-    )
+    )
+
+
+
+
+@user_router.post('/deactivate', status_code=200)
+async def deactivate_account(
+    request: Request, 
+    schema: AdminDeleteUserSchema, 
+    db: Session = Depends(get_db), 
+    super_admin: User = Depends(user_service.get_current_super_admin)
+):
+    '''Endpoint for super admin to deactivate a user account by user ID'''
+
+
+    if not super_admin.is_superadmin:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only super admins can deactivate users"
+        )
+
+    user_to_deactivate = db.query(User).filter(User.id == schema.user_id).first()
+
+    if user_to_deactivate is None:
+        return JSONResponse(
+            status_code=404,
+            content={
+                "status": "error",
+                "status_code": 404,
+                "message": "User not found",
+                "data": {}
+            }
+        )
+
+    if not user_to_deactivate.is_active:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "status": "error",
+                "status_code": 400,
+                "message": "User is already deactivated",
+                "data": {}
+            }
+        )
+
+    user_to_deactivate.is_active = False
+
+    # Send email to user
+    mail_service.send_mail(
+        to=user_to_deactivate.email,
+        subject='Account Deactivation',
+        body=f'Hello {user_to_deactivate.first_name},\n\nYour account has been deactivated successfully.\n\nTo reactivate your account, visit:\n{request.url.hostname}/api/v1/users/accounts/reactivate\n\nThis link will expire in 15 minutes.'
+    )
+
+    db.commit()
+
+
+    return JSONResponse(
+        status_code=200,
+        content={
+            "status": "success",
+            "status_code": 200,
+            "message": "User account deactivated successfully",
+            "data": {}
+        }
+    )

api/v1/schemas/user.py

@@ -438,3 +438,13 @@ def role_validator(cls, value):
         if value not in ["admin", "user", "guest", "owner"]:
             raise ValueError("Role has to be one of admin, guest, user, or owner")
         return value
+
+
+
+
+class AdminDeleteUserSchema(BaseModel):
+    """Schema for admin for deleting a user account"""
+
+    user_id: str = Field(..., title="User ID", description="The ID of the user to be deleted")
+
+

tests/v1/superadmin/test_admin_to_delete_user.py

@@ -0,0 +1,206 @@
+import pytest
+from fastapi.testclient import TestClient
+from unittest.mock import MagicMock
+from main import app
+from api.v1.models.user import User
+from api.v1.schemas.user import AdminDeleteUserSchema
+from api.db.database import get_db
+from api.v1.services.user import user_service
+from api.core.dependencies.google_email import mail_service
+from datetime import datetime, timezone
+from uuid_extensions import uuid7
+from fastapi import status
+
+client = TestClient(app)
+
+@pytest.fixture
+def db_session_mock():
+    db_session = MagicMock()
+    yield db_session
+
+@pytest.fixture(autouse=True)
+def override_get_db(db_session_mock):
+    def get_db_override():
+        yield db_session_mock
+
+    app.dependency_overrides[get_db] = get_db_override
+    yield
+    app.dependency_overrides = {}
+
+@pytest.fixture
+def super_admin_user():
+    return User(
+        id=str(uuid7()),
+        email="[email protected]",
+        password=user_service.hash_password("Superpassword@123"),
+        first_name="Super",
+        last_name="Admin",
+        is_active=True,
+        is_superadmin=True,
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+@pytest.fixture
+def regular_user():
+    return User(
+        id=str(uuid7()),
+        email="[email protected]",
+        password=user_service.hash_password("Testpassword@123"),
+        first_name="Test",
+        last_name="User",
+        is_active=True,
+        is_superadmin=True,
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+@pytest.fixture
+def mock_get_current_user():
+    """Fixture to create a mock current user"""
+    with patch(
+        "api.v1.services.user.UserService.get_current_user", autospec=True
+    ) as mock_get_current_user:
+        yield mock_get_current_user
+
+from unittest.mock import patch, MagicMock
+
+
+@pytest.fixture
+def mock_db_session():
+    """Fixture to create a mock database session."
+
+    Yields:
+        MagicMock: mock database
+    """
+
+    with patch("api.v1.services.user.get_db", autospec=True) as mock_get_db:
+        mock_db = MagicMock()
+        app.dependency_overrides[get_db] = lambda: mock_db
+        yield mock_db
+    app.dependency_overrides = {}
+
+@pytest.fixture
+def mock_mail_service():
+    """Fixture to mock the mail service."""
+    with patch("api.core.dependencies.google_email.mail_service.send_mail", autospec=True) as mock_service:
+        yield mock_service
+
+def test_deactivate_account_success(mock_db_session, mock_mail_service):
+    """Test successful deactivation of a user account by a super admin"""
+    mock_id = "mock_user_id"
+    dummy_mock_user = User(
+        id=mock_id,
+        email="[email protected]",
+        password=user_service.hash_password("Testpassword@123"),
+        first_name="Mr",
+        last_name="Dummy",
+        is_active=True,
+        is_superadmin=True,
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+    app.dependency_overrides[user_service.get_current_super_admin] = lambda: dummy_mock_user
+
+
+
+    user = User(
+        id=str(uuid7()),
+        email="[email protected]",
+        password=user_service.hash_password("Testpassword@123"),
+        first_name="Test",
+        last_name="User",
+        is_active=True,
+        is_superadmin=True,
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+
+    response = client.post(
+        "/api/v1/users/deactivate",
+        json={"user_id": user.id, "token": "testtoken"},
+        headers={"Authorization": "Bearer testtoken"},
+    )
+    response_json = response.json()
+
+
+    mock_mail_service.assert_called_once()
+
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response_json.get("status_code") == status.HTTP_200_OK
+    assert response_json.get("message") == "User account deactivated successfully"
+
+
+def test_deactivate_account_user_not_found(mock_db_session):
+    """Test deactivation failure when the user ID does not exist"""
+    app.dependency_overrides[user_service.get_current_super_admin] = lambda: User(is_superadmin=True)
+
+    mock_db_session.query().filter().first.return_value = None
+
+    response = client.post(
+        "/api/v1/users/deactivate",
+        json={"user_id": "non_existent_user_id", "token": "testtoken"},
+        headers={"Authorization": "Bearer testtoken"},
+    )
+
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.json()["message"] == "User not found"
+def test_deactivate_account_already_deactivated(mock_db_session):
+    """Test that an already deactivated user cannot be deactivated again"""
+    user = User(
+        id="deactivated_user_id",
+        email="[email protected]",
+        password=user_service.hash_password("Testpassword@123"),
+        first_name="Inactive",
+        last_name="User",
+        is_active=False,  
+        is_superadmin=False,
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+    app.dependency_overrides[user_service.get_current_super_admin] = lambda: User(is_superadmin=True)
+
+
+    mock_db_session.query().filter().first.return_value = user
+
+    response = client.post(
+        "/api/v1/users/deactivate",
+        json={"user_id": "deactivated_user_id", "token": "testtoken"},
+        headers={"Authorization": "Bearer testtoken"},
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json()["message"] == "User is already deactivated"
+
+
+def test_deactivate_account_unauthorized(mock_db_session):
+    """Test that a non-super-admin cannot deactivate a user account"""
+    mock_id = "mock_user_id"
+    non_admin_user = User(
+        id=mock_id,
+        email="[email protected]",
+        password=user_service.hash_password("Testpassword@123"),
+        first_name="Regular",
+        last_name="User",
+        is_active=True,
+        is_superadmin=False,  
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+
+    app.dependency_overrides[user_service.get_current_super_admin] = lambda: non_admin_user
+
+    response = client.post(
+        "/api/v1/users/deactivate",
+        json={"user_id": "some_user_id", "token": "testtoken"},
+        headers={"Authorization": "Bearer testtoken"},
+    )
+
+
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert response.json()["message"] == "Only super admins can deactivate users"