Rewrite group unittest to pytest style (#42159)
This commit is contained in:
Tim Werner
GitHub
parent
3fa97f5fc0
commit
3f623ad453
2 changed files with 361 additions and 309 deletions
|
|
@ -1,7 +1,6 @@
|
||||||
"""The tests for the Group components."""
|
"""The tests for the Group components."""
|
||||||
# pylint: disable=protected-access
|
# pylint: disable=protected-access
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
import unittest
|
|
||||||
|
|
||||||
import homeassistant.components.group as group
|
import homeassistant.components.group as group
|
||||||
from homeassistant.const import (
|
from homeassistant.const import (
|
||||||
|
|
@ -18,307 +17,375 @@ from homeassistant.const import (
|
||||||
)
|
)
|
||||||
from homeassistant.core import CoreState
|
from homeassistant.core import CoreState
|
||||||
from homeassistant.helpers.event import TRACK_STATE_CHANGE_CALLBACKS
|
from homeassistant.helpers.event import TRACK_STATE_CHANGE_CALLBACKS
|
||||||
from homeassistant.setup import async_setup_component, setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
|
||||||
from tests.async_mock import patch
|
from tests.async_mock import patch
|
||||||
from tests.common import assert_setup_component, get_test_home_assistant
|
from tests.common import assert_setup_component
|
||||||
from tests.components.group import common
|
from tests.components.group import common
|
||||||
|
|
||||||
|
|
||||||
class TestComponentsGroup(unittest.TestCase):
|
async def test_setup_group_with_mixed_groupable_states(hass):
|
||||||
"""Test Group component."""
|
"""Try to set up a group with mixed groupable states."""
|
||||||
|
|
||||||
# pylint: disable=invalid-name
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
def setUp(self):
|
hass.states.async_set("device_tracker.Paulus", STATE_HOME)
|
||||||
"""Set up things to be run when tests are started."""
|
|
||||||
self.hass = get_test_home_assistant()
|
|
||||||
for domain in ["device_tracker", "light", "group", "sensor"]:
|
|
||||||
setup_component(self.hass, domain, {})
|
|
||||||
self.addCleanup(self.hass.stop)
|
|
||||||
|
|
||||||
def test_setup_group_with_mixed_groupable_states(self):
|
assert await async_setup_component(hass, "group", {})
|
||||||
"""Try to set up a group with mixed groupable states."""
|
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
|
||||||
self.hass.states.set("device_tracker.Paulus", STATE_HOME)
|
|
||||||
group.Group.create_group(
|
|
||||||
self.hass, "person_and_light", ["light.Bowl", "device_tracker.Paulus"]
|
|
||||||
)
|
|
||||||
|
|
||||||
assert (
|
await group.Group.async_create_group(
|
||||||
STATE_ON == self.hass.states.get(f"{group.DOMAIN}.person_and_light").state
|
hass, "person_and_light", ["light.Bowl", "device_tracker.Paulus"]
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_setup_group_with_a_non_existing_state(self):
|
await hass.async_block_till_done()
|
||||||
"""Try to set up a group with a non existing state."""
|
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
|
||||||
|
|
||||||
grp = group.Group.create_group(
|
assert STATE_ON == hass.states.get(f"{group.DOMAIN}.person_and_light").state
|
||||||
self.hass, "light_and_nothing", ["light.Bowl", "non.existing"]
|
|
||||||
)
|
|
||||||
|
|
||||||
assert STATE_ON == grp.state
|
|
||||||
|
|
||||||
def test_setup_group_with_non_groupable_states(self):
|
async def test_setup_group_with_a_non_existing_state(hass):
|
||||||
"""Test setup with groups which are not groupable."""
|
"""Try to set up a group with a non existing state."""
|
||||||
self.hass.states.set("cast.living_room", "Plex")
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
self.hass.states.set("cast.bedroom", "Netflix")
|
|
||||||
|
|
||||||
grp = group.Group.create_group(
|
assert await async_setup_component(hass, "group", {})
|
||||||
self.hass, "chromecasts", ["cast.living_room", "cast.bedroom"]
|
|
||||||
)
|
|
||||||
|
|
||||||
assert grp.state is None
|
grp = await group.Group.async_create_group(
|
||||||
|
hass, "light_and_nothing", ["light.Bowl", "non.existing"]
|
||||||
|
)
|
||||||
|
|
||||||
def test_setup_empty_group(self):
|
assert STATE_ON == grp.state
|
||||||
"""Try to set up an empty group."""
|
|
||||||
grp = group.Group.create_group(self.hass, "nothing", [])
|
|
||||||
|
|
||||||
assert grp.state is None
|
|
||||||
|
|
||||||
def test_monitor_group(self):
|
async def test_setup_group_with_non_groupable_states(hass):
|
||||||
"""Test if the group keeps track of states."""
|
"""Test setup with groups which are not groupable."""
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
hass.states.async_set("cast.living_room", "Plex")
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
hass.states.async_set("cast.bedroom", "Netflix")
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
|
||||||
)
|
|
||||||
|
|
||||||
# Test if group setup in our init mode is ok
|
assert await async_setup_component(hass, "group", {})
|
||||||
assert test_group.entity_id in self.hass.states.entity_ids()
|
|
||||||
|
|
||||||
group_state = self.hass.states.get(test_group.entity_id)
|
grp = await group.Group.async_create_group(
|
||||||
assert STATE_ON == group_state.state
|
hass, "chromecasts", ["cast.living_room", "cast.bedroom"]
|
||||||
assert group_state.attributes.get(group.ATTR_AUTO)
|
)
|
||||||
|
|
||||||
def test_group_turns_off_if_all_off(self):
|
assert grp.state is None
|
||||||
"""Test if turn off if the last device that was on turns off."""
|
|
||||||
self.hass.states.set("light.Bowl", STATE_OFF)
|
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
|
||||||
)
|
|
||||||
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
group_state = self.hass.states.get(test_group.entity_id)
|
async def test_setup_empty_group(hass):
|
||||||
assert STATE_OFF == group_state.state
|
"""Try to set up an empty group."""
|
||||||
|
grp = await group.Group.async_create_group(hass, "nothing", [])
|
||||||
|
|
||||||
def test_group_turns_on_if_all_are_off_and_one_turns_on(self):
|
assert grp.state is None
|
||||||
"""Test if turn on if all devices were turned off and one turns on."""
|
|
||||||
self.hass.states.set("light.Bowl", STATE_OFF)
|
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
|
||||||
)
|
|
||||||
|
|
||||||
# Turn one on
|
|
||||||
self.hass.states.set("light.Ceiling", STATE_ON)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
group_state = self.hass.states.get(test_group.entity_id)
|
async def test_monitor_group(hass):
|
||||||
assert STATE_ON == group_state.state
|
"""Test if the group keeps track of states."""
|
||||||
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
|
|
||||||
def test_allgroup_stays_off_if_all_are_off_and_one_turns_on(self):
|
assert await async_setup_component(hass, "group", {})
|
||||||
"""Group with all: true, stay off if one device turns on."""
|
|
||||||
self.hass.states.set("light.Bowl", STATE_OFF)
|
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "init_group", ["light.Bowl", "light.Ceiling"], False, mode=True
|
|
||||||
)
|
|
||||||
|
|
||||||
# Turn one on
|
test_group = await group.Group.async_create_group(
|
||||||
self.hass.states.set("light.Ceiling", STATE_ON)
|
hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
||||||
self.hass.block_till_done()
|
)
|
||||||
|
|
||||||
group_state = self.hass.states.get(test_group.entity_id)
|
# Test if group setup in our init mode is ok
|
||||||
assert STATE_OFF == group_state.state
|
assert test_group.entity_id in hass.states.async_entity_ids()
|
||||||
|
|
||||||
def test_allgroup_turn_on_if_last_turns_on(self):
|
group_state = hass.states.get(test_group.entity_id)
|
||||||
"""Group with all: true, turn on if all devices are on."""
|
assert STATE_ON == group_state.state
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
assert group_state.attributes.get(group.ATTR_AUTO)
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "init_group", ["light.Bowl", "light.Ceiling"], False, mode=True
|
|
||||||
)
|
|
||||||
|
|
||||||
# Turn one on
|
|
||||||
self.hass.states.set("light.Ceiling", STATE_ON)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
group_state = self.hass.states.get(test_group.entity_id)
|
async def test_group_turns_off_if_all_off(hass):
|
||||||
assert STATE_ON == group_state.state
|
"""Test if turn off if the last device that was on turns off."""
|
||||||
|
hass.states.async_set("light.Bowl", STATE_OFF)
|
||||||
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
|
|
||||||
def test_expand_entity_ids(self):
|
assert await async_setup_component(hass, "group", {})
|
||||||
"""Test expand_entity_ids method."""
|
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
|
||||||
)
|
|
||||||
|
|
||||||
assert sorted(["light.ceiling", "light.bowl"]) == sorted(
|
test_group = await group.Group.async_create_group(
|
||||||
group.expand_entity_ids(self.hass, [test_group.entity_id])
|
hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_expand_entity_ids_does_not_return_duplicates(self):
|
await hass.async_block_till_done()
|
||||||
"""Test that expand_entity_ids does not return duplicates."""
|
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
|
||||||
)
|
|
||||||
|
|
||||||
assert ["light.bowl", "light.ceiling"] == sorted(
|
group_state = hass.states.get(test_group.entity_id)
|
||||||
group.expand_entity_ids(self.hass, [test_group.entity_id, "light.Ceiling"])
|
assert STATE_OFF == group_state.state
|
||||||
)
|
|
||||||
|
|
||||||
assert ["light.bowl", "light.ceiling"] == sorted(
|
|
||||||
group.expand_entity_ids(self.hass, ["light.bowl", test_group.entity_id])
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_expand_entity_ids_recursive(self):
|
async def test_group_turns_on_if_all_are_off_and_one_turns_on(hass):
|
||||||
"""Test expand_entity_ids method with a group that contains itself."""
|
"""Test if turn on if all devices were turned off and one turns on."""
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
hass.states.async_set("light.Bowl", STATE_OFF)
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass,
|
|
||||||
"init_group",
|
|
||||||
["light.Bowl", "light.Ceiling", "group.init_group"],
|
|
||||||
False,
|
|
||||||
)
|
|
||||||
|
|
||||||
assert sorted(["light.ceiling", "light.bowl"]) == sorted(
|
assert await async_setup_component(hass, "group", {})
|
||||||
group.expand_entity_ids(self.hass, [test_group.entity_id])
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_expand_entity_ids_ignores_non_strings(self):
|
test_group = await group.Group.async_create_group(
|
||||||
"""Test that non string elements in lists are ignored."""
|
hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
||||||
assert [] == group.expand_entity_ids(self.hass, [5, True])
|
)
|
||||||
|
|
||||||
def test_get_entity_ids(self):
|
# Turn one on
|
||||||
"""Test get_entity_ids method."""
|
hass.states.async_set("light.Ceiling", STATE_ON)
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
await hass.async_block_till_done()
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
|
||||||
)
|
|
||||||
|
|
||||||
assert ["light.bowl", "light.ceiling"] == sorted(
|
group_state = hass.states.get(test_group.entity_id)
|
||||||
group.get_entity_ids(self.hass, test_group.entity_id)
|
assert STATE_ON == group_state.state
|
||||||
)
|
|
||||||
|
|
||||||
def test_get_entity_ids_with_domain_filter(self):
|
|
||||||
"""Test if get_entity_ids works with a domain_filter."""
|
|
||||||
self.hass.states.set("switch.AC", STATE_OFF)
|
|
||||||
|
|
||||||
mixed_group = group.Group.create_group(
|
async def test_allgroup_stays_off_if_all_are_off_and_one_turns_on(hass):
|
||||||
self.hass, "mixed_group", ["light.Bowl", "switch.AC"], False
|
"""Group with all: true, stay off if one device turns on."""
|
||||||
)
|
hass.states.async_set("light.Bowl", STATE_OFF)
|
||||||
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
|
|
||||||
assert ["switch.ac"] == group.get_entity_ids(
|
assert await async_setup_component(hass, "group", {})
|
||||||
self.hass, mixed_group.entity_id, domain_filter="switch"
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_get_entity_ids_with_non_existing_group_name(self):
|
test_group = await group.Group.async_create_group(
|
||||||
"""Test get_entity_ids with a non existing group."""
|
hass, "init_group", ["light.Bowl", "light.Ceiling"], False, mode=True
|
||||||
assert [] == group.get_entity_ids(self.hass, "non_existing")
|
)
|
||||||
|
|
||||||
def test_get_entity_ids_with_non_group_state(self):
|
# Turn one on
|
||||||
"""Test get_entity_ids with a non group state."""
|
hass.states.async_set("light.Ceiling", STATE_ON)
|
||||||
assert [] == group.get_entity_ids(self.hass, "switch.AC")
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
def test_group_being_init_before_first_tracked_state_is_set_to_on(self):
|
group_state = hass.states.get(test_group.entity_id)
|
||||||
"""Test if the groups turn on.
|
assert STATE_OFF == group_state.state
|
||||||
|
|
||||||
If no states existed and now a state it is tracking is being added
|
|
||||||
as ON.
|
|
||||||
"""
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "test group", ["light.not_there_1"]
|
|
||||||
)
|
|
||||||
|
|
||||||
self.hass.states.set("light.not_there_1", STATE_ON)
|
async def test_allgroup_turn_on_if_last_turns_on(hass):
|
||||||
|
"""Group with all: true, turn on if all devices are on."""
|
||||||
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
|
|
||||||
self.hass.block_till_done()
|
assert await async_setup_component(hass, "group", {})
|
||||||
|
|
||||||
group_state = self.hass.states.get(test_group.entity_id)
|
test_group = await group.Group.async_create_group(
|
||||||
assert STATE_ON == group_state.state
|
hass, "init_group", ["light.Bowl", "light.Ceiling"], False, mode=True
|
||||||
|
)
|
||||||
|
|
||||||
def test_group_being_init_before_first_tracked_state_is_set_to_off(self):
|
# Turn one on
|
||||||
"""Test if the group turns off.
|
hass.states.async_set("light.Ceiling", STATE_ON)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
If no states existed and now a state it is tracking is being added
|
group_state = hass.states.get(test_group.entity_id)
|
||||||
as OFF.
|
assert STATE_ON == group_state.state
|
||||||
"""
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "test group", ["light.not_there_1"]
|
|
||||||
)
|
|
||||||
|
|
||||||
self.hass.states.set("light.not_there_1", STATE_OFF)
|
|
||||||
|
|
||||||
self.hass.block_till_done()
|
async def test_expand_entity_ids(hass):
|
||||||
|
"""Test expand_entity_ids method."""
|
||||||
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
|
|
||||||
group_state = self.hass.states.get(test_group.entity_id)
|
assert await async_setup_component(hass, "group", {})
|
||||||
assert STATE_OFF == group_state.state
|
|
||||||
|
|
||||||
def test_groups_get_unique_names(self):
|
test_group = await group.Group.async_create_group(
|
||||||
"""Two groups with same name should both have a unique entity id."""
|
hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
||||||
grp1 = group.Group.create_group(self.hass, "Je suis Charlie")
|
)
|
||||||
grp2 = group.Group.create_group(self.hass, "Je suis Charlie")
|
|
||||||
|
|
||||||
assert grp1.entity_id != grp2.entity_id
|
assert sorted(["light.ceiling", "light.bowl"]) == sorted(
|
||||||
|
group.expand_entity_ids(hass, [test_group.entity_id])
|
||||||
|
)
|
||||||
|
|
||||||
def test_expand_entity_ids_expands_nested_groups(self):
|
|
||||||
"""Test if entity ids epands to nested groups."""
|
|
||||||
group.Group.create_group(self.hass, "light", ["light.test_1", "light.test_2"])
|
|
||||||
group.Group.create_group(
|
|
||||||
self.hass, "switch", ["switch.test_1", "switch.test_2"]
|
|
||||||
)
|
|
||||||
group.Group.create_group(
|
|
||||||
self.hass, "group_of_groups", ["group.light", "group.switch"]
|
|
||||||
)
|
|
||||||
|
|
||||||
assert [
|
async def test_expand_entity_ids_does_not_return_duplicates(hass):
|
||||||
"light.test_1",
|
"""Test that expand_entity_ids does not return duplicates."""
|
||||||
"light.test_2",
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
"switch.test_1",
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
"switch.test_2",
|
|
||||||
] == sorted(group.expand_entity_ids(self.hass, ["group.group_of_groups"]))
|
|
||||||
|
|
||||||
def test_set_assumed_state_based_on_tracked(self):
|
assert await async_setup_component(hass, "group", {})
|
||||||
"""Test assumed state."""
|
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
|
||||||
self.hass.states.set("light.Ceiling", STATE_OFF)
|
|
||||||
test_group = group.Group.create_group(
|
|
||||||
self.hass, "init_group", ["light.Bowl", "light.Ceiling", "sensor.no_exist"]
|
|
||||||
)
|
|
||||||
|
|
||||||
state = self.hass.states.get(test_group.entity_id)
|
test_group = await group.Group.async_create_group(
|
||||||
assert not state.attributes.get(ATTR_ASSUMED_STATE)
|
hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
||||||
|
)
|
||||||
|
|
||||||
self.hass.states.set("light.Bowl", STATE_ON, {ATTR_ASSUMED_STATE: True})
|
assert ["light.bowl", "light.ceiling"] == sorted(
|
||||||
self.hass.block_till_done()
|
group.expand_entity_ids(hass, [test_group.entity_id, "light.Ceiling"])
|
||||||
|
)
|
||||||
|
|
||||||
state = self.hass.states.get(test_group.entity_id)
|
assert ["light.bowl", "light.ceiling"] == sorted(
|
||||||
assert state.attributes.get(ATTR_ASSUMED_STATE)
|
group.expand_entity_ids(hass, ["light.bowl", test_group.entity_id])
|
||||||
|
)
|
||||||
|
|
||||||
self.hass.states.set("light.Bowl", STATE_ON)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
state = self.hass.states.get(test_group.entity_id)
|
async def test_expand_entity_ids_recursive(hass):
|
||||||
assert not state.attributes.get(ATTR_ASSUMED_STATE)
|
"""Test expand_entity_ids method with a group that contains itself."""
|
||||||
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
|
|
||||||
def test_group_updated_after_device_tracker_zone_change(self):
|
assert await async_setup_component(hass, "group", {})
|
||||||
"""Test group state when device tracker in group changes zone."""
|
|
||||||
self.hass.states.set("device_tracker.Adam", STATE_HOME)
|
test_group = await group.Group.async_create_group(
|
||||||
self.hass.states.set("device_tracker.Eve", STATE_NOT_HOME)
|
hass,
|
||||||
self.hass.block_till_done()
|
"init_group",
|
||||||
group.Group.create_group(
|
["light.Bowl", "light.Ceiling", "group.init_group"],
|
||||||
self.hass, "peeps", ["device_tracker.Adam", "device_tracker.Eve"]
|
False,
|
||||||
)
|
)
|
||||||
self.hass.states.set("device_tracker.Adam", "cool_state_not_home")
|
|
||||||
self.hass.block_till_done()
|
assert sorted(["light.ceiling", "light.bowl"]) == sorted(
|
||||||
assert STATE_NOT_HOME == self.hass.states.get(f"{group.DOMAIN}.peeps").state
|
group.expand_entity_ids(hass, [test_group.entity_id])
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_expand_entity_ids_ignores_non_strings(hass):
|
||||||
|
"""Test that non string elements in lists are ignored."""
|
||||||
|
assert [] == group.expand_entity_ids(hass, [5, True])
|
||||||
|
|
||||||
|
|
||||||
|
async def test_get_entity_ids(hass):
|
||||||
|
"""Test get_entity_ids method."""
|
||||||
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
|
|
||||||
|
assert await async_setup_component(hass, "group", {})
|
||||||
|
|
||||||
|
test_group = await group.Group.async_create_group(
|
||||||
|
hass, "init_group", ["light.Bowl", "light.Ceiling"], False
|
||||||
|
)
|
||||||
|
|
||||||
|
assert ["light.bowl", "light.ceiling"] == sorted(
|
||||||
|
group.get_entity_ids(hass, test_group.entity_id)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_get_entity_ids_with_domain_filter(hass):
|
||||||
|
"""Test if get_entity_ids works with a domain_filter."""
|
||||||
|
hass.states.async_set("switch.AC", STATE_OFF)
|
||||||
|
|
||||||
|
assert await async_setup_component(hass, "group", {})
|
||||||
|
|
||||||
|
mixed_group = await group.Group.async_create_group(
|
||||||
|
hass, "mixed_group", ["light.Bowl", "switch.AC"], False
|
||||||
|
)
|
||||||
|
|
||||||
|
assert ["switch.ac"] == group.get_entity_ids(
|
||||||
|
hass, mixed_group.entity_id, domain_filter="switch"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_get_entity_ids_with_non_existing_group_name(hass):
|
||||||
|
"""Test get_entity_ids with a non existing group."""
|
||||||
|
assert [] == group.get_entity_ids(hass, "non_existing")
|
||||||
|
|
||||||
|
|
||||||
|
async def test_get_entity_ids_with_non_group_state(hass):
|
||||||
|
"""Test get_entity_ids with a non group state."""
|
||||||
|
assert [] == group.get_entity_ids(hass, "switch.AC")
|
||||||
|
|
||||||
|
|
||||||
|
async def test_group_being_init_before_first_tracked_state_is_set_to_on(hass):
|
||||||
|
"""Test if the groups turn on.
|
||||||
|
|
||||||
|
If no states existed and now a state it is tracking is being added
|
||||||
|
as ON.
|
||||||
|
"""
|
||||||
|
|
||||||
|
assert await async_setup_component(hass, "group", {})
|
||||||
|
|
||||||
|
test_group = await group.Group.async_create_group(
|
||||||
|
hass, "test group", ["light.not_there_1"]
|
||||||
|
)
|
||||||
|
|
||||||
|
hass.states.async_set("light.not_there_1", STATE_ON)
|
||||||
|
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
group_state = hass.states.get(test_group.entity_id)
|
||||||
|
assert STATE_ON == group_state.state
|
||||||
|
|
||||||
|
|
||||||
|
async def test_group_being_init_before_first_tracked_state_is_set_to_off(hass):
|
||||||
|
"""Test if the group turns off.
|
||||||
|
|
||||||
|
If no states existed and now a state it is tracking is being added
|
||||||
|
as OFF.
|
||||||
|
"""
|
||||||
|
assert await async_setup_component(hass, "group", {})
|
||||||
|
test_group = await group.Group.async_create_group(
|
||||||
|
hass, "test group", ["light.not_there_1"]
|
||||||
|
)
|
||||||
|
|
||||||
|
hass.states.async_set("light.not_there_1", STATE_OFF)
|
||||||
|
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
group_state = hass.states.get(test_group.entity_id)
|
||||||
|
assert STATE_OFF == group_state.state
|
||||||
|
|
||||||
|
|
||||||
|
async def test_groups_get_unique_names(hass):
|
||||||
|
"""Two groups with same name should both have a unique entity id."""
|
||||||
|
|
||||||
|
assert await async_setup_component(hass, "group", {})
|
||||||
|
|
||||||
|
grp1 = await group.Group.async_create_group(hass, "Je suis Charlie")
|
||||||
|
grp2 = await group.Group.async_create_group(hass, "Je suis Charlie")
|
||||||
|
|
||||||
|
assert grp1.entity_id != grp2.entity_id
|
||||||
|
|
||||||
|
|
||||||
|
async def test_expand_entity_ids_expands_nested_groups(hass):
|
||||||
|
"""Test if entity ids epands to nested groups."""
|
||||||
|
|
||||||
|
assert await async_setup_component(hass, "group", {})
|
||||||
|
|
||||||
|
await group.Group.async_create_group(
|
||||||
|
hass, "light", ["light.test_1", "light.test_2"]
|
||||||
|
)
|
||||||
|
await group.Group.async_create_group(
|
||||||
|
hass, "switch", ["switch.test_1", "switch.test_2"]
|
||||||
|
)
|
||||||
|
await group.Group.async_create_group(
|
||||||
|
hass, "group_of_groups", ["group.light", "group.switch"]
|
||||||
|
)
|
||||||
|
|
||||||
|
assert [
|
||||||
|
"light.test_1",
|
||||||
|
"light.test_2",
|
||||||
|
"switch.test_1",
|
||||||
|
"switch.test_2",
|
||||||
|
] == sorted(group.expand_entity_ids(hass, ["group.group_of_groups"]))
|
||||||
|
|
||||||
|
|
||||||
|
async def test_set_assumed_state_based_on_tracked(hass):
|
||||||
|
"""Test assumed state."""
|
||||||
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
|
hass.states.async_set("light.Ceiling", STATE_OFF)
|
||||||
|
|
||||||
|
assert await async_setup_component(hass, "group", {})
|
||||||
|
|
||||||
|
test_group = await group.Group.async_create_group(
|
||||||
|
hass, "init_group", ["light.Bowl", "light.Ceiling", "sensor.no_exist"]
|
||||||
|
)
|
||||||
|
|
||||||
|
state = hass.states.get(test_group.entity_id)
|
||||||
|
assert not state.attributes.get(ATTR_ASSUMED_STATE)
|
||||||
|
|
||||||
|
hass.states.async_set("light.Bowl", STATE_ON, {ATTR_ASSUMED_STATE: True})
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
state = hass.states.get(test_group.entity_id)
|
||||||
|
assert state.attributes.get(ATTR_ASSUMED_STATE)
|
||||||
|
|
||||||
|
hass.states.async_set("light.Bowl", STATE_ON)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
state = hass.states.get(test_group.entity_id)
|
||||||
|
assert not state.attributes.get(ATTR_ASSUMED_STATE)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_group_updated_after_device_tracker_zone_change(hass):
|
||||||
|
"""Test group state when device tracker in group changes zone."""
|
||||||
|
hass.states.async_set("device_tracker.Adam", STATE_HOME)
|
||||||
|
hass.states.async_set("device_tracker.Eve", STATE_NOT_HOME)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
assert await async_setup_component(hass, "group", {})
|
||||||
|
assert await async_setup_component(hass, "device_tracker", {})
|
||||||
|
|
||||||
|
await group.Group.async_create_group(
|
||||||
|
hass, "peeps", ["device_tracker.Adam", "device_tracker.Eve"]
|
||||||
|
)
|
||||||
|
|
||||||
|
hass.states.async_set("device_tracker.Adam", "cool_state_not_home")
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
assert STATE_NOT_HOME == hass.states.get(f"{group.DOMAIN}.peeps").state
|
||||||
|
|
||||||
|
|
||||||
async def test_is_on(hass):
|
async def test_is_on(hass):
|
||||||
|
|
|
||||||
|
|
@ -1,98 +1,83 @@
|
||||||
"""The tests for the notify.group platform."""
|
"""The tests for the notify.group platform."""
|
||||||
import asyncio
|
|
||||||
from os import path
|
from os import path
|
||||||
import unittest
|
|
||||||
|
|
||||||
from homeassistant import config as hass_config
|
from homeassistant import config as hass_config
|
||||||
import homeassistant.components.demo.notify as demo
|
import homeassistant.components.demo.notify as demo
|
||||||
from homeassistant.components.group import SERVICE_RELOAD
|
from homeassistant.components.group import SERVICE_RELOAD
|
||||||
import homeassistant.components.group.notify as group
|
import homeassistant.components.group.notify as group
|
||||||
import homeassistant.components.notify as notify
|
import homeassistant.components.notify as notify
|
||||||
from homeassistant.setup import async_setup_component, setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
|
||||||
from tests.async_mock import MagicMock, patch
|
from tests.async_mock import MagicMock, patch
|
||||||
from tests.common import assert_setup_component, get_test_home_assistant
|
|
||||||
|
|
||||||
|
|
||||||
class TestNotifyGroup(unittest.TestCase):
|
async def test_send_message_with_data(hass):
|
||||||
"""Test the notify.group platform."""
|
"""Test sending a message with to a notify group."""
|
||||||
|
service1 = demo.DemoNotificationService(hass)
|
||||||
|
service2 = demo.DemoNotificationService(hass)
|
||||||
|
|
||||||
def setUp(self): # pylint: disable=invalid-name
|
service1.send_message = MagicMock(autospec=True)
|
||||||
"""Set up things to be run when tests are started."""
|
service2.send_message = MagicMock(autospec=True)
|
||||||
self.hass = get_test_home_assistant()
|
|
||||||
self.events = []
|
|
||||||
self.service1 = demo.DemoNotificationService(self.hass)
|
|
||||||
self.service2 = demo.DemoNotificationService(self.hass)
|
|
||||||
|
|
||||||
self.service1.send_message = MagicMock(autospec=True)
|
def mock_get_service(hass, config, discovery_info=None):
|
||||||
self.service2.send_message = MagicMock(autospec=True)
|
if config["name"] == "demo1":
|
||||||
|
return service1
|
||||||
|
return service2
|
||||||
|
|
||||||
def mock_get_service(hass, config, discovery_info=None):
|
assert await async_setup_component(
|
||||||
if config["name"] == "demo1":
|
hass,
|
||||||
return self.service1
|
"group",
|
||||||
return self.service2
|
{},
|
||||||
|
)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
with assert_setup_component(2, notify.DOMAIN), patch.object(
|
with patch.object(demo, "get_service", mock_get_service):
|
||||||
demo, "get_service", mock_get_service
|
await async_setup_component(
|
||||||
):
|
hass,
|
||||||
setup_component(
|
notify.DOMAIN,
|
||||||
self.hass,
|
{
|
||||||
notify.DOMAIN,
|
"notify": [
|
||||||
|
{"name": "demo1", "platform": "demo"},
|
||||||
|
{"name": "demo2", "platform": "demo"},
|
||||||
|
]
|
||||||
|
},
|
||||||
|
)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
service = await group.async_get_service(
|
||||||
|
hass,
|
||||||
|
{
|
||||||
|
"services": [
|
||||||
|
{"service": "demo1"},
|
||||||
{
|
{
|
||||||
"notify": [
|
"service": "demo2",
|
||||||
{"name": "demo1", "platform": "demo"},
|
"data": {
|
||||||
{"name": "demo2", "platform": "demo"},
|
"target": "unnamed device",
|
||||||
]
|
"data": {"test": "message"},
|
||||||
|
},
|
||||||
},
|
},
|
||||||
)
|
]
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
self.service = asyncio.run_coroutine_threadsafe(
|
"""Test sending a message with to a notify group."""
|
||||||
group.async_get_service(
|
await service.async_send_message(
|
||||||
self.hass,
|
"Hello", title="Test notification", data={"hello": "world"}
|
||||||
{
|
)
|
||||||
"services": [
|
|
||||||
{"service": "demo1"},
|
|
||||||
{
|
|
||||||
"service": "demo2",
|
|
||||||
"data": {
|
|
||||||
"target": "unnamed device",
|
|
||||||
"data": {"test": "message"},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
]
|
|
||||||
},
|
|
||||||
),
|
|
||||||
self.hass.loop,
|
|
||||||
).result()
|
|
||||||
|
|
||||||
assert self.service is not None
|
await hass.async_block_till_done()
|
||||||
self.addCleanup(self.tear_down_cleanup)
|
|
||||||
|
|
||||||
def tear_down_cleanup(self):
|
assert service1.send_message.mock_calls[0][1][0] == "Hello"
|
||||||
"""Stop everything that was started."""
|
assert service1.send_message.mock_calls[0][2] == {
|
||||||
self.hass.stop()
|
"title": "Test notification",
|
||||||
|
"data": {"hello": "world"},
|
||||||
def test_send_message_with_data(self):
|
}
|
||||||
"""Test sending a message with to a notify group."""
|
assert service2.send_message.mock_calls[0][1][0] == "Hello"
|
||||||
asyncio.run_coroutine_threadsafe(
|
assert service2.send_message.mock_calls[0][2] == {
|
||||||
self.service.async_send_message(
|
"target": ["unnamed device"],
|
||||||
"Hello", title="Test notification", data={"hello": "world"}
|
"title": "Test notification",
|
||||||
),
|
"data": {"hello": "world", "test": "message"},
|
||||||
self.hass.loop,
|
}
|
||||||
).result()
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
assert self.service1.send_message.mock_calls[0][1][0] == "Hello"
|
|
||||||
assert self.service1.send_message.mock_calls[0][2] == {
|
|
||||||
"title": "Test notification",
|
|
||||||
"data": {"hello": "world"},
|
|
||||||
}
|
|
||||||
assert self.service2.send_message.mock_calls[0][1][0] == "Hello"
|
|
||||||
assert self.service2.send_message.mock_calls[0][2] == {
|
|
||||||
"target": ["unnamed device"],
|
|
||||||
"title": "Test notification",
|
|
||||||
"data": {"hello": "world", "test": "message"},
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
async def test_reload_notify(hass):
|
async def test_reload_notify(hass):
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue