Move yaml configuration to integration key for command_line (#92824)

* Inital init commit

* bs and cover

* notify

* sensor

* switch

* Issues

* Finalize __init__

* First pass tests

* Fix Binary sensors

* Test cover

* Test notify

* Test sensor

* Tests switch

* Fix coverage

* Add codeowner

* Fix caplog

* test issue

* Flaky test notify

* Fix async

* Adjust yaml structure

* Change yaml format again

* Issue strings

* Fix tests

* string review comment
This commit is contained in:
G Johansson 2023-05-29 08:00:50 +02:00 committed by GitHub
parent 20d1a0fc77
commit 46e7486ce6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 1461 additions and 489 deletions

View file

@ -213,6 +213,8 @@ build.json @home-assistant/supervisor
/tests/components/color_extractor/ @GenericStudent /tests/components/color_extractor/ @GenericStudent
/homeassistant/components/comfoconnect/ @michaelarnauts /homeassistant/components/comfoconnect/ @michaelarnauts
/tests/components/comfoconnect/ @michaelarnauts /tests/components/comfoconnect/ @michaelarnauts
/homeassistant/components/command_line/ @gjohansson-ST
/tests/components/command_line/ @gjohansson-ST
/homeassistant/components/compensation/ @Petro31 /homeassistant/components/compensation/ @Petro31
/tests/components/compensation/ @Petro31 /tests/components/compensation/ @Petro31
/homeassistant/components/config/ @home-assistant/core /homeassistant/components/config/ @home-assistant/core

View file

@ -1 +1,177 @@
"""The command_line component.""" """The command_line component."""
from __future__ import annotations
import asyncio
from collections.abc import Coroutine
import logging
from typing import Any
import voluptuous as vol
from homeassistant.components.binary_sensor import (
DEVICE_CLASSES_SCHEMA as BINARY_SENSOR_DEVICE_CLASSES_SCHEMA,
DOMAIN as BINARY_SENSOR_DOMAIN,
)
from homeassistant.components.cover import DOMAIN as COVER_DOMAIN
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
from homeassistant.components.sensor import (
CONF_STATE_CLASS,
DEVICE_CLASSES_SCHEMA as SENSOR_DEVICE_CLASSES_SCHEMA,
DOMAIN as SENSOR_DOMAIN,
STATE_CLASSES_SCHEMA as SENSOR_STATE_CLASSES_SCHEMA,
)
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
CONF_COMMAND,
CONF_COMMAND_CLOSE,
CONF_COMMAND_OFF,
CONF_COMMAND_ON,
CONF_COMMAND_OPEN,
CONF_COMMAND_STATE,
CONF_COMMAND_STOP,
CONF_DEVICE_CLASS,
CONF_ICON,
CONF_NAME,
CONF_PAYLOAD_OFF,
CONF_PAYLOAD_ON,
CONF_UNIQUE_ID,
CONF_UNIT_OF_MEASUREMENT,
CONF_VALUE_TEMPLATE,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.reload import async_setup_reload_service
from homeassistant.helpers.typing import ConfigType
from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN
BINARY_SENSOR_DEFAULT_NAME = "Binary Command Sensor"
DEFAULT_PAYLOAD_ON = "ON"
DEFAULT_PAYLOAD_OFF = "OFF"
CONF_JSON_ATTRIBUTES = "json_attributes"
SENSOR_DEFAULT_NAME = "Command Sensor"
CONF_NOTIFIERS = "notifiers"
PLATFORM_MAPPING = {
BINARY_SENSOR_DOMAIN: Platform.BINARY_SENSOR,
COVER_DOMAIN: Platform.COVER,
NOTIFY_DOMAIN: Platform.NOTIFY,
SENSOR_DOMAIN: Platform.SENSOR,
SWITCH_DOMAIN: Platform.SWITCH,
}
_LOGGER = logging.getLogger(__name__)
BINARY_SENSOR_SCHEMA = vol.Schema(
{
vol.Required(CONF_COMMAND): cv.string,
vol.Optional(CONF_NAME, default=BINARY_SENSOR_DEFAULT_NAME): cv.string,
vol.Optional(CONF_PAYLOAD_OFF, default=DEFAULT_PAYLOAD_OFF): cv.string,
vol.Optional(CONF_PAYLOAD_ON, default=DEFAULT_PAYLOAD_ON): cv.string,
vol.Optional(CONF_DEVICE_CLASS): BINARY_SENSOR_DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
vol.Optional(CONF_COMMAND_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
vol.Optional(CONF_UNIQUE_ID): cv.string,
}
)
COVER_SCHEMA = vol.Schema(
{
vol.Optional(CONF_COMMAND_CLOSE, default="true"): cv.string,
vol.Optional(CONF_COMMAND_OPEN, default="true"): cv.string,
vol.Optional(CONF_COMMAND_STATE): cv.string,
vol.Optional(CONF_COMMAND_STOP, default="true"): cv.string,
vol.Required(CONF_NAME): cv.string,
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
vol.Optional(CONF_COMMAND_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
vol.Optional(CONF_UNIQUE_ID): cv.string,
}
)
NOTIFY_SCHEMA = vol.Schema(
{
vol.Required(CONF_COMMAND): cv.string,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_COMMAND_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
}
)
SENSOR_SCHEMA = vol.Schema(
{
vol.Required(CONF_COMMAND): cv.string,
vol.Optional(CONF_COMMAND_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
vol.Optional(CONF_JSON_ATTRIBUTES): cv.ensure_list_csv,
vol.Optional(CONF_NAME, default=SENSOR_DEFAULT_NAME): cv.string,
vol.Optional(CONF_UNIT_OF_MEASUREMENT): cv.string,
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
vol.Optional(CONF_UNIQUE_ID): cv.string,
vol.Optional(CONF_DEVICE_CLASS): SENSOR_DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_STATE_CLASS): SENSOR_STATE_CLASSES_SCHEMA,
}
)
SWITCH_SCHEMA = vol.Schema(
{
vol.Optional(CONF_COMMAND_OFF, default="true"): cv.string,
vol.Optional(CONF_COMMAND_ON, default="true"): cv.string,
vol.Optional(CONF_COMMAND_STATE): cv.string,
vol.Required(CONF_NAME): cv.string,
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
vol.Optional(CONF_ICON): cv.template,
vol.Optional(CONF_COMMAND_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
vol.Optional(CONF_UNIQUE_ID): cv.string,
}
)
COMBINED_SCHEMA = vol.Schema(
{
vol.Optional(BINARY_SENSOR_DOMAIN): BINARY_SENSOR_SCHEMA,
vol.Optional(COVER_DOMAIN): COVER_SCHEMA,
vol.Optional(NOTIFY_DOMAIN): NOTIFY_SCHEMA,
vol.Optional(SENSOR_DOMAIN): SENSOR_SCHEMA,
vol.Optional(SWITCH_DOMAIN): SWITCH_SCHEMA,
}
)
CONFIG_SCHEMA = vol.Schema(
{
vol.Optional(DOMAIN): vol.All(
cv.ensure_list,
[COMBINED_SCHEMA],
)
},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Command Line from yaml config."""
command_line_config: list[dict[str, dict[str, Any]]] = config.get(DOMAIN, {})
if not command_line_config:
return True
_LOGGER.debug("Full config loaded: %s", command_line_config)
load_coroutines: list[Coroutine[Any, Any, None]] = []
platforms: list[Platform] = []
for platform_config in command_line_config:
for platform, _config in platform_config.items():
platforms.append(PLATFORM_MAPPING[platform])
_LOGGER.debug(
"Loading config %s for platform %s",
platform_config,
PLATFORM_MAPPING[platform],
)
load_coroutines.append(
discovery.async_load_platform(
hass,
PLATFORM_MAPPING[platform],
DOMAIN,
_config,
config,
)
)
await async_setup_reload_service(hass, DOMAIN, platforms)
if load_coroutines:
_LOGGER.debug("Loading platforms: %s", platforms)
await asyncio.gather(*load_coroutines)
return True

View file

@ -7,6 +7,7 @@ import voluptuous as vol
from homeassistant.components.binary_sensor import ( from homeassistant.components.binary_sensor import (
DEVICE_CLASSES_SCHEMA, DEVICE_CLASSES_SCHEMA,
DOMAIN as BINARY_SENSOR_DOMAIN,
PLATFORM_SCHEMA, PLATFORM_SCHEMA,
BinarySensorDeviceClass, BinarySensorDeviceClass,
BinarySensorEntity, BinarySensorEntity,
@ -23,11 +24,11 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.reload import async_setup_reload_service from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.template import Template from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN, PLATFORMS from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN
from .sensor import CommandSensorData from .sensor import CommandSensorData
DEFAULT_NAME = "Binary Command Sensor" DEFAULT_NAME = "Binary Command Sensor"
@ -59,16 +60,30 @@ async def async_setup_platform(
) -> None: ) -> None:
"""Set up the Command line Binary Sensor.""" """Set up the Command line Binary Sensor."""
await async_setup_reload_service(hass, DOMAIN, PLATFORMS) if binary_sensor_config := config:
async_create_issue(
hass,
DOMAIN,
"deprecated_yaml_binary_sensor",
breaks_in_ha_version="2023.8.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_platform_yaml",
translation_placeholders={"platform": BINARY_SENSOR_DOMAIN},
)
if discovery_info:
binary_sensor_config = discovery_info
name: str = config.get(CONF_NAME, DEFAULT_NAME) name: str = binary_sensor_config.get(CONF_NAME, DEFAULT_NAME)
command: str = config[CONF_COMMAND] command: str = binary_sensor_config[CONF_COMMAND]
payload_off: str = config[CONF_PAYLOAD_OFF] payload_off: str = binary_sensor_config[CONF_PAYLOAD_OFF]
payload_on: str = config[CONF_PAYLOAD_ON] payload_on: str = binary_sensor_config[CONF_PAYLOAD_ON]
device_class: BinarySensorDeviceClass | None = config.get(CONF_DEVICE_CLASS) device_class: BinarySensorDeviceClass | None = binary_sensor_config.get(
value_template: Template | None = config.get(CONF_VALUE_TEMPLATE) CONF_DEVICE_CLASS
command_timeout: int = config[CONF_COMMAND_TIMEOUT] )
unique_id: str | None = config.get(CONF_UNIQUE_ID) value_template: Template | None = binary_sensor_config.get(CONF_VALUE_TEMPLATE)
command_timeout: int = binary_sensor_config[CONF_COMMAND_TIMEOUT]
unique_id: str | None = binary_sensor_config.get(CONF_UNIQUE_ID)
if value_template is not None: if value_template is not None:
value_template.hass = hass value_template.hass = hass
data = CommandSensorData(hass, command, command_timeout) data = CommandSensorData(hass, command, command_timeout)

View file

@ -6,7 +6,11 @@ from typing import TYPE_CHECKING, Any
import voluptuous as vol import voluptuous as vol
from homeassistant.components.cover import PLATFORM_SCHEMA, CoverEntity from homeassistant.components.cover import (
DOMAIN as COVER_DOMAIN,
PLATFORM_SCHEMA,
CoverEntity,
)
from homeassistant.const import ( from homeassistant.const import (
CONF_COMMAND_CLOSE, CONF_COMMAND_CLOSE,
CONF_COMMAND_OPEN, CONF_COMMAND_OPEN,
@ -14,17 +18,19 @@ from homeassistant.const import (
CONF_COMMAND_STOP, CONF_COMMAND_STOP,
CONF_COVERS, CONF_COVERS,
CONF_FRIENDLY_NAME, CONF_FRIENDLY_NAME,
CONF_NAME,
CONF_UNIQUE_ID, CONF_UNIQUE_ID,
CONF_VALUE_TEMPLATE, CONF_VALUE_TEMPLATE,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.reload import async_setup_reload_service from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.template import Template from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import slugify
from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN, PLATFORMS from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN
from .utils import call_shell_with_timeout, check_output_or_log from .utils import call_shell_with_timeout, check_output_or_log
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -55,19 +61,35 @@ async def async_setup_platform(
) -> None: ) -> None:
"""Set up cover controlled by shell commands.""" """Set up cover controlled by shell commands."""
await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
devices: dict[str, Any] = config.get(CONF_COVERS, {})
covers = [] covers = []
if discovery_info:
entities: dict[str, Any] = {slugify(discovery_info[CONF_NAME]): discovery_info}
else:
async_create_issue(
hass,
DOMAIN,
"deprecated_yaml_cover",
breaks_in_ha_version="2023.8.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_platform_yaml",
translation_placeholders={"platform": COVER_DOMAIN},
)
entities = config.get(CONF_COVERS, {})
for device_name, device_config in devices.items(): for device_name, device_config in entities.items():
value_template: Template | None = device_config.get(CONF_VALUE_TEMPLATE) value_template: Template | None = device_config.get(CONF_VALUE_TEMPLATE)
if value_template is not None: if value_template is not None:
value_template.hass = hass value_template.hass = hass
if name := device_config.get(
CONF_FRIENDLY_NAME
): # Backward compatibility. Can be removed after deprecation
device_config[CONF_NAME] = name
covers.append( covers.append(
CommandCover( CommandCover(
device_config.get(CONF_FRIENDLY_NAME, device_name), device_config.get(CONF_NAME, device_name),
device_config[CONF_COMMAND_OPEN], device_config[CONF_COMMAND_OPEN],
device_config[CONF_COMMAND_CLOSE], device_config[CONF_COMMAND_CLOSE],
device_config[CONF_COMMAND_STOP], device_config[CONF_COMMAND_STOP],

View file

@ -1,7 +1,7 @@
{ {
"domain": "command_line", "domain": "command_line",
"name": "Command Line", "name": "Command Line",
"codeowners": [], "codeowners": ["@gjohansson-ST"],
"documentation": "https://www.home-assistant.io/integrations/command_line", "documentation": "https://www.home-assistant.io/integrations/command_line",
"iot_class": "local_polling" "iot_class": "local_polling"
} }

View file

@ -7,14 +7,19 @@ from typing import Any
import voluptuous as vol import voluptuous as vol
from homeassistant.components.notify import PLATFORM_SCHEMA, BaseNotificationService from homeassistant.components.notify import (
DOMAIN as NOTIFY_DOMAIN,
PLATFORM_SCHEMA,
BaseNotificationService,
)
from homeassistant.const import CONF_COMMAND, CONF_NAME from homeassistant.const import CONF_COMMAND, CONF_NAME
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.issue_registry import IssueSeverity, create_issue
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util.process import kill_subprocess from homeassistant.util.process import kill_subprocess
from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -33,8 +38,21 @@ def get_service(
discovery_info: DiscoveryInfoType | None = None, discovery_info: DiscoveryInfoType | None = None,
) -> CommandLineNotificationService: ) -> CommandLineNotificationService:
"""Get the Command Line notification service.""" """Get the Command Line notification service."""
command: str = config[CONF_COMMAND] if notify_config := config:
timeout: int = config[CONF_COMMAND_TIMEOUT] create_issue(
hass,
DOMAIN,
"deprecated_yaml_notify",
breaks_in_ha_version="2023.8.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_platform_yaml",
translation_placeholders={"platform": NOTIFY_DOMAIN},
)
if discovery_info:
notify_config = discovery_info
command: str = notify_config[CONF_COMMAND]
timeout: int = notify_config[CONF_COMMAND_TIMEOUT]
return CommandLineNotificationService(command, timeout) return CommandLineNotificationService(command, timeout)

View file

@ -11,6 +11,7 @@ import voluptuous as vol
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
CONF_STATE_CLASS, CONF_STATE_CLASS,
DEVICE_CLASSES_SCHEMA, DEVICE_CLASSES_SCHEMA,
DOMAIN as SENSOR_DOMAIN,
PLATFORM_SCHEMA, PLATFORM_SCHEMA,
STATE_CLASSES_SCHEMA, STATE_CLASSES_SCHEMA,
SensorEntity, SensorEntity,
@ -27,11 +28,11 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import TemplateError from homeassistant.exceptions import TemplateError
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.reload import async_setup_reload_service from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.template import Template from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN, PLATFORMS from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN
from .utils import check_output_or_log from .utils import check_output_or_log
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -64,18 +65,29 @@ async def async_setup_platform(
discovery_info: DiscoveryInfoType | None = None, discovery_info: DiscoveryInfoType | None = None,
) -> None: ) -> None:
"""Set up the Command Sensor.""" """Set up the Command Sensor."""
if sensor_config := config:
async_create_issue(
hass,
DOMAIN,
"deprecated_yaml_sensor",
breaks_in_ha_version="2023.8.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_platform_yaml",
translation_placeholders={"platform": SENSOR_DOMAIN},
)
if discovery_info:
sensor_config = discovery_info
await async_setup_reload_service(hass, DOMAIN, PLATFORMS) name: str = sensor_config[CONF_NAME]
command: str = sensor_config[CONF_COMMAND]
name: str = config[CONF_NAME] unit: str | None = sensor_config.get(CONF_UNIT_OF_MEASUREMENT)
command: str = config[CONF_COMMAND] value_template: Template | None = sensor_config.get(CONF_VALUE_TEMPLATE)
unit: str | None = config.get(CONF_UNIT_OF_MEASUREMENT) command_timeout: int = sensor_config[CONF_COMMAND_TIMEOUT]
value_template: Template | None = config.get(CONF_VALUE_TEMPLATE) unique_id: str | None = sensor_config.get(CONF_UNIQUE_ID)
command_timeout: int = config[CONF_COMMAND_TIMEOUT]
unique_id: str | None = config.get(CONF_UNIQUE_ID)
if value_template is not None: if value_template is not None:
value_template.hass = hass value_template.hass = hass
json_attributes: list[str] | None = config.get(CONF_JSON_ATTRIBUTES) json_attributes: list[str] | None = sensor_config.get(CONF_JSON_ATTRIBUTES)
data = CommandSensorData(hass, command, command_timeout) data = CommandSensorData(hass, command, command_timeout)
async_add_entities( async_add_entities(

View file

@ -0,0 +1,8 @@
{
"issues": {
"deprecated_platform_yaml": {
"title": "Command Line YAML configuration has been deprecated",
"description": "Configuring Command Line `{platform}` using YAML has been deprecated.\n\nConsult the documentation to move your YAML configuration to integration key and restart Home Assistant to resolve this issue."
}
}
}

View file

@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Any
import voluptuous as vol import voluptuous as vol
from homeassistant.components.switch import ( from homeassistant.components.switch import (
DOMAIN as SWITCH_DOMAIN,
ENTITY_ID_FORMAT, ENTITY_ID_FORMAT,
PLATFORM_SCHEMA, PLATFORM_SCHEMA,
SwitchEntity, SwitchEntity,
@ -26,12 +27,13 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.reload import async_setup_reload_service from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.template import Template from homeassistant.helpers.template import Template
from homeassistant.helpers.template_entity import ManualTriggerEntity from homeassistant.helpers.template_entity import ManualTriggerEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import slugify
from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN, PLATFORMS from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN
from .utils import call_shell_with_timeout, check_output_or_log from .utils import call_shell_with_timeout, check_output_or_log
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -62,16 +64,38 @@ async def async_setup_platform(
) -> None: ) -> None:
"""Find and return switches controlled by shell commands.""" """Find and return switches controlled by shell commands."""
await async_setup_reload_service(hass, DOMAIN, PLATFORMS) if discovery_info:
entities: dict[str, Any] = {slugify(discovery_info[CONF_NAME]): discovery_info}
else:
async_create_issue(
hass,
DOMAIN,
"deprecated_yaml_switch",
breaks_in_ha_version="2023.8.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_platform_yaml",
translation_placeholders={"platform": SWITCH_DOMAIN},
)
entities = config.get(CONF_SWITCHES, {})
devices: dict[str, Any] = config.get(CONF_SWITCHES, {})
switches = [] switches = []
for object_id, device_config in devices.items(): for object_id, device_config in entities.items():
if name := device_config.get(
CONF_FRIENDLY_NAME
): # Backward compatibility. Can be removed after deprecation
device_config[CONF_NAME] = name
if icon := device_config.get(
CONF_ICON_TEMPLATE
): # Backward compatibility. Can be removed after deprecation
device_config[CONF_ICON] = icon
trigger_entity_config = { trigger_entity_config = {
CONF_UNIQUE_ID: device_config.get(CONF_UNIQUE_ID), CONF_UNIQUE_ID: device_config.get(CONF_UNIQUE_ID),
CONF_NAME: Template(device_config.get(CONF_FRIENDLY_NAME, object_id), hass), CONF_NAME: Template(device_config.get(CONF_NAME, object_id), hass),
CONF_ICON: device_config.get(CONF_ICON_TEMPLATE), CONF_ICON: device_config.get(CONF_ICON),
} }
value_template: Template | None = device_config.get(CONF_VALUE_TEMPLATE) value_template: Template | None = device_config.get(CONF_VALUE_TEMPLATE)

View file

@ -0,0 +1,72 @@
"""Fixtures for command_line."""
from typing import Any
import pytest
from homeassistant import setup
from homeassistant.components.command_line.const import DOMAIN
from homeassistant.core import HomeAssistant
@pytest.fixture(name="get_config")
async def get_config_to_integration_load() -> dict[str, Any]:
"""Return default minimal configuration.
To override the config, tests can be marked with:
@pytest.mark.parametrize("get_config", [{...}])
"""
return {
"command_line": [
{
"binary_sensor": {
"name": "Test",
"command": "echo 1",
"payload_on": "1",
"payload_off": "0",
"command_timeout": 15,
}
},
{
"cover": {
"name": "Test",
"command_state": "echo 1",
"command_timeout": 15,
}
},
{
"notify": {
"name": "Test",
"command": "echo 1",
"command_timeout": 15,
}
},
{
"sensor": {
"name": "Test",
"command": "echo 5",
"unit_of_measurement": "in",
"command_timeout": 15,
}
},
{
"switch": {
"name": "Test",
"command_state": "echo 1",
"command_timeout": 15,
}
},
]
}
@pytest.fixture(name="load_yaml_integration")
async def load_int(hass: HomeAssistant, get_config: dict[str, Any]) -> None:
"""Set up the Command Line integration in Home Assistant."""
await setup.async_setup_component(
hass,
DOMAIN,
get_config,
)
await hass.async_block_till_done()

View file

@ -6,32 +6,63 @@ from typing import Any
import pytest import pytest
from homeassistant import setup from homeassistant import setup
from homeassistant.components.binary_sensor import DOMAIN from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.command_line.const import DOMAIN
from homeassistant.const import STATE_OFF, STATE_ON from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
import homeassistant.helpers.issue_registry as ir
async def setup_test_entity(hass: HomeAssistant, config_dict: dict[str, Any]) -> None: async def test_setup_platform_yaml(hass: HomeAssistant) -> None:
"""Set up a test command line binary_sensor entity.""" """Test sensor setup."""
assert await setup.async_setup_component( assert await setup.async_setup_component(
hass, hass,
DOMAIN, BINARY_SENSOR_DOMAIN,
{DOMAIN: {"platform": "command_line", "name": "Test", **config_dict}}, {
BINARY_SENSOR_DOMAIN: {
"platform": "command_line",
"name": "Test",
"command": "echo 1",
"payload_on": "1",
"payload_off": "0",
}
},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
entity_state = hass.states.get("binary_sensor.test")
assert entity_state
assert entity_state.state == STATE_ON
assert entity_state.name == "Test"
async def test_setup(hass: HomeAssistant) -> None: issue_registry = ir.async_get(hass)
"""Test sensor setup.""" issue = issue_registry.async_get_issue(DOMAIN, "deprecated_yaml_binary_sensor")
await setup_test_entity( assert issue.translation_key == "deprecated_platform_yaml"
hass,
@pytest.mark.parametrize(
"get_config",
[
{ {
"command": "echo 1", "command_line": [
"payload_on": "1", {
"payload_off": "0", "binary_sensor": {
}, "name": "Test",
) "command": "echo 1",
"payload_on": "1",
"payload_off": "0",
"command_timeout": 15,
}
}
]
}
],
)
async def test_setup_integration_yaml(
hass: HomeAssistant, load_yaml_integration: None
) -> None:
"""Test sensor setup."""
entity_state = hass.states.get("binary_sensor.test") entity_state = hass.states.get("binary_sensor.test")
assert entity_state assert entity_state
@ -39,67 +70,88 @@ async def test_setup(hass: HomeAssistant) -> None:
assert entity_state.name == "Test" assert entity_state.name == "Test"
async def test_template(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Test setting the state with a template.""" "get_config",
[
await setup_test_entity(
hass,
{ {
"command": "echo 10", "command_line": [
"payload_on": "1.0", {
"payload_off": "0", "binary_sensor": {
"value_template": "{{ value | multiply(0.1) }}", "name": "Test",
}, "command": "echo 10",
) "payload_on": "1.0",
"payload_off": "0",
"value_template": "{{ value | multiply(0.1) }}",
}
}
]
}
],
)
async def test_template(hass: HomeAssistant, load_yaml_integration: None) -> None:
"""Test setting the state with a template."""
entity_state = hass.states.get("binary_sensor.test") entity_state = hass.states.get("binary_sensor.test")
assert entity_state assert entity_state
assert entity_state.state == STATE_ON assert entity_state.state == STATE_ON
async def test_sensor_off(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Test setting the state with a template.""" "get_config",
await setup_test_entity( [
hass,
{ {
"command": "echo 0", "command_line": [
"payload_on": "1", {
"payload_off": "0", "binary_sensor": {
}, "name": "Test",
) "command": "echo 0",
"payload_on": "1",
"payload_off": "0",
}
}
]
}
],
)
async def test_sensor_off(hass: HomeAssistant, load_yaml_integration: None) -> None:
"""Test setting the state with a template."""
entity_state = hass.states.get("binary_sensor.test") entity_state = hass.states.get("binary_sensor.test")
assert entity_state assert entity_state
assert entity_state.state == STATE_OFF assert entity_state.state == STATE_OFF
async def test_unique_id( @pytest.mark.parametrize(
hass: HomeAssistant, entity_registry: er.EntityRegistry "get_config",
) -> None: [
"""Test unique_id option and if it only creates one binary sensor per id."""
assert await setup.async_setup_component(
hass,
DOMAIN,
{ {
DOMAIN: [ "command_line": [
{ {
"platform": "command_line", "binary_sensor": {
"unique_id": "unique", "unique_id": "unique",
"command": "echo 0", "command": "echo 0",
}
}, },
{ {
"platform": "command_line", "binary_sensor": {
"unique_id": "not-so-unique-anymore", "unique_id": "not-so-unique-anymore",
"command": "echo 1", "command": "echo 1",
}
}, },
{ {
"platform": "command_line", "binary_sensor": {
"unique_id": "not-so-unique-anymore", "unique_id": "not-so-unique-anymore",
"command": "echo 2", "command": "echo 2",
}
}, },
] ]
}, }
) ],
await hass.async_block_till_done() )
async def test_unique_id(
hass: HomeAssistant, entity_registry: er.EntityRegistry, load_yaml_integration: None
) -> None:
"""Test unique_id option and if it only creates one binary sensor per id."""
assert len(hass.states.async_all()) == 2 assert len(hass.states.async_all()) == 2
@ -112,14 +164,28 @@ async def test_unique_id(
) )
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"binary_sensor": {
"command": "exit 33",
}
}
]
}
],
)
async def test_return_code( async def test_return_code(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant hass: HomeAssistant, caplog: pytest.LogCaptureFixture, get_config: dict[str, Any]
) -> None: ) -> None:
"""Test setting the state with a template.""" """Test setting the state with a template."""
await setup_test_entity( await setup.async_setup_component(
hass, hass,
{ DOMAIN,
"command": "exit 33", get_config,
},
) )
await hass.async_block_till_done()
assert "return code 33" in caplog.text assert "return code 33" in caplog.text

View file

@ -3,13 +3,13 @@ from __future__ import annotations
import os import os
import tempfile import tempfile
from typing import Any
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from homeassistant import config as hass_config, setup from homeassistant import config as hass_config, setup
from homeassistant.components.cover import DOMAIN, SCAN_INTERVAL from homeassistant.components.command_line import DOMAIN
from homeassistant.components.cover import DOMAIN as COVER_DOMAIN, SCAN_INTERVAL
from homeassistant.const import ( from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_ENTITY_ID,
SERVICE_CLOSE_COVER, SERVICE_CLOSE_COVER,
@ -19,35 +19,79 @@ from homeassistant.const import (
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
import homeassistant.helpers.issue_registry as ir
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from tests.common import async_fire_time_changed, get_fixture_path from tests.common import async_fire_time_changed, get_fixture_path
async def setup_test_entity(hass: HomeAssistant, config_dict: dict[str, Any]) -> None: async def test_no_covers_platform_yaml(
"""Set up a test command line notify service.""" caplog: pytest.LogCaptureFixture, hass: HomeAssistant
assert await setup.async_setup_component( ) -> None:
hass,
DOMAIN,
{
DOMAIN: [
{"platform": "command_line", "covers": config_dict},
]
},
)
await hass.async_block_till_done()
async def test_no_covers(caplog: pytest.LogCaptureFixture, hass: HomeAssistant) -> None:
"""Test that the cover does not polls when there's no state command.""" """Test that the cover does not polls when there's no state command."""
with patch( with patch(
"homeassistant.components.command_line.utils.subprocess.check_output", "homeassistant.components.command_line.utils.subprocess.check_output",
return_value=b"50\n", return_value=b"50\n",
): ):
await setup_test_entity(hass, {}) assert await setup.async_setup_component(
hass,
COVER_DOMAIN,
{
COVER_DOMAIN: [
{"platform": "command_line", "covers": {}},
]
},
)
await hass.async_block_till_done()
assert "No covers added" in caplog.text assert "No covers added" in caplog.text
issue_registry = ir.async_get(hass)
issue = issue_registry.async_get_issue(DOMAIN, "deprecated_yaml_cover")
assert issue.translation_key == "deprecated_platform_yaml"
async def test_state_value_platform_yaml(hass: HomeAssistant) -> None:
"""Test with state value."""
with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, "cover_status")
assert await setup.async_setup_component(
hass,
COVER_DOMAIN,
{
COVER_DOMAIN: [
{
"platform": "command_line",
"covers": {
"test": {
"command_state": f"cat {path}",
"command_open": f"echo 1 > {path}",
"command_close": f"echo 1 > {path}",
"command_stop": f"echo 0 > {path}",
"value_template": "{{ value }}",
"friendly_name": "Test",
},
},
},
]
},
)
await hass.async_block_till_done()
entity_state = hass.states.get("cover.test")
assert entity_state
assert entity_state.state == "unknown"
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: "cover.test"},
blocking=True,
)
entity_state = hass.states.get("cover.test")
assert entity_state
assert entity_state.state == "open"
async def test_no_poll_when_cover_has_no_command_state(hass: HomeAssistant) -> None: async def test_no_poll_when_cover_has_no_command_state(hass: HomeAssistant) -> None:
"""Test that the cover does not polls when there's no state command.""" """Test that the cover does not polls when there's no state command."""
@ -56,20 +100,44 @@ async def test_no_poll_when_cover_has_no_command_state(hass: HomeAssistant) -> N
"homeassistant.components.command_line.utils.subprocess.check_output", "homeassistant.components.command_line.utils.subprocess.check_output",
return_value=b"50\n", return_value=b"50\n",
) as check_output: ) as check_output:
await setup_test_entity(hass, {"test": {}}) assert await setup.async_setup_component(
hass,
COVER_DOMAIN,
{
COVER_DOMAIN: [
{"platform": "command_line", "covers": {"test": {}}},
]
},
)
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
assert not check_output.called assert not check_output.called
async def test_poll_when_cover_has_command_state(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"cover": {
"command_state": "echo state",
"name": "Test",
},
}
]
}
],
)
async def test_poll_when_cover_has_command_state(
hass: HomeAssistant, load_yaml_integration: None
) -> None:
"""Test that the cover polls when there's a state command.""" """Test that the cover polls when there's a state command."""
with patch( with patch(
"homeassistant.components.command_line.utils.subprocess.check_output", "homeassistant.components.command_line.utils.subprocess.check_output",
return_value=b"50\n", return_value=b"50\n",
) as check_output: ) as check_output:
await setup_test_entity(hass, {"test": {"command_state": "echo state"}})
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
check_output.assert_called_once_with( check_output.assert_called_once_with(
@ -84,57 +152,80 @@ async def test_state_value(hass: HomeAssistant) -> None:
"""Test with state value.""" """Test with state value."""
with tempfile.TemporaryDirectory() as tempdirname: with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, "cover_status") path = os.path.join(tempdirname, "cover_status")
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_state": f"cat {path}", {
"command_open": f"echo 1 > {path}", "cover": {
"command_close": f"echo 1 > {path}", "command_state": f"cat {path}",
"command_stop": f"echo 0 > {path}", "command_open": f"echo 1 > {path}",
"value_template": "{{ value }}", "command_close": f"echo 1 > {path}",
} "command_stop": f"echo 0 > {path}",
"value_template": "{{ value }}",
"name": "Test",
}
}
]
}, },
) )
await hass.async_block_till_done()
entity_state = hass.states.get("cover.test") entity_state = hass.states.get("cover.test")
assert entity_state assert entity_state
assert entity_state.state == "unknown" assert entity_state.state == "unknown"
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SERVICE_OPEN_COVER, {ATTR_ENTITY_ID: "cover.test"}, blocking=True COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: "cover.test"},
blocking=True,
) )
entity_state = hass.states.get("cover.test") entity_state = hass.states.get("cover.test")
assert entity_state assert entity_state
assert entity_state.state == "open" assert entity_state.state == "open"
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SERVICE_CLOSE_COVER, {ATTR_ENTITY_ID: "cover.test"}, blocking=True COVER_DOMAIN,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: "cover.test"},
blocking=True,
) )
entity_state = hass.states.get("cover.test") entity_state = hass.states.get("cover.test")
assert entity_state assert entity_state
assert entity_state.state == "open" assert entity_state.state == "open"
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SERVICE_STOP_COVER, {ATTR_ENTITY_ID: "cover.test"}, blocking=True COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: "cover.test"},
blocking=True,
) )
entity_state = hass.states.get("cover.test") entity_state = hass.states.get("cover.test")
assert entity_state assert entity_state
assert entity_state.state == "closed" assert entity_state.state == "closed"
async def test_reload(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"cover": {
"command_state": "echo open",
"value_template": "{{ value }}",
"name": "Test",
}
}
]
}
],
)
async def test_reload(hass: HomeAssistant, load_yaml_integration: None) -> None:
"""Verify we can reload command_line covers.""" """Verify we can reload command_line covers."""
await setup_test_entity(
hass,
{
"test": {
"command_state": "echo open",
"value_template": "{{ value }}",
}
},
)
entity_state = hass.states.get("cover.test") entity_state = hass.states.get("cover.test")
assert entity_state assert entity_state
assert entity_state.state == "unknown" assert entity_state.state == "unknown"
@ -155,50 +246,73 @@ async def test_reload(hass: HomeAssistant) -> None:
assert hass.states.get("cover.from_yaml") assert hass.states.get("cover.from_yaml")
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"cover": {
"command_open": "exit 1",
"name": "Test",
}
}
]
}
],
)
async def test_move_cover_failure( async def test_move_cover_failure(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, load_yaml_integration: None
) -> None: ) -> None:
"""Test command failure.""" """Test command failure."""
await setup_test_entity(
hass,
{"test": {"command_open": "exit 1"}},
)
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SERVICE_OPEN_COVER, {ATTR_ENTITY_ID: "cover.test"}, blocking=True COVER_DOMAIN, SERVICE_OPEN_COVER, {ATTR_ENTITY_ID: "cover.test"}, blocking=True
) )
assert "Command failed" in caplog.text assert "Command failed" in caplog.text
assert "return code 1" in caplog.text assert "return code 1" in caplog.text
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"cover": {
"command_open": "echo open",
"command_close": "echo close",
"command_stop": "echo stop",
"unique_id": "unique",
"name": "Test",
}
},
{
"cover": {
"command_open": "echo open",
"command_close": "echo close",
"command_stop": "echo stop",
"unique_id": "not-so-unique-anymore",
"name": "Test2",
}
},
{
"cover": {
"command_open": "echo open",
"command_close": "echo close",
"command_stop": "echo stop",
"unique_id": "not-so-unique-anymore",
"name": "Test3",
}
},
]
}
],
)
async def test_unique_id( async def test_unique_id(
hass: HomeAssistant, entity_registry: er.EntityRegistry hass: HomeAssistant, entity_registry: er.EntityRegistry, load_yaml_integration: None
) -> None: ) -> None:
"""Test unique_id option and if it only creates one cover per id.""" """Test unique_id option and if it only creates one cover per id."""
await setup_test_entity(
hass,
{
"unique": {
"command_open": "echo open",
"command_close": "echo close",
"command_stop": "echo stop",
"unique_id": "unique",
},
"not_unique_1": {
"command_open": "echo open",
"command_close": "echo close",
"command_stop": "echo stop",
"unique_id": "not-so-unique-anymore",
},
"not_unique_2": {
"command_open": "echo open",
"command_close": "echo close",
"command_stop": "echo stop",
"unique_id": "not-so-unique-anymore",
},
},
)
assert len(hass.states.async_all()) == 2 assert len(hass.states.async_all()) == 2
assert len(entity_registry.entities) == 2 assert len(entity_registry.entities) == 2

View file

@ -0,0 +1,27 @@
"""Test Command line component setup process."""
from __future__ import annotations
from datetime import timedelta
from homeassistant.const import STATE_ON, STATE_OPEN
from homeassistant.core import HomeAssistant
import homeassistant.util.dt as dt_util
from tests.common import async_fire_time_changed
async def test_setup_config(hass: HomeAssistant, load_yaml_integration: None) -> None:
"""Test setup from yaml."""
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=10))
await hass.async_block_till_done()
state_binary_sensor = hass.states.get("binary_sensor.test")
state_sensor = hass.states.get("sensor.test")
state_cover = hass.states.get("cover.test")
state_switch = hass.states.get("switch.test")
assert state_binary_sensor.state == STATE_ON
assert state_sensor.state == "5"
assert state_cover.state == STATE_OPEN
assert state_switch.state == STATE_ON

View file

@ -4,40 +4,71 @@ from __future__ import annotations
import os import os
import subprocess import subprocess
import tempfile import tempfile
from typing import Any
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from homeassistant import setup from homeassistant import setup
from homeassistant.components.notify import DOMAIN from homeassistant.components.command_line import DOMAIN
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
import homeassistant.helpers.issue_registry as ir
async def setup_test_service(hass: HomeAssistant, config_dict: dict[str, Any]) -> None: async def test_setup_platform_yaml(hass: HomeAssistant) -> None:
"""Set up a test command line notify service.""" """Test sensor setup."""
assert await setup.async_setup_component( assert await setup.async_setup_component(
hass, hass,
DOMAIN, NOTIFY_DOMAIN,
{ {
DOMAIN: [ NOTIFY_DOMAIN: [
{"platform": "command_line", "name": "Test", **config_dict}, {"platform": "command_line", "name": "Test1", "command": "exit 0"},
] ]
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.services.has_service(NOTIFY_DOMAIN, "test1")
issue_registry = ir.async_get(hass)
issue = issue_registry.async_get_issue(DOMAIN, "deprecated_yaml_notify")
assert issue.translation_key == "deprecated_platform_yaml"
async def test_setup(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"notify": {
"command": "exit 0",
"name": "Test2",
}
}
]
}
],
)
async def test_setup_integration_yaml(
hass: HomeAssistant, load_yaml_integration: None
) -> None:
"""Test sensor setup.""" """Test sensor setup."""
await setup_test_service(hass, {"command": "exit 0"}) assert hass.services.has_service(NOTIFY_DOMAIN, "test2")
assert hass.services.has_service(DOMAIN, "test")
async def test_bad_config(hass: HomeAssistant) -> None: async def test_bad_config(hass: HomeAssistant) -> None:
"""Test set up the platform with bad/missing configuration.""" """Test set up the platform with bad/missing configuration."""
await setup_test_service(hass, {}) assert await setup.async_setup_component(
assert not hass.services.has_service(DOMAIN, "test") hass,
NOTIFY_DOMAIN,
{
NOTIFY_DOMAIN: [
{"platform": "command_line"},
]
},
)
await hass.async_block_till_done()
assert not hass.services.has_service(NOTIFY_DOMAIN, "test")
async def test_command_line_output(hass: HomeAssistant) -> None: async def test_command_line_output(hass: HomeAssistant) -> None:
@ -45,58 +76,102 @@ async def test_command_line_output(hass: HomeAssistant) -> None:
with tempfile.TemporaryDirectory() as tempdirname: with tempfile.TemporaryDirectory() as tempdirname:
filename = os.path.join(tempdirname, "message.txt") filename = os.path.join(tempdirname, "message.txt")
message = "one, two, testing, testing" message = "one, two, testing, testing"
await setup_test_service( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"command": f"cat > {filename}", "command_line": [
{
"notify": {
"command": f"cat > {filename}",
"name": "Test3",
}
}
]
}, },
) )
await hass.async_block_till_done()
assert hass.services.has_service(DOMAIN, "test") assert hass.services.has_service(NOTIFY_DOMAIN, "test3")
assert await hass.services.async_call( assert await hass.services.async_call(
DOMAIN, "test", {"message": message}, blocking=True NOTIFY_DOMAIN, "test3", {"message": message}, blocking=True
) )
with open(filename) as handle: with open(filename, encoding="UTF-8") as handle:
# the echo command adds a line break # the echo command adds a line break
assert message == handle.read() assert message == handle.read()
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"notify": {
"command": "exit 1",
"name": "Test4",
}
}
]
}
],
)
async def test_error_for_none_zero_exit_code( async def test_error_for_none_zero_exit_code(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, load_yaml_integration: None
) -> None: ) -> None:
"""Test if an error is logged for non zero exit codes.""" """Test if an error is logged for non zero exit codes."""
await setup_test_service(
hass,
{
"command": "exit 1",
},
)
assert await hass.services.async_call( assert await hass.services.async_call(
DOMAIN, "test", {"message": "error"}, blocking=True NOTIFY_DOMAIN, "test4", {"message": "error"}, blocking=True
) )
assert "Command failed" in caplog.text assert "Command failed" in caplog.text
assert "return code 1" in caplog.text assert "return code 1" in caplog.text
async def test_timeout(caplog: pytest.LogCaptureFixture, hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Test blocking is not forever.""" "get_config",
await setup_test_service( [
hass,
{ {
"command": "sleep 10000", "command_line": [
"command_timeout": 0.0000001, {
}, "notify": {
) "command": "sleep 10000",
"command_timeout": 0.0000001,
"name": "Test5",
}
}
]
}
],
)
async def test_timeout(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant, load_yaml_integration: None
) -> None:
"""Test blocking is not forever."""
assert await hass.services.async_call( assert await hass.services.async_call(
DOMAIN, "test", {"message": "error"}, blocking=True NOTIFY_DOMAIN, "test5", {"message": "error"}, blocking=True
) )
assert "Timeout" in caplog.text assert "Timeout" in caplog.text
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"notify": {
"command": "exit 0",
"name": "Test6",
}
}
]
}
],
)
async def test_subprocess_exceptions( async def test_subprocess_exceptions(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, load_yaml_integration: None
) -> None: ) -> None:
"""Test that notify subprocess exceptions are handled correctly.""" """Test that notify subprocess exceptions are handled correctly."""
@ -110,15 +185,14 @@ async def test_subprocess_exceptions(
subprocess.SubprocessError(), subprocess.SubprocessError(),
] ]
await setup_test_service(hass, {"command": "exit 0"})
assert await hass.services.async_call( assert await hass.services.async_call(
DOMAIN, "test", {"message": "error"}, blocking=True NOTIFY_DOMAIN, "test6", {"message": "error"}, blocking=True
) )
assert check_output.call_count == 2 assert check_output.call_count == 2
assert "Timeout for command" in caplog.text assert "Timeout for command" in caplog.text
assert await hass.services.async_call( assert await hass.services.async_call(
DOMAIN, "test", {"message": "error"}, blocking=True NOTIFY_DOMAIN, "test6", {"message": "error"}, blocking=True
) )
assert check_output.call_count == 4 assert check_output.call_count == 4
assert "Error trying to exec command" in caplog.text assert "Error trying to exec command" in caplog.text

View file

@ -8,38 +8,65 @@ from unittest.mock import patch
import pytest import pytest
from homeassistant import setup from homeassistant import setup
from homeassistant.components.command_line import DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
import homeassistant.helpers.issue_registry as ir
from homeassistant.util import dt from homeassistant.util import dt
from tests.common import async_fire_time_changed from tests.common import async_fire_time_changed
async def setup_test_entities(hass: HomeAssistant, config_dict: dict[str, Any]) -> None: async def test_setup_platform_yaml(hass: HomeAssistant) -> None:
"""Set up a test command line sensor entity.""" """Test sensor setup."""
hass.states.async_set("sensor.input_sensor", "sensor_value")
assert await setup.async_setup_component( assert await setup.async_setup_component(
hass, hass,
SENSOR_DOMAIN, SENSOR_DOMAIN,
{ {
SENSOR_DOMAIN: [ SENSOR_DOMAIN: [
{"platform": "command_line", "name": "Test", **config_dict}, {
"platform": "command_line",
"name": "Test",
"command": "echo 5",
"unit_of_measurement": "in",
},
] ]
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
entity_state = hass.states.get("sensor.test")
assert entity_state
assert entity_state.state == "5"
assert entity_state.name == "Test"
assert entity_state.attributes["unit_of_measurement"] == "in"
issue_registry = ir.async_get(hass)
issue = issue_registry.async_get_issue(DOMAIN, "deprecated_yaml_sensor")
assert issue.translation_key == "deprecated_platform_yaml"
async def test_setup(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Test sensor setup.""" "get_config",
await setup_test_entities( [
hass,
{ {
"command": "echo 5", "command_line": [
"unit_of_measurement": "in", {
}, "sensor": {
) "name": "Test",
"command": "echo 5",
"unit_of_measurement": "in",
}
}
]
}
],
)
async def test_setup_integration_yaml(
hass: HomeAssistant, load_yaml_integration: None
) -> None:
"""Test sensor setup."""
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert entity_state.state == "5" assert entity_state.state == "5"
@ -47,30 +74,51 @@ async def test_setup(hass: HomeAssistant) -> None:
assert entity_state.attributes["unit_of_measurement"] == "in" assert entity_state.attributes["unit_of_measurement"] == "in"
async def test_template(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Test command sensor with template.""" "get_config",
await setup_test_entities( [
hass,
{ {
"command": "echo 50", "command_line": [
"unit_of_measurement": "in", {
"value_template": "{{ value | multiply(0.1) }}", "sensor": {
}, "name": "Test",
) "command": "echo 50",
"unit_of_measurement": "in",
"value_template": "{{ value | multiply(0.1) }}",
}
}
]
}
],
)
async def test_template(hass: HomeAssistant, load_yaml_integration: None) -> None:
"""Test command sensor with template."""
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert float(entity_state.state) == 5 assert float(entity_state.state) == 5
async def test_template_render(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Ensure command with templates get rendered properly.""" "get_config",
[
await setup_test_entities(
hass,
{ {
"command": "echo {{ states.sensor.input_sensor.state }}", "command_line": [
}, {
) "sensor": {
"name": "Test",
"command": "echo {{ states.sensor.input_sensor.state }}",
}
}
]
}
],
)
async def test_template_render(
hass: HomeAssistant, load_yaml_integration: None
) -> None:
"""Ensure command with templates get rendered properly."""
hass.states.async_set("sensor.input_sensor", "sensor_value")
# Give time for template to load # Give time for template to load
async_fire_time_changed( async_fire_time_changed(
@ -86,18 +134,27 @@ async def test_template_render(hass: HomeAssistant) -> None:
async def test_template_render_with_quote(hass: HomeAssistant) -> None: async def test_template_render_with_quote(hass: HomeAssistant) -> None:
"""Ensure command with templates and quotes get rendered properly.""" """Ensure command with templates and quotes get rendered properly."""
hass.states.async_set("sensor.input_sensor", "sensor_value")
await setup.async_setup_component(
hass,
DOMAIN,
{
"command_line": [
{
"sensor": {
"name": "Test",
"command": 'echo "{{ states.sensor.input_sensor.state }}" "3 4"',
}
}
]
},
)
await hass.async_block_till_done()
with patch( with patch(
"homeassistant.components.command_line.utils.subprocess.check_output", "homeassistant.components.command_line.utils.subprocess.check_output",
return_value=b"Works\n", return_value=b"Works\n",
) as check_output: ) as check_output:
await setup_test_entities(
hass,
{
"command": 'echo "{{ states.sensor.input_sensor.state }}" "3 4"',
},
)
# Give time for template to load # Give time for template to load
async_fire_time_changed( async_fire_time_changed(
hass, hass,
@ -105,7 +162,7 @@ async def test_template_render_with_quote(hass: HomeAssistant) -> None:
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(check_output.mock_calls) == 2 assert len(check_output.mock_calls) == 1
check_output.assert_called_with( check_output.assert_called_with(
'echo "sensor_value" "3 4"', 'echo "sensor_value" "3 4"',
shell=True, # nosec # shell by design shell=True, # nosec # shell by design
@ -114,59 +171,116 @@ async def test_template_render_with_quote(hass: HomeAssistant) -> None:
) )
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"sensor": {
"name": "Test",
"command": "echo {{ this template doesn't parse",
}
}
]
}
],
)
async def test_bad_template_render( async def test_bad_template_render(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, get_config: dict[str, Any]
) -> None: ) -> None:
"""Test rendering a broken template.""" """Test rendering a broken template."""
await setup.async_setup_component(
await setup_test_entities(
hass, hass,
{ DOMAIN,
"command": "echo {{ this template doesn't parse", get_config,
},
) )
await hass.async_block_till_done()
assert "Error rendering command template" in caplog.text assert "Error rendering command template" in caplog.text
async def test_bad_command(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Test bad command.""" "get_config",
await setup_test_entities( [
hass,
{ {
"command": "asdfasdf", "command_line": [
}, {
"sensor": {
"name": "Test",
"command": "asdfasdf",
}
}
]
}
],
)
async def test_bad_command(hass: HomeAssistant, get_config: dict[str, Any]) -> None:
"""Test bad command."""
await setup.async_setup_component(
hass,
DOMAIN,
get_config,
) )
await hass.async_block_till_done()
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert entity_state.state == "unknown" assert entity_state.state == "unknown"
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"sensor": {
"name": "Test",
"command": "exit 33",
}
}
]
}
],
)
async def test_return_code( async def test_return_code(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, get_config: dict[str, Any]
) -> None: ) -> None:
"""Test that an error return code is logged.""" """Test that an error return code is logged."""
await setup_test_entities( await setup.async_setup_component(
hass, hass,
{ DOMAIN,
"command": "exit 33", get_config,
},
) )
await hass.async_block_till_done()
assert "return code 33" in caplog.text assert "return code 33" in caplog.text
async def test_update_with_json_attrs(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Test attributes get extracted from a JSON result.""" "get_config",
await setup_test_entities( [
hass,
{ {
"command": ( "command_line": [
'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\": ' {
'\\"another_json_value\\", \\"key_three\\": \\"value_three\\" }' "sensor": {
), "name": "Test",
"json_attributes": ["key", "another_key", "key_three"], "command": (
}, 'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\": '
) '\\"another_json_value\\", \\"key_three\\": \\"value_three\\" }'
),
"json_attributes": ["key", "another_key", "key_three"],
}
}
]
}
],
)
async def test_update_with_json_attrs(
hass: HomeAssistant, load_yaml_integration: None
) -> None:
"""Test attributes get extracted from a JSON result."""
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert entity_state.state == "unknown" assert entity_state.state == "unknown"
@ -175,19 +289,30 @@ async def test_update_with_json_attrs(hass: HomeAssistant) -> None:
assert entity_state.attributes["key_three"] == "value_three" assert entity_state.attributes["key_three"] == "value_three"
async def test_update_with_json_attrs_and_value_template(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Test json_attributes can be used together with value_template.""" "get_config",
await setup_test_entities( [
hass,
{ {
"command": ( "command_line": [
'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\": ' {
'\\"another_json_value\\", \\"key_three\\": \\"value_three\\" }' "sensor": {
), "name": "Test",
"json_attributes": ["key", "another_key", "key_three"], "command": (
"value_template": '{{ value_json["key"] }}', 'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\": '
}, '\\"another_json_value\\", \\"key_three\\": \\"value_three\\" }'
) ),
"json_attributes": ["key", "another_key", "key_three"],
"value_template": '{{ value_json["key"] }}',
}
}
]
}
],
)
async def test_update_with_json_attrs_and_value_template(
hass: HomeAssistant, load_yaml_integration: None
) -> None:
"""Test json_attributes can be used together with value_template."""
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert entity_state.state == "some_json_value" assert entity_state.state == "some_json_value"
@ -196,75 +321,134 @@ async def test_update_with_json_attrs_and_value_template(hass: HomeAssistant) ->
assert entity_state.attributes["key_three"] == "value_three" assert entity_state.attributes["key_three"] == "value_three"
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"sensor": {
"name": "Test",
"command": "echo",
"json_attributes": ["key"],
}
}
]
}
],
)
async def test_update_with_json_attrs_no_data( async def test_update_with_json_attrs_no_data(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, get_config: dict[str, Any]
) -> None: ) -> None:
"""Test attributes when no JSON result fetched.""" """Test attributes when no JSON result fetched."""
await setup.async_setup_component(
await setup_test_entities(
hass, hass,
{ DOMAIN,
"command": "echo", get_config,
"json_attributes": ["key"],
},
) )
await hass.async_block_till_done()
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert "key" not in entity_state.attributes assert "key" not in entity_state.attributes
assert "Empty reply found when expecting JSON data" in caplog.text assert "Empty reply found when expecting JSON data" in caplog.text
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"sensor": {
"name": "Test",
"command": "echo [1, 2, 3]",
"json_attributes": ["key"],
}
}
]
}
],
)
async def test_update_with_json_attrs_not_dict( async def test_update_with_json_attrs_not_dict(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, get_config: dict[str, Any]
) -> None: ) -> None:
"""Test attributes when the return value not a dict.""" """Test attributes when the return value not a dict."""
await setup.async_setup_component(
await setup_test_entities(
hass, hass,
{ DOMAIN,
"command": "echo [1, 2, 3]", get_config,
"json_attributes": ["key"],
},
) )
await hass.async_block_till_done()
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert "key" not in entity_state.attributes assert "key" not in entity_state.attributes
assert "JSON result was not a dictionary" in caplog.text assert "JSON result was not a dictionary" in caplog.text
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"sensor": {
"name": "Test",
"command": "echo This is text rather than JSON data.",
"json_attributes": ["key"],
}
}
]
}
],
)
async def test_update_with_json_attrs_bad_json( async def test_update_with_json_attrs_bad_json(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, get_config: dict[str, Any]
) -> None: ) -> None:
"""Test attributes when the return value is invalid JSON.""" """Test attributes when the return value is invalid JSON."""
await setup.async_setup_component(
await setup_test_entities(
hass, hass,
{ DOMAIN,
"command": "echo This is text rather than JSON data.", get_config,
"json_attributes": ["key"],
},
) )
await hass.async_block_till_done()
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert "key" not in entity_state.attributes assert "key" not in entity_state.attributes
assert "Unable to parse output as JSON" in caplog.text assert "Unable to parse output as JSON" in caplog.text
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"sensor": {
"name": "Test",
"command": (
'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\": '
'\\"another_json_value\\", \\"key_three\\": \\"value_three\\" }'
),
"json_attributes": [
"key",
"another_key",
"key_three",
"missing_key",
],
}
}
]
}
],
)
async def test_update_with_missing_json_attrs( async def test_update_with_missing_json_attrs(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, load_yaml_integration: None
) -> None: ) -> None:
"""Test attributes when an expected key is missing.""" """Test attributes when an expected key is missing."""
await setup_test_entities(
hass,
{
"command": (
'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\": '
'\\"another_json_value\\", \\"key_three\\": \\"value_three\\" }'
),
"json_attributes": ["key", "another_key", "key_three", "missing_key"],
},
)
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert entity_state.attributes["key"] == "some_json_value" assert entity_state.attributes["key"] == "some_json_value"
@ -273,21 +457,30 @@ async def test_update_with_missing_json_attrs(
assert "missing_key" not in entity_state.attributes assert "missing_key" not in entity_state.attributes
@pytest.mark.parametrize(
"get_config",
[
{
"command_line": [
{
"sensor": {
"name": "Test",
"command": (
'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\": '
'\\"another_json_value\\", \\"key_three\\": \\"value_three\\" }'
),
"json_attributes": ["key", "another_key"],
}
}
]
}
],
)
async def test_update_with_unnecessary_json_attrs( async def test_update_with_unnecessary_json_attrs(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant, load_yaml_integration: None
) -> None: ) -> None:
"""Test attributes when an expected key is missing.""" """Test attributes when an expected key is missing."""
await setup_test_entities(
hass,
{
"command": (
'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\": '
'\\"another_json_value\\", \\"key_three\\": \\"value_three\\" }'
),
"json_attributes": ["key", "another_key"],
},
)
entity_state = hass.states.get("sensor.test") entity_state = hass.states.get("sensor.test")
assert entity_state assert entity_state
assert entity_state.attributes["key"] == "some_json_value" assert entity_state.attributes["key"] == "some_json_value"
@ -295,34 +488,40 @@ async def test_update_with_unnecessary_json_attrs(
assert "key_three" not in entity_state.attributes assert "key_three" not in entity_state.attributes
async def test_unique_id( @pytest.mark.parametrize(
hass: HomeAssistant, entity_registry: er.EntityRegistry "get_config",
) -> None: [
"""Test unique_id option and if it only creates one sensor per id."""
assert await setup.async_setup_component(
hass,
SENSOR_DOMAIN,
{ {
SENSOR_DOMAIN: [ "command_line": [
{ {
"platform": "command_line", "sensor": {
"unique_id": "unique", "name": "Test",
"command": "echo 0", "unique_id": "unique",
"command": "echo 0",
}
}, },
{ {
"platform": "command_line", "sensor": {
"unique_id": "not-so-unique-anymore", "name": "Test",
"command": "echo 1", "unique_id": "not-so-unique-anymore",
"command": "echo 1",
}
}, },
{ {
"platform": "command_line", "sensor": {
"unique_id": "not-so-unique-anymore", "name": "Test",
"command": "echo 2", "unique_id": "not-so-unique-anymore",
"command": "echo 2",
},
}, },
] ]
}, }
) ],
await hass.async_block_till_done() )
async def test_unique_id(
hass: HomeAssistant, entity_registry: er.EntityRegistry, load_yaml_integration: None
) -> None:
"""Test unique_id option and if it only creates one sensor per id."""
assert len(hass.states.async_all()) == 2 assert len(hass.states.async_all()) == 2

View file

@ -5,13 +5,13 @@ import json
import os import os
import subprocess import subprocess
import tempfile import tempfile
from typing import Any
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from homeassistant import setup from homeassistant import setup
from homeassistant.components.switch import DOMAIN, SCAN_INTERVAL from homeassistant.components.command_line import DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN, SCAN_INTERVAL
from homeassistant.const import ( from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_ENTITY_ID,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
@ -21,45 +21,45 @@ from homeassistant.const import (
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
import homeassistant.helpers.issue_registry as ir
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from tests.common import async_fire_time_changed from tests.common import async_fire_time_changed
async def setup_test_entity(hass: HomeAssistant, config_dict: dict[str, Any]) -> None: async def test_state_platform_yaml(hass: HomeAssistant) -> None:
"""Set up a test command line switch entity."""
assert await setup.async_setup_component(
hass,
DOMAIN,
{
DOMAIN: [
{"platform": "command_line", "switches": config_dict},
]
},
)
await hass.async_block_till_done()
async def test_state_none(hass: HomeAssistant) -> None:
"""Test with none state.""" """Test with none state."""
with tempfile.TemporaryDirectory() as tempdirname: with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, "switch_status") path = os.path.join(tempdirname, "switch_status")
await setup_test_entity( assert await setup.async_setup_component(
hass, hass,
SWITCH_DOMAIN,
{ {
"test": { SWITCH_DOMAIN: [
"command_on": f"echo 1 > {path}", {
"command_off": f"echo 0 > {path}", "platform": "command_line",
} "switches": {
"test": {
"command_on": f"echo 1 > {path}",
"command_off": f"echo 0 > {path}",
"friendly_name": "Test",
"icon_template": (
'{% if value=="1" %} mdi:on {% else %} mdi:off {% endif %}'
),
}
},
},
]
}, },
) )
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test") entity_state = hass.states.get("switch.test")
assert entity_state assert entity_state
assert entity_state.state == STATE_OFF assert entity_state.state == STATE_OFF
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_ON, SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
@ -70,7 +70,7 @@ async def test_state_none(hass: HomeAssistant) -> None:
assert entity_state.state == STATE_ON assert entity_state.state == STATE_ON
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
@ -80,32 +80,69 @@ async def test_state_none(hass: HomeAssistant) -> None:
assert entity_state assert entity_state
assert entity_state.state == STATE_OFF assert entity_state.state == STATE_OFF
issue_registry = ir.async_get(hass)
issue = issue_registry.async_get_issue(DOMAIN, "deprecated_yaml_switch")
assert issue.translation_key == "deprecated_platform_yaml"
async def test_state_integration_yaml(hass: HomeAssistant) -> None:
"""Test with none state."""
with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, "switch_status")
await setup.async_setup_component(
hass,
DOMAIN,
{
"command_line": [
{
"switch": {
"command_on": f"echo 1 > {path}",
"command_off": f"echo 0 > {path}",
"name": "Test",
}
}
]
},
)
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test")
assert entity_state
assert entity_state.state == STATE_OFF
async def test_state_value(hass: HomeAssistant) -> None: async def test_state_value(hass: HomeAssistant) -> None:
"""Test with state value.""" """Test with state value."""
with tempfile.TemporaryDirectory() as tempdirname: with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, "switch_status") path = os.path.join(tempdirname, "switch_status")
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_state": f"cat {path}", {
"command_on": f"echo 1 > {path}", "switch": {
"command_off": f"echo 0 > {path}", "command_state": f"cat {path}",
"value_template": '{{ value=="1" }}', "command_on": f"echo 1 > {path}",
"icon_template": ( "command_off": f"echo 0 > {path}",
'{% if value=="1" %} mdi:on {% else %} mdi:off {% endif %}' "value_template": '{{ value=="1" }}',
), "icon": (
} '{% if value=="1" %} mdi:on {% else %} mdi:off {% endif %}'
),
"name": "Test",
}
}
]
}, },
) )
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test") entity_state = hass.states.get("switch.test")
assert entity_state assert entity_state
assert entity_state.state == STATE_OFF assert entity_state.state == STATE_OFF
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_ON, SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
@ -117,7 +154,7 @@ async def test_state_value(hass: HomeAssistant) -> None:
assert entity_state.attributes.get("icon") == "mdi:on" assert entity_state.attributes.get("icon") == "mdi:on"
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
@ -136,28 +173,35 @@ async def test_state_json_value(hass: HomeAssistant) -> None:
oncmd = json.dumps({"status": "ok"}) oncmd = json.dumps({"status": "ok"})
offcmd = json.dumps({"status": "nope"}) offcmd = json.dumps({"status": "nope"})
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_state": f"cat {path}", {
"command_on": f"echo '{oncmd}' > {path}", "switch": {
"command_off": f"echo '{offcmd}' > {path}", "command_state": f"cat {path}",
"value_template": '{{ value_json.status=="ok" }}', "command_on": f"echo '{oncmd}' > {path}",
"icon_template": ( "command_off": f"echo '{offcmd}' > {path}",
'{% if value_json.status=="ok" %} mdi:on' "value_template": '{{ value_json.status=="ok" }}',
"{% else %} mdi:off {% endif %}" "icon": (
), '{% if value_json.status=="ok" %} mdi:on'
} "{% else %} mdi:off {% endif %}"
),
"name": "Test",
}
}
]
}, },
) )
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test") entity_state = hass.states.get("switch.test")
assert entity_state assert entity_state
assert entity_state.state == STATE_OFF assert entity_state.state == STATE_OFF
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_ON, SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
@ -169,7 +213,7 @@ async def test_state_json_value(hass: HomeAssistant) -> None:
assert entity_state.attributes.get("icon") == "mdi:on" assert entity_state.attributes.get("icon") == "mdi:on"
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
@ -185,23 +229,30 @@ async def test_state_code(hass: HomeAssistant) -> None:
"""Test with state code.""" """Test with state code."""
with tempfile.TemporaryDirectory() as tempdirname: with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, "switch_status") path = os.path.join(tempdirname, "switch_status")
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_state": f"cat {path}", {
"command_on": f"echo 1 > {path}", "switch": {
"command_off": f"echo 0 > {path}", "command_state": f"cat {path}",
} "command_on": f"echo 1 > {path}",
"command_off": f"echo 0 > {path}",
"name": "Test",
}
}
]
}, },
) )
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test") entity_state = hass.states.get("switch.test")
assert entity_state assert entity_state
assert entity_state.state == STATE_OFF assert entity_state.state == STATE_OFF
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_ON, SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
@ -212,7 +263,7 @@ async def test_state_code(hass: HomeAssistant) -> None:
assert entity_state.state == STATE_ON assert entity_state.state == STATE_ON
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
@ -228,15 +279,23 @@ async def test_assumed_state_should_be_true_if_command_state_is_none(
) -> None: ) -> None:
"""Test with state value.""" """Test with state value."""
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_on": "echo 'on command'", {
"command_off": "echo 'off command'", "switch": {
} "command_on": "echo 'on command'",
"command_off": "echo 'off command'",
"name": "Test",
}
}
]
}, },
) )
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test") entity_state = hass.states.get("switch.test")
assert entity_state assert entity_state
assert entity_state.attributes["assumed_state"] assert entity_state.attributes["assumed_state"]
@ -247,16 +306,24 @@ async def test_assumed_state_should_absent_if_command_state_present(
) -> None: ) -> None:
"""Test with state value.""" """Test with state value."""
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_on": "echo 'on command'", {
"command_off": "echo 'off command'", "switch": {
"command_state": "cat {}", "command_on": "echo 'on command'",
} "command_off": "echo 'off command'",
"command_state": "cat {}",
"name": "Test",
}
}
]
}, },
) )
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test") entity_state = hass.states.get("switch.test")
assert entity_state assert entity_state
assert "assumed_state" not in entity_state.attributes assert "assumed_state" not in entity_state.attributes
@ -264,18 +331,24 @@ async def test_assumed_state_should_absent_if_command_state_present(
async def test_name_is_set_correctly(hass: HomeAssistant) -> None: async def test_name_is_set_correctly(hass: HomeAssistant) -> None:
"""Test that name is set correctly.""" """Test that name is set correctly."""
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_on": "echo 'on command'", {
"command_off": "echo 'off command'", "switch": {
"friendly_name": "Test friendly name!", "command_on": "echo 'on command'",
} "command_off": "echo 'off command'",
"name": "Test friendly name!",
}
}
]
}, },
) )
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test") entity_state = hass.states.get("switch.test_friendly_name")
assert entity_state assert entity_state
assert entity_state.name == "Test friendly name!" assert entity_state.name == "Test friendly name!"
@ -284,16 +357,23 @@ async def test_switch_command_state_fail(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant
) -> None: ) -> None:
"""Test that switch failures are handled correctly.""" """Test that switch failures are handled correctly."""
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_on": "exit 0", {
"command_off": "exit 0'", "switch": {
"command_state": "echo 1", "command_on": "exit 0",
} "command_off": "exit 0'",
"command_state": "echo 1",
"name": "Test",
}
}
]
}, },
) )
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -303,7 +383,7 @@ async def test_switch_command_state_fail(
assert entity_state.state == "on" assert entity_state.state == "on"
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
@ -329,16 +409,24 @@ async def test_switch_command_state_code_exceptions(
subprocess.SubprocessError(), subprocess.SubprocessError(),
], ],
) as check_output: ) as check_output:
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_on": "exit 0", {
"command_off": "exit 0'", "switch": {
"command_state": "echo 1", "command_on": "exit 0",
} "command_off": "exit 0'",
"command_state": "echo 1",
"name": "Test",
}
}
]
}, },
) )
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
assert check_output.called assert check_output.called
@ -362,17 +450,25 @@ async def test_switch_command_state_value_exceptions(
subprocess.SubprocessError(), subprocess.SubprocessError(),
], ],
) as check_output: ) as check_output:
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_on": "exit 0", {
"command_off": "exit 0'", "switch": {
"command_state": "echo 1", "command_on": "exit 0",
"value_template": '{{ value=="1" }}', "command_off": "exit 0'",
} "command_state": "echo 1",
"value_template": '{{ value=="1" }}',
"name": "Test",
}
}
]
}, },
) )
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
assert check_output.call_count == 1 assert check_output.call_count == 1
@ -384,12 +480,24 @@ async def test_switch_command_state_value_exceptions(
assert "Error trying to exec command" in caplog.text assert "Error trying to exec command" in caplog.text
async def test_no_switches( async def test_no_switches_platform_yaml(
caplog: pytest.LogCaptureFixture, hass: HomeAssistant caplog: pytest.LogCaptureFixture, hass: HomeAssistant
) -> None: ) -> None:
"""Test with no switches.""" """Test with no switches."""
await setup_test_entity(hass, {}) assert await setup.async_setup_component(
hass,
SWITCH_DOMAIN,
{
SWITCH_DOMAIN: [
{
"platform": "command_line",
"switches": {},
},
]
},
)
await hass.async_block_till_done()
assert "No switches" in caplog.text assert "No switches" in caplog.text
@ -397,26 +505,39 @@ async def test_unique_id(
hass: HomeAssistant, entity_registry: er.EntityRegistry hass: HomeAssistant, entity_registry: er.EntityRegistry
) -> None: ) -> None:
"""Test unique_id option and if it only creates one switch per id.""" """Test unique_id option and if it only creates one switch per id."""
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"unique": { "command_line": [
"command_on": "echo on", {
"command_off": "echo off", "switch": {
"unique_id": "unique", "command_on": "echo on",
}, "command_off": "echo off",
"not_unique_1": { "unique_id": "unique",
"command_on": "echo on", "name": "Test",
"command_off": "echo off", }
"unique_id": "not-so-unique-anymore", },
}, {
"not_unique_2": { "switch": {
"command_on": "echo on", "command_on": "echo on",
"command_off": "echo off", "command_off": "echo off",
"unique_id": "not-so-unique-anymore", "unique_id": "not-so-unique-anymore",
}, "name": "Test2",
}
},
{
"switch": {
"command_on": "echo on",
"command_off": "echo off",
"unique_id": "not-so-unique-anymore",
"name": "Test3",
},
},
]
}, },
) )
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 2 assert len(hass.states.async_all()) == 2
@ -432,12 +553,24 @@ async def test_command_failure(
) -> None: ) -> None:
"""Test command failure.""" """Test command failure."""
await setup_test_entity( await setup.async_setup_component(
hass, hass,
{"test": {"command_off": "exit 33"}}, DOMAIN,
{
"command_line": [
{
"switch": {
"command_off": "exit 33",
"name": "Test",
}
}
]
},
) )
await hass.async_block_till_done()
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: "switch.test"}, blocking=True SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: "switch.test"}, blocking=True
) )
assert "return code 33" in caplog.text assert "return code 33" in caplog.text
@ -446,29 +579,39 @@ async def test_templating(hass: HomeAssistant) -> None:
"""Test with templating.""" """Test with templating."""
with tempfile.TemporaryDirectory() as tempdirname: with tempfile.TemporaryDirectory() as tempdirname:
path = os.path.join(tempdirname, "switch_status") path = os.path.join(tempdirname, "switch_status")
await setup_test_entity( await setup.async_setup_component(
hass, hass,
DOMAIN,
{ {
"test": { "command_line": [
"command_state": f"cat {path}", {
"command_on": f"echo 1 > {path}", "switch": {
"command_off": f"echo 0 > {path}", "command_state": f"cat {path}",
"value_template": '{{ value=="1" }}', "command_on": f"echo 1 > {path}",
"icon_template": ( "command_off": f"echo 0 > {path}",
'{% if this.state=="on" %} mdi:on {% else %} mdi:off {% endif %}' "value_template": '{{ value=="1" }}',
), "icon": (
}, '{% if this.state=="on" %} mdi:on {% else %} mdi:off {% endif %}'
"test2": { ),
"command_state": f"cat {path}", "name": "Test",
"command_on": f"echo 1 > {path}", }
"command_off": f"echo 0 > {path}", },
"value_template": '{{ value=="1" }}', {
"icon_template": ( "switch": {
'{% if states("switch.test2")=="on" %} mdi:on {% else %} mdi:off {% endif %}' "command_state": f"cat {path}",
), "command_on": f"echo 1 > {path}",
}, "command_off": f"echo 0 > {path}",
"value_template": '{{ value=="1" }}',
"icon": (
'{% if states("switch.test2")=="on" %} mdi:on {% else %} mdi:off {% endif %}'
),
"name": "Test2",
},
},
]
}, },
) )
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test") entity_state = hass.states.get("switch.test")
entity_state2 = hass.states.get("switch.test2") entity_state2 = hass.states.get("switch.test2")
@ -476,13 +619,13 @@ async def test_templating(hass: HomeAssistant) -> None:
assert entity_state2.state == STATE_OFF assert entity_state2.state == STATE_OFF
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_ON, SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.test"}, {ATTR_ENTITY_ID: "switch.test"},
blocking=True, blocking=True,
) )
await hass.services.async_call( await hass.services.async_call(
DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_ON, SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.test2"}, {ATTR_ENTITY_ID: "switch.test2"},
blocking=True, blocking=True,