diff --git a/homeassistant/components/command_line/cover.py b/homeassistant/components/command_line/cover.py index 05d2b9634f2..961d9a31f4e 100644 --- a/homeassistant/components/command_line/cover.py +++ b/homeassistant/components/command_line/cover.py @@ -107,11 +107,6 @@ class CommandCover(CoverEntity): return success - def _query_state_value(self, command): - """Execute state command for return value.""" - _LOGGER.info("Running state value command: %s", command) - return check_output_or_log(command, self._timeout) - @property def should_poll(self): """Only poll if we have state command.""" @@ -138,10 +133,8 @@ class CommandCover(CoverEntity): def _query_state(self): """Query for the state.""" - if not self._command_state: - _LOGGER.error("No state command specified") - return - return self._query_state_value(self._command_state) + _LOGGER.info("Running state value command: %s", self._command_state) + return check_output_or_log(self._command_state, self._timeout) def update(self): """Update device state.""" diff --git a/homeassistant/components/command_line/sensor.py b/homeassistant/components/command_line/sensor.py index 35f7c5a4811..3e6f384cca3 100644 --- a/homeassistant/components/command_line/sensor.py +++ b/homeassistant/components/command_line/sensor.py @@ -145,19 +145,14 @@ class CommandSensorData: def update(self): """Get the latest data with a shell command.""" command = self.command - cache = {} - if command in cache: - prog, args, args_compiled = cache[command] - elif " " not in command: + if " " not in command: prog = command args = None args_compiled = None - cache[command] = (prog, args, args_compiled) else: prog, args = command.split(" ", 1) args_compiled = template.Template(args, self.hass) - cache[command] = (prog, args, args_compiled) if args_compiled: try: diff --git a/homeassistant/components/command_line/switch.py b/homeassistant/components/command_line/switch.py index ce46cd4f2cd..ae6c1c0c925 100644 --- a/homeassistant/components/command_line/switch.py +++ b/homeassistant/components/command_line/switch.py @@ -144,9 +144,6 @@ class CommandSwitch(SwitchEntity): def _query_state(self): """Query for state.""" - if not self._command_state: - _LOGGER.error("No state command specified") - return if self._value_template: return self._query_state_value(self._command_state) return self._query_state_code(self._command_state) diff --git a/tests/components/command_line/test_binary_sensor.py b/tests/components/command_line/test_binary_sensor.py index 90871faaf78..21209a8b60d 100644 --- a/tests/components/command_line/test_binary_sensor.py +++ b/tests/components/command_line/test_binary_sensor.py @@ -1,68 +1,65 @@ """The tests for the Command line Binary sensor platform.""" -import unittest - -from homeassistant.components.command_line import binary_sensor as command_line +from homeassistant import setup +from homeassistant.components.binary_sensor import DOMAIN from homeassistant.const import STATE_OFF, STATE_ON -from homeassistant.helpers import template - -from tests.common import get_test_home_assistant +from homeassistant.helpers.typing import Any, Dict, HomeAssistantType -class TestCommandSensorBinarySensor(unittest.TestCase): - """Test the Command line Binary sensor.""" +async def setup_test_entity( + hass: HomeAssistantType, config_dict: Dict[str, Any] +) -> None: + """Set up a test command line binary_sensor entity.""" + assert await setup.async_setup_component( + hass, + DOMAIN, + {DOMAIN: {"platform": "command_line", "name": "Test", **config_dict}}, + ) + await hass.async_block_till_done() - def setUp(self): - """Set up things to be run when tests are started.""" - self.hass = get_test_home_assistant() - self.addCleanup(self.hass.stop) - def test_setup(self): - """Test sensor setup.""" - config = { - "name": "Test", +async def test_setup(hass: HomeAssistantType) -> None: + """Test sensor setup.""" + await setup_test_entity( + hass, + { "command": "echo 1", "payload_on": "1", "payload_off": "0", - "command_timeout": 15, - } + }, + ) - devices = [] + entity_state = hass.states.get("binary_sensor.test") + assert entity_state + assert entity_state.state == STATE_ON + assert entity_state.name == "Test" - def add_dev_callback(devs, update): - """Add callback to add devices.""" - for dev in devs: - devices.append(dev) - command_line.setup_platform(self.hass, config, add_dev_callback) +async def test_template(hass: HomeAssistantType) -> None: + """Test setting the state with a template.""" - assert 1 == len(devices) - entity = devices[0] - entity.update() - assert "Test" == entity.name - assert STATE_ON == entity.state + await setup_test_entity( + hass, + { + "command": "echo 10", + "payload_on": "1.0", + "payload_off": "0", + "value_template": "{{ value | multiply(0.1) }}", + }, + ) - def test_template(self): - """Test setting the state with a template.""" - data = command_line.CommandSensorData(self.hass, "echo 10", 15) + entity_state = hass.states.get("binary_sensor.test") + assert entity_state.state == STATE_ON - entity = command_line.CommandBinarySensor( - self.hass, - data, - "test", - None, - "1.0", - "0", - template.Template("{{ value | multiply(0.1) }}", self.hass), - ) - entity.update() - assert STATE_ON == entity.state - def test_sensor_off(self): - """Test setting the state with a template.""" - data = command_line.CommandSensorData(self.hass, "echo 0", 15) - - entity = command_line.CommandBinarySensor( - self.hass, data, "test", None, "1", "0", None - ) - entity.update() - assert STATE_OFF == entity.state +async def test_sensor_off(hass: HomeAssistantType) -> None: + """Test setting the state with a template.""" + await setup_test_entity( + hass, + { + "command": "echo 0", + "payload_on": "1", + "payload_off": "0", + }, + ) + entity_state = hass.states.get("binary_sensor.test") + assert entity_state.state == STATE_OFF diff --git a/tests/components/command_line/test_cover.py b/tests/components/command_line/test_cover.py index ee692413bcd..093c1e86212 100644 --- a/tests/components/command_line/test_cover.py +++ b/tests/components/command_line/test_cover.py @@ -1,15 +1,10 @@ """The tests the cover command line platform.""" import os -from os import path import tempfile -from unittest import mock from unittest.mock import patch -import pytest - -from homeassistant import config as hass_config -import homeassistant.components.command_line.cover as cmd_rs -from homeassistant.components.cover import DOMAIN +from homeassistant import config as hass_config, setup +from homeassistant.components.cover import DOMAIN, SCAN_INTERVAL from homeassistant.const import ( ATTR_ENTITY_ID, SERVICE_CLOSE_COVER, @@ -17,101 +12,128 @@ from homeassistant.const import ( SERVICE_RELOAD, SERVICE_STOP_COVER, ) -from homeassistant.setup import async_setup_component +from homeassistant.helpers.typing import Any, Dict, HomeAssistantType +import homeassistant.util.dt as dt_util + +from tests.common import async_fire_time_changed -@pytest.fixture -def rs(hass): - """Return CommandCover instance.""" - return cmd_rs.CommandCover( +async def setup_test_entity( + hass: HomeAssistantType, config_dict: Dict[str, Any] +) -> None: + """Set up a test command line notify service.""" + assert await setup.async_setup_component( hass, - "foo", - "command_open", - "command_close", - "command_stop", - "command_state", - None, - 15, + DOMAIN, + { + DOMAIN: [ + {"platform": "command_line", "covers": config_dict}, + ] + }, ) + await hass.async_block_till_done() -def test_should_poll_new(rs): - """Test the setting of polling.""" - assert rs.should_poll is True - rs._command_state = None - assert rs.should_poll is False +async def test_no_covers(caplog: Any, hass: HomeAssistantType) -> None: + """Test that the cover does not polls when there's no state command.""" + + with patch( + "homeassistant.components.command_line.subprocess.check_output", + return_value=b"50\n", + ): + await setup_test_entity(hass, {}) + assert "No covers added" in caplog.text -def test_query_state_value(rs): - """Test with state value.""" - with mock.patch("subprocess.check_output") as mock_run: - mock_run.return_value = b" foo bar " - result = rs._query_state_value("runme") - assert "foo bar" == result - assert mock_run.call_count == 1 - assert mock_run.call_args == mock.call( - "runme", shell=True, timeout=15 # nosec # shell by design +async def test_no_poll_when_cover_has_no_command_state(hass: HomeAssistantType) -> None: + """Test that the cover does not polls when there's no state command.""" + + with patch( + "homeassistant.components.command_line.subprocess.check_output", + return_value=b"50\n", + ) as check_output: + await setup_test_entity(hass, {"test": {}}) + async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) + await hass.async_block_till_done() + assert not check_output.called + + +async def test_poll_when_cover_has_command_state(hass: HomeAssistantType) -> None: + """Test that the cover polls when there's a state command.""" + + with patch( + "homeassistant.components.command_line.subprocess.check_output", + return_value=b"50\n", + ) as check_output: + await setup_test_entity(hass, {"test": {"command_state": "echo state"}}) + async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) + await hass.async_block_till_done() + check_output.assert_called_once_with( + "echo state", shell=True, timeout=15 # nosec # shell by design ) -async def test_state_value(hass): +async def test_state_value(hass: HomeAssistantType) -> None: """Test with state value.""" with tempfile.TemporaryDirectory() as tempdirname: path = os.path.join(tempdirname, "cover_status") - test_cover = { - "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 }}", - } - assert ( - await async_setup_component( - hass, - DOMAIN, - {"cover": {"platform": "command_line", "covers": {"test": test_cover}}}, - ) - is True + await setup_test_entity( + hass, + { + "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 }}", + } + }, ) - await hass.async_block_till_done() - assert "unknown" == hass.states.get("cover.test").state + entity_state = hass.states.get("cover.test") + assert entity_state + assert entity_state.state == "unknown" await hass.services.async_call( DOMAIN, SERVICE_OPEN_COVER, {ATTR_ENTITY_ID: "cover.test"}, blocking=True ) - assert "open" == hass.states.get("cover.test").state + entity_state = hass.states.get("cover.test") + assert entity_state + assert entity_state.state == "open" await hass.services.async_call( DOMAIN, SERVICE_CLOSE_COVER, {ATTR_ENTITY_ID: "cover.test"}, blocking=True ) - assert "open" == hass.states.get("cover.test").state + entity_state = hass.states.get("cover.test") + assert entity_state + assert entity_state.state == "open" await hass.services.async_call( DOMAIN, SERVICE_STOP_COVER, {ATTR_ENTITY_ID: "cover.test"}, blocking=True ) - assert "closed" == hass.states.get("cover.test").state + entity_state = hass.states.get("cover.test") + assert entity_state + assert entity_state.state == "closed" -async def test_reload(hass): +async def test_reload(hass: HomeAssistantType) -> None: """Verify we can reload command_line covers.""" - test_cover = { - "command_state": "echo open", - "value_template": "{{ value }}", - } - await async_setup_component( + await setup_test_entity( hass, - DOMAIN, - {"cover": {"platform": "command_line", "covers": {"test": test_cover}}}, + { + "test": { + "command_state": "echo open", + "value_template": "{{ value }}", + } + }, ) - await hass.async_block_till_done() + entity_state = hass.states.get("cover.test") + assert entity_state + assert entity_state.state == "unknown" - assert len(hass.states.async_all()) == 1 - assert hass.states.get("cover.test").state - - yaml_path = path.join( - _get_fixtures_base_path(), + yaml_path = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "fixtures", "command_line/configuration.yaml", ) @@ -126,9 +148,18 @@ async def test_reload(hass): assert len(hass.states.async_all()) == 1 - assert hass.states.get("cover.test") is None + assert not hass.states.get("cover.test") assert hass.states.get("cover.from_yaml") -def _get_fixtures_base_path(): - return path.dirname(path.dirname(path.dirname(__file__))) +async def test_move_cover_failure(caplog: Any, hass: HomeAssistantType) -> None: + """Test with state value.""" + + await setup_test_entity( + hass, + {"test": {"command_open": "exit 1"}}, + ) + await hass.services.async_call( + DOMAIN, SERVICE_OPEN_COVER, {ATTR_ENTITY_ID: "cover.test"}, blocking=True + ) + assert "Command failed" in caplog.text diff --git a/tests/components/command_line/test_notify.py b/tests/components/command_line/test_notify.py index 3dcb521cfd2..4166b9e8bbf 100644 --- a/tests/components/command_line/test_notify.py +++ b/tests/components/command_line/test_notify.py @@ -1,117 +1,115 @@ """The tests for the command line notification platform.""" import os +import subprocess import tempfile -import unittest from unittest.mock import patch -import homeassistant.components.notify as notify -from homeassistant.setup import async_setup_component, setup_component - -from tests.common import assert_setup_component, get_test_home_assistant +from homeassistant import setup +from homeassistant.components.notify import DOMAIN +from homeassistant.helpers.typing import Any, Dict, HomeAssistantType -class TestCommandLine(unittest.TestCase): - """Test the command line notifications.""" - - def setUp(self): # pylint: disable=invalid-name - """Set up things to be run when tests are started.""" - self.hass = get_test_home_assistant() - self.addCleanup(self.tear_down_cleanup) - - def tear_down_cleanup(self): - """Stop down everything that was started.""" - self.hass.stop() - - def test_setup(self): - """Test setup.""" - with assert_setup_component(1) as handle_config: - assert setup_component( - self.hass, - "notify", - { - "notify": { - "name": "test", - "platform": "command_line", - "command": "echo $(cat); exit 1", - } - }, - ) - assert handle_config[notify.DOMAIN] - - def test_bad_config(self): - """Test set up the platform with bad/missing configuration.""" - config = {notify.DOMAIN: {"name": "test", "platform": "command_line"}} - with assert_setup_component(0) as handle_config: - assert setup_component(self.hass, notify.DOMAIN, config) - assert not handle_config[notify.DOMAIN] - - def test_command_line_output(self): - """Test the command line output.""" - with tempfile.TemporaryDirectory() as tempdirname: - filename = os.path.join(tempdirname, "message.txt") - message = "one, two, testing, testing" - with assert_setup_component(1) as handle_config: - assert setup_component( - self.hass, - notify.DOMAIN, - { - "notify": { - "name": "test", - "platform": "command_line", - "command": f"echo $(cat) > {filename}", - } - }, - ) - assert handle_config[notify.DOMAIN] - - assert self.hass.services.call( - "notify", "test", {"message": message}, blocking=True - ) - - with open(filename) as fil: - # the echo command adds a line break - assert fil.read() == f"{message}\n" - - @patch("homeassistant.components.command_line.notify._LOGGER.error") - def test_error_for_none_zero_exit_code(self, mock_error): - """Test if an error is logged for non zero exit codes.""" - with assert_setup_component(1) as handle_config: - assert setup_component( - self.hass, - notify.DOMAIN, - { - "notify": { - "name": "test", - "platform": "command_line", - "command": "echo $(cat); exit 1", - } - }, - ) - assert handle_config[notify.DOMAIN] - - assert self.hass.services.call( - "notify", "test", {"message": "error"}, blocking=True - ) - assert mock_error.call_count == 1 - - -async def test_timeout(hass, caplog): - """Test we do not block forever.""" - assert await async_setup_component( +async def setup_test_service( + hass: HomeAssistantType, config_dict: Dict[str, Any] +) -> None: + """Set up a test command line notify service.""" + assert await setup.async_setup_component( hass, - notify.DOMAIN, + DOMAIN, { - "notify": { - "name": "test", - "platform": "command_line", - "command": "sleep 10000", - "command_timeout": 0.0000001, - } + DOMAIN: [ + {"platform": "command_line", "name": "Test", **config_dict}, + ] }, ) await hass.async_block_till_done() - assert await hass.services.async_call( - "notify", "test", {"message": "error"}, blocking=True + + +async def test_setup(hass: HomeAssistantType) -> None: + """Test sensor setup.""" + await setup_test_service(hass, {"command": "exit 0"}) + assert hass.services.has_service(DOMAIN, "test") + + +async def test_bad_config(hass: HomeAssistantType) -> None: + """Test set up the platform with bad/missing configuration.""" + await setup_test_service(hass, {}) + assert not hass.services.has_service(DOMAIN, "test") + + +async def test_command_line_output(hass: HomeAssistantType) -> None: + """Test the command line output.""" + with tempfile.TemporaryDirectory() as tempdirname: + filename = os.path.join(tempdirname, "message.txt") + message = "one, two, testing, testing" + await setup_test_service( + hass, + { + "command": f"cat > {filename}", + }, + ) + + assert hass.services.has_service(DOMAIN, "test") + + assert await hass.services.async_call( + DOMAIN, "test", {"message": message}, blocking=True + ) + with open(filename) as handle: + # the echo command adds a line break + assert message == handle.read() + + +async def test_error_for_none_zero_exit_code( + caplog: Any, hass: HomeAssistantType +) -> None: + """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( + DOMAIN, "test", {"message": "error"}, blocking=True + ) + assert "Command failed" in caplog.text + + +async def test_timeout(caplog: Any, hass: HomeAssistantType) -> None: + """Test blocking is not forever.""" + await setup_test_service( + hass, + { + "command": "sleep 10000", + "command_timeout": 0.0000001, + }, + ) + assert await hass.services.async_call( + DOMAIN, "test", {"message": "error"}, blocking=True ) - await hass.async_block_till_done() assert "Timeout" in caplog.text + + +async def test_subprocess_exceptions(caplog: Any, hass: HomeAssistantType) -> None: + """Test that notify subprocess exceptions are handled correctly.""" + + with patch( + "homeassistant.components.command_line.notify.subprocess.Popen", + side_effect=[ + subprocess.TimeoutExpired("cmd", 10), + subprocess.SubprocessError(), + ], + ) as check_output: + await setup_test_service(hass, {"command": "exit 0"}) + assert await hass.services.async_call( + DOMAIN, "test", {"message": "error"}, blocking=True + ) + assert check_output.call_count == 1 + assert "Timeout for command" in caplog.text + + assert await hass.services.async_call( + DOMAIN, "test", {"message": "error"}, blocking=True + ) + assert check_output.call_count == 2 + assert "Error trying to exec command" in caplog.text diff --git a/tests/components/command_line/test_sensor.py b/tests/components/command_line/test_sensor.py index 042c9acf432..66472c5feba 100644 --- a/tests/components/command_line/test_sensor.py +++ b/tests/components/command_line/test_sensor.py @@ -1,201 +1,224 @@ """The tests for the Command line sensor platform.""" -import unittest from unittest.mock import patch -from homeassistant.components.command_line import sensor as command_line -from homeassistant.helpers.template import Template - -from tests.common import get_test_home_assistant +from homeassistant import setup +from homeassistant.components.sensor import DOMAIN +from homeassistant.helpers.typing import Any, Dict, HomeAssistantType -class TestCommandSensorSensor(unittest.TestCase): - """Test the Command line sensor.""" +async def setup_test_entities( + hass: HomeAssistantType, config_dict: Dict[str, Any] +) -> None: + """Set up a test command line sensor entity.""" + assert await setup.async_setup_component( + hass, + DOMAIN, + { + DOMAIN: [ + { + "platform": "template", + "sensors": { + "template_sensor": { + "value_template": "template_value", + } + }, + }, + {"platform": "command_line", "name": "Test", **config_dict}, + ] + }, + ) + await hass.async_block_till_done() - def setUp(self): - """Set up things to be run when tests are started.""" - self.hass = get_test_home_assistant() - self.addCleanup(self.hass.stop) - def update_side_effect(self, data): - """Side effect function for mocking CommandSensorData.update().""" - self.commandline.data = data - - def test_setup(self): - """Test sensor setup.""" - config = { - "name": "Test", - "unit_of_measurement": "in", +async def test_setup(hass: HomeAssistantType) -> None: + """Test sensor setup.""" + await setup_test_entities( + hass, + { "command": "echo 5", - "command_timeout": 15, - } - devices = [] + "unit_of_measurement": "in", + }, + ) + 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" - def add_dev_callback(devs, update): - """Add callback to add devices.""" - for dev in devs: - devices.append(dev) - command_line.setup_platform(self.hass, config, add_dev_callback) +async def test_template(hass: HomeAssistantType) -> None: + """Test command sensor with template.""" + await setup_test_entities( + hass, + { + "command": "echo 50", + "unit_of_measurement": "in", + "value_template": "{{ value | multiply(0.1) }}", + }, + ) + entity_state = hass.states.get("sensor.test") + assert entity_state + assert float(entity_state.state) == 5 - assert len(devices) == 1 - entity = devices[0] - entity.update() - assert entity.name == "Test" - assert entity.unit_of_measurement == "in" - assert entity.state == "5" - def test_template(self): - """Test command sensor with template.""" - data = command_line.CommandSensorData(self.hass, "echo 50", 15) +async def test_template_render(hass: HomeAssistantType) -> None: + """Ensure command with templates get rendered properly.""" - entity = command_line.CommandSensor( - self.hass, - data, - "test", - "in", - Template("{{ value | multiply(0.1) }}", self.hass), - [], + await setup_test_entities( + hass, + { + "command": "echo {{ states.sensor.template_sensor.state }}", + }, + ) + entity_state = hass.states.get("sensor.test") + assert entity_state + assert entity_state.state == "template_value" + + +async def test_template_render_with_quote(hass: HomeAssistantType) -> None: + """Ensure command with templates and quotes get rendered properly.""" + + with patch( + "homeassistant.components.command_line.subprocess.check_output", + return_value=b"Works\n", + ) as check_output: + await setup_test_entities( + hass, + { + "command": 'echo "{{ states.sensor.template_sensor.state }}" "3 4"', + }, ) - entity.update() - assert float(entity.state) == 5 - - def test_template_render(self): - """Ensure command with templates get rendered properly.""" - self.hass.states.set("sensor.test_state", "Works") - data = command_line.CommandSensorData( - self.hass, "echo {{ states.sensor.test_state.state }}", 15 - ) - data.update() - - assert data.value == "Works" - - def test_template_render_with_quote(self): - """Ensure command with templates and quotes get rendered properly.""" - self.hass.states.set("sensor.test_state", "Works 2") - with patch( - "homeassistant.components.command_line.subprocess.check_output", - return_value=b"Works\n", - ) as check_output: - data = command_line.CommandSensorData( - self.hass, - 'echo "{{ states.sensor.test_state.state }}" "3 4"', - 15, - ) - data.update() - - assert data.value == "Works" check_output.assert_called_once_with( - 'echo "Works 2" "3 4"', shell=True, timeout=15 # nosec # shell by design + 'echo "template_value" "3 4"', + shell=True, # nosec # shell by design + timeout=15, ) - def test_bad_command(self): - """Test bad command.""" - data = command_line.CommandSensorData(self.hass, "asdfasdf", 15) - data.update() - assert data.value is None +async def test_bad_template_render(caplog: Any, hass: HomeAssistantType) -> None: + """Test rendering a broken template.""" - def test_update_with_json_attrs(self): - """Test attributes get extracted from a JSON result.""" - data = command_line.CommandSensorData( - self.hass, - ( - 'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\":\ - \\"another_json_value\\", \\"key_three\\": \\"value_three\\" }' - ), - 15, - ) + await setup_test_entities( + hass, + { + "command": "echo {{ this template doesn't parse", + }, + ) - self.sensor = command_line.CommandSensor( - self.hass, data, "test", None, None, ["key", "another_key", "key_three"] - ) - self.sensor.update() - assert self.sensor.device_state_attributes["key"] == "some_json_value" - assert ( - self.sensor.device_state_attributes["another_key"] == "another_json_value" - ) - assert self.sensor.device_state_attributes["key_three"] == "value_three" + assert "Error rendering command template" in caplog.text - @patch("homeassistant.components.command_line.sensor._LOGGER") - def test_update_with_json_attrs_no_data(self, mock_logger): - """Test attributes when no JSON result fetched.""" - data = command_line.CommandSensorData(self.hass, "echo ", 15) - self.sensor = command_line.CommandSensor( - self.hass, data, "test", None, None, ["key"] - ) - self.sensor.update() - assert {} == self.sensor.device_state_attributes - assert mock_logger.warning.called - @patch("homeassistant.components.command_line.sensor._LOGGER") - def test_update_with_json_attrs_not_dict(self, mock_logger): - """Test attributes get extracted from a JSON result.""" - data = command_line.CommandSensorData(self.hass, "echo [1, 2, 3]", 15) - self.sensor = command_line.CommandSensor( - self.hass, data, "test", None, None, ["key"] - ) - self.sensor.update() - assert {} == self.sensor.device_state_attributes - assert mock_logger.warning.called +async def test_bad_command(hass: HomeAssistantType) -> None: + """Test bad command.""" + await setup_test_entities( + hass, + { + "command": "asdfasdf", + }, + ) + entity_state = hass.states.get("sensor.test") + assert entity_state + assert entity_state.state == "unknown" - @patch("homeassistant.components.command_line.sensor._LOGGER") - def test_update_with_json_attrs_bad_JSON(self, mock_logger): - """Test attributes get extracted from a JSON result.""" - data = command_line.CommandSensorData( - self.hass, "echo This is text rather than JSON data.", 15 - ) - self.sensor = command_line.CommandSensor( - self.hass, data, "test", None, None, ["key"] - ) - self.sensor.update() - assert {} == self.sensor.device_state_attributes - assert mock_logger.warning.called - def test_update_with_missing_json_attrs(self): - """Test attributes get extracted from a JSON result.""" - data = command_line.CommandSensorData( - self.hass, - ( - 'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\":\ - \\"another_json_value\\", \\"key_three\\": \\"value_three\\" }' - ), - 15, - ) +async def test_update_with_json_attrs(hass: HomeAssistantType) -> None: + """Test attributes get extracted from a JSON result.""" + 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"], + }, + ) + entity_state = hass.states.get("sensor.test") + assert entity_state + assert entity_state.attributes["key"] == "some_json_value" + assert entity_state.attributes["another_key"] == "another_json_value" + assert entity_state.attributes["key_three"] == "value_three" - self.sensor = command_line.CommandSensor( - self.hass, - data, - "test", - None, - None, - ["key", "another_key", "key_three", "special_key"], - ) - self.sensor.update() - assert self.sensor.device_state_attributes["key"] == "some_json_value" - assert ( - self.sensor.device_state_attributes["another_key"] == "another_json_value" - ) - assert self.sensor.device_state_attributes["key_three"] == "value_three" - assert "special_key" not in self.sensor.device_state_attributes - def test_update_with_unnecessary_json_attrs(self): - """Test attributes get extracted from a JSON result.""" - data = command_line.CommandSensorData( - self.hass, - ( - 'echo { \\"key\\": \\"some_json_value\\", \\"another_key\\":\ - \\"another_json_value\\", \\"key_three\\": \\"value_three\\" }' - ), - 15, - ) +async def test_update_with_json_attrs_no_data(caplog, hass: HomeAssistantType) -> None: # type: ignore[no-untyped-def] + """Test attributes when no JSON result fetched.""" - self.sensor = command_line.CommandSensor( - self.hass, data, "test", None, None, ["key", "another_key"] - ) - self.sensor.update() - assert self.sensor.device_state_attributes["key"] == "some_json_value" - assert ( - self.sensor.device_state_attributes["another_key"] == "another_json_value" - ) - assert "key_three" not in self.sensor.device_state_attributes + await setup_test_entities( + hass, + { + "command": "echo", + "json_attributes": ["key"], + }, + ) + entity_state = hass.states.get("sensor.test") + assert entity_state + assert "key" not in entity_state.attributes + assert "Empty reply found when expecting JSON data" in caplog.text + + +async def test_update_with_json_attrs_not_dict(caplog, hass: HomeAssistantType) -> None: # type: ignore[no-untyped-def] + """Test attributes when the return value not a dict.""" + + await setup_test_entities( + hass, + { + "command": "echo [1, 2, 3]", + "json_attributes": ["key"], + }, + ) + entity_state = hass.states.get("sensor.test") + assert entity_state + assert "key" not in entity_state.attributes + assert "JSON result was not a dictionary" in caplog.text + + +async def test_update_with_json_attrs_bad_json(caplog, hass: HomeAssistantType) -> None: # type: ignore[no-untyped-def] + """Test attributes when the return value is invalid JSON.""" + + await setup_test_entities( + hass, + { + "command": "echo This is text rather than JSON data.", + "json_attributes": ["key"], + }, + ) + entity_state = hass.states.get("sensor.test") + assert entity_state + assert "key" not in entity_state.attributes + assert "Unable to parse output as JSON" in caplog.text + + +async def test_update_with_missing_json_attrs(caplog, hass: HomeAssistantType) -> None: # type: ignore[no-untyped-def] + """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") + assert entity_state + assert entity_state.attributes["key"] == "some_json_value" + assert entity_state.attributes["another_key"] == "another_json_value" + assert entity_state.attributes["key_three"] == "value_three" + assert "missing_key" not in entity_state.attributes + + +async def test_update_with_unnecessary_json_attrs(caplog, hass: HomeAssistantType) -> None: # type: ignore[no-untyped-def] + """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") + assert entity_state + assert entity_state.attributes["key"] == "some_json_value" + assert entity_state.attributes["another_key"] == "another_json_value" + assert "key_three" not in entity_state.attributes diff --git a/tests/components/command_line/test_switch.py b/tests/components/command_line/test_switch.py index c6d315b05b5..0e31999f928 100644 --- a/tests/components/command_line/test_switch.py +++ b/tests/components/command_line/test_switch.py @@ -1,10 +1,12 @@ """The tests for the Command line switch platform.""" import json import os +import subprocess import tempfile +from unittest.mock import patch -import homeassistant.components.command_line.switch as command_line -import homeassistant.components.switch as switch +from homeassistant import setup +from homeassistant.components.switch import DOMAIN, SCAN_INTERVAL from homeassistant.const import ( ATTR_ENTITY_ID, SERVICE_TURN_OFF, @@ -12,230 +14,358 @@ from homeassistant.const import ( STATE_OFF, STATE_ON, ) -from homeassistant.setup import async_setup_component +from homeassistant.helpers.typing import Any, Dict, HomeAssistantType +import homeassistant.util.dt as dt_util + +from tests.common import async_fire_time_changed -async def test_state_none(hass): +async def setup_test_entity( + hass: HomeAssistantType, config_dict: Dict[str, Any] +) -> 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: HomeAssistantType) -> None: """Test with none state.""" with tempfile.TemporaryDirectory() as tempdirname: path = os.path.join(tempdirname, "switch_status") - test_switch = { - "command_on": f"echo 1 > {path}", - "command_off": f"echo 0 > {path}", - } - assert await async_setup_component( + await setup_test_entity( hass, - switch.DOMAIN, { - "switch": { - "platform": "command_line", - "switches": {"test": test_switch}, + "test": { + "command_on": f"echo 1 > {path}", + "command_off": f"echo 0 > {path}", } }, ) - await hass.async_block_till_done() - state = hass.states.get("switch.test") - assert STATE_OFF == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_OFF await hass.services.async_call( - switch.DOMAIN, + DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: "switch.test"}, blocking=True, ) - state = hass.states.get("switch.test") - assert STATE_ON == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_ON await hass.services.async_call( - switch.DOMAIN, + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: "switch.test"}, blocking=True, ) - state = hass.states.get("switch.test") - assert STATE_OFF == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_OFF -async def test_state_value(hass): +async def test_state_value(hass: HomeAssistantType) -> None: """Test with state value.""" with tempfile.TemporaryDirectory() as tempdirname: path = os.path.join(tempdirname, "switch_status") - test_switch = { - "command_state": f"cat {path}", - "command_on": f"echo 1 > {path}", - "command_off": f"echo 0 > {path}", - "value_template": '{{ value=="1" }}', - } - assert await async_setup_component( + await setup_test_entity( hass, - switch.DOMAIN, { - "switch": { - "platform": "command_line", - "switches": {"test": test_switch}, + "test": { + "command_state": f"cat {path}", + "command_on": f"echo 1 > {path}", + "command_off": f"echo 0 > {path}", + "value_template": '{{ value=="1" }}', } }, ) - await hass.async_block_till_done() - state = hass.states.get("switch.test") - assert STATE_OFF == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_OFF await hass.services.async_call( - switch.DOMAIN, + DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: "switch.test"}, blocking=True, ) - state = hass.states.get("switch.test") - assert STATE_ON == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_ON await hass.services.async_call( - switch.DOMAIN, + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: "switch.test"}, blocking=True, ) - state = hass.states.get("switch.test") - assert STATE_OFF == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_OFF -async def test_state_json_value(hass): +async def test_state_json_value(hass: HomeAssistantType) -> None: """Test with state JSON value.""" with tempfile.TemporaryDirectory() as tempdirname: path = os.path.join(tempdirname, "switch_status") oncmd = json.dumps({"status": "ok"}) offcmd = json.dumps({"status": "nope"}) - test_switch = { - "command_state": f"cat {path}", - "command_on": f"echo '{oncmd}' > {path}", - "command_off": f"echo '{offcmd}' > {path}", - "value_template": '{{ value_json.status=="ok" }}', - } - assert await async_setup_component( + + await setup_test_entity( hass, - switch.DOMAIN, { - "switch": { - "platform": "command_line", - "switches": {"test": test_switch}, + "test": { + "command_state": f"cat {path}", + "command_on": f"echo '{oncmd}' > {path}", + "command_off": f"echo '{offcmd}' > {path}", + "value_template": '{{ value_json.status=="ok" }}', } }, ) - await hass.async_block_till_done() - state = hass.states.get("switch.test") - assert STATE_OFF == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_OFF await hass.services.async_call( - switch.DOMAIN, + DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: "switch.test"}, blocking=True, ) - state = hass.states.get("switch.test") - assert STATE_ON == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_ON await hass.services.async_call( - switch.DOMAIN, + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: "switch.test"}, blocking=True, ) - state = hass.states.get("switch.test") - assert STATE_OFF == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_OFF -async def test_state_code(hass): +async def test_state_code(hass: HomeAssistantType) -> None: """Test with state code.""" with tempfile.TemporaryDirectory() as tempdirname: path = os.path.join(tempdirname, "switch_status") - test_switch = { - "command_state": f"cat {path}", - "command_on": f"echo 1 > {path}", - "command_off": f"echo 0 > {path}", - } - assert await async_setup_component( + await setup_test_entity( hass, - switch.DOMAIN, { - "switch": { - "platform": "command_line", - "switches": {"test": test_switch}, + "test": { + "command_state": f"cat {path}", + "command_on": f"echo 1 > {path}", + "command_off": f"echo 0 > {path}", } }, ) - await hass.async_block_till_done() - state = hass.states.get("switch.test") - assert STATE_OFF == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_OFF await hass.services.async_call( - switch.DOMAIN, + DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: "switch.test"}, blocking=True, ) - state = hass.states.get("switch.test") - assert STATE_ON == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_ON await hass.services.async_call( - switch.DOMAIN, + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: "switch.test"}, blocking=True, ) - state = hass.states.get("switch.test") - assert STATE_ON == state.state + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.state == STATE_ON -def test_assumed_state_should_be_true_if_command_state_is_none(hass): +async def test_assumed_state_should_be_true_if_command_state_is_none( + hass: HomeAssistantType, +) -> None: """Test with state value.""" - # args: hass, device_name, friendly_name, command_on, command_off, - # command_state, value_template - init_args = [ + + await setup_test_entity( hass, - "test_device_name", - "Test friendly name!", - "echo 'on command'", - "echo 'off command'", - None, - None, - 15, - ] - - no_state_device = command_line.CommandSwitch(*init_args) - assert no_state_device.assumed_state - - # Set state command - init_args[-3] = "cat {}" - - state_device = command_line.CommandSwitch(*init_args) - assert not state_device.assumed_state + { + "test": { + "command_on": "echo 'on command'", + "command_off": "echo 'off command'", + } + }, + ) + entity_state = hass.states.get("switch.test") + assert entity_state + assert entity_state.attributes["assumed_state"] -def test_entity_id_set_correctly(hass): - """Test that entity_id is set correctly from object_id.""" - init_args = [ +async def test_assumed_state_should_absent_if_command_state_present( + hass: HomeAssistantType, +) -> None: + """Test with state value.""" + + await setup_test_entity( hass, - "test_device_name", - "Test friendly name!", - "echo 'on command'", - "echo 'off command'", - False, - None, - 15, - ] + { + "test": { + "command_on": "echo 'on command'", + "command_off": "echo 'off command'", + "command_state": "cat {}", + } + }, + ) + entity_state = hass.states.get("switch.test") + assert entity_state + assert "assumed_state" not in entity_state.attributes - test_switch = command_line.CommandSwitch(*init_args) - assert test_switch.entity_id == "switch.test_device_name" - assert test_switch.name == "Test friendly name!" + +async def test_name_is_set_correctly(hass: HomeAssistantType) -> None: + """Test that name is set correctly.""" + await setup_test_entity( + hass, + { + "test": { + "command_on": "echo 'on command'", + "command_off": "echo 'off command'", + "friendly_name": "Test friendly name!", + } + }, + ) + + entity_state = hass.states.get("switch.test") + assert entity_state.name == "Test friendly name!" + + +async def test_switch_command_state_fail(caplog: Any, hass: HomeAssistantType) -> None: + """Test that switch failures are handled correctly.""" + await setup_test_entity( + hass, + { + "test": { + "command_on": "exit 0", + "command_off": "exit 0'", + "command_state": "echo 1", + } + }, + ) + + async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) + await hass.async_block_till_done() + + entity_state = hass.states.get("switch.test") + assert entity_state.state == "on" + + await hass.services.async_call( + DOMAIN, + SERVICE_TURN_OFF, + {ATTR_ENTITY_ID: "switch.test"}, + blocking=True, + ) + await hass.async_block_till_done() + + entity_state = hass.states.get("switch.test") + assert entity_state.state == "on" + + assert "Command failed" in caplog.text + + +async def test_switch_command_state_code_exceptions( + caplog: Any, hass: HomeAssistantType +) -> None: + """Test that switch state code exceptions are handled correctly.""" + + with patch( + "homeassistant.components.command_line.subprocess.check_output", + side_effect=[ + subprocess.TimeoutExpired("cmd", 10), + subprocess.SubprocessError(), + ], + ) as check_output: + await setup_test_entity( + hass, + { + "test": { + "command_on": "exit 0", + "command_off": "exit 0'", + "command_state": "echo 1", + } + }, + ) + async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) + await hass.async_block_till_done() + assert check_output.called + assert "Timeout for command" in caplog.text + + async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL * 2) + await hass.async_block_till_done() + assert check_output.called + assert "Error trying to exec command" in caplog.text + + +async def test_switch_command_state_value_exceptions( + caplog: Any, hass: HomeAssistantType +) -> None: + """Test that switch state value exceptions are handled correctly.""" + + with patch( + "homeassistant.components.command_line.subprocess.check_output", + side_effect=[ + subprocess.TimeoutExpired("cmd", 10), + subprocess.SubprocessError(), + ], + ) as check_output: + await setup_test_entity( + hass, + { + "test": { + "command_on": "exit 0", + "command_off": "exit 0'", + "command_state": "echo 1", + "value_template": '{{ value=="1" }}', + } + }, + ) + async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) + await hass.async_block_till_done() + assert check_output.call_count == 1 + assert "Timeout for command" in caplog.text + + async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL * 2) + await hass.async_block_till_done() + assert check_output.call_count == 2 + assert "Error trying to exec command" in caplog.text + + +async def test_no_switches(caplog: Any, hass: HomeAssistantType) -> None: + """Test with no switches.""" + + await setup_test_entity(hass, {}) + assert "No switches" in caplog.text