from unittest import mock
import pytest
+import pywebpush
+import requests
from .utils import create_user, create_user_and_song
pushmock.assert_not_called()
-# notification for activity when enabled
-# no notification for activity when disabled
+@mock.patch("pywebpush.webpush")
+def test_notification_for_comments_enabled(pushmock, app, client, user, subid, vapid_keys):
+ _enable_notifications(client, subid, comments=True, songs=False)
+
+ # user2 comments on user's profile
+ create_user(client, "user2", login=True)
+ response = client.get("/comment?threadid=1", headers={"Referer": "/users/user1"})
+ response = client.post("/comment?threadid=1", data={"content": "comment on profile"}, follow_redirects=True)
+ lsp.push_notifications.wait_all()
-# notification for jam when enabled
-# no notification for jam when disabled
+ # user receives notification
+ pushmock.assert_called_once_with(
+ json.loads(SUBDATA),
+ '{"title": "Comment from user2", "body": "comment on profile", "url": "/activity"}',
+ vapid_private_key="vapid-private-key",
+ vapid_claims={"sub": "mailto:littlesongplace@gmail.com"})
+@mock.patch("pywebpush.webpush")
+def test_notification_for_comments_disabled(pushmock, app, client, user, subid, vapid_keys):
+ _enable_notifications(client, subid, comments=False, songs=False)
+ create_user(client, "user2", login=True)
+ response = client.get("/comment?threadid=1", headers={"Referer": "/users/user1"})
+ response = client.post("/comment?threadid=1", data={"content": "comment on profile"}, follow_redirects=True)
+ lsp.push_notifications.wait_all()
+ pushmock.assert_not_called()
################################################################################
# Subscription expiry
-# @mock.patch("pywebpush.webpush")
-# def test_subscription_deleted_on_410(client, user, subid):
-# ...
+@mock.patch("pywebpush.webpush")
+def test_subscription_deleted_on_410(pushmock, app, client, user, subid, vapid_keys):
+ _enable_notifications(client, subid, comments=False, songs=True)
+ create_user_and_song(client, "user2")
+
+ with app.app_context():
+ # Works first time
+ lsp.push_notifications.notify_new_songs_cmd([], standalone_mode=False)
+ lsp.push_notifications.wait_all()
+ pushmock.assert_called_once()
+ pushmock.reset_mock()
+
+ # Called second time, raises 410 Gone response
+ fail_response = requests.Response()
+ fail_response.status_code = 410
+ pushmock.side_effect = pywebpush.WebPushException("Push failed", fail_response)
+
+ lsp.push_notifications.notify_new_songs_cmd([], standalone_mode=False)
+ lsp.push_notifications.wait_all()
+ pushmock.assert_called_once()
+ pushmock.reset_mock()
+
+ # After 410 response, is not called again (subscription deleted)
+ lsp.push_notifications.notify_new_songs_cmd([], standalone_mode=False)
+ lsp.push_notifications.wait_all()
+ pushmock.assert_not_called()
+
+@mock.patch("pywebpush.webpush")
+def test_subscription_not_deleted_on_500(pushmock, app, client, user, subid, vapid_keys):
+ _enable_notifications(client, subid, comments=False, songs=True)
+ create_user_and_song(client, "user2")
+
+ with app.app_context():
+ # Works first time
+ lsp.push_notifications.notify_new_songs_cmd([], standalone_mode=False)
+ lsp.push_notifications.wait_all()
+ pushmock.assert_called_once()
+ pushmock.reset_mock()
+
+ # Called second time, raises 500 Internal Server Error response
+ fail_response = requests.Response()
+ fail_response.status_code = 500
+ pushmock.side_effect = pywebpush.WebPushException("Push failed", fail_response)
+
+ lsp.push_notifications.notify_new_songs_cmd([], standalone_mode=False)
+ lsp.push_notifications.wait_all()
+ pushmock.assert_called_once()
+ pushmock.reset_mock()
+
+ # After 500 response, is still called again (subscription NOT deleted)
+ lsp.push_notifications.notify_new_songs_cmd([], standalone_mode=False)
+ lsp.push_notifications.wait_all()
+ pushmock.assert_called_once()
-# subscription deleted on 410
-# subscription not deleted on 500