Chim/feature/add notification service

.gitignore

@@ -245,4 +245,7 @@ packages/react-devtools-extensions/shared/build
 packages/react-devtools-extensions/.tempUserDataDir
 packages/react-devtools-inline/dist
 packages/react-devtools-shell/dist
-packages/react-devtools-scheduling-profiler/dist
+packages/react-devtools-scheduling-profiler/dist
+
+# Ignore Firebase service account key
+google-credentials.json

Pipfile

@@ -49,6 +49,7 @@ werkzeug = ">=1.0.1"
 numpy = "==1.24.4"
 pytest = "*"
 dataclasses = "*"
+firebase-admin = "*"
 
 [dev-packages]
 

repository/fcm_token_repository.py

@@ -1,9 +1,11 @@
 import logging
+from typing import List, Optional
 from datetime import datetime
 from domain.entity.fcm_tokens import FCMToken
 from domain import session_scope
 from exception import InvalidTokenError
 from sqlalchemy.exc import SQLAlchemyError
+from services.notification_service import NotificationService
 
 
 class FCMTokenRepository:
@@ -55,3 +57,70 @@ def unregister_token(self, user_id: int, fcm_token: str) -> None:
                 logging.error(f"Database error while unregistering FCM token for user {user_id}:{e}")
                 session.rollback()
                 raise e
+
+    def get_fcm_token(self, user_id: int) -> List[str]:
+        """
+        Get FCM tokens for a user
+
+        :param user_id: The user id
+        :return: A list of FCM tokens for a given user.
+        """
+        with session_scope() as session:
+            try:
+                # Query to get all tokens for the given user_id
+                existing_tokens = session.query(FCMToken).filter_by(user_id=user_id).all()
+
+                if not existing_tokens:
+                    logging.info(f"No active FCM token found for user {user_id}")
+                    return []
+
+                fcm_token_list = [token.fcm_token for token in existing_tokens]
+                return fcm_token_list
+
+            except SQLAlchemyError as e:
+                logging.error(f"Database error while retrieving FCM token for user {user_id}: {e}")
+
+                raise e
+
+    def notify_user(
+            self,
+            user_id: Optional[int] = None,
+            fcm_token_list: Optional[List[str]] = None,
+            title: Optional[str] = None,
+            body: Optional[str] = None,
+            data: Optional[dict] = None
+    ) -> None:
+        """
+        Notify a user by sending a notification to their device.
+
+        This function either accepts a user id to look up tokens or take a list of FCM tokens.
+
+        :param user_id: The user id (optional if fcm token list is provided)
+        :param fcm_token_list: A list of FCM tokens (optional if user id is provided)
+        :param title: The title of the notification (required)
+        :param body: The body of the notification (required)
+        :param data: The data of the notification (optional)
+        """
+
+        # Ensure either user_id or fcm_token_list is provided
+        if user_id is None and fcm_token_list is None:
+            raise ValueError("Either user_id or fcm_token_list must be provided")
+
+        if title is None or body is None:
+            raise ValueError("Title and body must be provided")
+
+        # Get the token list if user_id is provided, otherwise use fcm_token_list
+        if user_id is not None:
+            token_list = self.get_fcm_token(user_id)
+        else:
+            token_list = fcm_token_list
+
+        # Ensure token_list is not empty
+        if not token_list:
+            raise ValueError("No valid FCM tokens found or provided")
+
+        # Call notification service function
+        notification_service = NotificationService()
+        notification_service.send_notification(
+            token_list, title, body, data
+        )

services/notification_service.py

@@ -0,0 +1,43 @@
+import logging
+import os
+import firebase_admin
+from firebase_admin import credentials, messaging, exceptions
+from typing import List, Optional
+
+
+class NotificationService:
+
+    def __init__(self):
+
+        if not firebase_admin._apps:
+            cred_path = f"{os.getcwd()}/google-credentials.json"
+            cred = credentials.Certificate(cred_path)
+            firebase_admin.initialize_app(cred)
+
+    def send_notification(self, fcm_token_list: List[str], title: str, body: str, data: Optional[dict] = None) -> None:
+        """
+        Send message to the user device given a list of FCM tokens.
+
+        :param fcm_token_list: List of FCM tokens to send the notification to.
+        :param title: The title of the notification.
+        :param body: The body content of the notification.
+        :param data: Optional additional data for the notification.
+        """
+        if not fcm_token_list:
+            logging.warning("No FCM tokens provided. Cannot send notification.")
+
+        for token in fcm_token_list:
+
+            message = messaging.Message(
+                notification=messaging.Notification(title=title, body=body),
+                data=data or {},
+                token=token
+            )
+
+            try:
+                response = messaging.send(message)
+                logging.info(f"Successfully sent message to token: {token}")
+            except exceptions.FirebaseError as e:
+                logging.error(f"Firebase error for token {token}: {e}")
+            except Exception as e:
+                logging.error(f"Error sending message to token {token}: {e}")

tests/unit/test_fcm_repository.py

@@ -0,0 +1,106 @@
+import unittest
+from datetime import datetime
+from unittest.mock import patch, MagicMock
+from repository.fcm_token_repository import FCMTokenRepository
+from domain.entity.fcm_tokens import FCMToken
+
+
+class TestFCMTokenRepository(unittest.TestCase):
+
+    @patch('repository.fcm_token_repository.session_scope')
+    def test_get_fcm_token_success(self, mock_session_scope):
+        # Set up the mock session and query
+        mock_session = MagicMock()
+        mock_query = mock_session.query.return_value
+        mock_session_scope.return_value.__enter__.return_value = mock_session
+
+        # Create a mock FCMToken object
+        mock_fcm_token = FCMToken(
+            user_id=6,
+            fcm_token="mock_token_1",
+            device_type="android",
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+            is_active=True
+        )
+
+        # Simulate tokens found in the database
+        mock_query.filter_by.return_value.all.return_value = [mock_fcm_token]
+
+        # Instantiate the repository and call the method
+        fcm_token_repo = FCMTokenRepository()
+        token_list = fcm_token_repo.get_fcm_token(6)
+
+        # Assertions
+        mock_query.filter_by.assert_called_once_with(user_id=6)
+        self.assertEqual(token_list, ["mock_token_1"])
+
+    @patch('repository.fcm_token_repository.session_scope')
+    def test_get_fcm_token_no_tokens(self, mock_session_scope):
+        # Set up the mock session and query
+        mock_session = MagicMock()  # Mock session
+        mock_query = mock_session.query.return_value
+        mock_session_scope.return_value.__enter__.return_value = mock_session
+
+        # Simulate no tokens found in the database
+        mock_query.filter_by.return_value.all.return_value = []
+
+        # Instantiate the repository and call the method
+        fcm_token_repo = FCMTokenRepository()
+        token_list = fcm_token_repo.get_fcm_token(6)
+
+        # Assertions
+        mock_query.filter_by.assert_called_once_with(user_id=6)  # Ensure correct query
+        self.assertEqual(token_list, [])
+
+    @patch('firebase_admin.credentials.Certificate')
+    @patch('firebase_admin.initialize_app')
+    @patch('repository.fcm_token_repository.NotificationService.send_notification')
+    @patch('repository.fcm_token_repository.session_scope')
+    def test_notify_user_with_user_id(self, mock_session_scope, mock_send_notification, mock_initialize_app, mock_certificate):
+
+        # Mock the session and query
+        mock_session = MagicMock()
+        mock_query = mock_session.query.return_value
+        mock_session_scope.return_value.__enter__.return_value = mock_session
+
+        # Create a mock FCMToken object
+        mock_fcm_token = FCMToken(
+            user_id=6,
+            fcm_token="mock_token_1",
+            device_type="android",
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+            is_active=True
+        )
+
+        # Simulate tokens found in the database
+        mock_query.filter_by.return_value.all.return_value = [mock_fcm_token]
+
+        # Instantiate the repository and call notify_user with user_id
+        fcm_token_repo = FCMTokenRepository()
+        fcm_token_repo.notify_user(user_id=6, fcm_token_list=None, title="Test Title", body="Test Body")
+
+        # Assert get_fcm_token was called
+        mock_query.filter_by.assert_called_once_with(user_id=6)
+
+        # Assert send_notification was called with correct arguments
+        mock_send_notification.assert_called_once_with(
+            ['mock_token_1'], "Test Title", "Test Body", None
+        )
+
+    # Test case when both user_id and fcm_token_list are None
+    def test_notify_user_missing_arguments(self):
+        # Instantiate the repository
+        fcm_token_repo = FCMTokenRepository()
+
+        # Call notify_user without user_id or fcm_token_list
+        with self.assertRaises(ValueError) as context:
+            fcm_token_repo.notify_user(user_id=None, fcm_token_list=None, title="Test Title", body="Test Body")
+
+        # Assert that the correct exception is raised
+        self.assertEqual(str(context.exception), "Either user_id or fcm_token_list must be provided")
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/unit/test_notification.py

@@ -0,0 +1,69 @@
+import unittest
+import os
+from unittest.mock import patch
+from repository.fcm_token_repository import FCMTokenRepository
+from services.notification_service import NotificationService
+
+
+class TestNotificationService(unittest.TestCase):
+
+
+    @patch('firebase_admin.initialize_app')
+    @patch('firebase_admin.credentials.Certificate')
+    def test_credentials_loading(self, mock_certificate, mock_initialize_app):
+
+        # Mock the Certificate and initialize_app calls
+        mock_certificate.return_value = "mocked_credential"
+
+        # Calculate the expected path dynamically
+        expected_cred_path = f"{os.getcwd()}/google-credentials.json"
+
+        # Instantiate the NotificationService, which should trigger credentials loading
+        notification_service = NotificationService()
+
+        # Assert the credentials were loaded correctly with the dynamic path
+        mock_certificate.assert_called_once_with(expected_cred_path)
+        mock_initialize_app.assert_called_once_with("mocked_credential")
+
+    @patch('firebase_admin.initialize_app')
+    @patch('firebase_admin.credentials.Certificate')
+    @patch('firebase_admin.messaging.send')
+    @patch.object(FCMTokenRepository, 'get_fcm_token')
+    def test_notify_user_with_user_id(self, mock_get_fcm_token, mock_send, mock_certificate, mock_initialize_app):
+        # Mock the get_fcm_token method to return a list of FCM tokens
+        mock_get_fcm_token.return_value = ['mock_token_1', 'mock_token_2']
+
+        # Mock the NotificationService's send_notification method
+        mock_send.return_value = "mock_response"
+
+        # Mock the certificate loading to avoid FileNotFoundError
+        mock_certificate.return_value = "mocked_credential"
+        mock_initialize_app.return_value = None
+
+        # Instantiate the repository and call notify_user with a user_id
+        fcm_token_repo = FCMTokenRepository()
+        fcm_token_repo.notify_user(user_id=6, fcm_token_list=None, title="Test Title", body="Test Body")
+
+        # Check that get_fcm_token was called with the correct user_id
+        mock_get_fcm_token.assert_called_once_with(6)
+
+        # Define a helper function to check attributes of the messaging.Message
+        def message_matcher(expected_title, expected_body, expected_token):
+            def match(message):
+                return (message.notification.title == expected_title and
+                        message.notification.body == expected_body and
+                        message.token == expected_token)
+            return match
+
+        # Check that messaging.send was called for each token
+        self.assertTrue(any(message_matcher("Test Title", "Test Body", "mock_token_1")(call[0][0])
+                            for call in mock_send.call_args_list))
+        self.assertTrue(any(message_matcher("Test Title", "Test Body", "mock_token_2")(call[0][0])
+                            for call in mock_send.call_args_list))
+
+        # Ensure that messaging.send was called twice (once for each token)
+        self.assertEqual(mock_send.call_count, 2)
+
+
+if __name__ == '__main__':
+    unittest.main()