diff --git a/api/v1/routes/user.py b/api/v1/routes/user.py index cec064283..fcc7484d7 100644 --- a/api/v1/routes/user.py +++ b/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'] ) - ) \ No newline at end of file + ) + + + + +@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": {} + } + ) diff --git a/api/v1/schemas/user.py b/api/v1/schemas/user.py index 095135e11..2cbd07511 100644 --- a/api/v1/schemas/user.py +++ b/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") + + diff --git a/hello.db b/hello.db new file mode 100644 index 000000000..ff89a5f2f Binary files /dev/null and b/hello.db differ diff --git a/tests/v1/superadmin/test_admin_to_delete_user.py b/tests/v1/superadmin/test_admin_to_delete_user.py new file mode 100644 index 000000000..e1ec85521 --- /dev/null +++ b/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="superadmin@gmail.com", + 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="testuser@gmail.com", + 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="dummyuser1@gmail.com", + 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="testuser@gmail.com", + 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="inactiveuser@gmail.com", + 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="regularuser@gmail.com", + 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"