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

Switch history tests to pytest (#42318)

Browse Source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Matthew Donoughe 2021-03-17 10:32:44 -04:00 committed by GitHub
parent 6a24ec7a30
commit 9011a54e7f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 711 additions and 699 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
tests/components/history/conftest.py Normal file
View File

@ -0,0 +1,49 @@
"""Fixtures for history tests."""
import pytest
from homeassistant.components import history
from homeassistant.components.recorder.const import DATA_INSTANCE
from homeassistant.setup import setup_component
from tests.common import get_test_home_assistant, init_recorder_component
@pytest.fixture
def hass_recorder():
"""Home Assistant fixture with in-memory recorder."""
hass = get_test_home_assistant()
def setup_recorder(config=None):
"""Set up with params."""
init_recorder_component(hass, config)
hass.start()
hass.block_till_done()
hass.data[DATA_INSTANCE].block_till_done()
return hass
yield setup_recorder
hass.stop()
@pytest.fixture
def hass_history(hass_recorder):
"""Home Assistant fixture with history."""
hass = hass_recorder()
config = history.CONFIG_SCHEMA(
{
history.DOMAIN: {
history.CONF_INCLUDE: {
history.CONF_DOMAINS: ["media_player"],
history.CONF_ENTITIES: ["thermostat.test"],
},
history.CONF_EXCLUDE: {
history.CONF_DOMAINS: ["thermostat"],
history.CONF_ENTITIES: ["media_player.test"],
},
}
}
)
assert setup_component(hass, history.DOMAIN, config)
yield hass

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

@ -3,71 +3,35 @@
from copy import copy from copy import copy
from datetime import timedelta from datetime import timedelta
import json import json
import unittest
from unittest.mock import patch, sentinel from unittest.mock import patch, sentinel
import pytest
from homeassistant.components import history, recorder from homeassistant.components import history, recorder
from homeassistant.components.recorder.models import process_timestamp from homeassistant.components.recorder.models import process_timestamp
import homeassistant.core as ha import homeassistant.core as ha
from homeassistant.helpers.json import JSONEncoder from homeassistant.helpers.json import JSONEncoder
from homeassistant.setup import async_setup_component, setup_component from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from tests.common import ( from tests.common import init_recorder_component, mock_state_change_event
get_test_home_assistant,
init_recorder_component,
mock_state_change_event,
)
from tests.components.recorder.common import trigger_db_commit, wait_recording_done from tests.components.recorder.common import trigger_db_commit, wait_recording_done
class TestComponentHistory(unittest.TestCase): @pytest.mark.usefixtures("hass_history")
"""Test History component.""" def test_setup():
def setUp(self): # pylint: disable=invalid-name
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.addCleanup(self.tear_down_cleanup)
def tear_down_cleanup(self):
"""Stop everything that was started."""
self.hass.stop()
def init_recorder(self):
"""Initialize the recorder."""
init_recorder_component(self.hass)
self.hass.start()
wait_recording_done(self.hass)
def test_setup(self):
"""Test setup method of history.""" """Test setup method of history."""
config = history.CONFIG_SCHEMA( # Verification occurs in the fixture
{ pass
# ha.DOMAIN: {},
history.DOMAIN: {
history.CONF_INCLUDE: {
history.CONF_DOMAINS: ["media_player"],
history.CONF_ENTITIES: ["thermostat.test"],
},
history.CONF_EXCLUDE: {
history.CONF_DOMAINS: ["thermostat"],
history.CONF_ENTITIES: ["media_player.test"],
},
}
}
)
self.init_recorder()
assert setup_component(self.hass, history.DOMAIN, config)
def test_get_states(self):
def test_get_states(hass_history):
"""Test getting states at a specific point in time.""" """Test getting states at a specific point in time."""
self.test_setup() hass = hass_history
states = [] states = []
now = dt_util.utcnow() now = dt_util.utcnow()
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=now):
"homeassistant.components.recorder.dt_util.utcnow", return_value=now
):
for i in range(5): for i in range(5):
state = ha.State( state = ha.State(
"test.point_in_time_{}".format(i % 5), "test.point_in_time_{}".format(i % 5),
@ -75,16 +39,14 @@ class TestComponentHistory(unittest.TestCase):
{"attribute_test": i}, {"attribute_test": i},
) )
mock_state_change_event(self.hass, state) mock_state_change_event(hass, state)
states.append(state) states.append(state)
wait_recording_done(self.hass) wait_recording_done(hass)
future = now + timedelta(seconds=1) future = now + timedelta(seconds=1)
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=future):
"homeassistant.components.recorder.dt_util.utcnow", return_value=future
):
for i in range(5): for i in range(5):
state = ha.State( state = ha.State(
"test.point_in_time_{}".format(i % 5), "test.point_in_time_{}".format(i % 5),
@ -92,51 +54,46 @@ class TestComponentHistory(unittest.TestCase):
{"attribute_test": i}, {"attribute_test": i},
) )
mock_state_change_event(self.hass, state) mock_state_change_event(hass, state)
wait_recording_done(self.hass) wait_recording_done(hass)
# Get states returns everything before POINT # Get states returns everything before POINT
for state1, state2 in zip( for state1, state2 in zip(
states, states,
sorted( sorted(history.get_states(hass, future), key=lambda state: state.entity_id),
history.get_states(self.hass, future), key=lambda state: state.entity_id
),
): ):
assert state1 == state2 assert state1 == state2
# Test get_state here because we have a DB setup # Test get_state here because we have a DB setup
assert states[0] == history.get_state(self.hass, future, states[0].entity_id) assert states[0] == history.get_state(hass, future, states[0].entity_id)
time_before_recorder_ran = now - timedelta(days=1000) time_before_recorder_ran = now - timedelta(days=1000)
assert history.get_states(self.hass, time_before_recorder_ran) == [] assert history.get_states(hass, time_before_recorder_ran) == []
assert history.get_state(self.hass, time_before_recorder_ran, "demo.id") is None assert history.get_state(hass, time_before_recorder_ran, "demo.id") is None
def test_state_changes_during_period(self):
def test_state_changes_during_period(hass_history):
"""Test state change during period.""" """Test state change during period."""
self.test_setup() hass = hass_history
entity_id = "media_player.test" entity_id = "media_player.test"
def set_state(state): def set_state(state):
"""Set the state.""" """Set the state."""
self.hass.states.set(entity_id, state) hass.states.set(entity_id, state)
wait_recording_done(self.hass) wait_recording_done(hass)
return self.hass.states.get(entity_id) return hass.states.get(entity_id)
start = dt_util.utcnow() start = dt_util.utcnow()
point = start + timedelta(seconds=1) point = start + timedelta(seconds=1)
end = point + timedelta(seconds=1) end = point + timedelta(seconds=1)
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=start):
"homeassistant.components.recorder.dt_util.utcnow", return_value=start
):
set_state("idle") set_state("idle")
set_state("YouTube") set_state("YouTube")
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=point):
"homeassistant.components.recorder.dt_util.utcnow", return_value=point
):
states = [ states = [
set_state("idle"), set_state("idle"),
set_state("Netflix"), set_state("Netflix"),
@ -144,98 +101,89 @@ class TestComponentHistory(unittest.TestCase):
set_state("YouTube"), set_state("YouTube"),
] ]
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=end):
"homeassistant.components.recorder.dt_util.utcnow", return_value=end
):
set_state("Netflix") set_state("Netflix")
set_state("Plex") set_state("Plex")
hist = history.state_changes_during_period(self.hass, start, end, entity_id) hist = history.state_changes_during_period(hass, start, end, entity_id)
assert states == hist[entity_id] assert states == hist[entity_id]
def test_get_last_state_changes(self):
def test_get_last_state_changes(hass_history):
"""Test number of state changes.""" """Test number of state changes."""
self.test_setup() hass = hass_history
entity_id = "sensor.test" entity_id = "sensor.test"
def set_state(state): def set_state(state):
"""Set the state.""" """Set the state."""
self.hass.states.set(entity_id, state) hass.states.set(entity_id, state)
wait_recording_done(self.hass) wait_recording_done(hass)
return self.hass.states.get(entity_id) return hass.states.get(entity_id)
start = dt_util.utcnow() - timedelta(minutes=2) start = dt_util.utcnow() - timedelta(minutes=2)
point = start + timedelta(minutes=1) point = start + timedelta(minutes=1)
point2 = point + timedelta(minutes=1) point2 = point + timedelta(minutes=1)
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=start):
"homeassistant.components.recorder.dt_util.utcnow", return_value=start
):
set_state("1") set_state("1")
states = [] states = []
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=point):
"homeassistant.components.recorder.dt_util.utcnow", return_value=point
):
states.append(set_state("2")) states.append(set_state("2"))
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=point2):
"homeassistant.components.recorder.dt_util.utcnow", return_value=point2
):
states.append(set_state("3")) states.append(set_state("3"))
hist = history.get_last_state_changes(self.hass, 2, entity_id) hist = history.get_last_state_changes(hass, 2, entity_id)
assert states == hist[entity_id] assert states == hist[entity_id]
def test_ensure_state_can_be_copied(self):
def test_ensure_state_can_be_copied(hass_history):
"""Ensure a state can pass though copy(). """Ensure a state can pass though copy().
The filter integration uses copy() on states The filter integration uses copy() on states
from history. from history.
""" """
self.test_setup() hass = hass_history
entity_id = "sensor.test" entity_id = "sensor.test"
def set_state(state): def set_state(state):
"""Set the state.""" """Set the state."""
self.hass.states.set(entity_id, state) hass.states.set(entity_id, state)
wait_recording_done(self.hass) wait_recording_done(hass)
return self.hass.states.get(entity_id) return hass.states.get(entity_id)
start = dt_util.utcnow() - timedelta(minutes=2) start = dt_util.utcnow() - timedelta(minutes=2)
point = start + timedelta(minutes=1) point = start + timedelta(minutes=1)
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=start):
"homeassistant.components.recorder.dt_util.utcnow", return_value=start
):
set_state("1") set_state("1")
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=point):
"homeassistant.components.recorder.dt_util.utcnow", return_value=point
):
set_state("2") set_state("2")
hist = history.get_last_state_changes(self.hass, 2, entity_id) hist = history.get_last_state_changes(hass, 2, entity_id)
assert copy(hist[entity_id][0]) == hist[entity_id][0] assert copy(hist[entity_id][0]) == hist[entity_id][0]
assert copy(hist[entity_id][1]) == hist[entity_id][1] assert copy(hist[entity_id][1]) == hist[entity_id][1]
def test_get_significant_states(self):
def test_get_significant_states(hass_history):
"""Test that only significant states are returned. """Test that only significant states are returned.
We should get back every thermostat change that We should get back every thermostat change that
includes an attribute change, but only the state updates for includes an attribute change, but only the state updates for
media player (attribute changes are not significant and not returned). media player (attribute changes are not significant and not returned).
""" """
zero, four, states = self.record_states() hass = hass_history
hist = history.get_significant_states( zero, four, states = record_states(hass)
self.hass, zero, four, filters=history.Filters() hist = history.get_significant_states(hass, zero, four, filters=history.Filters())
)
assert states == hist assert states == hist
def test_get_significant_states_minimal_response(self):
def test_get_significant_states_minimal_response(hass_history):
"""Test that only significant states are returned. """Test that only significant states are returned.
When minimal responses is set only the first and When minimal responses is set only the first and
@ -245,9 +193,10 @@ class TestComponentHistory(unittest.TestCase):
includes an attribute change, but only the state updates for includes an attribute change, but only the state updates for
media player (attribute changes are not significant and not returned). media player (attribute changes are not significant and not returned).
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
hist = history.get_significant_states( hist = history.get_significant_states(
self.hass, zero, four, filters=history.Filters(), minimal_response=True hass, zero, four, filters=history.Filters(), minimal_response=True
) )
# The second media_player.test state is reduced # The second media_player.test state is reduced
@ -268,14 +217,16 @@ class TestComponentHistory(unittest.TestCase):
assert states == hist assert states == hist
def test_get_significant_states_with_initial(self):
def test_get_significant_states_with_initial(hass_history):
"""Test that only significant states are returned. """Test that only significant states are returned.
We should get back every thermostat change that We should get back every thermostat change that
includes an attribute change, but only the state updates for includes an attribute change, but only the state updates for
media player (attribute changes are not significant and not returned). media player (attribute changes are not significant and not returned).
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
one = zero + timedelta(seconds=1) one = zero + timedelta(seconds=1)
one_and_half = zero + timedelta(seconds=1.5) one_and_half = zero + timedelta(seconds=1.5)
for entity_id in states: for entity_id in states:
@ -286,7 +237,7 @@ class TestComponentHistory(unittest.TestCase):
state.last_changed = one_and_half state.last_changed = one_and_half
hist = history.get_significant_states( hist = history.get_significant_states(
self.hass, hass,
one_and_half, one_and_half,
four, four,
filters=history.Filters(), filters=history.Filters(),
@ -294,14 +245,16 @@ class TestComponentHistory(unittest.TestCase):
) )
assert states == hist assert states == hist
def test_get_significant_states_without_initial(self):
def test_get_significant_states_without_initial(hass_history):
"""Test that only significant states are returned. """Test that only significant states are returned.
We should get back every thermostat change that We should get back every thermostat change that
includes an attribute change, but only the state updates for includes an attribute change, but only the state updates for
media player (attribute changes are not significant and not returned). media player (attribute changes are not significant and not returned).
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
one = zero + timedelta(seconds=1) one = zero + timedelta(seconds=1)
one_and_half = zero + timedelta(seconds=1.5) one_and_half = zero + timedelta(seconds=1.5)
for entity_id in states: for entity_id in states:
@ -311,7 +264,7 @@ class TestComponentHistory(unittest.TestCase):
del states["media_player.test2"] del states["media_player.test2"]
hist = history.get_significant_states( hist = history.get_significant_states(
self.hass, hass,
one_and_half, one_and_half,
four, four,
filters=history.Filters(), filters=history.Filters(),
@ -319,9 +272,11 @@ class TestComponentHistory(unittest.TestCase):
) )
assert states == hist assert states == hist
def test_get_significant_states_entity_id(self):
def test_get_significant_states_entity_id(hass_history):
"""Test that only significant states are returned for one entity.""" """Test that only significant states are returned for one entity."""
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test2"] del states["media_player.test2"]
del states["media_player.test3"] del states["media_player.test3"]
del states["thermostat.test"] del states["thermostat.test"]
@ -329,20 +284,22 @@ class TestComponentHistory(unittest.TestCase):
del states["script.can_cancel_this_one"] del states["script.can_cancel_this_one"]
hist = history.get_significant_states( hist = history.get_significant_states(
self.hass, zero, four, ["media_player.test"], filters=history.Filters() hass, zero, four, ["media_player.test"], filters=history.Filters()
) )
assert states == hist assert states == hist
def test_get_significant_states_multiple_entity_ids(self):
def test_get_significant_states_multiple_entity_ids(hass_history):
"""Test that only significant states are returned for one entity.""" """Test that only significant states are returned for one entity."""
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test2"] del states["media_player.test2"]
del states["media_player.test3"] del states["media_player.test3"]
del states["thermostat.test2"] del states["thermostat.test2"]
del states["script.can_cancel_this_one"] del states["script.can_cancel_this_one"]
hist = history.get_significant_states( hist = history.get_significant_states(
self.hass, hass,
zero, zero,
four, four,
["media_player.test", "thermostat.test"], ["media_player.test", "thermostat.test"],
@ -350,13 +307,15 @@ class TestComponentHistory(unittest.TestCase):
) )
assert states == hist assert states == hist
def test_get_significant_states_exclude_domain(self):
def test_get_significant_states_exclude_domain(hass_history):
"""Test if significant states are returned when excluding domains. """Test if significant states are returned when excluding domains.
We should get back every thermostat change that includes an attribute We should get back every thermostat change that includes an attribute
change, but no media player changes. change, but no media player changes.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"] del states["media_player.test"]
del states["media_player.test2"] del states["media_player.test2"]
del states["media_player.test3"] del states["media_player.test3"]
@ -369,15 +328,17 @@ class TestComponentHistory(unittest.TestCase):
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_exclude_entity(self):
def test_get_significant_states_exclude_entity(hass_history):
"""Test if significant states are returned when excluding entities. """Test if significant states are returned when excluding entities.
We should get back every thermostat and script changes, but no media We should get back every thermostat and script changes, but no media
player changes. player changes.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"] del states["media_player.test"]
config = history.CONFIG_SCHEMA( config = history.CONFIG_SCHEMA(
@ -388,14 +349,16 @@ class TestComponentHistory(unittest.TestCase):
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_exclude(self):
def test_get_significant_states_exclude(hass_history):
"""Test significant states when excluding entities and domains. """Test significant states when excluding entities and domains.
We should not get back every thermostat and media player test changes. We should not get back every thermostat and media player test changes.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"] del states["media_player.test"]
del states["thermostat.test"] del states["thermostat.test"]
del states["thermostat.test2"] del states["thermostat.test2"]
@ -411,14 +374,16 @@ class TestComponentHistory(unittest.TestCase):
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_exclude_include_entity(self):
def test_get_significant_states_exclude_include_entity(hass_history):
"""Test significant states when excluding domains and include entities. """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 and media player test changes.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test2"] del states["media_player.test2"]
del states["media_player.test3"] del states["media_player.test3"]
del states["thermostat.test"] del states["thermostat.test"]
@ -436,15 +401,17 @@ class TestComponentHistory(unittest.TestCase):
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_include_domain(self):
def test_get_significant_states_include_domain(hass_history):
"""Test if significant states are returned when including domains. """Test if significant states are returned when including domains.
We should get back every thermostat and script changes, but no media We should get back every thermostat and script changes, but no media
player changes. player changes.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"] del states["media_player.test"]
del states["media_player.test2"] del states["media_player.test2"]
del states["media_player.test3"] del states["media_player.test3"]
@ -453,20 +420,20 @@ class TestComponentHistory(unittest.TestCase):
{ {
ha.DOMAIN: {}, ha.DOMAIN: {},
history.DOMAIN: { history.DOMAIN: {
history.CONF_INCLUDE: { history.CONF_INCLUDE: {history.CONF_DOMAINS: ["thermostat", "script"]}
history.CONF_DOMAINS: ["thermostat", "script"]
}
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_include_entity(self):
def test_get_significant_states_include_entity(hass_history):
"""Test if significant states are returned when including entities. """Test if significant states are returned when including entities.
We should only get back changes of the media_player.test entity. We should only get back changes of the media_player.test entity.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test2"] del states["media_player.test2"]
del states["media_player.test3"] del states["media_player.test3"]
del states["thermostat.test"] del states["thermostat.test"]
@ -481,15 +448,17 @@ class TestComponentHistory(unittest.TestCase):
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_include(self):
def test_get_significant_states_include(hass_history):
"""Test significant states when including domains and entities. """Test significant states when including domains and entities.
We should only get back changes of the media_player.test entity and the We should only get back changes of the media_player.test entity and the
thermostat domain. thermostat domain.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test2"] del states["media_player.test2"]
del states["media_player.test3"] del states["media_player.test3"]
del states["script.can_cancel_this_one"] del states["script.can_cancel_this_one"]
@ -505,15 +474,17 @@ class TestComponentHistory(unittest.TestCase):
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_include_exclude_domain(self):
def test_get_significant_states_include_exclude_domain(hass_history):
"""Test if significant states when excluding and including domains. """Test if significant states when excluding and including domains.
We should not get back any changes since we include only the We should not get back any changes since we include only the
media_player domain but also exclude it. media_player domain but also exclude it.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"] del states["media_player.test"]
del states["media_player.test2"] del states["media_player.test2"]
del states["media_player.test3"] del states["media_player.test3"]
@ -530,15 +501,17 @@ class TestComponentHistory(unittest.TestCase):
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_include_exclude_entity(self):
def test_get_significant_states_include_exclude_entity(hass_history):
"""Test if significant states when excluding and including domains. """Test if significant states when excluding and including domains.
We should not get back any changes since we include only We should not get back any changes since we include only
media_player.test but also exclude it. media_player.test but also exclude it.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"] del states["media_player.test"]
del states["media_player.test2"] del states["media_player.test2"]
del states["media_player.test3"] del states["media_player.test3"]
@ -550,23 +523,21 @@ class TestComponentHistory(unittest.TestCase):
{ {
ha.DOMAIN: {}, ha.DOMAIN: {},
history.DOMAIN: { history.DOMAIN: {
history.CONF_INCLUDE: { history.CONF_INCLUDE: {history.CONF_ENTITIES: ["media_player.test"]},
history.CONF_ENTITIES: ["media_player.test"] history.CONF_EXCLUDE: {history.CONF_ENTITIES: ["media_player.test"]},
},
history.CONF_EXCLUDE: {
history.CONF_ENTITIES: ["media_player.test"]
},
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_include_exclude(self):
def test_get_significant_states_include_exclude(hass_history):
"""Test if significant states when in/excluding domains and entities. """Test if significant states when in/excluding domains and entities.
We should only get back changes of the media_player.test2 entity. We should only get back changes of the media_player.test2 entity.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, states = record_states(hass)
del states["media_player.test"] del states["media_player.test"]
del states["thermostat.test"] del states["thermostat.test"]
del states["thermostat.test2"] del states["thermostat.test2"]
@ -587,36 +558,39 @@ class TestComponentHistory(unittest.TestCase):
}, },
} }
) )
self.check_significant_states(zero, four, states, config) check_significant_states(hass, zero, four, states, config)
def test_get_significant_states_are_ordered(self):
def test_get_significant_states_are_ordered(hass_history):
"""Test order of results from get_significant_states. """Test order of results from get_significant_states.
When entity ids are given, the results should be returned with the data When entity ids are given, the results should be returned with the data
in the same order. in the same order.
""" """
zero, four, states = self.record_states() hass = hass_history
zero, four, _states = record_states(hass)
entity_ids = ["media_player.test", "media_player.test2"] entity_ids = ["media_player.test", "media_player.test2"]
hist = history.get_significant_states( hist = history.get_significant_states(
self.hass, zero, four, entity_ids, filters=history.Filters() hass, zero, four, entity_ids, filters=history.Filters()
) )
assert list(hist.keys()) == entity_ids assert list(hist.keys()) == entity_ids
entity_ids = ["media_player.test2", "media_player.test"] entity_ids = ["media_player.test2", "media_player.test"]
hist = history.get_significant_states( hist = history.get_significant_states(
self.hass, zero, four, entity_ids, filters=history.Filters() hass, zero, four, entity_ids, filters=history.Filters()
) )
assert list(hist.keys()) == entity_ids assert list(hist.keys()) == entity_ids
def test_get_significant_states_only(self):
def test_get_significant_states_only(hass_history):
"""Test significant states when significant_states_only is set.""" """Test significant states when significant_states_only is set."""
self.test_setup() hass = hass_history
entity_id = "sensor.test" entity_id = "sensor.test"
def set_state(state, **kwargs): def set_state(state, **kwargs):
"""Set the state.""" """Set the state."""
self.hass.states.set(entity_id, state, **kwargs) hass.states.set(entity_id, state, **kwargs)
wait_recording_done(self.hass) wait_recording_done(hass)
return self.hass.states.get(entity_id) return hass.states.get(entity_id)
start = dt_util.utcnow() - timedelta(minutes=4) start = dt_util.utcnow() - timedelta(minutes=4)
points = [] points = []
@ -624,9 +598,7 @@ class TestComponentHistory(unittest.TestCase):
points.append(start + timedelta(minutes=i)) points.append(start + timedelta(minutes=i))
states = [] states = []
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=start):
"homeassistant.components.recorder.dt_util.utcnow", return_value=start
):
set_state("123", attributes={"attribute": 10.64}) set_state("123", attributes={"attribute": 10.64})
with patch( with patch(
@ -647,23 +619,20 @@ class TestComponentHistory(unittest.TestCase):
# everything is different # everything is different
states.append(set_state("412", attributes={"attribute": 54.23})) states.append(set_state("412", attributes={"attribute": 54.23}))
hist = history.get_significant_states( hist = history.get_significant_states(hass, start, significant_changes_only=True)
self.hass, start, significant_changes_only=True
)
assert len(hist[entity_id]) == 2 assert len(hist[entity_id]) == 2
assert states[0] not in hist[entity_id] assert states[0] not in hist[entity_id]
assert states[1] in hist[entity_id] assert states[1] in hist[entity_id]
assert states[2] in hist[entity_id] assert states[2] in hist[entity_id]
hist = history.get_significant_states( hist = history.get_significant_states(hass, start, significant_changes_only=False)
self.hass, start, significant_changes_only=False
)
assert len(hist[entity_id]) == 3 assert len(hist[entity_id]) == 3
assert states == hist[entity_id] assert states == hist[entity_id]
def check_significant_states(self, zero, four, states, config):
def check_significant_states(hass, zero, four, states, config):
"""Check if significant states are retrieved.""" """Check if significant states are retrieved."""
filters = history.Filters() filters = history.Filters()
exclude = config[history.DOMAIN].get(history.CONF_EXCLUDE) exclude = config[history.DOMAIN].get(history.CONF_EXCLUDE)
@ -675,16 +644,16 @@ class TestComponentHistory(unittest.TestCase):
filters.included_entities = include.get(history.CONF_ENTITIES, []) filters.included_entities = include.get(history.CONF_ENTITIES, [])
filters.included_domains = include.get(history.CONF_DOMAINS, []) filters.included_domains = include.get(history.CONF_DOMAINS, [])
hist = history.get_significant_states(self.hass, zero, four, filters=filters) hist = history.get_significant_states(hass, zero, four, filters=filters)
assert states == hist assert states == hist
def record_states(self):
def record_states(hass):
"""Record some test states. """Record some test states.
We inject a bunch of state updates from media player, zone and We inject a bunch of state updates from media player, zone and
thermostat. thermostat.
""" """
self.test_setup()
mp = "media_player.test" mp = "media_player.test"
mp2 = "media_player.test2" mp2 = "media_player.test2"
mp3 = "media_player.test3" mp3 = "media_player.test3"
@ -695,9 +664,9 @@ class TestComponentHistory(unittest.TestCase):
def set_state(entity_id, state, **kwargs): def set_state(entity_id, state, **kwargs):
"""Set the state.""" """Set the state."""
self.hass.states.set(entity_id, state, **kwargs) hass.states.set(entity_id, state, **kwargs)
wait_recording_done(self.hass) wait_recording_done(hass)
return self.hass.states.get(entity_id) return hass.states.get(entity_id)
zero = dt_util.utcnow() zero = dt_util.utcnow()
one = zero + timedelta(seconds=1) one = zero + timedelta(seconds=1)
@ -706,9 +675,7 @@ class TestComponentHistory(unittest.TestCase):
four = three + timedelta(seconds=1) four = three + timedelta(seconds=1)
states = {therm: [], therm2: [], mp: [], mp2: [], mp3: [], script_c: []} states = {therm: [], therm2: [], mp: [], mp2: [], mp3: [], script_c: []}
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=one):
"homeassistant.components.recorder.dt_util.utcnow", return_value=one
):
states[mp].append( states[mp].append(
set_state(mp, "idle", attributes={"media_title": str(sentinel.mt1)}) set_state(mp, "idle", attributes={"media_title": str(sentinel.mt1)})
) )
@ -725,9 +692,7 @@ class TestComponentHistory(unittest.TestCase):
set_state(therm, 20, attributes={"current_temperature": 19.5}) set_state(therm, 20, attributes={"current_temperature": 19.5})
) )
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=two):
"homeassistant.components.recorder.dt_util.utcnow", return_value=two
):
# This state will be skipped only different in time # This state will be skipped only different in time
set_state(mp, "YouTube", attributes={"media_title": str(sentinel.mt3)}) set_state(mp, "YouTube", attributes={"media_title": str(sentinel.mt3)})
# This state will be skipped because domain is excluded # This state will be skipped because domain is excluded
@ -742,9 +707,7 @@ class TestComponentHistory(unittest.TestCase):
set_state(therm2, 20, attributes={"current_temperature": 19}) set_state(therm2, 20, attributes={"current_temperature": 19})
) )
with patch( with patch("homeassistant.components.recorder.dt_util.utcnow", return_value=three):
"homeassistant.components.recorder.dt_util.utcnow", return_value=three
):
states[mp].append( states[mp].append(
set_state(mp, "Netflix", attributes={"media_title": str(sentinel.mt4)}) set_state(mp, "Netflix", attributes={"media_title": str(sentinel.mt4)})
) )