jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix misalignments between sql based filtering with the entityfilter based filtering (#72936)

Browse Source
This commit is contained in:
J. Nick Koston 2022-06-02 17:52:53 -10:00 committed by GitHub
parent f52fa3599f
commit 5b31414225
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 620 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

110
homeassistant/components/recorder/filters.py
View File

@ -88,14 +88,32 @@ class Filters:
self.included_domains: Iterable[str] = []
self.included_entity_globs: Iterable[str] = []
def __repr__(self) -> str:
"""Return human readable excludes/includes."""
return (
f"<Filters excluded_entities={self.excluded_entities} excluded_domains={self.excluded_domains} "
f"excluded_entity_globs={self.excluded_entity_globs} "
f"included_entities={self.included_entities} included_domains={self.included_domains} "
f"included_entity_globs={self.included_entity_globs}>"
)
@property
def has_config(self) -> bool:
"""Determine if there is any filter configuration."""
return bool(self._have_exclude or self._have_include)
@property
def _have_exclude(self) -> bool:
return bool(
self.excluded_entities
or self.excluded_domains
or self.excluded_entity_globs
or self.included_entities
)
@property
def _have_include(self) -> bool:
return bool(
self.included_entities
or self.included_domains
or self.included_entity_globs
)
@ -103,36 +121,67 @@ class Filters:
def _generate_filter_for_columns(
self, columns: Iterable[Column], encoder: Callable[[Any], Any]
) -> ClauseList:
includes = []
if self.included_domains:
includes.append(_domain_matcher(self.included_domains, columns, encoder))
if self.included_entities:
includes.append(_entity_matcher(self.included_entities, columns, encoder))
if self.included_entity_globs:
includes.append(
_globs_to_like(self.included_entity_globs, columns, encoder)
)
"""Generate a filter from pre-comuted sets and pattern lists.
excludes = []
if self.excluded_domains:
excludes.append(_domain_matcher(self.excluded_domains, columns, encoder))
if self.excluded_entities:
excludes.append(_entity_matcher(self.excluded_entities, columns, encoder))
if self.excluded_entity_globs:
excludes.append(
_globs_to_like(self.excluded_entity_globs, columns, encoder)
)
This must match exactly how homeassistant.helpers.entityfilter works.
"""
i_domains = _domain_matcher(self.included_domains, columns, encoder)
i_entities = _entity_matcher(self.included_entities, columns, encoder)
i_entity_globs = _globs_to_like(self.included_entity_globs, columns, encoder)
includes = [i_domains, i_entities, i_entity_globs]
if not includes and not excludes:
e_domains = _domain_matcher(self.excluded_domains, columns, encoder)
e_entities = _entity_matcher(self.excluded_entities, columns, encoder)
e_entity_globs = _globs_to_like(self.excluded_entity_globs, columns, encoder)
excludes = [e_domains, e_entities, e_entity_globs]
have_exclude = self._have_exclude
have_include = self._have_include
# Case 1 - no includes or excludes - pass all entities
if not have_include and not have_exclude:
return None
if includes and not excludes:
# Case 2 - includes, no excludes - only include specified entities
if have_include and not have_exclude:
return or_(*includes).self_group()
if not includes and excludes:
# Case 3 - excludes, no includes - only exclude specified entities
if not have_include and have_exclude:
return not_(or_(*excludes).self_group())
return or_(*includes).self_group() & not_(or_(*excludes).self_group())
# Case 4 - both includes and excludes specified
# Case 4a - include domain or glob specified
# - if domain is included, pass if entity not excluded
# - if glob is included, pass if entity and domain not excluded
# - if domain and glob are not included, pass if entity is included
# note: if both include domain matches then exclude domains ignored.
# If glob matches then exclude domains and glob checked
if self.included_domains or self.included_entity_globs:
return or_(
(i_domains & ~(e_entities | e_entity_globs)),
(
~i_domains
& or_(
(i_entity_globs & ~(or_(*excludes))),
(~i_entity_globs & i_entities),
)
),
).self_group()
# Case 4b - exclude domain or glob specified, include has no domain or glob
# In this one case the traditional include logic is inverted. Even though an
# include is specified since its only a list of entity IDs its used only to
# expose specific entities excluded by domain or glob. Any entities not
# excluded are then presumed included. Logic is as follows
# - if domain or glob is excluded, pass if entity is included
# - if domain is not excluded, pass if entity not excluded by ID
if self.excluded_domains or self.excluded_entity_globs:
return (not_(or_(*excludes)) | i_entities).self_group()
# Case 4c - neither include or exclude domain specified
# - Only pass if entity is included. Ignore entity excludes.
return i_entities
def states_entity_filter(self) -> ClauseList:
"""Generate the entity filter query."""
@ -158,29 +207,32 @@ def _globs_to_like(
glob_strs: Iterable[str], columns: Iterable[Column], encoder: Callable[[Any], Any]
) -> ClauseList:
"""Translate glob to sql."""
return or_(
matchers = [
cast(column, Text()).like(
encoder(glob_str).translate(GLOB_TO_SQL_CHARS), escape="\\"
)
for glob_str in glob_strs
for column in columns
)
]
return or_(*matchers) if matchers else or_(False)
def _entity_matcher(
entity_ids: Iterable[str], columns: Iterable[Column], encoder: Callable[[Any], Any]
) -> ClauseList:
return or_(
matchers = [
cast(column, Text()).in_([encoder(entity_id) for entity_id in entity_ids])
for column in columns
)
]
return or_(*matchers) if matchers else or_(False)
def _domain_matcher(
domains: Iterable[str], columns: Iterable[Column], encoder: Callable[[Any], Any]
) -> ClauseList:
return or_(
matchers = [
cast(column, Text()).like(encoder(f"{domain}.%"))
for domain in domains
for column in columns
)
]
return or_(*matchers) if matchers else or_(False)

6
homeassistant/components/recorder/history.py
View File

@ -237,7 +237,9 @@ def _significant_states_stmt(
stmt += _ignore_domains_filter
if filters and filters.has_config:
entity_filter = filters.states_entity_filter()
stmt += lambda q: q.filter(entity_filter)
stmt = stmt.add_criteria(
lambda q: q.filter(entity_filter), track_on=[filters]
)
stmt += lambda q: q.filter(States.last_updated > start_time)
if end_time:
@ -529,7 +531,7 @@ def _get_states_for_all_stmt(
stmt += _ignore_domains_filter
if filters and filters.has_config:
entity_filter = filters.states_entity_filter()
stmt += lambda q: q.filter(entity_filter)
stmt = stmt.add_criteria(lambda q: q.filter(entity_filter), track_on=[filters])
if join_attributes:
stmt += lambda q: q.outerjoin(
StateAttributes, (States.attributes_id == StateAttributes.attributes_id)

19
tests/components/history/test_init.py
View File

@ -246,15 +246,11 @@ def test_get_significant_states_exclude(hass_history):
def test_get_significant_states_exclude_include_entity(hass_history):
"""Test significant states when excluding domains and include entities.
We should not get back every thermostat and media player test changes.
We should not get back every thermostat change unless its specifically included
"""
hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test2"]
del states["media_player.test3"]
del states["thermostat.test"]
del states["thermostat.test2"]
del states["script.can_cancel_this_one"]
config = history.CONFIG_SCHEMA(
{
@ -340,14 +336,12 @@ def test_get_significant_states_include(hass_history):
def test_get_significant_states_include_exclude_domain(hass_history):
"""Test if significant states when excluding and including domains.
We should not get back any changes since we include only the
media_player domain but also exclude it.
We should get back all the media_player domain changes
only since the include wins over the exclude but will
exclude everything else.
"""
hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"]
del states["media_player.test2"]
del states["media_player.test3"]
del states["thermostat.test"]
del states["thermostat.test2"]
del states["script.can_cancel_this_one"]
@ -372,7 +366,6 @@ def test_get_significant_states_include_exclude_entity(hass_history):
"""
hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"]
del states["media_player.test2"]
del states["media_player.test3"]
del states["thermostat.test"]
@ -394,12 +387,12 @@ def test_get_significant_states_include_exclude_entity(hass_history):
def test_get_significant_states_include_exclude(hass_history):
"""Test if significant states when in/excluding domains and entities.
We should only get back changes of the media_player.test2 entity.
We should get back changes of the media_player.test2, media_player.test3,
and thermostat.test.
"""
hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"]
del states["thermostat.test"]
del states["thermostat.test2"]
del states["script.can_cancel_this_one"]

22
tests/components/logbook/test_init.py
View File

@ -2037,7 +2037,7 @@ async def test_include_events_domain_glob(hass, hass_client, recorder_mock):
_assert_entry(entries[3], name="included", entity_id=entity_id3)
async def test_include_exclude_events(hass, hass_client, recorder_mock):
async def test_include_exclude_events_no_globs(hass, hass_client, recorder_mock):
"""Test if events are filtered if include and exclude is configured."""
entity_id = "switch.bla"
entity_id2 = "sensor.blu"
@ -2082,13 +2082,15 @@ async def test_include_exclude_events(hass, hass_client, recorder_mock):
client = await hass_client()
entries = await _async_fetch_logbook(client)
assert len(entries) == 4
assert len(entries) == 6
_assert_entry(
entries[0], name="Home Assistant", message="started", domain=ha.DOMAIN
)
_assert_entry(entries[1], name="blu", entity_id=entity_id2, state="10")
_assert_entry(entries[2], name="blu", entity_id=entity_id2, state="20")
_assert_entry(entries[3], name="keep", entity_id=entity_id4, state="10")
_assert_entry(entries[1], name="bla", entity_id=entity_id, state="10")
_assert_entry(entries[2], name="blu", entity_id=entity_id2, state="10")
_assert_entry(entries[3], name="bla", entity_id=entity_id, state="20")
_assert_entry(entries[4], name="blu", entity_id=entity_id2, state="20")
_assert_entry(entries[5], name="keep", entity_id=entity_id4, state="10")
async def test_include_exclude_events_with_glob_filters(
@ -2145,13 +2147,15 @@ async def test_include_exclude_events_with_glob_filters(
client = await hass_client()
entries = await _async_fetch_logbook(client)
assert len(entries) == 4
assert len(entries) == 6
_assert_entry(
entries[0], name="Home Assistant", message="started", domain=ha.DOMAIN
)
_assert_entry(entries[1], name="blu", entity_id=entity_id2, state="10")
_assert_entry(entries[2], name="blu", entity_id=entity_id2, state="20")
_assert_entry(entries[3], name="included", entity_id=entity_id4, state="30")
_assert_entry(entries[1], name="bla", entity_id=entity_id, state="10")
_assert_entry(entries[2], name="blu", entity_id=entity_id2, state="10")
_assert_entry(entries[3], name="bla", entity_id=entity_id, state="20")
_assert_entry(entries[4], name="blu", entity_id=entity_id2, state="20")
_assert_entry(entries[5], name="included", entity_id=entity_id4, state="30")
async def test_empty_config(hass, hass_client, recorder_mock):

516
tests/components/recorder/test_filters_with_entityfilter.py Normal file
View File

@ -0,0 +1,516 @@
"""The tests for the recorder filter matching the EntityFilter component."""
import json
from sqlalchemy import select
from sqlalchemy.engine.row import Row
from homeassistant.components.recorder import get_instance
from homeassistant.components.recorder.filters import (
Filters,
extract_include_exclude_filter_conf,
sqlalchemy_filter_from_include_exclude_conf,
)
from homeassistant.components.recorder.models import EventData, States
from homeassistant.components.recorder.util import session_scope
from homeassistant.const import ATTR_ENTITY_ID, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entityfilter import (
CONF_DOMAINS,
CONF_ENTITIES,
CONF_ENTITY_GLOBS,
CONF_EXCLUDE,
CONF_INCLUDE,
convert_include_exclude_filter,
)
from .common import async_wait_recording_done
async def _async_get_states_and_events_with_filter(
hass: HomeAssistant, sqlalchemy_filter: Filters, entity_ids: set[str]
) -> tuple[list[Row], list[Row]]:
"""Get states from the database based on a filter."""
for entity_id in entity_ids:
hass.states.async_set(entity_id, STATE_ON)
hass.bus.async_fire("any", {ATTR_ENTITY_ID: entity_id})
await async_wait_recording_done(hass)
def _get_states_with_session():
with session_scope(hass=hass) as session:
return session.execute(
select(States.entity_id).filter(
sqlalchemy_filter.states_entity_filter()
)
).all()
filtered_states_entity_ids = {
row[0]
for row in await get_instance(hass).async_add_executor_job(
_get_states_with_session
)
}
def _get_events_with_session():
with session_scope(hass=hass) as session:
return session.execute(
select(EventData.shared_data).filter(
sqlalchemy_filter.events_entity_filter()
)
).all()
filtered_events_entity_ids = set()
for row in await get_instance(hass).async_add_executor_job(
_get_events_with_session
):
event_data = json.loads(row[0])
if ATTR_ENTITY_ID not in event_data:
continue
filtered_events_entity_ids.add(json.loads(row[0])[ATTR_ENTITY_ID])
return filtered_states_entity_ids, filtered_events_entity_ids
async def test_included_and_excluded_simple_case_no_domains(hass, recorder_mock):
"""Test filters with included and excluded without domains."""
filter_accept = {"sensor.kitchen4", "switch.kitchen"}
filter_reject = {
"light.any",
"switch.other",
"cover.any",
"sensor.weather5",
"light.kitchen",
}
conf = {
CONF_INCLUDE: {
CONF_ENTITY_GLOBS: ["sensor.kitchen*"],
CONF_ENTITIES: ["switch.kitchen"],
},
CONF_EXCLUDE: {
CONF_ENTITY_GLOBS: ["sensor.weather*"],
CONF_ENTITIES: ["light.kitchen"],
},
}
extracted_filter = extract_include_exclude_filter_conf(conf)
entity_filter = convert_include_exclude_filter(extracted_filter)
sqlalchemy_filter = sqlalchemy_filter_from_include_exclude_conf(extracted_filter)
assert sqlalchemy_filter is not None
for entity_id in filter_accept:
assert entity_filter(entity_id) is True
for entity_id in filter_reject:
assert entity_filter(entity_id) is False
assert not entity_filter.explicitly_included("light.any")
assert not entity_filter.explicitly_included("switch.other")
assert entity_filter.explicitly_included("sensor.kitchen4")
assert entity_filter.explicitly_included("switch.kitchen")
assert not entity_filter.explicitly_excluded("light.any")
assert not entity_filter.explicitly_excluded("switch.other")
assert entity_filter.explicitly_excluded("sensor.weather5")
assert entity_filter.explicitly_excluded("light.kitchen")
(
filtered_states_entity_ids,
filtered_events_entity_ids,
) = await _async_get_states_and_events_with_filter(
hass, sqlalchemy_filter, filter_accept | filter_reject
)
assert filtered_states_entity_ids == filter_accept
assert not filtered_states_entity_ids.intersection(filter_reject)
assert filtered_events_entity_ids == filter_accept
assert not filtered_events_entity_ids.intersection(filter_reject)
async def test_included_and_excluded_simple_case_no_globs(hass, recorder_mock):
"""Test filters with included and excluded without globs."""
filter_accept = {"switch.bla", "sensor.blu", "sensor.keep"}
filter_reject = {"sensor.bli"}
conf = {
CONF_INCLUDE: {
CONF_DOMAINS: ["sensor", "homeassistant"],
CONF_ENTITIES: ["switch.bla"],
},
CONF_EXCLUDE: {
CONF_DOMAINS: ["switch"],
CONF_ENTITIES: ["sensor.bli"],
},
}
extracted_filter = extract_include_exclude_filter_conf(conf)
entity_filter = convert_include_exclude_filter(extracted_filter)
sqlalchemy_filter = sqlalchemy_filter_from_include_exclude_conf(extracted_filter)
assert sqlalchemy_filter is not None
for entity_id in filter_accept:
assert entity_filter(entity_id) is True
for entity_id in filter_reject:
assert entity_filter(entity_id) is False
(
filtered_states_entity_ids,
filtered_events_entity_ids,
) = await _async_get_states_and_events_with_filter(
hass, sqlalchemy_filter, filter_accept | filter_reject
)
assert filtered_states_entity_ids == filter_accept
assert not filtered_states_entity_ids.intersection(filter_reject)
assert filtered_events_entity_ids == filter_accept
assert not filtered_events_entity_ids.intersection(filter_reject)
async def test_included_and_excluded_simple_case_without_underscores(
hass, recorder_mock
):
"""Test filters with included and excluded without underscores."""
filter_accept = {"light.any", "sensor.kitchen4", "switch.kitchen"}
filter_reject = {"switch.other", "cover.any", "sensor.weather5", "light.kitchen"}
conf = {
CONF_INCLUDE: {
CONF_DOMAINS: ["light"],
CONF_ENTITY_GLOBS: ["sensor.kitchen*"],
CONF_ENTITIES: ["switch.kitchen"],
},
CONF_EXCLUDE: {
CONF_DOMAINS: ["cover"],
CONF_ENTITY_GLOBS: ["sensor.weather*"],
CONF_ENTITIES: ["light.kitchen"],
},
}
extracted_filter = extract_include_exclude_filter_conf(conf)
entity_filter = convert_include_exclude_filter(extracted_filter)
sqlalchemy_filter = sqlalchemy_filter_from_include_exclude_conf(extracted_filter)
assert sqlalchemy_filter is not None
for entity_id in filter_accept:
assert entity_filter(entity_id) is True
for entity_id in filter_reject:
assert entity_filter(entity_id) is False
assert not entity_filter.explicitly_included("light.any")
assert not entity_filter.explicitly_included("switch.other")
assert entity_filter.explicitly_included("sensor.kitchen4")
assert entity_filter.explicitly_included("switch.kitchen")
assert not entity_filter.explicitly_excluded("light.any")
assert not entity_filter.explicitly_excluded("switch.other")
assert entity_filter.explicitly_excluded("sensor.weather5")
assert entity_filter.explicitly_excluded("light.kitchen")
(
filtered_states_entity_ids,
filtered_events_entity_ids,
) = await _async_get_states_and_events_with_filter(
hass, sqlalchemy_filter, filter_accept | filter_reject
)
assert filtered_states_entity_ids == filter_accept
assert not filtered_states_entity_ids.intersection(filter_reject)
assert filtered_events_entity_ids == filter_accept
assert not filtered_events_entity_ids.intersection(filter_reject)
async def test_included_and_excluded_simple_case_with_underscores(hass, recorder_mock):
"""Test filters with included and excluded with underscores."""
filter_accept = {"light.any", "sensor.kitchen_4", "switch.kitchen"}
filter_reject = {"switch.other", "cover.any", "sensor.weather_5", "light.kitchen"}
conf = {
CONF_INCLUDE: {
CONF_DOMAINS: ["light"],
CONF_ENTITY_GLOBS: ["sensor.kitchen_*"],
CONF_ENTITIES: ["switch.kitchen"],
},
CONF_EXCLUDE: {
CONF_DOMAINS: ["cover"],
CONF_ENTITY_GLOBS: ["sensor.weather_*"],
CONF_ENTITIES: ["light.kitchen"],
},
}
extracted_filter = extract_include_exclude_filter_conf(conf)
entity_filter = convert_include_exclude_filter(extracted_filter)
sqlalchemy_filter = sqlalchemy_filter_from_include_exclude_conf(extracted_filter)
assert sqlalchemy_filter is not None
for entity_id in filter_accept:
assert entity_filter(entity_id) is True
for entity_id in filter_reject:
assert entity_filter(entity_id) is False
assert not entity_filter.explicitly_included("light.any")
assert not entity_filter.explicitly_included("switch.other")
assert entity_filter.explicitly_included("sensor.kitchen_4")
assert entity_filter.explicitly_included("switch.kitchen")
assert not entity_filter.explicitly_excluded("light.any")
assert not entity_filter.explicitly_excluded("switch.other")
assert entity_filter.explicitly_excluded("sensor.weather_5")
assert entity_filter.explicitly_excluded("light.kitchen")
(
filtered_states_entity_ids,
filtered_events_entity_ids,
) = await _async_get_states_and_events_with_filter(
hass, sqlalchemy_filter, filter_accept | filter_reject
)
assert filtered_states_entity_ids == filter_accept
assert not filtered_states_entity_ids.intersection(filter_reject)
assert filtered_events_entity_ids == filter_accept
assert not filtered_events_entity_ids.intersection(filter_reject)
async def test_included_and_excluded_complex_case(hass, recorder_mock):
"""Test filters with included and excluded with a complex filter."""
filter_accept = {"light.any", "sensor.kitchen_4", "switch.kitchen"}
filter_reject = {
"camera.one",
"notify.any",
"automation.update_readme",
"automation.update_utilities_cost",
"binary_sensor.iss",
}
conf = {
CONF_INCLUDE: {
CONF_ENTITIES: ["group.trackers"],
},
CONF_EXCLUDE: {
CONF_ENTITIES: [
"automation.update_readme",
"automation.update_utilities_cost",
"binary_sensor.iss",
],
CONF_DOMAINS: [
"camera",
"group",
"media_player",
"notify",
"scene",
"sun",
"zone",
],
},
}
extracted_filter = extract_include_exclude_filter_conf(conf)
entity_filter = convert_include_exclude_filter(extracted_filter)
sqlalchemy_filter = sqlalchemy_filter_from_include_exclude_conf(extracted_filter)
assert sqlalchemy_filter is not None
for entity_id in filter_accept:
assert entity_filter(entity_id) is True
for entity_id in filter_reject:
assert entity_filter(entity_id) is False
(
filtered_states_entity_ids,
filtered_events_entity_ids,
) = await _async_get_states_and_events_with_filter(
hass, sqlalchemy_filter, filter_accept | filter_reject
)
assert filtered_states_entity_ids == filter_accept
assert not filtered_states_entity_ids.intersection(filter_reject)
assert filtered_events_entity_ids == filter_accept
assert not filtered_events_entity_ids.intersection(filter_reject)
async def test_included_entities_and_excluded_domain(hass, recorder_mock):
"""Test filters with included entities and excluded domain."""
filter_accept = {
"media_player.test",
"media_player.test3",
"thermostat.test",
"zone.home",
"script.can_cancel_this_one",
}
filter_reject = {
"thermostat.test2",
}
conf = {
CONF_INCLUDE: {
CONF_ENTITIES: ["media_player.test", "thermostat.test"],
},
CONF_EXCLUDE: {
CONF_DOMAINS: ["thermostat"],
},
}
extracted_filter = extract_include_exclude_filter_conf(conf)
entity_filter = convert_include_exclude_filter(extracted_filter)
sqlalchemy_filter = sqlalchemy_filter_from_include_exclude_conf(extracted_filter)
assert sqlalchemy_filter is not None
for entity_id in filter_accept:
assert entity_filter(entity_id) is True
for entity_id in filter_reject:
assert entity_filter(entity_id) is False
(
filtered_states_entity_ids,
filtered_events_entity_ids,
) = await _async_get_states_and_events_with_filter(
hass, sqlalchemy_filter, filter_accept | filter_reject
)
assert filtered_states_entity_ids == filter_accept
assert not filtered_states_entity_ids.intersection(filter_reject)
assert filtered_events_entity_ids == filter_accept
assert not filtered_events_entity_ids.intersection(filter_reject)
async def test_same_domain_included_excluded(hass, recorder_mock):
"""Test filters with the same domain included and excluded."""
filter_accept = {
"media_player.test",
"media_player.test3",
}
filter_reject = {
"thermostat.test2",
"thermostat.test",
"zone.home",
"script.can_cancel_this_one",
}
conf = {
CONF_INCLUDE: {
CONF_DOMAINS: ["media_player"],
},
CONF_EXCLUDE: {
CONF_DOMAINS: ["media_player"],
},
}
extracted_filter = extract_include_exclude_filter_conf(conf)
entity_filter = convert_include_exclude_filter(extracted_filter)
sqlalchemy_filter = sqlalchemy_filter_from_include_exclude_conf(extracted_filter)
assert sqlalchemy_filter is not None
for entity_id in filter_accept:
assert entity_filter(entity_id) is True
for entity_id in filter_reject:
assert entity_filter(entity_id) is False
(
filtered_states_entity_ids,
filtered_events_entity_ids,
) = await _async_get_states_and_events_with_filter(
hass, sqlalchemy_filter, filter_accept | filter_reject
)
assert filtered_states_entity_ids == filter_accept
assert not filtered_states_entity_ids.intersection(filter_reject)
assert filtered_events_entity_ids == filter_accept
assert not filtered_events_entity_ids.intersection(filter_reject)
async def test_same_entity_included_excluded(hass, recorder_mock):
"""Test filters with the same entity included and excluded."""
filter_accept = {
"media_player.test",
}
filter_reject = {
"media_player.test3",
"thermostat.test2",
"thermostat.test",
"zone.home",
"script.can_cancel_this_one",
}
conf = {
CONF_INCLUDE: {
CONF_ENTITIES: ["media_player.test"],
},
CONF_EXCLUDE: {
CONF_ENTITIES: ["media_player.test"],
},
}
extracted_filter = extract_include_exclude_filter_conf(conf)
entity_filter = convert_include_exclude_filter(extracted_filter)
sqlalchemy_filter = sqlalchemy_filter_from_include_exclude_conf(extracted_filter)
assert sqlalchemy_filter is not None
for entity_id in filter_accept:
assert entity_filter(entity_id) is True
for entity_id in filter_reject:
assert entity_filter(entity_id) is False
(
filtered_states_entity_ids,
filtered_events_entity_ids,
) = await _async_get_states_and_events_with_filter(
hass, sqlalchemy_filter, filter_accept | filter_reject
)
assert filtered_states_entity_ids == filter_accept
assert not filtered_states_entity_ids.intersection(filter_reject)
assert filtered_events_entity_ids == filter_accept
assert not filtered_events_entity_ids.intersection(filter_reject)
async def test_same_entity_included_excluded_include_domain_wins(hass, recorder_mock):
"""Test filters with domain and entities and the include domain wins."""
filter_accept = {
"media_player.test2",
"media_player.test3",
"thermostat.test",
}
filter_reject = {
"thermostat.test2",
"zone.home",
"script.can_cancel_this_one",
}
conf = {
CONF_INCLUDE: {
CONF_DOMAINS: ["media_player"],
CONF_ENTITIES: ["thermostat.test"],
},
CONF_EXCLUDE: {
CONF_DOMAINS: ["thermostat"],
CONF_ENTITIES: ["media_player.test"],
},
}
extracted_filter = extract_include_exclude_filter_conf(conf)
entity_filter = convert_include_exclude_filter(extracted_filter)
sqlalchemy_filter = sqlalchemy_filter_from_include_exclude_conf(extracted_filter)
assert sqlalchemy_filter is not None
for entity_id in filter_accept:
assert entity_filter(entity_id) is True
for entity_id in filter_reject:
assert entity_filter(entity_id) is False
(
filtered_states_entity_ids,
filtered_events_entity_ids,
) = await _async_get_states_and_events_with_filter(
hass, sqlalchemy_filter, filter_accept | filter_reject
)
assert filtered_states_entity_ids == filter_accept
assert not filtered_states_entity_ids.intersection(filter_reject)
assert filtered_events_entity_ids == filter_accept
assert not filtered_events_entity_ids.intersection(filter_reject)