Expose notify platform targets as individual services (#2837)

* First pass on providing individual services for all possible targets that a notification platform supports.

* Add a quite hacky first version of notification groups

* Add a docstring for get_targets

* Register group service under notifygroup/ and safely check for notifygroups in config

* Remove notifygroups, because it belongs in its own PR

* Make @balloob requested changes

* get_targets()->targets

* Add tests for notify targets exposed as individual services

* If we dont have a platform name set in configuration, lets use the name of the platform instead of notify

* Fix test docstring.

* Dont use a dictionary for just one value

* No need to double slugify

* targets is now just a list of strings instead of a dict
This commit is contained in:
Robbie Trencheny 2016-08-16 22:05:41 -07:00 committed by Paulus Schoutsen
parent 37561765ff
commit c1ce6855c5
4 changed files with 64 additions and 3 deletions

View File

@ -64,6 +64,7 @@ def send_message(hass, message, title=None, data=None):
hass.services.call(DOMAIN, SERVICE_NOTIFY, info)
# pylint: disable=too-many-locals
def setup(hass, config):
"""Setup the notify services."""
success = False
@ -71,6 +72,8 @@ def setup(hass, config):
descriptions = load_yaml_config_file(
os.path.join(os.path.dirname(__file__), 'services.yaml'))
targets = {}
for platform, p_config in config_per_platform(config, DOMAIN):
notify_implementation = bootstrap.prepare_setup_platform(
hass, config, DOMAIN, platform)
@ -92,7 +95,10 @@ def setup(hass, config):
title = template.render(
hass, call.data.get(ATTR_TITLE, ATTR_TITLE_DEFAULT))
target = call.data.get(ATTR_TARGET)
if targets.get(call.service) is not None:
target = targets[call.service]
else:
target = call.data.get(ATTR_TARGET)
message = template.render(hass, message)
data = call.data.get(ATTR_DATA)
@ -100,8 +106,22 @@ def setup(hass, config):
data=data)
service_call_handler = partial(notify_message, notify_service)
service_name = slugify(p_config.get(CONF_NAME) or SERVICE_NOTIFY)
hass.services.register(DOMAIN, service_name, service_call_handler,
if hasattr(notify_service, 'targets'):
platform_name = (p_config.get(CONF_NAME) or platform)
for target in notify_service.targets:
target_name = slugify("{}_{}".format(platform_name, target))
targets[target_name] = target
hass.services.register(DOMAIN, target_name,
service_call_handler,
descriptions.get(SERVICE_NOTIFY),
schema=NOTIFY_SERVICE_SCHEMA)
platform_name = (p_config.get(CONF_NAME) or SERVICE_NOTIFY)
platform_name_slug = slugify(platform_name)
hass.services.register(DOMAIN, platform_name_slug,
service_call_handler,
descriptions.get(SERVICE_NOTIFY),
schema=NOTIFY_SERVICE_SCHEMA)
success = True

View File

@ -22,6 +22,11 @@ class DemoNotificationService(BaseNotificationService):
"""Initialize the service."""
self.hass = hass
@property
def targets(self):
"""Return a dictionary of registered targets."""
return ["test target"]
def send_message(self, message="", **kwargs):
"""Send a message to a user."""
kwargs['message'] = message

View File

@ -142,6 +142,11 @@ class HTML5NotificationService(BaseNotificationService):
self._gcm_key = gcm_key
self.registrations = registrations
@property
def targets(self):
"""Return a dictionary of registered targets."""
return self.registrations.keys()
def send_message(self, message="", **kwargs):
"""Send a message to a user."""
from pywebpush import WebPusher

View File

@ -22,6 +22,7 @@ class TestNotifyDemo(unittest.TestCase):
}
}))
self.events = []
self.calls = []
def record_event(event):
"""Record event to send notification."""
@ -33,6 +34,10 @@ class TestNotifyDemo(unittest.TestCase):
""""Stop down everything that was started."""
self.hass.stop()
def record_calls(self, *args):
"""Helper for recording calls."""
self.calls.append(args)
def test_sending_none_message(self):
"""Test send with None as message."""
notify.send_message(self.hass, None)
@ -93,3 +98,29 @@ data_template:
'sound':
'US-EN-Morgan-Freeman-Roommate-Is-Arriving.wav'}}
} == self.events[0].data
def test_targets_are_services(self):
"""Test that all targets are exposed as individual services."""
self.assertIsNotNone(self.hass.services.has_service("notify", "demo"))
service = "demo_test_target"
self.assertIsNotNone(self.hass.services.has_service("notify", service))
def test_messages_to_targets_route(self):
"""Test message routing to specific target services."""
self.hass.bus.listen_once("notify", self.record_calls)
self.hass.services.call("notify", "demo_test_target",
{'message': 'my message',
'title': 'my title',
'data': {'hello': 'world'}})
self.hass.pool.block_till_done()
data = self.calls[0][0].data
assert {
'message': 'my message',
'target': 'test target',
'title': 'my title',
'data': {'hello': 'world'}
} == data