jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-26 02:07:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add support for newer SamsungTV models (#31537)

Browse Source 
* Added support for newer SamsungTV models

* Fixed legacy port

* store token in HA config directory

* Change token name

* rebasing and exception handling

* implement update

* fix error creating mediaplayer

* Debug logging

* Increase timeout

* Restore update timeout

* Store token_file path in config_entry

* Introduction of samsung bridge class

* Added bridge class functions

* Code cleanup

* more fixes

* Begin testing

* samsungtvws 1.2.0

* Config flow tests 0.1

* Fixed some mediaplayer tests

* Fixed fixture in media player

* use of constants and turn off

* more media player tests

* samsungtvws 1.3.1 and other fixes

* WS tv update rewritten

* more tests

* test_init

* fixed tests

* removed reset mock

* tests reset mock

* close_remote and tests

* deprecate port config

* deprecate port config 2

* deprecate port config 3

* save token only if needed

* cleanup

* better websocket protocol detection

* config removal
This commit is contained in:
Paolo Tuninetto 2020-03-10 11:48:09 +01:00 committed by GitHub
parent 2e802c88f8
commit a8758ed3a1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 646 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -23,6 +23,7 @@ CONFIG_SCHEMA = vol.Schema(
DOMAIN: vol.All(
cv.ensure_list,
[
cv.deprecated(CONF_PORT),
vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
@ -30,7 +31,7 @@ CONFIG_SCHEMA = vol.Schema(
vol.Optional(CONF_PORT): cv.port,
vol.Optional(CONF_ON_ACTION): cv.SCRIPT_SCHEMA,
}
)
),
],
ensure_unique_hosts,
)

254
homeassistant/components/samsungtv/bridge.py Normal file
View File

@ -0,0 +1,254 @@
"""samsungctl and samsungtvws bridge classes."""
from abc import ABC, abstractmethod
from samsungctl import Remote
from samsungctl.exceptions import AccessDenied, ConnectionClosed, UnhandledResponse
from samsungtvws import SamsungTVWS
from samsungtvws.exceptions import ConnectionFailure
from websocket import WebSocketException
from homeassistant.const import (
CONF_HOST,
CONF_ID,
CONF_METHOD,
CONF_NAME,
CONF_PORT,
CONF_TIMEOUT,
CONF_TOKEN,
)
from .const import (
CONF_DESCRIPTION,
LOGGER,
METHOD_LEGACY,
RESULT_AUTH_MISSING,
RESULT_NOT_SUCCESSFUL,
RESULT_NOT_SUPPORTED,
RESULT_SUCCESS,
VALUE_CONF_ID,
VALUE_CONF_NAME,
)
class SamsungTVBridge(ABC):
"""The Base Bridge abstract class."""
@staticmethod
def get_bridge(method, host, port=None, token=None):
"""Get Bridge instance."""
if method == METHOD_LEGACY:
return SamsungTVLegacyBridge(method, host, port)
return SamsungTVWSBridge(method, host, port, token)
def __init__(self, method, host, port):
"""Initialize Bridge."""
self.port = port
self.method = method
self.host = host
self.token = None
self._remote = None
self._callback = None
def register_reauth_callback(self, func):
"""Register a callback function."""
self._callback = func
@abstractmethod
def try_connect(self):
"""Try to connect to the TV."""
def is_on(self):
"""Tells if the TV is on."""
self.close_remote()
try:
return self._get_remote() is not None
except (
UnhandledResponse,
AccessDenied,
ConnectionFailure,
):
# We got a response so it's working.
return True
except OSError:
# Different reasons, e.g. hostname not resolveable
return False
def send_key(self, key):
"""Send a key to the tv and handles exceptions."""
try:
# recreate connection if connection was dead
retry_count = 1
for _ in range(retry_count + 1):
try:
self._send_key(key)
break
except (
ConnectionClosed,
BrokenPipeError,
WebSocketException,
):
# BrokenPipe can occur when the commands is sent to fast
# WebSocketException can occur when timed out
self._remote = None
except (UnhandledResponse, AccessDenied):
# We got a response so it's on.
LOGGER.debug("Failed sending command %s", key, exc_info=True)
except OSError:
# Different reasons, e.g. hostname not resolveable
pass
@abstractmethod
def _send_key(self, key):
"""Send the key."""
@abstractmethod
def _get_remote(self):
"""Get Remote object."""
def close_remote(self):
"""Close remote object."""
try:
if self._remote is not None:
# Close the current remote connection
self._remote.close()
self._remote = None
except OSError:
LOGGER.debug("Could not establish connection")
def _notify_callback(self):
"""Notify access denied callback."""
if self._callback:
self._callback()
class SamsungTVLegacyBridge(SamsungTVBridge):
"""The Bridge for Legacy TVs."""
def __init__(self, method, host, port):
"""Initialize Bridge."""
super().__init__(method, host, None)
self.config = {
CONF_NAME: VALUE_CONF_NAME,
CONF_ID: VALUE_CONF_ID,
CONF_DESCRIPTION: VALUE_CONF_NAME,
CONF_METHOD: method,
CONF_HOST: host,
CONF_TIMEOUT: 1,
}
def try_connect(self):
"""Try to connect to the Legacy TV."""
config = {
CONF_NAME: VALUE_CONF_NAME,
CONF_DESCRIPTION: VALUE_CONF_NAME,
CONF_ID: VALUE_CONF_ID,
CONF_HOST: self.host,
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,
}
try:
LOGGER.debug("Try config: %s", config)
with Remote(config.copy()):
LOGGER.debug("Working config: %s", config)
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 as err:
LOGGER.debug("Failing config: %s, error: %s", config, err)
return RESULT_NOT_SUCCESSFUL
def _get_remote(self):
"""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 SamsungRemote")
self._remote = Remote(self.config.copy())
# This is only happening when the auth was switched to DENY
# A removed auth will lead to socket timeout because waiting for auth popup is just an open socket
except AccessDenied:
self._notify_callback()
raise
return self._remote
def _send_key(self, key):
"""Send the key using legacy protocol."""
self._get_remote().control(key)
class SamsungTVWSBridge(SamsungTVBridge):
"""The Bridge for WebSocket TVs."""
def __init__(self, method, host, port, token=None):
"""Initialize Bridge."""
super().__init__(method, host, port)
self.token = token
def try_connect(self):
"""Try to connect to the Websocket TV."""
for self.port in (8001, 8002):
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,
}
try:
LOGGER.debug("Try config: %s", config)
with SamsungTVWS(
host=self.host,
port=self.port,
token=self.token,
timeout=config[CONF_TIMEOUT],
name=config[CONF_NAME],
) as remote:
remote.open()
self.token = remote.token
if self.token:
config[CONF_TOKEN] = "*****"
LOGGER.debug("Working config: %s", config)
return RESULT_SUCCESS
except WebSocketException:
LOGGER.debug("Working but unsupported config: %s", config)
return RESULT_NOT_SUPPORTED
except (OSError, ConnectionFailure) as err:
LOGGER.debug("Failing config: %s, error: %s", config, err)
return RESULT_NOT_SUCCESSFUL
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):
"""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")
self._remote = SamsungTVWS(
host=self.host,
port=self.port,
token=self.token,
timeout=1,
name=VALUE_CONF_NAME,
)
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
return self._remote

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

@ -2,10 +2,7 @@
import socket
from urllib.parse import urlparse
from samsungctl import Remote
from samsungctl.exceptions import AccessDenied, UnhandledResponse
import voluptuous as vol
from websocket import WebSocketException
from homeassistant import config_entries
from homeassistant.components.ssdp import (
@ -21,23 +18,25 @@ from homeassistant.const import (
CONF_METHOD,
CONF_NAME,
CONF_PORT,
CONF_TOKEN,
)
# pylint:disable=unused-import
from .const import CONF_MANUFACTURER, CONF_MODEL, DOMAIN, LOGGER
from .bridge import SamsungTVBridge
from .const import (
CONF_MANUFACTURER,
CONF_MODEL,
DOMAIN,
LOGGER,
METHOD_LEGACY,
METHOD_WEBSOCKET,
RESULT_AUTH_MISSING,
RESULT_NOT_SUCCESSFUL,
RESULT_SUCCESS,
)
DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str, vol.Required(CONF_NAME): str})
RESULT_AUTH_MISSING = "auth_missing"
RESULT_SUCCESS = "success"
RESULT_NOT_SUCCESSFUL = "not_successful"
RESULT_NOT_SUPPORTED = "not_supported"
SUPPORTED_METHODS = (
{"method": "websocket", "timeout": 1},
# We need this high timeout because waiting for auth popup is just an open socket
{"method": "legacy", "timeout": 31},
)
SUPPORTED_METHODS = [METHOD_LEGACY, METHOD_WEBSOCKET]
def _get_ip(host):
@ -59,61 +58,39 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._host = None
self._ip = None
self._manufacturer = None
self._method = None
self._model = None
self._name = None
self._port = None
self._title = None
self._id = None
self._bridge = None
def _get_entry(self):
return self.async_create_entry(
title=self._title,
data={
CONF_HOST: self._host,
CONF_ID: self._id,
CONF_IP_ADDRESS: self._ip,
CONF_MANUFACTURER: self._manufacturer,
CONF_METHOD: self._method,
CONF_MODEL: self._model,
CONF_NAME: self._name,
CONF_PORT: self._port,
},
)
data = {
CONF_HOST: self._host,
CONF_ID: self._id,
CONF_IP_ADDRESS: self._ip,
CONF_MANUFACTURER: self._manufacturer,
CONF_METHOD: self._bridge.method,
CONF_MODEL: self._model,
CONF_NAME: self._name,
CONF_PORT: self._bridge.port,
}
if self._bridge.token:
data[CONF_TOKEN] = self._bridge.token
return self.async_create_entry(title=self._title, data=data,)
def _try_connect(self):
"""Try to connect and check auth."""
for cfg in SUPPORTED_METHODS:
config = {
"name": "HomeAssistant",
"description": "HomeAssistant",
"id": "ha.component.samsung",
"host": self._host,
"port": self._port,
}
config.update(cfg)
try:
LOGGER.debug("Try config: %s", config)
with Remote(config.copy()):
LOGGER.debug("Working config: %s", config)
self._method = cfg["method"]
return RESULT_SUCCESS
except AccessDenied:
LOGGER.debug("Working but denied config: %s", config)
return RESULT_AUTH_MISSING
except (UnhandledResponse, WebSocketException):
LOGGER.debug("Working but unsupported config: %s", config)
return RESULT_NOT_SUPPORTED
except OSError as err:
LOGGER.debug("Failing config: %s, error: %s", config, err)
for method in SUPPORTED_METHODS:
self._bridge = SamsungTVBridge.get_bridge(method, self._host)
result = self._bridge.try_connect()
if result != RESULT_NOT_SUCCESSFUL:
return result
LOGGER.debug("No working config found")
return RESULT_NOT_SUCCESSFUL
async def async_step_import(self, user_input=None):
"""Handle configuration by yaml file."""
self._port = user_input.get(CONF_PORT)
return await self.async_step_user(user_input)
async def async_step_user(self, user_input=None):
@ -191,7 +168,6 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._manufacturer = user_input.get(CONF_MANUFACTURER)
self._model = user_input.get(CONF_MODEL)
self._name = user_input.get(CONF_NAME)
self._port = user_input.get(CONF_PORT)
self._title = self._model or self._name
await self.async_set_unique_id(self._ip)

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

@ -6,6 +6,18 @@ DOMAIN = "samsungtv"
DEFAULT_NAME = "Samsung TV"
VALUE_CONF_NAME = "HomeAssistant"
VALUE_CONF_ID = "ha.component.samsung"
CONF_DESCRIPTION = "description"
CONF_MANUFACTURER = "manufacturer"
CONF_MODEL = "model"
CONF_ON_ACTION = "turn_on_action"
RESULT_AUTH_MISSING = "auth_missing"
RESULT_SUCCESS = "success"
RESULT_NOT_SUCCESSFUL = "not_successful"
RESULT_NOT_SUPPORTED = "not_supported"
METHOD_LEGACY = "legacy"
METHOD_WEBSOCKET = "websocket"

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

@ -3,7 +3,8 @@
"name": "Samsung Smart TV",
"documentation": "https://www.home-assistant.io/integrations/samsungtv",
"requirements": [
"samsungctl[websocket]==0.7.1"
"samsungctl[websocket]==0.7.1",
"samsungtvws[websocket]==1.4.0"
],
"ssdp": [
{
@ -15,4 +16,4 @@
"@escoand"
],
"config_flow": true
}
}

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

@ -2,9 +2,7 @@
import asyncio
from datetime import timedelta
from samsungctl import Remote as SamsungRemote, exceptions as samsung_exceptions
import voluptuous as vol
from websocket import WebSocketException
from homeassistant.components.media_player import DEVICE_CLASS_TV, MediaPlayerDevice
from homeassistant.components.media_player.const import (
@ -27,6 +25,7 @@ from homeassistant.const import (
CONF_METHOD,
CONF_NAME,
CONF_PORT,
CONF_TOKEN,
STATE_OFF,
STATE_ON,
)
@ -34,6 +33,7 @@ import homeassistant.helpers.config_validation as cv
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, CONF_ON_ACTION, DOMAIN, LOGGER
KEY_PRESS_TIMEOUT = 1.2
@ -90,91 +90,40 @@ class SamsungTVDevice(MediaPlayerDevice):
# Assume that the TV is in Play mode
self._playing = True
self._state = None
self._remote = None
# Mark the end of a shutdown command (need to wait 15 seconds before
# sending the next command to avoid turning the TV back ON).
self._end_of_power_off = None
# Generate a configuration for the Samsung library
self._config = {
"name": "HomeAssistant",
"description": "HomeAssistant",
"id": "ha.component.samsung",
"method": config_entry.data[CONF_METHOD],
"port": config_entry.data.get(CONF_PORT),
"host": config_entry.data[CONF_HOST],
"timeout": 1,
}
# Initialize bridge
self._bridge = SamsungTVBridge.get_bridge(
config_entry.data[CONF_METHOD],
config_entry.data[CONF_HOST],
config_entry.data[CONF_PORT],
config_entry.data.get(CONF_TOKEN),
)
self._bridge.register_reauth_callback(self.access_denied)
def access_denied(self):
"""Access denied callbck."""
LOGGER.debug("Access denied in getting remote object")
self.hass.add_job(
self.hass.config_entries.flow.async_init(
DOMAIN, context={"source": "reauth"}, data=self._config_entry.data,
)
)
def update(self):
"""Update state of device."""
if self._power_off_in_progress():
self._state = STATE_OFF
else:
if self._remote is not None:
# Close the current remote connection
self._remote.close()
self._remote = None
try:
self.get_remote()
if self._remote:
self._state = STATE_ON
except (
samsung_exceptions.UnhandledResponse,
samsung_exceptions.AccessDenied,
):
# We got a response so it's working.
self._state = STATE_ON
except (OSError, WebSocketException):
# Different reasons, e.g. hostname not resolveable
self._state = STATE_OFF
def get_remote(self):
"""Create or return a remote control instance."""
if self._remote is None:
# We need to create a new instance to reconnect.
try:
self._remote = SamsungRemote(self._config.copy())
# This is only happening when the auth was switched to DENY
# A removed auth will lead to socket timeout because waiting for auth popup is just an open socket
except samsung_exceptions.AccessDenied:
self.hass.async_create_task(
self.hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "reauth"},
data=self._config_entry.data,
)
)
raise
return self._remote
self._state = STATE_ON if self._bridge.is_on() else STATE_OFF
def send_key(self, key):
"""Send a key to the tv and handles exceptions."""
if self._power_off_in_progress() and key not in ("KEY_POWER", "KEY_POWEROFF"):
if self._power_off_in_progress() and key != "KEY_POWEROFF":
LOGGER.info("TV is powering off, not sending command: %s", key)
return
try:
# recreate connection if connection was dead
retry_count = 1
for _ in range(retry_count + 1):
try:
self.get_remote().control(key)
break
except (
samsung_exceptions.ConnectionClosed,
BrokenPipeError,
WebSocketException,
):
# BrokenPipe can occur when the commands is sent to fast
# WebSocketException can occur when timed out
self._remote = None
except (samsung_exceptions.UnhandledResponse, samsung_exceptions.AccessDenied):
# We got a response so it's on.
LOGGER.debug("Failed sending command %s", key, exc_info=True)
except OSError:
# Different reasons, e.g. hostname not resolveable
pass
self._bridge.send_key(key)
def _power_off_in_progress(self):
return (
@ -233,16 +182,9 @@ class SamsungTVDevice(MediaPlayerDevice):
"""Turn off media player."""
self._end_of_power_off = dt_util.utcnow() + timedelta(seconds=15)
if self._config["method"] == "websocket":
self.send_key("KEY_POWER")
else:
self.send_key("KEY_POWEROFF")
self.send_key("KEY_POWEROFF")
# Force closing of remote session to provide instant UI feedback
try:
self.get_remote().close()
self._remote = None
except OSError:
LOGGER.debug("Could not establish connection.")
self._bridge.close_remote()
def volume_up(self):
"""Volume up the media player."""

3
requirements_all.txt
View File

@ -1815,6 +1815,9 @@ saltbox==0.1.3
# homeassistant.components.samsungtv
samsungctl[websocket]==0.7.1
# homeassistant.components.samsungtv
samsungtvws[websocket]==1.4.0
# homeassistant.components.satel_integra
satel_integra==0.3.4

3
requirements_test_all.txt
View File

@ -625,6 +625,9 @@ rxv==0.6.0
# homeassistant.components.samsungtv
samsungctl[websocket]==0.7.1
# homeassistant.components.samsungtv
samsungtvws[websocket]==1.4.0
# homeassistant.components.sense
sense_energy==0.7.0

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

@ -4,6 +4,7 @@ from unittest.mock import call, patch
from asynctest import mock
import pytest
from samsungctl.exceptions import AccessDenied, UnhandledResponse
from samsungtvws.exceptions import ConnectionFailure
from websocket import WebSocketProtocolException
from homeassistant.components.samsungtv.const import (
@ -36,15 +37,6 @@ MOCK_SSDP_DATA_NOPREFIX = {
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",
@ -59,7 +51,9 @@ AUTODETECT_LEGACY = {
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch("samsungctl.Remote") as remote_class, patch(
with patch(
"homeassistant.components.samsungtv.bridge.Remote"
) as remote_class, patch(
"homeassistant.components.samsungtv.config_flow.socket"
) as socket_class:
remote = mock.Mock()
@ -71,9 +65,25 @@ def remote_fixture():
yield remote
async def test_user(hass, remote):
"""Test starting a flow by user."""
@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"
) as socket_class:
remotews = mock.Mock()
remotews.__enter__ = mock.Mock()
remotews.__exit__ = mock.Mock()
remotews_class.return_value = remotews
socket = mock.Mock()
socket_class.return_value = socket
yield remotews
async def test_user_legacy(hass, remote):
"""Test starting a flow by user."""
# show form
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}
@ -85,23 +95,51 @@ async def test_user(hass, remote):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=MOCK_USER_DATA
)
# legacy tv entry created
assert result["type"] == "create_entry"
assert result["title"] == "fake_name"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake_name"
assert result["data"][CONF_METHOD] == "legacy"
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):
async def test_user_websocket(hass, remotews):
"""Test starting a flow by user."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom")
):
# 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
)
# legacy tv entry created
assert result["type"] == "create_entry"
assert result["title"] == "fake_name"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake_name"
assert result["data"][CONF_METHOD] == "websocket"
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_legacy_missing_auth(hass):
"""Test starting a flow by user with authentication."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=AccessDenied("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
# missing authentication
# legacy device missing authentication
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
@ -109,14 +147,31 @@ async def test_user_missing_auth(hass):
assert result["reason"] == "auth_missing"
async def test_user_not_supported(hass):
async def test_user_legacy_not_supported(hass):
"""Test starting a flow by user for not supported device."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=UnhandledResponse("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
# legacy 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"
# device not supported
async def test_user_websocket_not_supported(hass):
"""Test starting a flow by user for not supported device."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS",
side_effect=WebSocketProtocolException("Boom"),
), patch(
"homeassistant.components.samsungtv.config_flow.socket"
):
# websocket device not supported
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
@ -127,11 +182,30 @@ async def test_user_not_supported(hass):
async def test_user_not_successful(hass):
"""Test starting a flow by user but no connection found."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS",
side_effect=OSError("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
), patch(
"homeassistant.components.samsungtv.config_flow.socket"
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "not_successful"
# device not connectable
async def test_user_not_successful_2(hass):
"""Test starting a flow by user but no connection found."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS",
side_effect=ConnectionFailure("Boom"),
), patch(
"homeassistant.components.samsungtv.config_flow.socket"
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
@ -202,10 +276,10 @@ async def test_ssdp_noprefix(hass, remote):
assert result["data"][CONF_ID] == "fake2_uuid"
async def test_ssdp_missing_auth(hass):
async def test_ssdp_legacy_missing_auth(hass):
"""Test starting a flow from discovery with authentication."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=AccessDenied("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
@ -224,10 +298,10 @@ async def test_ssdp_missing_auth(hass):
assert result["reason"] == "auth_missing"
async def test_ssdp_not_supported(hass):
async def test_ssdp_legacy_not_supported(hass):
"""Test starting a flow from discovery for not supported device."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=UnhandledResponse("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
@ -246,13 +320,16 @@ async def test_ssdp_not_supported(hass):
assert result["reason"] == "not_supported"
async def test_ssdp_not_supported_2(hass):
async def test_ssdp_websocket_not_supported(hass):
"""Test starting a flow from discovery for not supported device."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS",
side_effect=WebSocketProtocolException("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
), 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
@ -271,9 +348,39 @@ async def test_ssdp_not_supported_2(hass):
async def test_ssdp_not_successful(hass):
"""Test starting a flow from discovery but no device found."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS",
side_effect=OSError("Boom"),
), patch("homeassistant.components.samsungtv.config_flow.socket"):
), 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_successful"
async def test_ssdp_not_successful_2(hass):
"""Test starting a flow from discovery but no device found."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS",
side_effect=ConnectionFailure("Boom"),
), patch(
"homeassistant.components.samsungtv.config_flow.socket"
):
# confirm to add the entry
result = await hass.config_entries.flow.async_init(
@ -334,22 +441,32 @@ async def test_ssdp_already_configured(hass, remote):
assert entry.data[CONF_ID] == "fake_uuid"
async def test_autodetect_websocket(hass, remote):
async def test_autodetect_websocket(hass, remote, remotews):
"""Test for send key with autodetection of protocol."""
with patch("homeassistant.components.samsungtv.config_flow.Remote") as remote:
with patch(
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
), patch("homeassistant.components.samsungtv.bridge.SamsungTVWS") as remotews:
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)]
assert remotews.call_count == 1
assert remotews.call_args_list == [
call(
host="fake_host",
name="HomeAssistant",
port=8001,
timeout=31,
token=None,
)
]
async def test_autodetect_auth_missing(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[AccessDenied("Boom")],
) as remote:
result = await hass.config_entries.flow.async_init(
@ -358,13 +475,13 @@ async def test_autodetect_auth_missing(hass, remote):
assert result["type"] == "abort"
assert result["reason"] == "auth_missing"
assert remote.call_count == 1
assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)]
assert remote.call_args_list == [call(AUTODETECT_LEGACY)]
async def test_autodetect_not_supported(hass, remote):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[UnhandledResponse("Boom")],
) as remote:
result = await hass.config_entries.flow.async_init(
@ -373,40 +490,52 @@ async def test_autodetect_not_supported(hass, remote):
assert result["type"] == "abort"
assert result["reason"] == "not_supported"
assert remote.call_count == 1
assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)]
assert remote.call_args_list == [call(AUTODETECT_LEGACY)]
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:
with patch("homeassistant.components.samsungtv.bridge.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] == "legacy"
assert remote.call_count == 2
assert remote.call_args_list == [
call(AUTODETECT_WEBSOCKET),
call(AUTODETECT_LEGACY),
]
assert remote.call_count == 1
assert remote.call_args_list == [call(AUTODETECT_LEGACY)]
async def test_autodetect_none(hass, remote):
async def test_autodetect_none(hass, remote, remotews):
"""Test for send key with autodetection of protocol."""
with patch(
"homeassistant.components.samsungtv.config_flow.Remote",
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"),
) as remote, patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS",
side_effect=OSError("Boom"),
) as remote:
) as remotews:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA
)
assert result["type"] == "abort"
assert result["reason"] == "not_successful"
assert remote.call_count == 2
assert remote.call_count == 1
assert remote.call_args_list == [
call(AUTODETECT_WEBSOCKET),
call(AUTODETECT_LEGACY),
]
assert remotews.call_count == 2
assert remotews.call_args_list == [
call(
host="fake_host",
name="HomeAssistant",
port=8001,
timeout=31,
token=None,
),
call(
host="fake_host",
name="HomeAssistant",
port=8002,
timeout=31,
token=None,
),
]

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

@ -1,6 +1,6 @@
"""Tests for the Samsung TV Integration."""
from unittest.mock import call, patch
from asynctest import mock
from asynctest.mock import call, patch
import pytest
from homeassistant.components.media_player.const import DOMAIN, SUPPORT_TURN_ON
@ -14,7 +14,6 @@ from homeassistant.const import (
ATTR_SUPPORTED_FEATURES,
CONF_HOST,
CONF_NAME,
CONF_PORT,
SERVICE_VOLUME_UP,
)
from homeassistant.setup import async_setup_component
@ -25,7 +24,6 @@ MOCK_CONFIG = {
{
CONF_HOST: "fake_host",
CONF_NAME: "fake_name",
CONF_PORT: 1234,
CONF_ON_ACTION: [{"delay": "00:00:01"}],
}
]
@ -34,8 +32,7 @@ REMOTE_CALL = {
"name": "HomeAssistant",
"description": "HomeAssistant",
"id": "ha.component.samsung",
"method": "websocket",
"port": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_PORT],
"method": "legacy",
"host": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_HOST],
"timeout": 1,
}
@ -44,11 +41,17 @@ REMOTE_CALL = {
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch("homeassistant.components.samsungtv.socket") as socket1, patch(
with patch(
"homeassistant.components.samsungtv.bridge.Remote"
) as remote_class, patch(
"homeassistant.components.samsungtv.config_flow.socket"
) as socket2, patch("homeassistant.components.samsungtv.config_flow.Remote"), patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote:
) as socket1, patch(
"homeassistant.components.samsungtv.socket"
) as socket2:
remote = mock.Mock()
remote.__enter__ = mock.Mock()
remote.__exit__ = mock.Mock()
remote_class.return_value = remote
socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS"
socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS"
yield remote
@ -56,22 +59,24 @@ def remote_fixture():
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)
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
assert state
assert state.name == "fake_name"
assert (
state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON
)
# 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)
# test host and port
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
assert remote.call_args == call(REMOTE_CALL)
async def test_setup_duplicate_config(hass, remote, caplog):

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

@ -7,6 +7,7 @@ from asynctest import mock
from asynctest.mock import call, patch
import pytest
from samsungctl import exceptions
from samsungtvws.exceptions import ConnectionFailure
from websocket import WebSocketException
from homeassistant.components.media_player import DEVICE_CLASS_TV
@ -54,6 +55,17 @@ from tests.common import async_fire_time_changed
ENTITY_ID = f"{DOMAIN}.fake"
MOCK_CONFIG = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake",
CONF_NAME: "fake",
CONF_PORT: 55000,
CONF_ON_ACTION: [{"delay": "00:00:01"}],
}
]
}
MOCK_CONFIGWS = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake",
@ -75,14 +87,35 @@ MOCK_CONFIG_NOTURNON = {
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch("homeassistant.components.samsungtv.config_flow.Remote"), patch(
with patch(
"homeassistant.components.samsungtv.bridge.Remote"
) as remote_class, patch(
"homeassistant.components.samsungtv.config_flow.socket"
) as socket1, patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote_class, patch(
"homeassistant.components.samsungtv.socket"
) as socket2:
remote = mock.Mock()
remote.__enter__ = mock.Mock()
remote.__exit__ = mock.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.Mock()
remote.__enter__ = mock.Mock()
remote.__exit__ = mock.Mock()
remote_class.return_value = remote
socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS"
socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS"
@ -140,7 +173,7 @@ async def test_update_off(hass, remote, mock_now):
await setup_samsungtv(hass, MOCK_CONFIG)
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote",
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[OSError("Boom"), mock.DEFAULT],
):
@ -154,14 +187,13 @@ async def test_update_off(hass, remote, mock_now):
async def test_update_access_denied(hass, remote, mock_now):
"""Testing update tv unhandled response exception."""
"""Testing update tv access denied exception."""
await setup_samsungtv(hass, MOCK_CONFIG)
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote",
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=exceptions.AccessDenied("Boom"),
):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
@ -174,12 +206,36 @@ async def test_update_access_denied(hass, remote, mock_now):
]
async def test_update_connection_failure(hass, remotews, mock_now):
"""Testing update tv connection failure exception."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[OSError("Boom"), mock.DEFAULT],
):
await setup_samsungtv(hass, MOCK_CONFIGWS)
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS",
side_effect=ConnectionFailure("Boom"),
):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert [
flow
for flow in hass.config_entries.flow.async_progress()
if flow["context"]["source"] == "reauth"
]
async def test_update_unhandled_response(hass, remote, mock_now):
"""Testing update tv unhandled response exception."""
await setup_samsungtv(hass, MOCK_CONFIG)
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote",
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[exceptions.UnhandledResponse("Boom"), mock.DEFAULT],
):
@ -334,36 +390,30 @@ async def test_device_class(hass, remote):
assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_TV
async def test_turn_off_websocket(hass, remote):
async def test_turn_off_websocket(hass, remotews):
"""Test for turn_off."""
await setup_samsungtv(hass, MOCK_CONFIG)
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[OSError("Boom"), mock.DEFAULT],
):
await setup_samsungtv(hass, MOCK_CONFIGWS)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# key called
assert remotews.send_key.call_count == 1
assert remotews.send_key.call_args_list == [call("KEY_POWER")]
async def test_turn_off_legacy(hass, remote):
"""Test for turn_off."""
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
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_POWER")]
async def test_turn_off_legacy(hass):
"""Test for turn_off."""
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")]
assert remote.control.call_args_list == [call("KEY_POWEROFF")]
async def test_turn_off_os_error(hass, remote, caplog):
@ -374,7 +424,7 @@ async def test_turn_off_os_error(hass, remote, caplog):
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
assert "Could not establish connection." in caplog.text
assert "Could not establish connection" in caplog.text
async def test_volume_up(hass, remote):
@ -526,11 +576,12 @@ async def test_play_media(hass, remote):
async def test_play_media_invalid_type(hass, remote):
"""Test for play_media with invalid media type."""
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote, patch("homeassistant.components.samsungtv.config_flow.socket"):
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote, patch(
"homeassistant.components.samsungtv.config_flow.socket"
):
url = "https://example.com"
await setup_samsungtv(hass, MOCK_CONFIG)
remote.reset_mock()
assert await hass.services.async_call(
DOMAIN,
SERVICE_PLAY_MEDIA,
@ -549,11 +600,12 @@ async def test_play_media_invalid_type(hass, remote):
async def test_play_media_channel_as_string(hass, remote):
"""Test for play_media with invalid channel as string."""
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote, patch("homeassistant.components.samsungtv.config_flow.socket"):
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote, patch(
"homeassistant.components.samsungtv.config_flow.socket"
):
url = "https://example.com"
await setup_samsungtv(hass, MOCK_CONFIG)
remote.reset_mock()
assert await hass.services.async_call(
DOMAIN,
SERVICE_PLAY_MEDIA,
@ -572,10 +624,11 @@ async def test_play_media_channel_as_string(hass, remote):
async def test_play_media_channel_as_non_positive(hass, remote):
"""Test for play_media with invalid channel as non positive integer."""
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote, patch("homeassistant.components.samsungtv.config_flow.socket"):
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote, patch(
"homeassistant.components.samsungtv.config_flow.socket"
):
await setup_samsungtv(hass, MOCK_CONFIG)
remote.reset_mock()
assert await hass.services.async_call(
DOMAIN,
SERVICE_PLAY_MEDIA,
@ -610,10 +663,11 @@ async def test_select_source(hass, remote):
async def test_select_source_invalid_source(hass, remote):
"""Test for select_source with invalid source."""
with patch(
"homeassistant.components.samsungtv.media_player.SamsungRemote"
) as remote, patch("homeassistant.components.samsungtv.config_flow.socket"):
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote, patch(
"homeassistant.components.samsungtv.config_flow.socket"
):
await setup_samsungtv(hass, MOCK_CONFIG)
remote.reset_mock()
assert await hass.services.async_call(
DOMAIN,
SERVICE_SELECT_SOURCE,