jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add samsungtv dhcp and zeroconf discovery (#48022)

Browse Source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Simone Chemelli 2021-05-22 17:41:18 +02:00 committed by GitHub
parent aa9b99713c
commit b9a0fb93eb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 1174 additions and 415 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -873,6 +873,7 @@ omit =
homeassistant/components/russound_rnet/media_player.py
homeassistant/components/sabnzbd/*
homeassistant/components/saj/sensor.py
homeassistant/components/samsungtv/bridge.py
homeassistant/components/satel_integra/*
homeassistant/components/schluter/*
homeassistant/components/scrape/sensor.py

2
CODEOWNERS
View File

@ -412,7 +412,7 @@ homeassistant/components/rpi_power/* @shenxn @swetoast
homeassistant/components/ruckus_unleashed/* @gabe565
homeassistant/components/safe_mode/* @home-assistant/core
homeassistant/components/saj/* @fredericvl
homeassistant/components/samsungtv/* @escoand
homeassistant/components/samsungtv/* @escoand @chemelli74
homeassistant/components/scene/* @home-assistant/core
homeassistant/components/schluter/* @prairieapps
homeassistant/components/scrape/* @fabaff

110
homeassistant/components/samsungtv/__init__.py
View File

@ -5,20 +5,31 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.media_player.const import DOMAIN as MP_DOMAIN
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
from homeassistant.const import (
CONF_HOST,
CONF_METHOD,
CONF_NAME,
CONF_PORT,
CONF_TOKEN,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from .const import CONF_ON_ACTION, DEFAULT_NAME, DOMAIN
from .bridge import SamsungTVBridge
from .const import CONF_ON_ACTION, DEFAULT_NAME, DOMAIN, LOGGER
def ensure_unique_hosts(value):
"""Validate that all configs have a unique host."""
vol.Schema(vol.Unique("duplicate host entries found"))(
[socket.gethostbyname(entry[CONF_HOST]) for entry in value]
[entry[CONF_HOST] for entry in value]
)
return value
PLATFORMS = [MP_DOMAIN]
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
@ -43,30 +54,87 @@ 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": config_entries.SOURCE_IMPORT},
data=entry_config,
)
)
hass.data[DOMAIN] = {}
if DOMAIN not in config:
return True
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": config_entries.SOURCE_IMPORT},
data=entry_config,
)
)
return True
@callback
def _async_get_device_bridge(data):
"""Get device bridge."""
return SamsungTVBridge.get_bridge(
data[CONF_METHOD],
data[CONF_HOST],
data[CONF_PORT],
data.get(CONF_TOKEN),
)
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, MP_DOMAIN)
# Initialize bridge
data = entry.data.copy()
bridge = _async_get_device_bridge(data)
if bridge.port is None and bridge.default_port is not None:
# For backward compat, set default port for websocket tv
data[CONF_PORT] = bridge.default_port
hass.config_entries.async_update_entry(entry, data=data)
bridge = _async_get_device_bridge(data)
def stop_bridge(event):
"""Stop SamsungTV bridge connection."""
bridge.stop()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_bridge)
)
hass.data[DOMAIN][entry.entry_id] = bridge
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass, entry):
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN][entry.entry_id].stop()
return unload_ok
async def async_migrate_entry(hass, config_entry):
"""Migrate old entry."""
version = config_entry.version
LOGGER.debug("Migrating from version %s", version)
# 1 -> 2: Unique ID format changed, so delete and re-import:
if version == 1:
dev_reg = await hass.helpers.device_registry.async_get_registry()
dev_reg.async_clear_config_entry(config_entry)
en_reg = await hass.helpers.entity_registry.async_get_registry()
en_reg.async_clear_config_entry(config_entry)
version = config_entry.version = 2
hass.config_entries.async_update_entry(config_entry)
LOGGER.debug("Migration to version %s successful", version)
return True

61
homeassistant/components/samsungtv/bridge.py
View File

@ -1,10 +1,11 @@
"""samsungctl and samsungtvws bridge classes."""
from abc import ABC, abstractmethod
import contextlib
from samsungctl import Remote
from samsungctl.exceptions import AccessDenied, ConnectionClosed, UnhandledResponse
from samsungtvws import SamsungTVWS
from samsungtvws.exceptions import ConnectionFailure
from samsungtvws.exceptions import ConnectionFailure, HttpApiError
from websocket import WebSocketException
from homeassistant.const import (
@ -25,8 +26,11 @@ from .const import (
RESULT_CANNOT_CONNECT,
RESULT_NOT_SUPPORTED,
RESULT_SUCCESS,
TIMEOUT_REQUEST,
TIMEOUT_WEBSOCKET,
VALUE_CONF_ID,
VALUE_CONF_NAME,
WEBSOCKET_PORTS,
)
@ -58,9 +62,14 @@ class SamsungTVBridge(ABC):
def try_connect(self):
"""Try to connect to the TV."""
@abstractmethod
def device_info(self):
"""Try to gather infos of this TV."""
def is_on(self):
"""Tells if the TV is on."""
self.close_remote()
if self._remote:
self.close_remote()
try:
return self._get_remote() is not None
@ -104,7 +113,7 @@ class SamsungTVBridge(ABC):
"""Send the key."""
@abstractmethod
def _get_remote(self):
def _get_remote(self, avoid_open: bool = False):
"""Get Remote object."""
def close_remote(self):
@ -149,7 +158,7 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
CONF_METHOD: self.method,
CONF_PORT: None,
# We need this high timeout because waiting for auth popup is just an open socket
CONF_TIMEOUT: 31,
CONF_TIMEOUT: TIMEOUT_REQUEST,
}
try:
LOGGER.debug("Try config: %s", config)
@ -162,11 +171,15 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
except UnhandledResponse:
LOGGER.debug("Working but unsupported config: %s", config)
return RESULT_NOT_SUPPORTED
except OSError as err:
except (ConnectionClosed, OSError) as err:
LOGGER.debug("Failing config: %s, error: %s", config, err)
return RESULT_CANNOT_CONNECT
def _get_remote(self):
def device_info(self):
"""Try to gather infos of this device."""
return None
def _get_remote(self, avoid_open: bool = False):
"""Create or return a remote control instance."""
if self._remote is None:
# We need to create a new instance to reconnect.
@ -184,6 +197,11 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
"""Send the key using legacy protocol."""
self._get_remote().control(key)
def stop(self):
"""Stop Bridge."""
LOGGER.debug("Stopping SamsungRemote")
self.close_remote()
class SamsungTVWSBridge(SamsungTVBridge):
"""The Bridge for WebSocket TVs."""
@ -196,14 +214,14 @@ class SamsungTVWSBridge(SamsungTVBridge):
def try_connect(self):
"""Try to connect to the Websocket TV."""
for self.port in (8001, 8002):
for self.port in WEBSOCKET_PORTS:
config = {
CONF_NAME: VALUE_CONF_NAME,
CONF_HOST: self.host,
CONF_METHOD: self.method,
CONF_PORT: self.port,
# We need this high timeout because waiting for auth popup is just an open socket
CONF_TIMEOUT: 31,
CONF_TIMEOUT: TIMEOUT_REQUEST,
}
result = None
@ -234,31 +252,46 @@ class SamsungTVWSBridge(SamsungTVBridge):
return RESULT_CANNOT_CONNECT
def device_info(self):
"""Try to gather infos of this TV."""
remote = self._get_remote(avoid_open=True)
if not remote:
return None
with contextlib.suppress(HttpApiError):
return remote.rest_device_info()
def _send_key(self, key):
"""Send the key using websocket protocol."""
if key == "KEY_POWEROFF":
key = "KEY_POWER"
self._get_remote().send_key(key)
def _get_remote(self):
def _get_remote(self, avoid_open: bool = False):
"""Create or return a remote control instance."""
if self._remote is None:
# We need to create a new instance to reconnect.
try:
LOGGER.debug("Create SamsungTVWS")
LOGGER.debug(
"Create SamsungTVWS for %s (%s)", VALUE_CONF_NAME, self.host
)
self._remote = SamsungTVWS(
host=self.host,
port=self.port,
token=self.token,
timeout=8,
timeout=TIMEOUT_WEBSOCKET,
name=VALUE_CONF_NAME,
)
self._remote.open()
if not avoid_open:
self._remote.open()
# 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 ConnectionFailure:
self._notify_callback()
raise
except WebSocketException:
except (WebSocketException, OSError):
self._remote = None
return self._remote
def stop(self):
"""Stop Bridge."""
LOGGER.debug("Stopping SamsungTVWS")
self.close_remote()

280
homeassistant/components/samsungtv/config_flow.py
View File

@ -4,7 +4,8 @@ from urllib.parse import urlparse
import voluptuous as vol
from homeassistant import config_entries
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.dhcp import IP_ADDRESS, MAC_ADDRESS
from homeassistant.components.ssdp import (
ATTR_SSDP_LOCATION,
ATTR_UPNP_MANUFACTURER,
@ -13,59 +14,85 @@ from homeassistant.components.ssdp import (
)
from homeassistant.const import (
CONF_HOST,
CONF_ID,
CONF_IP_ADDRESS,
CONF_MAC,
CONF_METHOD,
CONF_NAME,
CONF_PORT,
CONF_TOKEN,
)
from homeassistant.core import callback
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.typing import DiscoveryInfoType
from .bridge import SamsungTVBridge
from .const import (
ATTR_PROPERTIES,
CONF_MANUFACTURER,
CONF_MODEL,
DEFAULT_MANUFACTURER,
DOMAIN,
LEGACY_PORT,
LOGGER,
METHOD_LEGACY,
METHOD_WEBSOCKET,
RESULT_AUTH_MISSING,
RESULT_CANNOT_CONNECT,
RESULT_NOT_SUPPORTED,
RESULT_SUCCESS,
RESULT_UNKNOWN_HOST,
WEBSOCKET_PORTS,
)
DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str, vol.Required(CONF_NAME): str})
SUPPORTED_METHODS = [METHOD_LEGACY, METHOD_WEBSOCKET]
def _get_ip(host):
if host is None:
return None
return socket.gethostbyname(host)
def _get_device_info(host):
"""Fetch device info by any websocket method."""
for port in WEBSOCKET_PORTS:
bridge = SamsungTVBridge.get_bridge(METHOD_WEBSOCKET, host, port)
if info := bridge.device_info():
return info
return None
async def async_get_device_info(hass, bridge, host):
"""Fetch device info from bridge or websocket."""
if bridge:
return await hass.async_add_executor_job(bridge.device_info)
return await hass.async_add_executor_job(_get_device_info, host)
def _strip_uuid(udn):
return udn[5:] if udn.startswith("uuid:") else udn
class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a Samsung TV config flow."""
VERSION = 1
VERSION = 2
def __init__(self):
"""Initialize flow."""
self._reauth_entry = None
self._host = None
self._ip = None
self._mac = None
self._udn = None
self._manufacturer = None
self._model = None
self._name = None
self._title = None
self._id = None
self._bridge = None
self._device_info = None
def _get_entry(self):
def _get_entry_from_bridge(self):
"""Get device entry."""
data = {
CONF_HOST: self._host,
CONF_ID: self._id,
CONF_IP_ADDRESS: self._ip,
CONF_MANUFACTURER: self._manufacturer,
CONF_MAC: self._mac,
CONF_MANUFACTURER: self._manufacturer or DEFAULT_MANUFACTURER,
CONF_METHOD: self._bridge.method,
CONF_MODEL: self._model,
CONF_NAME: self._name,
@ -78,98 +105,205 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data=data,
)
async def _async_set_device_unique_id(self, raise_on_progress=True):
"""Set device unique_id."""
await self._async_get_and_check_device_info()
await self._async_set_unique_id_from_udn(raise_on_progress)
async def _async_set_unique_id_from_udn(self, raise_on_progress=True):
"""Set the unique id from the udn."""
await self.async_set_unique_id(self._udn, raise_on_progress=raise_on_progress)
self._async_update_existing_host_entry(self._host)
updates = {CONF_HOST: self._host}
if self._mac:
updates[CONF_MAC] = self._mac
self._abort_if_unique_id_configured(updates=updates)
def _try_connect(self):
"""Try to connect and check auth."""
for method in SUPPORTED_METHODS:
self._bridge = SamsungTVBridge.get_bridge(method, self._host)
result = self._bridge.try_connect()
if result == RESULT_SUCCESS:
return
if result != RESULT_CANNOT_CONNECT:
return result
raise data_entry_flow.AbortFlow(result)
LOGGER.debug("No working config found")
return RESULT_CANNOT_CONNECT
raise data_entry_flow.AbortFlow(RESULT_CANNOT_CONNECT)
async def _async_get_and_check_device_info(self):
"""Try to get the device info."""
info = await async_get_device_info(self.hass, self._bridge, self._host)
if not info:
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED)
dev_info = info.get("device", {})
device_type = dev_info.get("type")
if device_type != "Samsung SmartTV":
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED)
self._model = dev_info.get("modelName")
name = dev_info.get("name")
self._name = name.replace("[TV] ", "") if name else device_type
self._title = f"{self._name} ({self._model})"
self._udn = _strip_uuid(dev_info.get("udn", info["id"]))
if dev_info.get("networkType") == "wireless" and dev_info.get("wifiMac"):
self._mac = format_mac(dev_info.get("wifiMac"))
self._device_info = info
async def async_step_import(self, user_input=None):
"""Handle configuration by yaml file."""
return await self.async_step_user(user_input)
# We need to import even if we cannot validate
# since the TV may be off at startup
await self._async_set_name_host_from_input(user_input)
self._async_abort_entries_match({CONF_HOST: self._host})
if user_input.get(CONF_PORT) in WEBSOCKET_PORTS:
user_input[CONF_METHOD] = METHOD_WEBSOCKET
else:
user_input[CONF_METHOD] = METHOD_LEGACY
user_input[CONF_PORT] = LEGACY_PORT
user_input[CONF_MANUFACTURER] = DEFAULT_MANUFACTURER
return self.async_create_entry(
title=self._title,
data=user_input,
)
async def _async_set_name_host_from_input(self, user_input):
try:
self._host = await self.hass.async_add_executor_job(
socket.gethostbyname, user_input[CONF_HOST]
)
except socket.gaierror as err:
raise data_entry_flow.AbortFlow(RESULT_UNKNOWN_HOST) from err
self._name = user_input[CONF_NAME]
self._title = self._name
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
if user_input is not None:
ip_address = await self.hass.async_add_executor_job(
_get_ip, user_input[CONF_HOST]
)
await self.async_set_unique_id(ip_address)
self._abort_if_unique_id_configured()
self._host = user_input.get(CONF_HOST)
self._ip = self.context[CONF_IP_ADDRESS] = ip_address
self._name = user_input.get(CONF_NAME)
self._title = self._name
result = await self.hass.async_add_executor_job(self._try_connect)
if result != RESULT_SUCCESS:
return self.async_abort(reason=result)
return self._get_entry()
await self._async_set_name_host_from_input(user_input)
await self.hass.async_add_executor_job(self._try_connect)
self._async_abort_entries_match({CONF_HOST: self._host})
if self._bridge.method != METHOD_LEGACY:
# Legacy bridge does not provide device info
await self._async_set_device_unique_id(raise_on_progress=False)
return self._get_entry_from_bridge()
return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA)
async def async_step_ssdp(self, discovery_info):
"""Handle a flow initialized by discovery."""
host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
ip_address = await self.hass.async_add_executor_job(_get_ip, host)
@callback
def _async_update_existing_host_entry(self, host):
for entry in self._async_current_entries(include_ignore=False):
if entry.data[CONF_HOST] != host:
continue
entry_kw_args = {}
if self.unique_id and entry.unique_id is None:
entry_kw_args["unique_id"] = self.unique_id
if self._mac and not entry.data.get(CONF_MAC):
data_copy = dict(entry.data)
data_copy[CONF_MAC] = self._mac
entry_kw_args["data"] = data_copy
if entry_kw_args:
self.hass.config_entries.async_update_entry(entry, **entry_kw_args)
return entry
return None
async def _async_start_discovery_for_host(self, host):
"""Start discovery for a host."""
if entry := self._async_update_existing_host_entry(host):
if entry.unique_id:
# Let the flow continue to fill the missing
# unique id as we may be able to obtain it
# in the next step
raise data_entry_flow.AbortFlow("already_configured")
self.context[CONF_HOST] = host
for progress in self._async_in_progress():
if progress.get("context", {}).get(CONF_HOST) == host:
raise data_entry_flow.AbortFlow("already_in_progress")
self._host = host
self._ip = self.context[CONF_IP_ADDRESS] = ip_address
self._manufacturer = discovery_info.get(ATTR_UPNP_MANUFACTURER)
self._model = discovery_info.get(ATTR_UPNP_MODEL_NAME)
self._name = f"Samsung {self._model}"
self._id = discovery_info.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:]
await self.async_set_unique_id(ip_address)
self._abort_if_unique_id_configured(
{
CONF_ID: self._id,
CONF_MANUFACTURER: self._manufacturer,
CONF_MODEL: self._model,
}
async def async_step_ssdp(self, discovery_info: DiscoveryInfoType):
"""Handle a flow initialized by ssdp discovery."""
self._udn = _strip_uuid(discovery_info[ATTR_UPNP_UDN])
await self._async_set_unique_id_from_udn()
await self._async_start_discovery_for_host(
urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
)
self._manufacturer = discovery_info[ATTR_UPNP_MANUFACTURER]
if not self._manufacturer or not self._manufacturer.lower().startswith(
"samsung"
):
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED)
self._name = self._title = self._model = discovery_info.get(
ATTR_UPNP_MODEL_NAME
)
self.context["title_placeholders"] = {"device": self._title}
return await self.async_step_confirm()
self.context["title_placeholders"] = {"model": self._model}
async def async_step_dhcp(self, discovery_info: DiscoveryInfoType):
"""Handle a flow initialized by dhcp discovery."""
self._mac = discovery_info[MAC_ADDRESS]
await self._async_start_discovery_for_host(discovery_info[IP_ADDRESS])
await self._async_set_device_unique_id()
self.context["title_placeholders"] = {"device": self._title}
return await self.async_step_confirm()
async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType):
"""Handle a flow initialized by zeroconf discovery."""
self._mac = format_mac(discovery_info[ATTR_PROPERTIES]["deviceid"])
await self._async_start_discovery_for_host(discovery_info[CONF_HOST])
await self._async_set_device_unique_id()
self.context["title_placeholders"] = {"device": self._title}
return await self.async_step_confirm()
async def async_step_confirm(self, user_input=None):
"""Handle user-confirmation of discovered node."""
if user_input is not None:
result = await self.hass.async_add_executor_job(self._try_connect)
if result != RESULT_SUCCESS:
return self.async_abort(reason=result)
return self._get_entry()
await self.hass.async_add_executor_job(self._try_connect)
return self._get_entry_from_bridge()
self._set_confirm_only()
return self.async_show_form(
step_id="confirm", description_placeholders={"model": self._model}
step_id="confirm", description_placeholders={"device": self._title}
)
async def async_step_reauth(self, user_input=None):
async def async_step_reauth(self, data):
"""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._title = self._model or self._name
self._reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
data = self._reauth_entry.data
if data.get(CONF_MODEL) and data.get(CONF_NAME):
self._title = f"{data[CONF_NAME]} ({data[CONF_MODEL]})"
else:
self._title = data.get(CONF_NAME) or data[CONF_HOST]
return await self.async_step_reauth_confirm()
await self.async_set_unique_id(self._ip)
self.context["title_placeholders"] = {"model": self._title}
async def async_step_reauth_confirm(self, user_input=None):
"""Confirm reauth."""
errors = {}
if user_input is not None:
bridge = SamsungTVBridge.get_bridge(
self._reauth_entry.data[CONF_METHOD], self._reauth_entry.data[CONF_HOST]
)
result = bridge.try_connect()
if result == RESULT_SUCCESS:
new_data = dict(self._reauth_entry.data)
new_data[CONF_TOKEN] = bridge.token
self.hass.config_entries.async_update_entry(
self._reauth_entry, data=new_data
)
return self.async_abort(reason="reauth_successful")
if result not in (RESULT_AUTH_MISSING, RESULT_CANNOT_CONNECT):
return self.async_abort(reason=result)
return await self.async_step_confirm()
# On websocket we will get RESULT_CANNOT_CONNECT when auth is missing
errors = {"base": RESULT_AUTH_MISSING}
self.context["title_placeholders"] = {"device": self._title}
return self.async_show_form(
step_id="reauth_confirm",
errors=errors,
description_placeholders={"device": self._title},
)

10
homeassistant/components/samsungtv/const.py
View File

@ -4,7 +4,10 @@ import logging
LOGGER = logging.getLogger(__package__)
DOMAIN = "samsungtv"
ATTR_PROPERTIES = "properties"
DEFAULT_NAME = "Samsung TV"
DEFAULT_MANUFACTURER = "Samsung"
VALUE_CONF_NAME = "HomeAssistant"
VALUE_CONF_ID = "ha.component.samsung"
@ -18,6 +21,13 @@ RESULT_AUTH_MISSING = "auth_missing"
RESULT_SUCCESS = "success"
RESULT_CANNOT_CONNECT = "cannot_connect"
RESULT_NOT_SUPPORTED = "not_supported"
RESULT_UNKNOWN_HOST = "unknown"
METHOD_LEGACY = "legacy"
METHOD_WEBSOCKET = "websocket"
TIMEOUT_REQUEST = 31
TIMEOUT_WEBSOCKET = 5
LEGACY_PORT = 55000
WEBSOCKET_PORTS = (8002, 8001)

18
homeassistant/components/samsungtv/manifest.json
View File

@ -2,13 +2,27 @@
"domain": "samsungtv",
"name": "Samsung Smart TV",
"documentation": "https://www.home-assistant.io/integrations/samsungtv",
"requirements": ["samsungctl[websocket]==0.7.1", "samsungtvws==1.6.0"],
"requirements": [
"samsungctl[websocket]==0.7.1",
"samsungtvws==1.6.0"
],
"ssdp": [
{
"st": "urn:samsung.com:device:RemoteControlReceiver:1"
}
],
"codeowners": ["@escoand"],
"zeroconf": [
{"type":"_airplay._tcp.local.","manufacturer":"samsung*"}
],
"dhcp": [
{
"hostname": "tizen*"
}
],
"codeowners": [
"@escoand",
"@chemelli74"
],
"config_flow": true,
"iot_class": "local_polling"
}

76
homeassistant/components/samsungtv/media_player.py
View File

@ -19,22 +19,12 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_STEP,
)
from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.const import (
CONF_HOST,
CONF_ID,
CONF_IP_ADDRESS,
CONF_METHOD,
CONF_NAME,
CONF_PORT,
CONF_TOKEN,
STATE_OFF,
STATE_ON,
)
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, STATE_OFF, STATE_ON
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.script import Script
from homeassistant.util import dt as dt_util
from .bridge import SamsungTVBridge
from .const import (
CONF_MANUFACTURER,
CONF_MODEL,
@ -60,41 +50,19 @@ SUPPORT_SAMSUNGTV = (
)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up the Samsung TV from a config entry."""
ip_address = config_entry.data[CONF_IP_ADDRESS]
bridge = hass.data[DOMAIN][entry.entry_id]
host = entry.data[CONF_HOST]
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]
data = hass.data[DOMAIN]
if turn_on_action := data.get(host, {}).get(CONF_ON_ACTION):
on_script = Script(
hass, turn_on_action, config_entry.data.get(CONF_NAME, DEFAULT_NAME), DOMAIN
hass, turn_on_action, entry.data.get(CONF_NAME, DEFAULT_NAME), DOMAIN
)
# Initialize bridge
data = config_entry.data.copy()
bridge = SamsungTVBridge.get_bridge(
data[CONF_METHOD],
data[CONF_HOST],
data[CONF_PORT],
data.get(CONF_TOKEN),
)
if bridge.port is None and bridge.default_port is not None:
# For backward compat, set default port for websocket tv
data[CONF_PORT] = bridge.default_port
hass.config_entries.async_update_entry(config_entry, data=data)
bridge = SamsungTVBridge.get_bridge(
data[CONF_METHOD],
data[CONF_HOST],
data[CONF_PORT],
data.get(CONF_TOKEN),
)
async_add_entities([SamsungTVDevice(bridge, config_entry, on_script)])
async_add_entities([SamsungTVDevice(bridge, entry, on_script)], True)
class SamsungTVDevice(MediaPlayerEntity):
@ -103,11 +71,12 @@ class SamsungTVDevice(MediaPlayerEntity):
def __init__(self, bridge, config_entry, on_script):
"""Initialize the Samsung device."""
self._config_entry = config_entry
self._mac = config_entry.data.get(CONF_MAC)
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._uuid = config_entry.data.get(CONF_ID)
self._uuid = config_entry.unique_id
# Assume that the TV is not muted
self._muted = False
# Assume that the TV is in Play mode
@ -117,21 +86,28 @@ class SamsungTVDevice(MediaPlayerEntity):
# sending the next command to avoid turning the TV back ON).
self._end_of_power_off = None
self._bridge = bridge
self._auth_failed = False
self._bridge.register_reauth_callback(self.access_denied)
def access_denied(self):
"""Access denied callback."""
LOGGER.debug("Access denied in getting remote object")
self._auth_failed = True
self.hass.add_job(
self.hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH},
context={
"source": SOURCE_REAUTH,
"entry_id": self._config_entry.entry_id,
},
data=self._config_entry.data,
)
)
def update(self):
"""Update state of device."""
if self._auth_failed:
return
if self._power_off_in_progress():
self._state = STATE_OFF
else:
@ -165,15 +141,25 @@ class SamsungTVDevice(MediaPlayerEntity):
"""Return the state of the device."""
return self._state
@property
def available(self):
"""Return the availability of the device."""
if self._auth_failed:
return False
return self._state == STATE_ON or self._on_script
@property
def device_info(self):
"""Return device specific attributes."""
return {
info = {
"name": self.name,
"identifiers": {(DOMAIN, self.unique_id)},
"manufacturer": self._manufacturer,
"model": self._model,
}
if self._mac:
info["connections"] = {(CONNECTION_NETWORK_MAC, self._mac)}
return info
@property
def is_volume_muted(self):

22
homeassistant/components/samsungtv/strings.json
View File

@ -1,6 +1,6 @@
{
"config": {
"flow_title": "{model}",
"flow_title": "{device}",
"step": {
"user": {
"description": "Enter your Samsung TV information. If you never connected Home Assistant before you should see a popup on your TV asking for authorization.",
@ -10,16 +10,24 @@
}
},
"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 authorization. Manual configurations for this TV will be overwritten."
}
"description": "Do you want to set up {device}? If you never connected Home Assistant before you should see a popup on your TV asking for authorization."
},
"reauth_confirm": {
"description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds."
}
},
"error": {
"auth_missing": "[%key:component::samsungtv::config::abort::auth_missing%]"
},
"abort": {
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Please check your TV's settings to authorize Home Assistant.",
"auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Check your TV's External Device Manager settings to authorize Home Assistant.",
"id_missing": "This Samsung device doesn't have a SerialNumber.",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"not_supported": "This Samsung TV device is currently not supported."
"not_supported": "This Samsung device is currently not supported.",
"unknown": "[%key:common::config_flow::error::unknown%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
}
}
}

18
homeassistant/components/samsungtv/translations/en.json
View File

@ -3,15 +3,23 @@
"abort": {
"already_configured": "Device is already configured",
"already_in_progress": "Configuration flow is already in progress",
"auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Please check your TV's settings to authorize Home Assistant.",
"auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Check your TV's External Device Manager settings to authorize Home Assistant.",
"cannot_connect": "Failed to connect",
"not_supported": "This Samsung TV device is currently not supported."
"id_missing": "This Samsung device doesn't have a SerialNumber.",
"not_supported": "This Samsung device is currently not supported.",
"reauth_successful": "Re-authentication was successful",
"unknown": "Unexpected error"
},
"flow_title": "{model}",
"error": {
"auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Check your TV's External Device Manager settings to authorize Home Assistant."
},
"flow_title": "{device}",
"step": {
"confirm": {
"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.",
"title": "Samsung TV"
"description": "Do you want to set up {device}? If you never connected Home Assistant before you should see a popup on your TV asking for authorization."
},
"reauth_confirm": {
"description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds."
},
"user": {
"data": {

4
homeassistant/generated/dhcp.py
View File

@ -171,6 +171,10 @@ DHCP = [
"hostname": "roomba-*",
"macaddress": "80A589*"
},
{
"domain": "samsungtv",
"hostname": "tizen*"
},
{
"domain": "screenlogic",
"hostname": "pentair: *",

6
homeassistant/generated/zeroconf.py
View File

@ -11,6 +11,12 @@ ZEROCONF = {
"domain": "volumio"
}
],
"_airplay._tcp.local.": [
{
"domain": "samsungtv",
"manufacturer": "samsung*"
}
],
"_api._udp.local.": [
{
"domain": "guardian"

14
tests/components/samsungtv/__init__.py
View File

@ -1 +1,15 @@
"""Tests for the samsungtv component."""
from homeassistant.components.samsungtv.const import DOMAIN as SAMSUNGTV_DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
async def setup_samsungtv(hass: HomeAssistant, config: dict):
"""Set up mock Samsung TV."""
await async_setup_component(hass, "persistent_notification", {})
entry = MockConfigEntry(domain=SAMSUNGTV_DOMAIN, data=config)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()

112
tests/components/samsungtv/conftest.py Normal file
View File

@ -0,0 +1,112 @@
"""Fixtures for Samsung TV."""
from unittest.mock import Mock, patch
import pytest
import homeassistant.util.dt as dt_util
RESULT_ALREADY_CONFIGURED = "already_configured"
RESULT_ALREADY_IN_PROGRESS = "already_in_progress"
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote"
) as remote_class, patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
return_value="fake_host",
):
remote = Mock()
remote.__enter__ = Mock()
remote.__exit__ = Mock()
remote_class.return_value = remote
yield remote
@pytest.fixture(name="remotews")
def remotews_fixture():
"""Patch the samsungtvws SamsungTVWS."""
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remotews_class, patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
return_value="fake_host",
):
remotews = Mock()
remotews.__enter__ = Mock()
remotews.__exit__ = Mock()
remotews.rest_device_info.return_value = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
"wifiMac": "aa:bb:cc:dd:ee:ff",
"name": "[TV] Living Room",
"type": "Samsung SmartTV",
"networkType": "wireless",
},
}
remotews_class.return_value = remotews
remotews_class().__enter__().token = "FAKE_TOKEN"
yield remotews
@pytest.fixture(name="remotews_no_device_info")
def remotews_no_device_info_fixture():
"""Patch the samsungtvws SamsungTVWS."""
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remotews_class, patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
return_value="fake_host",
):
remotews = Mock()
remotews.__enter__ = Mock()
remotews.__exit__ = Mock()
remotews.rest_device_info.return_value = None
remotews_class.return_value = remotews
remotews_class().__enter__().token = "FAKE_TOKEN"
yield remotews
@pytest.fixture(name="remotews_soundbar")
def remotews_soundbar_fixture():
"""Patch the samsungtvws SamsungTVWS."""
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remotews_class, patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
return_value="fake_host",
):
remotews = Mock()
remotews.__enter__ = Mock()
remotews.__exit__ = Mock()
remotews.rest_device_info.return_value = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
"wifiMac": "aa:bb:cc:dd:ee:ff",
"mac": "aa:bb:cc:dd:ee:ff",
"name": "[TV] Living Room",
"type": "Samsung SoundBar",
},
}
remotews_class.return_value = remotews
remotews_class().__enter__().token = "FAKE_TOKEN"
yield remotews
@pytest.fixture(name="delay")
def delay_fixture():
"""Patch the delay script function."""
with patch(
"homeassistant.components.samsungtv.media_player.Script.async_run"
) as delay:
yield delay
@pytest.fixture
def mock_now():
"""Fixture for dtutil.now."""
return dt_util.utcnow()

709
tests/components/samsungtv/test_config_flow.py
View File

File diff suppressed because it is too large Load Diff

43
tests/components/samsungtv/test_init.py
View File

@ -1,21 +1,22 @@
"""Tests for the Samsung TV Integration."""
from unittest.mock import Mock, call, patch
import pytest
from homeassistant.components.media_player.const import DOMAIN, SUPPORT_TURN_ON
from homeassistant.components.samsungtv.const import (
CONF_ON_ACTION,
DOMAIN as SAMSUNGTV_DOMAIN,
METHOD_WEBSOCKET,
)
from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_SUPPORTED_FEATURES,
CONF_HOST,
CONF_METHOD,
CONF_NAME,
SERVICE_VOLUME_UP,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
ENTITY_ID = f"{DOMAIN}.fake_name"
@ -25,6 +26,7 @@ MOCK_CONFIG = {
CONF_HOST: "fake_host",
CONF_NAME: "fake_name",
CONF_ON_ACTION: [{"delay": "00:00:01"}],
CONF_METHOD: METHOD_WEBSOCKET,
}
]
}
@ -32,37 +34,22 @@ REMOTE_CALL = {
"name": "HomeAssistant",
"description": "HomeAssistant",
"id": "ha.component.samsung",
"method": "legacy",
"host": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_HOST],
"method": "legacy",
"port": None,
"timeout": 1,
}
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote"
) as remote_class, patch(
"homeassistant.components.samsungtv.config_flow.socket"
) as socket1, patch(
"homeassistant.components.samsungtv.socket"
) as socket2:
remote = Mock()
remote.__enter__ = Mock()
remote.__exit__ = Mock()
remote_class.return_value = remote
socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS"
socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS"
yield remote
async def test_setup(hass, remote):
async def test_setup(hass: HomeAssistant, remote: Mock):
"""Test Samsung TV integration is setup."""
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote:
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote, patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
return_value="fake_host",
):
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote:
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
# test name and turn_on
@ -80,7 +67,7 @@ async def test_setup(hass, remote):
assert remote.call_args == call(REMOTE_CALL)
async def test_setup_duplicate_config(hass, remote, caplog):
async def test_setup_duplicate_config(hass: HomeAssistant, remote: Mock, caplog):
"""Test duplicate setup of platform."""
DUPLICATE = {
SAMSUNGTV_DOMAIN: [
@ -95,7 +82,7 @@ async def test_setup_duplicate_config(hass, remote, caplog):
assert "duplicate host entries found" in caplog.text
async def test_setup_duplicate_entries(hass, remote, caplog):
async def test_setup_duplicate_entries(hass: HomeAssistant, remote: Mock, caplog):
"""Test duplicate setup of platform."""
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()

103
tests/components/samsungtv/test_media_player.py
View File

@ -25,6 +25,7 @@ from homeassistant.components.media_player.const import (
from homeassistant.components.samsungtv.const import (
CONF_ON_ACTION,
DOMAIN as SAMSUNGTV_DOMAIN,
TIMEOUT_WEBSOCKET,
)
from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV
from homeassistant.const import (
@ -37,10 +38,12 @@ from homeassistant.const import (
CONF_METHOD,
CONF_NAME,
CONF_PORT,
CONF_TIMEOUT,
CONF_TOKEN,
SERVICE_MEDIA_NEXT_TRACK,
SERVICE_MEDIA_PAUSE,
SERVICE_MEDIA_PLAY,
SERVICE_MEDIA_PLAY_PAUSE,
SERVICE_MEDIA_PREVIOUS_TRACK,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
@ -49,6 +52,7 @@ from homeassistant.const import (
SERVICE_VOLUME_UP,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
@ -59,7 +63,7 @@ ENTITY_ID = f"{DOMAIN}.fake"
MOCK_CONFIG = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake",
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_PORT: 55000,
CONF_ON_ACTION: [{"delay": "00:00:01"}],
@ -69,7 +73,7 @@ MOCK_CONFIG = {
MOCK_CONFIGWS = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake",
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_PORT: 8001,
CONF_TOKEN: "123456789",
@ -78,27 +82,20 @@ MOCK_CONFIGWS = {
]
}
MOCK_CALLS_WS = {
"host": "fake",
"port": 8001,
"token": None,
"timeout": 31,
"name": "HomeAssistant",
CONF_HOST: "fake_host",
CONF_PORT: 8001,
CONF_TOKEN: "123456789",
CONF_TIMEOUT: TIMEOUT_WEBSOCKET,
CONF_NAME: "HomeAssistant",
}
MOCK_ENTRY_WS = {
CONF_IP_ADDRESS: "test",
CONF_HOST: "fake",
CONF_HOST: "fake_host",
CONF_METHOD: "websocket",
CONF_NAME: "fake",
CONF_PORT: 8001,
CONF_TOKEN: "abcde",
}
MOCK_CALLS_ENTRY_WS = {
"host": "fake",
"name": "HomeAssistant",
"port": 8001,
"timeout": 8,
"token": "abcde",
CONF_TOKEN: "123456789",
}
ENTITY_ID_NOTURNON = f"{DOMAIN}.fake_noturnon"
@ -109,45 +106,6 @@ MOCK_CONFIG_NOTURNON = {
}
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote"
) as remote_class, patch(
"homeassistant.components.samsungtv.config_flow.socket"
) as socket1, patch(
"homeassistant.components.samsungtv.socket"
) as socket2:
remote = Mock()
remote.__enter__ = Mock()
remote.__exit__ = Mock()
remote_class.return_value = remote
socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS"
socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS"
yield remote
@pytest.fixture(name="remotews")
def remotews_fixture():
"""Patch the samsungtvws SamsungTVWS."""
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remote_class, patch(
"homeassistant.components.samsungtv.config_flow.socket"
) as socket1, patch(
"homeassistant.components.samsungtv.socket"
) as socket2:
remote = Mock()
remote.__enter__ = Mock()
remote.__exit__ = Mock()
remote_class.return_value = remote
remote_class().__enter__().token = "FAKE_TOKEN"
socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS"
socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS"
yield remote
@pytest.fixture(name="delay")
def delay_fixture():
"""Patch the delay script function."""
@ -226,7 +184,7 @@ async def test_setup_websocket_2(hass, mock_now):
state = hass.states.get(entity_id)
assert state
assert remote.call_count == 1
assert remote.call_args_list == [call(**MOCK_CALLS_ENTRY_WS)]
assert remote.call_args_list == [call(**MOCK_CALLS_WS)]
async def test_update_on(hass, remote, mock_now):
@ -272,12 +230,18 @@ async def test_update_access_denied(hass, remote, mock_now):
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
next_update = mock_now + timedelta(minutes=10)
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"
]
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_UNAVAILABLE
async def test_update_connection_failure(hass, remotews, mock_now):
@ -296,12 +260,18 @@ async def test_update_connection_failure(hass, remotews, mock_now):
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
next_update = mock_now + timedelta(minutes=10)
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"
]
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_UNAVAILABLE
async def test_update_unhandled_response(hass, remote, mock_now):
@ -438,7 +408,8 @@ async def test_state_without_turnon(hass, remote):
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True
)
state = hass.states.get(ENTITY_ID_NOTURNON)
assert state.state == STATE_OFF
# Should be STATE_UNAVAILABLE since there is no way to turn it back on
assert state.state == STATE_UNAVAILABLE
async def test_supported_features_with_turnon(hass, remote):
@ -555,6 +526,15 @@ async def test_media_play(hass, remote):
assert remote.close.call_count == 1
assert remote.close.call_args_list == [call()]
assert await hass.services.async_call(
DOMAIN, SERVICE_MEDIA_PLAY_PAUSE, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# key and update called
assert remote.control.call_count == 2
assert remote.control.call_args_list == [call("KEY_PLAY"), call("KEY_PAUSE")]
assert remote.close.call_count == 2
assert remote.close.call_args_list == [call(), call()]
async def test_media_pause(hass, remote):
"""Test for media_pause."""
@ -568,6 +548,15 @@ async def test_media_pause(hass, remote):
assert remote.close.call_count == 1
assert remote.close.call_args_list == [call()]
assert await hass.services.async_call(
DOMAIN, SERVICE_MEDIA_PLAY_PAUSE, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# key and update called
assert remote.control.call_count == 2
assert remote.control.call_args_list == [call("KEY_PAUSE"), call("KEY_PLAY")]
assert remote.close.call_count == 2
assert remote.close.call_args_list == [call(), call()]
async def test_media_next_track(hass, remote):
"""Test for media_next_track."""