jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Overhaul command_line tests (#46682)

Browse Source
This commit is contained in:
Dermot Duffy 2021-03-01 08:27:04 -08:00 committed by GitHub
parent 3ebd5aff98
commit be8584c0bc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 699 additions and 535 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
homeassistant/components/command_line/cover.py
View File

@ -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."""

7
homeassistant/components/command_line/sensor.py
View File

@ -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:

3
homeassistant/components/command_line/switch.py
View File

@ -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)

101
tests/components/command_line/test_binary_sensor.py
View File

@ -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

171
tests/components/command_line/test_cover.py
View File

@ -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

204
tests/components/command_line/test_notify.py
View File

@ -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

369
tests/components/command_line/test_sensor.py
View File

@ -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

368
tests/components/command_line/test_switch.py
View File

@ -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