|
|
@ -409,4 +409,50 @@ class WebPushService(): |
|
|
|
return jsonify(success=False, message="Subscription not found"), 404 |
|
|
|
|
|
|
|
|
|
|
|
@staticmethod |
|
|
|
@app.route('/web-push/send-test', methods=['POST']) |
|
|
|
def send_test() -> Tuple[str, int]: |
|
|
|
""" |
|
|
|
Endpoint to send a test push notification to a specific client. |
|
|
|
|
|
|
|
This method: |
|
|
|
1. Retrieves the subscription information from the incoming request. |
|
|
|
2. Looks up the subscription in the database. |
|
|
|
3. Calls the _send_push_notification method to send a test push notification. |
|
|
|
4. Returns the result of the _send_push_notification call. |
|
|
|
|
|
|
|
URL: /web-push/send-test |
|
|
|
Method: POST |
|
|
|
|
|
|
|
Returns: |
|
|
|
- A JSON response with the result of the _send_push_notification call. |
|
|
|
|
|
|
|
Notes: |
|
|
|
- The incoming request should contain the parameters "endpoint", "p256dh", and "auth". |
|
|
|
""" |
|
|
|
|
|
|
|
# Retrieving the subscription information from the incoming request |
|
|
|
content = request.json |
|
|
|
endpoint = content['endpoint'] |
|
|
|
p256dh = content['p256dh'] |
|
|
|
auth = content['auth'] |
|
|
|
|
|
|
|
# Looking up the subscription in the database |
|
|
|
subscription = Subscription.query.filter_by(endpoint=endpoint, p256dh=p256dh, auth=auth).first() |
|
|
|
|
|
|
|
# If the subscription is found, call the _send_push_notification method |
|
|
|
if subscription: |
|
|
|
subscription_info = { |
|
|
|
"endpoint": subscription.endpoint, |
|
|
|
"keys": { |
|
|
|
"p256dh": subscription.p256dh, |
|
|
|
"auth": subscription.auth |
|
|
|
} |
|
|
|
} |
|
|
|
vapid_key = VAPIDKey.query.first() |
|
|
|
result = WebPushService._send_push_notification(subscription_info, {"title": "Test Notification", "message": "This is a test notification"}, vapid_key) |
|
|
|
return jsonify(result) |
|
|
|
else: |
|
|
|
return jsonify({"success": False, "message": "Subscription not found"}), 404 |
|
|
|
|
|
|
|
web_push_service = WebPushService(app, "app") |
|
|
|