mirror of
https://github.com/home-assistant/core.git
synced 2025-07-27 07:07:28 +00:00
Improve command_line switch tests (#40749)
This commit is contained in:
Franck Nijhof
GitHub
parent
fc0344399b
commit
27350f41c9
@ -2,209 +2,240 @@
|
|||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import tempfile
|
import tempfile
|
||||||
import unittest
|
|
||||||
|
|
||||||
import homeassistant.components.command_line.switch as command_line
|
import homeassistant.components.command_line.switch as command_line
|
||||||
import homeassistant.components.switch as switch
|
import homeassistant.components.switch as switch
|
||||||
from homeassistant.const import STATE_OFF, STATE_ON
|
from homeassistant.const import (
|
||||||
from homeassistant.setup import setup_component
|
ATTR_ENTITY_ID,
|
||||||
|
SERVICE_TURN_OFF,
|
||||||
from tests.common import get_test_home_assistant
|
SERVICE_TURN_ON,
|
||||||
from tests.components.switch import common
|
STATE_OFF,
|
||||||
|
STATE_ON,
|
||||||
|
)
|
||||||
|
from homeassistant.setup import async_setup_component
|
||||||
|
|
||||||
|
|
||||||
# pylint: disable=invalid-name
|
async def test_state_none(hass):
|
||||||
class TestCommandSwitch(unittest.TestCase):
|
"""Test with none state."""
|
||||||
"""Test the command switch."""
|
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(
|
||||||
|
hass,
|
||||||
|
switch.DOMAIN,
|
||||||
|
{
|
||||||
|
"switch": {
|
||||||
|
"platform": "command_line",
|
||||||
|
"switches": {"test": test_switch},
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
def setUp(self):
|
state = hass.states.get("switch.test")
|
||||||
"""Set up things to be run when tests are started."""
|
assert STATE_OFF == state.state
|
||||||
self.hass = get_test_home_assistant()
|
|
||||||
self.addCleanup(self.hass.stop)
|
|
||||||
|
|
||||||
def test_state_none(self):
|
await hass.services.async_call(
|
||||||
"""Test with none state."""
|
switch.DOMAIN,
|
||||||
with tempfile.TemporaryDirectory() as tempdirname:
|
SERVICE_TURN_ON,
|
||||||
path = os.path.join(tempdirname, "switch_status")
|
{ATTR_ENTITY_ID: "switch.test"},
|
||||||
test_switch = {
|
blocking=True,
|
||||||
"command_on": f"echo 1 > {path}",
|
)
|
||||||
"command_off": f"echo 0 > {path}",
|
|
||||||
}
|
|
||||||
assert setup_component(
|
|
||||||
self.hass,
|
|
||||||
switch.DOMAIN,
|
|
||||||
{
|
|
||||||
"switch": {
|
|
||||||
"platform": "command_line",
|
|
||||||
"switches": {"test": test_switch},
|
|
||||||
}
|
|
||||||
},
|
|
||||||
)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
state = hass.states.get("switch.test")
|
||||||
assert STATE_OFF == state.state
|
assert STATE_ON == state.state
|
||||||
|
|
||||||
common.turn_on(self.hass, "switch.test")
|
await hass.services.async_call(
|
||||||
self.hass.block_till_done()
|
switch.DOMAIN,
|
||||||
|
SERVICE_TURN_OFF,
|
||||||
|
{ATTR_ENTITY_ID: "switch.test"},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
state = hass.states.get("switch.test")
|
||||||
assert STATE_ON == state.state
|
assert STATE_OFF == state.state
|
||||||
|
|
||||||
common.turn_off(self.hass, "switch.test")
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
async def test_state_value(hass):
|
||||||
assert STATE_OFF == state.state
|
"""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(
|
||||||
|
hass,
|
||||||
|
switch.DOMAIN,
|
||||||
|
{
|
||||||
|
"switch": {
|
||||||
|
"platform": "command_line",
|
||||||
|
"switches": {"test": test_switch},
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
def test_state_value(self):
|
state = hass.states.get("switch.test")
|
||||||
"""Test with state value."""
|
assert STATE_OFF == state.state
|
||||||
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 setup_component(
|
|
||||||
self.hass,
|
|
||||||
switch.DOMAIN,
|
|
||||||
{
|
|
||||||
"switch": {
|
|
||||||
"platform": "command_line",
|
|
||||||
"switches": {"test": test_switch},
|
|
||||||
}
|
|
||||||
},
|
|
||||||
)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
await hass.services.async_call(
|
||||||
assert STATE_OFF == state.state
|
switch.DOMAIN,
|
||||||
|
SERVICE_TURN_ON,
|
||||||
|
{ATTR_ENTITY_ID: "switch.test"},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
|
||||||
common.turn_on(self.hass, "switch.test")
|
state = hass.states.get("switch.test")
|
||||||
self.hass.block_till_done()
|
assert STATE_ON == state.state
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
await hass.services.async_call(
|
||||||
assert STATE_ON == state.state
|
switch.DOMAIN,
|
||||||
|
SERVICE_TURN_OFF,
|
||||||
|
{ATTR_ENTITY_ID: "switch.test"},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
|
||||||
common.turn_off(self.hass, "switch.test")
|
state = hass.states.get("switch.test")
|
||||||
self.hass.block_till_done()
|
assert STATE_OFF == state.state
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
|
||||||
assert STATE_OFF == state.state
|
|
||||||
|
|
||||||
def test_state_json_value(self):
|
async def test_state_json_value(hass):
|
||||||
"""Test with state JSON value."""
|
"""Test with state JSON 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")
|
||||||
oncmd = json.dumps({"status": "ok"})
|
oncmd = json.dumps({"status": "ok"})
|
||||||
offcmd = json.dumps({"status": "nope"})
|
offcmd = json.dumps({"status": "nope"})
|
||||||
test_switch = {
|
test_switch = {
|
||||||
"command_state": f"cat {path}",
|
"command_state": f"cat {path}",
|
||||||
"command_on": f"echo '{oncmd}' > {path}",
|
"command_on": f"echo '{oncmd}' > {path}",
|
||||||
"command_off": f"echo '{offcmd}' > {path}",
|
"command_off": f"echo '{offcmd}' > {path}",
|
||||||
"value_template": '{{ value_json.status=="ok" }}',
|
"value_template": '{{ value_json.status=="ok" }}',
|
||||||
}
|
}
|
||||||
assert setup_component(
|
assert await async_setup_component(
|
||||||
self.hass,
|
hass,
|
||||||
switch.DOMAIN,
|
switch.DOMAIN,
|
||||||
{
|
{
|
||||||
"switch": {
|
"switch": {
|
||||||
"platform": "command_line",
|
"platform": "command_line",
|
||||||
"switches": {"test": test_switch},
|
"switches": {"test": test_switch},
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
self.hass.block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
state = hass.states.get("switch.test")
|
||||||
assert STATE_OFF == state.state
|
assert STATE_OFF == state.state
|
||||||
|
|
||||||
common.turn_on(self.hass, "switch.test")
|
await hass.services.async_call(
|
||||||
self.hass.block_till_done()
|
switch.DOMAIN,
|
||||||
|
SERVICE_TURN_ON,
|
||||||
|
{ATTR_ENTITY_ID: "switch.test"},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
state = hass.states.get("switch.test")
|
||||||
assert STATE_ON == state.state
|
assert STATE_ON == state.state
|
||||||
|
|
||||||
common.turn_off(self.hass, "switch.test")
|
await hass.services.async_call(
|
||||||
self.hass.block_till_done()
|
switch.DOMAIN,
|
||||||
|
SERVICE_TURN_OFF,
|
||||||
|
{ATTR_ENTITY_ID: "switch.test"},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
state = hass.states.get("switch.test")
|
||||||
assert STATE_OFF == state.state
|
assert STATE_OFF == state.state
|
||||||
|
|
||||||
def test_state_code(self):
|
|
||||||
"""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 setup_component(
|
|
||||||
self.hass,
|
|
||||||
switch.DOMAIN,
|
|
||||||
{
|
|
||||||
"switch": {
|
|
||||||
"platform": "command_line",
|
|
||||||
"switches": {"test": test_switch},
|
|
||||||
}
|
|
||||||
},
|
|
||||||
)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
state = self.hass.states.get("switch.test")
|
|
||||||
assert STATE_OFF == state.state
|
|
||||||
|
|
||||||
common.turn_on(self.hass, "switch.test")
|
async def test_state_code(hass):
|
||||||
self.hass.block_till_done()
|
"""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(
|
||||||
|
hass,
|
||||||
|
switch.DOMAIN,
|
||||||
|
{
|
||||||
|
"switch": {
|
||||||
|
"platform": "command_line",
|
||||||
|
"switches": {"test": test_switch},
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
state = hass.states.get("switch.test")
|
||||||
assert STATE_ON == state.state
|
assert STATE_OFF == state.state
|
||||||
|
|
||||||
common.turn_off(self.hass, "switch.test")
|
await hass.services.async_call(
|
||||||
self.hass.block_till_done()
|
switch.DOMAIN,
|
||||||
|
SERVICE_TURN_ON,
|
||||||
|
{ATTR_ENTITY_ID: "switch.test"},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
|
||||||
state = self.hass.states.get("switch.test")
|
state = hass.states.get("switch.test")
|
||||||
assert STATE_ON == state.state
|
assert STATE_ON == state.state
|
||||||
|
|
||||||
def test_assumed_state_should_be_true_if_command_state_is_none(self):
|
await hass.services.async_call(
|
||||||
"""Test with state value."""
|
switch.DOMAIN,
|
||||||
# args: hass, device_name, friendly_name, command_on, command_off,
|
SERVICE_TURN_OFF,
|
||||||
# command_state, value_template
|
{ATTR_ENTITY_ID: "switch.test"},
|
||||||
init_args = [
|
blocking=True,
|
||||||
self.hass,
|
)
|
||||||
"test_device_name",
|
|
||||||
"Test friendly name!",
|
|
||||||
"echo 'on command'",
|
|
||||||
"echo 'off command'",
|
|
||||||
None,
|
|
||||||
None,
|
|
||||||
15,
|
|
||||||
]
|
|
||||||
|
|
||||||
no_state_device = command_line.CommandSwitch(*init_args)
|
state = hass.states.get("switch.test")
|
||||||
assert no_state_device.assumed_state
|
assert STATE_ON == state.state
|
||||||
|
|
||||||
# Set state command
|
|
||||||
init_args[-3] = "cat {}"
|
|
||||||
|
|
||||||
state_device = command_line.CommandSwitch(*init_args)
|
def test_assumed_state_should_be_true_if_command_state_is_none(hass):
|
||||||
assert not state_device.assumed_state
|
"""Test with state value."""
|
||||||
|
# args: hass, device_name, friendly_name, command_on, command_off,
|
||||||
|
# command_state, value_template
|
||||||
|
init_args = [
|
||||||
|
hass,
|
||||||
|
"test_device_name",
|
||||||
|
"Test friendly name!",
|
||||||
|
"echo 'on command'",
|
||||||
|
"echo 'off command'",
|
||||||
|
None,
|
||||||
|
None,
|
||||||
|
15,
|
||||||
|
]
|
||||||
|
|
||||||
def test_entity_id_set_correctly(self):
|
no_state_device = command_line.CommandSwitch(*init_args)
|
||||||
"""Test that entity_id is set correctly from object_id."""
|
assert no_state_device.assumed_state
|
||||||
init_args = [
|
|
||||||
self.hass,
|
|
||||||
"test_device_name",
|
|
||||||
"Test friendly name!",
|
|
||||||
"echo 'on command'",
|
|
||||||
"echo 'off command'",
|
|
||||||
False,
|
|
||||||
None,
|
|
||||||
15,
|
|
||||||
]
|
|
||||||
|
|
||||||
test_switch = command_line.CommandSwitch(*init_args)
|
# Set state command
|
||||||
assert test_switch.entity_id == "switch.test_device_name"
|
init_args[-3] = "cat {}"
|
||||||
assert test_switch.name == "Test friendly name!"
|
|
||||||
|
state_device = command_line.CommandSwitch(*init_args)
|
||||||
|
assert not state_device.assumed_state
|
||||||
|
|
||||||
|
|
||||||
|
def test_entity_id_set_correctly(hass):
|
||||||
|
"""Test that entity_id is set correctly from object_id."""
|
||||||
|
init_args = [
|
||||||
|
hass,
|
||||||
|
"test_device_name",
|
||||||
|
"Test friendly name!",
|
||||||
|
"echo 'on command'",
|
||||||
|
"echo 'off command'",
|
||||||
|
False,
|
||||||
|
None,
|
||||||
|
15,
|
||||||
|
]
|
||||||
|
|
||||||
|
test_switch = command_line.CommandSwitch(*init_args)
|
||||||
|
assert test_switch.entity_id == "switch.test_device_name"
|
||||||
|
assert test_switch.name == "Test friendly name!"
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user