diff --git a/.coveragerc b/.coveragerc index e378f0291cc..b87a91716cf 100644 --- a/.coveragerc +++ b/.coveragerc @@ -616,7 +616,6 @@ omit = homeassistant/components/launch_library/sensor.py homeassistant/components/lcn/binary_sensor.py homeassistant/components/lcn/climate.py - homeassistant/components/lcn/cover.py homeassistant/components/lcn/helpers.py homeassistant/components/lcn/scene.py homeassistant/components/lcn/sensor.py diff --git a/tests/components/lcn/fixtures/config.json b/tests/components/lcn/fixtures/config.json index e9278d8e2cd..cc615b6083b 100644 --- a/tests/components/lcn/fixtures/config.json +++ b/tests/components/lcn/fixtures/config.json @@ -77,6 +77,19 @@ "address": "s0.g5", "output": "relay1" } + ], + "covers": [ + { + "name": "Cover_Ouputs", + "address": "s0.m7", + "motor": "outputs", + "reverse_time": "rt1200" + }, + { + "name": "Cover_Relays", + "address": "s0.m7", + "motor": "motor1" + } ] } } diff --git a/tests/components/lcn/fixtures/config_entry_pchk.json b/tests/components/lcn/fixtures/config_entry_pchk.json index 9d34add37ff..620bbb673f5 100644 --- a/tests/components/lcn/fixtures/config_entry_pchk.json +++ b/tests/components/lcn/fixtures/config_entry_pchk.json @@ -100,6 +100,26 @@ "domain_data": { "output": "RELAY1" } + }, + { + "address": [0, 7, false], + "name": "Cover_Outputs", + "resource": "outputs", + "domain": "cover", + "domain_data": { + "motor": "OUTPUTS", + "reverse_time": "RT1200" + } + }, + { + "address": [0, 7, false], + "name": "Cover_Relays", + "resource": "motor1", + "domain": "cover", + "domain_data": { + "motor": "MOTOR1", + "reverse_time": "RT1200" + } } ] } diff --git a/tests/components/lcn/test_cover.py b/tests/components/lcn/test_cover.py new file mode 100644 index 00000000000..f89cfa41071 --- /dev/null +++ b/tests/components/lcn/test_cover.py @@ -0,0 +1,390 @@ +"""Test for the LCN cover platform.""" +from unittest.mock import patch + +from pypck.inputs import ModStatusOutput, ModStatusRelays +from pypck.lcn_addr import LcnAddr +from pypck.lcn_defs import MotorReverseTime, MotorStateModifier + +from homeassistant.components.cover import DOMAIN as DOMAIN_COVER +from homeassistant.components.lcn.helpers import get_device_connection +from homeassistant.const import ( + ATTR_ENTITY_ID, + SERVICE_CLOSE_COVER, + SERVICE_OPEN_COVER, + SERVICE_STOP_COVER, + STATE_CLOSED, + STATE_CLOSING, + STATE_OPEN, + STATE_OPENING, + STATE_UNAVAILABLE, +) + +from .conftest import MockModuleConnection + + +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", + ): + state = hass.states.get(entity_id) + assert state is not None + assert state.state == STATE_OPEN + + +async def test_entity_attributes(hass, entry, lcn_connection): + """Test the attributes of an entity.""" + entity_registry = await hass.helpers.entity_registry.async_get_registry() + + entity_outputs = entity_registry.async_get("cover.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") + + assert entity_relays + assert entity_relays.unique_id == f"{entry.entry_id}-m000007-motor1" + assert entity_relays.original_name == "Cover_Relays" + + +@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.state = STATE_CLOSED + + # command failed + control_motors_outputs.return_value = False + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_OPEN_COVER, + {ATTR_ENTITY_ID: "cover.cover_outputs"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_outputs.assert_awaited_with( + MotorStateModifier.UP, MotorReverseTime.RT1200 + ) + + state = hass.states.get("cover.cover_outputs") + assert state is not None + assert state.state != STATE_OPENING + + # command success + control_motors_outputs.reset_mock(return_value=True) + control_motors_outputs.return_value = True + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_OPEN_COVER, + {ATTR_ENTITY_ID: "cover.cover_outputs"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_outputs.assert_awaited_with( + MotorStateModifier.UP, MotorReverseTime.RT1200 + ) + + state = hass.states.get("cover.cover_outputs") + assert state is not None + assert state.state == STATE_OPENING + + +@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.state = STATE_OPEN + + # command failed + control_motors_outputs.return_value = False + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_CLOSE_COVER, + {ATTR_ENTITY_ID: "cover.cover_outputs"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_outputs.assert_awaited_with( + MotorStateModifier.DOWN, MotorReverseTime.RT1200 + ) + + state = hass.states.get("cover.cover_outputs") + assert state is not None + assert state.state != STATE_CLOSING + + # command success + control_motors_outputs.reset_mock(return_value=True) + control_motors_outputs.return_value = True + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_CLOSE_COVER, + {ATTR_ENTITY_ID: "cover.cover_outputs"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_outputs.assert_awaited_with( + MotorStateModifier.DOWN, MotorReverseTime.RT1200 + ) + + state = hass.states.get("cover.cover_outputs") + assert state is not None + assert state.state == STATE_CLOSING + + +@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.state = STATE_CLOSING + + # command failed + control_motors_outputs.return_value = False + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_STOP_COVER, + {ATTR_ENTITY_ID: "cover.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") + assert state is not None + assert state.state == STATE_CLOSING + + # command success + control_motors_outputs.reset_mock(return_value=True) + control_motors_outputs.return_value = True + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_STOP_COVER, + {ATTR_ENTITY_ID: "cover.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") + assert state is not None + assert state.state not in (STATE_CLOSING, STATE_OPENING) + + +@patch.object(MockModuleConnection, "control_motors_relays") +async def test_relays_open(control_motors_relays, hass, lcn_connection): + """Test the relays cover opens.""" + states = [MotorStateModifier.NOCHANGE] * 4 + states[0] = MotorStateModifier.UP + + state = hass.states.get("cover.cover_relays") + state.state = STATE_CLOSED + + # command failed + control_motors_relays.return_value = False + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_OPEN_COVER, + {ATTR_ENTITY_ID: "cover.cover_relays"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_relays.assert_awaited_with(states) + + state = hass.states.get("cover.cover_relays") + assert state is not None + assert state.state != STATE_OPENING + + # command success + control_motors_relays.reset_mock(return_value=True) + control_motors_relays.return_value = True + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_OPEN_COVER, + {ATTR_ENTITY_ID: "cover.cover_relays"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_relays.assert_awaited_with(states) + + state = hass.states.get("cover.cover_relays") + assert state is not None + assert state.state == STATE_OPENING + + +@patch.object(MockModuleConnection, "control_motors_relays") +async def test_relays_close(control_motors_relays, hass, lcn_connection): + """Test the relays cover closes.""" + states = [MotorStateModifier.NOCHANGE] * 4 + states[0] = MotorStateModifier.DOWN + + state = hass.states.get("cover.cover_relays") + state.state = STATE_OPEN + + # command failed + control_motors_relays.return_value = False + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_CLOSE_COVER, + {ATTR_ENTITY_ID: "cover.cover_relays"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_relays.assert_awaited_with(states) + + state = hass.states.get("cover.cover_relays") + assert state is not None + assert state.state != STATE_CLOSING + + # command success + control_motors_relays.reset_mock(return_value=True) + control_motors_relays.return_value = True + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_CLOSE_COVER, + {ATTR_ENTITY_ID: "cover.cover_relays"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_relays.assert_awaited_with(states) + + state = hass.states.get("cover.cover_relays") + assert state is not None + assert state.state == STATE_CLOSING + + +@patch.object(MockModuleConnection, "control_motors_relays") +async def test_relays_stop(control_motors_relays, hass, lcn_connection): + """Test the relays cover stops.""" + states = [MotorStateModifier.NOCHANGE] * 4 + states[0] = MotorStateModifier.STOP + + state = hass.states.get("cover.cover_relays") + state.state = STATE_CLOSING + + # command failed + control_motors_relays.return_value = False + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_STOP_COVER, + {ATTR_ENTITY_ID: "cover.cover_relays"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_relays.assert_awaited_with(states) + + state = hass.states.get("cover.cover_relays") + assert state is not None + assert state.state == STATE_CLOSING + + # command success + control_motors_relays.reset_mock(return_value=True) + control_motors_relays.return_value = True + + await hass.services.async_call( + DOMAIN_COVER, + SERVICE_STOP_COVER, + {ATTR_ENTITY_ID: "cover.cover_relays"}, + blocking=True, + ) + await hass.async_block_till_done() + control_motors_relays.assert_awaited_with(states) + + state = hass.states.get("cover.cover_relays") + assert state is not None + assert state.state not in (STATE_CLOSING, STATE_OPENING) + + +async def test_pushed_outputs_status_change(hass, entry, lcn_connection): + """Test the outputs cover changes its state on status received.""" + device_connection = get_device_connection(hass, (0, 7, False), entry) + address = LcnAddr(0, 7, False) + + state = hass.states.get("cover.cover_outputs") + state.state = STATE_CLOSED + + # push status "open" + input = ModStatusOutput(address, 0, 100) + await device_connection.async_process_input(input) + await hass.async_block_till_done() + + state = hass.states.get("cover.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) + await hass.async_block_till_done() + + state = hass.states.get("cover.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) + await hass.async_block_till_done() + + state = hass.states.get("cover.cover_outputs") + assert state is not None + assert state.state == STATE_CLOSING + + +async def test_pushed_relays_status_change(hass, entry, lcn_connection): + """Test the relays cover changes its state on status received.""" + device_connection = get_device_connection(hass, (0, 7, False), entry) + address = LcnAddr(0, 7, False) + states = [False] * 8 + + state = hass.states.get("cover.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) + await hass.async_block_till_done() + + state = hass.states.get("cover.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) + await hass.async_block_till_done() + + state = hass.states.get("cover.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) + await hass.async_block_till_done() + + state = hass.states.get("cover.cover_relays") + assert state is not None + assert state.state == STATE_CLOSING + + +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