Compare commits

..

8 Commits

Author SHA1 Message Date
copilot-swe-agent[bot] afb90c91db Merge remote-tracking branch 'origin/dev' into repair_trigger_behavior
# Conflicts:
#	tests/helpers/test_trigger.py
2026-06-17 05:57:00 +00:00
epenet d9e2b49c0c Fix incorrect use of entity component constants in template (#172532) 2026-06-17 07:55:57 +02:00
renovate[bot] 4f9051464d Update cryptography to 48.0.1 (#174096) 2026-06-17 07:34:00 +02:00
Paulus Schoutsen 87894fd623 Activate venv before running python commands (#174093) 2026-06-17 07:32:22 +02:00
Franck Nijhof 34a70a9210 Clean up deprecated solar_rising entity from sun integration (#174079) 2026-06-17 06:44:16 +02:00
Paulus Schoutsen c9fb6a13fb Remove stale requirements_test_all.txt reference (#174095) 2026-06-17 05:08:20 +02:00
Erik 58f247afca Address review comments 2026-06-16 07:51:39 +02:00
Erik b9688b7fb2 Open repair issue when deprecated trigger behavior is used 2026-06-08 10:13:36 +02:00
35 changed files with 153 additions and 1401 deletions
-1
View File
@@ -496,7 +496,6 @@ 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
@@ -1555,8 +1555,6 @@ CLAUDE.md @home-assistant/core
/homeassistant/components/sabnzbd/ @shaiu @jpbede
/tests/components/sabnzbd/ @shaiu @jpbede
/homeassistant/components/saj/ @fredericvl
/homeassistant/components/samsung_exlink/ @balloob
/tests/components/samsung_exlink/ @balloob
/homeassistant/components/samsung_infrared/ @lmaertin
/tests/components/samsung_infrared/ @lmaertin
/homeassistant/components/samsungtv/ @chemelli74
+1 -7
View File
@@ -1,11 +1,5 @@
{
"domain": "samsung",
"name": "Samsung",
"integrations": [
"familyhub",
"samsung_exlink",
"samsung_infrared",
"samsungtv",
"syncthru"
]
"integrations": ["familyhub", "samsung_infrared", "samsungtv", "syncthru"]
}
@@ -135,6 +135,10 @@
"description": "The {integration_title} integration is being removed as it depends on system packages that can only be installed on systems running a deprecated architecture. To resolve this, remove the {domain} entry from your configuration.yaml file and restart Home Assistant.",
"title": "The {integration_title} integration is being removed"
},
"deprecated_trigger_behavior": {
"description": "An automation, script or template entity uses the trigger behavior option `{deprecated_behavior}`, which has been renamed to `{new_behavior}`. The old value still works for now, but support for it will be removed in a future release.\n\nTo fix this issue, edit the affected automations and scripts and change the behavior option from `behavior: {deprecated_behavior}` to `behavior: {new_behavior}`, then restart Home Assistant.",
"title": "Deprecated trigger behavior option in use"
},
"deprecated_yaml": {
"description": "Configuring {integration_title} using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the `{domain}` configuration from your configuration.yaml file and restart Home Assistant to fix this issue.",
"title": "The {integration_title} YAML configuration is being removed"
@@ -1,61 +0,0 @@
"""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
@@ -1,113 +0,0 @@
"""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")
@@ -1,12 +0,0 @@
"""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]
@@ -1,13 +0,0 @@
{
"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"]
}
@@ -1,196 +0,0 @@
"""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])
@@ -1,84 +0,0 @@
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
@@ -1,72 +0,0 @@
{
"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}"
}
}
}
+8 -1
View File
@@ -5,7 +5,7 @@ import logging
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
@@ -50,6 +50,13 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: SunConfigEntry) -> bool:
"""Set up from a config entry."""
# Remove deprecated solar_rising sensor entity (removed in 2026.1)
ent_reg = er.async_get(hass)
if entity_id := ent_reg.async_get_entity_id(
Platform.SENSOR, DOMAIN, f"{entry.entry_id}-solar_rising"
):
ent_reg.async_remove(entity_id)
sun = Sun(hass)
component = EntityComponent[Sun](_LOGGER, DOMAIN, hass)
await component.async_add_entities([sun])
+16 -10
View File
@@ -1,15 +1,12 @@
"""Support for Template fans."""
from enum import StrEnum
import logging
from typing import TYPE_CHECKING, Any
import voluptuous as vol
from homeassistant.components.fan import (
ATTR_DIRECTION,
ATTR_OSCILLATING,
ATTR_PERCENTAGE,
ATTR_PRESET_MODE,
DIRECTION_FORWARD,
DIRECTION_REVERSE,
DOMAIN as FAN_DOMAIN,
@@ -100,6 +97,15 @@ FAN_CONFIG_ENTRY_SCHEMA = FAN_COMMON_SCHEMA.extend(
)
class FanScriptVariable(StrEnum):
"""Variables for scripts."""
DIRECTION = "direction"
OSCILLATING = "oscillating"
PERCENTAGE = "percentage"
PRESET_MODE = "preset_mode"
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
@@ -235,8 +241,8 @@ class AbstractTemplateFan(AbstractTemplateEntity, FanEntity):
await self.async_run_script(
self._action_scripts[CONF_ON_ACTION],
run_variables={
ATTR_PERCENTAGE: percentage,
ATTR_PRESET_MODE: preset_mode,
FanScriptVariable.PERCENTAGE: percentage,
FanScriptVariable.PRESET_MODE: preset_mode,
},
context=self._context,
)
@@ -267,7 +273,7 @@ class AbstractTemplateFan(AbstractTemplateEntity, FanEntity):
if script := self._action_scripts.get(CONF_SET_PERCENTAGE_ACTION):
await self.async_run_script(
script,
run_variables={ATTR_PERCENTAGE: self._attr_percentage},
run_variables={FanScriptVariable.PERCENTAGE: self._attr_percentage},
context=self._context,
)
@@ -284,7 +290,7 @@ class AbstractTemplateFan(AbstractTemplateEntity, FanEntity):
if script := self._action_scripts.get(CONF_SET_PRESET_MODE_ACTION):
await self.async_run_script(
script,
run_variables={ATTR_PRESET_MODE: self._attr_preset_mode},
run_variables={FanScriptVariable.PRESET_MODE: self._attr_preset_mode},
context=self._context,
)
@@ -302,7 +308,7 @@ class AbstractTemplateFan(AbstractTemplateEntity, FanEntity):
) is not None:
await self.async_run_script(
script,
run_variables={ATTR_OSCILLATING: self.oscillating},
run_variables={FanScriptVariable.OSCILLATING: self.oscillating},
context=self._context,
)
@@ -318,7 +324,7 @@ class AbstractTemplateFan(AbstractTemplateEntity, FanEntity):
) is not None:
await self.async_run_script(
script,
run_variables={ATTR_DIRECTION: direction},
run_variables={FanScriptVariable.DIRECTION: direction},
context=self._context,
)
if CONF_DIRECTION not in self._templates:
+1 -2
View File
@@ -5,7 +5,6 @@ from typing import TYPE_CHECKING, Any
import voluptuous as vol
from homeassistant.components.number import (
ATTR_VALUE,
DEFAULT_MAX_VALUE,
DEFAULT_MIN_VALUE,
DEFAULT_STEP,
@@ -161,7 +160,7 @@ class AbstractTemplateNumber(AbstractTemplateEntity, NumberEntity):
if set_value := self._action_scripts.get(CONF_SET_VALUE):
await self.async_run_script(
set_value,
run_variables={ATTR_VALUE: value},
run_variables={"value": value},
context=self._context,
)
+3 -5
View File
@@ -6,8 +6,6 @@ from typing import TYPE_CHECKING, Any
import voluptuous as vol
from homeassistant.components.select import (
ATTR_OPTION,
ATTR_OPTIONS,
DOMAIN as SELECT_DOMAIN,
ENTITY_ID_FORMAT,
SelectEntity,
@@ -48,7 +46,7 @@ SCRIPT_FIELDS = (CONF_SELECT_OPTION,)
SELECT_COMMON_SCHEMA = vol.Schema(
{
vol.Required(ATTR_OPTIONS): cv.template,
vol.Required(CONF_OPTIONS): cv.template,
vol.Optional(CONF_SELECT_OPTION): cv.SCRIPT_SCHEMA,
vol.Optional(CONF_STATE): cv.template,
}
@@ -147,7 +145,7 @@ class AbstractTemplateSelect(AbstractTemplateEntity, SelectEntity):
if select_option := self._action_scripts.get(CONF_SELECT_OPTION):
await self.async_run_script(
select_option,
run_variables={ATTR_OPTION: option},
run_variables={"option": option},
context=self._context,
)
@@ -175,7 +173,7 @@ class TriggerSelectEntity(TriggerEntity, AbstractTemplateSelect):
"""Select entity based on trigger data."""
domain = SELECT_DOMAIN
extra_template_keys_complex = (ATTR_OPTIONS,)
extra_template_keys_complex = (CONF_OPTIONS,)
def __init__(
self,
+5 -5
View File
@@ -9,7 +9,6 @@ from typing import Any
import voluptuous as vol
from homeassistant.components.sensor import (
ATTR_LAST_RESET,
CONF_STATE_CLASS,
DEVICE_CLASSES_SCHEMA,
DOMAIN as SENSOR_DOMAIN,
@@ -50,13 +49,14 @@ from .schemas import (
from .template_entity import TemplateEntity
from .trigger_entity import TriggerEntity
CONF_LAST_RESET = "last_reset"
DEFAULT_NAME = "Template Sensor"
def validate_last_reset(val):
"""Run extra validation checks."""
if (
val.get(ATTR_LAST_RESET) is not None
val.get(CONF_LAST_RESET) is not None
and val.get(CONF_STATE_CLASS) != SensorStateClass.TOTAL
):
raise vol.Invalid(
@@ -78,7 +78,7 @@ SENSOR_COMMON_SCHEMA = vol.Schema(
SENSOR_YAML_SCHEMA = vol.All(
vol.Schema(
{
vol.Optional(ATTR_LAST_RESET): cv.template,
vol.Optional(CONF_LAST_RESET): cv.template,
}
)
.extend(SENSOR_COMMON_SCHEMA.schema)
@@ -204,10 +204,10 @@ class AbstractTemplateSensor(AbstractTemplateEntity, RestoreSensor):
self._validate_state,
)
self.setup_template(
ATTR_LAST_RESET,
CONF_LAST_RESET,
"_attr_last_reset",
validate_datetime(
self, ATTR_LAST_RESET, SensorDeviceClass.TIMESTAMP, require_tzinfo=False
self, CONF_LAST_RESET, SensorDeviceClass.TIMESTAMP, require_tzinfo=False
),
)
+1 -2
View File
@@ -7,7 +7,6 @@ from typing import TYPE_CHECKING, Any
import voluptuous as vol
from homeassistant.components.vacuum import (
ATTR_FAN_SPEED,
DOMAIN as VACUUM_DOMAIN,
SERVICE_CLEAN_SPOT,
SERVICE_LOCATE,
@@ -389,7 +388,7 @@ class AbstractTemplateVacuum(AbstractTemplateEntity, StateVacuumEntity):
if script := self._action_scripts.get(SERVICE_SET_FAN_SPEED):
await self.async_run_script(
script, run_variables={ATTR_FAN_SPEED: fan_speed}, context=self._context
script, run_variables={"fan_speed": fan_speed}, context=self._context
)
-1
View File
@@ -654,7 +654,6 @@ FLOWS = {
"ruuvitag_ble",
"rympro",
"sabnzbd",
"samsung_exlink",
"samsung_infrared",
"samsungtv",
"sanix",
@@ -6170,12 +6170,6 @@
"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,
+28
View File
@@ -44,11 +44,13 @@ from homeassistant.const import (
)
from homeassistant.core import (
CALLBACK_TYPE,
DOMAIN as HOMEASSISTANT_DOMAIN,
Context,
HassJob,
HassJobType,
HomeAssistant,
State,
async_get_hass_or_none,
callback,
get_hassjob_callable_job_type,
is_callback,
@@ -331,11 +333,37 @@ BEHAVIOR_ALL: Final = "all"
BEHAVIOR_EACH: Final = "each"
def _create_deprecated_behavior_issue(deprecated: str, replacement: str) -> None:
"""Inform the user a renamed trigger behavior value is still in use."""
# Returns None when called from the wrong thread or before hass is set up
# (e.g. a `check_config` run), in which case there's nothing to report to.
if (hass := async_get_hass_or_none()) is None:
return
from .issue_registry import IssueSeverity, async_create_issue # noqa: PLC0415
async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_trigger_behavior_{deprecated}",
breaks_in_ha_version="2027.1",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_trigger_behavior",
translation_placeholders={
"deprecated_behavior": deprecated,
"new_behavior": replacement,
},
)
def _backwards_compatible_behavior(value: Any) -> Any:
"""Convert legacy behavior values to new ones."""
if value == "any":
_create_deprecated_behavior_issue("any", BEHAVIOR_EACH)
return BEHAVIOR_EACH
if value == "last":
_create_deprecated_behavior_issue("last", BEHAVIOR_ALL)
return BEHAVIOR_ALL
return value
+1 -1
View File
@@ -29,7 +29,7 @@ cached-ipaddress==1.1.2
certifi>=2021.5.30
ciso8601==2.3.3
cronsim==2.7
cryptography==48.0.0
cryptography==48.0.1
dbus-fast==5.0.16
file-read-backwards==2.0.0
fnv-hash-fast==2.0.3
Generated
-10
View File
@@ -4717,16 +4717,6 @@ 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
+1 -1
View File
@@ -57,7 +57,7 @@ dependencies = [
"lru-dict==1.4.1",
"PyJWT==2.12.1",
# PyJWT has loose dependency. We want the latest one.
"cryptography==48.0.0",
"cryptography==48.0.1",
"Pillow==12.2.0",
"propcache==0.5.2",
"pyOpenSSL==26.2.0",
+1 -1
View File
@@ -21,7 +21,7 @@ bcrypt==5.0.0
certifi>=2021.5.30
ciso8601==2.3.3
cronsim==2.7
cryptography==48.0.0
cryptography==48.0.1
fnv-hash-fast==2.0.3
ha-ffmpeg==3.2.2
hass-nabucasa==2.2.0
-3
View File
@@ -2949,9 +2949,6 @@ 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
+4
View File
@@ -6,6 +6,10 @@ set -e
cd "$(realpath "$(dirname "$0")/..")"
if [ ! -n "$VIRTUAL_ENV" ]; then
source .venv/bin/activate
fi
echo "Installing development dependencies..."
uv pip install \
-e . \
+1 -1
View File
@@ -17,7 +17,7 @@ from script.hassfest.model import Config, Integration
# Requirements which can't be installed on all systems because they
# rely on additional system packages. Requirements listed in
# EXCLUDED_REQUIREMENTS_ALL will be commented-out in
# requirements_all.txt and requirements_test_all.txt.
# requirements_all.txt.
EXCLUDED_REQUIREMENTS_ALL = {
"atenpdu", # depends on pysnmp which is not maintained at this time
"avion",
@@ -1,4 +0,0 @@
"""Tests for the Samsung ExLink integration."""
MOCK_DEVICE = "/dev/ttyUSB0"
MOCK_MODEL = "frame_2022"
-106
View File
@@ -1,106 +0,0 @@
"""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()
@@ -1,103 +0,0 @@
# serializer version: 1
# name: test_entities_created[media_player.samsung_tv-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'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({
'device_class': 'tv',
'friendly_name': 'Samsung TV',
'is_volume_muted': False,
'source': 'hdmi1',
'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',
]),
'supported_features': <MediaPlayerEntityFeature: 3468>,
'volume_level': 0.2,
}),
'context': <ANY>,
'entity_id': 'media_player.samsung_tv',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
@@ -1,221 +0,0 @@
"""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"
@@ -1,65 +0,0 @@
"""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()
@@ -1,291 +0,0 @@
"""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"]
)
+30 -1
View File
@@ -10,8 +10,9 @@ import pytest
from homeassistant.components import sun
from homeassistant.components.sun import entity
from homeassistant.const import EVENT_STATE_CHANGED
from homeassistant.const import EVENT_STATE_CHANGED, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
@@ -245,3 +246,31 @@ async def test_setup_and_remove_config_entry(hass: HomeAssistant) -> None:
await hass.async_block_till_done()
assert hass.states.get(entity.ENTITY_ID) is None
async def test_cleanup_deprecated_solar_rising(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that the deprecated solar_rising entity is removed on setup."""
config_entry = MockConfigEntry(domain=sun.DOMAIN)
config_entry.add_to_hass(hass)
entity_registry.async_get_or_create(
Platform.SENSOR,
sun.DOMAIN,
unique_id=f"{config_entry.entry_id}-solar_rising",
config_entry=config_entry,
)
assert entity_registry.async_get_entity_id(
Platform.SENSOR, sun.DOMAIN, f"{config_entry.entry_id}-solar_rising"
)
now = datetime(2016, 6, 1, 8, 0, 0, tzinfo=dt_util.UTC)
with freeze_time(now):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert not entity_registry.async_get_entity_id(
Platform.SENSOR, sun.DOMAIN, f"{config_entry.entry_id}-solar_rising"
)
+48
View File
@@ -37,6 +37,7 @@ from homeassistant.const import (
)
from homeassistant.core import (
CALLBACK_TYPE,
DOMAIN as HOMEASSISTANT_DOMAIN,
Context,
Event,
EventStateChangedData,
@@ -49,6 +50,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import (
config_validation as cv,
entity_registry as er,
issue_registry as ir,
label_registry as lr,
trigger,
)
@@ -5699,6 +5701,52 @@ def test_entity_state_trigger_schema_behavior_backwards_compatible(
assert validated[CONF_OPTIONS][ATTR_BEHAVIOR] == expected
@pytest.mark.parametrize(
("behavior", "expected"),
[
("any", BEHAVIOR_EACH),
("last", BEHAVIOR_ALL),
],
)
async def test_entity_state_trigger_legacy_behavior_creates_repair_issue(
issue_registry: ir.IssueRegistry,
behavior: str,
expected: str,
) -> None:
"""Test a repair issue is raised when a legacy behavior value is used."""
config = {
CONF_TARGET: {CONF_ENTITY_ID: "test.entity"},
CONF_OPTIONS: {ATTR_BEHAVIOR: behavior},
}
validated = ENTITY_STATE_TRIGGER_SCHEMA_WITH_BEHAVIOR(config)
assert validated[CONF_OPTIONS][ATTR_BEHAVIOR] == expected
issue = issue_registry.async_get_issue(
HOMEASSISTANT_DOMAIN, f"deprecated_trigger_behavior_{behavior}"
)
assert issue is not None
assert issue.translation_key == "deprecated_trigger_behavior"
assert issue.translation_placeholders == {
"deprecated_behavior": behavior,
"new_behavior": expected,
}
@pytest.mark.parametrize("behavior", [BEHAVIOR_EACH, BEHAVIOR_FIRST, BEHAVIOR_ALL])
async def test_entity_state_trigger_new_behavior_no_repair_issue(
issue_registry: ir.IssueRegistry,
behavior: str,
) -> None:
"""Test no repair issue is raised when a current behavior value is used."""
config = {
CONF_TARGET: {CONF_ENTITY_ID: "test.entity"},
CONF_OPTIONS: {ATTR_BEHAVIOR: behavior},
}
ENTITY_STATE_TRIGGER_SCHEMA_WITH_BEHAVIOR(config)
assert not issue_registry.issues
def test_entity_state_trigger_schema_behavior_default() -> None:
"""Test the behavior defaults to 'each' when omitted."""
config = {