Add config flow for braviatv integration (#33774)

* Run scripts

* Improvement strings

* Fix FlowOptions update listener

* Update .ceveragerc

* Add tests

* Better strings

* Add test for OptionsFlow

* Run gen_requirements_all.py once again

* Fix pylint errors

* Log error when there is no bravia.conf file during import

* Improvement strings

* Use braviarc object from hass.data in options flow

* Use async_add_executor_job for IO

* Fix options flow test

* Fix tests

* Remove host_reachable method

* Remove dependencies

* Change setup_platform method to async

* Remove calling system_info

* Save mac in the config entry

* Fix get ignore sources

* Fix read config from file

* Remove the side effect from init

* Fix user_input for user step

* Switch OrderedDict to dict

* New config_entry instance for each test

* Revert change

* Patch async_setup_entry in test_import

* Change a way to create source list

* Consolidate repeated block of code

* Update tests

* Suggested change

Co-Authored-By: Martin Hjelmare <marhje52@gmail.com>

* Suggested channge

Co-Authored-By: Martin Hjelmare <marhje52@gmail.com>

* Suggested change

* Patch async_setup_entry

* Remove unnecesary if

* suggested change

* Suggested change

* Fix tests

* Fix pylint error

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Maciej Bieniek 2020-04-15 01:04:06 +02:00 committed by GitHub
parent 075030f15a
commit 6dc6f2d099
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 642 additions and 124 deletions

View file

@ -80,6 +80,8 @@ omit =
homeassistant/components/bom/camera.py
homeassistant/components/bom/sensor.py
homeassistant/components/bom/weather.py
homeassistant/components/braviatv/__init__.py
homeassistant/components/braviatv/const.py
homeassistant/components/braviatv/media_player.py
homeassistant/components/broadlink/remote.py
homeassistant/components/broadlink/sensor.py

View file

@ -55,7 +55,7 @@ homeassistant/components/blink/* @fronzbot
homeassistant/components/bmp280/* @belidzs
homeassistant/components/bmw_connected_drive/* @gerard33
homeassistant/components/bom/* @maddenp
homeassistant/components/braviatv/* @robbiet480
homeassistant/components/braviatv/* @robbiet480 @bieniu
homeassistant/components/broadlink/* @danielhiversen @felipediel
homeassistant/components/brother/* @bieniu
homeassistant/components/brunt/* @eavanvalkenburg

View file

@ -1 +1,56 @@
"""The braviatv component."""
"""The Bravia TV component."""
import asyncio
from bravia_tv import BraviaRC
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN
from homeassistant.exceptions import ConfigEntryNotReady
from .const import CLIENTID_PREFIX, DOMAIN, NICKNAME
PLATFORMS = ["media_player"]
async def async_setup(hass, config):
"""Set up the Bravia TV component."""
return True
async def async_setup_entry(hass, config_entry):
"""Set up a config entry."""
host = config_entry.data[CONF_HOST]
mac = config_entry.data[CONF_MAC]
pin = config_entry.data[CONF_PIN]
braviarc = BraviaRC(host, mac)
await hass.async_add_executor_job(braviarc.connect, pin, CLIENTID_PREFIX, NICKNAME)
if not braviarc.is_connected():
raise ConfigEntryNotReady
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][config_entry.entry_id] = braviarc
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, component)
)
return True
async def async_unload_entry(hass, config_entry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN].pop(config_entry.entry_id)
return unload_ok

View file

@ -0,0 +1,190 @@
"""Adds config flow for Bravia TV integration."""
import ipaddress
import logging
import re
from bravia_tv import BraviaRC
import voluptuous as vol
from homeassistant import config_entries, exceptions
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from .const import ( # pylint:disable=unused-import
ATTR_CID,
ATTR_MAC,
ATTR_MODEL,
CLIENTID_PREFIX,
CONF_IGNORED_SOURCES,
DOMAIN,
NICKNAME,
)
_LOGGER = logging.getLogger(__name__)
def host_valid(host):
"""Return True if hostname or IP address is valid."""
try:
if ipaddress.ip_address(host).version == (4 or 6):
return True
except ValueError:
disallowed = re.compile(r"[^a-zA-Z\d\-]")
return all(x and not disallowed.search(x) for x in host.split("."))
class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for BraviaTV integration."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
"""Initialize."""
self.braviarc = None
self.host = None
self.title = None
self.mac = None
async def init_device(self, pin):
"""Initialize Bravia TV device."""
await self.hass.async_add_executor_job(
self.braviarc.connect, pin, CLIENTID_PREFIX, NICKNAME,
)
if not self.braviarc.is_connected():
raise CannotConnect()
try:
system_info = await self.hass.async_add_executor_job(
self.braviarc.get_system_info
)
except (KeyError, TypeError):
raise ModelNotSupported()
await self.async_set_unique_id(system_info[ATTR_CID].lower())
self._abort_if_unique_id_configured()
self.title = system_info[ATTR_MODEL]
self.mac = system_info[ATTR_MAC]
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Bravia TV options callback."""
return BraviaTVOptionsFlowHandler(config_entry)
async def async_step_import(self, user_input=None):
"""Handle configuration by yaml file."""
self.host = user_input[CONF_HOST]
self.braviarc = BraviaRC(self.host)
try:
await self.init_device(user_input[CONF_PIN])
except CannotConnect:
_LOGGER.error("Import aborted, cannot connect to %s", self.host)
return self.async_abort(reason="cannot_connect")
except ModelNotSupported:
_LOGGER.error("Import aborted, your TV is not supported")
return self.async_abort(reason="unsupported_model")
user_input[CONF_MAC] = self.mac
return self.async_create_entry(title=self.title, data=user_input)
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
if host_valid(user_input[CONF_HOST]):
self.host = user_input[CONF_HOST]
self.braviarc = BraviaRC(self.host)
return await self.async_step_authorize()
errors[CONF_HOST] = "invalid_host"
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_HOST, default=""): str}),
errors=errors,
)
async def async_step_authorize(self, user_input=None):
"""Get PIN from the Bravia TV device."""
errors = {}
if user_input is not None:
try:
await self.init_device(user_input[CONF_PIN])
except CannotConnect:
errors["base"] = "cannot_connect"
except ModelNotSupported:
errors["base"] = "unsupported_model"
else:
user_input[CONF_HOST] = self.host
user_input[CONF_MAC] = self.mac
return self.async_create_entry(title=self.title, data=user_input)
# Connecting with th PIN "0000" to start the pairing process on the TV.
await self.hass.async_add_executor_job(
self.braviarc.connect, "0000", CLIENTID_PREFIX, NICKNAME,
)
return self.async_show_form(
step_id="authorize",
data_schema=vol.Schema({vol.Required(CONF_PIN, default=""): str}),
errors=errors,
)
class BraviaTVOptionsFlowHandler(config_entries.OptionsFlow):
"""Config flow options for Bravia TV."""
def __init__(self, config_entry):
"""Initialize Bravia TV options flow."""
self.braviarc = None
self.config_entry = config_entry
self.pin = config_entry.data[CONF_PIN]
self.ignored_sources = config_entry.options.get(CONF_IGNORED_SOURCES)
self.source_list = []
async def async_step_init(self, user_input=None):
"""Manage the options."""
self.braviarc = self.hass.data[DOMAIN][self.config_entry.entry_id]
if not self.braviarc.is_connected():
await self.hass.async_add_executor_job(
self.braviarc.connect, self.pin, CLIENTID_PREFIX, NICKNAME,
)
content_mapping = await self.hass.async_add_executor_job(
self.braviarc.load_source_list
)
self.source_list = [*content_mapping]
return await self.async_step_user()
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Optional(
CONF_IGNORED_SOURCES, default=self.ignored_sources
): cv.multi_select(self.source_list)
}
),
)
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""
class ModelNotSupported(exceptions.HomeAssistantError):
"""Error to indicate not supported model."""

View file

@ -0,0 +1,13 @@
"""Constants for Bravia TV integration."""
ATTR_CID = "cid"
ATTR_MAC = "macAddr"
ATTR_MANUFACTURER = "Sony"
ATTR_MODEL = "model"
CONF_IGNORED_SOURCES = "ignored_sources"
BRAVIA_CONFIG_FILE = "bravia.conf"
CLIENTID_PREFIX = "HomeAssistant"
DEFAULT_NAME = f"{ATTR_MANUFACTURER} Bravia TV"
DOMAIN = "braviatv"
NICKNAME = "Home Assistant"

View file

@ -3,6 +3,6 @@
"name": "Sony Bravia TV",
"documentation": "https://www.home-assistant.io/integrations/braviatv",
"requirements": ["bravia-tv==1.0.1"],
"dependencies": ["configurator"],
"codeowners": ["@robbiet480"]
"codeowners": ["@robbiet480", "@bieniu"],
"config_flow": true
}

View file

@ -1,10 +1,13 @@
"""Support for interface with a Sony Bravia TV."""
"""Support for interface with a Bravia TV."""
import logging
from bravia_tv import BraviaRC
import voluptuous as vol
from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerDevice
from homeassistant.components.media_player import (
DEVICE_CLASS_TV,
PLATFORM_SCHEMA,
MediaPlayerDevice,
)
from homeassistant.components.media_player.const import (
SUPPORT_NEXT_TRACK,
SUPPORT_PAUSE,
@ -18,22 +21,20 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_SET,
SUPPORT_VOLUME_STEP,
)
from homeassistant.const import CONF_HOST, CONF_NAME, STATE_OFF, STATE_ON
from homeassistant.exceptions import PlatformNotReady
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PIN, STATE_OFF, STATE_ON
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import format_mac
from homeassistant.util.json import load_json, save_json
from homeassistant.util.json import load_json
BRAVIA_CONFIG_FILE = "bravia.conf"
CLIENTID_PREFIX = "HomeAssistant"
DEFAULT_NAME = "Sony Bravia TV"
NICKNAME = "Home Assistant"
# Map ip to request id for configuring
_CONFIGURING = {}
from .const import (
ATTR_MANUFACTURER,
BRAVIA_CONFIG_FILE,
CLIENTID_PREFIX,
CONF_IGNORED_SOURCES,
DEFAULT_NAME,
DOMAIN,
NICKNAME,
)
_LOGGER = logging.getLogger(__name__)
@ -59,116 +60,66 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Sony Bravia TV platform."""
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the Bravia TV platform."""
host = config[CONF_HOST]
if host is None:
return
pin = None
bravia_config = load_json(hass.config.path(BRAVIA_CONFIG_FILE))
while bravia_config:
# Set up a configured TV
host_ip, host_config = bravia_config.popitem()
if host_ip == host:
pin = host_config["pin"]
mac = host_config["mac"]
name = config[CONF_NAME]
braviarc = BraviaRC(host, mac)
if not braviarc.connect(pin, CLIENTID_PREFIX, NICKNAME):
raise PlatformNotReady
try:
unique_id = braviarc.get_system_info()["cid"].lower()
except TypeError:
raise PlatformNotReady
add_entities([BraviaTVDevice(braviarc, name, pin, unique_id)])
return
setup_bravia(config, pin, hass, add_entities)
def setup_bravia(config, pin, hass, add_entities):
"""Set up a Sony Bravia TV based on host parameter."""
host = config[CONF_HOST]
name = config[CONF_NAME]
if pin is None:
request_configuration(config, hass, add_entities)
return
# If we came here and configuring this host, mark as done
if host in _CONFIGURING:
request_id = _CONFIGURING.pop(host)
configurator = hass.components.configurator
configurator.request_done(request_id)
_LOGGER.info("Discovery configuration done")
braviarc = BraviaRC(host)
if not braviarc.connect(pin, CLIENTID_PREFIX, NICKNAME):
_LOGGER.error("Cannot connect to %s", host)
return
try:
system_info = braviarc.get_system_info()
except TypeError:
_LOGGER.error("Cannot retrieve system info from %s", host)
return
mac = format_mac(system_info["macAddr"])
unique_id = system_info["cid"].lower()
# Save config
save_json(
hass.config.path(BRAVIA_CONFIG_FILE),
{host: {"pin": pin, "host": host, "mac": mac}},
bravia_config_file_path = hass.config.path(BRAVIA_CONFIG_FILE)
bravia_config = await hass.async_add_executor_job(
load_json, bravia_config_file_path
)
add_entities([BraviaTVDevice(braviarc, name, pin, unique_id)])
def request_configuration(config, hass, add_entities):
"""Request configuration steps from the user."""
host = config[CONF_HOST]
name = config[CONF_NAME]
configurator = hass.components.configurator
# We got an error if this method is called while we are configuring
if host in _CONFIGURING:
configurator.notify_errors(
_CONFIGURING[host], "Failed to register, please try again."
if not bravia_config:
_LOGGER.error(
"Configuration import failed, there is no bravia.conf file in the configuration folder"
)
return
def bravia_configuration_callback(data):
"""Handle the entry of user PIN."""
while bravia_config:
# Import a configured TV
host_ip, host_config = bravia_config.popitem()
if host_ip == host:
pin = host_config[CONF_PIN]
pin = data.get("pin")
_braviarc = BraviaRC(host)
_braviarc.connect(pin, CLIENTID_PREFIX, NICKNAME)
if _braviarc.is_connected():
setup_bravia(config, pin, hass, add_entities)
else:
request_configuration(config, hass, add_entities)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={CONF_HOST: host, CONF_PIN: pin},
)
)
return
_CONFIGURING[host] = configurator.request_config(
name,
bravia_configuration_callback,
description=(
"Enter the Pin shown on your Sony Bravia TV."
"If no Pin is shown, enter 0000 to let TV show you a Pin."
),
description_image="/static/images/smart-tv.png",
submit_caption="Confirm",
fields=[{"id": "pin", "name": "Enter the pin", "type": ""}],
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Add BraviaTV entities from a config_entry."""
ignored_sources = []
pin = config_entry.data[CONF_PIN]
unique_id = config_entry.unique_id
device_info = {
"identifiers": {(DOMAIN, unique_id)},
"name": DEFAULT_NAME,
"manufacturer": ATTR_MANUFACTURER,
"model": config_entry.title,
}
braviarc = hass.data[DOMAIN][config_entry.entry_id]
ignored_sources = config_entry.options.get(CONF_IGNORED_SOURCES, [])
async_add_entities(
[
BraviaTVDevice(
braviarc, DEFAULT_NAME, pin, unique_id, device_info, ignored_sources
)
]
)
class BraviaTVDevice(MediaPlayerDevice):
"""Representation of a Sony Bravia TV."""
"""Representation of a Bravia TV."""
def __init__(self, client, name, pin, unique_id):
"""Initialize the Sony Bravia device."""
def __init__(self, client, name, pin, unique_id, device_info, ignored_sources):
"""Initialize the Bravia TV device."""
self._pin = pin
self._braviarc = client
@ -191,11 +142,8 @@ class BraviaTVDevice(MediaPlayerDevice):
self._max_volume = None
self._volume = None
self._unique_id = unique_id
if self._braviarc.is_connected():
self.update()
else:
self._state = STATE_OFF
self._device_info = device_info
self._ignored_sources = ignored_sources
def update(self):
"""Update TV info."""
@ -265,18 +213,29 @@ class BraviaTVDevice(MediaPlayerDevice):
self._content_mapping = self._braviarc.load_source_list()
self._source_list = []
for key in self._content_mapping:
self._source_list.append(key)
if key not in self._ignored_sources:
self._source_list.append(key)
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def device_class(self):
"""Set the device class to TV."""
return DEVICE_CLASS_TV
@property
def unique_id(self):
"""Return a unique_id for this entity."""
return self._unique_id
@property
def device_info(self):
"""Return the device info."""
return self._device_info
@property
def state(self):
"""Return the state of the device."""

View file

@ -0,0 +1,39 @@
{
"config": {
"title": "Sony Bravia TV",
"step": {
"user": {
"title": "Sony Bravia TV",
"description": "Set up Sony Bravia TV integration. If you have problems with configuration go to: https://www.home-assistant.io/integrations/braviatv \n\nEnsure that your TV is turned on.",
"data": {
"host": "TV hostname or IP address"
}
},
"authorize": {
"title": "Authorize Sony Bravia TV",
"description": "Enter the PIN code shown on the Sony Bravia TV. \n\nIf the PIN code is not shown, you have to unregister Home Assistant on your TV, go to: Settings -> Network -> Remote device settings -> Unregister remote device.",
"data": {
"pin": "PIN code"
}
}
},
"error": {
"invalid_host": "Invalid hostname or IP address.",
"cannot_connect": "Failed to connect, invalid host or PIN code.",
"unsupported_model": "Your TV model is not supported."
},
"abort": {
"already_configured": "This TV is already configured."
}
},
"options": {
"step": {
"user": {
"title": "Options for Sony Bravia TV",
"data": {
"ignored_sources": "List of ignored sources"
}
}
}
}
}

View file

@ -15,6 +15,7 @@ FLOWS = [
"ambient_station",
"august",
"axis",
"braviatv",
"brother",
"cast",
"cert_expiry",

View file

@ -136,6 +136,9 @@ bellows-homeassistant==0.15.2
# homeassistant.components.bom
bomradarloop==0.1.4
# homeassistant.components.braviatv
bravia-tv==1.0.1
# homeassistant.components.broadlink
broadlink==0.13.0

View file

@ -0,0 +1 @@
"""Tests for Bravia TV."""

View file

@ -0,0 +1,255 @@
"""Define tests for the Bravia TV config flow."""
from asynctest import patch
from homeassistant import data_entry_flow
from homeassistant.components.braviatv.const import CONF_IGNORED_SOURCES, DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN
from tests.common import MockConfigEntry
BRAVIA_SYSTEM_INFO = {
"product": "TV",
"region": "XEU",
"language": "pol",
"model": "TV-Model",
"serial": "serial_number",
"macAddr": "AA:BB:CC:DD:EE:FF",
"name": "BRAVIA",
"generation": "5.2.0",
"area": "POL",
"cid": "very_unique_string",
}
BRAVIA_SOURCE_LIST = {
"HDMI 1": "extInput:hdmi?port=1",
"HDMI 2": "extInput:hdmi?port=2",
"HDMI 3/ARC": "extInput:hdmi?port=3",
"HDMI 4": "extInput:hdmi?port=4",
"AV/Component": "extInput:component?port=1",
}
IMPORT_CONFIG_HOSTNAME = {
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
}
IMPORT_CONFIG_IP = {
CONF_HOST: "10.10.10.12",
CONF_PIN: "1234",
}
async def test_show_form(hass):
"""Test that the form is served with no input."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == SOURCE_USER
async def test_import(hass):
"""Test that the import works."""
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch(
"bravia_tv.BraviaRC.get_system_info", return_value=BRAVIA_SYSTEM_INFO
), patch(
"homeassistant.components.braviatv.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=IMPORT_CONFIG_HOSTNAME,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["result"].unique_id == "very_unique_string"
assert result["title"] == "TV-Model"
assert result["data"] == {
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
}
async def test_import_cannot_connect(hass):
"""Test that errors are shown when cannot connect to the host during import."""
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=False
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=IMPORT_CONFIG_HOSTNAME,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_import_model_unsupported(hass):
"""Test that errors are shown when the TV is not supported during import."""
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch("bravia_tv.BraviaRC.get_system_info", side_effect=KeyError):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=IMPORT_CONFIG_IP,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "unsupported_model"
async def test_import_duplicate_error(hass):
"""Test that errors are shown when duplicates are added during import."""
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id="very_unique_string",
data={
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
},
title="TV-Model",
)
config_entry.add_to_hass(hass)
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch("bravia_tv.BraviaRC.get_system_info", return_value=BRAVIA_SYSTEM_INFO):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=IMPORT_CONFIG_HOSTNAME,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_user_invalid_host(hass):
"""Test that errors are shown when the host is invalid."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "invalid/host"}
)
assert result["errors"] == {CONF_HOST: "invalid_host"}
async def test_authorize_cannot_connect(hass):
"""Test that errors are shown when cannot connect to host at the authorize step."""
with patch("bravia_tv.BraviaRC.connect", return_value=True):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "1234"}
)
assert result["errors"] == {"base": "cannot_connect"}
async def test_authorize_model_unsupported(hass):
"""Test that errors are shown when the TV is not supported at the authorize step."""
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch("bravia_tv.BraviaRC.get_system_info", side_effect=KeyError):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "10.10.10.12"},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "1234"}
)
assert result["errors"] == {"base": "unsupported_model"}
async def test_duplicate_error(hass):
"""Test that errors are shown when duplicates are added."""
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id="very_unique_string",
data={
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
},
title="TV-Model",
)
config_entry.add_to_hass(hass)
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch("bravia_tv.BraviaRC.get_system_info", return_value=BRAVIA_SYSTEM_INFO):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "1234"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_create_entry(hass):
"""Test that the user step works."""
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch(
"bravia_tv.BraviaRC.get_system_info", return_value=BRAVIA_SYSTEM_INFO
), patch(
"homeassistant.components.braviatv.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "authorize"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "1234"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["result"].unique_id == "very_unique_string"
assert result["title"] == "TV-Model"
assert result["data"] == {
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
}
async def test_options_flow(hass):
"""Test config flow options."""
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id="very_unique_string",
data={
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
},
title="TV-Model",
)
config_entry.add_to_hass(hass)
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch("bravia_tv.BraviaRC.get_system_info", return_value=BRAVIA_SYSTEM_INFO):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
with patch("bravia_tv.BraviaRC.load_source_list", return_value=BRAVIA_SOURCE_LIST):
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={CONF_IGNORED_SOURCES: ["HDMI 1", "HDMI 2"]}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert config_entry.options == {CONF_IGNORED_SOURCES: ["HDMI 1", "HDMI 2"]}