# Copyright 2017 Google Inc.

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

"""Test cases for the firebase_admin.messaging module."""

import datetime

import json

import numbers

from googleapiclient import http

from googleapiclient import _helpers

import pytest

import firebase_admin

from firebase_admin import exceptions

from firebase_admin import messaging

from firebase_admin import _http_client

from tests import testutils

NON_STRING_ARGS = [list(), tuple(), dict(), True, False, 1, 0]

NON_DICT_ARGS = ['', list(), tuple(), True, False, 1, 0, {1: 'foo'}, {'foo': 1}]

NON_OBJECT_ARGS = [list(), tuple(), dict(), 'foo', 0, 1, True, False]

NON_LIST_ARGS = ['', tuple(), dict(), True, False, 1, 0, [1], ['foo', 1]]

NON_UINT_ARGS = ['1.23s', list(), tuple(), dict(), -1.23]

NON_BOOL_ARGS = ['', list(), tuple(), dict(), 1, 0, [1], ['foo', 1], {1: 'foo'}, {'foo': 1}]

HTTP_ERROR_CODES = {

400: exceptions.InvalidArgumentError,

403: exceptions.PermissionDeniedError,

404: exceptions.NotFoundError,

500: exceptions.InternalError,

503: exceptions.UnavailableError,

}

FCM_ERROR_CODES = {

'APNS_AUTH_ERROR': messaging.ThirdPartyAuthError,

'QUOTA_EXCEEDED': messaging.QuotaExceededError,

'SENDER_ID_MISMATCH': messaging.SenderIdMismatchError,

'THIRD_PARTY_AUTH_ERROR': messaging.ThirdPartyAuthError,

'UNREGISTERED': messaging.UnregisteredError,

}

def check_encoding(msg, expected=None):

encoded = messaging._MessagingService.encode_message(msg)

if expected:

assert encoded == expected

def check_exception(exception, message, status):

assert isinstance(exception, exceptions.FirebaseError)

assert str(exception) == message

assert exception.cause is not None

assert exception.http_response is not None

assert exception.http_response.status_code == status

class TestMessageStr:

@pytest.mark.parametrize('msg', [

messaging.Message(),

messaging.Message(topic='topic', token='token'),

messaging.Message(topic='topic', condition='condition'),

messaging.Message(condition='condition', token='token'),

messaging.Message(topic='topic', token='token', condition='condition'),

])

def test_invalid_target_message(self, msg):

with pytest.raises(ValueError) as excinfo:

str(msg)

assert str(

excinfo.value) == 'Exactly one of token, topic or condition must be specified.'

def test_empty_message(self):

assert str(messaging.Message(token='value')) == '{"token": "value"}'

assert str(messaging.Message(topic='value')) == '{"topic": "value"}'

assert str(messaging.Message(condition='value')

) == '{"condition": "value"}'

def test_data_message(self):

assert str(messaging.Message(topic='topic', data={})

) == '{"topic": "topic"}'

assert str(messaging.Message(topic='topic', data={

'k1': 'v1', 'k2': 'v2'})) == '{"data": {"k1": "v1", "k2": "v2"}, "topic": "topic"}'

class TestMulticastMessage:

@pytest.mark.parametrize('tokens', NON_LIST_ARGS)

def test_invalid_tokens_type(self, tokens):

with pytest.raises(ValueError) as excinfo:

messaging.MulticastMessage(tokens=tokens)

if isinstance(tokens, list):

expected = 'MulticastMessage.tokens must not contain non-string values.'

assert str(excinfo.value) == expected

else:

expected = 'MulticastMessage.tokens must be a list of strings.'

assert str(excinfo.value) == expected

def test_tokens_over_500(self):

with pytest.raises(ValueError) as excinfo:

messaging.MulticastMessage(tokens=['token' for _ in range(0, 501)])

expected = 'MulticastMessage.tokens must not contain more than 500 tokens.'

assert str(excinfo.value) == expected

def test_tokens_type(self):

message = messaging.MulticastMessage(tokens=['token'])

assert len(message.tokens) == 1

message = messaging.MulticastMessage(tokens=['token' for _ in range(0, 500)])

assert len(message.tokens) == 500

class TestMessageEncoder:

@pytest.mark.parametrize('msg', [

messaging.Message(),

messaging.Message(topic='topic', token='token'),

messaging.Message(topic='topic', condition='condition'),

messaging.Message(condition='condition', token='token'),

messaging.Message(topic='topic', token='token', condition='condition'),

])

def test_invalid_target_message(self, msg):

with pytest.raises(ValueError) as excinfo:

check_encoding(msg)

assert str(excinfo.value) == 'Exactly one of token, topic or condition must be specified.'

@pytest.mark.parametrize('target', NON_STRING_ARGS + [''])

def test_invalid_token(self, target):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(token=target))

assert str(excinfo.value) == 'Message.token must be a non-empty string.'

@pytest.mark.parametrize('target', NON_STRING_ARGS + [''])

def test_invalid_topic(self, target):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(topic=target))

assert str(excinfo.value) == 'Message.topic must be a non-empty string.'

@pytest.mark.parametrize('target', NON_STRING_ARGS + [''])

def test_invalid_condition(self, target):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(condition=target))

assert str(excinfo.value) == 'Message.condition must be a non-empty string.'

@pytest.mark.parametrize('topic', ['/topics/', '/foo/bar', 'foo bar'])

def test_malformed_topic_name(self, topic):

with pytest.raises(ValueError):

check_encoding(messaging.Message(topic=topic))

def test_empty_message(self):

check_encoding(messaging.Message(token='value'), {'token': 'value'})

check_encoding(messaging.Message(topic='value'), {'topic': 'value'})

check_encoding(messaging.Message(condition='value'), {'condition': 'value'})

@pytest.mark.parametrize('data', NON_DICT_ARGS)

def test_invalid_data_message(self, data):

with pytest.raises(ValueError):

check_encoding(messaging.Message(topic='topic', data=data))

def test_data_message(self):

check_encoding(messaging.Message(topic='topic', data={}), {'topic': 'topic'})

check_encoding(

messaging.Message(topic='topic', data={'k1': 'v1', 'k2': 'v2'}),

{'topic': 'topic', 'data': {'k1': 'v1', 'k2': 'v2'}})

def test_prefixed_topic(self):

check_encoding(messaging.Message(topic='/topics/topic'), {'topic': 'topic'})

def test_fcm_options(self):

check_encoding(

messaging.Message(

topic='topic', fcm_options=messaging.FCMOptions('analytics_label_v1')),

{'topic': 'topic', 'fcm_options': {'analytics_label': 'analytics_label_v1'}})

check_encoding(

messaging.Message(topic='topic', fcm_options=messaging.FCMOptions()),

{'topic': 'topic'})

class TestNotificationEncoder:

@pytest.mark.parametrize('data', NON_OBJECT_ARGS)

def test_invalid_notification(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', notification=data))

expected = 'Message.notification must be an instance of Notification class.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_title(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', notification=messaging.Notification(title=data)))

assert str(excinfo.value) == 'Notification.title must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_body(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', notification=messaging.Notification(body=data)))

assert str(excinfo.value) == 'Notification.body must be a string.'

def test_notification_message(self):

check_encoding(

messaging.Message(topic='topic', notification=messaging.Notification()),

{'topic': 'topic'})

check_encoding(

messaging.Message(topic='topic', notification=messaging.Notification('t', 'b')),

{'topic': 'topic', 'notification': {'title': 't', 'body': 'b'}})

check_encoding(

messaging.Message(topic='topic', notification=messaging.Notification('t')),

{'topic': 'topic', 'notification': {'title': 't'}})

class TestFcmOptionEncoder:

@pytest.mark.parametrize('label', [

'!',

'THIS_IS_LONGER_THAN_50_CHARACTERS_WHICH_IS_NOT_ALLOWED',

'',

])

def test_invalid_fcm_options(self, label):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic',

fcm_options=messaging.FCMOptions(label)

))

expected = 'Malformed FCMOptions.analytics_label.'

assert str(excinfo.value) == expected

def test_fcm_options(self):

check_encoding(

messaging.Message(

topic='topic',

fcm_options=messaging.FCMOptions(),

android=messaging.AndroidConfig(fcm_options=messaging.AndroidFCMOptions()),

apns=messaging.APNSConfig(fcm_options=messaging.APNSFCMOptions())

),

{'topic': 'topic'})

check_encoding(

messaging.Message(

topic='topic',

fcm_options=messaging.FCMOptions('message-label'),

android=messaging.AndroidConfig(

fcm_options=messaging.AndroidFCMOptions('android-label'),

direct_boot_ok=False),

apns=messaging.APNSConfig(fcm_options=

messaging.APNSFCMOptions(

analytics_label='apns-label',

image='https://images.unsplash.com/photo-14944386399'

'46-1ebd1d20bf85?fit=crop&w=900&q=60'))

),

{

'topic': 'topic',

'fcm_options': {'analytics_label': 'message-label'},

'android': {'fcm_options': {'analytics_label': 'android-label'},

'direct_boot_ok': False},

'apns': {'fcm_options': {'analytics_label': 'apns-label',

'image': 'https://images.unsplash.com/photo-14944386399'

'46-1ebd1d20bf85?fit=crop&w=900&q=60'}},

})

class TestAndroidConfigEncoder:

@pytest.mark.parametrize('data', NON_OBJECT_ARGS)

def test_invalid_android(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', android=data))

expected = 'Message.android must be an instance of AndroidConfig class.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_collapse_key(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(collapse_key=data)))

assert str(excinfo.value) == 'AndroidConfig.collapse_key must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS + ['foo'])

def test_invalid_priority(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(priority=data)))

if isinstance(data, str):

assert str(excinfo.value) == 'AndroidConfig.priority must be "high" or "normal".'

else:

assert str(excinfo.value) == 'AndroidConfig.priority must be a non-empty string.'

@pytest.mark.parametrize('data', NON_UINT_ARGS)

def test_invalid_ttl(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(ttl=data)))

if isinstance(data, numbers.Number):

assert str(excinfo.value) == ('AndroidConfig.ttl must not be negative.')

else:

assert str(excinfo.value) == ('AndroidConfig.ttl must be a duration in seconds or '

'an instance of datetime.timedelta.')

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_package_name(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(restricted_package_name=data)))

assert str(excinfo.value) == 'AndroidConfig.restricted_package_name must be a string.'

@pytest.mark.parametrize('data', NON_DICT_ARGS)

def test_invalid_data(self, data):

with pytest.raises(ValueError):

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(data=data)))

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_analytics_label(self, data):

with pytest.raises(ValueError):

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(

fcm_options=messaging.AndroidFCMOptions(analytics_label=data))))

@pytest.mark.parametrize('data', NON_BOOL_ARGS)

def test_invalid_direct_boot_ok(self, data):

with pytest.raises(ValueError):

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(direct_boot_ok=data)))

def test_android_config(self):

msg = messaging.Message(

topic='topic',

android=messaging.AndroidConfig(

collapse_key='key',

restricted_package_name='package',

priority='high',

ttl=123,

data={'k1': 'v1', 'k2': 'v2'},

fcm_options=messaging.AndroidFCMOptions('analytics_label_v1'),

direct_boot_ok=True,

)

)

expected = {

'topic': 'topic',

'android': {

'collapse_key': 'key',

'restricted_package_name': 'package',

'priority': 'high',

'ttl': '123s',

'data': {

'k1': 'v1',

'k2': 'v2',

},

'fcm_options': {

'analytics_label': 'analytics_label_v1',

},

'direct_boot_ok': True,

},

}

check_encoding(msg, expected)

@pytest.mark.parametrize('ttl', [

(0.5, '0.500000000s'),

(123, '123s'),

(123.45, '123.450000000s'),

(datetime.timedelta(days=1, seconds=100), '86500s'),

])

def test_android_ttl(self, ttl):

msg = messaging.Message(

topic='topic',

android=messaging.AndroidConfig(ttl=ttl[0])

)

expected = {

'topic': 'topic',

'android': {

'ttl': ttl[1],

},

}

check_encoding(msg, expected)

class TestAndroidNotificationEncoder:

def _check_notification(self, notification):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(notification=notification)))

return excinfo

@pytest.mark.parametrize('data', NON_OBJECT_ARGS)

def test_invalid_android_notification(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(notification=data)))

expected = 'AndroidConfig.notification must be an instance of AndroidNotification class.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_title(self, data):

notification = messaging.AndroidNotification(title=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.title must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_body(self, data):

notification = messaging.AndroidNotification(body=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.body must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_icon(self, data):

notification = messaging.AndroidNotification(icon=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.icon must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS + ['foo', '#xxyyzz', '112233', '#11223'])

def test_invalid_color(self, data):

notification = messaging.AndroidNotification(color=data)

excinfo = self._check_notification(notification)

if isinstance(data, str):

assert str(excinfo.value) == 'AndroidNotification.color must be in the form #RRGGBB.'

else:

assert str(excinfo.value) == 'AndroidNotification.color must be a non-empty string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_sound(self, data):

notification = messaging.AndroidNotification(sound=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.sound must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_tag(self, data):

notification = messaging.AndroidNotification(tag=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.tag must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_click_action(self, data):

notification = messaging.AndroidNotification(click_action=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.click_action must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_title_loc_key(self, data):

notification = messaging.AndroidNotification(title_loc_key=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.title_loc_key must be a string.'

@pytest.mark.parametrize('data', NON_LIST_ARGS)

def test_invalid_title_loc_args(self, data):

notification = messaging.AndroidNotification(title_loc_key='foo', title_loc_args=data)

excinfo = self._check_notification(notification)

if isinstance(data, list):

expected = 'AndroidNotification.title_loc_args must not contain non-string values.'

assert str(excinfo.value) == expected

else:

expected = 'AndroidNotification.title_loc_args must be a list of strings.'

assert str(excinfo.value) == expected

def test_no_title_loc_key(self):

notification = messaging.AndroidNotification(title_loc_args=['foo'])

excinfo = self._check_notification(notification)

expected = 'AndroidNotification.title_loc_key is required when specifying title_loc_args.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_body_loc_key(self, data):

notification = messaging.AndroidNotification(body_loc_key=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.body_loc_key must be a string.'

@pytest.mark.parametrize('data', NON_LIST_ARGS)

def test_invalid_body_loc_args(self, data):

notification = messaging.AndroidNotification(body_loc_key='foo', body_loc_args=data)

excinfo = self._check_notification(notification)

if isinstance(data, list):

expected = 'AndroidNotification.body_loc_args must not contain non-string values.'

assert str(excinfo.value) == expected

else:

expected = 'AndroidNotification.body_loc_args must be a list of strings.'

assert str(excinfo.value) == expected

def test_no_body_loc_key(self):

notification = messaging.AndroidNotification(body_loc_args=['foo'])

excinfo = self._check_notification(notification)

expected = 'AndroidNotification.body_loc_key is required when specifying body_loc_args.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_channel_id(self, data):

notification = messaging.AndroidNotification(channel_id=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.channel_id must be a string.'

@pytest.mark.parametrize('timestamp', [100, '', 'foo', {}, [], list(), dict()])

def test_invalid_event_timestamp(self, timestamp):

notification = messaging.AndroidNotification(event_timestamp=timestamp)

excinfo = self._check_notification(notification)

expected = 'AndroidNotification.event_timestamp must be a datetime.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('priority', NON_STRING_ARGS + ['foo'])

def test_invalid_priority(self, priority):

notification = messaging.AndroidNotification(priority=priority)

excinfo = self._check_notification(notification)

if isinstance(priority, str):

if not priority:

expected = 'AndroidNotification.priority must be a non-empty string.'

else:

expected = ('AndroidNotification.priority must be "default", "min", "low", "high" '

'or "max".')

else:

expected = 'AndroidNotification.priority must be a non-empty string.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('visibility', NON_STRING_ARGS + ['foo'])

def test_invalid_visibility(self, visibility):

notification = messaging.AndroidNotification(visibility=visibility)

excinfo = self._check_notification(notification)

if isinstance(visibility, str):

if not visibility:

expected = 'AndroidNotification.visibility must be a non-empty string.'

else:

expected = ('AndroidNotification.visibility must be "private", "public" or'

' "secret".')

else:

expected = 'AndroidNotification.visibility must be a non-empty string.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('vibrate_timings', ['', 1, True, 'msec', ['500', 500], [0, 'abc']])

def test_invalid_vibrate_timings_millis(self, vibrate_timings):

notification = messaging.AndroidNotification(vibrate_timings_millis=vibrate_timings)

excinfo = self._check_notification(notification)

if isinstance(vibrate_timings, list):

expected = ('AndroidNotification.vibrate_timings_millis must not contain non-number '

'values.')

else:

expected = 'AndroidNotification.vibrate_timings_millis must be a list of numbers.'

assert str(excinfo.value) == expected

def test_negative_vibrate_timings_millis(self):

notification = messaging.AndroidNotification(

vibrate_timings_millis=[100, -20, 15])

excinfo = self._check_notification(notification)

expected = 'AndroidNotification.vibrate_timings_millis must not be negative.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('notification_count', ['', 'foo', list(), tuple(), dict()])

def test_invalid_notification_count(self, notification_count):

notification = messaging.AndroidNotification(notification_count=notification_count)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'AndroidNotification.notification_count must be a number.'

def test_android_notification(self):

msg = messaging.Message(

topic='topic',

android=messaging.AndroidConfig(

notification=messaging.AndroidNotification(

title='t', body='b', icon='i', color='#112233', sound='s', tag='t',

click_action='ca', title_loc_key='tlk', body_loc_key='blk',

title_loc_args=['t1', 't2'], body_loc_args=['b1', 'b2'], channel_id='c',

ticker='ticker', sticky=True,

event_timestamp=datetime.datetime(

2019, 10, 20, 15, 12, 23, 123,

tzinfo=datetime.timezone(datetime.timedelta(hours=-5))

),

local_only=False,

priority='high', vibrate_timings_millis=[100, 50, 250],

default_vibrate_timings=False, default_sound=True,

light_settings=messaging.LightSettings(

color='#AABBCCDD', light_on_duration_millis=200,

light_off_duration_millis=300,

),

default_light_settings=False, visibility='public', notification_count=1,

)

)

)

expected = {

'topic': 'topic',

'android': {

'notification': {

'title': 't',

'body': 'b',

'icon': 'i',

'color': '#112233',

'sound': 's',

'tag': 't',

'click_action': 'ca',

'title_loc_key': 'tlk',

'body_loc_key': 'blk',

'title_loc_args': ['t1', 't2'],

'body_loc_args': ['b1', 'b2'],

'channel_id': 'c',

'ticker': 'ticker',

'sticky': True,

'event_time': '2019-10-20T20:12:23.000123Z',

'local_only': False,

'notification_priority': 'PRIORITY_HIGH',

'vibrate_timings': ['0.100000000s', '0.050000000s', '0.250000000s'],

'default_vibrate_timings': False,

'default_sound': 1,

'light_settings': {

'color': {

'red': 0.6666666666666666,

'green': 0.7333333333333333,

'blue': 0.8,

'alpha': 0.8666666666666667,

},

'light_on_duration': '0.200000000s',

'light_off_duration': '0.300000000s',

},

'default_light_settings': False,

'visibility': 'PUBLIC',

'notification_count': 1,

},

},

}

check_encoding(msg, expected)

def test_android_notification_naive_event_timestamp(self):

event_time = datetime.datetime.now()

msg = messaging.Message(

topic='topic',

android=messaging.AndroidConfig(

notification=messaging.AndroidNotification(

title='t',

event_timestamp=event_time,

)

)

)

expected = {

'topic': 'topic',

'android': {

'notification': {

'title': 't',

'event_time': event_time.strftime('%Y-%m-%dT%H:%M:%S.%fZ')

},

},

}

check_encoding(msg, expected)

class TestLightSettingsEncoder:

def _check_light_settings(self, light_settings):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(

notification=messaging.AndroidNotification(

light_settings=light_settings

))))

return excinfo

@pytest.mark.parametrize('data', NON_OBJECT_ARGS)

def test_invalid_light_settings(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', android=messaging.AndroidConfig(

notification=messaging.AndroidNotification(

light_settings=data

))))

expected = 'AndroidNotification.light_settings must be an instance of LightSettings class.'

assert str(excinfo.value) == expected

def test_no_color(self):

light_settings = messaging.LightSettings(color=None, light_on_duration_millis=200,

light_off_duration_millis=200)

excinfo = self._check_light_settings(light_settings)

expected = 'LightSettings.color is required.'

assert str(excinfo.value) == expected

def test_no_light_on_duration_millis(self):

light_settings = messaging.LightSettings(color='#aabbcc', light_on_duration_millis=None,

light_off_duration_millis=200)

excinfo = self._check_light_settings(light_settings)

expected = 'LightSettings.light_on_duration_millis is required.'

assert str(excinfo.value) == expected

def test_no_light_off_duration_millis(self):

light_settings = messaging.LightSettings(color='#aabbcc', light_on_duration_millis=200,

light_off_duration_millis=None)

excinfo = self._check_light_settings(light_settings)

expected = 'LightSettings.light_off_duration_millis is required.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('data', NON_UINT_ARGS)

def test_invalid_light_off_duration_millis(self, data):

light_settings = messaging.LightSettings(color='#aabbcc',

light_on_duration_millis=200,

light_off_duration_millis=data)

excinfo = self._check_light_settings(light_settings)

if isinstance(data, numbers.Number):

assert str(excinfo.value) == ('LightSettings.light_off_duration_millis must not be '

'negative.')

else:

assert str(excinfo.value) == ('LightSettings.light_off_duration_millis must be a '

'duration in milliseconds or '

'an instance of datetime.timedelta.')

@pytest.mark.parametrize('data', NON_UINT_ARGS)

def test_invalid_light_on_duration_millis(self, data):

light_settings = messaging.LightSettings(color='#aabbcc',

light_on_duration_millis=data,

light_off_duration_millis=200)

excinfo = self._check_light_settings(light_settings)

if isinstance(data, numbers.Number):

assert str(excinfo.value) == ('LightSettings.light_on_duration_millis must not be '

'negative.')

else:

assert str(excinfo.value) == ('LightSettings.light_on_duration_millis must be a '

'duration in milliseconds or '

'an instance of datetime.timedelta.')

@pytest.mark.parametrize('data', NON_STRING_ARGS + ['foo', '#xxyyzz', '112233', '#11223'])

def test_invalid_color(self, data):

notification = messaging.LightSettings(color=data, light_on_duration_millis=300,

light_off_duration_millis=200)

excinfo = self._check_light_settings(notification)

if isinstance(data, str):

assert str(excinfo.value) == ('LightSettings.color must be in the form #RRGGBB or '

'#RRGGBBAA.')

else:

assert str(

excinfo.value) == 'LightSettings.color must be a non-empty string.'

def test_light_settings(self):

msg = messaging.Message(

topic='topic', android=messaging.AndroidConfig(

notification=messaging.AndroidNotification(

light_settings=messaging.LightSettings(

color="#aabbcc",

light_on_duration_millis=200,

light_off_duration_millis=300,

)

))

)

expected = {

'topic': 'topic',

'android': {

'notification': {

'light_settings': {

'color': {

'red': 0.6666666666666666,

'green': 0.7333333333333333,

'blue': 0.8,

'alpha': 1,

},

'light_on_duration': '0.200000000s',

'light_off_duration': '0.300000000s',

}

},

},

}

check_encoding(msg, expected)

class TestWebpushConfigEncoder:

@pytest.mark.parametrize('data', NON_OBJECT_ARGS)

def test_invalid_webpush(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', webpush=data))

expected = 'Message.webpush must be an instance of WebpushConfig class.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('data', NON_DICT_ARGS)

def test_invalid_headers(self, data):

with pytest.raises(ValueError):

check_encoding(messaging.Message(

topic='topic', webpush=messaging.WebpushConfig(headers=data)))

@pytest.mark.parametrize('data', NON_DICT_ARGS)

def test_invalid_data(self, data):

with pytest.raises(ValueError):

check_encoding(messaging.Message(

topic='topic', webpush=messaging.WebpushConfig(data=data)))

def test_webpush_config(self):

msg = messaging.Message(

topic='topic',

webpush=messaging.WebpushConfig(

headers={'h1': 'v1', 'h2': 'v2'},

data={'k1': 'v1', 'k2': 'v2'},

)

)

expected = {

'topic': 'topic',

'webpush': {

'headers': {

'h1': 'v1',

'h2': 'v2',

},

'data': {

'k1': 'v1',

'k2': 'v2',

},

},

}

check_encoding(msg, expected)

class TestWebpushFCMOptionsEncoder:

@pytest.mark.parametrize('data', NON_OBJECT_ARGS)

def test_invalid_webpush_fcm_options(self, data):

with pytest.raises(AttributeError):

check_encoding(messaging.Message(

topic='topic', webpush=messaging.WebpushConfig(fcm_options=data)))

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_link_type(self, data):

options = messaging.WebpushFCMOptions(link=data)

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', webpush=messaging.WebpushConfig(fcm_options=options)))

expected = 'WebpushConfig.fcm_options.link must be a string.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('data', ['', 'foo', 'http://example'])

def test_invalid_link_format(self, data):

options = messaging.WebpushFCMOptions(link=data)

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', webpush=messaging.WebpushConfig(fcm_options=options)))

expected = 'WebpushFCMOptions.link must be a HTTPS URL.'

assert str(excinfo.value) == expected

def test_webpush_options(self):

msg = messaging.Message(

topic='topic',

webpush=messaging.WebpushConfig(

fcm_options=messaging.WebpushFCMOptions(

link='https://example',

),

)

)

expected = {

'topic': 'topic',

'webpush': {

'fcm_options': {

'link': 'https://example',

},

},

}

check_encoding(msg, expected)

class TestWebpushNotificationEncoder:

def _check_notification(self, notification):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', webpush=messaging.WebpushConfig(notification=notification)))

return excinfo

@pytest.mark.parametrize('data', NON_OBJECT_ARGS)

def test_invalid_webpush_notification(self, data):

with pytest.raises(ValueError) as excinfo:

check_encoding(messaging.Message(

topic='topic', webpush=messaging.WebpushConfig(notification=data)))

expected = 'WebpushConfig.notification must be an instance of WebpushNotification class.'

assert str(excinfo.value) == expected

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_title(self, data):

notification = messaging.WebpushNotification(title=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'WebpushNotification.title must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_body(self, data):

notification = messaging.WebpushNotification(body=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'WebpushNotification.body must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_icon(self, data):

notification = messaging.WebpushNotification(icon=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'WebpushNotification.icon must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_badge(self, data):

notification = messaging.WebpushNotification(badge=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'WebpushNotification.badge must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS + ['foo'])

def test_invalid_direction(self, data):

notification = messaging.WebpushNotification(direction=data)

excinfo = self._check_notification(notification)

if isinstance(data, str):

assert str(excinfo.value) == ('WebpushNotification.direction must be "auto", '

'"ltr" or "rtl".')

else:

assert str(excinfo.value) == 'WebpushNotification.direction must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_image(self, data):

notification = messaging.WebpushNotification(image=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'WebpushNotification.image must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_language(self, data):

notification = messaging.WebpushNotification(language=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'WebpushNotification.language must be a string.'

@pytest.mark.parametrize('data', NON_STRING_ARGS)

def test_invalid_tag(self, data):

notification = messaging.WebpushNotification(tag=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'WebpushNotification.tag must be a string.'

@pytest.mark.parametrize('data', ['', 'foo', list(), tuple(), dict()])

def test_invalid_timestamp(self, data):

notification = messaging.WebpushNotification(timestamp_millis=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'WebpushNotification.timestamp_millis must be a number.'

@pytest.mark.parametrize('data', ['', list(), tuple(), True, False, 1, 0])

def test_invalid_custom_data(self, data):

notification = messaging.WebpushNotification(custom_data=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == 'WebpushNotification.custom_data must be a dict.'

@pytest.mark.parametrize('data', ['', dict(), tuple(), True, False, 1, 0, [1, 2]])

def test_invalid_actions(self, data):

notification = messaging.WebpushNotification(actions=data)

excinfo = self._check_notification(notification)

assert str(excinfo.value) == ('WebpushConfig.notification.actions must be a list of '

'WebpushNotificationAction instances.')

def test_webpush_notification(self):

msg = messaging.Message(

topic='topic',

webpush=messaging.WebpushConfig(

notification=messaging.WebpushNotification(

badge='badge',

body='body',

data={'foo': 'bar'},

icon='icon',

image='image',

language='language',

renotify=True,

require_interaction=True,

silent=True,

tag='tag',

timestamp_millis=100,

title='title',

vibrate=[100, 200, 100],

custom_data={'k1': 'v1', 'k2': 'v2'},

),

)

)

expected = {

'topic': 'topic',

'webpush': {

'notification': {

'badge': 'badge',

'body': 'body',

'data': {'foo': 'bar'},

'icon': 'icon',

'image': 'image',

'lang': 'language',

'renotify': True,

'requireInteraction': True,

'silent': True,

'tag': 'tag',

'timestamp': 100,

'vibrate': [100, 200, 100],

'title': 'title',

'k1': 'v1',

'k2': 'v2',

},

},

}

check_encoding(msg, expected)

def test_multiple_field_specifications(self):

notification = messaging.WebpushNotification(

badge='badge',

custom_data={'badge': 'other badge'},

)

excinfo = self._check_notification(notification)

expected = 'Multiple specifications for badge in WebpushNotification.'

assert str(excinfo.value) == expected

def test_webpush_notification_action(self):

msg = messaging.Message(

topic='topic',