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

Register mobile_app notification services when a new device is added ()

Browse source 
* Register mobile_app notification services when a new device is added

* targets and base service use their own patterns to generate the name
This commit is contained in:
J. Nick Koston 2020-08-28 12:18:02 -05:00 • committed by GitHub
parent 4b8217777e
commit 3377f6b12a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 129 additions and 52 deletions
homeassistant/components
mobile_app
__init__.py
notify
__init__.py
tests/components/mobile_app
test_notify.py

4
homeassistant/components/mobile_app/__init__.py
View file

@ -1,7 +1,7 @@
"""Integrates Native Apps to Home Assistant."""
import asyncio
from homeassistant.components import cloud
from homeassistant.components import cloud, notify as hass_notify
from homeassistant.components.webhook import (
async_register as webhook_register,
async_unregister as webhook_unregister,
@ -101,6 +101,8 @@ async def async_setup_entry(hass, entry):
hass.config_entries.async_forward_entry_setup(entry, domain)
)
await hass_notify.async_reload(hass, DOMAIN)
return True

148
homeassistant/components/notify/__init__.py
View file

@ -11,6 +11,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_per_platform, discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.loader import bind_hass
from homeassistant.setup import async_prepare_setup_platform
from homeassistant.util import slugify
@ -35,6 +36,12 @@ DOMAIN = "notify"
SERVICE_NOTIFY = "notify"
NOTIFY_SERVICES = "notify_services"
SERVICE = "service"
TARGETS = "targets"
FRIENDLY_NAME = "friendly_name"
TARGET_FRIENDLY_NAME = "target_friendly_name"
PLATFORM_SCHEMA = vol.Schema(
{vol.Required(CONF_PLATFORM): cv.string, vol.Optional(CONF_NAME): cv.string},
extra=vol.ALLOW_EXTRA,
@ -50,22 +57,89 @@ NOTIFY_SERVICE_SCHEMA = vol.Schema(
)
@bind_hass
async def async_reload(hass, integration_name):
"""Register notify services for an integration."""
if NOTIFY_SERVICES not in hass.data:
return
data = hass.data[NOTIFY_SERVICES][integration_name]
notify_service = data[SERVICE]
friendly_name = data[FRIENDLY_NAME]
targets = data[TARGETS]
async def _async_notify_message(service):
"""Handle sending notification message service calls."""
await _async_notify_message_service(hass, service, notify_service, targets)
if hasattr(notify_service, "targets"):
target_friendly_name = data[TARGET_FRIENDLY_NAME]
for name, target in notify_service.targets.items():
target_name = slugify(f"{target_friendly_name}_{name}")
if target_name in targets:
continue
targets[target_name] = target
hass.services.async_register(
DOMAIN,
target_name,
_async_notify_message,
schema=NOTIFY_SERVICE_SCHEMA,
)
friendly_name_slug = slugify(friendly_name)
if hass.services.has_service(DOMAIN, friendly_name_slug):
return
hass.services.async_register(
DOMAIN,
friendly_name_slug,
_async_notify_message,
schema=NOTIFY_SERVICE_SCHEMA,
)
async def _async_notify_message_service(hass, service, notify_service, targets):
"""Handle sending notification message service calls."""
kwargs = {}
message = service.data[ATTR_MESSAGE]
title = service.data.get(ATTR_TITLE)
if title:
title.hass = hass
kwargs[ATTR_TITLE] = title.async_render()
if targets.get(service.service) is not None:
kwargs[ATTR_TARGET] = [targets[service.service]]
elif service.data.get(ATTR_TARGET) is not None:
kwargs[ATTR_TARGET] = service.data.get(ATTR_TARGET)
message.hass = hass
kwargs[ATTR_MESSAGE] = message.async_render()
kwargs[ATTR_DATA] = service.data.get(ATTR_DATA)
await notify_service.async_send_message(**kwargs)
async def async_setup(hass, config):
"""Set up the notify services."""
targets = {}
hass.data.setdefault(NOTIFY_SERVICES, {})
async def async_setup_platform(p_type, p_config=None, discovery_info=None):
async def async_setup_platform(
integration_name, p_config=None, discovery_info=None
):
"""Set up a notify platform."""
if p_config is None:
p_config = {}
platform = await async_prepare_setup_platform(hass, config, DOMAIN, p_type)
platform = await async_prepare_setup_platform(
hass, config, DOMAIN, integration_name
)
if platform is None:
_LOGGER.error("Unknown notification service specified")
return
_LOGGER.info("Setting up %s.%s", DOMAIN, p_type)
_LOGGER.info("Setting up %s.%s", DOMAIN, integration_name)
notify_service = None
try:
if hasattr(platform, "async_get_service"):
@ -84,12 +158,12 @@ async def async_setup(hass, config):
# on discovery data.
if discovery_info is None:
_LOGGER.error(
"Failed to initialize notification service %s", p_type
"Failed to initialize notification service %s", integration_name
)
return
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Error setting up platform %s", p_type)
_LOGGER.exception("Error setting up platform %s", integration_name)
return
notify_service.hass = hass
@ -97,60 +171,32 @@ async def async_setup(hass, config):
if discovery_info is None:
discovery_info = {}
async def async_notify_message(service):
"""Handle sending notification message service calls."""
kwargs = {}
message = service.data[ATTR_MESSAGE]
title = service.data.get(ATTR_TITLE)
target_friendly_name = (
p_config.get(CONF_NAME) or discovery_info.get(CONF_NAME) or integration_name
)
if title:
title.hass = hass
kwargs[ATTR_TITLE] = title.async_render()
if targets.get(service.service) is not None:
kwargs[ATTR_TARGET] = [targets[service.service]]
elif service.data.get(ATTR_TARGET) is not None:
kwargs[ATTR_TARGET] = service.data.get(ATTR_TARGET)
message.hass = hass
kwargs[ATTR_MESSAGE] = message.async_render()
kwargs[ATTR_DATA] = service.data.get(ATTR_DATA)
await notify_service.async_send_message(**kwargs)
if hasattr(notify_service, "targets"):
platform_name = (
p_config.get(CONF_NAME) or discovery_info.get(CONF_NAME) or p_type
)
for name, target in notify_service.targets.items():
target_name = slugify(f"{platform_name}_{name}")
targets[target_name] = target
hass.services.async_register(
DOMAIN,
target_name,
async_notify_message,
schema=NOTIFY_SERVICE_SCHEMA,
)
platform_name = (
friendly_name = (
p_config.get(CONF_NAME) or discovery_info.get(CONF_NAME) or SERVICE_NOTIFY
)
platform_name_slug = slugify(platform_name)
hass.services.async_register(
DOMAIN,
platform_name_slug,
async_notify_message,
schema=NOTIFY_SERVICE_SCHEMA,
)
hass.data[NOTIFY_SERVICES][integration_name] = {
FRIENDLY_NAME: friendly_name,
# The targets use a slightly different friendly name
# selection pattern than the base service
TARGET_FRIENDLY_NAME: target_friendly_name,
SERVICE: notify_service,
TARGETS: {},
}
hass.config.components.add(f"{DOMAIN}.{p_type}")
await async_reload(hass, integration_name)
hass.config.components.add(f"{DOMAIN}.{integration_name}")
return True
setup_tasks = [
async_setup_platform(p_type, p_config)
for p_type, p_config in config_per_platform(config, DOMAIN)
async_setup_platform(integration_name, p_config)
for integration_name, p_config in config_per_platform(config, DOMAIN)
]
if setup_tasks:

29
tests/components/mobile_app/test_notify.py
View file

@ -61,6 +61,35 @@ async def setup_push_receiver(hass, aioclient_mock):
await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
await hass.async_block_till_done()
loaded_late_entry = MockConfigEntry(
connection_class="cloud_push",
data={
"app_data": {"push_token": "PUSH_TOKEN2", "push_url": f"{push_url}2"},
"app_id": "io.homeassistant.mobile_app",
"app_name": "mobile_app tests",
"app_version": "1.0",
"device_id": "4d5e6f2",
"device_name": "Loaded Late",
"manufacturer": "Home Assistant",
"model": "mobile_app",
"os_name": "Linux",
"os_version": "5.0.6",
"secret": "123abc2",
"supports_encryption": False,
"user_id": "1a2b3c2",
"webhook_id": "webhook_id_2",
},
domain=DOMAIN,
source="registration",
title="mobile_app 2 test entry",
version=1,
)
loaded_late_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(loaded_late_entry.entry_id)
await hass.async_block_till_done()
assert hass.services.has_service("notify", "mobile_app_loaded_late")
async def test_notify_works(hass, aioclient_mock, setup_push_receiver):
"""Test notify works."""