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_multiple_devices_data_message(push_service):
try:
push_service.multiple_devices_data_message(
data_message={"test": "Test"},
dry_run=True
)
assert False, "Should raise InvalidDataError without registration ids"
except errors.InvalidDataError:
pass
response = push_service.multiple_devices_data_message(
registration_ids=["Test"],
data_message={"test": "Test"},
dry_run=True
)
assert isinstance(response, dict)
if isinstance(body_loc_args, list):
fcm_payload['notification']['body_loc_args'] = body_loc_args
else:
raise InvalidDataError('body_loc_args should be an array')
# If title is present, use it
if message_title:
fcm_payload['notification']['title'] = message_title
# Else use title_loc_key and title_loc_args for title
else:
if title_loc_key:
fcm_payload['notification']['title_loc_key'] = title_loc_key
if title_loc_args:
if isinstance(title_loc_args, list):
fcm_payload['notification']['title_loc_args'] = title_loc_args
else:
raise InvalidDataError('title_loc_args should be an array')
if android_channel_id:
fcm_payload['notification']['android_channel_id'] = android_channel_id
# This is needed for iOS when we are sending only custom data messages
if content_available and isinstance(content_available, bool):
fcm_payload['content_available'] = content_available
if click_action:
fcm_payload['notification']['click_action'] = click_action
if isinstance(badge, int) and badge >= 0:
fcm_payload['notification']['badge'] = badge
if color:
fcm_payload['notification']['color'] = color
if tag:
fcm_payload['notification']['tag'] = tag
fcm_payload['to'] = '/topics/%s' % topic_name
# Revert to legacy API compatible priority
if low_priority:
fcm_payload['priority'] = self.FCM_LOW_PRIORITY
else:
fcm_payload['priority'] = self.FCM_HIGH_PRIORITY
if delay_while_idle:
fcm_payload['delay_while_idle'] = delay_while_idle
if collapse_key:
fcm_payload['collapse_key'] = collapse_key
if time_to_live is not None:
if isinstance(time_to_live, int):
fcm_payload['time_to_live'] = time_to_live
else:
raise InvalidDataError("Provided time_to_live is not an integer")
if restricted_package_name:
fcm_payload['restricted_package_name'] = restricted_package_name
if dry_run:
fcm_payload['dry_run'] = dry_run
if data_message:
if isinstance(data_message, dict):
fcm_payload['data'] = data_message
else:
raise InvalidDataError("Provided data_message is in the wrong format")
fcm_payload['notification'] = {}
if message_icon:
fcm_payload['notification']['icon'] = message_icon
# If body is present, use it
if message_body:
extra_notification_kwargs (dict, optional): More notification keyword arguments
extra_kwargs (dict, optional): More keyword arguments
Returns:
dict: Response from FCM server (`multicast_id`, `success`, `failure`, `canonical_ids`, `results`)
Raises:
AuthenticationError: If :attr:`api_key` is not set or provided
or there is an error authenticating the sender.
FCMServerError: Internal server error or timeout error on Firebase cloud messaging server
InvalidDataError: Invalid data provided
InternalPackageError: JSON parsing error, mostly from changes in the response of FCM,
create a new github issue to resolve it.
"""
if not isinstance(registration_ids, list):
raise InvalidDataError('Invalid registration IDs (should be list)')
payloads = []
registration_id_chunks = self.registration_id_chunks(registration_ids)
for registration_ids in registration_id_chunks:
# appends a payload with a chunk of registration ids here
payloads.append(self.parse_payload(
registration_ids=registration_ids,
condition=condition,
collapse_key=collapse_key,
delay_while_idle=delay_while_idle,
time_to_live=time_to_live,
restricted_package_name=restricted_package_name,
low_priority=low_priority,
dry_run=dry_run,
data_message=data_message,
content_available=content_available,
extra_notification_kwargs (dict, optional): More notification keyword arguments
extra_kwargs (dict, optional): More keyword arguments
Returns:
dict: Response from FCM server (`multicast_id`, `success`, `failure`, `canonical_ids`, `results`)
Raises:
AuthenticationError: If :attr:`api_key` is not set or provided
or there is an error authenticating the sender.
FCMServerError: Internal server error or timeout error on Firebase cloud messaging server
InvalidDataError: Invalid data provided
InternalPackageError: Mostly from changes in the response of FCM,
contact the project owner to resolve the issue
"""
if registration_id is None:
raise InvalidDataError('Invalid registration ID')
# [registration_id] cos we're sending to a single device
payload = self.parse_payload(
registration_ids=[registration_id],
condition=condition,
collapse_key=collapse_key,
delay_while_idle=delay_while_idle,
time_to_live=time_to_live,
restricted_package_name=restricted_package_name,
low_priority=low_priority,
dry_run=dry_run,
data_message=data_message,
content_available=content_available,
remove_notification=True,
android_channel_id=android_channel_id,
extra_notification_kwargs=extra_notification_kwargs,
**extra_kwargs
results = parsed_response.get('results', [])
message_id = parsed_response.get('message_id', None) # for topic messages
if message_id:
success = 1
if multicast_id:
response_dict['multicast_ids'].append(multicast_id)
response_dict['success'] += success
response_dict['failure'] += failure
response_dict['canonical_ids'] += canonical_ids
response_dict['results'].extend(results)
response_dict['topic_message_id'] = message_id
elif response.status_code == 401:
raise AuthenticationError("There was an error authenticating the sender account")
elif response.status_code == 400:
raise InvalidDataError(response.text)
elif response.status_code == 404:
raise FCMNotRegisteredError("Token not registered")
else:
raise FCMServerError("FCM server is temporarily unavailable")
return response_dict
fcm_payload['notification'] = {}
if message_icon:
fcm_payload['notification']['icon'] = message_icon
# If body is present, use it
if message_body:
fcm_payload['notification']['body'] = message_body
# Else use body_loc_key and body_loc_args for body
else:
if body_loc_key:
fcm_payload['notification']['body_loc_key'] = body_loc_key
if body_loc_args:
if isinstance(body_loc_args, list):
fcm_payload['notification']['body_loc_args'] = body_loc_args
else:
raise InvalidDataError('body_loc_args should be an array')
# If title is present, use it
if message_title:
fcm_payload['notification']['title'] = message_title
# Else use title_loc_key and title_loc_args for title
else:
if title_loc_key:
fcm_payload['notification']['title_loc_key'] = title_loc_key
if title_loc_args:
if isinstance(title_loc_args, list):
fcm_payload['notification']['title_loc_args'] = title_loc_args
else:
raise InvalidDataError('title_loc_args should be an array')
if android_channel_id:
fcm_payload['notification']['android_channel_id'] = android_channel_id
extra_notification_kwargs (dict, optional): More notification keyword arguments
extra_kwargs (dict, optional): More keyword arguments
Returns:
dict: Response from FCM server (`multicast_id`, `success`, `failure`, `canonical_ids`, `results`)
Raises:
AuthenticationError: If :attr:`api_key` is not set or provided
or there is an error authenticating the sender.
FCMServerError: Internal server error or timeout error on Firebase cloud messaging server
InvalidDataError: Invalid data provided
InternalPackageError: JSON parsing error, mostly from changes in the response of FCM,
create a new github issue to resolve it.
"""
if not isinstance(registration_ids, list):
raise InvalidDataError('Invalid registration IDs (should be list)')
payloads = []
registration_id_chunks = self.registration_id_chunks(registration_ids)
for registration_ids in registration_id_chunks:
# appends a payload with a chunk of registration ids here
payloads.append(self.parse_payload(
registration_ids=registration_ids,
message_body=message_body,
message_title=message_title,
message_icon=message_icon,
sound=sound,
condition=condition,
collapse_key=collapse_key,
delay_while_idle=delay_while_idle,
time_to_live=time_to_live,
Raises:
InvalidDataError: data sent to server was incorrectly formatted
FCMError: an error occured on the server
"""
url = 'https://iid.googleapis.com/iid/v1:batchAdd'
payload = {
'to': '/topics/' + topic_name,
'registration_tokens': registration_ids,
}
response = self.requests_session.post(url, json=payload)
if response.status_code == 200:
return True
elif response.status_code == 400:
error = response.json()
raise InvalidDataError(error['error'])
else:
raise FCMError()
extra_notification_kwargs (dict, optional): More notification keyword arguments
extra_kwargs (dict, optional): More keyword arguments
Returns:
dict: Response from FCM server (`multicast_id`, `success`, `failure`, `canonical_ids`, `results`)
Raises:
AuthenticationError: If :attr:`api_key` is not set or provided
or there is an error authenticating the sender.
FCMServerError: Internal server error or timeout error on Firebase cloud messaging server
InvalidDataError: Invalid data provided
InternalPackageError: Mostly from changes in the response of FCM,
contact the project owner to resolve the issue
"""
if registration_id is None:
raise InvalidDataError('Invalid registration ID')
# [registration_id] cos we're sending to a single device
payload = self.parse_payload(
registration_ids=[registration_id],
message_body=message_body,
message_title=message_title,
message_icon=message_icon,
sound=sound,
condition=condition,
collapse_key=collapse_key,
delay_while_idle=delay_while_idle,
time_to_live=time_to_live,
restricted_package_name=restricted_package_name,
low_priority=low_priority,
dry_run=dry_run, data_message=data_message, click_action=click_action,
badge=badge,
color=color,