Merge pull request #1190 from ObiFaith/fix/send-notification

fix: secure send_notification endpoint and ensure correct user association

Author: joboy-dev

api/v1/routes/notification.py

@@ -20,11 +20,14 @@
     status_code=status.HTTP_201_CREATED,
 )
 def send_notification(
-    notification_data: NotificationCreate, db: Session = Depends(get_db)
+    notification_data: NotificationCreate,
+    user: User = Depends(user_service.get_current_user),
+    db: Session = Depends(get_db)
 ):
     notification = notification_service.send_notification(
         title=notification_data.title,
         message=notification_data.message,
+        user=user,
         db=db,
     )
     return success_response(

api/v1/services/notification.py

@@ -10,10 +10,10 @@
 class NotificationService(Service):
 
     def send_notification(
-        self, title: str, message: str, db: Session = Depends(get_db)
+        self, title: str, message: str, user: User, db: Session = Depends(get_db)
     ):
         """Function to send a notification"""
-        new_notification = Notification(title=title, message=message, status="unread")
+        new_notification = Notification(user_id=user.id, title=title, message=message, status="unread")
         db.add(new_notification)
         db.commit()
         db.refresh(new_notification)

tests/v1/notification/test_notifications_service.py

@@ -8,6 +8,9 @@
 from api.db.database import get_db
 from api.utils.settings import settings
 import jwt
+from uuid_extensions import uuid7
+from api.v1.models.user import User
+from api.v1.services.user import user_service
 
 client = TestClient(app)
 
@@ -32,7 +35,21 @@ def create_test_token() -> str:
     return jwt.encode(data, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
 
 def test_send_notification(db_session_mock):
-    with patch("api.utils.dependencies.get_current_user", return_value=None):
+    user_id = uuid7()
+    user = User(
+        id=user_id,
+        email="[email protected]",
+        password=user_service.hash_password("Testpassword@123"),
+        first_name="Test",
+        last_name="User",
+        is_active=False,
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+    access_token = user_service.create_access_token(str(user_id))
+    headers = {"authorization": f"Bearer {access_token}"}
+
+    with patch("api.utils.dependencies.get_current_user", return_value=user):
         token = create_test_token()
 
         response = client.post(
@@ -41,16 +58,28 @@ def test_send_notification(db_session_mock):
                 "title": "Test Notification",
                 "message": "This is a test notification."
             },
-            headers={"Authorization": f"Bearer {token}"},
+            headers=headers,
         )
 
-    print(response.json())  # Debug print
     assert response.status_code == 201
     assert response.json()["message"] == "Notification sent successfully"
     assert response.json()["data"]["title"] == "Test Notification"
     assert response.json()["data"]["message"] == "This is a test notification."
     assert response.json()["data"]["status"] == "unread"
 
+def test_send_notification_unauthenticated_user(db_session_mock):
+    with patch("api.utils.dependencies.get_current_user", return_value=None):
+        response = client.post(
+            "/api/v1/notifications/send",
+            json={
+                "title": "Test Notification",
+                "message": "This is a test notification."
+            },
+        )
+
+    assert response.status_code == 401
+    assert response.json()["message"] == "Not authenticated"
+
 def test_get_notification_by_id(db_session_mock):
     notification = Notification(
         id="notification_id",