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

Add config flow to nfandroidtv (#51280)

Browse source 
Co-authored-by: Franck Nijhof <git@frenck.dev>
This commit is contained in:
Robert Hillis 2021-07-21 07:31:54 -04:00 • committed by GitHub
parent 02a7a2464a
commit 462db1b4b2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 512 additions and 185 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -693,6 +693,7 @@ omit =
homeassistant/components/neurio_energy/sensor.py
homeassistant/components/nexia/climate.py
homeassistant/components/nextcloud/*
homeassistant/components/nfandroidtv/__init__.py
homeassistant/components/nfandroidtv/notify.py
homeassistant/components/niko_home_control/light.py
homeassistant/components/nilu/air_quality.py

1
CODEOWNERS
View file

@ -332,6 +332,7 @@ homeassistant/components/netdata/* @fabaff
homeassistant/components/nexia/* @bdraco
homeassistant/components/nextbus/* @vividboarder
homeassistant/components/nextcloud/* @meichthys
homeassistant/components/nfandroidtv/* @tkdrob
homeassistant/components/nightscout/* @marciogranzotto
homeassistant/components/nilu/* @hfurubotten
homeassistant/components/nissan_leaf/* @filcole

70
homeassistant/components/nfandroidtv/__init__.py
View file

@ -1 +1,69 @@
"""The nfandroidtv component."""
"""The NFAndroidTV integration."""
import logging
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
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import discovery
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [NOTIFY]
async def async_setup(hass: HomeAssistant, config):
"""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
)
)
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)
except ConnectError as ex:
_LOGGER.warning("Failed to connect: %s", ex)
raise ConfigEntryNotReady 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, NOTIFY, DOMAIN, hass.data[DOMAIN][entry.entry_id], hass.data[DOMAIN]
)
)
return True
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

76
homeassistant/components/nfandroidtv/config_flow.py Normal file
View file

@ -0,0 +1,76 @@
"""Config flow for NFAndroidTV integration."""
from __future__ import annotations
import logging
from notifications_android_tv.notifications import ConnectError, Notifications
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.data_entry_flow import FlowResult
from .const import DEFAULT_NAME, DOMAIN
_LOGGER = logging.getLogger(__name__)
class NFAndroidTVFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for NFAndroidTV."""
async def async_step_user(self, user_input=None) -> FlowResult:
"""Handle a flow initiated by the user."""
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:
return self.async_create_entry(
title=name,
data={CONF_HOST: host, CONF_NAME: name},
)
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,
}
),
errors=errors,
)
async def async_step_import(self, import_config):
"""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):
"""Try connecting to Android TV / Fire TV."""
try:
await self.hass.async_add_executor_job(Notifications, host)
except ConnectError:
_LOGGER.error("Error connecting to device at %s", host)
return "cannot_connect"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return "unknown"
return

28
homeassistant/components/nfandroidtv/const.py Normal file
View file

@ -0,0 +1,28 @@
"""Constants for the NFAndroidTV integration."""
DOMAIN: str = "nfandroidtv"
CONF_DURATION = "duration"
CONF_FONTSIZE = "fontsize"
CONF_POSITION = "position"
CONF_TRANSPARENCY = "transparency"
CONF_COLOR = "color"
CONF_INTERRUPT = "interrupt"
DEFAULT_NAME = "Android TV / Fire TV"
DEFAULT_TIMEOUT = 5
ATTR_DURATION = "duration"
ATTR_FONTSIZE = "fontsize"
ATTR_POSITION = "position"
ATTR_TRANSPARENCY = "transparency"
ATTR_COLOR = "color"
ATTR_BKGCOLOR = "bkgcolor"
ATTR_INTERRUPT = "interrupt"
ATTR_FILE = "file"
# Attributes contained in file
ATTR_FILE_URL = "url"
ATTR_FILE_PATH = "path"
ATTR_FILE_USERNAME = "username"
ATTR_FILE_PASSWORD = "password"
ATTR_FILE_AUTH = "auth"
# Any other value or absence of 'auth' lead to basic authentication being used
ATTR_FILE_AUTH_DIGEST = "digest"

6
homeassistant/components/nfandroidtv/manifest.json
View file

@ -1,7 +1,9 @@
{
"domain": "nfandroidtv",
"name": "Notifications for Android TV / FireTV",
"name": "Notifications for Android TV / Fire TV",
"documentation": "https://www.home-assistant.io/integrations/nfandroidtv",
"codeowners": [],
"requirements": ["notifications-android-tv==0.1.2"],
"codeowners": ["@tkdrob"],
"config_flow": true,
"iot_class": "local_push"
}

300
homeassistant/components/nfandroidtv/notify.py
View file

@ -1,8 +1,7 @@
"""Notifications for Android TV notification service."""
import base64
import io
import logging
from notifications_android_tv import Notifications
import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
import voluptuous as vol
@ -14,115 +13,69 @@ from homeassistant.components.notify import (
PLATFORM_SCHEMA,
BaseNotificationService,
)
from homeassistant.const import CONF_HOST, CONF_TIMEOUT, HTTP_OK, PERCENTAGE
from homeassistant.const import ATTR_ICON, CONF_HOST, CONF_TIMEOUT
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from .const import (
ATTR_COLOR,
ATTR_DURATION,
ATTR_FILE,
ATTR_FILE_AUTH,
ATTR_FILE_AUTH_DIGEST,
ATTR_FILE_PASSWORD,
ATTR_FILE_PATH,
ATTR_FILE_URL,
ATTR_FILE_USERNAME,
ATTR_FONTSIZE,
ATTR_INTERRUPT,
ATTR_POSITION,
ATTR_TRANSPARENCY,
CONF_COLOR,
CONF_DURATION,
CONF_FONTSIZE,
CONF_INTERRUPT,
CONF_POSITION,
CONF_TRANSPARENCY,
DEFAULT_TIMEOUT,
)
_LOGGER = logging.getLogger(__name__)
CONF_DURATION = "duration"
CONF_FONTSIZE = "fontsize"
CONF_POSITION = "position"
CONF_TRANSPARENCY = "transparency"
CONF_COLOR = "color"
CONF_INTERRUPT = "interrupt"
DEFAULT_DURATION = 5
DEFAULT_FONTSIZE = "medium"
DEFAULT_POSITION = "bottom-right"
DEFAULT_TRANSPARENCY = "default"
DEFAULT_COLOR = "grey"
DEFAULT_INTERRUPT = False
DEFAULT_TIMEOUT = 5
DEFAULT_ICON = (
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR4nGP6zwAAAgcBApo"
"cMXEAAAAASUVORK5CYII="
)
ATTR_DURATION = "duration"
ATTR_FONTSIZE = "fontsize"
ATTR_POSITION = "position"
ATTR_TRANSPARENCY = "transparency"
ATTR_COLOR = "color"
ATTR_BKGCOLOR = "bkgcolor"
ATTR_INTERRUPT = "interrupt"
ATTR_IMAGE = "filename2"
ATTR_FILE = "file"
# Attributes contained in file
ATTR_FILE_URL = "url"
ATTR_FILE_PATH = "path"
ATTR_FILE_USERNAME = "username"
ATTR_FILE_PASSWORD = "password"
ATTR_FILE_AUTH = "auth"
# Any other value or absence of 'auth' lead to basic authentication being used
ATTR_FILE_AUTH_DIGEST = "digest"
FONTSIZES = {"small": 1, "medium": 0, "large": 2, "max": 3}
POSITIONS = {
"bottom-right": 0,
"bottom-left": 1,
"top-right": 2,
"top-left": 3,
"center": 4,
}
TRANSPARENCIES = {
"default": 0,
f"0{PERCENTAGE}": 1,
f"25{PERCENTAGE}": 2,
f"50{PERCENTAGE}": 3,
f"75{PERCENTAGE}": 4,
f"100{PERCENTAGE}": 5,
}
COLORS = {
"grey": "#607d8b",
"black": "#000000",
"indigo": "#303F9F",
"green": "#4CAF50",
"red": "#F44336",
"cyan": "#00BCD4",
"teal": "#009688",
"amber": "#FFC107",
"pink": "#E91E63",
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_DURATION, default=DEFAULT_DURATION): vol.Coerce(int),
vol.Optional(CONF_FONTSIZE, default=DEFAULT_FONTSIZE): vol.In(FONTSIZES.keys()),
vol.Optional(CONF_POSITION, default=DEFAULT_POSITION): vol.In(POSITIONS.keys()),
vol.Optional(CONF_TRANSPARENCY, default=DEFAULT_TRANSPARENCY): vol.In(
TRANSPARENCIES.keys()
# 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,
}
),
vol.Optional(CONF_COLOR, default=DEFAULT_COLOR): vol.In(COLORS.keys()),
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): vol.Coerce(int),
vol.Optional(CONF_INTERRUPT, default=DEFAULT_INTERRUPT): cv.boolean,
}
)
)
def get_service(hass, config, discovery_info=None):
"""Get the Notifications for Android TV notification service."""
remoteip = config.get(CONF_HOST)
duration = config.get(CONF_DURATION)
fontsize = config.get(CONF_FONTSIZE)
position = config.get(CONF_POSITION)
transparency = config.get(CONF_TRANSPARENCY)
color = config.get(CONF_COLOR)
interrupt = config.get(CONF_INTERRUPT)
timeout = config.get(CONF_TIMEOUT)
async def async_get_service(hass: HomeAssistant, config, discovery_info=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))
return NFAndroidTVNotificationService(
remoteip,
duration,
fontsize,
position,
transparency,
color,
interrupt,
timeout,
notify,
hass.config.is_allowed_path,
)
@ -132,116 +85,98 @@ class NFAndroidTVNotificationService(BaseNotificationService):
def __init__(
self,
remoteip,
duration,
fontsize,
position,
transparency,
color,
interrupt,
timeout,
notify: Notifications,
is_allowed_path,
):
"""Initialize the service."""
self._target = f"http://{remoteip}:7676"
self._default_duration = duration
self._default_fontsize = fontsize
self._default_position = position
self._default_transparency = transparency
self._default_color = color
self._default_interrupt = interrupt
self._timeout = timeout
self._icon_file = io.BytesIO(base64.b64decode(DEFAULT_ICON))
self.notify = notify
self.is_allowed_path = is_allowed_path
def send_message(self, message="", **kwargs):
"""Send a message to a Android TV device."""
_LOGGER.debug("Sending notification to: %s", self._target)
payload = {
"filename": (
"icon.png",
self._icon_file,
"application/octet-stream",
{"Expires": "0"},
),
"type": "0",
"title": kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT),
"msg": message,
"duration": "%i" % self._default_duration,
"fontsize": "%i" % FONTSIZES.get(self._default_fontsize),
"position": "%i" % POSITIONS.get(self._default_position),
"bkgcolor": "%s" % COLORS.get(self._default_color),
"transparency": "%i" % TRANSPARENCIES.get(self._default_transparency),
"offset": "0",
"app": ATTR_TITLE_DEFAULT,
"force": "true",
"interrupt": "%i" % self._default_interrupt,
}
data = kwargs.get(ATTR_DATA)
title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
duration = None
fontsize = None
position = None
transparency = None
bkgcolor = None
interrupt = None
icon = None
image_file = None
if data:
if ATTR_DURATION in data:
duration = data.get(ATTR_DURATION)
try:
payload[ATTR_DURATION] = "%i" % int(duration)
duration = int(data.get(ATTR_DURATION))
except ValueError:
_LOGGER.warning("Invalid duration-value: %s", str(duration))
_LOGGER.warning(
"Invalid duration-value: %s", str(data.get(ATTR_DURATION))
)
if ATTR_FONTSIZE in data:
fontsize = data.get(ATTR_FONTSIZE)
if fontsize in FONTSIZES:
payload[ATTR_FONTSIZE] = "%i" % FONTSIZES.get(fontsize)
if data.get(ATTR_FONTSIZE) in Notifications.FONTSIZES:
fontsize = data.get(ATTR_FONTSIZE)
else:
_LOGGER.warning("Invalid fontsize-value: %s", str(fontsize))
_LOGGER.warning(
"Invalid fontsize-value: %s", str(data.get(ATTR_FONTSIZE))
)
if ATTR_POSITION in data:
position = data.get(ATTR_POSITION)
if position in POSITIONS:
payload[ATTR_POSITION] = "%i" % POSITIONS.get(position)
if data.get(ATTR_POSITION) in Notifications.POSITIONS:
position = data.get(ATTR_POSITION)
else:
_LOGGER.warning("Invalid position-value: %s", str(position))
_LOGGER.warning(
"Invalid position-value: %s", str(data.get(ATTR_POSITION))
)
if ATTR_TRANSPARENCY in data:
transparency = data.get(ATTR_TRANSPARENCY)
if transparency in TRANSPARENCIES:
payload[ATTR_TRANSPARENCY] = "%i" % TRANSPARENCIES.get(transparency)
if data.get(ATTR_TRANSPARENCY) in Notifications.TRANSPARENCIES:
transparency = data.get(ATTR_TRANSPARENCY)
else:
_LOGGER.warning("Invalid transparency-value: %s", str(transparency))
_LOGGER.warning(
"Invalid transparency-value: %s",
str(data.get(ATTR_TRANSPARENCY)),
)
if ATTR_COLOR in data:
color = data.get(ATTR_COLOR)
if color in COLORS:
payload[ATTR_BKGCOLOR] = "%s" % COLORS.get(color)
if data.get(ATTR_COLOR) in Notifications.BKG_COLORS:
bkgcolor = data.get(ATTR_COLOR)
else:
_LOGGER.warning("Invalid color-value: %s", str(color))
_LOGGER.warning(
"Invalid color-value: %s", str(data.get(ATTR_COLOR))
)
if ATTR_INTERRUPT in data:
interrupt = data.get(ATTR_INTERRUPT)
try:
payload[ATTR_INTERRUPT] = "%i" % cv.boolean(interrupt)
interrupt = cv.boolean(data.get(ATTR_INTERRUPT))
except vol.Invalid:
_LOGGER.warning("Invalid interrupt-value: %s", str(interrupt))
_LOGGER.warning(
"Invalid interrupt-value: %s", str(data.get(ATTR_INTERRUPT))
)
filedata = data.get(ATTR_FILE) if data else None
if filedata is not None:
# Load from file or URL
file_as_bytes = self.load_file(
if ATTR_ICON in filedata:
icon = self.load_file(
url=filedata.get(ATTR_ICON),
local_path=filedata.get(ATTR_FILE_PATH),
username=filedata.get(ATTR_FILE_USERNAME),
password=filedata.get(ATTR_FILE_PASSWORD),
auth=filedata.get(ATTR_FILE_AUTH),
)
image_file = self.load_file(
url=filedata.get(ATTR_FILE_URL),
local_path=filedata.get(ATTR_FILE_PATH),
username=filedata.get(ATTR_FILE_USERNAME),
password=filedata.get(ATTR_FILE_PASSWORD),
auth=filedata.get(ATTR_FILE_AUTH),
)
if file_as_bytes:
payload[ATTR_IMAGE] = (
"image",
file_as_bytes,
"application/octet-stream",
{"Expires": "0"},
)
try:
_LOGGER.debug("Payload: %s", str(payload))
response = requests.post(self._target, files=payload, timeout=self._timeout)
if response.status_code != HTTP_OK:
_LOGGER.error("Error sending message: %s", str(response))
except requests.exceptions.ConnectionError as err:
_LOGGER.error("Error communicating with %s: %s", self._target, str(err))
self.notify.send(
message,
title=title,
duration=duration,
fontsize=fontsize,
position=position,
bkgcolor=bkgcolor,
transparency=transparency,
interrupt=interrupt,
icon=icon,
image_file=image_file,
)
def load_file(
self, url=None, local_path=None, username=None, password=None, auth=None
@ -266,7 +201,8 @@ class NFAndroidTVNotificationService(BaseNotificationService):
if local_path is not None:
# Check whether path is whitelisted in configuration.yaml
if self.is_allowed_path(local_path):
return open(local_path, "rb") # pylint: disable=consider-using-with
with open(local_path, "rb") as path_handle:
return path_handle
_LOGGER.warning("'%s' is not secure to load data from!", local_path)
else:
_LOGGER.warning("Neither URL nor local path found in params!")

21
homeassistant/components/nfandroidtv/strings.json Normal file
View file

@ -0,0 +1,21 @@
{
"config": {
"step": {
"user": {
"title": "Notifications for Android TV / Fire TV",
"description": "This integration requires the Notifications for Android TV app.\n\nFor Android TV: https://play.google.com/store/apps/details?id=de.cyberdream.androidtv.notifications.google\nFor Fire TV: https://www.amazon.com/Christian-Fees-Notifications-for-Fire/dp/B00OESCXEK\n\nYou should set up either DHCP reservation on your router (refer to your router's user manual) or a static IP address on the device. If not, the device will eventually become unavailable.",
"data": {
"host": "[%key:common::config_flow::data::host%]",
"name": "[%key:common::config_flow::data::name%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

21
homeassistant/components/nfandroidtv/translations/en.json Normal file
View file

@ -0,0 +1,21 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"host": "Host",
"name": "Name"
},
"description": "This integration requires the Notifications for Android TV app.\n\nFor Android TV: https://play.google.com/store/apps/details?id=de.cyberdream.androidtv.notifications.google\nFor Fire TV: https://www.amazon.com/Christian-Fees-Notifications-for-Fire/dp/B00OESCXEK\n\nYou should set up either DHCP reservation on your router (refer to your router's user manual) or a static IP address on the device. If not, the device will eventually become unavailable.",
"title": "Notifications for Android TV / Fire TV"
}
}
}
}

1
homeassistant/generated/config_flows.py
View file

@ -178,6 +178,7 @@ FLOWS = [
"nest",
"netatmo",
"nexia",
"nfandroidtv",
"nightscout",
"notion",
"nuheat",

3
requirements_all.txt
View file

@ -1032,6 +1032,9 @@ niluclient==0.1.2
# homeassistant.components.noaa_tides
noaa-coops==0.1.8
# homeassistant.components.nfandroidtv
notifications-android-tv==0.1.2
# homeassistant.components.notify_events
notify-events==1.0.4

3
requirements_test_all.txt
View file

@ -577,6 +577,9 @@ nettigo-air-monitor==1.0.0
# homeassistant.components.nexia
nexia==0.9.10
# homeassistant.components.nfandroidtv
notifications-android-tv==0.1.2
# homeassistant.components.notify_events
notify-events==1.0.4

31
tests/components/nfandroidtv/__init__.py Normal file
View file

@ -0,0 +1,31 @@
"""Tests for the NFAndroidTV integration."""
from unittest.mock import AsyncMock, patch
from homeassistant.const import CONF_HOST, CONF_NAME
HOST = "1.2.3.4"
NAME = "Android TV / Fire TV"
CONF_DATA = {
CONF_HOST: HOST,
CONF_NAME: NAME,
}
CONF_CONFIG_FLOW = {
CONF_HOST: HOST,
CONF_NAME: NAME,
}
async def _create_mocked_tv(raise_exception=False):
mocked_tv = AsyncMock()
mocked_tv.get_state = AsyncMock()
return mocked_tv
def _patch_config_flow_tv(mocked_tv):
return patch(
"homeassistant.components.nfandroidtv.config_flow.Notifications",
return_value=mocked_tv,
)

135
tests/components/nfandroidtv/test_config_flow.py Normal file
View file

@ -0,0 +1,135 @@
"""Test NFAndroidTV config flow."""
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 . import (
CONF_CONFIG_FLOW,
CONF_DATA,
HOST,
NAME,
_create_mocked_tv,
_patch_config_flow_tv,
)
from tests.common import MockConfigEntry
def _patch_setup():
return patch(
"homeassistant.components.nfandroidtv.async_setup_entry",
return_value=True,
)
async def test_flow_user(hass):
"""Test user initialized flow."""
mocked_tv = await _create_mocked_tv()
with _patch_config_flow_tv(mocked_tv), _patch_setup():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_CONFIG_FLOW,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == NAME
assert result["data"] == CONF_DATA
async def test_flow_user_already_configured(hass):
"""Test user initialized flow with duplicate server."""
entry = MockConfigEntry(
domain=DOMAIN,
data=CONF_CONFIG_FLOW,
unique_id=HOST,
)
entry.add_to_hass(hass)
mocked_tv = await _create_mocked_tv()
with _patch_config_flow_tv(mocked_tv), _patch_setup():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_CONFIG_FLOW,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_flow_user_cannot_connect(hass):
"""Test user initialized flow with unreachable server."""
mocked_tv = await _create_mocked_tv(True)
with _patch_config_flow_tv(mocked_tv) as tvmock:
tvmock.side_effect = ConnectError
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=CONF_CONFIG_FLOW,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
async def test_flow_user_unknown_error(hass):
"""Test user initialized flow with unreachable server."""
mocked_tv = await _create_mocked_tv(True)
with _patch_config_flow_tv(mocked_tv) as tvmock:
tvmock.side_effect = Exception
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=CONF_CONFIG_FLOW,
)
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}"}