Cleanup deprecated YAML from samsungtv (#93392)

This commit is contained in:
epenet 2023-05-23 11:33:51 +02:00 committed by GitHub
parent d0d74eb30e
commit 1dccb8a9a9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 226 additions and 710 deletions

View File

@ -3,14 +3,11 @@ from __future__ import annotations
from collections.abc import Coroutine, Mapping
from functools import partial
import socket
from typing import Any
from urllib.parse import urlparse
import getmac
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@ -18,7 +15,6 @@ from homeassistant.const import (
CONF_MAC,
CONF_METHOD,
CONF_MODEL,
CONF_NAME,
CONF_PORT,
CONF_TOKEN,
EVENT_HOMEASSISTANT_STOP,
@ -30,10 +26,8 @@ from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_registry as er,
issue_registry as ir,
)
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.typing import ConfigType
from .bridge import (
SamsungTVBridge,
@ -42,11 +36,9 @@ from .bridge import (
model_requires_encryption,
)
from .const import (
CONF_ON_ACTION,
CONF_SESSION_ID,
CONF_SSDP_MAIN_TV_AGENT_LOCATION,
CONF_SSDP_RENDERING_CONTROL_LOCATION,
DEFAULT_NAME,
DOMAIN,
ENTRY_RELOAD_COOLDOWN,
LEGACY_PORT,
@ -57,73 +49,9 @@ from .const import (
UPNP_SVC_RENDERING_CONTROL,
)
def ensure_unique_hosts(value: dict[Any, Any]) -> dict[Any, Any]:
"""Validate that all configs have a unique host."""
vol.Schema(vol.Unique("duplicate host entries found"))(
[entry[CONF_HOST] for entry in value]
)
return value
PLATFORMS = [Platform.MEDIA_PLAYER]
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
cv.ensure_list,
[
cv.deprecated(CONF_PORT),
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: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Samsung TV integration."""
hass.data[DOMAIN] = {}
if DOMAIN not in config:
return True
ir.async_create_issue(
hass,
DOMAIN,
"deprecated_yaml",
breaks_in_ha_version="2023.6.0",
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"on_action_url": "https://www.home-assistant.io/integrations/samsungtv/#turn-on-action"
},
learn_more_url="https://www.home-assistant.io/integrations/samsungtv/#turn-on-action",
)
for entry_config in config[DOMAIN]:
ip_address = await hass.async_add_executor_job(
socket.gethostbyname, entry_config[CONF_HOST]
)
hass.data[DOMAIN][ip_address] = {
CONF_ON_ACTION: entry_config.get(CONF_ON_ACTION)
}
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=entry_config,
)
)
return True
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
@callback
@ -195,6 +123,7 @@ async def _async_update_ssdp_locations(hass: HomeAssistant, entry: ConfigEntry)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up the Samsung TV platform."""
hass.data.setdefault(DOMAIN, {})
# Initialize bridge
if entry.data.get(CONF_METHOD) == METHOD_ENCRYPTED_WEBSOCKET:

View File

@ -35,12 +35,9 @@ from .const import (
CONF_SSDP_RENDERING_CONTROL_LOCATION,
DEFAULT_MANUFACTURER,
DOMAIN,
ENCRYPTED_WEBSOCKET_PORT,
LEGACY_PORT,
LOGGER,
METHOD_ENCRYPTED_WEBSOCKET,
METHOD_LEGACY,
METHOD_WEBSOCKET,
RESULT_AUTH_MISSING,
RESULT_CANNOT_CONNECT,
RESULT_INVALID_PIN,
@ -50,7 +47,6 @@ from .const import (
SUCCESSFUL_RESULTS,
UPNP_SVC_MAIN_TV_AGENT,
UPNP_SVC_RENDERING_CONTROL,
WEBSOCKET_PORTS,
)
DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str, vol.Required(CONF_NAME): str})
@ -231,25 +227,6 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._mac = mac
return True
async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
"""Handle configuration by yaml file."""
# We need to import even if we cannot validate
# since the TV may be off at startup
await self._async_set_name_host_from_input(user_input)
self._async_abort_entries_match({CONF_HOST: self._host})
port = user_input.get(CONF_PORT)
if port in WEBSOCKET_PORTS:
user_input[CONF_METHOD] = METHOD_WEBSOCKET
elif port == ENCRYPTED_WEBSOCKET_PORT:
user_input[CONF_METHOD] = METHOD_ENCRYPTED_WEBSOCKET
elif port == LEGACY_PORT:
user_input[CONF_METHOD] = METHOD_LEGACY
user_input[CONF_MANUFACTURER] = DEFAULT_MANUFACTURER
return self.async_create_entry(
title=self._title,
data=user_input,
)
async def _async_set_name_host_from_input(self, user_input: dict[str, Any]) -> None:
try:
self._host = await self.hass.async_add_executor_job(

View File

@ -6,7 +6,6 @@ DOMAIN = "samsungtv"
ATTR_PROPERTIES = "properties"
DEFAULT_NAME = "Samsung TV"
DEFAULT_MANUFACTURER = "Samsung"
VALUE_CONF_NAME = "HomeAssistant"
@ -16,7 +15,6 @@ CONF_DESCRIPTION = "description"
CONF_MANUFACTURER = "manufacturer"
CONF_SSDP_RENDERING_CONTROL_LOCATION = "ssdp_rendering_control_location"
CONF_SSDP_MAIN_TV_AGENT_LOCATION = "ssdp_main_tv_agent_location"
CONF_ON_ACTION = "turn_on_action"
CONF_SESSION_ID = "session_id"
RESULT_AUTH_MISSING = "auth_missing"

View File

@ -41,16 +41,13 @@ from homeassistant.helpers import (
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.script import Script
from homeassistant.helpers.trigger import PluggableAction
from homeassistant.util import dt as dt_util
from .bridge import SamsungTVBridge, SamsungTVWSBridge
from .const import (
CONF_MANUFACTURER,
CONF_ON_ACTION,
CONF_SSDP_RENDERING_CONTROL_LOCATION,
DEFAULT_NAME,
DOMAIN,
LOGGER,
)
@ -86,16 +83,7 @@ async def async_setup_entry(
) -> None:
"""Set up the Samsung TV from a config entry."""
bridge = hass.data[DOMAIN][entry.entry_id]
host = entry.data[CONF_HOST]
on_script = None
data = hass.data[DOMAIN]
if turn_on_action := data.get(host, {}).get(CONF_ON_ACTION):
on_script = Script(
hass, turn_on_action, entry.data.get(CONF_NAME, DEFAULT_NAME), DOMAIN
)
async_add_entities([SamsungTVDevice(bridge, entry, on_script)], True)
async_add_entities([SamsungTVDevice(bridge, entry)], True)
class SamsungTVDevice(MediaPlayerEntity):
@ -107,7 +95,6 @@ class SamsungTVDevice(MediaPlayerEntity):
self,
bridge: SamsungTVBridge,
config_entry: ConfigEntry,
on_script: Script | None,
) -> None:
"""Initialize the Samsung device."""
self._config_entry = config_entry
@ -117,7 +104,6 @@ class SamsungTVDevice(MediaPlayerEntity):
CONF_SSDP_RENDERING_CONTROL_LOCATION
)
self._turn_on = PluggableAction(self.async_write_ha_state)
self._on_script = on_script
# Assume that the TV is in Play mode
self._playing: bool = True
@ -130,8 +116,8 @@ class SamsungTVDevice(MediaPlayerEntity):
self._app_list_event: asyncio.Event = asyncio.Event()
self._attr_supported_features = SUPPORT_SAMSUNGTV
if self._on_script or self._mac:
# (deprecated) add turn-on if on_script YAML or mac is available
if self._mac:
# (deprecated) add turn-on if mac is available
# Triggers have not yet been registered so this is adjusted in the property
self._attr_supported_features |= MediaPlayerEntityFeature.TURN_ON
if self._ssdp_rendering_control_location:
@ -375,7 +361,6 @@ class SamsungTVDevice(MediaPlayerEntity):
return (
self.state == MediaPlayerState.ON
or bool(self._turn_on)
or self._on_script is not None
or self._mac is not None
or self._power_off_in_progress()
)
@ -477,9 +462,6 @@ class SamsungTVDevice(MediaPlayerEntity):
"""Turn the media player on."""
if self._turn_on:
await self._turn_on.async_run(self.hass, self._context)
elif self._on_script:
# YAML on_script is deprecated - replaced by turn_on trigger
await self._on_script.async_run(context=self._context)
elif self._mac:
await self.hass.async_add_executor_job(self._wake_on_lan)

View File

@ -7,7 +7,6 @@ from homeassistant.components.samsungtv.const import DOMAIN, ENTRY_RELOAD_COOLDO
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed
@ -29,7 +28,7 @@ async def setup_samsungtv_entry(hass: HomeAssistant, data: ConfigType) -> Config
)
entry.add_to_hass(hass)
await async_setup_component(hass, DOMAIN, {})
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

View File

@ -290,15 +290,6 @@ def remoteencws_fixture() -> Mock:
yield remoteencws
@pytest.fixture(name="delay")
def delay_fixture() -> Mock:
"""Patch the delay script function."""
with patch(
"homeassistant.components.samsungtv.media_player.Script.async_run"
) as delay:
yield delay
@pytest.fixture
def mock_now() -> datetime:
"""Fixture for dtutil.now."""

View File

@ -1,10 +0,0 @@
# serializer version: 1
# name: test_setup
IssueRegistryItemSnapshot({
'created': <ANY>,
'dismissed_version': None,
'domain': 'samsungtv',
'is_persistent': False,
'issue_id': 'deprecated_yaml',
})
# ---

View File

@ -1,5 +1,4 @@
"""Tests for Samsung TV config flow."""
import socket
from unittest.mock import ANY, AsyncMock, Mock, call, patch
import pytest
@ -28,13 +27,9 @@ from homeassistant.components.samsungtv.const import (
DEFAULT_MANUFACTURER,
DOMAIN,
LEGACY_PORT,
METHOD_ENCRYPTED_WEBSOCKET,
METHOD_LEGACY,
METHOD_WEBSOCKET,
RESULT_AUTH_MISSING,
RESULT_CANNOT_CONNECT,
RESULT_NOT_SUPPORTED,
RESULT_UNKNOWN_HOST,
TIMEOUT_REQUEST,
TIMEOUT_WEBSOCKET,
)
@ -61,7 +56,6 @@ from homeassistant.data_entry_flow import FlowResultType
from homeassistant.setup import async_setup_component
from .const import (
MOCK_CONFIG_ENCRYPTED_WS,
MOCK_ENTRYDATA_ENCRYPTED_WS,
MOCK_ENTRYDATA_WS,
MOCK_SSDP_DATA_MAIN_TV_AGENT_ST,
@ -907,136 +901,6 @@ async def test_ssdp_already_configured(hass: HomeAssistant) -> None:
assert entry.unique_id == "123"
@pytest.mark.usefixtures("remote")
async def test_import_legacy(hass: HomeAssistant) -> None:
"""Test importing from yaml with hostname."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_IMPORT_DATA,
)
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == "fake"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake"
assert result["data"][CONF_MANUFACTURER] == "Samsung"
assert result["result"].unique_id is None
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].data[CONF_METHOD] == METHOD_LEGACY
assert entries[0].data[CONF_PORT] == LEGACY_PORT
@pytest.mark.usefixtures("remote", "remotews", "rest_api_failing")
async def test_import_legacy_without_name(
hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test importing from yaml without a name."""
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote.start_listening",
side_effect=WebSocketProtocolError("Boom"),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_IMPORT_DATA_WITHOUT_NAME,
)
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == "fake_host"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_MANUFACTURER] == "Samsung"
assert result["result"].unique_id is None
mock_setup_entry.assert_called_once()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
# METHOD / PORT failed during import
# They will get checked/set on setup
assert CONF_METHOD not in entries[0].data
assert CONF_PORT not in entries[0].data
@pytest.mark.usefixtures("remotews", "rest_api")
async def test_import_websocket(hass: HomeAssistant) -> None:
"""Test importing from yaml with hostname."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_IMPORT_WSDATA,
)
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == "fake"
assert result["data"][CONF_METHOD] == METHOD_WEBSOCKET
assert result["data"][CONF_PORT] == 8002
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake"
assert result["data"][CONF_MANUFACTURER] == "Samsung"
assert result["result"].unique_id is None
@pytest.mark.usefixtures("remoteencws")
async def test_import_websocket_encrypted(hass: HomeAssistant) -> None:
"""Test importing from yaml with hostname."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_CONFIG_ENCRYPTED_WS,
)
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == "fake"
assert result["data"][CONF_METHOD] == METHOD_ENCRYPTED_WEBSOCKET
assert result["data"][CONF_PORT] == 8000
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake"
assert result["data"][CONF_MANUFACTURER] == "Samsung"
assert result["result"].unique_id is None
@pytest.mark.usefixtures("remotews", "rest_api")
async def test_import_websocket_without_port(hass: HomeAssistant) -> None:
"""Test importing from yaml with hostname by no port."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_IMPORT_WSDATA,
)
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == "fake"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake"
assert result["data"][CONF_MANUFACTURER] == "Samsung"
assert result["result"].unique_id is None
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].data[CONF_METHOD] == METHOD_WEBSOCKET
assert entries[0].data[CONF_PORT] == 8002
@pytest.mark.usefixtures("remotews")
async def test_import_unknown_host(hass: HomeAssistant) -> None:
"""Test importing from yaml with hostname that does not resolve."""
with patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
side_effect=socket.gaierror,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_IMPORT_DATA,
)
await hass.async_block_till_done()
assert result["type"] == "abort"
assert result["reason"] == RESULT_UNKNOWN_HOST
@pytest.mark.usefixtures("remotews", "rest_api_non_ssl_only", "remoteencws_failing")
async def test_dhcp_wireless(hass: HomeAssistant) -> None:
"""Test starting a flow from dhcp."""
@ -1389,18 +1253,14 @@ async def test_update_missing_mac_unique_id_added_from_dhcp(
"""Test missing mac and unique id added."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_OLD_ENTRY, unique_id=None)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=MOCK_DHCP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=MOCK_DHCP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -1415,18 +1275,15 @@ async def test_update_missing_mac_unique_id_added_from_zeroconf(
"""Test missing mac and unique id added."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_OLD_ENTRY, unique_id=None)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
assert entry.data[CONF_MAC] == "aa:bb:zz:ee:rr:oo"
@ -1444,18 +1301,14 @@ async def test_update_missing_model_added_from_ssdp(
unique_id=None,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -1469,18 +1322,14 @@ async def test_update_missing_mac_unique_id_ssdp_location_added_from_ssdp(
"""Test missing mac, ssdp_location, and unique id added via ssdp."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_OLD_ENTRY, unique_id=None)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -1529,18 +1378,14 @@ async def test_update_missing_mac_unique_id_added_ssdp_location_updated_from_ssd
unique_id=None,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -1566,18 +1411,14 @@ async def test_update_missing_mac_unique_id_added_ssdp_location_rendering_st_upd
unique_id=None,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -1605,18 +1446,14 @@ async def test_update_missing_mac_unique_id_added_ssdp_location_main_tv_agent_st
unique_id=None,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_MAIN_TV_AGENT_ST,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_MAIN_TV_AGENT_ST,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -1644,18 +1481,14 @@ async def test_update_ssdp_location_rendering_st_updated_from_ssdp(
unique_id="be9554b9-c9fb-41f4-8920-22da015376a4",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -1679,18 +1512,14 @@ async def test_update_main_tv_ssdp_location_rendering_st_updated_from_ssdp(
unique_id="be9554b9-c9fb-41f4-8920-22da015376a4",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_MAIN_TV_AGENT_ST,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_MAIN_TV_AGENT_ST,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -1714,18 +1543,15 @@ async def test_update_missing_mac_added_unique_id_preserved_from_zeroconf(
unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
data=MOCK_ZEROCONF_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
assert entry.data[CONF_MAC] == "aa:bb:zz:ee:rr:oo"
@ -1743,20 +1569,17 @@ async def test_update_legacy_missing_mac_from_dhcp(
unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp.DhcpServiceInfo(
ip=EXISTING_IP, macaddress="aa:bb:cc:dd:ee:ff", hostname="fake_hostname"
),
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp.DhcpServiceInfo(
ip=EXISTING_IP, macaddress="aa:bb:cc:dd:ee:ff", hostname="fake_hostname"
),
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
assert entry.data[CONF_MAC] == "aa:bb:cc:dd:ee:ff"
@ -1780,10 +1603,7 @@ async def test_update_legacy_missing_mac_from_dhcp_no_unique_id(
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote.start_listening",
side_effect=WebSocketProtocolError("Boom"),
), patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
@ -1792,8 +1612,8 @@ async def test_update_legacy_missing_mac_from_dhcp_no_unique_id(
),
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "not_supported"
assert entry.data[CONF_MAC] == "aa:bb:cc:dd:ee:ff"
@ -1811,18 +1631,14 @@ async def test_update_ssdp_location_unique_id_added_from_ssdp(
unique_id=None,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -1843,18 +1659,14 @@ async def test_update_ssdp_location_unique_id_added_from_ssdp_with_rendering_con
unique_id=None,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -2060,18 +1872,14 @@ async def test_update_incorrect_udn_matching_upnp_udn_unique_id_added_from_ssdp(
unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -2090,18 +1898,14 @@ async def test_update_incorrect_udn_matching_mac_unique_id_added_from_ssdp(
unique_id=None,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_SSDP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -2121,18 +1925,14 @@ async def test_update_incorrect_udn_matching_mac_from_dhcp(
unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=MOCK_DHCP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=MOCK_DHCP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -2152,18 +1952,14 @@ async def test_no_update_incorrect_udn_not_matching_mac_from_dhcp(
unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.async_setup",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=MOCK_DHCP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 0
assert len(mock_setup_entry.mock_calls) == 0
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=MOCK_DHCP_DATA,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 0
assert result["type"] == "form"
assert result["step_id"] == "confirm"

View File

@ -2,12 +2,10 @@
from unittest.mock import Mock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.media_player import DOMAIN, MediaPlayerEntityFeature
from homeassistant.components.samsungtv.const import (
CONF_MANUFACTURER,
CONF_ON_ACTION,
CONF_SESSION_ID,
CONF_SSDP_MAIN_TV_AGENT_LOCATION,
CONF_SSDP_RENDERING_CONTROL_LOCATION,
@ -32,8 +30,6 @@ from homeassistant.const import (
SERVICE_VOLUME_UP,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
from . import setup_samsungtv_entry
from .const import (
@ -48,43 +44,16 @@ from tests.common import MockConfigEntry
ENTITY_ID = f"{DOMAIN}.fake_name"
MOCK_CONFIG = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake_host",
CONF_NAME: "fake_name",
CONF_ON_ACTION: [{"delay": "00:00:01"}],
CONF_METHOD: METHOD_WEBSOCKET,
}
]
}
MOCK_CONFIG_WITHOUT_PORT = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_ON_ACTION: [{"delay": "00:00:01"}],
}
]
}
REMOTE_CALL = {
"name": "HomeAssistant",
"description": "HomeAssistant",
"id": "ha.component.samsung",
"host": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_HOST],
"method": "legacy",
"port": None,
"timeout": 1,
CONF_HOST: "fake_host",
CONF_NAME: "fake_name",
CONF_METHOD: METHOD_WEBSOCKET,
}
@pytest.mark.usefixtures("remotews", "remoteencws_failing", "rest_api")
async def test_setup(
hass: HomeAssistant, issue_registry: ir.IssueRegistry, snapshot: SnapshotAssertion
) -> None:
async def test_setup(hass: HomeAssistant) -> None:
"""Test Samsung TV integration is setup."""
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
await setup_samsungtv_entry(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
# test name and turn_on
@ -100,12 +69,8 @@ async def test_setup(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# ensure deprecated_yaml issue is raised
issue = issue_registry.async_get_issue(SAMSUNGTV_DOMAIN, "deprecated_yaml")
assert issue == snapshot
async def test_setup_from_yaml_without_port_device_offline(hass: HomeAssistant) -> None:
async def test_setup_without_port_device_offline(hass: HomeAssistant) -> None:
"""Test import from yaml when the device is offline."""
with patch(
"homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError
@ -119,8 +84,7 @@ async def test_setup_from_yaml_without_port_device_offline(hass: HomeAssistant)
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info",
return_value=None,
):
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
await setup_samsungtv_entry(hass, MOCK_CONFIG)
config_entries_domain = hass.config_entries.async_entries(SAMSUNGTV_DOMAIN)
assert len(config_entries_domain) == 1
@ -128,52 +92,22 @@ async def test_setup_from_yaml_without_port_device_offline(hass: HomeAssistant)
@pytest.mark.usefixtures("remotews", "remoteencws_failing", "rest_api")
async def test_setup_from_yaml_without_port_device_online(hass: HomeAssistant) -> None:
async def test_setup_without_port_device_online(hass: HomeAssistant) -> None:
"""Test import from yaml when the device is online."""
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
await setup_samsungtv_entry(hass, MOCK_CONFIG)
config_entries_domain = hass.config_entries.async_entries(SAMSUNGTV_DOMAIN)
assert len(config_entries_domain) == 1
assert config_entries_domain[0].data[CONF_MAC] == "aa:bb:ww:ii:ff:ii"
@pytest.mark.usefixtures("remote")
async def test_setup_duplicate_config(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""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("media_player")) == 0
assert "duplicate host entries found" in caplog.text
@pytest.mark.usefixtures("remotews", "remoteencws_failing", "rest_api")
async def test_setup_duplicate_entries(hass: HomeAssistant) -> None:
"""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("media_player")) == 1
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
assert len(hass.states.async_all("media_player")) == 1
@pytest.mark.usefixtures("remotews", "remoteencws_failing")
async def test_setup_h_j_model(
hass: HomeAssistant, rest_api: Mock, caplog: pytest.LogCaptureFixture
) -> None:
"""Test Samsung TV integration is setup."""
rest_api.rest_device_info.return_value = SAMPLE_DEVICE_INFO_UE48JU6400
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state

View File

@ -33,15 +33,14 @@ from homeassistant.components.media_player import (
SERVICE_PLAY_MEDIA,
SERVICE_SELECT_SOURCE,
MediaPlayerDeviceClass,
MediaPlayerEntityFeature,
MediaType,
)
from homeassistant.components.samsungtv.const import (
CONF_ON_ACTION,
CONF_SSDP_RENDERING_CONTROL_LOCATION,
DOMAIN as SAMSUNGTV_DOMAIN,
ENCRYPTED_WEBSOCKET_PORT,
METHOD_ENCRYPTED_WEBSOCKET,
METHOD_LEGACY,
METHOD_WEBSOCKET,
TIMEOUT_WEBSOCKET,
)
@ -77,8 +76,6 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from . import async_wait_config_entry_reload, setup_samsungtv_entry
@ -93,25 +90,17 @@ from tests.common import MockConfigEntry, async_fire_time_changed
ENTITY_ID = f"{DOMAIN}.fake"
MOCK_CONFIG = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_PORT: 55000,
CONF_ON_ACTION: [{"delay": "00:00:01"}],
}
]
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_PORT: 55000,
CONF_METHOD: METHOD_LEGACY,
}
MOCK_CONFIGWS = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_PORT: 8001,
CONF_TOKEN: "123456789",
CONF_ON_ACTION: [{"delay": "00:00:01"}],
}
]
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_PORT: 8001,
CONF_TOKEN: "123456789",
CONF_METHOD: METHOD_WEBSOCKET,
}
MOCK_CALLS_WS = {
CONF_HOST: "fake_host",
@ -143,43 +132,13 @@ MOCK_ENTRY_WS_WITH_MAC = {
}
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="delay")
def delay_fixture():
"""Patch the delay script function."""
with patch(
"homeassistant.components.samsungtv.media_player.Script.async_run"
) as delay:
yield delay
async def setup_samsungtv(hass: HomeAssistant, config: ConfigType) -> None:
"""Set up mock Samsung TV."""
await async_setup_component(hass, SAMSUNGTV_DOMAIN, config)
await hass.async_block_till_done()
@pytest.mark.usefixtures("remote")
async def test_setup_with_turnon(hass: HomeAssistant) -> None:
async def test_setup(hass: HomeAssistant) -> None:
"""Test setup of platform."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert hass.states.get(ENTITY_ID)
@pytest.mark.usefixtures("remote")
async def test_setup_without_turnon(hass: HomeAssistant) -> None:
"""Test setup of platform."""
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
assert hass.states.get(ENTITY_ID_NOTURNON)
@pytest.mark.usefixtures("remotews", "rest_api")
async def test_setup_websocket(hass: HomeAssistant) -> None:
"""Test setup of platform."""
@ -192,7 +151,7 @@ async def test_setup_websocket(hass: HomeAssistant) -> None:
remote.token = "123456789"
remote_class.return_value = remote
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
assert remote_class.call_count == 1
assert remote_class.call_args_list == [call(**MOCK_CALLS_WS)]
@ -229,7 +188,7 @@ async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> Non
remote.__aexit__ = AsyncMock()
remote.token = "987654321"
remote_class.return_value = remote
assert await async_setup_component(hass, SAMSUNGTV_DOMAIN, {})
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert config_entries[0].data[CONF_MAC] == "aa:bb:ww:ii:ff:ii"
@ -272,7 +231,7 @@ async def test_setup_encrypted_websocket(
@pytest.mark.usefixtures("remote")
async def test_update_on(hass: HomeAssistant, mock_now: datetime) -> None:
"""Testing update tv on."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
@ -286,7 +245,7 @@ async def test_update_on(hass: HomeAssistant, mock_now: datetime) -> None:
@pytest.mark.usefixtures("remote")
async def test_update_off(hass: HomeAssistant, mock_now: datetime) -> None:
"""Testing update tv off."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
@ -298,14 +257,14 @@ async def test_update_off(hass: HomeAssistant, mock_now: datetime) -> None:
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
assert state.state == STATE_UNAVAILABLE
async def test_update_off_ws_no_power_state(
hass: HomeAssistant, remotews: Mock, rest_api: Mock, mock_now: datetime
) -> None:
"""Testing update tv off."""
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
# device_info should only get called once, as part of the setup
rest_api.rest_device_info.assert_called_once()
rest_api.rest_device_info.reset_mock()
@ -336,13 +295,13 @@ async def test_update_off_ws_with_power_state(
) as mock_device_info, patch.object(
remotews, "start_listening", side_effect=WebSocketException("Boom")
) as mock_start_listening:
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
mock_device_info.assert_called_once()
mock_start_listening.assert_called_once()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
assert state.state == STATE_UNAVAILABLE
# First update uses start_listening once, and initialises device_info
device_info = deepcopy(SAMPLE_DEVICE_INFO_WIFI)
@ -385,7 +344,7 @@ async def test_update_off_ws_with_power_state(
rest_api.rest_device_info.assert_called_once()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
assert state.state == STATE_UNAVAILABLE
remotews.start_listening.assert_not_called()
@ -417,7 +376,7 @@ async def test_update_off_encryptedws(
@pytest.mark.usefixtures("remote")
async def test_update_access_denied(hass: HomeAssistant, mock_now: datetime) -> None:
"""Testing update tv access denied exception."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
@ -449,7 +408,7 @@ async def test_update_ws_connection_failure(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Testing update tv connection failure exception."""
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
with patch.object(
remotews,
@ -476,7 +435,7 @@ async def test_update_ws_connection_closed(
hass: HomeAssistant, mock_now: datetime, remotews: Mock
) -> None:
"""Testing update tv connection failure exception."""
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
with patch.object(
remotews, "start_listening", side_effect=ConnectionClosedError(None, None)
@ -495,7 +454,7 @@ async def test_update_ws_unauthorized_error(
hass: HomeAssistant, mock_now: datetime, remotews: Mock
) -> None:
"""Testing update tv unauthorized failure exception."""
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
with patch.object(
remotews, "start_listening", side_effect=UnauthorizedError
@ -519,7 +478,7 @@ async def test_update_unhandled_response(
hass: HomeAssistant, mock_now: datetime
) -> None:
"""Testing update tv unhandled response exception."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
@ -539,7 +498,7 @@ async def test_connection_closed_during_update_can_recover(
hass: HomeAssistant, mock_now: datetime
) -> None:
"""Testing update tv connection closed exception can recover."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
@ -551,7 +510,7 @@ async def test_connection_closed_during_update_can_recover(
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
assert state.state == STATE_UNAVAILABLE
next_update = mock_now + timedelta(minutes=10)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
@ -564,7 +523,7 @@ async def test_connection_closed_during_update_can_recover(
async def test_send_key(hass: HomeAssistant, remote: Mock) -> None:
"""Test for send key."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
@ -579,7 +538,7 @@ async def test_send_key(hass: HomeAssistant, remote: Mock) -> None:
async def test_send_key_broken_pipe(hass: HomeAssistant, remote: Mock) -> None:
"""Testing broken pipe Exception."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
remote.control = Mock(side_effect=BrokenPipeError("Boom"))
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
@ -592,7 +551,7 @@ async def test_send_key_connection_closed_retry_succeed(
hass: HomeAssistant, remote: Mock
) -> None:
"""Test retry on connection closed."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
remote.control = Mock(
side_effect=[exceptions.ConnectionClosed("Boom"), DEFAULT_MOCK, DEFAULT_MOCK]
)
@ -613,7 +572,7 @@ async def test_send_key_connection_closed_retry_succeed(
async def test_send_key_unhandled_response(hass: HomeAssistant, remote: Mock) -> None:
"""Testing unhandled response exception."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
remote.control = Mock(side_effect=exceptions.UnhandledResponse("Boom"))
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
@ -625,7 +584,7 @@ async def test_send_key_unhandled_response(hass: HomeAssistant, remote: Mock) ->
@pytest.mark.usefixtures("rest_api")
async def test_send_key_websocketexception(hass: HomeAssistant, remotews: Mock) -> None:
"""Testing unhandled response exception."""
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
remotews.send_commands = Mock(side_effect=WebSocketException("Boom"))
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
@ -651,7 +610,7 @@ async def test_send_key_websocketexception_encrypted(
@pytest.mark.usefixtures("rest_api")
async def test_send_key_os_error_ws(hass: HomeAssistant, remotews: Mock) -> None:
"""Testing unhandled response exception."""
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
remotews.send_commands = Mock(side_effect=OSError("Boom"))
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
@ -676,7 +635,7 @@ async def test_send_key_os_error_ws_encrypted(
async def test_send_key_os_error(hass: HomeAssistant, remote: Mock) -> None:
"""Testing broken pipe Exception."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
remote.control = Mock(side_effect=OSError("Boom"))
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
@ -688,42 +647,24 @@ async def test_send_key_os_error(hass: HomeAssistant, remote: Mock) -> None:
@pytest.mark.usefixtures("remote")
async def test_name(hass: HomeAssistant) -> None:
"""Test for name property."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert state.attributes[ATTR_FRIENDLY_NAME] == "fake"
@pytest.mark.usefixtures("remote")
async def test_state_with_turnon(hass: HomeAssistant, delay: Mock) -> None:
async def test_state(hass: HomeAssistant) -> None:
"""Test for state property."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
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
)
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
@pytest.mark.usefixtures("remote")
async def test_state_without_turnon(hass: HomeAssistant) -> None:
"""Test for state property."""
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True
)
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_NOTURNON}, True
)
state = hass.states.get(ENTITY_ID_NOTURNON)
# Should be STATE_UNAVAILABLE after the timer expires
assert state.state == STATE_OFF
@ -735,34 +676,23 @@ async def test_state_without_turnon(hass: HomeAssistant) -> None:
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID_NOTURNON)
state = hass.states.get(ENTITY_ID)
# Should be STATE_UNAVAILABLE since there is no way to turn it back on
assert state.state == STATE_UNAVAILABLE
@pytest.mark.usefixtures("remote")
async def test_supported_features_with_turnon(hass: HomeAssistant) -> None:
async def test_supported_features(hass: HomeAssistant) -> None:
"""Test for supported_features property."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert (
state.attributes[ATTR_SUPPORTED_FEATURES]
== SUPPORT_SAMSUNGTV | MediaPlayerEntityFeature.TURN_ON
)
@pytest.mark.usefixtures("remote")
async def test_supported_features_without_turnon(hass: HomeAssistant) -> None:
"""Test for supported_features property."""
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
state = hass.states.get(ENTITY_ID_NOTURNON)
assert state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV
@pytest.mark.usefixtures("remote")
async def test_device_class(hass: HomeAssistant) -> None:
"""Test for device_class property."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
state = hass.states.get(ENTITY_ID)
assert state.attributes[ATTR_DEVICE_CLASS] is MediaPlayerDeviceClass.TV.value
@ -777,7 +707,7 @@ async def test_turn_off_websocket(
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[OSError("Boom"), DEFAULT_MOCK],
):
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
remotews.send_commands.reset_mock()
@ -816,7 +746,7 @@ async def test_turn_off_websocket_frame(
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=[OSError("Boom"), DEFAULT_MOCK],
):
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
remotews.send_commands.reset_mock()
@ -903,9 +833,9 @@ async def test_turn_off_encrypted_websocket_key_type(
async def test_turn_off_legacy(hass: HomeAssistant, remote: Mock) -> None:
"""Test for turn_off."""
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# key called
assert remote.control.call_count == 1
@ -917,7 +847,7 @@ async def test_turn_off_os_error(
) -> None:
"""Test for turn_off with OSError."""
caplog.set_level(logging.DEBUG)
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
remote.close = Mock(side_effect=OSError("BOOM"))
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
@ -931,7 +861,7 @@ async def test_turn_off_ws_os_error(
) -> None:
"""Test for turn_off with OSError."""
caplog.set_level(logging.DEBUG)
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
remotews.close = Mock(side_effect=OSError("BOOM"))
assert await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
@ -955,7 +885,7 @@ async def test_turn_off_encryptedws_os_error(
async def test_volume_up(hass: HomeAssistant, remote: Mock) -> None:
"""Test for volume_up."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
@ -968,7 +898,7 @@ async def test_volume_up(hass: HomeAssistant, remote: Mock) -> None:
async def test_volume_down(hass: HomeAssistant, remote: Mock) -> None:
"""Test for volume_down."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_DOWN, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
@ -981,7 +911,7 @@ async def test_volume_down(hass: HomeAssistant, remote: Mock) -> None:
async def test_mute_volume(hass: HomeAssistant, remote: Mock) -> None:
"""Test for mute_volume."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN,
SERVICE_VOLUME_MUTE,
@ -997,7 +927,7 @@ async def test_mute_volume(hass: HomeAssistant, remote: Mock) -> None:
async def test_media_play(hass: HomeAssistant, remote: Mock) -> None:
"""Test for media_play."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_MEDIA_PLAY, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
@ -1019,7 +949,7 @@ async def test_media_play(hass: HomeAssistant, remote: Mock) -> None:
async def test_media_pause(hass: HomeAssistant, remote: Mock) -> None:
"""Test for media_pause."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_MEDIA_PAUSE, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
@ -1041,7 +971,7 @@ async def test_media_pause(hass: HomeAssistant, remote: Mock) -> None:
async def test_media_next_track(hass: HomeAssistant, remote: Mock) -> None:
"""Test for media_next_track."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_MEDIA_NEXT_TRACK, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
@ -1054,7 +984,7 @@ async def test_media_next_track(hass: HomeAssistant, remote: Mock) -> None:
async def test_media_previous_track(hass: HomeAssistant, remote: Mock) -> None:
"""Test for media_previous_track."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN, SERVICE_MEDIA_PREVIOUS_TRACK, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
@ -1065,16 +995,6 @@ async def test_media_previous_track(hass: HomeAssistant, remote: Mock) -> None:
assert remote.close.call_args_list == [call()]
@pytest.mark.usefixtures("remote")
async def test_turn_on_with_turnon(hass: HomeAssistant, delay: Mock) -> None:
"""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
)
assert delay.call_count == 1
@pytest.mark.usefixtures("remotews", "rest_api")
async def test_turn_on_wol(hass: HomeAssistant) -> None:
"""Test turn on."""
@ -1084,7 +1004,7 @@ async def test_turn_on_wol(hass: HomeAssistant) -> None:
unique_id="any",
)
entry.add_to_hass(hass)
assert await async_setup_component(hass, SAMSUNGTV_DOMAIN, {})
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with patch(
"homeassistant.components.samsungtv.media_player.send_magic_packet"
@ -1098,10 +1018,10 @@ async def test_turn_on_wol(hass: HomeAssistant) -> None:
async def test_turn_on_without_turnon(hass: HomeAssistant, remote: Mock) -> None:
"""Test turn on."""
await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True
DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
# nothing called as not supported feature
assert remote.control.call_count == 0
@ -1109,7 +1029,7 @@ async def test_turn_on_without_turnon(hass: HomeAssistant, remote: Mock) -> None
async def test_play_media(hass: HomeAssistant, remote: Mock) -> None:
"""Test for play_media."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
with patch("homeassistant.components.samsungtv.bridge.asyncio.sleep") as sleep:
assert await hass.services.async_call(
DOMAIN,
@ -1138,7 +1058,7 @@ async def test_play_media_invalid_type(hass: HomeAssistant) -> None:
"""Test for play_media with invalid media type."""
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote:
url = "https://example.com"
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
remote.reset_mock()
assert await hass.services.async_call(
DOMAIN,
@ -1160,7 +1080,7 @@ async def test_play_media_channel_as_string(hass: HomeAssistant) -> None:
"""Test for play_media with invalid channel as string."""
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote:
url = "https://example.com"
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
remote.reset_mock()
assert await hass.services.async_call(
DOMAIN,
@ -1181,7 +1101,7 @@ async def test_play_media_channel_as_string(hass: HomeAssistant) -> None:
async def test_play_media_channel_as_non_positive(hass: HomeAssistant) -> None:
"""Test for play_media with invalid channel as non positive integer."""
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote:
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
remote.reset_mock()
assert await hass.services.async_call(
DOMAIN,
@ -1201,7 +1121,7 @@ async def test_play_media_channel_as_non_positive(hass: HomeAssistant) -> None:
async def test_select_source(hass: HomeAssistant, remote: Mock) -> None:
"""Test for select_source."""
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
assert await hass.services.async_call(
DOMAIN,
SERVICE_SELECT_SOURCE,
@ -1218,7 +1138,7 @@ async def test_select_source(hass: HomeAssistant, remote: Mock) -> None:
async def test_select_source_invalid_source(hass: HomeAssistant) -> None:
"""Test for select_source with invalid source."""
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote:
await setup_samsungtv(hass, MOCK_CONFIG)
await setup_samsungtv_entry(hass, MOCK_CONFIG)
remote.reset_mock()
assert await hass.services.async_call(
DOMAIN,
@ -1235,7 +1155,7 @@ async def test_select_source_invalid_source(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("rest_api")
async def test_play_media_app(hass: HomeAssistant, remotews: Mock) -> None:
"""Test for play_media."""
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
remotews.send_commands.reset_mock()
assert await hass.services.async_call(
@ -1259,7 +1179,7 @@ async def test_play_media_app(hass: HomeAssistant, remotews: Mock) -> None:
async def test_select_source_app(hass: HomeAssistant, remotews: Mock) -> None:
"""Test for select_source."""
remotews.app_list_data = SAMPLE_EVENT_ED_INSTALLED_APP
await setup_samsungtv(hass, MOCK_CONFIGWS)
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
remotews.send_commands.reset_mock()
assert await hass.services.async_call(