Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def send_push(msg, user_ids=None, extras=None, title=None):
'''
user_ids is a list of user_id [1, 2, ...]
if user_ids is None then send to all
'''
app_key = settings.JPUSH_APP_KEY
master_secret = settings.JPUSH_MASTER_SECRET
_jpush = jpush.JPush(app_key, master_secret)
push = _jpush.create_push()
ios_msg = jpush.ios(alert=msg, extras=extras)
android_msg = jpush.android(alert=msg, extras=extras, title=title)
push.notification = jpush.notification(
alert=msg, android=android_msg, ios=ios_msg)
push.platform = jpush.all_
# for ios dev or prd env
options = dict()
options['apns_production'] = settings.APNS_PRODUCTION
jpush.options(options)
if user_ids is None:
push.audience = jpush.all_
return str(push.send())
elif len(user_ids) > 1000:
ans = []
for i in range(len(user_ids) // 1000 + 1):
options['apns_production'] = settings.APNS_PRODUCTION
jpush.options(options)
if user_ids is None:
push.audience = jpush.all_
return str(push.send())
elif len(user_ids) > 1000:
ans = []
for i in range(len(user_ids) // 1000 + 1):
ret = send_push(msg, user_ids[i * 1000: (i + 1) * 1000])
ans.append(ret)
return ans
elif len(user_ids) == 0:
return ''
else:
push.audience = jpush.audience(
jpush.alias(*user_ids)
)
return str(push.send())
jpush.options(options)
if user_ids is None:
push.audience = jpush.all_
return str(push.send())
elif len(user_ids) > 1000:
ans = []
for i in range(len(user_ids) // 1000 + 1):
ret = send_push(msg, user_ids[i * 1000: (i + 1) * 1000])
ans.append(ret)
return ans
elif len(user_ids) == 0:
return ''
else:
push.audience = jpush.audience(
jpush.alias(*user_ids)
)
return str(push.send())
import unittest
from tests.conf import app_key, master_secret
from jpush import device
from jpush import common
import jpush as jpush
_jpush = jpush.JPush(app_key, master_secret)
device = _jpush.create_device()
_jpush.set_logging("DEBUG")
class TestEntity(unittest.TestCase):
def test_create_device(self):
reg_id = '1507bfd3f7c466c355c'
entity = jpush.device_tag(jpush.add("ddd", "tageee"))
result = device.set_devicemobile(reg_id, entity)
self.assertEqual(result.status_code, 200)
def test_aliasuser(self):
alias = "alias1"
platform = "android,ios"
result = device.get_aliasuser(alias, platform)
self.assertEqual(result.status_code, 200)
def test_simple_alert(self):
self.assertEqual(jpush.notification(alert='中文'), {'alert':'中文'})
def test_audience(self):
_jpush = jpush.JPush(app_key, master_secret)
push = _jpush.create_push()
push.audience = jpush.audience(
jpush.tag("tag1", "tag2"),
jpush.alias("alias1", "alias2")
)
push.notification = jpush.notification(alert="Hello world with audience!")
push.platform = jpush.all_
try:
response = push.send()
print response.status_code
self.assertEqual(response.status_code, 200)
except common.Unauthorized, e:
self.assertFalse(isinstance(e, common.Unauthorized))
raise common.Unauthorized("Unauthorized")
except common.APIConnectionException, e:
self.assertFalse(isinstance(e, common.APIConnectionException))
raise common.APIConnectionException("conn")
except common.JPushFailure, e:
self.assertFalse(isinstance(e, common.JPushFailure))
print "JPushFailure"
except:
self.assertFalse(1)
def test_audience(self):
_jpush = jpush.JPush(app_key, master_secret)
push = _jpush.create_push()
push.audience = jpush.audience(
jpush.tag("tag1", "tag2"),
jpush.alias("alias1", "alias2")
)
push.notification = jpush.notification(alert="Hello world with audience!")
push.platform = jpush.all_
try:
response = push.send()
print response.status_code
self.assertEqual(response.status_code, 200)
except common.Unauthorized, e:
self.assertFalse(isinstance(e, common.Unauthorized))
raise common.Unauthorized("Unauthorized")
except common.APIConnectionException, e:
self.assertFalse(isinstance(e, common.APIConnectionException))
raise common.APIConnectionException("conn")
except common.JPushFailure, e:
self.assertFalse(isinstance(e, common.JPushFailure))
print "JPushFailure"
except:
self.assertFalse(1)
print "Exception"
def test_get_schedule_by_invalid_id(self):
try:
result = report.get_users("DAY","2016-04-10","3")
self.assertEqual(result.status_code, 200)
except common.JPushFailure, e:
print e
self.assertIsInstance(e, common.JPushFailure)
def test_get_schedule_by_invalid_id(self):
try:
result = report.get_users("DAY","2016-04-10","3")
self.assertEqual(result.status_code, 200)
except common.JPushFailure, e:
print e
self.assertIsInstance(e, common.JPushFailure)
def test_clear_tag(self):
reg_id = '090c1f59f89'
entity = jpush.device_tag("")
try:
device.set_deviceinfo(reg_id, entity)
except common.JPushFailure:
self.assertEqual(1, 1)
except:
self.assertEqual(1, 0)