evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Support for Apple Push Notification Service (#3756)

Browse source 
* added push notification implementation

* some lint changes

* added docs

* added push notification implementation

* some lint changes

* added docs

* Fixed comment formatting issues

* Added requirments

* Update requirements_all.txt

* Update apns.py

* re-generated requirments_all.txt

* Added link to online docs

* added push notification implementation

* some lint changes

* added docs

* added push notification implementation

* some lint changes

* added docs

* Fixed comment formatting issues

* Added requirments

* Update requirements_all.txt

* Update apns.py

* re-generated requirments_all.txt

* Added link to online docs

* changed to use http/2 library for push notifications

* fixed lint issue

* fixed test that fails on CI

* another go at fixing test that fails on CI

* another go at fixing test that fails on CI

* another go at fixing test that fails on CI

* added missing docstring

* moved service description to main services.yaml file

* renamed apns service
This commit is contained in:
sam-io 2016-10-18 03:41:49 +01:00 • committed by Paulus Schoutsen
parent 4c8d1d9d2f
commit ae8a8e22ad
4 changed files with 662 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

289
homeassistant/components/notify/apns.py Normal file
View file

@ -0,0 +1,289 @@
"""
APNS Notification platform.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/notify.apns/
"""
import logging
import os
import voluptuous as vol
from homeassistant.helpers.event import track_state_change
from homeassistant.config import load_yaml_config_file
from homeassistant.components.notify import (
ATTR_TARGET, ATTR_DATA, BaseNotificationService)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import template as template_helper
DOMAIN = "apns"
APNS_DEVICES = "apns.yaml"
DEVICE_TRACKER_DOMAIN = "device_tracker"
SERVICE_REGISTER = "apns_register"
ATTR_PUSH_ID = "push_id"
ATTR_NAME = "name"
REGISTER_SERVICE_SCHEMA = vol.Schema({
vol.Required(ATTR_PUSH_ID): cv.string,
vol.Optional(ATTR_NAME, default=None): cv.string,
})
REQUIREMENTS = ["apns2==0.1.1"]
def get_service(hass, config):
"""Return push service."""
descriptions = load_yaml_config_file(
os.path.join(os.path.dirname(__file__), 'services.yaml'))
name = config.get("name")
if name is None:
logging.error("Name must be specified.")
return None
cert_file = config.get('cert_file')
if cert_file is None:
logging.error("Certificate must be specified.")
return None
topic = config.get('topic')
if topic is None:
logging.error("Topic must be specified.")
return None
sandbox = bool(config.get('sandbox', False))
service = ApnsNotificationService(hass, name, topic, sandbox, cert_file)
hass.services.register(DOMAIN,
name,
service.register,
descriptions.get(SERVICE_REGISTER),
schema=REGISTER_SERVICE_SCHEMA)
return service
class ApnsDevice(object):
"""
Apns Device class.
Stores information about a device that is
registered for push notifications.
"""
def __init__(self, push_id, name, tracking_device_id=None, disabled=False):
"""Initialize Apns Device."""
self.device_push_id = push_id
self.device_name = name
self.tracking_id = tracking_device_id
self.device_disabled = disabled
@property
def push_id(self):
"""The apns id for the device."""
return self.device_push_id
@property
def name(self):
"""The friendly name for the device."""
return self.device_name
@property
def tracking_device_id(self):
"""
Device Id.
The id of a device that is tracked by the device
tracking component.
"""
return self.tracking_id
@property
def full_tracking_device_id(self):
"""
Fully qualified device id.
The full id of a device that is tracked by the device
tracking component.
"""
return DEVICE_TRACKER_DOMAIN + '.' + self.tracking_id
@property
def disabled(self):
"""Should receive notifications."""
return self.device_disabled
def disable(self):
"""Disable the device from recieving notifications."""
self.device_disabled = True
def __eq__(self, other):
"""Return the comparision."""
if isinstance(other, self.__class__):
return self.push_id == other.push_id and self.name == other.name
return NotImplemented
def __ne__(self, other):
"""Return the comparision."""
return not self.__eq__(other)
class ApnsNotificationService(BaseNotificationService):
"""Implement the notification service for the APNS service."""
# pylint: disable=too-many-arguments
# pylint: disable=too-many-instance-attributes
def __init__(self, hass, app_name, topic, sandbox, cert_file):
"""Initialize APNS application."""
self.hass = hass
self.app_name = app_name
self.sandbox = sandbox
self.certificate = cert_file
self.yaml_path = hass.config.path(app_name + '_' + APNS_DEVICES)
self.devices = {}
self.device_states = {}
self.topic = topic
if os.path.isfile(self.yaml_path):
self.devices = {
str(key): ApnsDevice(
str(key),
value.get('name'),
value.get('tracking_device_id'),
value.get('disabled', False)
)
for (key, value) in
load_yaml_config_file(self.yaml_path).items()
}
tracking_ids = [
device.full_tracking_device_id
for (key, device) in self.devices.items()
if device.tracking_device_id is not None
]
track_state_change(
hass,
tracking_ids,
self.device_state_changed_listener)
def device_state_changed_listener(self, entity_id, from_s, to_s):
"""
Listener for sate change.
Track device state change if a device
has a tracking id specified.
"""
self.device_states[entity_id] = str(to_s.state)
return
@staticmethod
def write_device(out, device):
"""Write a single device to file."""
attributes = []
if device.name is not None:
attributes.append(
'name: {}'.format(device.name))
if device.tracking_device_id is not None:
attributes.append(
'tracking_device_id: {}'.format(device.tracking_device_id))
if device.disabled:
attributes.append('disabled: True')
out.write(device.push_id)
out.write(": {")
if len(attributes) > 0:
separator = ", "
out.write(separator.join(attributes))
out.write("}\n")
def write_devices(self):
"""Write all known devices to file."""
with open(self.yaml_path, 'w+') as out:
for _, device in self.devices.items():
ApnsNotificationService.write_device(out, device)
def register(self, call):
"""Register a device to receive push messages."""
push_id = call.data.get(ATTR_PUSH_ID)
if push_id is None:
return False
device_name = call.data.get(ATTR_NAME)
current_device = self.devices.get(push_id)
current_tracking_id = None if current_device is None \
else current_device.tracking_device_id
device = ApnsDevice(
push_id,
device_name,
current_tracking_id)
if current_device is None:
self.devices[push_id] = device
with open(self.yaml_path, 'a') as out:
self.write_device(out, device)
return True
if device != current_device:
self.devices[push_id] = device
self.write_devices()
return True
def send_message(self, message=None, **kwargs):
"""Send push message to registered devices."""
from apns2.client import APNsClient
from apns2.payload import Payload
from apns2.errors import Unregistered
apns = APNsClient(
self.certificate,
use_sandbox=self.sandbox,
use_alternative_port=False)
device_state = kwargs.get(ATTR_TARGET)
message_data = kwargs.get(ATTR_DATA)
if message_data is None:
message_data = {}
if isinstance(message, str):
rendered_message = message
elif isinstance(message, template_helper.Template):
rendered_message = message.render()
else:
rendered_message = ""
payload = Payload(
alert=rendered_message,
badge=message_data.get("badge"),
sound=message_data.get("sound"),
category=message_data.get("category"),
custom=message_data.get("custom", {}),
content_available=message_data.get("content_available", False))
device_update = False
for push_id, device in self.devices.items():
if not device.disabled:
state = None
if device.tracking_device_id is not None:
state = self.device_states.get(
device.full_tracking_device_id)
if device_state is None or state == str(device_state):
try:
apns.send_notification(
push_id,
payload,
topic=self.topic)
except Unregistered:
logging.error(
"Device %s has unregistered.",
push_id)
device_update = True
device.disable()
if device_update:
self.write_devices()
return True

12
homeassistant/components/notify/services.yaml
View file

@ -17,3 +17,15 @@ notify:
data: data:
description: Extended information for notification. Optional depending on the platform. description: Extended information for notification. Optional depending on the platform.
example: platform specific example: platform specific
apns_register:
description: Registers a device to receive push notifications.
fields:
push_id:
description: The device token, a 64 character hex string (256 bits). The device token is provided to you by your client app, which receives the token after registering itself with the remote notification service.
example: '72f2a8633655c5ce574fdc9b2b34ff8abdfc3b739b6ceb7a9ff06c1cbbf99f62'
name:
description: A friendly name for the device (optional).
example: 'Sam''s iPhone'

3
requirements_all.txt
View file

@ -35,6 +35,9 @@ Werkzeug==0.11.11
# homeassistant.components.apcupsd # homeassistant.components.apcupsd
apcaccess==0.0.4 apcaccess==0.0.4
# homeassistant.components.notify.apns
apns2==0.1.1
# homeassistant.components.sun # homeassistant.components.sun
astral==1.2 astral==1.2

358
tests/components/notify/test_apns.py Normal file
View file

@ -0,0 +1,358 @@
"""The tests for the APNS component."""
import unittest
import os
import homeassistant.components.notify as notify
from homeassistant.core import State
from homeassistant.components.notify.apns import ApnsNotificationService
from tests.common import get_test_home_assistant
from homeassistant.config import load_yaml_config_file
from unittest.mock import patch
from apns2.errors import Unregistered
class TestApns(unittest.TestCase):
"""Test the APNS component."""
def test_apns_setup_full(self):
"""Test setup with all data."""
config = {
'notify': {
'platform': 'apns',
'name': 'test_app',
'sandbox': 'True',
'topic': 'testapp.appname',
'cert_file': 'test_app.pem'
}
}
hass = get_test_home_assistant()
self.assertTrue(notify.setup(hass, config))
def test_apns_setup_missing_name(self):
"""Test setup with missing name."""
config = {
'notify': {
'platform': 'apns',
'sandbox': 'True',
'topic': 'testapp.appname',
'cert_file': 'test_app.pem'
}
}
hass = get_test_home_assistant()
self.assertFalse(notify.setup(hass, config))
def test_apns_setup_missing_certificate(self):
"""Test setup with missing name."""
config = {
'notify': {
'platform': 'apns',
'topic': 'testapp.appname',
'name': 'test_app'
}
}
hass = get_test_home_assistant()
self.assertFalse(notify.setup(hass, config))
def test_apns_setup_missing_topic(self):
"""Test setup with missing topic."""
config = {
'notify': {
'platform': 'apns',
'cert_file': 'test_app.pem',
'name': 'test_app'
}
}
hass = get_test_home_assistant()
self.assertFalse(notify.setup(hass, config))
def test_register_new_device(self):
"""Test registering a new device with a name."""
config = {
'notify': {
'platform': 'apns',
'name': 'test_app',
'topic': 'testapp.appname',
'cert_file': 'test_app.pem'
}
}
hass = get_test_home_assistant()
devices_path = hass.config.path('test_app_apns.yaml')
with open(devices_path, 'w+') as out:
out.write('5678: {name: test device 2}\n')
notify.setup(hass, config)
self.assertTrue(hass.services.call('apns',
'test_app',
{'push_id': '1234',
'name': 'test device'},
blocking=True))
devices = {str(key): value for (key, value) in
load_yaml_config_file(devices_path).items()}
test_device_1 = devices.get('1234')
test_device_2 = devices.get('5678')
self.assertIsNotNone(test_device_1)
self.assertIsNotNone(test_device_2)
self.assertEqual('test device', test_device_1.get('name'))
os.remove(devices_path)
def test_register_device_without_name(self):
"""Test registering a without a name."""
config = {
'notify': {
'platform': 'apns',
'name': 'test_app',
'topic': 'testapp.appname',
'cert_file': 'test_app.pem'
}
}
hass = get_test_home_assistant()
devices_path = hass.config.path('test_app_apns.yaml')
with open(devices_path, 'w+') as out:
out.write('5678: {name: test device 2}\n')
notify.setup(hass, config)
self.assertTrue(hass.services.call('apns', 'test_app',
{'push_id': '1234'},
blocking=True))
devices = {str(key): value for (key, value) in
load_yaml_config_file(devices_path).items()}
test_device = devices.get('1234')
self.assertIsNotNone(test_device)
self.assertIsNone(test_device.get('name'))
os.remove(devices_path)
def test_update_existing_device(self):
"""Test updating an existing device."""
config = {
'notify': {
'platform': 'apns',
'name': 'test_app',
'topic': 'testapp.appname',
'cert_file': 'test_app.pem'
}
}
hass = get_test_home_assistant()
devices_path = hass.config.path('test_app_apns.yaml')
with open(devices_path, 'w+') as out:
out.write('1234: {name: test device 1}\n')
out.write('5678: {name: test device 2}\n')
notify.setup(hass, config)
self.assertTrue(hass.services.call('apns',
'test_app',
{'push_id': '1234',
'name': 'updated device 1'},
blocking=True))
devices = {str(key): value for (key, value) in
load_yaml_config_file(devices_path).items()}
test_device_1 = devices.get('1234')
test_device_2 = devices.get('5678')
self.assertIsNotNone(test_device_1)
self.assertIsNotNone(test_device_2)
self.assertEqual('updated device 1', test_device_1.get('name'))
os.remove(devices_path)
def test_update_existing_device_with_tracking_id(self):
"""Test updating an existing device that has a tracking id."""
config = {
'notify': {
'platform': 'apns',
'name': 'test_app',
'topic': 'testapp.appname',
'cert_file': 'test_app.pem'
}
}
hass = get_test_home_assistant()
devices_path = hass.config.path('test_app_apns.yaml')
with open(devices_path, 'w+') as out:
out.write('1234: {name: test device 1, tracking_device_id: tracking123}\n') # nopep8
out.write('5678: {name: test device 2, tracking_device_id: tracking456}\n') # nopep8
notify.setup(hass, config)
self.assertTrue(hass.services.call('apns',
'test_app',
{'push_id': '1234',
'name': 'updated device 1'},
blocking=True))
devices = {str(key): value for (key, value) in
load_yaml_config_file(devices_path).items()}
test_device_1 = devices.get('1234')
test_device_2 = devices.get('5678')
self.assertIsNotNone(test_device_1)
self.assertIsNotNone(test_device_2)
self.assertEqual('tracking123',
test_device_1.get('tracking_device_id'))
self.assertEqual('tracking456',
test_device_2.get('tracking_device_id'))
os.remove(devices_path)
@patch('apns2.client.APNsClient')
def test_send(self, mock_client):
"""Test updating an existing device."""
send = mock_client.return_value.send_notification
config = {
'notify': {
'platform': 'apns',
'name': 'test_app',
'topic': 'testapp.appname',
'cert_file': 'test_app.pem'
}
}
hass = get_test_home_assistant()
devices_path = hass.config.path('test_app_apns.yaml')
with open(devices_path, 'w+') as out:
out.write('1234: {name: test device 1}\n')
notify.setup(hass, config)
self.assertTrue(hass.services.call('notify', 'test_app',
{'message': 'Hello',
'data': {
'badge': 1,
'sound': 'test.mp3',
'category': 'testing'
}
},
blocking=True))
self.assertTrue(send.called)
self.assertEqual(1, len(send.mock_calls))
target = send.mock_calls[0][1][0]
payload = send.mock_calls[0][1][1]
self.assertEqual('1234', target)
self.assertEqual('Hello', payload.alert)
self.assertEqual(1, payload.badge)
self.assertEqual('test.mp3', payload.sound)
self.assertEqual('testing', payload.category)
@patch('apns2.client.APNsClient')
def test_send_when_disabled(self, mock_client):
"""Test updating an existing device."""
send = mock_client.return_value.send_notification
config = {
'notify': {
'platform': 'apns',
'name': 'test_app',
'topic': 'testapp.appname',
'cert_file': 'test_app.pem'
}
}
hass = get_test_home_assistant()
devices_path = hass.config.path('test_app_apns.yaml')
with open(devices_path, 'w+') as out:
out.write('1234: {name: test device 1, disabled: True}\n')
notify.setup(hass, config)
self.assertTrue(hass.services.call('notify', 'test_app',
{'message': 'Hello',
'data': {
'badge': 1,
'sound': 'test.mp3',
'category': 'testing'
}
},
blocking=True))
self.assertFalse(send.called)
@patch('apns2.client.APNsClient')
def test_send_with_state(self, mock_client):
"""Test updating an existing device."""
send = mock_client.return_value.send_notification
hass = get_test_home_assistant()
devices_path = hass.config.path('test_app_apns.yaml')
with open(devices_path, 'w+') as out:
out.write('1234: {name: test device 1, tracking_device_id: tracking123}\n') # nopep8
out.write('5678: {name: test device 2, tracking_device_id: tracking456}\n') # nopep8
notify_service = ApnsNotificationService(
hass,
'test_app',
'testapp.appname',
False,
'test_app.pem'
)
notify_service.device_state_changed_listener(
'device_tracker.tracking456',
State('device_tracker.tracking456', None),
State('device_tracker.tracking456', 'home'))
hass.block_till_done()
notify_service.send_message(message='Hello', target='home')
self.assertTrue(send.called)
self.assertEqual(1, len(send.mock_calls))
target = send.mock_calls[0][1][0]
payload = send.mock_calls[0][1][1]
self.assertEqual('5678', target)
self.assertEqual('Hello', payload.alert)
@patch('apns2.client.APNsClient')
def test_disable_when_unregistered(self, mock_client):
"""Test disabling a device when it is unregistered."""
send = mock_client.return_value.send_notification
send.side_effect = Unregistered()
config = {
'notify': {
'platform': 'apns',
'name': 'test_app',
'topic': 'testapp.appname',
'cert_file': 'test_app.pem'
}
}
hass = get_test_home_assistant()
devices_path = hass.config.path('test_app_apns.yaml')
with open(devices_path, 'w+') as out:
out.write('1234: {name: test device 1}\n')
notify.setup(hass, config)
self.assertTrue(hass.services.call('notify', 'test_app',
{'message': 'Hello'},
blocking=True))
devices = {str(key): value for (key, value) in
load_yaml_config_file(devices_path).items()}
test_device_1 = devices.get('1234')
self.assertIsNotNone(test_device_1)
self.assertEqual(True, test_device_1.get('disabled'))
os.remove(devices_path)