mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 00:37:53 +00:00
Add Samsung TV automatic protocol detection (#27492)
* added automatic protocol detection * fix logger tests * fix async tests * add missin const.py * fix log formatting * wait for first update call * migrate first tests * migrated all test functions * started to use state machine * updated all tests to use async_setup_component * slove hints * update tests * get state at correct position * remove impossible tests * fix autodetect tests * use caplog fixture * add test for duplicate * catch concrete exceptions * don't mock samsungctl exceptions * add test for discovery * get state when possible * add test for autodetect without connection
This commit is contained in:
escoand
Martin Hjelmare
parent
95295791bd
commit
9661efc312
@ -247,6 +247,7 @@ homeassistant/components/rfxtrx/* @danielhiversen
|
||||
homeassistant/components/rmvtransport/* @cgtobi
|
||||
homeassistant/components/roomba/* @pschmitt
|
||||
homeassistant/components/saj/* @fredericvl
|
||||
homeassistant/components/samsungtv/* @escoand
|
||||
homeassistant/components/scene/* @home-assistant/core
|
||||
homeassistant/components/scrape/* @fabaff
|
||||
homeassistant/components/script/* @home-assistant/core
|
||||
|
||||
@ -1 +1 @@
|
||||
"""The samsungtv component."""
|
||||
"""The Samsung TV integration."""
|
||||
|
||||
5
homeassistant/components/samsungtv/const.py
Normal file
5
homeassistant/components/samsungtv/const.py
Normal file
@ -0,0 +1,5 @@
|
||||
"""Constants for the Samsung TV integration."""
|
||||
import logging
|
||||
|
||||
LOGGER = logging.getLogger(__package__)
|
||||
DOMAIN = "samsungtv"
|
||||
@ -1,11 +1,11 @@
|
||||
{
|
||||
"domain": "samsungtv",
|
||||
"name": "Samsungtv",
|
||||
"name": "Samsung TV",
|
||||
"documentation": "https://www.home-assistant.io/integrations/samsungtv",
|
||||
"requirements": [
|
||||
"samsungctl[websocket]==0.7.1",
|
||||
"wakeonlan==1.1.6"
|
||||
],
|
||||
"dependencies": [],
|
||||
"codeowners": []
|
||||
"codeowners": ["@escoand"]
|
||||
}
|
||||
|
||||
@ -1,7 +1,6 @@
|
||||
"""Support for interface with an Samsung TV."""
|
||||
import asyncio
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
import socket
|
||||
|
||||
import voluptuous as vol
|
||||
@ -36,14 +35,14 @@ from homeassistant.const import (
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
from .const import LOGGER
|
||||
|
||||
DEFAULT_NAME = "Samsung TV Remote"
|
||||
DEFAULT_PORT = 55000
|
||||
DEFAULT_TIMEOUT = 1
|
||||
|
||||
KEY_PRESS_TIMEOUT = 1.2
|
||||
KNOWN_DEVICES_KEY = "samsungtv_known_devices"
|
||||
METHODS = ("websocket", "legacy")
|
||||
SOURCES = {"TV": "KEY_TV", "HDMI": "KEY_HDMI"}
|
||||
|
||||
SUPPORT_SAMSUNGTV = (
|
||||
@ -62,7 +61,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_HOST): cv.string,
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
|
||||
vol.Optional(CONF_PORT): cv.port,
|
||||
vol.Optional(CONF_MAC): cv.string,
|
||||
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
|
||||
}
|
||||
@ -89,15 +88,14 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
model = discovery_info.get("model_name")
|
||||
host = discovery_info.get("host")
|
||||
name = f"{tv_name} ({model})"
|
||||
port = DEFAULT_PORT
|
||||
if name.startswith("[TV]"):
|
||||
name = name[4:]
|
||||
port = None
|
||||
timeout = DEFAULT_TIMEOUT
|
||||
mac = None
|
||||
udn = discovery_info.get("udn")
|
||||
if udn and udn.startswith("uuid:"):
|
||||
uuid = udn[len("uuid:") :]
|
||||
else:
|
||||
_LOGGER.warning("Cannot determine device")
|
||||
return
|
||||
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.
|
||||
@ -105,9 +103,9 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
if ip_addr not in known_devices:
|
||||
known_devices.add(ip_addr)
|
||||
add_entities([SamsungTVDevice(host, port, name, timeout, mac, uuid)])
|
||||
_LOGGER.info("Samsung TV %s:%d added as '%s'", host, port, name)
|
||||
LOGGER.info("Samsung TV %s added as '%s'", host, name)
|
||||
else:
|
||||
_LOGGER.info("Ignoring duplicate Samsung TV %s:%d", host, port)
|
||||
LOGGER.info("Ignoring duplicate Samsung TV %s", host)
|
||||
|
||||
|
||||
class SamsungTVDevice(MediaPlayerDevice):
|
||||
@ -140,14 +138,16 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
"name": "HomeAssistant",
|
||||
"description": name,
|
||||
"id": "ha.component.samsung",
|
||||
"method": None,
|
||||
"port": port,
|
||||
"host": host,
|
||||
"timeout": timeout,
|
||||
}
|
||||
|
||||
# Select method by port number, mainly for fallback
|
||||
if self._config["port"] in (8001, 8002):
|
||||
self._config["method"] = "websocket"
|
||||
else:
|
||||
elif self._config["port"] == 55000:
|
||||
self._config["method"] = "legacy"
|
||||
|
||||
def update(self):
|
||||
@ -156,16 +156,47 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
|
||||
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 = self._remote_class(self._config.copy())
|
||||
self._state = STATE_ON
|
||||
LOGGER.debug("Found working config: %s", self._config)
|
||||
break
|
||||
except (
|
||||
self._exceptions_class.UnhandledResponse,
|
||||
self._exceptions_class.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 = self._remote_class(self._config)
|
||||
self._remote = self._remote_class(self._config.copy())
|
||||
|
||||
return self._remote
|
||||
|
||||
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"):
|
||||
_LOGGER.info("TV is powering off, not sending command: %s", key)
|
||||
LOGGER.info("TV is powering off, not sending command: %s", key)
|
||||
return
|
||||
try:
|
||||
# recreate connection if connection was dead
|
||||
@ -178,6 +209,9 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
# BrokenPipe can occur when the commands is sent to fast
|
||||
self._remote = None
|
||||
self._state = STATE_ON
|
||||
except AttributeError:
|
||||
# Auto-detect could not find working config yet
|
||||
pass
|
||||
except (
|
||||
self._exceptions_class.UnhandledResponse,
|
||||
self._exceptions_class.AccessDenied,
|
||||
@ -185,7 +219,7 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
# We got a response so it's on.
|
||||
self._state = STATE_ON
|
||||
self._remote = None
|
||||
_LOGGER.debug("Failed sending command %s", key, exc_info=True)
|
||||
LOGGER.debug("Failed sending command %s", key, exc_info=True)
|
||||
return
|
||||
except OSError:
|
||||
self._state = STATE_OFF
|
||||
@ -249,7 +283,7 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
self.get_remote().close()
|
||||
self._remote = None
|
||||
except OSError:
|
||||
_LOGGER.debug("Could not establish connection.")
|
||||
LOGGER.debug("Could not establish connection.")
|
||||
|
||||
def volume_up(self):
|
||||
"""Volume up the media player."""
|
||||
@ -291,14 +325,14 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
async def async_play_media(self, media_type, media_id, **kwargs):
|
||||
"""Support changing a channel."""
|
||||
if media_type != MEDIA_TYPE_CHANNEL:
|
||||
_LOGGER.error("Unsupported media type")
|
||||
LOGGER.error("Unsupported media type")
|
||||
return
|
||||
|
||||
# media_id should only be a channel number
|
||||
try:
|
||||
cv.positive_int(media_id)
|
||||
except vol.Invalid:
|
||||
_LOGGER.error("Media ID must be positive integer")
|
||||
LOGGER.error("Media ID must be positive integer")
|
||||
return
|
||||
|
||||
for digit in media_id:
|
||||
@ -316,7 +350,7 @@ class SamsungTVDevice(MediaPlayerDevice):
|
||||
async def async_select_source(self, source):
|
||||
"""Select input source."""
|
||||
if source not in SOURCES:
|
||||
_LOGGER.error("Unsupported source")
|
||||
LOGGER.error("Unsupported source")
|
||||
return
|
||||
|
||||
await self.hass.async_add_job(self.send_key, SOURCES[source])
|
||||
|
||||
@ -1,309 +1,573 @@
|
||||
"""Tests for samsungtv Components."""
|
||||
import asyncio
|
||||
import unittest
|
||||
from unittest.mock import call, patch, MagicMock
|
||||
|
||||
from asynctest import mock
|
||||
|
||||
from datetime import timedelta
|
||||
import pytest
|
||||
from samsungctl import exceptions
|
||||
from tests.common import MockDependency, async_fire_time_changed
|
||||
from unittest.mock import call, patch
|
||||
|
||||
import tests.common
|
||||
from homeassistant.components.media_player import DEVICE_CLASS_TV
|
||||
from homeassistant.components.media_player.const import (
|
||||
ATTR_INPUT_SOURCE,
|
||||
ATTR_MEDIA_CONTENT_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE,
|
||||
ATTR_MEDIA_VOLUME_MUTED,
|
||||
DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
SERVICE_SELECT_SOURCE,
|
||||
SUPPORT_TURN_ON,
|
||||
MEDIA_TYPE_CHANNEL,
|
||||
MEDIA_TYPE_URL,
|
||||
)
|
||||
from homeassistant.components.samsungtv.const import DOMAIN as SAMSUNGTV_DOMAIN
|
||||
from homeassistant.components.samsungtv.media_player import (
|
||||
setup_platform,
|
||||
CONF_TIMEOUT,
|
||||
SamsungTVDevice,
|
||||
SUPPORT_SAMSUNGTV,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_DEVICE_CLASS,
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_FRIENDLY_NAME,
|
||||
ATTR_SUPPORTED_FEATURES,
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
CONF_PORT,
|
||||
STATE_ON,
|
||||
CONF_MAC,
|
||||
CONF_NAME,
|
||||
CONF_PLATFORM,
|
||||
CONF_PORT,
|
||||
SERVICE_MEDIA_NEXT_TRACK,
|
||||
SERVICE_MEDIA_PAUSE,
|
||||
SERVICE_MEDIA_PLAY,
|
||||
SERVICE_MEDIA_PREVIOUS_TRACK,
|
||||
SERVICE_TURN_OFF,
|
||||
SERVICE_TURN_ON,
|
||||
SERVICE_VOLUME_DOWN,
|
||||
SERVICE_VOLUME_MUTE,
|
||||
SERVICE_VOLUME_UP,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
from tests.common import MockDependency
|
||||
from homeassistant.util import dt as dt_util
|
||||
from datetime import timedelta
|
||||
from homeassistant.helpers.discovery import async_load_platform
|
||||
from homeassistant.setup import async_setup_component
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
WORKING_CONFIG = {
|
||||
CONF_HOST: "fake",
|
||||
CONF_NAME: "fake",
|
||||
CONF_PORT: 8001,
|
||||
CONF_TIMEOUT: 10,
|
||||
CONF_MAC: "fake",
|
||||
"uuid": None,
|
||||
|
||||
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: "fake",
|
||||
}
|
||||
}
|
||||
|
||||
DISCOVERY_INFO = {"name": "fake", "model_name": "fake", "host": "fake"}
|
||||
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,
|
||||
}
|
||||
|
||||
|
||||
class AccessDenied(Exception):
|
||||
"""Dummy Exception."""
|
||||
@pytest.fixture(name="remote")
|
||||
def remote_fixture():
|
||||
"""Patch the samsungctl Remote."""
|
||||
with patch("samsungctl.Remote") as remote_class, patch(
|
||||
"homeassistant.components.samsungtv.media_player.socket"
|
||||
) as socket_class:
|
||||
remote = mock.Mock()
|
||||
remote_class.return_value = remote
|
||||
socket = mock.Mock()
|
||||
socket_class.return_value = socket
|
||||
yield remote
|
||||
|
||||
|
||||
class ConnectionClosed(Exception):
|
||||
"""Dummy Exception."""
|
||||
|
||||
|
||||
class UnhandledResponse(Exception):
|
||||
"""Dummy Exception."""
|
||||
|
||||
|
||||
class TestSamsungTv(unittest.TestCase):
|
||||
"""Testing Samsungtv component."""
|
||||
|
||||
@MockDependency("samsungctl")
|
||||
@MockDependency("wakeonlan")
|
||||
def setUp(self, samsung_mock, wol_mock):
|
||||
"""Set up test environment."""
|
||||
self.hass = tests.common.get_test_home_assistant()
|
||||
self.hass.start()
|
||||
self.hass.block_till_done()
|
||||
self.device = SamsungTVDevice(**WORKING_CONFIG)
|
||||
self.device._exceptions_class = mock.Mock()
|
||||
self.device._exceptions_class.UnhandledResponse = UnhandledResponse
|
||||
self.device._exceptions_class.AccessDenied = AccessDenied
|
||||
self.device._exceptions_class.ConnectionClosed = ConnectionClosed
|
||||
|
||||
def tearDown(self):
|
||||
"""Tear down test data."""
|
||||
self.hass.stop()
|
||||
|
||||
@MockDependency("samsungctl")
|
||||
@MockDependency("wakeonlan")
|
||||
def test_setup(self, samsung_mock, wol_mock):
|
||||
"""Testing setup of platform."""
|
||||
with mock.patch("homeassistant.components.samsungtv.media_player.socket"):
|
||||
add_entities = mock.Mock()
|
||||
setup_platform(self.hass, WORKING_CONFIG, add_entities)
|
||||
|
||||
@MockDependency("samsungctl")
|
||||
@MockDependency("wakeonlan")
|
||||
def test_setup_discovery(self, samsung_mock, wol_mock):
|
||||
"""Testing setup of platform with discovery."""
|
||||
with mock.patch("homeassistant.components.samsungtv.media_player.socket"):
|
||||
add_entities = mock.Mock()
|
||||
setup_platform(self.hass, {}, add_entities, discovery_info=DISCOVERY_INFO)
|
||||
|
||||
@MockDependency("samsungctl")
|
||||
@MockDependency("wakeonlan")
|
||||
@mock.patch("homeassistant.components.samsungtv.media_player._LOGGER.warning")
|
||||
def test_setup_none(self, samsung_mock, wol_mock, mocked_warn):
|
||||
"""Testing setup of platform with no data."""
|
||||
with mock.patch("homeassistant.components.samsungtv.media_player.socket"):
|
||||
add_entities = mock.Mock()
|
||||
setup_platform(self.hass, {}, add_entities, discovery_info=None)
|
||||
mocked_warn.assert_called_once_with("Cannot determine device")
|
||||
add_entities.assert_not_called()
|
||||
|
||||
def test_update_on(self):
|
||||
"""Testing update tv on."""
|
||||
self.device.update()
|
||||
self.assertEqual(STATE_ON, self.device._state)
|
||||
|
||||
def test_update_off(self):
|
||||
"""Testing update tv off."""
|
||||
_remote = mock.Mock()
|
||||
_remote.control = mock.Mock(side_effect=OSError("Boom"))
|
||||
self.device.get_remote = mock.Mock(return_value=_remote)
|
||||
self.device.update()
|
||||
assert STATE_OFF == self.device._state
|
||||
|
||||
def test_send_key(self):
|
||||
"""Test for send key."""
|
||||
self.device.send_key("KEY_POWER")
|
||||
self.assertEqual(STATE_ON, self.device._state)
|
||||
|
||||
def test_send_key_broken_pipe(self):
|
||||
"""Testing broken pipe Exception."""
|
||||
_remote = mock.Mock()
|
||||
_remote.control = mock.Mock(side_effect=BrokenPipeError("Boom"))
|
||||
self.device.get_remote = mock.Mock(return_value=_remote)
|
||||
self.device.send_key("HELLO")
|
||||
self.assertIsNone(self.device._remote)
|
||||
self.assertEqual(STATE_ON, self.device._state)
|
||||
|
||||
def test_send_key_connection_closed_retry_succeed(self):
|
||||
"""Test retry on connection closed."""
|
||||
_remote = mock.Mock()
|
||||
_remote.control = mock.Mock(
|
||||
side_effect=[
|
||||
self.device._exceptions_class.ConnectionClosed("Boom"),
|
||||
mock.DEFAULT,
|
||||
]
|
||||
)
|
||||
self.device.get_remote = mock.Mock(return_value=_remote)
|
||||
command = "HELLO"
|
||||
self.device.send_key(command)
|
||||
self.assertEqual(STATE_ON, self.device._state)
|
||||
# verify that _remote.control() get called twice because of retry logic
|
||||
expected = [mock.call(command), mock.call(command)]
|
||||
assert expected == _remote.control.call_args_list
|
||||
|
||||
def test_send_key_unhandled_response(self):
|
||||
"""Testing unhandled response exception."""
|
||||
_remote = mock.Mock()
|
||||
_remote.control = mock.Mock(
|
||||
side_effect=self.device._exceptions_class.UnhandledResponse("Boom")
|
||||
)
|
||||
self.device.get_remote = mock.Mock(return_value=_remote)
|
||||
self.device.send_key("HELLO")
|
||||
self.assertIsNone(self.device._remote)
|
||||
self.assertEqual(STATE_ON, self.device._state)
|
||||
|
||||
def test_send_key_os_error(self):
|
||||
"""Testing broken pipe Exception."""
|
||||
_remote = mock.Mock()
|
||||
_remote.control = mock.Mock(side_effect=OSError("Boom"))
|
||||
self.device.get_remote = mock.Mock(return_value=_remote)
|
||||
self.device.send_key("HELLO")
|
||||
assert self.device._remote is None
|
||||
assert STATE_OFF == self.device._state
|
||||
|
||||
def test_power_off_in_progress(self):
|
||||
"""Test for power_off_in_progress."""
|
||||
assert not self.device._power_off_in_progress()
|
||||
self.device._end_of_power_off = dt_util.utcnow() + timedelta(seconds=15)
|
||||
assert self.device._power_off_in_progress()
|
||||
|
||||
def test_name(self):
|
||||
"""Test for name property."""
|
||||
assert "fake" == self.device.name
|
||||
|
||||
def test_state(self):
|
||||
"""Test for state property."""
|
||||
self.device._state = STATE_ON
|
||||
self.assertEqual(STATE_ON, self.device.state)
|
||||
self.device._state = STATE_OFF
|
||||
assert STATE_OFF == self.device.state
|
||||
|
||||
def test_is_volume_muted(self):
|
||||
"""Test for is_volume_muted property."""
|
||||
self.device._muted = False
|
||||
assert not self.device.is_volume_muted
|
||||
self.device._muted = True
|
||||
assert self.device.is_volume_muted
|
||||
|
||||
def test_supported_features(self):
|
||||
"""Test for supported_features property."""
|
||||
self.device._mac = None
|
||||
assert SUPPORT_SAMSUNGTV == self.device.supported_features
|
||||
self.device._mac = "fake"
|
||||
assert SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON == self.device.supported_features
|
||||
|
||||
def test_device_class(self):
|
||||
"""Test for device_class property."""
|
||||
assert DEVICE_CLASS_TV == self.device.device_class
|
||||
|
||||
def test_turn_off(self):
|
||||
"""Test for turn_off."""
|
||||
self.device.send_key = mock.Mock()
|
||||
_remote = mock.Mock()
|
||||
_remote.close = mock.Mock()
|
||||
self.get_remote = mock.Mock(return_value=_remote)
|
||||
self.device._end_of_power_off = None
|
||||
self.device.turn_off()
|
||||
assert self.device._end_of_power_off is not None
|
||||
self.device.send_key.assert_called_once_with("KEY_POWER")
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device._config["method"] = "legacy"
|
||||
self.device.turn_off()
|
||||
self.device.send_key.assert_called_once_with("KEY_POWEROFF")
|
||||
|
||||
@mock.patch("homeassistant.components.samsungtv.media_player._LOGGER.debug")
|
||||
def test_turn_off_os_error(self, mocked_debug):
|
||||
"""Test for turn_off with OSError."""
|
||||
_remote = mock.Mock()
|
||||
_remote.close = mock.Mock(side_effect=OSError("BOOM"))
|
||||
self.device.get_remote = mock.Mock(return_value=_remote)
|
||||
self.device.turn_off()
|
||||
mocked_debug.assert_called_once_with("Could not establish connection.")
|
||||
|
||||
def test_volume_up(self):
|
||||
"""Test for volume_up."""
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device.volume_up()
|
||||
self.device.send_key.assert_called_once_with("KEY_VOLUP")
|
||||
|
||||
def test_volume_down(self):
|
||||
"""Test for volume_down."""
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device.volume_down()
|
||||
self.device.send_key.assert_called_once_with("KEY_VOLDOWN")
|
||||
|
||||
def test_mute_volume(self):
|
||||
"""Test for mute_volume."""
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device.mute_volume(True)
|
||||
self.device.send_key.assert_called_once_with("KEY_MUTE")
|
||||
|
||||
def test_media_play_pause(self):
|
||||
"""Test for media_next_track."""
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device._playing = False
|
||||
self.device.media_play_pause()
|
||||
self.device.send_key.assert_called_once_with("KEY_PLAY")
|
||||
assert self.device._playing
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device.media_play_pause()
|
||||
self.device.send_key.assert_called_once_with("KEY_PAUSE")
|
||||
assert not self.device._playing
|
||||
|
||||
def test_media_play(self):
|
||||
"""Test for media_play."""
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device._playing = False
|
||||
self.device.media_play()
|
||||
self.device.send_key.assert_called_once_with("KEY_PLAY")
|
||||
assert self.device._playing
|
||||
|
||||
def test_media_pause(self):
|
||||
"""Test for media_pause."""
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device._playing = True
|
||||
self.device.media_pause()
|
||||
self.device.send_key.assert_called_once_with("KEY_PAUSE")
|
||||
assert not self.device._playing
|
||||
|
||||
def test_media_next_track(self):
|
||||
"""Test for media_next_track."""
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device.media_next_track()
|
||||
self.device.send_key.assert_called_once_with("KEY_FF")
|
||||
|
||||
def test_media_previous_track(self):
|
||||
"""Test for media_previous_track."""
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device.media_previous_track()
|
||||
self.device.send_key.assert_called_once_with("KEY_REWIND")
|
||||
|
||||
def test_turn_on(self):
|
||||
"""Test turn on."""
|
||||
self.device.send_key = mock.Mock()
|
||||
self.device._mac = None
|
||||
self.device.turn_on()
|
||||
self.device.send_key.assert_called_once_with("KEY_POWERON")
|
||||
self.device._wol.send_magic_packet = mock.Mock()
|
||||
self.device._mac = "fake"
|
||||
self.device.turn_on()
|
||||
self.device._wol.send_magic_packet.assert_called_once_with("fake")
|
||||
@pytest.fixture(name="wakeonlan")
|
||||
def wakeonlan_fixture():
|
||||
"""Patch the wakeonlan Remote."""
|
||||
with MockDependency("wakeonlan") as wakeonlan:
|
||||
yield wakeonlan
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def samsung_mock():
|
||||
"""Mock samsungctl."""
|
||||
with patch.dict("sys.modules", {"samsungctl": MagicMock()}):
|
||||
yield
|
||||
def mock_now():
|
||||
"""Fixture for dtutil.now."""
|
||||
return dt_util.utcnow()
|
||||
|
||||
|
||||
async def test_play_media(hass, samsung_mock):
|
||||
async def setup_samsungtv(hass, config):
|
||||
"""Set up mock Samsung TV."""
|
||||
await async_setup_component(hass, "media_player", config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def test_setup_with_mac(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):
|
||||
"""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"
|
||||
|
||||
|
||||
async def test_update_on(hass, remote, mock_now):
|
||||
"""Testing update tv on."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
|
||||
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()
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_update_off(hass, remote, mock_now):
|
||||
"""Testing update tv off."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
remote.control = mock.Mock(side_effect=OSError("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()
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
|
||||
async def test_send_key(hass, remote, wakeonlan):
|
||||
"""Test for send key."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
# key and update called
|
||||
assert remote.control.call_count == 2
|
||||
assert remote.control.call_args_list == [call("KEY_VOLUP"), call("KEY")]
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_send_key_autodetect_websocket(hass, remote):
|
||||
"""Test for send key with autodetection of protocol."""
|
||||
with patch("samsungctl.Remote") 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."""
|
||||
with patch(
|
||||
"samsungctl.Remote", 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(
|
||||
"samsungctl.Remote", 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("samsungctl.Remote", 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)
|
||||
remote.control = mock.Mock(side_effect=BrokenPipeError("Boom"))
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_send_key_connection_closed_retry_succeed(hass, remote):
|
||||
"""Test retry on connection closed."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
remote.control = mock.Mock(
|
||||
side_effect=[exceptions.ConnectionClosed("Boom"), mock.DEFAULT, mock.DEFAULT]
|
||||
)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
# key because of retry two times and update called
|
||||
assert remote.control.call_count == 3
|
||||
assert remote.control.call_args_list == [
|
||||
call("KEY_VOLUP"),
|
||||
call("KEY_VOLUP"),
|
||||
call("KEY"),
|
||||
]
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_send_key_unhandled_response(hass, remote):
|
||||
"""Testing unhandled response exception."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
remote.control = mock.Mock(side_effect=exceptions.UnhandledResponse("Boom"))
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_send_key_os_error(hass, remote):
|
||||
"""Testing broken pipe Exception."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
remote.control = mock.Mock(side_effect=OSError("Boom"))
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
|
||||
async def test_name(hass, remote):
|
||||
"""Test for name property."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.attributes[ATTR_FRIENDLY_NAME] == "fake"
|
||||
|
||||
|
||||
async def test_state_with_mac(hass, remote, wakeonlan):
|
||||
"""Test for state property."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
|
||||
async def test_state_without_mac(hass, remote):
|
||||
"""Test for state property."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG_NOMAC)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True
|
||||
)
|
||||
state = hass.states.get(ENTITY_ID_NOMAC)
|
||||
assert state.state == STATE_ON
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True
|
||||
)
|
||||
state = hass.states.get(ENTITY_ID_NOMAC)
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
|
||||
async def test_supported_features_with_mac(hass, remote):
|
||||
"""Test for supported_features property."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert (
|
||||
state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON
|
||||
)
|
||||
|
||||
|
||||
async def test_supported_features_without_mac(hass, remote):
|
||||
"""Test for supported_features property."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG_NOMAC)
|
||||
state = hass.states.get(ENTITY_ID_NOMAC)
|
||||
assert state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV
|
||||
|
||||
|
||||
async def test_device_class(hass, remote):
|
||||
"""Test for device_class property."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_TV
|
||||
|
||||
|
||||
async def test_turn_off_websocket(hass, remote):
|
||||
"""Test for turn_off."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, 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, remote):
|
||||
"""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")]
|
||||
|
||||
|
||||
async def test_turn_off_os_error(hass, remote, caplog):
|
||||
"""Test for turn_off with OSError."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
remote.close = mock.Mock(side_effect=OSError("BOOM"))
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
assert "Could not establish connection." in caplog.text
|
||||
|
||||
|
||||
async def test_volume_up(hass, remote):
|
||||
"""Test for volume_up."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
# key and update called
|
||||
assert remote.control.call_count == 2
|
||||
assert remote.control.call_args_list == [call("KEY_VOLUP"), call("KEY")]
|
||||
|
||||
|
||||
async def test_volume_down(hass, remote):
|
||||
"""Test for volume_down."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_VOLUME_DOWN, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
# key and update called
|
||||
assert remote.control.call_count == 2
|
||||
assert remote.control.call_args_list == [call("KEY_VOLDOWN"), call("KEY")]
|
||||
|
||||
|
||||
async def test_mute_volume(hass, remote):
|
||||
"""Test for mute_volume."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_VOLUME_MUTE,
|
||||
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: True},
|
||||
True,
|
||||
)
|
||||
# key and update called
|
||||
assert remote.control.call_count == 2
|
||||
assert remote.control.call_args_list == [call("KEY_MUTE"), call("KEY")]
|
||||
|
||||
|
||||
async def test_media_play(hass, remote):
|
||||
"""Test for media_play."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_MEDIA_PLAY, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
# key and update called
|
||||
assert remote.control.call_count == 2
|
||||
assert remote.control.call_args_list == [call("KEY_PLAY"), call("KEY")]
|
||||
|
||||
|
||||
async def test_media_pause(hass, remote):
|
||||
"""Test for media_pause."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_MEDIA_PAUSE, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
# key and update called
|
||||
assert remote.control.call_count == 2
|
||||
assert remote.control.call_args_list == [call("KEY_PAUSE"), call("KEY")]
|
||||
|
||||
|
||||
async def test_media_next_track(hass, remote):
|
||||
"""Test for media_next_track."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_MEDIA_NEXT_TRACK, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
# key and update called
|
||||
assert remote.control.call_count == 2
|
||||
assert remote.control.call_args_list == [call("KEY_FF"), call("KEY")]
|
||||
|
||||
|
||||
async def test_media_previous_track(hass, remote):
|
||||
"""Test for media_previous_track."""
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN, SERVICE_MEDIA_PREVIOUS_TRACK, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
# key and update called
|
||||
assert remote.control.call_count == 2
|
||||
assert remote.control.call_args_list == [call("KEY_REWIND"), call("KEY")]
|
||||
|
||||
|
||||
async def test_turn_on_with_mac(hass, remote, wakeonlan):
|
||||
"""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("fake")]
|
||||
|
||||
|
||||
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
|
||||
)
|
||||
# nothing called as not supported feature
|
||||
assert remote.control.call_count == 0
|
||||
|
||||
|
||||
async def test_play_media(hass, remote):
|
||||
"""Test for play_media."""
|
||||
asyncio_sleep = asyncio.sleep
|
||||
sleeps = []
|
||||
@ -312,57 +576,109 @@ async def test_play_media(hass, samsung_mock):
|
||||
sleeps.append(duration)
|
||||
await asyncio_sleep(0, loop=loop)
|
||||
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
with patch("asyncio.sleep", new=sleep):
|
||||
device = SamsungTVDevice(**WORKING_CONFIG)
|
||||
device.hass = hass
|
||||
|
||||
device.send_key = mock.Mock()
|
||||
await device.async_play_media(MEDIA_TYPE_CHANNEL, "576")
|
||||
|
||||
exp = [call("KEY_5"), call("KEY_7"), call("KEY_6"), call("KEY_ENTER")]
|
||||
assert device.send_key.call_args_list == exp
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_CHANNEL,
|
||||
ATTR_MEDIA_CONTENT_ID: "576",
|
||||
},
|
||||
True,
|
||||
)
|
||||
# keys and update called
|
||||
assert remote.control.call_count == 5
|
||||
assert remote.control.call_args_list == [
|
||||
call("KEY_5"),
|
||||
call("KEY_7"),
|
||||
call("KEY_6"),
|
||||
call("KEY_ENTER"),
|
||||
call("KEY"),
|
||||
]
|
||||
assert len(sleeps) == 3
|
||||
|
||||
|
||||
async def test_play_media_invalid_type(hass, samsung_mock):
|
||||
async def test_play_media_invalid_type(hass, remote):
|
||||
"""Test for play_media with invalid media type."""
|
||||
url = "https://example.com"
|
||||
device = SamsungTVDevice(**WORKING_CONFIG)
|
||||
device.send_key = mock.Mock()
|
||||
await device.async_play_media(MEDIA_TYPE_URL, url)
|
||||
assert device.send_key.call_count == 0
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_URL,
|
||||
ATTR_MEDIA_CONTENT_ID: url,
|
||||
},
|
||||
True,
|
||||
)
|
||||
# only update called
|
||||
assert remote.control.call_count == 1
|
||||
assert remote.control.call_args_list == [call("KEY")]
|
||||
|
||||
|
||||
async def test_play_media_channel_as_string(hass, samsung_mock):
|
||||
async def test_play_media_channel_as_string(hass, remote):
|
||||
"""Test for play_media with invalid channel as string."""
|
||||
url = "https://example.com"
|
||||
device = SamsungTVDevice(**WORKING_CONFIG)
|
||||
device.send_key = mock.Mock()
|
||||
await device.async_play_media(MEDIA_TYPE_CHANNEL, url)
|
||||
assert device.send_key.call_count == 0
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_CHANNEL,
|
||||
ATTR_MEDIA_CONTENT_ID: url,
|
||||
},
|
||||
True,
|
||||
)
|
||||
# only update called
|
||||
assert remote.control.call_count == 1
|
||||
assert remote.control.call_args_list == [call("KEY")]
|
||||
|
||||
|
||||
async def test_play_media_channel_as_non_positive(hass, samsung_mock):
|
||||
async def test_play_media_channel_as_non_positive(hass, remote):
|
||||
"""Test for play_media with invalid channel as non positive integer."""
|
||||
device = SamsungTVDevice(**WORKING_CONFIG)
|
||||
device.send_key = mock.Mock()
|
||||
await device.async_play_media(MEDIA_TYPE_CHANNEL, "-4")
|
||||
assert device.send_key.call_count == 0
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_PLAY_MEDIA,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_CHANNEL,
|
||||
ATTR_MEDIA_CONTENT_ID: "-4",
|
||||
},
|
||||
True,
|
||||
)
|
||||
# only update called
|
||||
assert remote.control.call_count == 1
|
||||
assert remote.control.call_args_list == [call("KEY")]
|
||||
|
||||
|
||||
async def test_select_source(hass, samsung_mock):
|
||||
async def test_select_source(hass, remote):
|
||||
"""Test for select_source."""
|
||||
device = SamsungTVDevice(**WORKING_CONFIG)
|
||||
device.hass = hass
|
||||
device.send_key = mock.Mock()
|
||||
await device.async_select_source("HDMI")
|
||||
exp = [call("KEY_HDMI")]
|
||||
assert device.send_key.call_args_list == exp
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_SELECT_SOURCE,
|
||||
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "HDMI"},
|
||||
True,
|
||||
)
|
||||
# key and update called
|
||||
assert remote.control.call_count == 2
|
||||
assert remote.control.call_args_list == [call("KEY_HDMI"), call("KEY")]
|
||||
|
||||
|
||||
async def test_select_source_invalid_source(hass, samsung_mock):
|
||||
async def test_select_source_invalid_source(hass, remote):
|
||||
"""Test for select_source with invalid source."""
|
||||
device = SamsungTVDevice(**WORKING_CONFIG)
|
||||
device.send_key = mock.Mock()
|
||||
await device.async_select_source("INVALID")
|
||||
assert device.send_key.call_count == 0
|
||||
await setup_samsungtv(hass, MOCK_CONFIG)
|
||||
assert await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_SELECT_SOURCE,
|
||||
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "INVALID"},
|
||||
True,
|
||||
)
|
||||
# only update called
|
||||
assert remote.control.call_count == 1
|
||||
assert remote.control.call_args_list == [call("KEY")]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user