extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-10-05 01:32:08 +02:00
Code Issues Packages Projects Releases Wiki Activity

FCM Plugin Supports more Payload Options (#489)

Browse Source
This commit is contained in:
Chris Caron 2022-01-25 18:19:24 -05:00 committed by GitHub
parent 6fbe23832b
commit 5e2a293195
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 992 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
apprise/plugins/NotifyFCM/__init__.py
View File

@ -52,8 +52,13 @@ from ..NotifyBase import NotifyBase
from ...common import NotifyType
from ...utils import validate_regex
from ...utils import parse_list
from ...utils import parse_bool
from ...common import NotifyImageSize
from ...AppriseAttachment import AppriseAttachment
from ...AppriseLocale import gettext_lazy as _
from .common import (FCMMode, FCM_MODES)
from .priority import (FCM_PRIORITIES, FCMPriorityManager)
from .color import FCMColorManager
# Default our global support flag
NOTIFY_FCM_SUPPORT_ENABLED = False
@ -80,26 +85,6 @@ FCM_HTTP_ERROR_MAP = {
}
class FCMMode(object):
"""
Define the Firebase Cloud Messaging Modes
"""
# The legacy way of sending a message
Legacy = "legacy"
# The new API
OAuth2 = "oauth2"
# FCM Modes
FCM_MODES = (
# Legacy API
FCMMode.Legacy,
# HTTP v1 URL
FCMMode.OAuth2,
)
class NotifyFCM(NotifyBase):
"""
A wrapper for Google's Firebase Cloud Messaging Notifications
@ -136,13 +121,12 @@ class NotifyFCM(NotifyBase):
# If it is more than this, then it is not accepted.
max_fcm_keyfile_size = 5000
# Allows the user to specify the NotifyImageSize object
image_size = NotifyImageSize.XY_256
# The maximum length of the body
body_maxlen = 1024
# A title can not be used for SMS Messages. Setting this to zero will
# cause any title (if defined) to get placed into the message body.
title_maxlen = 0
# Define object templates
templates = (
# OAuth2
@ -163,12 +147,6 @@ class NotifyFCM(NotifyBase):
'type': 'string',
'private': True,
},
'mode': {
'name': _('Mode'),
'type': 'choice:string',
'values': FCM_MODES,
'default': FCMMode.Legacy,
},
'project': {
'name': _('Project ID'),
'type': 'string',
@ -195,10 +173,47 @@ class NotifyFCM(NotifyBase):
'to': {
'alias_of': 'targets',
},
'mode': {
'name': _('Mode'),
'type': 'choice:string',
'values': FCM_MODES,
'default': FCMMode.Legacy,
},
'priority': {
'name': _('Mode'),
'type': 'choice:string',
'values': FCM_PRIORITIES,
},
'image_url': {
'name': _('Custom Image URL'),
'type': 'string',
},
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': False,
'map_to': 'include_image',
},
# Color can either be yes, no, or a #rrggbb (
# rrggbb without hashtag is accepted to)
'color': {
'name': _('Notification Color'),
'type': 'string',
'default': 'yes',
},
})
# Define our data entry
template_kwargs = {
'data_kwargs': {
'name': _('Data Entries'),
'prefix': '+',
},
}
def __init__(self, project, apikey, targets=None, mode=None, keyfile=None,
**kwargs):
data_kwargs=None, image_url=None, include_image=False,
color=None, priority=None, **kwargs):
"""
Initialize Firebase Cloud Messaging
@ -214,7 +229,7 @@ class NotifyFCM(NotifyBase):
self.mode = NotifyFCM.template_tokens['mode']['default'] \
if not isinstance(mode, six.string_types) else mode.lower()
if self.mode and self.mode not in FCM_MODES:
msg = 'The mode specified ({}) is invalid.'.format(mode)
msg = 'The FCM mode specified ({}) is invalid.'.format(mode)
self.logger.warning(msg)
raise TypeError(msg)
@ -267,6 +282,29 @@ class NotifyFCM(NotifyBase):
# Acquire Device IDs to notify
self.targets = parse_list(targets)
# Our data Keyword/Arguments to include in our outbound payload
self.data_kwargs = {}
if isinstance(data_kwargs, dict):
self.data_kwargs.update(data_kwargs)
# Include the image as part of the payload
self.include_image = include_image
# A Custom Image URL
# FCM allows you to provide a remote https?:// URL to an image_url
# located on the internet that it will download and include in the
# payload.
#
# self.image_url() is reserved as an internal function name; so we
# jsut store it into a different variable for now
self.image_src = image_url
# Initialize our priority
self.priority = FCMPriorityManager(self.mode, priority)
# Initialize our color
self.color = FCMColorManager(color, asset=self.asset)
return
@property
@ -335,6 +373,10 @@ class NotifyFCM(NotifyBase):
# Prepare our notify URL
notify_url = self.notify_legacy_url
# Acquire image url
image = self.image_url(notify_type) \
if not self.image_src else self.image_src
has_error = False
# Create a copy of the targets list
targets = list(self.targets)
@ -352,6 +394,17 @@ class NotifyFCM(NotifyBase):
}
}
if self.color:
# Acquire our color
payload['message']['android'] = {
'notification': {'color': self.color.get(notify_type)}}
if self.include_image and image:
payload['message']['notification']['image'] = image
if self.data_kwargs:
payload['message']['data'] = self.data_kwargs
if recipient[0] == '#':
payload['message']['topic'] = recipient[1:]
self.logger.debug(
@ -373,6 +426,18 @@ class NotifyFCM(NotifyBase):
}
}
}
if self.color:
# Acquire our color
payload['notification']['notification']['color'] = \
self.color.get(notify_type)
if self.include_image and image:
payload['notification']['notification']['image'] = image
if self.data_kwargs:
payload['data'] = self.data_kwargs
if recipient[0] == '#':
payload['to'] = '/topics/{}'.format(recipient)
self.logger.debug(
@ -385,6 +450,18 @@ class NotifyFCM(NotifyBase):
"FCM recipient %s parsed as a device token",
recipient)
#
# Apply our priority configuration (if set)
#
def merge(d1, d2):
for k in d2:
if k in d1 and isinstance(d1[k], dict) \
and isinstance(d2[k], dict):
merge(d1[k], d2[k])
else:
d1[k] = d2[k]
merge(payload, self.priority.payload())
self.logger.debug(
'FCM %s POST URL: %s (cert_verify=%r)',
self.mode, notify_url, self.verify_certificate,
@ -443,16 +520,30 @@ class NotifyFCM(NotifyBase):
# Define any URL parameters
params = {
'mode': self.mode,
'image': 'yes' if self.include_image else 'no',
'color': str(self.color),
}
if self.priority:
# Store our priority if one was defined
params['priority'] = str(self.priority)
if self.keyfile:
# Include our keyfile if specified
params['keyfile'] = NotifyFCM.quote(
self.keyfile[0].url(privacy=privacy), safe='')
if self.image_src:
# Include our image path as part of our URL payload
params['image_url'] = self.image_src
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Add our data keyword/args into our URL response
params.update(
{'+{}'.format(k): v for k, v in self.data_kwargs.items()})
reference = NotifyFCM.quote(self.project) \
if self.mode == FCMMode.OAuth2 \
else self.pprint(self.apikey, privacy, safe='')
@ -507,4 +598,30 @@ class NotifyFCM(NotifyBase):
results['keyfile'] = \
NotifyFCM.unquote(results['qsd']['keyfile'])
# Our Priority
if 'priority' in results['qsd'] and results['qsd']['priority']:
results['priority'] = \
NotifyFCM.unquote(results['qsd']['priority'])
# Our Color
if 'color' in results['qsd'] and results['qsd']['color']:
results['color'] = \
NotifyFCM.unquote(results['qsd']['color'])
# Boolean to include an image or not
results['include_image'] = parse_bool(results['qsd'].get(
'image', NotifyFCM.template_args['image']['default']))
# Extract image_url if it was specified
if 'image_url' in results['qsd']:
results['image_url'] = \
NotifyFCM.unquote(results['qsd']['image_url'])
if 'image' not in results['qsd']:
# Toggle default behaviour if a custom image was provided
# but ONLY if the `image` boolean was not set
results['include_image'] = True
# Store our data keyword/args if specified
results['data_kwargs'] = results['qsd+']
return results

127
apprise/plugins/NotifyFCM/color.py Normal file
View File

@ -0,0 +1,127 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2021 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
# New priorities are defined here:
# - https://firebase.google.com/docs/reference/fcm/rest/v1/\
# projects.messages#NotificationPriority
# Legacy color payload example here:
# https://firebase.google.com/docs/reference/fcm/rest/v1/\
# projects.messages#androidnotification
import re
import six
from ...utils import parse_bool
from ...common import NotifyType
from ...AppriseAsset import AppriseAsset
class FCMColorManager(object):
"""
A Simple object to accept either a boolean value
- True: Use colors provided by Apprise
- False: Do not use colors at all
- rrggbb: where you provide the rgb values (hence #333333)
- rgb: is also accepted as rgb values (hence #333)
For RGB colors, the hashtag is optional
"""
__color_rgb = re.compile(
r'#?((?P<r1>[0-9A-F]{2})(?P<g1>[0-9A-F]{2})(?P<b1>[0-9A-F]{2})'
r'|(?P<r2>[0-9A-F])(?P<g2>[0-9A-F])(?P<b2>[0-9A-F]))', re.IGNORECASE)
def __init__(self, color, asset=None):
"""
Parses the color object accordingly
"""
# Initialize an asset object if one isn't otherwise defined
self.asset = asset \
if isinstance(asset, AppriseAsset) else AppriseAsset()
# Prepare our color
self.color = color
if isinstance(color, six.string_types):
self.color = self.__color_rgb.match(color)
if self.color:
# Store our RGB value as #rrggbb
self.color = '{red}{green}{blue}'.format(
red=self.color.group('r1'),
green=self.color.group('g1'),
blue=self.color.group('b1')).lower() \
if self.color.group('r1') else \
'{red1}{red2}{green1}{green2}{blue1}{blue2}'.format(
red1=self.color.group('r2'),
red2=self.color.group('r2'),
green1=self.color.group('g2'),
green2=self.color.group('g2'),
blue1=self.color.group('b2'),
blue2=self.color.group('b2')).lower()
if self.color is None:
# Color not determined, so base it on boolean parser
self.color = parse_bool(color)
def get(self, notify_type=NotifyType.INFO):
"""
Returns color or true/false value based on configuration
"""
if isinstance(self.color, bool) and self.color:
# We want to use the asset value
return self.asset.color(notify_type=notify_type)
elif self.color:
# return our color as is
return '#' + self.color
# No color to return
return None
def __str__(self):
"""
our color representation
"""
if isinstance(self.color, bool):
return 'yes' if self.color else 'no'
# otherwise return our color
return self.color
def __bool__(self):
"""
Allows this object to be wrapped in an Python 3.x based 'if
statement'. True is returned if a color was loaded
"""
return True if self.color is True or \
isinstance(self.color, six.string_types) else False
def __nonzero__(self):
"""
Allows this object to be wrapped in an Python 2.x based 'if
statement'. True is returned if a color was loaded
"""
return True if self.color is True or \
isinstance(self.color, six.string_types) else False

42
apprise/plugins/NotifyFCM/common.py Normal file
View File

@ -0,0 +1,42 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2021 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
class FCMMode(object):
"""
Define the Firebase Cloud Messaging Modes
"""
# The legacy way of sending a message
Legacy = "legacy"
# The new API
OAuth2 = "oauth2"
# FCM Modes
FCM_MODES = (
# Legacy API
FCMMode.Legacy,
# HTTP v1 URL
FCMMode.OAuth2,
)

255
apprise/plugins/NotifyFCM/priority.py Normal file
View File

@ -0,0 +1,255 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2021 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
# New priorities are defined here:
# - https://firebase.google.com/docs/reference/fcm/rest/v1/\
# projects.messages#NotificationPriority
# Legacy priorities are defined here:
# - https://firebase.google.com/docs/cloud-messaging/http-server-ref
from .common import (FCMMode, FCM_MODES)
from ...logger import logger
class NotificationPriority(object):
"""
Defines the Notification Priorities as described on:
https://firebase.google.com/docs/reference/fcm/rest/v1/\
projects.messages#androidmessagepriority
NORMAL:
Default priority for data messages. Normal priority messages won't
open network connections on a sleeping device, and their delivery
may be delayed to conserve the battery. For less time-sensitive
messages, such as notifications of new email or other data to sync,
choose normal delivery priority.
HIGH:
Default priority for notification messages. FCM attempts to
deliver high priority messages immediately, allowing the FCM
service to wake a sleeping device when possible and open a network
connection to your app server. Apps with instant messaging, chat,
or voice call alerts, for example, generally need to open a
network connection and make sure FCM delivers the message to the
device without delay. Set high priority if the message is
time-critical and requires the user's immediate interaction, but
beware that setting your messages to high priority contributes
more to battery drain compared with normal priority messages.
"""
NORMAL = 'NORMAL'
HIGH = 'HIGH'
class FCMPriority(object):
"""
Defines our accepted priorites
"""
MIN = "min"
LOW = "low"
NORMAL = "normal"
HIGH = "high"
MAX = "max"
FCM_PRIORITIES = (
FCMPriority.MIN,
FCMPriority.LOW,
FCMPriority.NORMAL,
FCMPriority.HIGH,
FCMPriority.MAX,
)
class FCMPriorityManager(object):
"""
A Simple object to make it easier to work with FCM set priorities
"""
priority_map = {
FCMPriority.MIN: {
FCMMode.OAuth2: {
'message': {
'android': {
'priority': NotificationPriority.NORMAL
},
'apns': {
'headers': {
'apns-priority': "5"
}
},
'webpush': {
'headers': {
'Urgency': 'very-low'
}
},
}
},
FCMMode.Legacy: {
'priority': 'normal',
}
},
FCMPriority.LOW: {
FCMMode.OAuth2: {
'message': {
'android': {
'priority': NotificationPriority.NORMAL
},
'apns': {
'headers': {
'apns-priority': "5"
}
},
'webpush': {
'headers': {
'Urgency': 'low'
}
}
}
},
FCMMode.Legacy: {
'priority': 'normal',
}
},
FCMPriority.NORMAL: {
FCMMode.OAuth2: {
'message': {
'android': {
'priority': NotificationPriority.NORMAL
},
'apns': {
'headers': {
'apns-priority': "5"
}
},
'webpush': {
'headers': {
'Urgency': 'normal'
}
}
}
},
FCMMode.Legacy: {
'priority': 'normal',
}
},
FCMPriority.HIGH: {
FCMMode.OAuth2: {
'message': {
'android': {
'priority': NotificationPriority.HIGH
},
'apns': {
'headers': {
'apns-priority': "10"
}
},
'webpush': {
'headers': {
'Urgency': 'high'
}
}
}
},
FCMMode.Legacy: {
'priority': 'high',
}
},
FCMPriority.MAX: {
FCMMode.OAuth2: {
'message': {
'android': {
'priority': NotificationPriority.HIGH
},
'apns': {
'headers': {
'apns-priority': "10"
}
},
'webpush': {
'headers': {
'Urgency': 'high'
}
}
}
},
FCMMode.Legacy: {
'priority': 'high',
}
}
}
def __init__(self, mode, priority=None):
"""
Takes a FCMMode and Priority
"""
self.mode = mode
if self.mode not in FCM_MODES:
msg = 'The FCM mode specified ({}) is invalid.'.format(mode)
logger.warning(msg)
raise TypeError(msg)
self.priority = None
if priority:
self.priority = \
next((p for p in FCM_PRIORITIES
if p.startswith(priority[:2].lower())), None)
if not self.priority:
msg = 'An invalid FCM Priority ' \
'({}) was specified.'.format(priority)
logger.warning(msg)
raise TypeError(msg)
def payload(self):
"""
Returns our payload depending on our mode
"""
return self.priority_map[self.priority][self.mode] \
if self.priority else {}
def __str__(self):
"""
our priority representation
"""
return self.priority if self.priority else ''
def __bool__(self):
"""
Allows this object to be wrapped in an Python 3.x based 'if
statement'. True is returned if a priority was loaded
"""
return True if self.priority else False
def __nonzero__(self):
"""
Allows this object to be wrapped in an Python 2.x based 'if
statement'. True is returned if a priority was loaded
"""
return True if self.priority else False

422
test/test_plugin_fcm.py
View File

@ -22,8 +22,16 @@
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# Great Resources:
# - Dev/Legacy API:
# https://firebase.google.com/docs/cloud-messaging/http-server-ref
# - Legacy API (v1) -> OAuth
# - https://firebase.google.com/docs/cloud-messaging/migrate-v1
import io
import os
import six
import sys
import mock
import pytest
@ -35,6 +43,10 @@ from helpers import AppriseURLTester
try:
from apprise.plugins.NotifyFCM.oauth import GoogleOAuth
from apprise.plugins.NotifyFCM.common import FCM_MODES
from apprise.plugins.NotifyFCM.priority import (
FCMPriorityManager, FCM_PRIORITIES)
from apprise.plugins.NotifyFCM.color import FCMColorManager
from cryptography.exceptions import UnsupportedAlgorithm
except ImportError:
@ -101,6 +113,30 @@ apprise_url_tests = (
# Test apikey= to=
'instance': plugins.NotifyFCM,
}),
('fcm://?apikey=abc123&to=device&image=yes', {
# Test image boolean
'instance': plugins.NotifyFCM,
}),
('fcm://?apikey=abc123&to=device&color=no', {
# Disable colors
'instance': plugins.NotifyFCM,
}),
('fcm://?apikey=abc123&to=device&color=aabbcc', {
# custom colors
'instance': plugins.NotifyFCM,
}),
('fcm://?apikey=abc123&to=device'
'&image_url=http://example.com/interesting.jpg', {
# Test image_url
'instance': plugins.NotifyFCM}),
('fcm://?apikey=abc123&to=device'
'&image_url=http://example.com/interesting.jpg&image=no', {
# Test image_url but set to no
'instance': plugins.NotifyFCM}),
('fcm://?apikey=abc123&to=device&+key=value&+key2=value2', {
# Test apikey= to= and data arguments
'instance': plugins.NotifyFCM,
}),
('fcm://%20?to=device&keyfile=/invalid/path', {
# invalid Project ID
'instance': TypeError,
@ -176,9 +212,136 @@ def test_plugin_fcm_urls():
@pytest.mark.skipif(
'cryptography' not in sys.modules, reason="Requires cryptography")
@mock.patch('requests.post')
def test_plugin_fcm_general(mock_post):
def test_plugin_fcm_general_legacy(mock_post):
"""
NotifyFCM() General Checks
NotifyFCM() General Legacy/APIKey Checks
"""
# Disable Throttling to speed testing
plugins.NotifyBase.request_rate_per_sec = 0
# Prepare a good response
response = mock.Mock()
response.status_code = requests.codes.ok
mock_post.return_value = response
# A valid Legacy URL
obj = Apprise.instantiate(
'fcm://abc123/device/'
'?+key=value&+key2=value2'
'&image_url=https://example.com/interesting.png')
# Send our notification
assert obj.notify("test") is True
# Test our call count
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://fcm.googleapis.com/fcm/send'
payload = mock_post.mock_calls[0][2]
data = json.loads(payload['data'])
assert 'data' in data
assert isinstance(data, dict)
assert 'key' in data['data']
assert data['data']['key'] == 'value'
assert 'key2' in data['data']
assert data['data']['key2'] == 'value2'
assert 'notification' in data
assert isinstance(data['notification'], dict)
assert 'notification' in data['notification']
assert isinstance(data['notification']['notification'], dict)
assert 'image' in data['notification']['notification']
assert data['notification']['notification']['image'] == \
'https://example.com/interesting.png'
#
# Test priorities
#
mock_post.reset_mock()
obj = Apprise.instantiate(
'fcm://abc123/device/?priority=low')
assert mock_post.call_count == 0
# Send our notification
assert obj.notify(title="title", body="body") is True
# Test our call count
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://fcm.googleapis.com/fcm/send'
payload = mock_post.mock_calls[0][2]
data = json.loads(payload['data'])
assert 'data' not in data
assert 'notification' in data
assert isinstance(data['notification'], dict)
assert 'notification' in data['notification']
assert isinstance(data['notification']['notification'], dict)
assert 'image' not in data['notification']['notification']
assert 'priority' in data
# legacy can only switch between high/low
assert data['priority'] == "normal"
#
# Test colors
#
mock_post.reset_mock()
obj = Apprise.instantiate(
'fcm://abc123/device/?color=no')
assert mock_post.call_count == 0
# Send our notification
assert obj.notify(title="title", body="body") is True
# Test our call count
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://fcm.googleapis.com/fcm/send'
payload = mock_post.mock_calls[0][2]
data = json.loads(payload['data'])
assert 'data' not in data
assert 'notification' in data
assert isinstance(data['notification'], dict)
assert 'notification' in data['notification']
assert isinstance(data['notification']['notification'], dict)
assert 'image' not in data['notification']['notification']
assert 'color' not in data['notification']['notification']
mock_post.reset_mock()
obj = Apprise.instantiate(
'fcm://abc123/device/?color=AA001b')
assert mock_post.call_count == 0
# Send our notification
assert obj.notify(title="title", body="body") is True
# Test our call count
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://fcm.googleapis.com/fcm/send'
payload = mock_post.mock_calls[0][2]
data = json.loads(payload['data'])
assert 'data' not in data
assert 'notification' in data
assert isinstance(data['notification'], dict)
assert 'notification' in data['notification']
assert isinstance(data['notification']['notification'], dict)
assert 'image' not in data['notification']['notification']
assert 'color' in data['notification']['notification']
assert data['notification']['notification']['color'] == '#aa001b'
@pytest.mark.skipif(
'cryptography' not in sys.modules, reason="Requires cryptography")
@mock.patch('requests.post')
def test_plugin_fcm_general_oauth(mock_post):
"""
NotifyFCM() General OAuth Checks
"""
@ -222,7 +385,7 @@ def test_plugin_fcm_general(mock_post):
obj = Apprise.instantiate(
'fcm://mock-project-id/device/#topic/?keyfile={}'.format(str(path)))
# we'll fail as a result
# send our notification
assert obj.notify("test") is True
# Test our call count
@ -234,6 +397,146 @@ def test_plugin_fcm_general(mock_post):
assert mock_post.call_args_list[2][0][0] == \
'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send'
mock_post.reset_mock()
# Now we test using a valid Project ID and data parameters
obj = Apprise.instantiate(
'fcm://mock-project-id/device/#topic/?keyfile={}'
'&+key=value&+key2=value2'
'&image_url=https://example.com/interesting.png'.format(str(path)))
assert mock_post.call_count == 0
# send our notification
assert obj.notify("test") is True
# Test our call count
assert mock_post.call_count == 3
assert mock_post.call_args_list[0][0][0] == \
'https://accounts.google.com/o/oauth2/token'
assert mock_post.call_args_list[1][0][0] == \
'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send'
payload = mock_post.mock_calls[1][2]
data = json.loads(payload['data'])
assert 'message' in data
assert isinstance(data['message'], dict)
assert 'data' in data['message']
assert isinstance(data['message']['data'], dict)
assert 'key' in data['message']['data']
assert data['message']['data']['key'] == 'value'
assert 'key2' in data['message']['data']
assert data['message']['data']['key2'] == 'value2'
assert 'notification' in data['message']
assert isinstance(data['message']['notification'], dict)
assert 'image' in data['message']['notification']
assert data['message']['notification']['image'] == \
'https://example.com/interesting.png'
assert mock_post.call_args_list[2][0][0] == \
'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send'
payload = mock_post.mock_calls[2][2]
data = json.loads(payload['data'])
assert 'message' in data
assert isinstance(data['message'], dict)
assert 'data' in data['message']
assert isinstance(data['message']['data'], dict)
assert 'key' in data['message']['data']
assert data['message']['data']['key'] == 'value'
assert 'key2' in data['message']['data']
assert data['message']['data']['key2'] == 'value2'
assert 'notification' in data['message']
assert isinstance(data['message']['notification'], dict)
assert 'image' in data['message']['notification']
assert data['message']['notification']['image'] == \
'https://example.com/interesting.png'
#
# Test priorities
#
mock_post.reset_mock()
obj = Apprise.instantiate(
'fcm://mock-project-id/device/?keyfile={}'
'&priority=high'.format(str(path)))
assert mock_post.call_count == 0
# Send our notification
assert obj.notify(title="title", body="body") is True
# Test our call count
assert mock_post.call_count == 2
assert mock_post.call_args_list[0][0][0] == \
'https://accounts.google.com/o/oauth2/token'
assert mock_post.call_args_list[1][0][0] == \
'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send'
payload = mock_post.mock_calls[1][2]
data = json.loads(payload['data'])
assert 'message' in data
assert isinstance(data['message'], dict)
assert 'data' not in data['message']
assert 'notification' in data['message']
assert isinstance(data['message']['notification'], dict)
assert 'image' not in data['message']['notification']
assert data['message']['apns']['headers']['apns-priority'] == "10"
assert data['message']['webpush']['headers']['Urgency'] == "high"
assert data['message']['android']['priority'] == "HIGH"
#
# Test colors
#
mock_post.reset_mock()
obj = Apprise.instantiate(
'fcm://mock-project-id/device/?keyfile={}'
'&color=no'.format(str(path)))
assert mock_post.call_count == 0
# Send our notification
assert obj.notify(title="title", body="body") is True
# Test our call count
assert mock_post.call_count == 2
assert mock_post.call_args_list[0][0][0] == \
'https://accounts.google.com/o/oauth2/token'
assert mock_post.call_args_list[1][0][0] == \
'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send'
payload = mock_post.mock_calls[1][2]
data = json.loads(payload['data'])
assert 'message' in data
assert isinstance(data['message'], dict)
assert 'data' not in data['message']
assert 'notification' in data['message']
assert isinstance(data['message']['notification'], dict)
assert 'color' not in data['message']['notification']
mock_post.reset_mock()
obj = Apprise.instantiate(
'fcm://mock-project-id/device/?keyfile={}'
'&color=#12AAbb'.format(str(path)))
assert mock_post.call_count == 0
# Send our notification
assert obj.notify(title="title", body="body") is True
# Test our call count
assert mock_post.call_count == 2
assert mock_post.call_args_list[0][0][0] == \
'https://accounts.google.com/o/oauth2/token'
assert mock_post.call_args_list[1][0][0] == \
'https://fcm.googleapis.com/v1/projects/mock-project-id/messages:send'
payload = mock_post.mock_calls[1][2]
data = json.loads(payload['data'])
assert 'message' in data
assert isinstance(data['message'], dict)
assert 'data' not in data['message']
assert 'notification' in data['message']
assert isinstance(data['message']['notification'], dict)
assert 'color' in data['message']['android']['notification']
assert data['message']['android']['notification']['color'] == '#12aabb'
@pytest.mark.skipif(
'cryptography' not in sys.modules, reason="Requires cryptography")
@ -443,6 +746,96 @@ def test_plugin_fcm_keyfile_missing_entries_parse(tmpdir):
assert oauth.load(str(path)) is False
@pytest.mark.skipif(
'cryptography' not in sys.modules, reason="Requires cryptography")
def test_plugin_fcm_priorities():
"""
NotifyFCM() FCMPriorityManager() Testing
"""
for mode in FCM_MODES:
for priority in FCM_PRIORITIES:
instance = FCMPriorityManager(mode, priority)
assert isinstance(instance.payload(), dict)
# Verify it's not empty
assert bool(instance)
assert instance.payload()
assert str(instance) == priority
# We do not have to set a priority
instance = FCMPriorityManager(mode)
assert isinstance(instance.payload(), dict)
# Dictionary is empty though
assert not bool(instance)
assert not instance.payload()
assert str(instance) == ''
with pytest.raises(TypeError):
instance = FCMPriorityManager(mode, 'invalid')
with pytest.raises(TypeError):
instance = FCMPriorityManager('invald', 'high')
# mode validation is done at the higher NotifyFCM() level so
# it is not tested here (not required)
@pytest.mark.skipif(
'cryptography' not in sys.modules, reason="Requires cryptography")
def test_plugin_fcm_colors():
"""
NotifyFCM() FCMColorManager() Testing
"""
# No colors
instance = FCMColorManager('no')
assert bool(instance) is False
assert instance.get() is None
# We'll return that we are not defined
assert str(instance) == 'no'
# Asset colors
instance = FCMColorManager('yes')
assert isinstance(instance.get(), six.string_types)
# Output: #rrggbb
assert len(instance.get()) == 7
# Starts with has symbol
assert instance.get()[0] == '#'
# We'll return that we are defined but using default configuration
assert str(instance) == 'yes'
# We will be `true` because we can acquire a color based on what was
# passed in
assert bool(instance) is True
# Custom color
instance = FCMColorManager('#A2B3A4')
assert isinstance(instance.get(), six.string_types)
assert instance.get() == '#a2b3a4'
assert bool(instance) is True
# str() response does not include hashtag
assert str(instance) == 'a2b3a4'
# Custom color (no hashtag)
instance = FCMColorManager('A2B3A4')
assert isinstance(instance.get(), six.string_types)
# Hashtag is always part of output
assert instance.get() == '#a2b3a4'
assert bool(instance) is True
# str() response does not include hashtag
assert str(instance) == 'a2b3a4'
# Custom color (no hashtag) but only using 3 letter rgb values
instance = FCMColorManager('AC4')
assert isinstance(instance.get(), six.string_types)
# Hashtag is always part of output
assert instance.get() == '#aacc44'
assert bool(instance) is True
# str() response does not include hashtag
assert str(instance) == 'aacc44'
@pytest.mark.skipif(
'cryptography' in sys.modules,
reason="Requires that cryptography NOT be installed")
@ -460,3 +853,26 @@ def test_plugin_fcm_cryptography_import_error():
# It's not possible because our cryptography depedancy is missing
assert obj is None
@pytest.mark.skipif(
'cryptography' not in sys.modules, reason="Requires cryptography")
@mock.patch('requests.post')
def test_plugin_fcm_edge_cases(mock_post):
"""
NotifyFCM() Edge Cases
"""
# Disable Throttling to speed testing
plugins.NotifyBase.request_rate_per_sec = 0
# Prepare a good response
response = mock.Mock()
response.status_code = requests.codes.ok
mock_post.return_value = response
# this tests an edge case where verify if the data_kwargs is a dictionary
# or not. Below, we don't even define it, so it will be None (causing
# the check to go). We'll still correctly instantiate a plugin:
obj = plugins.NotifyFCM("project", "api:123", targets='device')
assert obj is not None