notification test

ktvmanager/lib/database.py

@@ -222,14 +222,18 @@ def save_push_subscription(user_id: int, subscription_json: str) -> None:
     _execute_query(query, params)
 
 
-def get_push_subscriptions(user_id: int) -> List[Dict[str, Any]]:
-    """Retrieves all push subscriptions for a given user ID.
+def get_push_subscriptions(user_id: Optional[int] = None) -> List[Dict[str, Any]]:
+    """Retrieves all push subscriptions for a given user ID, or all if no user_id is provided.
 
     Args:
-        user_id: The ID of the user.
+        user_id: The ID of the user (optional).
 
     Returns:
         A list of push subscriptions.
     """
-    query = "SELECT * FROM push_subscriptions WHERE user_id = %s"
-    return _execute_query(query, (user_id,))
+    if user_id:
+        query = "SELECT * FROM push_subscriptions WHERE user_id = %s"
+        return _execute_query(query, (user_id,))
+    else:
+        query = "SELECT * FROM push_subscriptions"
+        return _execute_query(query)

routes/api.py

@@ -223,16 +223,15 @@ def send_expiry_notifications_route(username: str, password: str) -> Response:
 @api_blueprint.route("/send-test-notification", methods=["POST"])
 @requires_basic_auth
 def send_test_notification_route(username: str, password: str) -> Response:
-    """Sends a test push notification to the user."""
-    user_id = get_user_id_from_username(username)
-    if not user_id:
-        return jsonify({"message": "User not found"}), 404
+    """Sends a test push notification to all users."""
+    data = request.get_json()
+    message = data.get("message", "Ktv Test") if data else "Ktv Test"
 
-    subscriptions = get_push_subscriptions(user_id)
+    subscriptions = get_push_subscriptions()  # Get all subscriptions
     if not subscriptions:
-        return jsonify({"message": "No push subscriptions found for this user."}), 404
+        return jsonify({"message": "No push subscriptions found."}), 404
 
-    message_body = json.dumps({"title": "Test Notification", "body": "This is a test notification."})
+    message_body = json.dumps({"title": "KTVManager", "body": message})
 
     for sub in subscriptions:
         try: