github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add tests for LCN sensor and binary_sensor platforms (#67263)

This commit is contained in:
Andre Lengwenus 2022-06-29 06:59:19 +02:00 committed by GitHub
parent 22b8afe966
commit b7b8feda0f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 532 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -633,11 +633,9 @@ omit =
homeassistant/components/launch_library/const.py
homeassistant/components/launch_library/diagnostics.py
homeassistant/components/launch_library/sensor.py
homeassistant/components/lcn/binary_sensor.py
homeassistant/components/lcn/climate.py
homeassistant/components/lcn/helpers.py
homeassistant/components/lcn/scene.py
homeassistant/components/lcn/sensor.py
homeassistant/components/lcn/services.py
homeassistant/components/lg_netcast/media_player.py
homeassistant/components/lg_soundbar/media_player.py

41
tests/components/lcn/fixtures/config.json
View File

@ -90,6 +90,47 @@
"address": "s0.m7",
"motor": "motor1"
}
],
"binary_sensors": [
{
"name": "Sensor_LockRegulator1",
"address": "s0.m7",
"source": "r1varsetpoint"
},
{
"name": "Binary_Sensor1",
"address": "s0.m7",
"source": "binsensor1"
},
{
"name": "Sensor_KeyLock",
"address": "s0.m7",
"source": "a5"
}
],
"sensors": [
{
"name": "Sensor_Var1",
"address": "s0.m7",
"source": "var1",
"unit_of_measurement": "°C"
},
{
"name": "Sensor_Setpoint1",
"address": "s0.m7",
"source": "r1varsetpoint",
"unit_of_measurement": "°C"
},
{
"name": "Sensor_Led6",
"address": "s0.m7",
"source": "led6"
},
{
"name": "Sensor_LogicOp1",
"address": "s0.m7",
"source": "logicop1"
}
]
}
}

67
tests/components/lcn/fixtures/config_entry_pchk.json
View File

@ -120,6 +120,73 @@
"motor": "MOTOR1",
"reverse_time": "RT1200"
}
},
{
"address": [0, 7, false],
"name": "Sensor_LockRegulator1",
"resource": "r1varsetpoint",
"domain": "binary_sensor",
"domain_data": {
"source": "R1VARSETPOINT"
}
},
{
"address": [0, 7, false],
"name": "Binary_Sensor1",
"resource": "binsensor1",
"domain": "binary_sensor",
"domain_data": {
"source": "BINSENSOR1"
}
},
{
"address": [0, 7, false],
"name": "Sensor_KeyLock",
"resource": "a5",
"domain": "binary_sensor",
"domain_data": {
"source": "A5"
}
},
{
"address": [0, 7, false],
"name": "Sensor_Var1",
"resource": "var1",
"domain": "sensor",
"domain_data": {
"source": "VAR1",
"unit_of_measurement": "°C"
}
},
{
"address": [0, 7, false],
"name": "Sensor_Setpoint1",
"resource": "r1varsetpoint",
"domain": "sensor",
"domain_data": {
"source": "R1VARSETPOINT",
"unit_of_measurement": "°C"
}
},
{
"address": [0, 7, false],
"name": "Sensor_Led6",
"resource": "led6",
"domain": "sensor",
"domain_data": {
"source": "LED6",
"unit_of_measurement": "NATIVE"
}
},
{
"address": [0, 7, false],
"name": "Sensor_LogicOp1",
"resource": "logicop1",
"domain": "sensor",
"domain_data": {
"source": "LOGICOP1",
"unit_of_measurement": "NATIVE"
}
}
]
}

140
tests/components/lcn/test_binary_sensor.py Normal file
View File

@ -0,0 +1,140 @@
"""Test for the LCN binary sensor platform."""
from pypck.inputs import ModStatusBinSensors, ModStatusKeyLocks, ModStatusVar
from pypck.lcn_addr import LcnAddr
from pypck.lcn_defs import Var, VarValue
from homeassistant.components.lcn.helpers import get_device_connection
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.helpers import entity_registry as er
BINARY_SENSOR_LOCKREGULATOR1 = "binary_sensor.sensor_lockregulator1"
BINARY_SENSOR_SENSOR1 = "binary_sensor.binary_sensor1"
BINARY_SENSOR_KEYLOCK = "binary_sensor.sensor_keylock"
async def test_setup_lcn_binary_sensor(hass, lcn_connection):
"""Test the setup of binary sensor."""
for entity_id in (
BINARY_SENSOR_LOCKREGULATOR1,
BINARY_SENSOR_SENSOR1,
BINARY_SENSOR_KEYLOCK,
):
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_UNKNOWN
async def test_entity_state(hass, lcn_connection):
"""Test state of entity."""
state = hass.states.get(BINARY_SENSOR_LOCKREGULATOR1)
assert state
state = hass.states.get(BINARY_SENSOR_SENSOR1)
assert state
state = hass.states.get(BINARY_SENSOR_KEYLOCK)
assert state
async def test_entity_attributes(hass, entry, lcn_connection):
"""Test the attributes of an entity."""
entity_registry = er.async_get(hass)
entity_setpoint1 = entity_registry.async_get(BINARY_SENSOR_LOCKREGULATOR1)
assert entity_setpoint1
assert entity_setpoint1.unique_id == f"{entry.entry_id}-m000007-r1varsetpoint"
assert entity_setpoint1.original_name == "Sensor_LockRegulator1"
entity_binsensor1 = entity_registry.async_get(BINARY_SENSOR_SENSOR1)
assert entity_binsensor1
assert entity_binsensor1.unique_id == f"{entry.entry_id}-m000007-binsensor1"
assert entity_binsensor1.original_name == "Binary_Sensor1"
entity_keylock = entity_registry.async_get(BINARY_SENSOR_KEYLOCK)
assert entity_keylock
assert entity_keylock.unique_id == f"{entry.entry_id}-m000007-a5"
assert entity_keylock.original_name == "Sensor_KeyLock"
async def test_pushed_lock_setpoint_status_change(hass, entry, lcn_connection):
"""Test the lock setpoint sensor changes its state on status received."""
device_connection = get_device_connection(hass, (0, 7, False), entry)
address = LcnAddr(0, 7, False)
# push status lock setpoint
inp = ModStatusVar(address, Var.R1VARSETPOINT, VarValue(0x8000))
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_LOCKREGULATOR1)
assert state is not None
assert state.state == STATE_ON
# push status unlock setpoint
inp = ModStatusVar(address, Var.R1VARSETPOINT, VarValue(0x7FFF))
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_LOCKREGULATOR1)
assert state is not None
assert state.state == STATE_OFF
async def test_pushed_binsensor_status_change(hass, entry, lcn_connection):
"""Test the binary port sensor changes its state on status received."""
device_connection = get_device_connection(hass, (0, 7, False), entry)
address = LcnAddr(0, 7, False)
states = [False] * 8
# push status binary port "off"
inp = ModStatusBinSensors(address, states)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_SENSOR1)
assert state is not None
assert state.state == STATE_OFF
# push status binary port "on"
states[0] = True
inp = ModStatusBinSensors(address, states)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_SENSOR1)
assert state is not None
assert state.state == STATE_ON
async def test_pushed_keylock_status_change(hass, entry, lcn_connection):
"""Test the keylock sensor changes its state on status received."""
device_connection = get_device_connection(hass, (0, 7, False), entry)
address = LcnAddr(0, 7, False)
states = [[False] * 8 for i in range(4)]
# push status keylock "off"
inp = ModStatusKeyLocks(address, states)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_KEYLOCK)
assert state is not None
assert state.state == STATE_OFF
# push status keylock "on"
states[0][4] = True
inp = ModStatusKeyLocks(address, states)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get(BINARY_SENSOR_KEYLOCK)
assert state is not None
assert state.state == STATE_ON
async def test_unload_config_entry(hass, entry, lcn_connection):
"""Test the binary sensor is removed when the config entry is unloaded."""
await hass.config_entries.async_unload(entry.entry_id)
assert hass.states.get(BINARY_SENSOR_LOCKREGULATOR1).state == STATE_UNAVAILABLE
assert hass.states.get(BINARY_SENSOR_SENSOR1).state == STATE_UNAVAILABLE
assert hass.states.get(BINARY_SENSOR_KEYLOCK).state == STATE_UNAVAILABLE

115
tests/components/lcn/test_cover.py
View File

@ -22,12 +22,15 @@ from homeassistant.helpers import entity_registry as er
from .conftest import MockModuleConnection
COVER_OUTPUTS = "cover.cover_outputs"
COVER_RELAYS = "cover.cover_relays"
async def test_setup_lcn_cover(hass, entry, lcn_connection):
"""Test the setup of cover."""
for entity_id in (
"cover.cover_outputs",
"cover.cover_relays",
COVER_OUTPUTS,
COVER_RELAYS,
):
state = hass.states.get(entity_id)
assert state is not None
@ -38,13 +41,13 @@ async def test_entity_attributes(hass, entry, lcn_connection):
"""Test the attributes of an entity."""
entity_registry = er.async_get(hass)
entity_outputs = entity_registry.async_get("cover.cover_outputs")
entity_outputs = entity_registry.async_get(COVER_OUTPUTS)
assert entity_outputs
assert entity_outputs.unique_id == f"{entry.entry_id}-m000007-outputs"
assert entity_outputs.original_name == "Cover_Outputs"
entity_relays = entity_registry.async_get("cover.cover_relays")
entity_relays = entity_registry.async_get(COVER_RELAYS)
assert entity_relays
assert entity_relays.unique_id == f"{entry.entry_id}-m000007-motor1"
@ -54,7 +57,7 @@ async def test_entity_attributes(hass, entry, lcn_connection):
@patch.object(MockModuleConnection, "control_motors_outputs")
async def test_outputs_open(control_motors_outputs, hass, lcn_connection):
"""Test the outputs cover opens."""
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
state.state = STATE_CLOSED
# command failed
@ -63,7 +66,7 @@ async def test_outputs_open(control_motors_outputs, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: "cover.cover_outputs"},
{ATTR_ENTITY_ID: COVER_OUTPUTS},
blocking=True,
)
await hass.async_block_till_done()
@ -71,7 +74,7 @@ async def test_outputs_open(control_motors_outputs, hass, lcn_connection):
MotorStateModifier.UP, MotorReverseTime.RT1200
)
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
assert state is not None
assert state.state != STATE_OPENING
@ -82,7 +85,7 @@ async def test_outputs_open(control_motors_outputs, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: "cover.cover_outputs"},
{ATTR_ENTITY_ID: COVER_OUTPUTS},
blocking=True,
)
await hass.async_block_till_done()
@ -90,7 +93,7 @@ async def test_outputs_open(control_motors_outputs, hass, lcn_connection):
MotorStateModifier.UP, MotorReverseTime.RT1200
)
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
assert state is not None
assert state.state == STATE_OPENING
@ -98,7 +101,7 @@ async def test_outputs_open(control_motors_outputs, hass, lcn_connection):
@patch.object(MockModuleConnection, "control_motors_outputs")
async def test_outputs_close(control_motors_outputs, hass, lcn_connection):
"""Test the outputs cover closes."""
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
state.state = STATE_OPEN
# command failed
@ -107,7 +110,7 @@ async def test_outputs_close(control_motors_outputs, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: "cover.cover_outputs"},
{ATTR_ENTITY_ID: COVER_OUTPUTS},
blocking=True,
)
await hass.async_block_till_done()
@ -115,7 +118,7 @@ async def test_outputs_close(control_motors_outputs, hass, lcn_connection):
MotorStateModifier.DOWN, MotorReverseTime.RT1200
)
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
assert state is not None
assert state.state != STATE_CLOSING
@ -126,7 +129,7 @@ async def test_outputs_close(control_motors_outputs, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: "cover.cover_outputs"},
{ATTR_ENTITY_ID: COVER_OUTPUTS},
blocking=True,
)
await hass.async_block_till_done()
@ -134,7 +137,7 @@ async def test_outputs_close(control_motors_outputs, hass, lcn_connection):
MotorStateModifier.DOWN, MotorReverseTime.RT1200
)
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
assert state is not None
assert state.state == STATE_CLOSING
@ -142,7 +145,7 @@ async def test_outputs_close(control_motors_outputs, hass, lcn_connection):
@patch.object(MockModuleConnection, "control_motors_outputs")
async def test_outputs_stop(control_motors_outputs, hass, lcn_connection):
"""Test the outputs cover stops."""
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
state.state = STATE_CLOSING
# command failed
@ -151,13 +154,13 @@ async def test_outputs_stop(control_motors_outputs, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: "cover.cover_outputs"},
{ATTR_ENTITY_ID: COVER_OUTPUTS},
blocking=True,
)
await hass.async_block_till_done()
control_motors_outputs.assert_awaited_with(MotorStateModifier.STOP)
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
assert state is not None
assert state.state == STATE_CLOSING
@ -168,13 +171,13 @@ async def test_outputs_stop(control_motors_outputs, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: "cover.cover_outputs"},
{ATTR_ENTITY_ID: COVER_OUTPUTS},
blocking=True,
)
await hass.async_block_till_done()
control_motors_outputs.assert_awaited_with(MotorStateModifier.STOP)
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
assert state is not None
assert state.state not in (STATE_CLOSING, STATE_OPENING)
@ -185,7 +188,7 @@ async def test_relays_open(control_motors_relays, hass, lcn_connection):
states = [MotorStateModifier.NOCHANGE] * 4
states[0] = MotorStateModifier.UP
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
state.state = STATE_CLOSED
# command failed
@ -194,13 +197,13 @@ async def test_relays_open(control_motors_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: "cover.cover_relays"},
{ATTR_ENTITY_ID: COVER_RELAYS},
blocking=True,
)
await hass.async_block_till_done()
control_motors_relays.assert_awaited_with(states)
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
assert state is not None
assert state.state != STATE_OPENING
@ -211,13 +214,13 @@ async def test_relays_open(control_motors_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: "cover.cover_relays"},
{ATTR_ENTITY_ID: COVER_RELAYS},
blocking=True,
)
await hass.async_block_till_done()
control_motors_relays.assert_awaited_with(states)
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
assert state is not None
assert state.state == STATE_OPENING
@ -228,7 +231,7 @@ async def test_relays_close(control_motors_relays, hass, lcn_connection):
states = [MotorStateModifier.NOCHANGE] * 4
states[0] = MotorStateModifier.DOWN
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
state.state = STATE_OPEN
# command failed
@ -237,13 +240,13 @@ async def test_relays_close(control_motors_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: "cover.cover_relays"},
{ATTR_ENTITY_ID: COVER_RELAYS},
blocking=True,
)
await hass.async_block_till_done()
control_motors_relays.assert_awaited_with(states)
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
assert state is not None
assert state.state != STATE_CLOSING
@ -254,13 +257,13 @@ async def test_relays_close(control_motors_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: "cover.cover_relays"},
{ATTR_ENTITY_ID: COVER_RELAYS},
blocking=True,
)
await hass.async_block_till_done()
control_motors_relays.assert_awaited_with(states)
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
assert state is not None
assert state.state == STATE_CLOSING
@ -271,7 +274,7 @@ async def test_relays_stop(control_motors_relays, hass, lcn_connection):
states = [MotorStateModifier.NOCHANGE] * 4
states[0] = MotorStateModifier.STOP
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
state.state = STATE_CLOSING
# command failed
@ -280,13 +283,13 @@ async def test_relays_stop(control_motors_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: "cover.cover_relays"},
{ATTR_ENTITY_ID: COVER_RELAYS},
blocking=True,
)
await hass.async_block_till_done()
control_motors_relays.assert_awaited_with(states)
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
assert state is not None
assert state.state == STATE_CLOSING
@ -297,13 +300,13 @@ async def test_relays_stop(control_motors_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: "cover.cover_relays"},
{ATTR_ENTITY_ID: COVER_RELAYS},
blocking=True,
)
await hass.async_block_till_done()
control_motors_relays.assert_awaited_with(states)
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
assert state is not None
assert state.state not in (STATE_CLOSING, STATE_OPENING)
@ -313,33 +316,33 @@ async def test_pushed_outputs_status_change(hass, entry, lcn_connection):
device_connection = get_device_connection(hass, (0, 7, False), entry)
address = LcnAddr(0, 7, False)
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
state.state = STATE_CLOSED
# push status "open"
input = ModStatusOutput(address, 0, 100)
await device_connection.async_process_input(input)
inp = ModStatusOutput(address, 0, 100)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
assert state is not None
assert state.state == STATE_OPENING
# push status "stop"
input = ModStatusOutput(address, 0, 0)
await device_connection.async_process_input(input)
inp = ModStatusOutput(address, 0, 0)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
assert state is not None
assert state.state not in (STATE_OPENING, STATE_CLOSING)
# push status "close"
input = ModStatusOutput(address, 1, 100)
await device_connection.async_process_input(input)
inp = ModStatusOutput(address, 1, 100)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("cover.cover_outputs")
state = hass.states.get(COVER_OUTPUTS)
assert state is not None
assert state.state == STATE_CLOSING
@ -350,36 +353,36 @@ async def test_pushed_relays_status_change(hass, entry, lcn_connection):
address = LcnAddr(0, 7, False)
states = [False] * 8
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
state.state = STATE_CLOSED
# push status "open"
states[0:2] = [True, False]
input = ModStatusRelays(address, states)
await device_connection.async_process_input(input)
inp = ModStatusRelays(address, states)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
assert state is not None
assert state.state == STATE_OPENING
# push status "stop"
states[0] = False
input = ModStatusRelays(address, states)
await device_connection.async_process_input(input)
inp = ModStatusRelays(address, states)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
assert state is not None
assert state.state not in (STATE_OPENING, STATE_CLOSING)
# push status "close"
states[0:2] = [True, True]
input = ModStatusRelays(address, states)
await device_connection.async_process_input(input)
inp = ModStatusRelays(address, states)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("cover.cover_relays")
state = hass.states.get(COVER_RELAYS)
assert state is not None
assert state.state == STATE_CLOSING
@ -387,5 +390,5 @@ async def test_pushed_relays_status_change(hass, entry, lcn_connection):
async def test_unload_config_entry(hass, entry, lcn_connection):
"""Test the cover is removed when the config entry is unloaded."""
await hass.config_entries.async_unload(entry.entry_id)
assert hass.states.get("cover.cover_outputs").state == STATE_UNAVAILABLE
assert hass.states.get("cover.cover_relays").state == STATE_UNAVAILABLE
assert hass.states.get(COVER_OUTPUTS).state == STATE_UNAVAILABLE
assert hass.states.get(COVER_RELAYS).state == STATE_UNAVAILABLE

37
tests/components/lcn/test_events.py
View File

@ -5,10 +5,15 @@ from pypck.lcn_defs import AccessControlPeriphery, KeyAction, SendKeyCommand
from tests.common import async_capture_events
LCN_TRANSPONDER = "lcn_transponder"
LCN_FINGERPRINT = "lcn_fingerprint"
LCN_TRANSMITTER = "lcn_transmitter"
LCN_SEND_KEYS = "lcn_send_keys"
async def test_fire_transponder_event(hass, lcn_connection):
"""Test the transponder event is fired."""
events = async_capture_events(hass, "lcn_transponder")
events = async_capture_events(hass, LCN_TRANSPONDER)
inp = ModStatusAccessControl(
LcnAddr(0, 7, False),
@ -20,13 +25,13 @@ async def test_fire_transponder_event(hass, lcn_connection):
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].event_type == "lcn_transponder"
assert events[0].event_type == LCN_TRANSPONDER
assert events[0].data["code"] == "aabbcc"
async def test_fire_fingerprint_event(hass, lcn_connection):
"""Test the fingerprint event is fired."""
events = async_capture_events(hass, "lcn_fingerprint")
events = async_capture_events(hass, LCN_FINGERPRINT)
inp = ModStatusAccessControl(
LcnAddr(0, 7, False),
@ -38,7 +43,7 @@ async def test_fire_fingerprint_event(hass, lcn_connection):
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].event_type == "lcn_fingerprint"
assert events[0].event_type == LCN_FINGERPRINT
assert events[0].data["code"] == "aabbcc"
@ -62,7 +67,7 @@ async def test_fire_codelock_event(hass, lcn_connection):
async def test_fire_transmitter_event(hass, lcn_connection):
"""Test the transmitter event is fired."""
events = async_capture_events(hass, "lcn_transmitter")
events = async_capture_events(hass, LCN_TRANSMITTER)
inp = ModStatusAccessControl(
LcnAddr(0, 7, False),
@ -77,7 +82,7 @@ async def test_fire_transmitter_event(hass, lcn_connection):
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].event_type == "lcn_transmitter"
assert events[0].event_type == LCN_TRANSMITTER
assert events[0].data["code"] == "aabbcc"
assert events[0].data["level"] == 0
assert events[0].data["key"] == 0
@ -86,7 +91,7 @@ async def test_fire_transmitter_event(hass, lcn_connection):
async def test_fire_sendkeys_event(hass, lcn_connection):
"""Test the send_keys event is fired."""
events = async_capture_events(hass, "lcn_send_keys")
events = async_capture_events(hass, LCN_SEND_KEYS)
inp = ModSendKeysHost(
LcnAddr(0, 7, False),
@ -98,16 +103,16 @@ async def test_fire_sendkeys_event(hass, lcn_connection):
await hass.async_block_till_done()
assert len(events) == 4
assert events[0].event_type == "lcn_send_keys"
assert events[0].event_type == LCN_SEND_KEYS
assert events[0].data["key"] == "a1"
assert events[0].data["action"] == "hit"
assert events[1].event_type == "lcn_send_keys"
assert events[1].event_type == LCN_SEND_KEYS
assert events[1].data["key"] == "a2"
assert events[1].data["action"] == "hit"
assert events[2].event_type == "lcn_send_keys"
assert events[2].event_type == LCN_SEND_KEYS
assert events[2].data["key"] == "b1"
assert events[2].data["action"] == "make"
assert events[3].event_type == "lcn_send_keys"
assert events[3].event_type == LCN_SEND_KEYS
assert events[3].data["key"] == "b2"
assert events[3].data["action"] == "make"
@ -117,10 +122,10 @@ async def test_dont_fire_on_non_module_input(hass, lcn_connection):
inp = Input()
for event_name in (
"lcn_transponder",
"lcn_fingerprint",
"lcn_transmitter",
"lcn_send_keys",
LCN_TRANSPONDER,
LCN_FINGERPRINT,
LCN_TRANSMITTER,
LCN_SEND_KEYS,
):
events = async_capture_events(hass, event_name)
await lcn_connection.async_process_input(inp)
@ -136,7 +141,7 @@ async def test_dont_fire_on_unknown_module(hass, lcn_connection):
code="aabbcc",
)
events = async_capture_events(hass, "lcn_fingerprint")
events = async_capture_events(hass, LCN_FINGERPRINT)
await lcn_connection.async_process_input(inp)
await hass.async_block_till_done()
assert len(events) == 0

74
tests/components/lcn/test_light.py
View File

@ -27,13 +27,17 @@ from homeassistant.helpers import entity_registry as er
from .conftest import MockModuleConnection
LIGHT_OUTPUT1 = "light.light_output1"
LIGHT_OUTPUT2 = "light.light_output2"
LIGHT_RELAY1 = "light.light_relay1"
async def test_setup_lcn_light(hass, lcn_connection):
"""Test the setup of light."""
for entity_id in (
"light.light_output1",
"light.light_output2",
"light.light_relay1",
LIGHT_OUTPUT1,
LIGHT_OUTPUT2,
LIGHT_RELAY1,
):
state = hass.states.get(entity_id)
assert state is not None
@ -42,12 +46,12 @@ async def test_setup_lcn_light(hass, lcn_connection):
async def test_entity_state(hass, lcn_connection):
"""Test state of entity."""
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
assert state
assert state.attributes[ATTR_SUPPORTED_FEATURES] == LightEntityFeature.TRANSITION
assert state.attributes[ATTR_SUPPORTED_COLOR_MODES] == [ColorMode.BRIGHTNESS]
state = hass.states.get("light.light_output2")
state = hass.states.get(LIGHT_OUTPUT2)
assert state
assert state.attributes[ATTR_SUPPORTED_FEATURES] == LightEntityFeature.TRANSITION
assert state.attributes[ATTR_SUPPORTED_COLOR_MODES] == [ColorMode.ONOFF]
@ -57,13 +61,13 @@ async def test_entity_attributes(hass, entry, lcn_connection):
"""Test the attributes of an entity."""
entity_registry = er.async_get(hass)
entity_output = entity_registry.async_get("light.light_output1")
entity_output = entity_registry.async_get(LIGHT_OUTPUT1)
assert entity_output
assert entity_output.unique_id == f"{entry.entry_id}-m000007-output1"
assert entity_output.original_name == "Light_Output1"
entity_relay = entity_registry.async_get("light.light_relay1")
entity_relay = entity_registry.async_get(LIGHT_RELAY1)
assert entity_relay
assert entity_relay.unique_id == f"{entry.entry_id}-m000007-relay1"
@ -79,13 +83,13 @@ async def test_output_turn_on(dim_output, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "light.light_output1"},
{ATTR_ENTITY_ID: LIGHT_OUTPUT1},
blocking=True,
)
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 100, 9)
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
assert state is not None
assert state.state != STATE_ON
@ -96,13 +100,13 @@ async def test_output_turn_on(dim_output, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "light.light_output1"},
{ATTR_ENTITY_ID: LIGHT_OUTPUT1},
blocking=True,
)
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 100, 9)
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
assert state is not None
assert state.state == STATE_ON
@ -116,7 +120,7 @@ async def test_output_turn_on_with_attributes(dim_output, hass, lcn_connection):
DOMAIN_LIGHT,
SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: "light.light_output1",
ATTR_ENTITY_ID: LIGHT_OUTPUT1,
ATTR_BRIGHTNESS: 50,
ATTR_TRANSITION: 2,
},
@ -125,7 +129,7 @@ async def test_output_turn_on_with_attributes(dim_output, hass, lcn_connection):
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 19, 6)
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
assert state is not None
assert state.state == STATE_ON
@ -133,7 +137,7 @@ async def test_output_turn_on_with_attributes(dim_output, hass, lcn_connection):
@patch.object(MockModuleConnection, "dim_output")
async def test_output_turn_off(dim_output, hass, lcn_connection):
"""Test the output light turns off."""
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
state.state = STATE_ON
# command failed
@ -142,13 +146,13 @@ async def test_output_turn_off(dim_output, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "light.light_output1"},
{ATTR_ENTITY_ID: LIGHT_OUTPUT1},
blocking=True,
)
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 0, 9)
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
assert state is not None
assert state.state != STATE_OFF
@ -159,13 +163,13 @@ async def test_output_turn_off(dim_output, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "light.light_output1"},
{ATTR_ENTITY_ID: LIGHT_OUTPUT1},
blocking=True,
)
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 0, 9)
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
assert state is not None
assert state.state == STATE_OFF
@ -175,14 +179,14 @@ async def test_output_turn_off_with_attributes(dim_output, hass, lcn_connection)
"""Test the output light turns off."""
dim_output.return_value = True
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
state.state = STATE_ON
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_OFF,
{
ATTR_ENTITY_ID: "light.light_output1",
ATTR_ENTITY_ID: LIGHT_OUTPUT1,
ATTR_TRANSITION: 2,
},
blocking=True,
@ -190,7 +194,7 @@ async def test_output_turn_off_with_attributes(dim_output, hass, lcn_connection)
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 0, 6)
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
assert state is not None
assert state.state == STATE_OFF
@ -207,13 +211,13 @@ async def test_relay_turn_on(control_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "light.light_relay1"},
{ATTR_ENTITY_ID: LIGHT_RELAY1},
blocking=True,
)
await hass.async_block_till_done()
control_relays.assert_awaited_with(states)
state = hass.states.get("light.light_relay1")
state = hass.states.get(LIGHT_RELAY1)
assert state is not None
assert state.state != STATE_ON
@ -224,13 +228,13 @@ async def test_relay_turn_on(control_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "light.light_relay1"},
{ATTR_ENTITY_ID: LIGHT_RELAY1},
blocking=True,
)
await hass.async_block_till_done()
control_relays.assert_awaited_with(states)
state = hass.states.get("light.light_relay1")
state = hass.states.get(LIGHT_RELAY1)
assert state is not None
assert state.state == STATE_ON
@ -241,7 +245,7 @@ async def test_relay_turn_off(control_relays, hass, lcn_connection):
states = [RelayStateModifier.NOCHANGE] * 8
states[0] = RelayStateModifier.OFF
state = hass.states.get("light.light_relay1")
state = hass.states.get(LIGHT_RELAY1)
state.state = STATE_ON
# command failed
@ -250,13 +254,13 @@ async def test_relay_turn_off(control_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "light.light_relay1"},
{ATTR_ENTITY_ID: LIGHT_RELAY1},
blocking=True,
)
await hass.async_block_till_done()
control_relays.assert_awaited_with(states)
state = hass.states.get("light.light_relay1")
state = hass.states.get(LIGHT_RELAY1)
assert state is not None
assert state.state != STATE_OFF
@ -267,13 +271,13 @@ async def test_relay_turn_off(control_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "light.light_relay1"},
{ATTR_ENTITY_ID: LIGHT_RELAY1},
blocking=True,
)
await hass.async_block_till_done()
control_relays.assert_awaited_with(states)
state = hass.states.get("light.light_relay1")
state = hass.states.get(LIGHT_RELAY1)
assert state is not None
assert state.state == STATE_OFF
@ -288,7 +292,7 @@ async def test_pushed_output_status_change(hass, entry, lcn_connection):
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
assert state is not None
assert state.state == STATE_ON
assert state.attributes[ATTR_BRIGHTNESS] == 127
@ -298,7 +302,7 @@ async def test_pushed_output_status_change(hass, entry, lcn_connection):
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("light.light_output1")
state = hass.states.get(LIGHT_OUTPUT1)
assert state is not None
assert state.state == STATE_OFF
@ -315,7 +319,7 @@ async def test_pushed_relay_status_change(hass, entry, lcn_connection):
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("light.light_relay1")
state = hass.states.get(LIGHT_RELAY1)
assert state is not None
assert state.state == STATE_ON
@ -325,7 +329,7 @@ async def test_pushed_relay_status_change(hass, entry, lcn_connection):
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("light.light_relay1")
state = hass.states.get(LIGHT_RELAY1)
assert state is not None
assert state.state == STATE_OFF
@ -333,4 +337,4 @@ async def test_pushed_relay_status_change(hass, entry, lcn_connection):
async def test_unload_config_entry(hass, entry, lcn_connection):
"""Test the light is removed when the config entry is unloaded."""
await hass.config_entries.async_unload(entry.entry_id)
assert hass.states.get("light.light_output1").state == STATE_UNAVAILABLE
assert hass.states.get(LIGHT_OUTPUT1).state == STATE_UNAVAILABLE

131
tests/components/lcn/test_sensor.py Normal file
View File

@ -0,0 +1,131 @@
"""Test for the LCN sensor platform."""
from pypck.inputs import ModStatusLedsAndLogicOps, ModStatusVar
from pypck.lcn_addr import LcnAddr
from pypck.lcn_defs import LedStatus, LogicOpStatus, Var, VarValue
from homeassistant.components.lcn.helpers import get_device_connection
from homeassistant.const import (
ATTR_UNIT_OF_MEASUREMENT,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
TEMP_CELSIUS,
)
from homeassistant.helpers import entity_registry as er
SENSOR_VAR1 = "sensor.sensor_var1"
SENSOR_SETPOINT1 = "sensor.sensor_setpoint1"
SENSOR_LED6 = "sensor.sensor_led6"
SENSOR_LOGICOP1 = "sensor.sensor_logicop1"
async def test_setup_lcn_sensor(hass, entry, lcn_connection):
"""Test the setup of sensor."""
for entity_id in (
SENSOR_VAR1,
SENSOR_SETPOINT1,
SENSOR_LED6,
SENSOR_LOGICOP1,
):
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_UNKNOWN
async def test_entity_state(hass, lcn_connection):
"""Test state of entity."""
state = hass.states.get(SENSOR_VAR1)
assert state
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == TEMP_CELSIUS
state = hass.states.get(SENSOR_SETPOINT1)
assert state
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == TEMP_CELSIUS
state = hass.states.get(SENSOR_LED6)
assert state
state = hass.states.get(SENSOR_LOGICOP1)
assert state
async def test_entity_attributes(hass, entry, lcn_connection):
"""Test the attributes of an entity."""
entity_registry = er.async_get(hass)
entity_var1 = entity_registry.async_get(SENSOR_VAR1)
assert entity_var1
assert entity_var1.unique_id == f"{entry.entry_id}-m000007-var1"
assert entity_var1.original_name == "Sensor_Var1"
entity_r1varsetpoint = entity_registry.async_get(SENSOR_SETPOINT1)
assert entity_r1varsetpoint
assert entity_r1varsetpoint.unique_id == f"{entry.entry_id}-m000007-r1varsetpoint"
assert entity_r1varsetpoint.original_name == "Sensor_Setpoint1"
entity_led6 = entity_registry.async_get(SENSOR_LED6)
assert entity_led6
assert entity_led6.unique_id == f"{entry.entry_id}-m000007-led6"
assert entity_led6.original_name == "Sensor_Led6"
entity_logicop1 = entity_registry.async_get(SENSOR_LOGICOP1)
assert entity_logicop1
assert entity_logicop1.unique_id == f"{entry.entry_id}-m000007-logicop1"
assert entity_logicop1.original_name == "Sensor_LogicOp1"
async def test_pushed_variable_status_change(hass, entry, lcn_connection):
"""Test the variable sensor changes its state on status received."""
device_connection = get_device_connection(hass, (0, 7, False), entry)
address = LcnAddr(0, 7, False)
# push status variable
inp = ModStatusVar(address, Var.VAR1, VarValue.from_celsius(42))
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get(SENSOR_VAR1)
assert state is not None
assert float(state.state) == 42.0
# push status setpoint
inp = ModStatusVar(address, Var.R1VARSETPOINT, VarValue.from_celsius(42))
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get(SENSOR_SETPOINT1)
assert state is not None
assert float(state.state) == 42.0
async def test_pushed_ledlogicop_status_change(hass, entry, lcn_connection):
"""Test the led and logicop sensor changes its state on status received."""
device_connection = get_device_connection(hass, (0, 7, False), entry)
address = LcnAddr(0, 7, False)
states_led = [LedStatus.OFF] * 12
states_logicop = [LogicOpStatus.NONE] * 4
states_led[5] = LedStatus.ON
states_logicop[0] = LogicOpStatus.ALL
# push status led and logicop
inp = ModStatusLedsAndLogicOps(address, states_led, states_logicop)
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get(SENSOR_LED6)
assert state is not None
assert state.state == "on"
state = hass.states.get(SENSOR_LOGICOP1)
assert state is not None
assert state.state == "all"
async def test_unload_config_entry(hass, entry, lcn_connection):
"""Test the sensor is removed when the config entry is unloaded."""
await hass.config_entries.async_unload(entry.entry_id)
assert hass.states.get(SENSOR_VAR1).state == STATE_UNAVAILABLE
assert hass.states.get(SENSOR_SETPOINT1).state == STATE_UNAVAILABLE
assert hass.states.get(SENSOR_LED6).state == STATE_UNAVAILABLE
assert hass.states.get(SENSOR_LOGICOP1).state == STATE_UNAVAILABLE

63
tests/components/lcn/test_switch.py
View File

@ -19,14 +19,19 @@ from homeassistant.helpers import entity_registry as er
from .conftest import MockModuleConnection
SWITCH_OUTPUT1 = "switch.switch_output1"
SWITCH_OUTPUT2 = "switch.switch_output2"
SWITCH_RELAY1 = "switch.switch_relay1"
SWITCH_RELAY2 = "switch.switch_relay2"
async def test_setup_lcn_switch(hass, lcn_connection):
"""Test the setup of switch."""
for entity_id in (
"switch.switch_output1",
"switch.switch_output2",
"switch.switch_relay1",
"switch.switch_relay2",
SWITCH_OUTPUT1,
SWITCH_OUTPUT2,
SWITCH_RELAY1,
SWITCH_RELAY2,
):
state = hass.states.get(entity_id)
assert state is not None
@ -37,13 +42,13 @@ async def test_entity_attributes(hass, entry, lcn_connection):
"""Test the attributes of an entity."""
entity_registry = er.async_get(hass)
entity_output = entity_registry.async_get("switch.switch_output1")
entity_output = entity_registry.async_get(SWITCH_OUTPUT1)
assert entity_output
assert entity_output.unique_id == f"{entry.entry_id}-m000007-output1"
assert entity_output.original_name == "Switch_Output1"
entity_relay = entity_registry.async_get("switch.switch_relay1")
entity_relay = entity_registry.async_get(SWITCH_RELAY1)
assert entity_relay
assert entity_relay.unique_id == f"{entry.entry_id}-m000007-relay1"
@ -59,13 +64,13 @@ async def test_output_turn_on(dim_output, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.switch_output1"},
{ATTR_ENTITY_ID: SWITCH_OUTPUT1},
blocking=True,
)
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 100, 0)
state = hass.states.get("switch.switch_output1")
state = hass.states.get(SWITCH_OUTPUT1)
assert state.state == STATE_OFF
# command success
@ -75,20 +80,20 @@ async def test_output_turn_on(dim_output, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.switch_output1"},
{ATTR_ENTITY_ID: SWITCH_OUTPUT1},
blocking=True,
)
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 100, 0)
state = hass.states.get("switch.switch_output1")
state = hass.states.get(SWITCH_OUTPUT1)
assert state.state == STATE_ON
@patch.object(MockModuleConnection, "dim_output")
async def test_output_turn_off(dim_output, hass, lcn_connection):
"""Test the output switch turns off."""
state = hass.states.get("switch.switch_output1")
state = hass.states.get(SWITCH_OUTPUT1)
state.state = STATE_ON
# command failed
@ -97,13 +102,13 @@ async def test_output_turn_off(dim_output, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.switch_output1"},
{ATTR_ENTITY_ID: SWITCH_OUTPUT1},
blocking=True,
)
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 0, 0)
state = hass.states.get("switch.switch_output1")
state = hass.states.get(SWITCH_OUTPUT1)
assert state.state == STATE_ON
# command success
@ -113,13 +118,13 @@ async def test_output_turn_off(dim_output, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.switch_output1"},
{ATTR_ENTITY_ID: SWITCH_OUTPUT1},
blocking=True,
)
await hass.async_block_till_done()
dim_output.assert_awaited_with(0, 0, 0)
state = hass.states.get("switch.switch_output1")
state = hass.states.get(SWITCH_OUTPUT1)
assert state.state == STATE_OFF
@ -135,13 +140,13 @@ async def test_relay_turn_on(control_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.switch_relay1"},
{ATTR_ENTITY_ID: SWITCH_RELAY1},
blocking=True,
)
await hass.async_block_till_done()
control_relays.assert_awaited_with(states)
state = hass.states.get("switch.switch_relay1")
state = hass.states.get(SWITCH_RELAY1)
assert state.state == STATE_OFF
# command success
@ -151,13 +156,13 @@ async def test_relay_turn_on(control_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.switch_relay1"},
{ATTR_ENTITY_ID: SWITCH_RELAY1},
blocking=True,
)
await hass.async_block_till_done()
control_relays.assert_awaited_with(states)
state = hass.states.get("switch.switch_relay1")
state = hass.states.get(SWITCH_RELAY1)
assert state.state == STATE_ON
@ -167,7 +172,7 @@ async def test_relay_turn_off(control_relays, hass, lcn_connection):
states = [RelayStateModifier.NOCHANGE] * 8
states[0] = RelayStateModifier.OFF
state = hass.states.get("switch.switch_relay1")
state = hass.states.get(SWITCH_RELAY1)
state.state = STATE_ON
# command failed
@ -176,13 +181,13 @@ async def test_relay_turn_off(control_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.switch_relay1"},
{ATTR_ENTITY_ID: SWITCH_RELAY1},
blocking=True,
)
await hass.async_block_till_done()
control_relays.assert_awaited_with(states)
state = hass.states.get("switch.switch_relay1")
state = hass.states.get(SWITCH_RELAY1)
assert state.state == STATE_ON
# command success
@ -192,13 +197,13 @@ async def test_relay_turn_off(control_relays, hass, lcn_connection):
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.switch_relay1"},
{ATTR_ENTITY_ID: SWITCH_RELAY1},
blocking=True,
)
await hass.async_block_till_done()
control_relays.assert_awaited_with(states)
state = hass.states.get("switch.switch_relay1")
state = hass.states.get(SWITCH_RELAY1)
assert state.state == STATE_OFF
@ -212,7 +217,7 @@ async def test_pushed_output_status_change(hass, entry, lcn_connection):
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("switch.switch_output1")
state = hass.states.get(SWITCH_OUTPUT1)
assert state.state == STATE_ON
# push status "off"
@ -220,7 +225,7 @@ async def test_pushed_output_status_change(hass, entry, lcn_connection):
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("switch.switch_output1")
state = hass.states.get(SWITCH_OUTPUT1)
assert state.state == STATE_OFF
@ -236,7 +241,7 @@ async def test_pushed_relay_status_change(hass, entry, lcn_connection):
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("switch.switch_relay1")
state = hass.states.get(SWITCH_RELAY1)
assert state.state == STATE_ON
# push status "off"
@ -245,11 +250,11 @@ async def test_pushed_relay_status_change(hass, entry, lcn_connection):
await device_connection.async_process_input(inp)
await hass.async_block_till_done()
state = hass.states.get("switch.switch_relay1")
state = hass.states.get(SWITCH_RELAY1)
assert state.state == STATE_OFF
async def test_unload_config_entry(hass, entry, lcn_connection):
"""Test the switch is removed when the config entry is unloaded."""
await hass.config_entries.async_unload(entry.entry_id)
assert hass.states.get("switch.switch_output1").state == STATE_UNAVAILABLE
assert hass.states.get(SWITCH_OUTPUT1).state == STATE_UNAVAILABLE