mirror of
https://github.com/home-assistant/core.git
synced 2025-07-24 05:37:44 +00:00
Samsung TV refinements (#31248)
* use st not deviceType * show model in flow title * Update strings.json * add re-auth to entity * add re-auth to config_flow * handle auth popup better * use media player domain const * fix tests * rename not_found to not_successful * authz not authn * Update media_player.py * Update config_flow.py * Update media_player.py * Update test_media_player.py * finalize re-auth * fix ssd tests * better naming * fix ip-address-mock serialization * fix turn_on_action serialization * add type of hass object * fix acces denied test * remove half-added typing * async get ip address * fix pylint
This commit is contained in:
parent
03642d9029
commit
4550968316
@ -3,6 +3,7 @@ import socket
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.media_player.const import DOMAIN as MP_DOMAIN
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
@ -41,7 +42,14 @@ CONFIG_SCHEMA = vol.Schema(
|
||||
async def async_setup(hass, config):
|
||||
"""Set up the Samsung TV integration."""
|
||||
if DOMAIN in config:
|
||||
hass.data[DOMAIN] = {}
|
||||
for entry_config in config[DOMAIN]:
|
||||
ip_address = await hass.async_add_executor_job(
|
||||
socket.gethostbyname, entry_config[CONF_HOST]
|
||||
)
|
||||
hass.data[DOMAIN][ip_address] = {
|
||||
CONF_ON_ACTION: entry_config.get(CONF_ON_ACTION)
|
||||
}
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "import"}, data=entry_config
|
||||
@ -54,7 +62,7 @@ async def async_setup(hass, config):
|
||||
async def async_setup_entry(hass, entry):
|
||||
"""Set up the Samsung TV platform."""
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(entry, "media_player")
|
||||
hass.config_entries.async_forward_entry_setup(entry, MP_DOMAIN)
|
||||
)
|
||||
|
||||
return True
|
||||
|
@ -9,7 +9,6 @@ import voluptuous as vol
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.ssdp import (
|
||||
ATTR_SSDP_LOCATION,
|
||||
ATTR_UPNP_FRIENDLY_NAME,
|
||||
ATTR_UPNP_MANUFACTURER,
|
||||
ATTR_UPNP_MODEL_NAME,
|
||||
ATTR_UPNP_UDN,
|
||||
@ -24,20 +23,13 @@ from homeassistant.const import (
|
||||
)
|
||||
|
||||
# pylint:disable=unused-import
|
||||
from .const import (
|
||||
CONF_MANUFACTURER,
|
||||
CONF_MODEL,
|
||||
CONF_ON_ACTION,
|
||||
DOMAIN,
|
||||
LOGGER,
|
||||
METHODS,
|
||||
)
|
||||
from .const import CONF_MANUFACTURER, CONF_MODEL, DOMAIN, LOGGER, METHODS
|
||||
|
||||
DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str, vol.Required(CONF_NAME): str})
|
||||
|
||||
RESULT_AUTH_MISSING = "auth_missing"
|
||||
RESULT_SUCCESS = "success"
|
||||
RESULT_NOT_FOUND = "not_found"
|
||||
RESULT_NOT_SUCCESSFUL = "not_successful"
|
||||
RESULT_NOT_SUPPORTED = "not_supported"
|
||||
|
||||
|
||||
@ -63,23 +55,21 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
self._method = None
|
||||
self._model = None
|
||||
self._name = None
|
||||
self._on_script = None
|
||||
self._port = None
|
||||
self._title = None
|
||||
self._uuid = None
|
||||
self._id = None
|
||||
|
||||
def _get_entry(self):
|
||||
return self.async_create_entry(
|
||||
title=self._title,
|
||||
data={
|
||||
CONF_HOST: self._host,
|
||||
CONF_ID: self._uuid,
|
||||
CONF_ID: self._id,
|
||||
CONF_IP_ADDRESS: self._ip,
|
||||
CONF_MANUFACTURER: self._manufacturer,
|
||||
CONF_METHOD: self._method,
|
||||
CONF_MODEL: self._model,
|
||||
CONF_NAME: self._name,
|
||||
CONF_ON_ACTION: self._on_script,
|
||||
CONF_PORT: self._port,
|
||||
},
|
||||
)
|
||||
@ -94,7 +84,8 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"host": self._host,
|
||||
"method": method,
|
||||
"port": self._port,
|
||||
"timeout": 1,
|
||||
# We need this high timeout because waiting for auth popup is just an open socket
|
||||
"timeout": 31,
|
||||
}
|
||||
try:
|
||||
LOGGER.debug("Try config: %s", config)
|
||||
@ -108,15 +99,14 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
except UnhandledResponse:
|
||||
LOGGER.debug("Working but unsupported config: %s", config)
|
||||
return RESULT_NOT_SUPPORTED
|
||||
except (OSError):
|
||||
LOGGER.debug("Failing config: %s", config)
|
||||
except OSError as err:
|
||||
LOGGER.debug("Failing config: %s, error: %s", config, err)
|
||||
|
||||
LOGGER.debug("No working config found")
|
||||
return RESULT_NOT_FOUND
|
||||
return RESULT_NOT_SUCCESSFUL
|
||||
|
||||
async def async_step_import(self, user_input=None):
|
||||
"""Handle configuration by yaml file."""
|
||||
self._on_script = user_input.get(CONF_ON_ACTION)
|
||||
self._port = user_input.get(CONF_PORT)
|
||||
|
||||
return await self.async_step_user(user_input)
|
||||
@ -133,7 +123,8 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
self._host = user_input.get(CONF_HOST)
|
||||
self._ip = self.context[CONF_IP_ADDRESS] = ip_address
|
||||
self._title = user_input.get(CONF_NAME)
|
||||
self._name = user_input.get(CONF_NAME)
|
||||
self._title = self._name
|
||||
|
||||
result = await self.hass.async_add_executor_job(self._try_connect)
|
||||
|
||||
@ -150,24 +141,27 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
self._host = host
|
||||
self._ip = self.context[CONF_IP_ADDRESS] = ip_address
|
||||
self._manufacturer = user_input[ATTR_UPNP_MANUFACTURER]
|
||||
self._model = user_input[ATTR_UPNP_MODEL_NAME]
|
||||
self._name = user_input[ATTR_UPNP_FRIENDLY_NAME]
|
||||
if self._name.startswith("[TV]"):
|
||||
self._name = self._name[4:]
|
||||
self._title = f"{self._name} ({self._model})"
|
||||
self._uuid = user_input[ATTR_UPNP_UDN]
|
||||
if self._uuid.startswith("uuid:"):
|
||||
self._uuid = self._uuid[5:]
|
||||
self._manufacturer = user_input.get(ATTR_UPNP_MANUFACTURER)
|
||||
self._model = user_input.get(ATTR_UPNP_MODEL_NAME)
|
||||
self._name = f"Samsung {self._model}"
|
||||
self._id = user_input.get(ATTR_UPNP_UDN)
|
||||
self._title = self._model
|
||||
|
||||
# probably access denied
|
||||
if self._id is None:
|
||||
return self.async_abort(reason=RESULT_AUTH_MISSING)
|
||||
if self._id.startswith("uuid:"):
|
||||
self._id = self._id[5:]
|
||||
|
||||
config_entry = await self.async_set_unique_id(ip_address)
|
||||
if config_entry:
|
||||
config_entry.data[CONF_ID] = self._uuid
|
||||
config_entry.data[CONF_ID] = self._id
|
||||
config_entry.data[CONF_MANUFACTURER] = self._manufacturer
|
||||
config_entry.data[CONF_MODEL] = self._model
|
||||
self.hass.config_entries.async_update_entry(config_entry)
|
||||
return self.async_abort(reason="already_configured")
|
||||
|
||||
self.context["title_placeholders"] = {"model": self._model}
|
||||
return await self.async_step_confirm()
|
||||
|
||||
async def async_step_confirm(self, user_input=None):
|
||||
@ -182,3 +176,19 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
return self.async_show_form(
|
||||
step_id="confirm", description_placeholders={"model": self._model}
|
||||
)
|
||||
|
||||
async def async_step_reauth(self, user_input=None):
|
||||
"""Handle configuration by re-auth."""
|
||||
self._host = user_input[CONF_HOST]
|
||||
self._id = user_input.get(CONF_ID)
|
||||
self._ip = user_input[CONF_IP_ADDRESS]
|
||||
self._manufacturer = user_input.get(CONF_MANUFACTURER)
|
||||
self._model = user_input.get(CONF_MODEL)
|
||||
self._name = user_input.get(CONF_NAME)
|
||||
self._port = user_input.get(CONF_PORT)
|
||||
self._title = self._model or self._name
|
||||
|
||||
await self.async_set_unique_id(self._ip)
|
||||
self.context["title_placeholders"] = {"model": self._title}
|
||||
|
||||
return await self.async_step_confirm()
|
||||
|
@ -4,7 +4,7 @@ import logging
|
||||
LOGGER = logging.getLogger(__package__)
|
||||
DOMAIN = "samsungtv"
|
||||
|
||||
DEFAULT_NAME = "Samsung TV Remote"
|
||||
DEFAULT_NAME = "Samsung TV"
|
||||
|
||||
CONF_MANUFACTURER = "manufacturer"
|
||||
CONF_MODEL = "model"
|
||||
|
@ -7,7 +7,7 @@
|
||||
],
|
||||
"ssdp": [
|
||||
{
|
||||
"deviceType": "urn:samsung.com:device:RemoteControlReceiver:1"
|
||||
"st": "urn:samsung.com:device:RemoteControlReceiver:1"
|
||||
}
|
||||
],
|
||||
"dependencies": [],
|
||||
|
@ -23,7 +23,9 @@ from homeassistant.components.media_player.const import (
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_ID,
|
||||
CONF_IP_ADDRESS,
|
||||
CONF_METHOD,
|
||||
CONF_NAME,
|
||||
CONF_PORT,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
@ -59,8 +61,16 @@ async def async_setup_platform(
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Samsung TV from a config entry."""
|
||||
turn_on_action = config_entry.data.get(CONF_ON_ACTION)
|
||||
on_script = Script(hass, turn_on_action) if turn_on_action else None
|
||||
ip_address = config_entry.data[CONF_IP_ADDRESS]
|
||||
on_script = None
|
||||
if (
|
||||
DOMAIN in hass.data
|
||||
and ip_address in hass.data[DOMAIN]
|
||||
and CONF_ON_ACTION in hass.data[DOMAIN][ip_address]
|
||||
and hass.data[DOMAIN][ip_address][CONF_ON_ACTION]
|
||||
):
|
||||
turn_on_action = hass.data[DOMAIN][ip_address][CONF_ON_ACTION]
|
||||
on_script = Script(hass, turn_on_action)
|
||||
async_add_entities([SamsungTVDevice(config_entry, on_script)])
|
||||
|
||||
|
||||
@ -70,12 +80,11 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
def __init__(self, config_entry, on_script):
|
||||
"""Initialize the Samsung device."""
|
||||
self._config_entry = config_entry
|
||||
self._name = config_entry.title
|
||||
self._uuid = config_entry.data.get(CONF_ID)
|
||||
self._manufacturer = config_entry.data.get(CONF_MANUFACTURER)
|
||||
self._model = config_entry.data.get(CONF_MODEL)
|
||||
self._name = config_entry.data.get(CONF_NAME)
|
||||
self._on_script = on_script
|
||||
self._update_listener = None
|
||||
self._uuid = config_entry.data.get(CONF_ID)
|
||||
# Assume that the TV is not muted
|
||||
self._muted = False
|
||||
# Assume that the TV is in Play mode
|
||||
@ -88,7 +97,7 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
# Generate a configuration for the Samsung library
|
||||
self._config = {
|
||||
"name": "HomeAssistant",
|
||||
"description": self._name,
|
||||
"description": "HomeAssistant",
|
||||
"id": "ha.component.samsung",
|
||||
"method": config_entry.data[CONF_METHOD],
|
||||
"port": config_entry.data.get(CONF_PORT),
|
||||
@ -124,7 +133,19 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
"""Create or return a remote control instance."""
|
||||
if self._remote is None:
|
||||
# We need to create a new instance to reconnect.
|
||||
self._remote = SamsungRemote(self._config.copy())
|
||||
try:
|
||||
self._remote = SamsungRemote(self._config.copy())
|
||||
# This is only happening when the auth was switched to DENY
|
||||
# A removed auth will lead to socket timeout because waiting for auth popup is just an open socket
|
||||
except samsung_exceptions.AccessDenied:
|
||||
self.hass.async_create_task(
|
||||
self.hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": "reauth"},
|
||||
data=self._config_entry.data,
|
||||
)
|
||||
)
|
||||
raise
|
||||
|
||||
return self._remote
|
||||
|
||||
|
@ -1,10 +1,11 @@
|
||||
{
|
||||
"config": {
|
||||
"flow_title": "Samsung TV: {model}",
|
||||
"title": "Samsung TV",
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Samsung TV",
|
||||
"description": "Enter your Samsung TV information. If you never connected Home Assistant before you should see a popup on your TV asking for authentication.",
|
||||
"description": "Enter your Samsung TV information. If you never connected Home Assistant before you should see a popup on your TV asking for authorization.",
|
||||
"data": {
|
||||
"host": "Host or IP address",
|
||||
"name": "Name"
|
||||
@ -12,15 +13,15 @@
|
||||
},
|
||||
"confirm": {
|
||||
"title": "Samsung TV",
|
||||
"description": "Do you want to set up Samsung TV {model}? If you never connected Home Assistant before you should see a popup on your TV asking for authentication. Manual configurations for this TV will be overwritten."
|
||||
"description": "Do you want to set up Samsung TV {model}? If you never connected Home Assistant before you should see a popup on your TV asking for authorization. Manual configurations for this TV will be overwritten."
|
||||
}
|
||||
},
|
||||
"abort": {
|
||||
"already_in_progress": "Samsung TV configuration is already in progress.",
|
||||
"already_configured": "This Samsung TV is already configured.",
|
||||
"auth_missing": "Home Assistant is not authenticated to connect to this Samsung TV.",
|
||||
"not_found": "No supported Samsung TV devices found on the network.",
|
||||
"not_supported": "This Samsung TV devices is currently not supported."
|
||||
"auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Please check your TV's settings to authorize Home Assistant.",
|
||||
"not_successful": "Unable to connect to this Samsung TV device.",
|
||||
"not_supported": "This Samsung TV device is currently not supported."
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -38,7 +38,7 @@ SSDP = {
|
||||
],
|
||||
"samsungtv": [
|
||||
{
|
||||
"deviceType": "urn:samsung.com:device:RemoteControlReceiver:1"
|
||||
"st": "urn:samsung.com:device:RemoteControlReceiver:1"
|
||||
}
|
||||
],
|
||||
"sonos": [
|
||||
|
@ -42,7 +42,7 @@ AUTODETECT_WEBSOCKET = {
|
||||
"method": "websocket",
|
||||
"port": None,
|
||||
"host": "fake_host",
|
||||
"timeout": 1,
|
||||
"timeout": 31,
|
||||
}
|
||||
AUTODETECT_LEGACY = {
|
||||
"name": "HomeAssistant",
|
||||
@ -51,7 +51,7 @@ AUTODETECT_LEGACY = {
|
||||
"method": "legacy",
|
||||
"port": None,
|
||||
"host": "fake_host",
|
||||
"timeout": 1,
|
||||
"timeout": 31,
|
||||
}
|
||||
|
||||
|
||||
@ -87,7 +87,7 @@ async def test_user(hass, remote):
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == "fake_name"
|
||||
assert result["data"][CONF_HOST] == "fake_host"
|
||||
assert result["data"][CONF_NAME] is None
|
||||
assert result["data"][CONF_NAME] == "fake_name"
|
||||
assert result["data"][CONF_MANUFACTURER] is None
|
||||
assert result["data"][CONF_MODEL] is None
|
||||
assert result["data"][CONF_ID] is None
|
||||
@ -123,19 +123,19 @@ async def test_user_not_supported(hass):
|
||||
assert result["reason"] == "not_supported"
|
||||
|
||||
|
||||
async def test_user_not_found(hass):
|
||||
"""Test starting a flow by user but no device found."""
|
||||
async def test_user_not_successful(hass):
|
||||
"""Test starting a flow by user but no connection found."""
|
||||
with patch(
|
||||
"homeassistant.components.samsungtv.config_flow.Remote",
|
||||
side_effect=OSError("Boom"),
|
||||
), patch("homeassistant.components.samsungtv.config_flow.socket"):
|
||||
|
||||
# device not found
|
||||
# device not connectable
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
|
||||
)
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "not_found"
|
||||
assert result["reason"] == "not_successful"
|
||||
|
||||
|
||||
async def test_user_already_configured(hass, remote):
|
||||
@ -170,9 +170,9 @@ async def test_ssdp(hass, remote):
|
||||
result["flow_id"], user_input="whatever"
|
||||
)
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == "fake_name (fake_model)"
|
||||
assert result["title"] == "fake_model"
|
||||
assert result["data"][CONF_HOST] == "fake_host"
|
||||
assert result["data"][CONF_NAME] == "fake_name"
|
||||
assert result["data"][CONF_NAME] == "Samsung fake_model"
|
||||
assert result["data"][CONF_MANUFACTURER] == "fake_manufacturer"
|
||||
assert result["data"][CONF_MODEL] == "fake_model"
|
||||
assert result["data"][CONF_ID] == "fake_uuid"
|
||||
@ -193,9 +193,9 @@ async def test_ssdp_noprefix(hass, remote):
|
||||
result["flow_id"], user_input="whatever"
|
||||
)
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == "fake2_name (fake2_model)"
|
||||
assert result["title"] == "fake2_model"
|
||||
assert result["data"][CONF_HOST] == "fake2_host"
|
||||
assert result["data"][CONF_NAME] == "fake2_name"
|
||||
assert result["data"][CONF_NAME] == "Samsung fake2_model"
|
||||
assert result["data"][CONF_MANUFACTURER] == "fake2_manufacturer"
|
||||
assert result["data"][CONF_MODEL] == "fake2_model"
|
||||
assert result["data"][CONF_ID] == "fake2_uuid"
|
||||
@ -245,7 +245,7 @@ async def test_ssdp_not_supported(hass):
|
||||
assert result["reason"] == "not_supported"
|
||||
|
||||
|
||||
async def test_ssdp_not_found(hass):
|
||||
async def test_ssdp_not_successful(hass):
|
||||
"""Test starting a flow from discovery but no device found."""
|
||||
with patch(
|
||||
"homeassistant.components.samsungtv.config_flow.Remote",
|
||||
@ -264,7 +264,7 @@ async def test_ssdp_not_found(hass):
|
||||
result["flow_id"], user_input="whatever"
|
||||
)
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "not_found"
|
||||
assert result["reason"] == "not_successful"
|
||||
|
||||
|
||||
async def test_ssdp_already_in_progress(hass, remote):
|
||||
@ -380,7 +380,7 @@ async def test_autodetect_none(hass, remote):
|
||||
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
|
||||
)
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "not_found"
|
||||
assert result["reason"] == "not_successful"
|
||||
assert remote.call_count == 2
|
||||
assert remote.call_args_list == [
|
||||
call(AUTODETECT_WEBSOCKET),
|
||||
|
@ -32,7 +32,7 @@ MOCK_CONFIG = {
|
||||
}
|
||||
REMOTE_CALL = {
|
||||
"name": "HomeAssistant",
|
||||
"description": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_NAME],
|
||||
"description": "HomeAssistant",
|
||||
"id": "ha.component.samsung",
|
||||
"method": "websocket",
|
||||
"port": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_PORT],
|
||||
@ -44,11 +44,13 @@ REMOTE_CALL = {
|
||||
@pytest.fixture(name="remote")
|
||||
def remote_fixture():
|
||||
"""Patch the samsungctl Remote."""
|
||||
with patch("homeassistant.components.samsungtv.socket"), patch(
|
||||
with patch("homeassistant.components.samsungtv.socket") as socket1, patch(
|
||||
"homeassistant.components.samsungtv.config_flow.socket"
|
||||
), patch("homeassistant.components.samsungtv.config_flow.Remote"), patch(
|
||||
) as socket2, patch("homeassistant.components.samsungtv.config_flow.Remote"), patch(
|
||||
"homeassistant.components.samsungtv.media_player.SamsungRemote"
|
||||
) as remote:
|
||||
socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS"
|
||||
socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS"
|
||||
yield remote
|
||||
|
||||
|
||||
|
@ -75,15 +75,17 @@ MOCK_CONFIG_NOTURNON = {
|
||||
@pytest.fixture(name="remote")
|
||||
def remote_fixture():
|
||||
"""Patch the samsungctl Remote."""
|
||||
with patch("homeassistant.components.samsungtv.config_flow.socket"), patch(
|
||||
"homeassistant.components.samsungtv.config_flow.Remote"
|
||||
), patch(
|
||||
with patch("homeassistant.components.samsungtv.config_flow.Remote"), patch(
|
||||
"homeassistant.components.samsungtv.config_flow.socket"
|
||||
) as socket1, patch(
|
||||
"homeassistant.components.samsungtv.media_player.SamsungRemote"
|
||||
) as remote_class, patch(
|
||||
"homeassistant.components.samsungtv.socket"
|
||||
):
|
||||
) as socket2:
|
||||
remote = mock.Mock()
|
||||
remote_class.return_value = remote
|
||||
socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS"
|
||||
socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS"
|
||||
yield remote
|
||||
|
||||
|
||||
@ -135,11 +137,12 @@ async def test_update_on(hass, remote, mock_now):
|
||||
|
||||
async def test_update_off(hass, remote, mock_now):
|
||||
"""Testing update tv off."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.samsungtv.media_player.SamsungRemote",
|
||||
side_effect=[OSError("Boom"), mock.DEFAULT],
|
||||
), patch("homeassistant.components.samsungtv.config_flow.socket"):
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
):
|
||||
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
|
||||
@ -150,13 +153,35 @@ async def test_update_off(hass, remote, mock_now):
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
|
||||
async def test_update_access_denied(hass, remote, mock_now):
|
||||
"""Testing update tv unhandled response exception."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.samsungtv.media_player.SamsungRemote",
|
||||
side_effect=exceptions.AccessDenied("Boom"),
|
||||
):
|
||||
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
|
||||
async_fire_time_changed(hass, next_update)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert [
|
||||
flow
|
||||
for flow in hass.config_entries.flow.async_progress()
|
||||
if flow["context"]["source"] == "reauth"
|
||||
]
|
||||
|
||||
|
||||
async def test_update_unhandled_response(hass, remote, mock_now):
|
||||
"""Testing update tv unhandled response exception."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.samsungtv.media_player.SamsungRemote",
|
||||
side_effect=[exceptions.UnhandledResponse("Boom"), mock.DEFAULT],
|
||||
), patch("homeassistant.components.samsungtv.config_flow.socket"):
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
):
|
||||
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
|
||||
|
Loading…
x
Reference in New Issue
Block a user