Add Samsung TV via ExLink integration

Add a new integration to control Samsung consumer TVs over their RS-232
ExLink serial port, backed by the samsung-exlink library. Provides a
media player entity with power, volume, mute, and input source control.

A powered-off TV does not answer status queries; this is treated as the
TV being off rather than unavailable, and the source/volume/mute controls
are only offered while the TV is on.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Paulus Schoutsen
2026-06-16 22:27:10 -04:00
parent e5aa33502c
commit e12bc9b063
20 changed files with 1371 additions and 1 deletions
+1
View File
@@ -504,6 +504,7 @@ homeassistant.components.rss_feed_template.*
homeassistant.components.russound_rio.*
homeassistant.components.ruuvi_gateway.*
homeassistant.components.ruuvitag_ble.*
homeassistant.components.samsung_exlink.*
homeassistant.components.samsung_infrared.*
homeassistant.components.samsungtv.*
homeassistant.components.saunum.*
Generated
+2
View File
@@ -1588,6 +1588,8 @@ CLAUDE.md @home-assistant/core
/tests/components/sabnzbd/ @shaiu @jpbede
/homeassistant/components/saj/ @fredericvl @edurenye
/tests/components/saj/ @fredericvl @edurenye
/homeassistant/components/samsung_exlink/ @balloob
/tests/components/samsung_exlink/ @balloob
/homeassistant/components/samsung_infrared/ @lmaertin
/tests/components/samsung_infrared/ @lmaertin
/homeassistant/components/samsungtv/ @chemelli74
+7 -1
View File
@@ -1,5 +1,11 @@
{
"domain": "samsung",
"name": "Samsung",
"integrations": ["familyhub", "samsung_infrared", "samsungtv", "syncthru"]
"integrations": [
"familyhub",
"samsung_exlink",
"samsung_infrared",
"samsungtv",
"syncthru"
]
}
@@ -0,0 +1,61 @@
"""The Samsung ExLink integration."""
from samsung_exlink import MODELS, SamsungTV, TVState
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_DEVICE, CONF_MODEL, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from .const import DOMAIN, LOGGER, SamsungExLinkConfigEntry
PLATFORMS = [Platform.MEDIA_PLAYER]
async def async_setup_entry(
hass: HomeAssistant, entry: SamsungExLinkConfigEntry
) -> bool:
"""Set up Samsung ExLink from a config entry."""
port = entry.data[CONF_DEVICE]
tv = SamsungTV(port, model=MODELS.get(entry.data.get(CONF_MODEL, "")))
try:
await tv.connect()
# refresh() tolerates a powered-off TV; it only raises on a broken link.
await tv.refresh()
except (ConnectionError, OSError, TimeoutError) as err:
if tv.connected:
await tv.disconnect()
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": str(err)},
) from err
entry.runtime_data = tv
@callback
def _on_disconnect(state: TVState | None) -> None:
# Only reload if the entry is still loaded. During entry removal,
# disconnect() fires this callback but the entry is already gone.
if state is None and entry.state is ConfigEntryState.LOADED:
LOGGER.warning("Samsung TV disconnected, reloading config entry")
hass.config_entries.async_schedule_reload(entry.entry_id)
entry.async_on_unload(tv.subscribe(_on_disconnect))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(
hass: HomeAssistant, entry: SamsungExLinkConfigEntry
) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
await entry.runtime_data.disconnect()
return unload_ok
@@ -0,0 +1,113 @@
"""Config flow for the Samsung ExLink integration."""
from typing import Any
from samsung_exlink import MODELS, SamsungTV, SamsungTVError, TVModel
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_DEVICE, CONF_MODEL
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
SerialPortSelector,
)
from .const import DOMAIN, LOGGER
DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_DEVICE): SerialPortSelector(),
vol.Optional(CONF_MODEL): SelectSelector(
SelectSelectorConfig(
options=[
SelectOptionDict(value=key, label=model.name)
for key, model in MODELS.items()
],
mode=SelectSelectorMode.DROPDOWN,
)
),
}
)
# Outcome of _async_attempt_connect that means the serial port works but no
# Samsung TV answered it; this routes the user to the troubleshooting step.
RESULT_NO_TV = "no_tv"
async def _async_attempt_connect(port: str, model: TVModel | None) -> str | None:
"""Attempt to connect to the TV at the given port.
Returns None on success, otherwise an outcome key: "cannot_connect" when
the serial port could not be opened, RESULT_NO_TV when the port works but
no Samsung TV responded to it, or "unknown" for an unexpected error.
"""
tv = SamsungTV(port, model=model)
try:
await tv.connect()
except ValueError, ConnectionError, OSError, TimeoutError:
# The serial port itself could not be opened.
return "cannot_connect"
except Exception: # noqa: BLE001
LOGGER.exception("Unexpected exception")
return "unknown"
try:
await tv.query_power()
except TimeoutError, SamsungTVError:
# The port was opened but no Samsung TV responded to the power query.
return RESULT_NO_TV
except Exception: # noqa: BLE001
LOGGER.exception("Unexpected exception")
return "unknown"
finally:
await tv.disconnect()
return None
class SamsungExLinkConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Samsung ExLink."""
VERSION = 1
_user_input: dict[str, Any] | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
port = user_input[CONF_DEVICE]
model_key = user_input.get(CONF_MODEL)
self._async_abort_entries_match({CONF_DEVICE: port})
error = await _async_attempt_connect(port, MODELS.get(model_key or ""))
if error is None:
return self.async_create_entry(title="Samsung TV", data=user_input)
if error == RESULT_NO_TV:
self._user_input = user_input
return await self.async_step_troubleshoot()
errors["base"] = error
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
DATA_SCHEMA, user_input or self._user_input or {}
),
errors=errors,
)
async def async_step_troubleshoot(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Guide the user to enable ExLink control after a failed connection."""
if user_input is not None:
return await self.async_step_user()
return self.async_show_form(step_id="troubleshoot")
@@ -0,0 +1,12 @@
"""Constants for the Samsung ExLink integration."""
import logging
from samsung_exlink import SamsungTV
from homeassistant.config_entries import ConfigEntry
LOGGER = logging.getLogger(__package__)
DOMAIN = "samsung_exlink"
type SamsungExLinkConfigEntry = ConfigEntry[SamsungTV]
@@ -0,0 +1,13 @@
{
"domain": "samsung_exlink",
"name": "Samsung TV via ExLink",
"codeowners": ["@balloob"],
"config_flow": true,
"dependencies": ["usb"],
"documentation": "https://www.home-assistant.io/integrations/samsung_exlink",
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["samsung_exlink"],
"quality_scale": "silver",
"requirements": ["samsung-exlink==1.1.0"]
}
@@ -0,0 +1,196 @@
"""Media player platform for the Samsung ExLink integration."""
from collections.abc import Callable, Coroutine
from datetime import timedelta
from functools import wraps
from typing import Any
from samsung_exlink import MAX_VOLUME, CommandRejected, InputSource, TVState
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN, SamsungExLinkConfigEntry
# Samsung TVs do not push state over RS-232, so the entity is polled.
SCAN_INTERVAL = timedelta(seconds=5)
PARALLEL_UPDATES = 1
INPUT_SOURCE_SAMSUNG_TO_HA: dict[InputSource, str] = {
InputSource.TV: "tv",
InputSource.AV1: "av1",
InputSource.AV2: "av2",
InputSource.AV3: "av3",
InputSource.S_VIDEO1: "s_video1",
InputSource.S_VIDEO2: "s_video2",
InputSource.S_VIDEO3: "s_video3",
InputSource.COMPONENT1: "component1",
InputSource.COMPONENT2: "component2",
InputSource.COMPONENT3: "component3",
InputSource.PC1: "pc1",
InputSource.PC2: "pc2",
InputSource.PC3: "pc3",
InputSource.HDMI1: "hdmi1",
InputSource.HDMI2: "hdmi2",
InputSource.HDMI3: "hdmi3",
InputSource.HDMI4: "hdmi4",
InputSource.DVI1: "dvi1",
InputSource.DVI2: "dvi2",
InputSource.DVI3: "dvi3",
InputSource.RVU: "rvu",
}
INPUT_SOURCE_HA_TO_SAMSUNG: dict[str, InputSource] = {
value: key for key, value in INPUT_SOURCE_SAMSUNG_TO_HA.items()
}
_BASE_SUPPORTED_FEATURES = (
MediaPlayerEntityFeature.TURN_ON | MediaPlayerEntityFeature.TURN_OFF
)
def catch_command_errors[**_P](
func: Callable[_P, Coroutine[Any, Any, None]],
) -> Callable[_P, Coroutine[Any, Any, None]]:
"""Translate Samsung library errors raised by an action into HomeAssistantError."""
@wraps(func)
async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
try:
await func(*args, **kwargs)
except CommandRejected as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_rejected",
translation_placeholders={"error": str(err)},
) from err
except (ConnectionError, OSError, TimeoutError) as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_failed",
translation_placeholders={"error": str(err)},
) from err
return wrapper
async def async_setup_entry(
hass: HomeAssistant,
config_entry: SamsungExLinkConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Samsung ExLink media player."""
async_add_entities([SamsungExLinkMediaPlayer(config_entry)])
class SamsungExLinkMediaPlayer(MediaPlayerEntity):
"""Representation of a Samsung TV controlled over ExLink (RS-232)."""
_attr_device_class = MediaPlayerDeviceClass.TV
_attr_has_entity_name = True
_attr_name = None
_attr_translation_key = "tv"
_attr_source_list = sorted(INPUT_SOURCE_SAMSUNG_TO_HA.values())
def __init__(self, config_entry: SamsungExLinkConfigEntry) -> None:
"""Initialize the media player."""
self._tv = config_entry.runtime_data
self._attr_unique_id = config_entry.entry_id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, config_entry.entry_id)},
manufacturer="Samsung",
)
self._async_update_from_state(self._tv.state)
async def async_added_to_hass(self) -> None:
"""Subscribe to TV state updates."""
self.async_on_remove(self._tv.subscribe(self._async_on_state_update))
async def async_update(self) -> None:
"""Poll the TV for its current state."""
await self._tv.refresh()
@callback
def _async_on_state_update(self, state: TVState | None) -> None:
"""Handle a state update from the TV."""
if state is None:
self._attr_available = False
else:
self._attr_available = True
self._async_update_from_state(state)
self.async_write_ha_state()
@callback
def _async_update_from_state(self, state: TVState) -> None:
"""Update entity attributes from a TV state snapshot."""
if state.power is None:
self._attr_state = None
else:
self._attr_state = (
MediaPlayerState.ON if state.power else MediaPlayerState.OFF
)
# A standby TV only accepts power-on over RS-232; source, volume, and
# mute commands time out. Those controls (and their attributes) are
# therefore only exposed while the TV is on, and volume/mute also
# require the value to be known.
features = _BASE_SUPPORTED_FEATURES
if state.power:
features |= MediaPlayerEntityFeature.SELECT_SOURCE
source = state.input_source
self._attr_source = (
INPUT_SOURCE_SAMSUNG_TO_HA.get(source) if source else None
)
if state.volume is None:
self._attr_volume_level = None
else:
features |= (
MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_STEP
)
self._attr_volume_level = state.volume / MAX_VOLUME
if state.mute is None:
self._attr_is_volume_muted = None
else:
features |= MediaPlayerEntityFeature.VOLUME_MUTE
self._attr_is_volume_muted = state.mute
else:
self._attr_source = None
self._attr_volume_level = None
self._attr_is_volume_muted = None
self._attr_supported_features = features
@catch_command_errors
async def async_turn_on(self) -> None:
"""Turn the TV on."""
await self._tv.power_on()
@catch_command_errors
async def async_turn_off(self) -> None:
"""Turn the TV off."""
await self._tv.power_off()
@catch_command_errors
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
await self._tv.set_volume(round(volume * MAX_VOLUME))
@catch_command_errors
async def async_mute_volume(self, mute: bool) -> None:
"""Mute or unmute the TV."""
await self._tv.set_mute(mute)
@catch_command_errors
async def async_select_source(self, source: str) -> None:
"""Select an input source."""
await self._tv.select_input_source(INPUT_SOURCE_HA_TO_SAMSUNG[source])
@@ -0,0 +1,84 @@
rules:
# Bronze
action-setup:
status: exempt
comment: The integration does not register custom actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: The integration does not register custom actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: done
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: The integration has no options to configure.
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow:
status: exempt
comment: The integration does not require authentication.
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: Serial devices are configured manually; there is no discovery.
discovery:
status: exempt
comment: RS-232 serial connections cannot be discovered.
docs-data-update: done
docs-examples: done
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices:
status: exempt
comment: The integration does not create dynamic devices.
entity-category: done
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: The integration only provides a single primary entity.
entity-translations: done
exception-translations: done
icon-translations:
status: exempt
comment: The media player entity uses its device class for its icon.
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: The integration has no user-actionable issues to repair.
stale-devices:
status: exempt
comment: The integration does not create devices that can become stale.
# Platinum
async-dependency: done
inject-websession:
status: exempt
comment: The integration does not make HTTP requests.
strict-typing: done
@@ -0,0 +1,72 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"troubleshoot": {
"description": "Home Assistant could not communicate with the Samsung TV over the serial port.\n\nThe most common cause is that ExLink control is not enabled. On TVs that use the USB-to-ExLink dongle, **EXT Link Support** and **USB Serial** must be enabled in the hidden service menu (on the remote, press **Mute → 1 → 8 → 2 → Power**); the exact path varies by model, so check your TV's documentation.\n\nAlso make sure that:\n- The TV is powered on. A TV in standby does not answer status queries.\n- The cable is connected to the TV's 3.5 mm ExLink jack (or the Samsung USB-to-ExLink dongle) and is fully seated.\n- The correct serial port was selected.\n\nSelect **Submit** to try again.",
"title": "Connection failed"
},
"user": {
"data": {
"device": "[%key:common::config_flow::data::port%]",
"model": "TV generation"
},
"data_description": {
"device": "Serial port path to connect to.",
"model": "Select your TV's generation to translate the active input back into a named source. Leave empty if your model is not listed; you can still switch sources."
},
"description": "Make sure the TV is powered on before continuing. Home Assistant queries the TV during setup to confirm the connection, and a TV in standby will not respond."
}
}
},
"entity": {
"media_player": {
"tv": {
"state_attributes": {
"source": {
"state": {
"av1": "AV 1",
"av2": "AV 2",
"av3": "AV 3",
"component1": "Component 1",
"component2": "Component 2",
"component3": "Component 3",
"dvi1": "DVI 1",
"dvi2": "DVI 2",
"dvi3": "DVI 3",
"hdmi1": "HDMI 1",
"hdmi2": "HDMI 2",
"hdmi3": "HDMI 3",
"hdmi4": "HDMI 4",
"pc1": "PC 1",
"pc2": "PC 2",
"pc3": "PC 3",
"rvu": "RVU",
"s_video1": "S-Video 1",
"s_video2": "S-Video 2",
"s_video3": "S-Video 3",
"tv": "TV"
}
}
}
}
}
},
"exceptions": {
"cannot_connect": {
"message": "Error connecting to the Samsung TV: {error}"
},
"command_failed": {
"message": "Failed to send the command to the TV: {error}"
},
"command_rejected": {
"message": "The TV rejected the command: {error}"
}
}
}
+1
View File
@@ -672,6 +672,7 @@ FLOWS = {
"rympro",
"sabnzbd",
"saj",
"samsung_exlink",
"samsung_infrared",
"samsungtv",
"sanix",
@@ -6207,6 +6207,12 @@
"iot_class": "local_polling",
"name": "Samsung Family Hub"
},
"samsung_exlink": {
"integration_type": "device",
"config_flow": true,
"iot_class": "local_polling",
"name": "Samsung TV via ExLink"
},
"samsung_infrared": {
"integration_type": "device",
"config_flow": true,
Generated
+10
View File
@@ -4797,6 +4797,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.samsung_exlink.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.samsung_infrared.*]
check_untyped_defs = true
disallow_incomplete_defs = true
+3
View File
@@ -2963,6 +2963,9 @@ ruuvitag-ble==0.4.0
# homeassistant.components.yamaha
rxv==0.7.0
# homeassistant.components.samsung_exlink
samsung-exlink==1.1.0
# homeassistant.components.samsungtv
samsungctl[websocket]==0.7.1
@@ -0,0 +1,4 @@
"""Tests for the Samsung ExLink integration."""
MOCK_DEVICE = "/dev/ttyUSB0"
MOCK_MODEL = "frame_2022"
+106
View File
@@ -0,0 +1,106 @@
"""Test fixtures for the Samsung ExLink integration."""
from unittest.mock import AsyncMock, patch
import pytest
from samsung_exlink import MODELS, InputSource, PowerState, SamsungTV, TVState
from homeassistant.components.samsung_exlink.const import DOMAIN
from homeassistant.const import CONF_DEVICE, CONF_MODEL
from homeassistant.core import HomeAssistant
from . import MOCK_DEVICE, MOCK_MODEL
from tests.common import MockConfigEntry
class MockSamsungTV(SamsungTV):
"""Samsung TV test double built on the real controller object."""
def __init__(self, state: TVState) -> None:
"""Initialize the mock TV."""
super().__init__(MOCK_DEVICE, model=MODELS[MOCK_MODEL])
self._connected = True
self._state = state
self.connect = AsyncMock(side_effect=self._mock_connect)
self.refresh = AsyncMock()
self.query_power = AsyncMock(return_value=PowerState.ON)
self.disconnect = AsyncMock(side_effect=self._mock_disconnect)
self.power_on = AsyncMock()
self.power_off = AsyncMock()
self.set_volume = AsyncMock()
self.set_mute = AsyncMock()
self.select_input_source = AsyncMock()
def mock_state(self, state: TVState | None) -> None:
"""Push a state update through the TV."""
self._connected = state is not None
if state is not None:
self._state = state
self._notify_subscribers()
async def _mock_connect(self) -> None:
"""Pretend to open the serial connection."""
self._connected = True
async def _mock_disconnect(self) -> None:
"""Pretend to close the serial connection."""
self._connected = False
self._notify_subscribers()
def _default_state() -> TVState:
"""Return a TVState with typical defaults."""
return TVState(
power=True,
input_source=InputSource.HDMI1,
volume=20,
mute=False,
)
@pytest.fixture
def initial_tv_state() -> TVState:
"""Return the initial TV state for a test."""
return _default_state()
@pytest.fixture
def mock_samsung_tv(initial_tv_state: TVState) -> MockSamsungTV:
"""Create a mock SamsungTV controller."""
return MockSamsungTV(initial_tv_state)
@pytest.fixture
def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Create a mock config entry."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
title="Samsung TV",
entry_id="01KPBBPM6WCQ8148EFR0TCG1WW",
)
entry.add_to_hass(hass)
return entry
@pytest.fixture(autouse=True)
async def mock_usb_component(hass: HomeAssistant) -> None:
"""Mock the USB component to prevent setup failures."""
hass.config.components.add("usb")
@pytest.fixture
async def init_components(
hass: HomeAssistant,
mock_usb_component: None,
mock_samsung_tv: MockSamsungTV,
mock_config_entry: MockConfigEntry,
) -> None:
"""Initialize the Samsung ExLink component."""
with patch(
"homeassistant.components.samsung_exlink.SamsungTV",
return_value=mock_samsung_tv,
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
@@ -0,0 +1,103 @@
# serializer version: 1
# name: test_entities_created[media_player.samsung_tv-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<MediaPlayerEntityCapabilityAttribute.INPUT_SOURCE_LIST: 'source_list'>: list([
'av1',
'av2',
'av3',
'component1',
'component2',
'component3',
'dvi1',
'dvi2',
'dvi3',
'hdmi1',
'hdmi2',
'hdmi3',
'hdmi4',
'pc1',
'pc2',
'pc3',
'rvu',
's_video1',
's_video2',
's_video3',
'tv',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'media_player',
'entity_category': None,
'entity_id': 'media_player.samsung_tv',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': <MediaPlayerDeviceClass.TV: 'tv'>,
'original_icon': None,
'original_name': None,
'platform': 'samsung_exlink',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 3468>,
'translation_key': 'tv',
'unique_id': '01KPBBPM6WCQ8148EFR0TCG1WW',
'unit_of_measurement': None,
})
# ---
# name: test_entities_created[media_player.samsung_tv-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'tv',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Samsung TV',
<MediaPlayerEntityStateAttribute.MEDIA_VOLUME_MUTED: 'is_volume_muted'>: False,
<MediaPlayerEntityStateAttribute.INPUT_SOURCE: 'source'>: 'hdmi1',
<MediaPlayerEntityCapabilityAttribute.INPUT_SOURCE_LIST: 'source_list'>: list([
'av1',
'av2',
'av3',
'component1',
'component2',
'component3',
'dvi1',
'dvi2',
'dvi3',
'hdmi1',
'hdmi2',
'hdmi3',
'hdmi4',
'pc1',
'pc2',
'pc3',
'rvu',
's_video1',
's_video2',
's_video3',
'tv',
]),
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <MediaPlayerEntityFeature: 3468>,
<MediaPlayerEntityStateAttribute.MEDIA_VOLUME_LEVEL: 'volume_level'>: 0.2,
}),
'context': <ANY>,
'entity_id': 'media_player.samsung_tv',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
@@ -0,0 +1,221 @@
"""Tests for the Samsung ExLink config flow."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from samsung_exlink import SamsungTVError
from homeassistant.components.samsung_exlink.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_DEVICE, CONF_MODEL
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import MOCK_DEVICE, MOCK_MODEL
from tests.common import MockConfigEntry
@pytest.fixture
def mock_async_setup_entry(mock_samsung_tv: MagicMock) -> Generator[AsyncMock]:
"""Prevent config-entry creation tests from setting up the integration."""
with patch(
"homeassistant.components.samsung_exlink.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
async def test_user_form_creates_entry(
hass: HomeAssistant,
mock_samsung_tv: MagicMock,
mock_async_setup_entry: AsyncMock,
) -> None:
"""Test successful config flow creates an entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.samsung_exlink.config_flow.SamsungTV",
return_value=mock_samsung_tv,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Samsung TV"
assert result["data"] == {CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}
mock_async_setup_entry.assert_awaited_once()
mock_samsung_tv.connect.assert_awaited_once()
mock_samsung_tv.query_power.assert_awaited_once()
mock_samsung_tv.disconnect.assert_awaited_once()
async def test_user_form_without_model(
hass: HomeAssistant,
mock_samsung_tv: MagicMock,
mock_async_setup_entry: AsyncMock,
) -> None:
"""Test the model field is optional."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.samsung_exlink.config_flow.SamsungTV",
return_value=mock_samsung_tv,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {CONF_DEVICE: MOCK_DEVICE}
@pytest.mark.parametrize(
"exception",
[TimeoutError, SamsungTVError("garbled response")],
)
async def test_user_form_no_tv_shows_troubleshooting(
hass: HomeAssistant,
mock_samsung_tv: MagicMock,
mock_async_setup_entry: AsyncMock,
exception: Exception,
) -> None:
"""Test a working port with no responding TV routes to troubleshooting."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_samsung_tv.query_power.side_effect = exception
with patch(
"homeassistant.components.samsung_exlink.config_flow.SamsungTV",
return_value=mock_samsung_tv,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "troubleshoot"
# Continuing from troubleshooting returns to the user step to retry.
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
mock_samsung_tv.query_power.side_effect = None
with patch(
"homeassistant.components.samsung_exlink.config_flow.SamsungTV",
return_value=mock_samsung_tv,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_user_form_query_unexpected_error(
hass: HomeAssistant,
mock_samsung_tv: MagicMock,
mock_async_setup_entry: AsyncMock,
) -> None:
"""Test an unexpected error while querying the TV shows the unknown error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_samsung_tv.query_power.side_effect = RuntimeError("boom")
with patch(
"homeassistant.components.samsung_exlink.config_flow.SamsungTV",
return_value=mock_samsung_tv,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "unknown"}
mock_samsung_tv.disconnect.assert_awaited_once()
@pytest.mark.parametrize(
("exception", "error"),
[
(ValueError("Invalid port"), "cannot_connect"),
(OSError("No such device"), "cannot_connect"),
(ConnectionRefusedError("Connection refused"), "cannot_connect"),
(RuntimeError("boom"), "unknown"),
],
)
async def test_user_form_bad_port_shows_error(
hass: HomeAssistant,
exception: Exception,
error: str,
mock_samsung_tv: MagicMock,
mock_async_setup_entry: AsyncMock,
) -> None:
"""Test a bad serial port keeps the user on the form with an error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_samsung_tv.connect.side_effect = exception
with patch(
"homeassistant.components.samsung_exlink.config_flow.SamsungTV",
return_value=mock_samsung_tv,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": error}
mock_samsung_tv.connect.side_effect = None
with patch(
"homeassistant.components.samsung_exlink.config_flow.SamsungTV",
return_value=mock_samsung_tv,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_user_duplicate_aborts(hass: HomeAssistant) -> None:
"""Test we abort if the same port is already configured."""
entry = MockConfigEntry(
domain=DOMAIN, data={CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL}
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: MOCK_DEVICE, CONF_MODEL: MOCK_MODEL},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@@ -0,0 +1,65 @@
"""Tests for the Samsung ExLink integration init."""
from unittest.mock import AsyncMock, patch
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from .conftest import MockSamsungTV
from tests.common import MockConfigEntry
async def test_setup_connection_error(
hass: HomeAssistant,
mock_samsung_tv: MockSamsungTV,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test a connection failure results in a retry."""
mock_samsung_tv.connect.side_effect = TimeoutError
with patch(
"homeassistant.components.samsung_exlink.SamsungTV",
return_value=mock_samsung_tv,
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_unload_entry(
hass: HomeAssistant,
mock_samsung_tv: MockSamsungTV,
init_components: None,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test unloading a config entry disconnects from the TV."""
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
mock_samsung_tv.disconnect.assert_awaited_once()
async def test_remove_entry_while_loaded(
hass: HomeAssistant,
mock_samsung_tv: MockSamsungTV,
init_components: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test removing a config entry does not schedule a reload.
When removing a loaded entry, disconnect() fires the subscriber callback
with state=None. The callback must not schedule a reload because the entry
is already being removed (state is no longer LOADED).
"""
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_remove(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
mock_samsung_tv.disconnect.assert_awaited_once()
@@ -0,0 +1,291 @@
"""Tests for the Samsung ExLink media player platform."""
from pathlib import Path
from unittest.mock import call
from freezegun.api import FrozenDateTimeFactory
import pytest
from samsung_exlink import CommandRejected, InputSource, TVState
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_INPUT_SOURCE_LIST,
ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED,
DOMAIN as MP_DOMAIN,
SERVICE_SELECT_SOURCE,
MediaPlayerEntityFeature,
)
from homeassistant.components.samsung_exlink.media_player import (
INPUT_SOURCE_SAMSUNG_TO_HA,
SCAN_INTERVAL,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_SUPPORTED_FEATURES,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_SET,
SERVICE_VOLUME_UP,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.util.json import load_json
from .conftest import MockSamsungTV, _default_state
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
ENTITY_ID = "media_player.samsung_tv"
@pytest.fixture(autouse=True)
async def auto_init_components(init_components: None) -> None:
"""Set up the component."""
async def test_entities_created(
hass: HomeAssistant,
mock_samsung_tv: MockSamsungTV,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test the media player entity is created through config entry setup."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_polling_updates_state(
hass: HomeAssistant,
mock_samsung_tv: MockSamsungTV,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the entity polls the TV on the scan interval and reflects changes."""
assert hass.states.get(ENTITY_ID).state == STATE_ON
off_state = _default_state()
off_state.power = False
mock_samsung_tv.refresh.side_effect = lambda: mock_samsung_tv.mock_state(off_state)
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
mock_samsung_tv.refresh.assert_awaited()
assert hass.states.get(ENTITY_ID).state == STATE_OFF
async def test_state_updates(
hass: HomeAssistant, mock_samsung_tv: MockSamsungTV
) -> None:
"""Test the entity updates from TV pushes and disconnects."""
assert hass.states.get(ENTITY_ID).state == STATE_ON
state = _default_state()
state.power = False
mock_samsung_tv.mock_state(state)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_OFF
mock_samsung_tv.mock_state(None)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
async def test_state_unknown_when_not_queried(
hass: HomeAssistant, mock_samsung_tv: MockSamsungTV
) -> None:
"""Test attributes are cleared when the TV reports no state."""
mock_samsung_tv.mock_state(TVState())
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_UNKNOWN
assert state.attributes.get(ATTR_MEDIA_VOLUME_LEVEL) is None
async def test_power_controls(
hass: HomeAssistant, mock_samsung_tv: MockSamsungTV
) -> None:
"""Test power services call the right methods."""
await hass.services.async_call(
MP_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
)
mock_samsung_tv.power_on.assert_awaited_once()
await hass.services.async_call(
MP_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
)
mock_samsung_tv.power_off.assert_awaited_once()
async def test_volume_controls(
hass: HomeAssistant, mock_samsung_tv: MockSamsungTV
) -> None:
"""Test volume state and controls."""
assert hass.states.get(ENTITY_ID).attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.2
await hass.services.async_call(
MP_DOMAIN,
SERVICE_VOLUME_SET,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_LEVEL: 0.5},
blocking=True,
)
assert mock_samsung_tv.set_volume.await_args == call(50)
# Volume up/down use the media player base class default step.
await hass.services.async_call(
MP_DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
)
assert mock_samsung_tv.set_volume.await_args == call(30)
await hass.services.async_call(
MP_DOMAIN, SERVICE_VOLUME_DOWN, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
)
assert mock_samsung_tv.set_volume.await_args == call(10)
async def test_mute_controls(
hass: HomeAssistant, mock_samsung_tv: MockSamsungTV
) -> None:
"""Test mute state and controls."""
assert hass.states.get(ENTITY_ID).attributes[ATTR_MEDIA_VOLUME_MUTED] is False
await hass.services.async_call(
MP_DOMAIN,
SERVICE_VOLUME_MUTE,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: True},
blocking=True,
)
assert mock_samsung_tv.set_mute.await_args == call(True)
await hass.services.async_call(
MP_DOMAIN,
SERVICE_VOLUME_MUTE,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: False},
blocking=True,
)
assert mock_samsung_tv.set_mute.await_args == call(False)
async def test_volume_features_depend_on_reported_state(
hass: HomeAssistant, mock_samsung_tv: MockSamsungTV
) -> None:
"""Test volume features drop when the TV does not report volume/mute.
A powered-off TV does not answer status queries, so volume and mute are
unknown; their controls and attributes are withheld until known.
"""
features = hass.states.get(ENTITY_ID).attributes[ATTR_SUPPORTED_FEATURES]
assert features & MediaPlayerEntityFeature.VOLUME_SET
assert features & MediaPlayerEntityFeature.VOLUME_STEP
assert features & MediaPlayerEntityFeature.VOLUME_MUTE
state = _default_state()
state.volume = None
state.mute = None
mock_samsung_tv.mock_state(state)
await hass.async_block_till_done()
entity_state = hass.states.get(ENTITY_ID)
features = entity_state.attributes[ATTR_SUPPORTED_FEATURES]
assert not features & MediaPlayerEntityFeature.VOLUME_SET
assert not features & MediaPlayerEntityFeature.VOLUME_STEP
assert not features & MediaPlayerEntityFeature.VOLUME_MUTE
assert features & MediaPlayerEntityFeature.SELECT_SOURCE
# The volume attributes are dropped along with the controls.
assert ATTR_MEDIA_VOLUME_LEVEL not in entity_state.attributes
assert ATTR_MEDIA_VOLUME_MUTED not in entity_state.attributes
async def test_no_source_or_volume_controls_when_off(
hass: HomeAssistant, mock_samsung_tv: MockSamsungTV
) -> None:
"""Test source/volume/mute controls are withheld while the TV is off.
A standby TV only accepts power-on over RS-232, so offering the other
controls would only produce failed commands.
"""
off_state = _default_state()
off_state.power = False
mock_samsung_tv.mock_state(off_state)
await hass.async_block_till_done()
entity_state = hass.states.get(ENTITY_ID)
features = entity_state.attributes[ATTR_SUPPORTED_FEATURES]
assert features & MediaPlayerEntityFeature.TURN_ON
assert features & MediaPlayerEntityFeature.TURN_OFF
assert not features & MediaPlayerEntityFeature.SELECT_SOURCE
assert not features & MediaPlayerEntityFeature.VOLUME_SET
assert not features & MediaPlayerEntityFeature.VOLUME_STEP
assert not features & MediaPlayerEntityFeature.VOLUME_MUTE
# Stale source/volume/mute attributes are cleared while off.
assert ATTR_INPUT_SOURCE not in entity_state.attributes
assert ATTR_MEDIA_VOLUME_LEVEL not in entity_state.attributes
assert ATTR_MEDIA_VOLUME_MUTED not in entity_state.attributes
async def test_source_state_and_controls(
hass: HomeAssistant, mock_samsung_tv: MockSamsungTV
) -> None:
"""Test source state and selection."""
entity_state = hass.states.get(ENTITY_ID)
assert entity_state.attributes[ATTR_INPUT_SOURCE] == "hdmi1"
source_list = entity_state.attributes[ATTR_INPUT_SOURCE_LIST]
assert "hdmi1" in source_list
assert "tv" in source_list
assert source_list == sorted(source_list)
state = _default_state()
state.input_source = InputSource.HDMI2
mock_samsung_tv.mock_state(state)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE] == "hdmi2"
await hass.services.async_call(
MP_DOMAIN,
SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "hdmi3"},
blocking=True,
)
assert mock_samsung_tv.select_input_source.await_args == call(InputSource.HDMI3)
@pytest.mark.parametrize(
"exception",
[CommandRejected("rejected"), ConnectionError("connection lost"), TimeoutError],
)
async def test_command_error_raises(
hass: HomeAssistant, mock_samsung_tv: MockSamsungTV, exception: Exception
) -> None:
"""Test library errors raised during an action surface as HomeAssistantError."""
mock_samsung_tv.power_on.side_effect = exception
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
MP_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
)
def test_input_source_translation_keys_cover_all_enum_members() -> None:
"""Test all input sources have a declared translation key."""
assert set(INPUT_SOURCE_SAMSUNG_TO_HA) == set(InputSource)
strings = load_json(Path("homeassistant/components/samsung_exlink/strings.json"))
assert set(INPUT_SOURCE_SAMSUNG_TO_HA.values()) == set(
strings["entity"]["media_player"]["tv"]["state_attributes"]["source"]["state"]
)