Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_init_baseapi():
try:
BaseAPI()
assert False, "Should raise AuthenticationError"
except errors.AuthenticationError:
pass
def test_push_service_without_credentials():
try:
FCMNotification()
assert False, "Should raise AuthenticationError without credentials"
except errors.AuthenticationError:
pass
def test_notify_topic_subscribers(push_service):
try:
push_service.notify_topic_subscribers(
message_body="Test",
dry_run=True
)
assert False, "Should raise InvalidDataError without topic"
except errors.InvalidDataError:
pass
response = push_service.notify_topic_subscribers(
topic_name="test",
message_body="Test",
message_title="Test",
dry_run=True
)
assert isinstance(response, dict)
elif 'encryption_key' in notification.headers:
data['enckey'] = notification.headers['encryption_key']
# registration_ids are the FCM instance tokens (specified during
# registration.
router_ttl = min(self.MAX_TTL,
max(self.min_ttl, notification.ttl or 0))
try:
result = self.fcm.notify_single_device(
collapse_key=self.collapseKey,
data_message=data,
dry_run=self.dryRun or ('dryrun' in router_data),
registration_id=regid,
time_to_live=router_ttl,
)
except pyfcm.errors.AuthenticationError as e:
self.log.error("Authentication Error: %s" % e)
raise RouterException("Server error", status_code=500)
except ConnectionError as e:
self.metrics.increment("notification.bridge.error",
tags=make_tags(
self._base_tags,
reason="connection_unavailable"))
self.log.warn("Could not connect to FCM server: %s" % e)
raise RouterException("Server error", status_code=502,
log_exception=False)
except Exception as e:
self.log.error("Unhandled FCM Error: %s" % e)
raise RouterException("Server error", status_code=500)
return self._process_reply(result, notification, router_data,
ttl=router_ttl)