Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve group tests (#73630)

pull/73926/head
Erik Montnemery 2022-06-23 21:38:17 +02:00 committed by GitHub
parent 186141ee4d
commit 9b8c3e37bb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 685 additions and 235 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/group/binary_sensor.py
View File

@ -132,7 +132,7 @@ class BinarySensorGroup(GroupEntity, BinarySensorEntity):
# filtered_states are members currently in the state machine
filtered_states: list[str] = [x.state for x in all_states if x is not None]
# Set group as unavailable if all members are unavailable
# Set group as unavailable if all members are unavailable or missing
self._attr_available = any(
state != STATE_UNAVAILABLE for state in filtered_states
)

169
tests/components/group/test_binary_sensor.py
View File

@ -50,7 +50,13 @@ async def test_default_state(hass):
async def test_state_reporting_all(hass):
"""Test the state reporting."""
"""Test the state reporting in 'all' mode.
The group state is unavailable if all group members are unavailable.
Otherwise, the group state is unknown if at least one group member is unknown or unavailable.
Otherwise, the group state is off if at least one group member is off.
Otherwise, the group state is on.
"""
await async_setup_component(
hass,
BINARY_SENSOR_DOMAIN,
@ -68,26 +74,12 @@ async def test_state_reporting_all(hass):
await hass.async_start()
await hass.async_block_till_done()
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNKNOWN
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
hass.states.async_set("binary_sensor.test1", STATE_OFF)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_ON)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
# Initial state with no group member in the state machine -> unavailable
assert (
hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNAVAILABLE
)
# All group members unavailable -> unavailable
hass.states.async_set("binary_sensor.test1", STATE_UNAVAILABLE)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
@ -95,6 +87,12 @@ async def test_state_reporting_all(hass):
hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNAVAILABLE
)
# At least one member unknown or unavailable -> group unknown
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNKNOWN
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
@ -105,9 +103,55 @@ async def test_state_reporting_all(hass):
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNKNOWN
hass.states.async_set("binary_sensor.test1", STATE_OFF)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNKNOWN
hass.states.async_set("binary_sensor.test1", STATE_OFF)
hass.states.async_set("binary_sensor.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNKNOWN
hass.states.async_set("binary_sensor.test1", STATE_UNKNOWN)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNKNOWN
# At least one member off -> group off
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
hass.states.async_set("binary_sensor.test1", STATE_OFF)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
# Otherwise -> on
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_ON)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
# All group members removed from the state machine -> unavailable
hass.states.async_remove("binary_sensor.test1")
hass.states.async_remove("binary_sensor.test2")
await hass.async_block_till_done()
assert (
hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNAVAILABLE
)
async def test_state_reporting_any(hass):
"""Test the state reporting."""
"""Test the state reporting in 'any' mode.
The group state is unavailable if all group members are unavailable.
Otherwise, the group state is unknown if all group members are unknown.
Otherwise, the group state is on if at least one group member is on.
Otherwise, the group state is off.
"""
await async_setup_component(
hass,
BINARY_SENSOR_DOMAIN,
@ -126,26 +170,17 @@ async def test_state_reporting_any(hass):
await hass.async_start()
await hass.async_block_till_done()
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
entity_registry = er.async_get(hass)
entry = entity_registry.async_get("binary_sensor.binary_sensor_group")
assert entry
assert entry.unique_id == "unique_identifier"
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
hass.states.async_set("binary_sensor.test1", STATE_OFF)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_ON)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
# Initial state with no group member in the state machine -> unavailable
assert (
hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNAVAILABLE
)
# All group members unavailable -> unavailable
hass.states.async_set("binary_sensor.test1", STATE_UNAVAILABLE)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
@ -153,17 +188,59 @@ async def test_state_reporting_any(hass):
hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNAVAILABLE
)
entity_registry = er.async_get(hass)
entry = entity_registry.async_get("binary_sensor.binary_sensor_group")
assert entry
assert entry.unique_id == "unique_identifier"
# All group members unknown -> unknown
hass.states.async_set("binary_sensor.test1", STATE_UNKNOWN)
hass.states.async_set("binary_sensor.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNKNOWN
# Group members unknown or unavailable -> unknown
hass.states.async_set("binary_sensor.test1", STATE_UNKNOWN)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNKNOWN
# At least one member on -> group on
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_ON)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
hass.states.async_set("binary_sensor.test1", STATE_ON)
hass.states.async_set("binary_sensor.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
hass.states.async_set("binary_sensor.test1", STATE_UNKNOWN)
hass.states.async_set("binary_sensor.test2", STATE_UNKNOWN)
# Otherwise -> off
hass.states.async_set("binary_sensor.test1", STATE_OFF)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNKNOWN
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
hass.states.async_set("binary_sensor.test1", STATE_UNKNOWN)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
hass.states.async_set("binary_sensor.test1", STATE_UNAVAILABLE)
hass.states.async_set("binary_sensor.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
# All group members removed from the state machine -> unavailable
hass.states.async_remove("binary_sensor.test1")
hass.states.async_remove("binary_sensor.test2")
await hass.async_block_till_done()
assert (
hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNAVAILABLE
)

192
tests/components/group/test_cover.py
View File

@ -33,6 +33,7 @@ from homeassistant.const import (
STATE_CLOSING,
STATE_OPEN,
STATE_OPENING,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.helpers import entity_registry as er
@ -99,7 +100,14 @@ async def setup_comp(hass, config_count):
@pytest.mark.parametrize("config_count", [(CONFIG_ATTRIBUTES, 1)])
async def test_state(hass, setup_comp):
"""Test handling of state."""
"""Test handling of state.
The group state is unknown if all group members are unknown or unavailable.
Otherwise, the group state is opening if at least one group member is opening.
Otherwise, the group state is closing if at least one group member is closing.
Otherwise, the group state is open if at least one group member is open.
Otherwise, the group state is closed.
"""
state = hass.states.get(COVER_GROUP)
# No entity has a valid state -> group state unknown
assert state.state == STATE_UNKNOWN
@ -115,87 +123,125 @@ async def test_state(hass, setup_comp):
assert ATTR_CURRENT_POSITION not in state.attributes
assert ATTR_CURRENT_TILT_POSITION not in state.attributes
# Set all entities as closed -> group state closed
hass.states.async_set(DEMO_COVER, STATE_CLOSED, {})
hass.states.async_set(DEMO_COVER_POS, STATE_CLOSED, {})
hass.states.async_set(DEMO_COVER_TILT, STATE_CLOSED, {})
hass.states.async_set(DEMO_TILT, STATE_CLOSED, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_CLOSED
# The group state is unknown if all group members are unknown or unavailable.
for state_1 in (STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_2 in (STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_3 in (STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set(DEMO_COVER, state_1, {})
hass.states.async_set(DEMO_COVER_POS, state_2, {})
hass.states.async_set(DEMO_COVER_TILT, state_3, {})
hass.states.async_set(DEMO_TILT, STATE_UNAVAILABLE, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_UNKNOWN
# Set all entities as open -> group state open
hass.states.async_set(DEMO_COVER, STATE_OPEN, {})
hass.states.async_set(DEMO_COVER_POS, STATE_OPEN, {})
hass.states.async_set(DEMO_COVER_TILT, STATE_OPEN, {})
hass.states.async_set(DEMO_TILT, STATE_OPEN, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPEN
for state_1 in (STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_2 in (STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_3 in (STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set(DEMO_COVER, state_1, {})
hass.states.async_set(DEMO_COVER_POS, state_2, {})
hass.states.async_set(DEMO_COVER_TILT, state_3, {})
hass.states.async_set(DEMO_TILT, STATE_UNKNOWN, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_UNKNOWN
# Set first entity as open -> group state open
hass.states.async_set(DEMO_COVER, STATE_OPEN, {})
hass.states.async_set(DEMO_COVER_POS, STATE_CLOSED, {})
hass.states.async_set(DEMO_COVER_TILT, STATE_CLOSED, {})
hass.states.async_set(DEMO_TILT, STATE_CLOSED, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPEN
# At least one member opening -> group opening
for state_1 in (
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_OPENING,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
for state_2 in (
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_OPENING,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
for state_3 in (
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_OPENING,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
hass.states.async_set(DEMO_COVER, state_1, {})
hass.states.async_set(DEMO_COVER_POS, state_2, {})
hass.states.async_set(DEMO_COVER_TILT, state_3, {})
hass.states.async_set(DEMO_TILT, STATE_OPENING, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPENING
# Set last entity as open -> group state open
hass.states.async_set(DEMO_COVER, STATE_OPEN, {})
hass.states.async_set(DEMO_COVER_POS, STATE_CLOSED, {})
hass.states.async_set(DEMO_COVER_TILT, STATE_CLOSED, {})
hass.states.async_set(DEMO_TILT, STATE_CLOSED, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPEN
# At least one member closing -> group closing
for state_1 in (
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
for state_2 in (
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
for state_3 in (
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
hass.states.async_set(DEMO_COVER, state_1, {})
hass.states.async_set(DEMO_COVER_POS, state_2, {})
hass.states.async_set(DEMO_COVER_TILT, state_3, {})
hass.states.async_set(DEMO_TILT, STATE_CLOSING, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_CLOSING
# Set conflicting valid states -> opening state has priority
hass.states.async_set(DEMO_COVER, STATE_OPEN, {})
hass.states.async_set(DEMO_COVER_POS, STATE_OPENING, {})
hass.states.async_set(DEMO_COVER_TILT, STATE_CLOSING, {})
hass.states.async_set(DEMO_TILT, STATE_CLOSED, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPENING
# At least one member open -> group open
for state_1 in (STATE_CLOSED, STATE_OPEN, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_2 in (STATE_CLOSED, STATE_OPEN, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_3 in (STATE_CLOSED, STATE_OPEN, STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set(DEMO_COVER, state_1, {})
hass.states.async_set(DEMO_COVER_POS, state_2, {})
hass.states.async_set(DEMO_COVER_TILT, state_3, {})
hass.states.async_set(DEMO_TILT, STATE_OPEN, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPEN
# Set all entities to unknown state -> group state unknown
hass.states.async_set(DEMO_COVER, STATE_UNKNOWN, {})
hass.states.async_set(DEMO_COVER_POS, STATE_UNKNOWN, {})
hass.states.async_set(DEMO_COVER_TILT, STATE_UNKNOWN, {})
hass.states.async_set(DEMO_TILT, STATE_UNKNOWN, {})
# At least one member closed -> group closed
for state_1 in (STATE_CLOSED, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_2 in (STATE_CLOSED, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_3 in (STATE_CLOSED, STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set(DEMO_COVER, state_1, {})
hass.states.async_set(DEMO_COVER_POS, state_2, {})
hass.states.async_set(DEMO_COVER_TILT, state_3, {})
hass.states.async_set(DEMO_TILT, STATE_CLOSED, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_CLOSED
# All group members removed from the state machine -> unknown
hass.states.async_remove(DEMO_COVER)
hass.states.async_remove(DEMO_COVER_POS)
hass.states.async_remove(DEMO_COVER_TILT)
hass.states.async_remove(DEMO_TILT)
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_UNKNOWN
# Set one entity to unknown state -> open state has priority
hass.states.async_set(DEMO_COVER, STATE_OPEN, {})
hass.states.async_set(DEMO_COVER_POS, STATE_UNKNOWN, {})
hass.states.async_set(DEMO_COVER_TILT, STATE_CLOSED, {})
hass.states.async_set(DEMO_TILT, STATE_OPEN, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPEN
# Set one entity to unknown state -> opening state has priority
hass.states.async_set(DEMO_COVER, STATE_OPEN, {})
hass.states.async_set(DEMO_COVER_POS, STATE_OPENING, {})
hass.states.async_set(DEMO_COVER_TILT, STATE_UNKNOWN, {})
hass.states.async_set(DEMO_TILT, STATE_CLOSED, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_OPENING
# Set one entity to unknown state -> closing state has priority
hass.states.async_set(DEMO_COVER, STATE_OPEN, {})
hass.states.async_set(DEMO_COVER_POS, STATE_UNKNOWN, {})
hass.states.async_set(DEMO_COVER_TILT, STATE_CLOSING, {})
hass.states.async_set(DEMO_TILT, STATE_CLOSED, {})
await hass.async_block_till_done()
state = hass.states.get(COVER_GROUP)
assert state.state == STATE_CLOSING
@pytest.mark.parametrize("config_count", [(CONFIG_ATTRIBUTES, 1)])
async def test_attributes(hass, setup_comp):

96
tests/components/group/test_fan.py
View File

@ -33,6 +33,8 @@ from homeassistant.const import (
CONF_UNIQUE_ID,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import CoreState
from homeassistant.helpers import entity_registry as er
@ -111,7 +113,11 @@ async def setup_comp(hass, config_count):
@pytest.mark.parametrize("config_count", [(CONFIG_ATTRIBUTES, 1)])
async def test_state(hass, setup_comp):
"""Test handling of state."""
"""Test handling of state.
The group state is on if at least one group member is on.
Otherwise, the group state is off.
"""
state = hass.states.get(FAN_GROUP)
# No entity has a valid state -> group state off
assert state.state == STATE_OFF
@ -123,41 +129,55 @@ async def test_state(hass, setup_comp):
assert ATTR_ASSUMED_STATE not in state.attributes
assert state.attributes[ATTR_SUPPORTED_FEATURES] == 0
# Set all entities as on -> group state on
hass.states.async_set(CEILING_FAN_ENTITY_ID, STATE_ON, {})
hass.states.async_set(LIVING_ROOM_FAN_ENTITY_ID, STATE_ON, {})
hass.states.async_set(PERCENTAGE_FULL_FAN_ENTITY_ID, STATE_ON, {})
hass.states.async_set(PERCENTAGE_LIMITED_FAN_ENTITY_ID, STATE_ON, {})
await hass.async_block_till_done()
state = hass.states.get(FAN_GROUP)
assert state.state == STATE_ON
# The group state is off if all group members are off, unknown or unavailable.
for state_1 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_2 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_3 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set(CEILING_FAN_ENTITY_ID, state_1, {})
hass.states.async_set(LIVING_ROOM_FAN_ENTITY_ID, state_2, {})
hass.states.async_set(PERCENTAGE_FULL_FAN_ENTITY_ID, state_3, {})
hass.states.async_set(PERCENTAGE_LIMITED_FAN_ENTITY_ID, STATE_OFF, {})
await hass.async_block_till_done()
state = hass.states.get(FAN_GROUP)
assert state.state == STATE_OFF
# Set all entities as off -> group state off
hass.states.async_set(CEILING_FAN_ENTITY_ID, STATE_OFF, {})
hass.states.async_set(LIVING_ROOM_FAN_ENTITY_ID, STATE_OFF, {})
hass.states.async_set(PERCENTAGE_FULL_FAN_ENTITY_ID, STATE_OFF, {})
hass.states.async_set(PERCENTAGE_LIMITED_FAN_ENTITY_ID, STATE_OFF, {})
await hass.async_block_till_done()
state = hass.states.get(FAN_GROUP)
assert state.state == STATE_OFF
for state_1 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_2 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_3 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set(CEILING_FAN_ENTITY_ID, state_1, {})
hass.states.async_set(LIVING_ROOM_FAN_ENTITY_ID, state_2, {})
hass.states.async_set(PERCENTAGE_FULL_FAN_ENTITY_ID, state_3, {})
hass.states.async_set(
PERCENTAGE_LIMITED_FAN_ENTITY_ID, STATE_UNAVAILABLE, {}
)
await hass.async_block_till_done()
state = hass.states.get(FAN_GROUP)
assert state.state == STATE_OFF
# Set first entity as on -> group state on
hass.states.async_set(CEILING_FAN_ENTITY_ID, STATE_ON, {})
hass.states.async_set(LIVING_ROOM_FAN_ENTITY_ID, STATE_OFF, {})
hass.states.async_set(PERCENTAGE_FULL_FAN_ENTITY_ID, STATE_OFF, {})
hass.states.async_set(PERCENTAGE_LIMITED_FAN_ENTITY_ID, STATE_OFF, {})
await hass.async_block_till_done()
state = hass.states.get(FAN_GROUP)
assert state.state == STATE_ON
for state_1 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_2 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_3 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set(CEILING_FAN_ENTITY_ID, state_1, {})
hass.states.async_set(LIVING_ROOM_FAN_ENTITY_ID, state_2, {})
hass.states.async_set(PERCENTAGE_FULL_FAN_ENTITY_ID, state_3, {})
hass.states.async_set(
PERCENTAGE_LIMITED_FAN_ENTITY_ID, STATE_UNKNOWN, {}
)
await hass.async_block_till_done()
state = hass.states.get(FAN_GROUP)
assert state.state == STATE_OFF
# Set last entity as on -> group state on
hass.states.async_set(CEILING_FAN_ENTITY_ID, STATE_OFF, {})
hass.states.async_set(LIVING_ROOM_FAN_ENTITY_ID, STATE_OFF, {})
hass.states.async_set(PERCENTAGE_FULL_FAN_ENTITY_ID, STATE_OFF, {})
hass.states.async_set(PERCENTAGE_LIMITED_FAN_ENTITY_ID, STATE_ON, {})
await hass.async_block_till_done()
state = hass.states.get(FAN_GROUP)
assert state.state == STATE_ON
# At least one member on -> group on
for state_1 in (STATE_OFF, STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_2 in (STATE_OFF, STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN):
for state_3 in (STATE_OFF, STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set(CEILING_FAN_ENTITY_ID, state_1, {})
hass.states.async_set(LIVING_ROOM_FAN_ENTITY_ID, state_2, {})
hass.states.async_set(PERCENTAGE_FULL_FAN_ENTITY_ID, state_3, {})
hass.states.async_set(PERCENTAGE_LIMITED_FAN_ENTITY_ID, STATE_ON, {})
await hass.async_block_till_done()
state = hass.states.get(FAN_GROUP)
assert state.state == STATE_ON
# now remove an entity
hass.states.async_remove(PERCENTAGE_LIMITED_FAN_ENTITY_ID)
@ -167,6 +187,16 @@ async def test_state(hass, setup_comp):
assert ATTR_ASSUMED_STATE not in state.attributes
assert state.attributes[ATTR_SUPPORTED_FEATURES] == 0
# now remove all entities
hass.states.async_remove(CEILING_FAN_ENTITY_ID)
hass.states.async_remove(LIVING_ROOM_FAN_ENTITY_ID)
hass.states.async_remove(PERCENTAGE_FULL_FAN_ENTITY_ID)
await hass.async_block_till_done()
state = hass.states.get(FAN_GROUP)
assert state.state == STATE_OFF
assert ATTR_ASSUMED_STATE not in state.attributes
assert state.attributes[ATTR_SUPPORTED_FEATURES] == 0
# Test entity registry integration
entity_registry = er.async_get(hass)
entry = entity_registry.async_get(FAN_GROUP)

132
tests/components/group/test_light.py
View File

@ -88,8 +88,14 @@ async def test_default_state(hass):
assert entry.unique_id == "unique_identifier"
async def test_state_reporting(hass):
"""Test the state reporting."""
async def test_state_reporting_any(hass):
"""Test the state reporting in 'any' mode.
The group state is unavailable if all group members are unavailable.
Otherwise, the group state is unknown if all group members are unknown.
Otherwise, the group state is on if at least one group member is on.
Otherwise, the group state is off.
"""
await async_setup_component(
hass,
LIGHT_DOMAIN,
@ -105,29 +111,79 @@ async def test_state_reporting(hass):
await hass.async_start()
await hass.async_block_till_done()
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_ON
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_ON
hass.states.async_set("light.test1", STATE_OFF)
hass.states.async_set("light.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_OFF
# Initial state with no group member in the state machine -> unavailable
assert hass.states.get("light.light_group").state == STATE_UNAVAILABLE
# All group members unavailable -> unavailable
hass.states.async_set("light.test1", STATE_UNAVAILABLE)
hass.states.async_set("light.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNAVAILABLE
# All group members unknown -> unknown
hass.states.async_set("light.test1", STATE_UNKNOWN)
hass.states.async_set("light.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNKNOWN
# Group members unknown or unavailable -> unknown
hass.states.async_set("light.test1", STATE_UNKNOWN)
hass.states.async_set("light.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNKNOWN
# At least one member on -> group on
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_ON
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_ON
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_ON)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_ON
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_ON
# Otherwise -> off
hass.states.async_set("light.test1", STATE_OFF)
hass.states.async_set("light.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_OFF
hass.states.async_set("light.test1", STATE_UNKNOWN)
hass.states.async_set("light.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_OFF
hass.states.async_set("light.test1", STATE_UNAVAILABLE)
hass.states.async_set("light.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_OFF
# All group members removed from the state machine -> unavailable
hass.states.async_remove("light.test1")
hass.states.async_remove("light.test2")
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNAVAILABLE
async def test_state_reporting_all(hass):
"""Test the state reporting."""
"""Test the state reporting in 'all' mode.
The group state is unavailable if all group members are unavailable.
Otherwise, the group state is unknown if at least one group member is unknown or unavailable.
Otherwise, the group state is off if at least one group member is off.
Otherwise, the group state is on.
"""
await async_setup_component(
hass,
LIGHT_DOMAIN,
@ -143,11 +199,47 @@ async def test_state_reporting_all(hass):
await hass.async_start()
await hass.async_block_till_done()
# Initial state with no group member in the state machine -> unavailable
assert hass.states.get("light.light_group").state == STATE_UNAVAILABLE
# All group members unavailable -> unavailable
hass.states.async_set("light.test1", STATE_UNAVAILABLE)
hass.states.async_set("light.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNAVAILABLE
# At least one member unknown or unavailable -> group unknown
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNKNOWN
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNKNOWN
hass.states.async_set("light.test1", STATE_UNKNOWN)
hass.states.async_set("light.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNKNOWN
hass.states.async_set("light.test1", STATE_OFF)
hass.states.async_set("light.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNKNOWN
hass.states.async_set("light.test1", STATE_OFF)
hass.states.async_set("light.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNKNOWN
hass.states.async_set("binary_sensor.test1", STATE_UNKNOWN)
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNKNOWN
# At least one member off -> group off
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_OFF)
await hass.async_block_till_done()
@ -158,13 +250,15 @@ async def test_state_reporting_all(hass):
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_OFF
# Otherwise -> on
hass.states.async_set("light.test1", STATE_ON)
hass.states.async_set("light.test2", STATE_ON)
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_ON
hass.states.async_set("light.test1", STATE_UNAVAILABLE)
hass.states.async_set("light.test2", STATE_UNAVAILABLE)
# All group members removed from the state machine -> unavailable
hass.states.async_remove("light.test1")
hass.states.async_remove("light.test2")
await hass.async_block_till_done()
assert hass.states.get("light.light_group").state == STATE_UNAVAILABLE

122
tests/components/group/test_lock.py
View File

@ -57,7 +57,16 @@ async def test_default_state(hass):
async def test_state_reporting(hass):
"""Test the state reporting."""
"""Test the state reporting.
The group state is unavailable if all group members are unavailable.
Otherwise, the group state is unknown if at least one group member is unknown or unavailable.
Otherwise, the group state is jammed if at least one group member is jammed.
Otherwise, the group state is locking if at least one group member is locking.
Otherwise, the group state is unlocking if at least one group member is unlocking.
Otherwise, the group state is unlocked if at least one group member is unlocked.
Otherwise, the group state is locked.
"""
await async_setup_component(
hass,
LOCK_DOMAIN,
@ -72,43 +81,98 @@ async def test_state_reporting(hass):
await hass.async_start()
await hass.async_block_till_done()
hass.states.async_set("lock.test1", STATE_LOCKED)
# Initial state with no group member in the state machine -> unavailable
assert hass.states.get("lock.lock_group").state == STATE_UNAVAILABLE
# All group members unavailable -> unavailable
hass.states.async_set("lock.test1", STATE_UNAVAILABLE)
hass.states.async_set("lock.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_UNKNOWN
assert hass.states.get("lock.lock_group").state == STATE_UNAVAILABLE
hass.states.async_set("lock.test1", STATE_LOCKED)
hass.states.async_set("lock.test2", STATE_UNLOCKED)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_UNLOCKED
# At least one member unknown or unavailable -> group unknown
for state_1 in (
STATE_JAMMED,
STATE_LOCKED,
STATE_LOCKING,
STATE_UNKNOWN,
STATE_UNLOCKED,
STATE_UNLOCKING,
):
hass.states.async_set("lock.test1", state_1)
hass.states.async_set("lock.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_UNKNOWN
for state_1 in (
STATE_JAMMED,
STATE_LOCKED,
STATE_LOCKING,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
STATE_UNLOCKED,
STATE_UNLOCKING,
):
hass.states.async_set("lock.test1", state_1)
hass.states.async_set("lock.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_UNKNOWN
# At least one member jammed -> group jammed
for state_1 in (
STATE_JAMMED,
STATE_LOCKED,
STATE_LOCKING,
STATE_UNLOCKED,
STATE_UNLOCKING,
):
hass.states.async_set("lock.test1", state_1)
hass.states.async_set("lock.test2", STATE_JAMMED)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_JAMMED
# At least one member locking -> group unlocking
for state_1 in (
STATE_LOCKED,
STATE_LOCKING,
STATE_UNLOCKED,
STATE_UNLOCKING,
):
hass.states.async_set("lock.test1", state_1)
hass.states.async_set("lock.test2", STATE_LOCKING)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_LOCKING
# At least one member unlocking -> group unlocking
for state_1 in (
STATE_LOCKED,
STATE_UNLOCKED,
STATE_UNLOCKING,
):
hass.states.async_set("lock.test1", state_1)
hass.states.async_set("lock.test2", STATE_UNLOCKING)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_UNLOCKING
# At least one member unlocked -> group unlocked
for state_1 in (
STATE_LOCKED,
STATE_UNLOCKED,
):
hass.states.async_set("lock.test1", state_1)
hass.states.async_set("lock.test2", STATE_UNLOCKED)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_UNLOCKED
# Otherwise -> locked
hass.states.async_set("lock.test1", STATE_LOCKED)
hass.states.async_set("lock.test2", STATE_LOCKED)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_LOCKED
hass.states.async_set("lock.test1", STATE_UNLOCKED)
hass.states.async_set("lock.test2", STATE_UNLOCKED)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_UNLOCKED
hass.states.async_set("lock.test1", STATE_UNLOCKED)
hass.states.async_set("lock.test2", STATE_JAMMED)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_JAMMED
hass.states.async_set("lock.test1", STATE_LOCKED)
hass.states.async_set("lock.test2", STATE_UNLOCKING)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_UNLOCKING
hass.states.async_set("lock.test1", STATE_UNLOCKED)
hass.states.async_set("lock.test2", STATE_LOCKING)
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_LOCKING
hass.states.async_set("lock.test1", STATE_UNAVAILABLE)
hass.states.async_set("lock.test2", STATE_UNAVAILABLE)
# All group members removed from the state machine -> unavailable
hass.states.async_remove("lock.test1")
hass.states.async_remove("lock.test2")
await hass.async_block_till_done()
assert hass.states.get("lock.lock_group").state == STATE_UNAVAILABLE

77
tests/components/group/test_media_player.py
View File

@ -43,6 +43,8 @@ from homeassistant.const import (
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_UP,
STATE_BUFFERING,
STATE_IDLE,
STATE_OFF,
STATE_ON,
STATE_PAUSED,
@ -99,7 +101,17 @@ async def test_default_state(hass):
async def test_state_reporting(hass):
"""Test the state reporting."""
"""Test the state reporting.
The group state is unavailable if all group members are unavailable.
Otherwise, the group state is unknown if all group members are unknown.
Otherwise, the group state is buffering if all group members are buffering.
Otherwise, the group state is idle if all group members are idle.
Otherwise, the group state is paused if all group members are paused.
Otherwise, the group state is playing if all group members are playing.
Otherwise, the group state is on if at least one group member is not off, unavailable or unknown.
Otherwise, the group state is off.
"""
await async_setup_component(
hass,
MEDIA_DOMAIN,
@ -114,27 +126,60 @@ async def test_state_reporting(hass):
await hass.async_start()
await hass.async_block_till_done()
# Initial state with no group member in the state machine -> unknown
assert hass.states.get("media_player.media_group").state == STATE_UNKNOWN
hass.states.async_set("media_player.player_1", STATE_ON)
hass.states.async_set("media_player.player_2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("media_player.media_group").state == STATE_ON
# All group members buffering -> buffering
# All group members idle -> idle
# All group members paused -> paused
# All group members playing -> playing
# All group members unavailable -> unavailable
# All group members unknown -> unknown
for state in (
STATE_BUFFERING,
STATE_IDLE,
STATE_PAUSED,
STATE_PLAYING,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
hass.states.async_set("media_player.player_1", state)
hass.states.async_set("media_player.player_2", state)
await hass.async_block_till_done()
assert hass.states.get("media_player.media_group").state == state
hass.states.async_set("media_player.player_1", STATE_ON)
hass.states.async_set("media_player.player_2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("media_player.media_group").state == STATE_ON
# At least one member not off, unavailable or unknown -> on
for state_1 in (STATE_BUFFERING, STATE_IDLE, STATE_ON, STATE_PAUSED, STATE_PLAYING):
for state_2 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set("media_player.player_1", state_1)
hass.states.async_set("media_player.player_2", state_2)
await hass.async_block_till_done()
assert hass.states.get("media_player.media_group").state == STATE_ON
hass.states.async_set("media_player.player_1", STATE_OFF)
hass.states.async_set("media_player.player_2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("media_player.media_group").state == STATE_OFF
for state_1 in (STATE_OFF, STATE_UNAVAILABLE, STATE_UNKNOWN):
hass.states.async_set("media_player.player_1", state_1)
hass.states.async_set("media_player.player_2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("media_player.media_group").state == STATE_OFF
hass.states.async_set("media_player.player_1", STATE_UNAVAILABLE)
hass.states.async_set("media_player.player_2", STATE_UNAVAILABLE)
# Otherwise off
for state_1 in (STATE_OFF, STATE_UNKNOWN):
hass.states.async_set("media_player.player_1", state_1)
hass.states.async_set("media_player.player_2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("media_player.media_group").state == STATE_OFF
for state_1 in (STATE_OFF, STATE_UNAVAILABLE):
hass.states.async_set("media_player.player_1", state_1)
hass.states.async_set("media_player.player_2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("media_player.media_group").state == STATE_OFF
# All group members removed from the state machine -> unknown
hass.states.async_remove("media_player.player_1")
hass.states.async_remove("media_player.player_2")
await hass.async_block_till_done()
assert hass.states.get("media_player.media_group").state == STATE_UNAVAILABLE
assert hass.states.get("media_player.media_group").state == STATE_UNKNOWN
async def test_supported_features(hass):

130
tests/components/group/test_switch.py
View File

@ -56,7 +56,13 @@ async def test_default_state(hass):
async def test_state_reporting(hass):
"""Test the state reporting."""
"""Test the state reporting in 'any' mode.
The group state is unavailable if all group members are unavailable.
Otherwise, the group state is unknown if all group members are unknown.
Otherwise, the group state is on if at least one group member is on.
Otherwise, the group state is off.
"""
await async_setup_component(
hass,
SWITCH_DOMAIN,
@ -72,29 +78,79 @@ async def test_state_reporting(hass):
await hass.async_start()
await hass.async_block_till_done()
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_ON
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_ON
hass.states.async_set("switch.test1", STATE_OFF)
hass.states.async_set("switch.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_OFF
# Initial state with no group member in the state machine -> unavailable
assert hass.states.get("switch.switch_group").state == STATE_UNAVAILABLE
# All group members unavailable -> unavailable
hass.states.async_set("switch.test1", STATE_UNAVAILABLE)
hass.states.async_set("switch.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNAVAILABLE
# All group members unknown -> unknown
hass.states.async_set("switch.test1", STATE_UNKNOWN)
hass.states.async_set("switch.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNKNOWN
# Group members unknown or unavailable -> unknown
hass.states.async_set("switch.test1", STATE_UNKNOWN)
hass.states.async_set("switch.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNKNOWN
# At least one member on -> group on
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_ON
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_ON
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_ON)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_ON
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_ON
# Otherwise -> off
hass.states.async_set("switch.test1", STATE_OFF)
hass.states.async_set("switch.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_OFF
hass.states.async_set("switch.test1", STATE_UNKNOWN)
hass.states.async_set("switch.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_OFF
hass.states.async_set("switch.test1", STATE_UNAVAILABLE)
hass.states.async_set("switch.test2", STATE_OFF)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_OFF
# All group members removed from the state machine -> unavailable
hass.states.async_remove("switch.test1")
hass.states.async_remove("switch.test2")
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNAVAILABLE
async def test_state_reporting_all(hass):
"""Test the state reporting."""
"""Test the state reporting in 'all' mode.
The group state is unavailable if all group members are unavailable.
Otherwise, the group state is unknown if at least one group member is unknown or unavailable.
Otherwise, the group state is off if at least one group member is off.
Otherwise, the group state is on.
"""
await async_setup_component(
hass,
SWITCH_DOMAIN,
@ -110,11 +166,47 @@ async def test_state_reporting_all(hass):
await hass.async_start()
await hass.async_block_till_done()
# Initial state with no group member in the state machine -> unavailable
assert hass.states.get("switch.switch_group").state == STATE_UNAVAILABLE
# All group members unavailable -> unavailable
hass.states.async_set("switch.test1", STATE_UNAVAILABLE)
hass.states.async_set("switch.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNAVAILABLE
# At least one member unknown or unavailable -> group unknown
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNKNOWN
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNKNOWN
hass.states.async_set("switch.test1", STATE_UNKNOWN)
hass.states.async_set("switch.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNKNOWN
hass.states.async_set("switch.test1", STATE_OFF)
hass.states.async_set("switch.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNKNOWN
hass.states.async_set("switch.test1", STATE_OFF)
hass.states.async_set("switch.test2", STATE_UNKNOWN)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNKNOWN
hass.states.async_set("switch.test1", STATE_UNKNOWN)
hass.states.async_set("switch.test2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNKNOWN
# At least one member off -> group off
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_OFF)
await hass.async_block_till_done()
@ -125,13 +217,15 @@ async def test_state_reporting_all(hass):
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_OFF
# Otherwise -> on
hass.states.async_set("switch.test1", STATE_ON)
hass.states.async_set("switch.test2", STATE_ON)
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_ON
hass.states.async_set("switch.test1", STATE_UNAVAILABLE)
hass.states.async_set("switch.test2", STATE_UNAVAILABLE)
# All group members removed from the state machine -> unavailable
hass.states.async_remove("switch.test1")
hass.states.async_remove("switch.test2")
await hass.async_block_till_done()
assert hass.states.get("switch.switch_group").state == STATE_UNAVAILABLE