paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-11-04 05:53:43 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
71edcb649054477d40a319f13d069c8090d87c67
zulip/zerver/tests/test_push_notifications.py
Umair Khan c24fbd28b2 Shift to python-gcm for Android push notifications. 
Fixes #2650
2016-12-15 12:17:07 -08:00

450 lines
17 KiB
Python
Raw Blame History

import mock
from mock import call
import time
from typing import Any, Union, SupportsInt, Text
import gcm
from django.test import TestCase
from django.conf import settings
from zerver.models import PushDeviceToken, UserProfile, Message
from zerver.models import get_user_profile_by_email, receives_online_notifications, \
receives_offline_notifications
from zerver.lib import push_notifications as apn
from zerver.lib.test_classes import (
ZulipTestCase,
)
class MockRedis(object):
data = {} # type: Dict[str, Any]
def hgetall(self, key):
# type: (str) -> Any
return self.data.get(key)
def exists(self, key):
# type: (str) -> bool
return key in self.data
def hmset(self, key, data):
# type: (str, Dict[Any, Any]) -> None
self.data[key] = data
def delete(self, key):
# type: (str) -> None
if self.exists(key):
del self.data[key]
def expire(self, *args, **kwargs):
# type: (Any, Any) -> None
pass
class PushNotificationTest(TestCase):
def setUp(self):
# type: () -> None
email = 'hamlet@zulip.com'
apn.connection = apn.get_connection('fake-cert', 'fake-key')
self.redis_client = apn.redis_client = MockRedis() # type: ignore
apn.dbx_connection = apn.get_connection('fake-cert', 'fake-key')
self.user_profile = get_user_profile_by_email(email)
self.tokens = [u'aaaa', u'bbbb']
for token in self.tokens:
PushDeviceToken.objects.create(
kind=PushDeviceToken.APNS,
token=apn.hex_to_b64(token),
user=self.user_profile,
ios_app_id=settings.ZULIP_IOS_APP_ID)
def tearDown(self):
# type: () -> None
for i in [100, 200]:
self.redis_client.delete(apn.get_apns_key(i))
class APNsMessageTest(PushNotificationTest):
@mock.patch('random.getrandbits', side_effect=[100, 200])
def test_apns_message(self, mock_getrandbits):
# type: (mock.MagicMock) -> None
apn.APNsMessage(self.user_profile, self.tokens, alert="test")
data = self.redis_client.hgetall(apn.get_apns_key(100))
self.assertEqual(data['token'], 'aaaa')
self.assertEqual(int(data['user_id']), self.user_profile.id)
data = self.redis_client.hgetall(apn.get_apns_key(200))
self.assertEqual(data['token'], 'bbbb')
self.assertEqual(int(data['user_id']), self.user_profile.id)
class ResponseListenerTest(PushNotificationTest):
def get_error_response(self, **kwargs):
# type: (Any) -> Dict[str, SupportsInt]
er = {'identifier': 0, 'status': 0} # type: Dict[str, SupportsInt]
er.update({k: v for k, v in kwargs.items() if k in er})
return er
def get_cache_value(self):
# type: () -> Dict[str, Union[str, int]]
return {'token': 'aaaa', 'user_id': self.user_profile.id}
@mock.patch('logging.warn')
def test_cache_does_not_exist(self, mock_warn):
# type: (mock.MagicMock) -> None
err_rsp = self.get_error_response(identifier=100, status=1)
apn.response_listener(err_rsp)
msg = "APNs key, apns:100, doesn't not exist."
mock_warn.assert_called_once_with(msg)
@mock.patch('logging.warn')
def test_cache_exists(self, mock_warn):
# type: (mock.MagicMock) -> None
self.redis_client.hmset(apn.get_apns_key(100), self.get_cache_value())
err_rsp = self.get_error_response(identifier=100, status=1)
apn.response_listener(err_rsp)
b64_token = apn.hex_to_b64('aaaa')
errmsg = apn.ERROR_CODES[int(err_rsp['status'])]
msg = ("APNS: Failed to deliver APNS notification to %s, "
"reason: %s" % (b64_token, errmsg))
mock_warn.assert_called_once_with(msg)
@mock.patch('logging.warn')
def test_error_code_eight(self, mock_warn):
# type: (mock.MagicMock) -> None
self.redis_client.hmset(apn.get_apns_key(100), self.get_cache_value())
err_rsp = self.get_error_response(identifier=100, status=8)
b64_token = apn.hex_to_b64('aaaa')
self.assertEqual(PushDeviceToken.objects.filter(
user=self.user_profile, token=b64_token).count(), 1)
apn.response_listener(err_rsp)
self.assertEqual(mock_warn.call_count, 2)
self.assertEqual(PushDeviceToken.objects.filter(
user=self.user_profile, token=b64_token).count(), 0)
class TestPushApi(ZulipTestCase):
def test_push_api(self):
# type: () -> None
email = "cordelia@zulip.com"
user = get_user_profile_by_email(email)
self.login(email)
endpoints = [
('/json/users/me/apns_device_token', 'apple-token'),
('/json/users/me/android_gcm_reg_id', 'android-token'),
]
# Test error handling
for endpoint, _ in endpoints:
# Try adding/removing tokens that are too big...
broken_token = "x" * 5000 # too big
result = self.client_post(endpoint, {'token': broken_token})
self.assert_json_error(result, 'Empty or invalid length token')
result = self.client_delete(endpoint, {'token': broken_token})
self.assert_json_error(result, 'Empty or invalid length token')
# Try to remove a non-existent token...
result = self.client_delete(endpoint, {'token': 'non-existent token'})
self.assert_json_error(result, 'Token does not exist')
# Add tokens
for endpoint, token in endpoints:
# Test that we can push twice
result = self.client_post(endpoint, {'token': token})
self.assert_json_success(result)
result = self.client_post(endpoint, {'token': token})
self.assert_json_success(result)
tokens = list(PushDeviceToken.objects.filter(user=user, token=token))
self.assertEqual(len(tokens), 1)
self.assertEqual(tokens[0].token, token)
# User should have tokens for both devices now.
tokens = list(PushDeviceToken.objects.filter(user=user))
self.assertEqual(len(tokens), 2)
# Remove tokens
for endpoint, token in endpoints:
result = self.client_delete(endpoint, {'token': token})
self.assert_json_success(result)
tokens = list(PushDeviceToken.objects.filter(user=user, token=token))
self.assertEqual(len(tokens), 0)
class SendNotificationTest(PushNotificationTest):
@mock.patch('logging.warn')
@mock.patch('logging.info')
@mock.patch('zerver.lib.push_notifications._do_push_to_apns_service')
def test_send_apple_push_notifiction(self, mock_send, mock_info, mock_warn):
# type: (mock.MagicMock, mock.MagicMock, mock.MagicMock) -> None
def test_send(user, message, alert):
# type: (UserProfile, Message, str) -> None
self.assertEqual(user.id, self.user_profile.id)
self.assertEqual(set(message.tokens), set(self.tokens))
mock_send.side_effect = test_send
apn.send_apple_push_notification(self.user_profile, "test alert")
self.assertEqual(mock_send.call_count, 1)
@mock.patch('apns.GatewayConnection.send_notification_multiple')
def test_do_push_to_apns_service(self, mock_push):
# type: (mock.MagicMock) -> None
msg = apn.APNsMessage(self.user_profile, self.tokens, alert="test")
def test_push(message):
# type: (Message) -> None
self.assertIs(message, msg.get_frame())
mock_push.side_effect = test_push
apn._do_push_to_apns_service(self.user_profile, msg, apn.connection)
@mock.patch('logging.warn')
@mock.patch('logging.info')
@mock.patch('apns.GatewayConnection.send_notification_multiple')
def test_connection_single_none(self, mock_push, mock_info, mock_warn):
# type: (mock.MagicMock, mock.MagicMock, mock.MagicMock) -> None
apn.connection = None
apn.send_apple_push_notification(self.user_profile, "test alert")
@mock.patch('logging.error')
@mock.patch('apns.GatewayConnection.send_notification_multiple')
def test_connection_both_none(self, mock_push, mock_error):
# type: (mock.MagicMock, mock.MagicMock) -> None
apn.connection = None
apn.dbx_connection = None
apn.send_apple_push_notification(self.user_profile, "test alert")
class APNsFeedbackTest(PushNotificationTest):
@mock.patch('logging.info')
@mock.patch('apns.FeedbackConnection.items')
def test_feedback(self, mock_items, mock_info):
# type: (mock.MagicMock, mock.MagicMock) -> None
update_time = apn.timestamp_to_datetime(int(time.time()) - 10000)
PushDeviceToken.objects.all().update(last_updated=update_time)
mock_items.return_value = [
('aaaa', int(time.time())),
]
self.assertEqual(PushDeviceToken.objects.all().count(), 2)
apn.check_apns_feedback()
self.assertEqual(PushDeviceToken.objects.all().count(), 1)
class GCMTest(PushNotificationTest):
def setUp(self):
# type: () -> None
super(GCMTest, self).setUp()
apn.gcm = gcm.GCM('fake key')
self.gcm_tokens = [u'1111', u'2222']
for token in self.gcm_tokens:
PushDeviceToken.objects.create(
kind=PushDeviceToken.GCM,
token=apn.hex_to_b64(token),
user=self.user_profile,
ios_app_id=None)
def get_gcm_data(self, **kwargs):
# type: (**Any) -> Dict[str, Any]
data = {
'key 1': 'Data 1',
'key 2': 'Data 2',
}
data.update(kwargs)
return data
class GCMNotSetTest(GCMTest):
@mock.patch('logging.error')
def test_gcm_is_none(self, mock_error):
# type: (mock.MagicMock) -> None
apn.gcm = None
apn.send_android_push_notification(self.user_profile, {})
mock_error.assert_called_with("Attempting to send a GCM push "
"notification, but no API key was "
"configured")
class GCMSuccessTest(GCMTest):
@mock.patch('logging.warning')
@mock.patch('logging.info')
@mock.patch('gcm.GCM.json_request')
def test_success(self, mock_send, mock_info, mock_warning):
# type: (mock.MagicMock, mock.MagicMock, mock.MagicMock) -> None
res = {}
res['success'] = {token: ind for ind, token in enumerate(self.gcm_tokens)}
mock_send.return_value = res
data = self.get_gcm_data()
apn.send_android_push_notification(self.user_profile, data)
self.assertEqual(mock_info.call_count, 2)
c1 = call("GCM: Sent 1111 as 0")
c2 = call("GCM: Sent 2222 as 1")
mock_info.assert_has_calls([c1, c2], any_order=True)
mock_warning.assert_not_called()
class GCMCanonicalTest(GCMTest):
@mock.patch('logging.warning')
@mock.patch('gcm.GCM.json_request')
def test_equal(self, mock_send, mock_warning):
# type: (mock.MagicMock, mock.MagicMock) -> None
res = {}
res['canonical'] = {1: 1}
mock_send.return_value = res
data = self.get_gcm_data()
apn.send_android_push_notification(self.user_profile, data)
mock_warning.assert_called_once_with("GCM: Got canonical ref but it "
"already matches our ID 1!")
@mock.patch('logging.warning')
@mock.patch('gcm.GCM.json_request')
def test_pushdevice_not_present(self, mock_send, mock_warning):
# type: (mock.MagicMock, mock.MagicMock) -> None
res = {}
t1 = apn.hex_to_b64(u'1111')
t2 = apn.hex_to_b64(u'3333')
res['canonical'] = {t1: t2}
mock_send.return_value = res
def get_count(hex_token):
# type: (Text) -> int
token = apn.hex_to_b64(hex_token)
return PushDeviceToken.objects.filter(
token=token, kind=PushDeviceToken.GCM).count()
self.assertEqual(get_count(u'1111'), 1)
self.assertEqual(get_count(u'3333'), 0)
data = self.get_gcm_data()
apn.send_android_push_notification(self.user_profile, data)
msg = ("GCM: Got canonical ref %s "
"replacing %s but new ID not "
"registered! Updating.")
mock_warning.assert_called_once_with(msg % (t2, t1))
self.assertEqual(get_count(u'1111'), 0)
self.assertEqual(get_count(u'3333'), 1)
@mock.patch('logging.info')
@mock.patch('gcm.GCM.json_request')
def test_pushdevice_different(self, mock_send, mock_info):
# type: (mock.MagicMock, mock.MagicMock) -> None
res = {}
old_token = apn.hex_to_b64(u'1111')
new_token = apn.hex_to_b64(u'2222')
res['canonical'] = {old_token: new_token}
mock_send.return_value = res
def get_count(hex_token):
# type: (Text) -> int
token = apn.hex_to_b64(hex_token)
return PushDeviceToken.objects.filter(
token=token, kind=PushDeviceToken.GCM).count()
self.assertEqual(get_count(u'1111'), 1)
self.assertEqual(get_count(u'2222'), 1)
data = self.get_gcm_data()
apn.send_android_push_notification(self.user_profile, data)
mock_info.assert_called_once_with(
"GCM: Got canonical ref %s, dropping %s" % (new_token, old_token))
self.assertEqual(get_count(u'1111'), 0)
self.assertEqual(get_count(u'2222'), 1)
class GCMNotRegisteredTest(GCMTest):
@mock.patch('logging.info')
@mock.patch('gcm.GCM.json_request')
def test_not_registered(self, mock_send, mock_info):
# type: (mock.MagicMock, mock.MagicMock) -> None
res = {}
token = apn.hex_to_b64(u'1111')
res['errors'] = {'NotRegistered': [token]}
mock_send.return_value = res
def get_count(hex_token):
# type: (Text) -> int
token = apn.hex_to_b64(hex_token)
return PushDeviceToken.objects.filter(
token=token, kind=PushDeviceToken.GCM).count()
self.assertEqual(get_count(u'1111'), 1)
data = self.get_gcm_data()
apn.send_android_push_notification(self.user_profile, data)
mock_info.assert_called_once_with("GCM: Removing %s" % (token,))
self.assertEqual(get_count(u'1111'), 0)
class GCMFailureTest(GCMTest):
@mock.patch('logging.warning')
@mock.patch('gcm.GCM.json_request')
def test_failure(self, mock_send, mock_warn):
# type: (mock.MagicMock, mock.MagicMock) -> None
res = {}
token = apn.hex_to_b64(u'1111')
res['errors'] = {'Failed': [token]}
mock_send.return_value = res
data = self.get_gcm_data()
apn.send_android_push_notification(self.user_profile, data)
c1 = call("GCM: Delivery to %s failed: Failed" % (token,))
mock_warn.assert_has_calls([c1], any_order=True)
class TestReceivesNotificationsFunctions(ZulipTestCase):
def setUp(self):
# type: () -> None
email = "cordelia@zulip.com"
self.user = get_user_profile_by_email(email)
def test_receivers_online_notifications_when_user_is_a_bot(self):
# type: () -> None
self.user.is_bot = True
self.user.enable_online_push_notifications = True
self.assertFalse(receives_online_notifications(self.user))
self.user.enable_online_push_notifications = False
self.assertFalse(receives_online_notifications(self.user))
def test_receivers_online_notifications_when_user_is_not_a_bot(self):
# type: () -> None
self.user.is_bot = False
self.user.enable_online_push_notifications = True
self.assertTrue(receives_online_notifications(self.user))
self.user.enable_online_push_notifications = False
self.assertFalse(receives_online_notifications(self.user))
def test_receivers_offline_notifications_when_user_is_a_bot(self):
# type: () -> None
self.user.is_bot = True
self.user.enable_offline_email_notifications = True
self.user.enable_offline_push_notifications = True
self.assertFalse(receives_offline_notifications(self.user))
self.user.enable_offline_email_notifications = False
self.user.enable_offline_push_notifications = False
self.assertFalse(receives_offline_notifications(self.user))
self.user.enable_offline_email_notifications = True
self.user.enable_offline_push_notifications = False
self.assertFalse(receives_offline_notifications(self.user))
self.user.enable_offline_email_notifications = False
self.user.enable_offline_push_notifications = True
self.assertFalse(receives_offline_notifications(self.user))
def test_receivers_offline_notifications_when_user_is_not_a_bot(self):
# type: () -> None
self.user.is_bot = False
self.user.enable_offline_email_notifications = True
self.user.enable_offline_push_notifications = True
self.assertTrue(receives_offline_notifications(self.user))
self.user.enable_offline_email_notifications = False
self.user.enable_offline_push_notifications = False
self.assertFalse(receives_offline_notifications(self.user))
self.user.enable_offline_email_notifications = True
self.user.enable_offline_push_notifications = False
self.assertTrue(receives_offline_notifications(self.user))
self.user.enable_offline_email_notifications = False
self.user.enable_offline_push_notifications = True
self.assertTrue(receives_offline_notifications(self.user))
Reference in New Issue View Git Blame Copy Permalink