Remove deprecated yaml and code cleanup for nfandroidtv (#73227)

This commit is contained in:
Rami Mosleh 2022-06-09 12:06:59 +03:00 committed by GitHub
parent 86723ea02b
commit a7398b8a73
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 38 additions and 147 deletions

View file

@ -1,58 +1,46 @@
"""The NFAndroidTV integration."""
from notifications_android_tv.notifications import ConnectError, Notifications
from homeassistant.components.notify import DOMAIN as NOTIFY
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PLATFORM, Platform
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN
from .const import DATA_HASS_CONFIG, DOMAIN
PLATFORMS = [Platform.NOTIFY]
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the NFAndroidTV component."""
hass.data.setdefault(DOMAIN, {})
# Iterate all entries for notify to only get nfandroidtv
if NOTIFY in config:
for entry in config[NOTIFY]:
if entry[CONF_PLATFORM] == DOMAIN:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=entry
)
)
hass.data[DATA_HASS_CONFIG] = config
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up NFAndroidTV from a config entry."""
host = entry.data[CONF_HOST]
name = entry.data[CONF_NAME]
try:
await hass.async_add_executor_job(Notifications, host)
await hass.async_add_executor_job(Notifications, entry.data[CONF_HOST])
except ConnectError as ex:
raise ConfigEntryNotReady("Failed to connect") from ex
raise ConfigEntryNotReady(
f"Failed to connect to host: {entry.data[CONF_HOST]}"
) from ex
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = {
CONF_HOST: host,
CONF_NAME: name,
}
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.NOTIFY,
DOMAIN,
hass.data[DOMAIN][entry.entry_id],
hass.data[DOMAIN],
dict(entry.data),
hass.data[DATA_HASS_CONFIG],
)
)
@ -61,9 +49,4 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View file

@ -26,46 +26,28 @@ class NFAndroidTVFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors = {}
if user_input is not None:
host = user_input[CONF_HOST]
name = user_input[CONF_NAME]
await self.async_set_unique_id(host)
self._abort_if_unique_id_configured()
error = await self._async_try_connect(host)
if error is None:
self._async_abort_entries_match(
{CONF_HOST: user_input[CONF_HOST], CONF_NAME: user_input[CONF_NAME]}
)
if not (error := await self._async_try_connect(user_input[CONF_HOST])):
return self.async_create_entry(
title=name,
data={CONF_HOST: host, CONF_NAME: name},
title=user_input[CONF_NAME],
data=user_input,
)
errors["base"] = error
user_input = user_input or {}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST, default=user_input.get(CONF_HOST)): str,
vol.Required(
CONF_NAME, default=user_input.get(CONF_NAME, DEFAULT_NAME)
): str,
vol.Required(CONF_HOST): str,
vol.Required(CONF_NAME, default=DEFAULT_NAME): str,
}
),
errors=errors,
)
async def async_step_import(self, import_config: dict[str, Any]) -> FlowResult:
"""Import a config entry from configuration.yaml."""
for entry in self._async_current_entries():
if entry.data[CONF_HOST] == import_config[CONF_HOST]:
_LOGGER.warning(
"Already configured. This yaml configuration has already been imported. Please remove it"
)
return self.async_abort(reason="already_configured")
if CONF_NAME not in import_config:
import_config[CONF_NAME] = f"{DEFAULT_NAME} {import_config[CONF_HOST]}"
return await self.async_step_user(import_config)
async def _async_try_connect(self, host: str) -> str | None:
"""Try connecting to Android TV / Fire TV."""
try:

View file

@ -7,6 +7,8 @@ CONF_TRANSPARENCY = "transparency"
CONF_COLOR = "color"
CONF_INTERRUPT = "interrupt"
DATA_HASS_CONFIG = "nfandroid_hass_config"
DEFAULT_NAME = "Android TV / Fire TV"
DEFAULT_TIMEOUT = 5

View file

@ -14,10 +14,9 @@ from homeassistant.components.notify import (
ATTR_DATA,
ATTR_TITLE,
ATTR_TITLE_DEFAULT,
PLATFORM_SCHEMA,
BaseNotificationService,
)
from homeassistant.const import CONF_HOST, CONF_TIMEOUT
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
@ -43,53 +42,21 @@ from .const import (
ATTR_INTERRUPT,
ATTR_POSITION,
ATTR_TRANSPARENCY,
CONF_COLOR,
CONF_DURATION,
CONF_FONTSIZE,
CONF_INTERRUPT,
CONF_POSITION,
CONF_TRANSPARENCY,
DEFAULT_TIMEOUT,
)
_LOGGER = logging.getLogger(__name__)
# Deprecated in Home Assistant 2021.8
PLATFORM_SCHEMA = cv.deprecated(
vol.All(
PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_DURATION): vol.Coerce(int),
vol.Optional(CONF_FONTSIZE): vol.In(Notifications.FONTSIZES.keys()),
vol.Optional(CONF_POSITION): vol.In(Notifications.POSITIONS.keys()),
vol.Optional(CONF_TRANSPARENCY): vol.In(
Notifications.TRANSPARENCIES.keys()
),
vol.Optional(CONF_COLOR): vol.In(Notifications.BKG_COLORS.keys()),
vol.Optional(CONF_TIMEOUT): vol.Coerce(int),
vol.Optional(CONF_INTERRUPT): cv.boolean,
}
),
)
)
async def async_get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> NFAndroidTVNotificationService:
) -> NFAndroidTVNotificationService | None:
"""Get the NFAndroidTV notification service."""
if discovery_info is not None:
notify = await hass.async_add_executor_job(
Notifications, discovery_info[CONF_HOST]
)
return NFAndroidTVNotificationService(
notify,
hass.config.is_allowed_path,
)
notify = await hass.async_add_executor_job(Notifications, config.get(CONF_HOST))
if discovery_info is None:
return None
notify = await hass.async_add_executor_job(Notifications, discovery_info[CONF_HOST])
return NFAndroidTVNotificationService(
notify,
hass.config.is_allowed_path,
@ -128,21 +95,21 @@ class NFAndroidTVNotificationService(BaseNotificationService):
)
except ValueError:
_LOGGER.warning(
"Invalid duration-value: %s", str(data.get(ATTR_DURATION))
"Invalid duration-value: %s", data.get(ATTR_DURATION)
)
if ATTR_FONTSIZE in data:
if data.get(ATTR_FONTSIZE) in Notifications.FONTSIZES:
fontsize = data.get(ATTR_FONTSIZE)
else:
_LOGGER.warning(
"Invalid fontsize-value: %s", str(data.get(ATTR_FONTSIZE))
"Invalid fontsize-value: %s", data.get(ATTR_FONTSIZE)
)
if ATTR_POSITION in data:
if data.get(ATTR_POSITION) in Notifications.POSITIONS:
position = data.get(ATTR_POSITION)
else:
_LOGGER.warning(
"Invalid position-value: %s", str(data.get(ATTR_POSITION))
"Invalid position-value: %s", data.get(ATTR_POSITION)
)
if ATTR_TRANSPARENCY in data:
if data.get(ATTR_TRANSPARENCY) in Notifications.TRANSPARENCIES:
@ -150,24 +117,21 @@ class NFAndroidTVNotificationService(BaseNotificationService):
else:
_LOGGER.warning(
"Invalid transparency-value: %s",
str(data.get(ATTR_TRANSPARENCY)),
data.get(ATTR_TRANSPARENCY),
)
if ATTR_COLOR in data:
if data.get(ATTR_COLOR) in Notifications.BKG_COLORS:
bkgcolor = data.get(ATTR_COLOR)
else:
_LOGGER.warning(
"Invalid color-value: %s", str(data.get(ATTR_COLOR))
)
_LOGGER.warning("Invalid color-value: %s", data.get(ATTR_COLOR))
if ATTR_INTERRUPT in data:
try:
interrupt = cv.boolean(data.get(ATTR_INTERRUPT))
except vol.Invalid:
_LOGGER.warning(
"Invalid interrupt-value: %s", str(data.get(ATTR_INTERRUPT))
"Invalid interrupt-value: %s", data.get(ATTR_INTERRUPT)
)
imagedata = data.get(ATTR_IMAGE) if data else None
if imagedata is not None:
if imagedata := data.get(ATTR_IMAGE):
image_file = self.load_file(
url=imagedata.get(ATTR_IMAGE_URL),
local_path=imagedata.get(ATTR_IMAGE_PATH),
@ -175,8 +139,7 @@ class NFAndroidTVNotificationService(BaseNotificationService):
password=imagedata.get(ATTR_IMAGE_PASSWORD),
auth=imagedata.get(ATTR_IMAGE_AUTH),
)
icondata = data.get(ATTR_ICON) if data else None
if icondata is not None:
if icondata := data.get(ATTR_ICON):
icon = self.load_file(
url=icondata.get(ATTR_ICON_URL),
local_path=icondata.get(ATTR_ICON_PATH),

View file

@ -4,8 +4,7 @@ from unittest.mock import patch
from notifications_android_tv.notifications import ConnectError
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.nfandroidtv.const import DEFAULT_NAME, DOMAIN
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.components.nfandroidtv.const import DOMAIN
from . import (
CONF_CONFIG_FLOW,
@ -95,41 +94,3 @@ async def test_flow_user_unknown_error(hass):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "unknown"}
async def test_flow_import(hass):
"""Test an import flow."""
mocked_tv = await _create_mocked_tv(True)
with _patch_config_flow_tv(mocked_tv), _patch_setup():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=CONF_CONFIG_FLOW,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == CONF_DATA
with _patch_config_flow_tv(mocked_tv), _patch_setup():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=CONF_CONFIG_FLOW,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_flow_import_missing_optional(hass):
"""Test an import flow with missing options."""
mocked_tv = await _create_mocked_tv(True)
with _patch_config_flow_tv(mocked_tv), _patch_setup():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={CONF_HOST: HOST},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {CONF_HOST: HOST, CONF_NAME: f"{DEFAULT_NAME} {HOST}"}