github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add Samsung TV config flow (#28306) 

* add config flow

* add tests

* add user step error handling

* remove unload function

* add missing test file

* handle authentication correctly

* remove old discovery mode

* better handling of remote class

* optimized abort messages

* add already configured test for user flow

* Import order

* use ip property instead context

* Black

* small syntax

* use snake_case

* Revert "use ip property instead context"

This reverts commit 91502407eb216f8a0b1b90e3e6fb165b81406f8f.

* disable wrong pylint errors

* disable wrong no-member

* Try to fix review comments

* Try to fix review comments

* Fix missing self

* Fix ip checks

* methods to functions

* simplify user check

* remove user errors

* use async_setup for config

* fix after rebase

* import config to user config flow

* patch all samsungctl

* fix after rebase

* fix notes

* remove unused variable

* ignore old setup function

* fix after merge

* pass configuration to import step

* isort

* fix recursion

* remove timeout config

* add turn on action (dry without testing)

* use upstream checks

* cleanup

* minor

* correctly await async method

* ignore unused import

* async call send_key

* Revert "async call send_key"

This reverts commit f37057819f.

* fix comments

* fix timeout test

* test turn on action

* Update media_player.py

* Update test_media_player.py

* Update test_media_player.py

* use async executor

* use newer ssdp data

* update manually configured with ssdp data

* dont setup component directly

* ensure list

* check updated device info

* Update config_flow.py

* Update __init__.py

* fix duplicate check

* simplified unique check

* move method detection to config_flow

* move unique test to init

* fix after real world test

* optimize config_validation

* update device_info on ssdp discovery

* cleaner update listener

* fix lint

* fix method signature

* add note for manual config to confirm message

* fix turn_on_action

* pass script

* patch delay

* remove device info update
This commit is contained in:
escoand 2020-01-10 03:19:10 +01:00 committed by Martin Hjelmare
parent 4fb36451c2
commit ef05aa2f39
14 changed files with 896 additions and 397 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
homeassistant/components/discovery/__init__.py
View File

@ -75,7 +75,6 @@ SERVICE_HANDLERS = {
"logitech_mediaserver": ("media_player", "squeezebox"),
"directv": ("media_player", "directv"),
"denonavr": ("media_player", "denonavr"),
"samsung_tv": ("media_player", "samsungtv"),
"frontier_silicon": ("media_player", "frontier_silicon"),
"openhome": ("media_player", "openhome"),
"harmony": ("remote", "harmony"),

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

@ -1 +1,60 @@
"""The Samsung TV integration."""
import socket
import voluptuous as vol
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
import homeassistant.helpers.config_validation as cv
from .const import CONF_ON_ACTION, DEFAULT_NAME, DOMAIN
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]
)
return value
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
cv.ensure_list,
[
vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PORT): cv.port,
vol.Optional(CONF_ON_ACTION): cv.SCRIPT_SCHEMA,
}
)
],
ensure_unique_hosts,
)
},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass, config):
"""Set up the Samsung TV integration."""
if DOMAIN in config:
for entry_config in config[DOMAIN]:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": "import"}, data=entry_config
)
)
return True
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")
)
return True

184
homeassistant/components/samsungtv/config_flow.py Normal file
View File

@ -0,0 +1,184 @@
"""Config flow for Samsung TV."""
import socket
from urllib.parse import urlparse
from samsungctl import Remote
from samsungctl.exceptions import AccessDenied, UnhandledResponse
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,
)
from homeassistant.const import (
CONF_HOST,
CONF_ID,
CONF_IP_ADDRESS,
CONF_METHOD,
CONF_NAME,
CONF_PORT,
)
# pylint:disable=unused-import
from .const import (
CONF_MANUFACTURER,
CONF_MODEL,
CONF_ON_ACTION,
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_SUPPORTED = "not_supported"
def _get_ip(host):
if host is None:
return None
return socket.gethostbyname(host)
class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a Samsung TV config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
def __init__(self):
"""Initialize flow."""
self._host = None
self._ip = None
self._manufacturer = None
self._method = None
self._model = None
self._name = None
self._on_script = None
self._port = None
self._title = None
self._uuid = None
def _get_entry(self):
return self.async_create_entry(
title=self._title,
data={
CONF_HOST: self._host,
CONF_ID: self._uuid,
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,
},
)
def _try_connect(self):
"""Try to connect and check auth."""
for method in METHODS:
config = {
"name": "HomeAssistant",
"description": "HomeAssistant",
"id": "ha.component.samsung",
"host": self._host,
"method": method,
"port": self._port,
"timeout": 1,
}
try:
LOGGER.debug("Try config: %s", config)
with Remote(config.copy()):
LOGGER.debug("Working config: %s", config)
self._method = method
return RESULT_SUCCESS
except AccessDenied:
LOGGER.debug("Working but denied config: %s", config)
return RESULT_AUTH_MISSING
except UnhandledResponse:
LOGGER.debug("Working but unsupported config: %s", config)
return RESULT_NOT_SUPPORTED
except (OSError):
LOGGER.debug("Failing config: %s", config)
LOGGER.debug("No working config found")
return RESULT_NOT_FOUND
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)
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._title = user_input.get(CONF_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()
return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA)
async def async_step_ssdp(self, user_input=None):
"""Handle a flow initialized by discovery."""
host = urlparse(user_input[ATTR_SSDP_LOCATION]).hostname
ip_address = await self.hass.async_add_executor_job(_get_ip, host)
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:]
config_entry = await self.async_set_unique_id(ip_address)
if config_entry:
config_entry.data[CONF_ID] = self._uuid
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")
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()
return self.async_show_form(
step_id="confirm", description_placeholders={"model": self._model}
)

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

@ -3,3 +3,11 @@ import logging
LOGGER = logging.getLogger(__package__)
DOMAIN = "samsungtv"
DEFAULT_NAME = "Samsung TV Remote"
CONF_MANUFACTURER = "manufacturer"
CONF_MODEL = "model"
CONF_ON_ACTION = "turn_on_action"
METHODS = ("websocket", "legacy")

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

@ -2,7 +2,17 @@
"domain": "samsungtv",
"name": "Samsung Smart TV",
"documentation": "https://www.home-assistant.io/integrations/samsungtv",
"requirements": ["samsungctl[websocket]==0.7.1", "wakeonlan==1.1.6"],
"requirements": [
"samsungctl[websocket]==0.7.1"
],
"ssdp": [
{
"deviceType": "urn:samsung.com:device:RemoteControlReceiver:1"
}
],
"dependencies": [],
"codeowners": ["@escoand"]
"codeowners": [
"@escoand"
],
"config_flow": true
}

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

@ -1,18 +1,12 @@
"""Support for interface with an Samsung TV."""
import asyncio
from datetime import timedelta
import socket
from samsungctl import Remote as SamsungRemote, exceptions as samsung_exceptions
import voluptuous as vol
import wakeonlan
from websocket import WebSocketException
from homeassistant.components.media_player import (
DEVICE_CLASS_TV,
PLATFORM_SCHEMA,
MediaPlayerDevice,
)
from homeassistant.components.media_player import DEVICE_CLASS_TV, MediaPlayerDevice
from homeassistant.components.media_player.const import (
MEDIA_TYPE_CHANNEL,
SUPPORT_NEXT_TRACK,
@ -27,27 +21,20 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_STEP,
)
from homeassistant.const import (
CONF_BROADCAST_ADDRESS,
CONF_HOST,
CONF_MAC,
CONF_NAME,
CONF_ID,
CONF_METHOD,
CONF_PORT,
CONF_TIMEOUT,
STATE_OFF,
STATE_ON,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.script import Script
from homeassistant.util import dt as dt_util
from .const import LOGGER
DEFAULT_NAME = "Samsung TV Remote"
DEFAULT_TIMEOUT = 1
DEFAULT_BROADCAST_ADDRESS = "255.255.255.255"
from .const import CONF_MANUFACTURER, CONF_MODEL, CONF_ON_ACTION, DOMAIN, LOGGER
KEY_PRESS_TIMEOUT = 1.2
KNOWN_DEVICES_KEY = "samsungtv_known_devices"
METHODS = ("websocket", "legacy")
SOURCES = {"TV": "KEY_TV", "HDMI": "KEY_HDMI"}
SUPPORT_SAMSUNGTV = (
@ -62,73 +49,33 @@ SUPPORT_SAMSUNGTV = (
| SUPPORT_PLAY_MEDIA
)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PORT): cv.port,
vol.Optional(CONF_MAC): cv.string,
vol.Optional(
CONF_BROADCAST_ADDRESS, default=DEFAULT_BROADCAST_ADDRESS
): cv.string,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
}
)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_platform(
hass, config, add_entities, discovery_info=None
): # pragma: no cover
"""Set up the Samsung TV platform."""
known_devices = hass.data.get(KNOWN_DEVICES_KEY)
if known_devices is None:
known_devices = set()
hass.data[KNOWN_DEVICES_KEY] = known_devices
pass
uuid = None
# Is this a manual configuration?
if config.get(CONF_HOST) is not None:
host = config.get(CONF_HOST)
port = config.get(CONF_PORT)
name = config.get(CONF_NAME)
mac = config.get(CONF_MAC)
broadcast = config.get(CONF_BROADCAST_ADDRESS)
timeout = config.get(CONF_TIMEOUT)
elif discovery_info is not None:
tv_name = discovery_info.get("name")
model = discovery_info.get("model_name")
host = discovery_info.get("host")
name = f"{tv_name} ({model})"
if name.startswith("[TV]"):
name = name[4:]
port = None
timeout = DEFAULT_TIMEOUT
mac = None
broadcast = DEFAULT_BROADCAST_ADDRESS
uuid = discovery_info.get("udn")
if uuid and uuid.startswith("uuid:"):
uuid = uuid[len("uuid:") :]
# Only add a device once, so discovered devices do not override manual
# config.
ip_addr = socket.gethostbyname(host)
if ip_addr not in known_devices:
known_devices.add(ip_addr)
add_entities([SamsungTVDevice(host, port, name, timeout, mac, broadcast, uuid)])
LOGGER.info("Samsung TV %s added as '%s'", host, name)
else:
LOGGER.info("Ignoring duplicate Samsung TV %s", host)
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
async_add_entities([SamsungTVDevice(config_entry, on_script)])
class SamsungTVDevice(MediaPlayerDevice):
"""Representation of a Samsung TV."""
def __init__(self, host, port, name, timeout, mac, broadcast, uuid):
def __init__(self, config_entry, on_script):
"""Initialize the Samsung device."""
# Save a reference to the imported classes
self._name = name
self._mac = mac
self._broadcast = broadcast
self._uuid = uuid
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._on_script = on_script
self._update_listener = None
# Assume that the TV is not muted
self._muted = False
# Assume that the TV is in Play mode
@ -141,57 +88,20 @@ class SamsungTVDevice(MediaPlayerDevice):
# Generate a configuration for the Samsung library
self._config = {
"name": "HomeAssistant",
"description": name,
"description": self._name,
"id": "ha.component.samsung",
"method": None,
"port": port,
"host": host,
"timeout": timeout,
"method": config_entry.data[CONF_METHOD],
"port": config_entry.data.get(CONF_PORT),
"host": config_entry.data[CONF_HOST],
"timeout": 1,
}
# Select method by port number, mainly for fallback
if self._config["port"] in (8001, 8002):
self._config["method"] = "websocket"
elif self._config["port"] == 55000:
self._config["method"] = "legacy"
def update(self):
"""Update state of device."""
self.send_key("KEY")
def get_remote(self):
"""Create or return a remote control instance."""
# Try to find correct method automatically
if self._config["method"] not in METHODS:
for method in METHODS:
try:
self._config["method"] = method
LOGGER.debug("Try config: %s", self._config)
self._remote = SamsungRemote(self._config.copy())
self._state = STATE_ON
LOGGER.debug("Found working config: %s", self._config)
break
except (
samsung_exceptions.UnhandledResponse,
samsung_exceptions.AccessDenied,
):
# We got a response so it's working.
self._state = STATE_ON
LOGGER.debug(
"Found working config without connection: %s", self._config
)
break
except OSError as err:
LOGGER.debug("Failing config: %s error was: %s", self._config, err)
self._config["method"] = None
# Unable to find working connection
if self._config["method"] is None:
self._remote = None
self._state = None
return None
if self._remote is None:
# We need to create a new instance to reconnect.
self._remote = SamsungRemote(self._config.copy())
@ -219,9 +129,6 @@ class SamsungTVDevice(MediaPlayerDevice):
# WebSocketException can occur when timed out
self._remote = None
self._state = STATE_ON
except AttributeError:
# Auto-detect could not find working config yet
pass
except (samsung_exceptions.UnhandledResponse, samsung_exceptions.AccessDenied):
# We got a response so it's on.
self._state = STATE_ON
@ -256,6 +163,16 @@ class SamsungTVDevice(MediaPlayerDevice):
"""Return the state of the device."""
return self._state
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(DOMAIN, self.unique_id)},
"manufacturer": self._manufacturer,
"model": self._model,
}
@property
def is_volume_muted(self):
"""Boolean if volume is currently muted."""
@ -269,7 +186,7 @@ class SamsungTVDevice(MediaPlayerDevice):
@property
def supported_features(self):
"""Flag media player features that are supported."""
if self._mac:
if self._on_script:
return SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON
return SUPPORT_SAMSUNGTV
@ -344,21 +261,19 @@ class SamsungTVDevice(MediaPlayerDevice):
return
for digit in media_id:
await self.hass.async_add_job(self.send_key, f"KEY_{digit}")
await self.hass.async_add_executor_job(self.send_key, f"KEY_{digit}")
await asyncio.sleep(KEY_PRESS_TIMEOUT, self.hass.loop)
await self.hass.async_add_job(self.send_key, "KEY_ENTER")
await self.hass.async_add_executor_job(self.send_key, "KEY_ENTER")
def turn_on(self):
async def async_turn_on(self):
"""Turn the media player on."""
if self._mac:
wakeonlan.send_magic_packet(self._mac, ip_address=self._broadcast)
else:
self.send_key("KEY_POWERON")
if self._on_script:
await self._on_script.async_run()
async def async_select_source(self, source):
def select_source(self, source):
"""Select input source."""
if source not in SOURCES:
LOGGER.error("Unsupported source")
return
await self.hass.async_add_job(self.send_key, SOURCES[source])
self.send_key(SOURCES[source])

26
homeassistant/components/samsungtv/strings.json Normal file
View File

@ -0,0 +1,26 @@
{
"config": {
"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.",
"data": {
"host": "Host or IP address",
"name": "Name"
}
},
"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."
}
},
"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."
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -66,6 +66,7 @@ FLOWS = [
"point",
"ps4",
"rainmachine",
"samsungtv",
"sentry",
"simplisafe",
"smartthings",

5
homeassistant/generated/ssdp.py
View File

@ -27,6 +27,11 @@ SSDP = {
"manufacturer": "Royal Philips Electronics"
}
],
"samsungtv": [
{
"deviceType": "urn:samsung.com:device:RemoteControlReceiver:1"
}
],
"sonos": [
{
"st": "urn:schemas-upnp-org:device:ZonePlayer:1"

1
requirements_all.txt
View File

@ -2035,7 +2035,6 @@ vtjp==0.1.14
vultr==0.1.2
# homeassistant.components.panasonic_viera
# homeassistant.components.samsungtv
# homeassistant.components.wake_on_lan
wakeonlan==1.1.6

1
requirements_test_all.txt
View File

@ -643,7 +643,6 @@ vsure==1.5.4
vultr==0.1.2
# homeassistant.components.panasonic_viera
# homeassistant.components.samsungtv
# homeassistant.components.wake_on_lan
wakeonlan==1.1.6

388
tests/components/samsungtv/test_config_flow.py Normal file
View File

@ -0,0 +1,388 @@
"""Tests for Samsung TV config flow."""
from unittest.mock import call, patch
from asynctest import mock
import pytest
from samsungctl.exceptions import AccessDenied, UnhandledResponse
from homeassistant.components.samsungtv.const import (
CONF_MANUFACTURER,
CONF_MODEL,
DOMAIN,
)
from homeassistant.components.ssdp import (
ATTR_SSDP_LOCATION,
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_UDN,
)
from homeassistant.const import CONF_HOST, CONF_ID, CONF_METHOD, CONF_NAME
MOCK_USER_DATA = {CONF_HOST: "fake_host", CONF_NAME: "fake_name"}
MOCK_SSDP_DATA = {
ATTR_SSDP_LOCATION: "https://fake_host:12345/test",
ATTR_UPNP_FRIENDLY_NAME: "[TV]fake_name",
ATTR_UPNP_MANUFACTURER: "fake_manufacturer",
ATTR_UPNP_MODEL_NAME: "fake_model",
ATTR_UPNP_UDN: "uuid:fake_uuid",
}
MOCK_SSDP_DATA_NOPREFIX = {
ATTR_SSDP_LOCATION: "http://fake2_host:12345/test",
ATTR_UPNP_FRIENDLY_NAME: "fake2_name",
ATTR_UPNP_MANUFACTURER: "fake2_manufacturer",
ATTR_UPNP_MODEL_NAME: "fake2_model",
ATTR_UPNP_UDN: "fake2_uuid",
}
AUTODETECT_WEBSOCKET = {
"name": "HomeAssistant",
"description": "HomeAssistant",
"id": "ha.component.samsung",
"method": "websocket",
"port": None,
"host": "fake_host",
"timeout": 1,
}
AUTODETECT_LEGACY = {
"name": "HomeAssistant",
"description": "HomeAssistant",
"id": "ha.component.samsung",
"method": "legacy",
"port": None,
"host": "fake_host",
"timeout": 1,
}
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch("samsungctl.Remote") as remote_class, patch(
"homeassistant.components.samsungtv.config_flow.socket"
) as socket_class:
remote = mock.Mock()
remote.__enter__ = mock.Mock()
remote.__exit__ = mock.Mock()
remote_class.return_value = remote
socket = mock.Mock()
socket_class.return_value = socket
yield remote
async def test_user(hass, remote):
"""Test starting a flow by user."""
# show form
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
# entry was added
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=MOCK_USER_DATA
)
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_MANUFACTURER] is None
assert result["data"][CONF_MODEL] is None
assert result["data"][CONF_ID] is None
async def test_user_missing_auth(hass):
"""Test starting a flow by user with authentication."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=AccessDenied("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
# missing authentication
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "auth_missing"
async def test_user_not_supported(hass):
"""Test starting a flow by user for not supported device."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=UnhandledResponse("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
# device not supported
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "not_supported"
async def test_user_not_found(hass):
"""Test starting a flow by user but no device found."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=OSError("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
# device not found
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"
async def test_user_already_configured(hass, remote):
"""Test starting a flow by user when already configured."""
# entry was added
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
# failed as already configured
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
async def test_ssdp(hass, remote):
"""Test starting a flow from discovery."""
# confirm to add the entry
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
# entry was added
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input="whatever"
)
assert result["type"] == "create_entry"
assert result["title"] == "fake_name (fake_model)"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake_name"
assert result["data"][CONF_MANUFACTURER] == "fake_manufacturer"
assert result["data"][CONF_MODEL] == "fake_model"
assert result["data"][CONF_ID] == "fake_uuid"
async def test_ssdp_noprefix(hass, remote):
"""Test starting a flow from discovery without prefixes."""
# confirm to add the entry
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA_NOPREFIX
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
# entry was added
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input="whatever"
)
assert result["type"] == "create_entry"
assert result["title"] == "fake2_name (fake2_model)"
assert result["data"][CONF_HOST] == "fake2_host"
assert result["data"][CONF_NAME] == "fake2_name"
assert result["data"][CONF_MANUFACTURER] == "fake2_manufacturer"
assert result["data"][CONF_MODEL] == "fake2_model"
assert result["data"][CONF_ID] == "fake2_uuid"
async def test_ssdp_missing_auth(hass):
"""Test starting a flow from discovery with authentication."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=AccessDenied("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
# confirm to add the entry
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
# missing authentication
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input="whatever"
)
assert result["type"] == "abort"
assert result["reason"] == "auth_missing"
async def test_ssdp_not_supported(hass):
"""Test starting a flow from discovery for not supported device."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=UnhandledResponse("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
# confirm to add the entry
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
# device not supported
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input="whatever"
)
assert result["type"] == "abort"
assert result["reason"] == "not_supported"
async def test_ssdp_not_found(hass):
"""Test starting a flow from discovery but no device found."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=OSError("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
# confirm to add the entry
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
# device not found
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input="whatever"
)
assert result["type"] == "abort"
assert result["reason"] == "not_found"
async def test_ssdp_already_in_progress(hass, remote):
"""Test starting a flow from discovery twice."""
# confirm to add the entry
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
# failed as already in progress
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "already_in_progress"
async def test_ssdp_already_configured(hass, remote):
"""Test starting a flow from discovery when already configured."""
# entry was added
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
assert result["data"][CONF_MANUFACTURER] is None
assert result["data"][CONF_MODEL] is None
assert result["data"][CONF_ID] is None
# failed as already configured
result2 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA
)
assert result2["type"] == "abort"
assert result2["reason"] == "already_configured"
# check updated device info
assert result["data"][CONF_MANUFACTURER] == "fake_manufacturer"
assert result["data"][CONF_MODEL] == "fake_model"
assert result["data"][CONF_ID] == "fake_uuid"
async def test_autodetect_websocket(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch("homeassistant.components.samsungtv.config_flow.Remote") as remote:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
assert result["data"][CONF_METHOD] == "websocket"
assert remote.call_count == 1
assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)]
async def test_autodetect_auth_missing(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=[AccessDenied("Boom")],
) as remote:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "auth_missing"
assert remote.call_count == 1
assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)]
async def test_autodetect_not_supported(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=[UnhandledResponse("Boom")],
) as remote:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "not_supported"
assert remote.call_count == 1
assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)]
async def test_autodetect_legacy(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=[OSError("Boom"), mock.DEFAULT],
) as remote:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
assert result["data"][CONF_METHOD] == "legacy"
assert remote.call_count == 2
assert remote.call_args_list == [
call(AUTODETECT_WEBSOCKET),
call(AUTODETECT_LEGACY),
]
async def test_autodetect_none(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=OSError("Boom"),
) as remote:
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 remote.call_count == 2
assert remote.call_args_list == [
call(AUTODETECT_WEBSOCKET),
call(AUTODETECT_LEGACY),
]

97
tests/components/samsungtv/test_init.py Normal file
View File

@ -0,0 +1,97 @@
"""Tests for the Samsung TV Integration."""
from unittest.mock import 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,
)
from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_SUPPORTED_FEATURES,
CONF_HOST,
CONF_NAME,
CONF_PORT,
SERVICE_VOLUME_UP,
)
from homeassistant.setup import async_setup_component
ENTITY_ID = f"{DOMAIN}.fake_name"
MOCK_CONFIG = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake_host",
CONF_NAME: "fake_name",
CONF_PORT: 1234,
CONF_ON_ACTION: [{"delay": "00:00:01"}],
}
]
}
REMOTE_CALL = {
"name": "HomeAssistant",
"description": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_NAME],
"id": "ha.component.samsung",
"method": "websocket",
"port": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_PORT],
"host": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_HOST],
"timeout": 1,
}
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch("homeassistant.components.samsungtv.socket"), patch(
"homeassistant.components.samsungtv.config_flow.socket"
), patch("homeassistant.components.samsungtv.config_flow.Remote"), patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote:
yield remote
async def test_setup(hass, remote):
"""Test Samsung TV integration is setup."""
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
assert state
assert state.name == "fake_name"
assert (
state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON
)
# test host and port
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
assert remote.mock_calls[0] == call(REMOTE_CALL)
async def test_setup_duplicate_config(hass, remote, caplog):
"""Test duplicate setup of platform."""
DUPLICATE = {
SAMSUNGTV_DOMAIN: [
MOCK_CONFIG[SAMSUNGTV_DOMAIN][0],
MOCK_CONFIG[SAMSUNGTV_DOMAIN][0],
]
}
await async_setup_component(hass, SAMSUNGTV_DOMAIN, DUPLICATE)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID) is None
assert len(hass.states.async_all()) == 0
assert "duplicate host entries found" in caplog.text
async def test_setup_duplicate_entries(hass, remote, caplog):
"""Test duplicate setup of platform."""
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID)
assert len(hass.states.async_all()) == 1
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
assert len(hass.states.async_all()) == 1

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

@ -2,9 +2,9 @@
import asyncio
from datetime import timedelta
import logging
from unittest.mock import call, patch
from asynctest import mock
from asynctest.mock import call, patch
import pytest
from samsungctl import exceptions
from websocket import WebSocketException
@ -22,21 +22,18 @@ from homeassistant.components.media_player.const import (
SERVICE_SELECT_SOURCE,
SUPPORT_TURN_ON,
)
from homeassistant.components.samsungtv.const import DOMAIN as SAMSUNGTV_DOMAIN
from homeassistant.components.samsungtv.media_player import (
CONF_TIMEOUT,
SUPPORT_SAMSUNGTV,
from homeassistant.components.samsungtv.const import (
CONF_ON_ACTION,
DOMAIN as SAMSUNGTV_DOMAIN,
)
from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_FRIENDLY_NAME,
ATTR_SUPPORTED_FEATURES,
CONF_BROADCAST_ADDRESS,
CONF_HOST,
CONF_MAC,
CONF_NAME,
CONF_PLATFORM,
CONF_PORT,
SERVICE_MEDIA_NEXT_TRACK,
SERVICE_MEDIA_PAUSE,
@ -49,9 +46,7 @@ from homeassistant.const import (
SERVICE_VOLUME_UP,
STATE_OFF,
STATE_ON,
STATE_UNKNOWN,
)
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
@ -59,107 +54,46 @@ from tests.common import async_fire_time_changed
ENTITY_ID = f"{DOMAIN}.fake"
MOCK_CONFIG = {
DOMAIN: {
CONF_PLATFORM: SAMSUNGTV_DOMAIN,
CONF_HOST: "fake",
CONF_NAME: "fake",
CONF_PORT: 8001,
CONF_TIMEOUT: 10,
CONF_MAC: "38:f9:d3:82:b4:f1",
}
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake",
CONF_NAME: "fake",
CONF_PORT: 8001,
CONF_ON_ACTION: [{"delay": "00:00:01"}],
}
]
}
ENTITY_ID_BROADCAST = f"{DOMAIN}.fake_broadcast"
MOCK_CONFIG_BROADCAST = {
DOMAIN: {
CONF_PLATFORM: SAMSUNGTV_DOMAIN,
CONF_HOST: "fake_broadcast",
CONF_NAME: "fake_broadcast",
CONF_PORT: 8001,
CONF_TIMEOUT: 10,
CONF_MAC: "38:f9:d3:82:b4:f1",
CONF_BROADCAST_ADDRESS: "192.168.5.255",
}
}
ENTITY_ID_NOMAC = f"{DOMAIN}.fake_nomac"
MOCK_CONFIG_NOMAC = {
DOMAIN: {
CONF_PLATFORM: SAMSUNGTV_DOMAIN,
CONF_HOST: "fake_nomac",
CONF_NAME: "fake_nomac",
CONF_PORT: 55000,
CONF_TIMEOUT: 10,
}
}
ENTITY_ID_AUTO = f"{DOMAIN}.fake_auto"
MOCK_CONFIG_AUTO = {
DOMAIN: {
CONF_PLATFORM: SAMSUNGTV_DOMAIN,
CONF_HOST: "fake_auto",
CONF_NAME: "fake_auto",
}
}
ENTITY_ID_DISCOVERY = f"{DOMAIN}.fake_discovery_fake_model"
MOCK_CONFIG_DISCOVERY = {
"name": "fake_discovery",
"model_name": "fake_model",
"host": "fake_host",
"udn": "fake_uuid",
}
ENTITY_ID_DISCOVERY_PREFIX = f"{DOMAIN}.fake_discovery_prefix_fake_model_prefix"
MOCK_CONFIG_DISCOVERY_PREFIX = {
"name": "[TV]fake_discovery_prefix",
"model_name": "fake_model_prefix",
"host": "fake_host_prefix",
"udn": "uuid:fake_uuid_prefix",
}
AUTODETECT_WEBSOCKET = {
"name": "HomeAssistant",
"description": "fake_auto",
"id": "ha.component.samsung",
"method": "websocket",
"port": None,
"host": "fake_auto",
"timeout": 1,
}
AUTODETECT_LEGACY = {
"name": "HomeAssistant",
"description": "fake_auto",
"id": "ha.component.samsung",
"method": "legacy",
"port": None,
"host": "fake_auto",
"timeout": 1,
ENTITY_ID_NOTURNON = f"{DOMAIN}.fake_noturnon"
MOCK_CONFIG_NOTURNON = {
SAMSUNGTV_DOMAIN: [
{CONF_HOST: "fake_noturnon", CONF_NAME: "fake_noturnon", CONF_PORT: 55000}
]
}
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch(
with patch("homeassistant.components.samsungtv.config_flow.socket"), patch(
"homeassistant.components.samsungtv.config_flow.Remote"
), patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote_class, patch(
"homeassistant.components.samsungtv.media_player.socket"
) as socket_class:
"homeassistant.components.samsungtv.socket"
):
remote = mock.Mock()
remote_class.return_value = remote
socket = mock.Mock()
socket_class.return_value = socket
yield remote
@pytest.fixture(name="wakeonlan")
def wakeonlan_fixture():
"""Patch the wakeonlan Remote."""
@pytest.fixture(name="delay")
def delay_fixture():
"""Patch the delay script function."""
with patch(
"homeassistant.components.samsungtv.media_player.wakeonlan"
) as wakeonlan_module:
yield wakeonlan_module
"homeassistant.components.samsungtv.media_player.Script.async_run"
) as delay:
yield delay
@pytest.fixture
@ -170,61 +104,20 @@ def mock_now():
async def setup_samsungtv(hass, config):
"""Set up mock Samsung TV."""
await async_setup_component(hass, "media_player", config)
await async_setup_component(hass, SAMSUNGTV_DOMAIN, config)
await hass.async_block_till_done()
async def test_setup_with_mac(hass, remote):
async def test_setup_with_turnon(hass, remote):
"""Test setup of platform."""
await setup_samsungtv(hass, MOCK_CONFIG)
assert hass.states.get(ENTITY_ID)
async def test_setup_duplicate(hass, remote, caplog):
"""Test duplicate setup of platform."""
DUPLICATE = {DOMAIN: [MOCK_CONFIG[DOMAIN], MOCK_CONFIG[DOMAIN]]}
await setup_samsungtv(hass, DUPLICATE)
assert "Ignoring duplicate Samsung TV fake" in caplog.text
async def test_setup_without_mac(hass, remote):
async def test_setup_without_turnon(hass, remote):
"""Test setup of platform."""
await setup_samsungtv(hass, MOCK_CONFIG_NOMAC)
assert hass.states.get(ENTITY_ID_NOMAC)
async def test_setup_discovery(hass, remote):
"""Test setup of platform with discovery."""
hass.async_create_task(
async_load_platform(
hass, DOMAIN, SAMSUNGTV_DOMAIN, MOCK_CONFIG_DISCOVERY, {DOMAIN: {}}
)
)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID_DISCOVERY)
assert state
assert state.name == "fake_discovery (fake_model)"
entity_registry = await hass.helpers.entity_registry.async_get_registry()
entry = entity_registry.async_get(ENTITY_ID_DISCOVERY)
assert entry
assert entry.unique_id == "fake_uuid"
async def test_setup_discovery_prefix(hass, remote):
"""Test setup of platform with discovery."""
hass.async_create_task(
async_load_platform(
hass, DOMAIN, SAMSUNGTV_DOMAIN, MOCK_CONFIG_DISCOVERY_PREFIX, {DOMAIN: {}}
)
)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID_DISCOVERY_PREFIX)
assert state
assert state.name == "fake_discovery_prefix (fake_model_prefix)"
entity_registry = await hass.helpers.entity_registry.async_get_registry()
entry = entity_registry.async_get(ENTITY_ID_DISCOVERY_PREFIX)
assert entry
assert entry.unique_id == "fake_uuid_prefix"
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
assert hass.states.get(ENTITY_ID_NOTURNON)
async def test_update_on(hass, remote, mock_now):
@ -254,7 +147,7 @@ async def test_update_off(hass, remote, mock_now):
assert state.state == STATE_OFF
async def test_send_key(hass, remote, wakeonlan):
async def test_send_key(hass, remote):
"""Test for send key."""
await setup_samsungtv(hass, MOCK_CONFIG)
assert await hass.services.async_call(
@ -267,85 +160,6 @@ async def test_send_key(hass, remote, wakeonlan):
assert state.state == STATE_ON
async def test_send_key_autodetect_websocket(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote, patch("homeassistant.components.samsungtv.media_player.socket"):
await setup_samsungtv(hass, MOCK_CONFIG_AUTO)
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_AUTO}, True
)
state = hass.states.get(ENTITY_ID_AUTO)
assert remote.call_count == 1
assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)]
assert state.state == STATE_ON
async def test_send_key_autodetect_websocket_exception(hass, caplog):
"""Test for send key with autodetection of protocol."""
caplog.set_level(logging.DEBUG)
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote",
side_effect=[exceptions.AccessDenied("Boom"), mock.DEFAULT],
) as remote, patch("homeassistant.components.samsungtv.media_player.socket"):
await setup_samsungtv(hass, MOCK_CONFIG_AUTO)
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_AUTO}, True
)
state = hass.states.get(ENTITY_ID_AUTO)
# called 2 times because of the exception and the send key
assert remote.call_count == 2
assert remote.call_args_list == [
call(AUTODETECT_WEBSOCKET),
call(AUTODETECT_WEBSOCKET),
]
assert state.state == STATE_ON
assert "Found working config without connection: " in caplog.text
assert "Failing config: " not in caplog.text
async def test_send_key_autodetect_legacy(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote",
side_effect=[OSError("Boom"), mock.DEFAULT],
) as remote, patch("homeassistant.components.samsungtv.media_player.socket"):
await setup_samsungtv(hass, MOCK_CONFIG_AUTO)
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_AUTO}, True
)
state = hass.states.get(ENTITY_ID_AUTO)
assert remote.call_count == 2
assert remote.call_args_list == [
call(AUTODETECT_WEBSOCKET),
call(AUTODETECT_LEGACY),
]
assert state.state == STATE_ON
async def test_send_key_autodetect_none(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote",
side_effect=OSError("Boom"),
) as remote, patch("homeassistant.components.samsungtv.media_player.socket"):
await setup_samsungtv(hass, MOCK_CONFIG_AUTO)
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_AUTO}, True
)
state = hass.states.get(ENTITY_ID_AUTO)
# 4 calls because of retry
assert remote.call_count == 4
assert remote.call_args_list == [
call(AUTODETECT_WEBSOCKET),
call(AUTODETECT_LEGACY),
call(AUTODETECT_WEBSOCKET),
call(AUTODETECT_LEGACY),
]
assert state.state == STATE_UNKNOWN
async def test_send_key_broken_pipe(hass, remote):
"""Testing broken pipe Exception."""
await setup_samsungtv(hass, MOCK_CONFIG)
@ -417,7 +231,7 @@ async def test_name(hass, remote):
assert state.attributes[ATTR_FRIENDLY_NAME] == "fake"
async def test_state_with_mac(hass, remote, wakeonlan):
async def test_state_with_turnon(hass, remote, delay):
"""Test for state property."""
await setup_samsungtv(hass, MOCK_CONFIG)
assert await hass.services.async_call(
@ -425,6 +239,8 @@ async def test_state_with_mac(hass, remote, wakeonlan):
)
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
assert delay.call_count == 1
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
@ -432,22 +248,22 @@ async def test_state_with_mac(hass, remote, wakeonlan):
assert state.state == STATE_OFF
async def test_state_without_mac(hass, remote):
async def test_state_without_turnon(hass, remote):
"""Test for state property."""
await setup_samsungtv(hass, MOCK_CONFIG_NOMAC)
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True
)
state = hass.states.get(ENTITY_ID_NOMAC)
state = hass.states.get(ENTITY_ID_NOTURNON)
assert state.state == STATE_ON
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True
)
state = hass.states.get(ENTITY_ID_NOMAC)
state = hass.states.get(ENTITY_ID_NOTURNON)
assert state.state == STATE_OFF
async def test_supported_features_with_mac(hass, remote):
async def test_supported_features_with_turnon(hass, remote):
"""Test for supported_features property."""
await setup_samsungtv(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
@ -456,10 +272,10 @@ async def test_supported_features_with_mac(hass, remote):
)
async def test_supported_features_without_mac(hass, remote):
async def test_supported_features_without_turnon(hass, remote):
"""Test for supported_features property."""
await setup_samsungtv(hass, MOCK_CONFIG_NOMAC)
state = hass.states.get(ENTITY_ID_NOMAC)
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
state = hass.states.get(ENTITY_ID_NOTURNON)
assert state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV
@ -481,15 +297,25 @@ async def test_turn_off_websocket(hass, remote):
assert remote.control.call_args_list == [call("KEY_POWER")]
async def test_turn_off_legacy(hass, remote):
async def test_turn_off_legacy(hass):
"""Test for turn_off."""
await setup_samsungtv(hass, MOCK_CONFIG_NOMAC)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True
)
# key called
assert remote.control.call_count == 1
assert remote.control.call_args_list == [call("KEY_POWEROFF")]
with patch("homeassistant.components.samsungtv.config_flow.socket"), patch(
"homeassistant.components.samsungtv.config_flow.Remote",
side_effect=[OSError("Boom"), mock.DEFAULT],
), patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote_class, patch(
"homeassistant.components.samsungtv.socket"
):
remote = mock.Mock()
remote_class.return_value = remote
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True
)
# key called
assert remote.control.call_count == 1
assert remote.control.call_args_list == [call("KEY_POWEROFF")]
async def test_turn_off_os_error(hass, remote, caplog):
@ -583,37 +409,20 @@ async def test_media_previous_track(hass, remote):
assert remote.control.call_args_list == [call("KEY_CHDOWN"), call("KEY")]
async def test_turn_on_with_mac(hass, remote, wakeonlan):
async def test_turn_on_with_turnon(hass, remote, delay):
"""Test turn on."""
await setup_samsungtv(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# key and update called
assert wakeonlan.send_magic_packet.call_count == 1
assert wakeonlan.send_magic_packet.call_args_list == [
call("38:f9:d3:82:b4:f1", ip_address="255.255.255.255")
]
assert delay.call_count == 1
async def test_turn_on_with_mac_and_broadcast(hass, remote, wakeonlan):
async def test_turn_on_without_turnon(hass, remote):
"""Test turn on."""
await setup_samsungtv(hass, MOCK_CONFIG_BROADCAST)
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID_BROADCAST}, True
)
# key and update called
assert wakeonlan.send_magic_packet.call_count == 1
assert wakeonlan.send_magic_packet.call_args_list == [
call("38:f9:d3:82:b4:f1", ip_address="192.168.5.255")
]
async def test_turn_on_without_mac(hass, remote):
"""Test turn on."""
await setup_samsungtv(hass, MOCK_CONFIG_NOMAC)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True
)
# nothing called as not supported feature
assert remote.control.call_count == 0