test: Add user_service test class

Author: cevheri

app/api/user_api.py

@@ -93,8 +93,8 @@ async def create(
     The endpoint creates a new user with the provided user data and returns the created user's details.
     """
     token_data = request.state.jwt_user
-    _log.debug(f"UserApi Creating user: {user_create_data}  {token_data}")
-    result = await user_service.create(user_create_data)
+    _log.debug(f"UserApi Creating user: {user_create_data}  TokenUser:{token_data}")
+    result = await user_service.create(user_create_data, token_data)
     if result is None:
         _log.error(f"UserApi User not created")
         raise HTTPException(status_code=400, detail="User not created")

app/entity/user_entity.py

@@ -1,80 +1,27 @@
-from datetime import datetime, timezone
-from typing import Optional
+from datetime import datetime
 
 from beanie import Document
 from pydantic import EmailStr
 
-from app.schema.user_dto import UserDTO, UserCreate, UserUpdate
-
 
 class User(Document):
-    user_id: str
+    user_id: str | None = None
     username: str
     first_name: str
     last_name: str
     email: EmailStr
-    hashed_password: Optional[str] = None
-    is_active: bool = False
+    hashed_password: str | None = None
+    is_active: bool | None = None
     roles: list[str] | None = None
-    created_by: Optional[str] = None
-    created_date: Optional[datetime] = None
+    created_by: str | None = None
+    created_date: datetime | None = None
     last_updated_by: str | None = None
-    last_updated_date: datetime = datetime.now(timezone.utc)
-    age: Optional[int] = None
+    last_updated_date: datetime | None = None
+    age: int | None = None
 
     class Settings:
         name = "app_user"
         validate_on_save = True
 
-    @staticmethod
-    def from_create(user_create: UserCreate):
-        return User(
-            user_id="",
-            username=user_create.username,
-            first_name=user_create.first_name,
-            last_name=user_create.last_name,
-            email=user_create.email,
-            hashed_password="",
-            is_active=user_create.is_active,
-            roles=user_create.roles,
-            created_by="system",
-            created_date=datetime.now().isoformat(),
-            last_updated_by="system",
-            last_updated_date=datetime.now().isoformat(),
-        )
-
-    @staticmethod
-    def from_update(user_update: UserUpdate):
-        return User(
-            user_id="",  # Not allowed to update
-            username="",  # Not allowed to update
-            first_name=user_update.first_name,
-            last_name=user_update.last_name,
-            email=user_update.email,
-            hashed_password="",  # Not allowed to update
-            is_active=user_update.is_active,
-            roles=user_update.roles,
-            created_by="",  # Not allowed to update
-            created_date=None,  # Not allowed to update
-            last_updated_by="system",
-            last_updated_date=datetime.now(timezone.utc),
-        )
-
-    @staticmethod
-    def from_dto(user_dto: UserDTO):
-        return User(
-            user_id=user_dto.user_id,
-            username=user_dto.username,
-            first_name=user_dto.first_name,
-            last_name=user_dto.last_name,
-            email=user_dto.email,
-            is_active=user_dto.is_active,
-            roles=user_dto.roles,
-            created_by=user_dto.created_by,
-            created_date=user_dto.created_date,
-            last_updated_by=user_dto.last_updated_by,
-            last_updated_date=user_dto.last_updated_date,
-        )
-
     def __str__(self):
         return f"User: {self.user_id}, {self.username}, {self.first_name} {self.last_name}, {self.email}, {self.is_active}, {self.roles}"

app/middleware/security_middleware.py

@@ -4,6 +4,7 @@
 
 from app.conf.app_settings import security_settings, server_settings
 from app.security import auth_handler
+from app.security.jwt_token import JWTUser
 
 API_PREFIX = server_settings.CONTEXT_PATH
 ALLOWED_PATHS = [resource for resource in security_settings.ALLOWED_PATHS]
@@ -38,6 +39,6 @@ def _is_valid_token(token: str) -> bool:
         return token is not None and auth_handler.is_valid_token(token.replace("Bearer ", ""))
 
     @staticmethod
-    def _get_user_from_token(token: str):
+    def _get_user_from_token(token: str) -> JWTUser:
         """Retrieve user information from the token."""
         return auth_handler.get_jwt_user_from_token(token.replace("Bearer ", ""))

app/repository/user_repository.py

@@ -28,6 +28,8 @@ async def create(self, user: User) -> User:
 
     async def update(self, user: User) -> User | None:
         _log.debug(f"UserRepository Updating user")
+        if user.user_id is None:
+            raise BusinessException(ErrorCodes.INVALID_PAYLOAD, "User id is required for update")
         result = await user.replace()
         _log.debug(f"UserRepository User updated")
         return result
@@ -74,7 +76,7 @@ async def find(self, query: str | None = None, page: int = 0, size: int = 10, so
         else:
             query = json.loads(query)
 
-        total_count = await self.count(query)
+        total_count = await User.find(query).count()
         if total_count == 0:
             return PageResponse(content=[], page=page, size=size, total=total_count)
 

app/security/jwt_token.py

@@ -9,7 +9,7 @@ class JWTUser(BaseModel):
     sub: str = Field(alias="sub", title="Username")
     email: str = Field(alias="email", title="Email")
     scopes: list[str] = Field(alias="scopes", title="Roles")
-    expires: float = Field(alias="exp", title="Expires of the token")
+    exp: float = Field(alias="exp", title="Expires of the token")
     token: str = Field(alias="token", title="Access Token")
 
 

app/service/user_service.py

@@ -10,6 +10,7 @@
 from app.errors.business_exception import BusinessException, ErrorCodes
 from app.repository.user_repository import UserRepository
 from app.schema.user_dto import UserDTO, UserCreate, UserUpdate
+from app.security.jwt_token import JWTUser
 from app.service import email_service
 from app.utils.pass_util import PasswordUtil
 
@@ -62,13 +63,16 @@ async def user_create_validation(self, user_create: UserCreate):
             raise BusinessException(ErrorCodes.ALREADY_EXISTS,
                                     f"User with email already exists: {user_create.email}")
 
-    async def create(self, user_create: UserCreate) -> UserDTO:
+    async def create(self, user_create: UserCreate, token_data: JWTUser) -> UserDTO:
         _log.debug(f"UserService Creating user: {user_create} with: {type(user_create)}")
 
         try:
-            user = User.from_create(user_create)
+            hashed_password = PasswordUtil().hash_password(user_create.password)
+            user = User(**user_create.model_dump(), hashed_password = hashed_password)
+            user.created_by = token_data.sub
+            user.last_updated_by = token_data.sub
             user.user_id = str(uuid.uuid4())
-            user.hashed_password = PasswordUtil().hash_password(user_create.password)
+
             await self.user_create_validation(user_create)
             final_user = await self.repository.create(user)
             result = UserDTO.model_validate(final_user)
@@ -107,7 +111,7 @@ async def find(self, query, page: int, size: int, sort: str) -> PageResponse[Use
         _log.debug("UserService Users retrieved")
         return page_response
 
-    async def update(self, user_id: str, user_update: UserUpdate) -> Optional[UserDTO]:
+    async def update(self, user_id: str, user_update: UserUpdate, token_data: JWTUser) -> Optional[UserDTO]:
         _log.debug(f"UserService Updating user: {user_id} with: {type(user_update)}")
 
         if not user_update:
@@ -123,6 +127,7 @@ async def update(self, user_id: str, user_update: UserUpdate) -> Optional[UserDT
             if getattr(user_update, attr) is not None:
                 setattr(user_entity, attr, getattr(user_update, attr))
 
+        user_entity.last_updated_by = token_data.sub
         final_user = await self.repository.update(user_entity)
         result = UserDTO.model_validate(final_user)
         _log.debug("UserService User updated")

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+asyncio_default_fixture_loop_scope = function

test/service/test_user_service.py

@@ -1,15 +1,16 @@
 # python unittest for user_service layer with mocking repository
+import unittest
 from datetime import datetime
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, patch, MagicMock
 
-import pytest
+from beanie import init_beanie
+from mongomock_motor import AsyncMongoMockClient
 
-from app.schema.user_dto import UserCreate, UserUpdate, UserDTO
+from app.entity import User
+from app.schema.user_dto import UserCreate, UserUpdate
+from app.schema.user_dto import UserDTO
 from app.security.jwt_token import JWTUser
 from app.service.user_service import UserService
-from unittest.mock import AsyncMock, patch
-from app.schema.user_dto import UserDTO
-from app.service.user_service import send_creation_email
 
 
 # region init entity
@@ -20,7 +21,7 @@ def _get_jwt_user():
         sub="system",
         email="system@system.com",
         scopes=["system"],
-        expires=1000,
+        exp=1000,
         token="token",
     )
 
@@ -63,54 +64,59 @@ def _get_dto():
     )
 
 
+def _get_mock_dto():
+    dto = MagicMock()
+    dto.first_name = "test"
+    dto.last_name = "test"
+    dto.email = "test@test.com"
+    dto.is_active = True
+    dto.roles = ["test"]
+    dto.user_id = "test"
+    dto.username = "test"
+    dto.created_by = "system"
+    dto.created_date = datetime(2024, 1, 1)
+    dto.last_updated_by = "system"
+    dto.last_updated_date = datetime(2024, 1, 1)
+    dto.password = "password"
+    return dto
+
+
 def _get_service(repository):
     return UserService(user_repository=repository)
 
 
 # endregion init entity
 
 
-# success case for create
-@pytest.mark.asyncio
-async def test_given_valid_create_entity_when_create_then_return_dto():
-    # Arrange 
-    @pytest.mark.asyncio
-    async def test_send_creation_email_success():
+# region test_service
+class TestUserService(unittest.IsolatedAsyncioTestCase):
+
+    async def asyncSetUp(self):
+        self.mock_repository = AsyncMock()
+        self.service = _get_service(self.mock_repository)
+        self.client = AsyncMongoMockClient()
+        await init_beanie(document_models=[User], database=self.client.get_database(name="pyfapi"))
+
+    async def asyncTearDown(self):
+        await self.client.drop_database("pyfapi")
+
+    @patch("app.service.user_service.send_creation_email", new_callable=AsyncMock)
+    async def test_given_create_entity_when_create_then_send_email_success(self, mock_send_email):
         # Arrange
-        user = _get_dto()
-        with patch("app.service.email_service.send_email", new_callable=AsyncMock) as mock_send_email:
-            # Act
-            await send_creation_email(user)
-
-            # Assert
-            mock_send_email.assert_called_once_with(
-                "john.doe@example.com",
-                "Welcome to the TestApp",
-                "Hello John,\n\n"
-                "Welcome to the TestApp. Your account has been created successfully.\n\n"
-                "Please visit http://testapp.com to login to your account.\n\n"
-                "TestApp Team."
-            )
-
-
-@pytest.mark.asyncio
-async def test_send_creation_email_failure():
-    # Arrange
-    user = _get_dto()
-    with patch("app.service.email_service.send_email", new_callable=AsyncMock) as mock_send_email:
-        mock_send_email.side_effect = Exception("Email service failure")
+        create_entity = _get_create_entity()
+        jwt_user = _get_jwt_user()
+
+        fvo = _get_mock_dto()
+
+        self.mock_repository.create.return_value = fvo
 
         # Act
-        with pytest.raises(Exception) as exc_info:
-            await send_creation_email(user)
+        result = await self.service.create(create_entity, jwt_user)
 
         # Assert
-        assert str(exc_info.value) == "Email service failure"
-        mock_send_email.assert_called_once_with(
-            "john.doe@example.com",
-            "Welcome to the TestApp",
-            "Hello John,\n\n"
-            "Welcome to the TestApp. Your account has been created successfully.\n\n"
-            "Please visit http://testapp.com to login to your account.\n\n"
-            "TestApp Team."
-        )
+        self.mock_repository.create.assert_called_once()
+        mock_send_email.assert_called_once_with(UserDTO.model_validate(fvo))
+        self.assertEqual(result.user_id, fvo.user_id)
+        self.assertEqual(result, UserDTO.model_validate(fvo))
+
+# endregion test_service