Rewrite recorder unittest tests to pytest style test function (#41264)
This commit is contained in:
Ariana Hlavaty
GitHub
parent
d35e33610a
commit
1e9e40bf71
4 changed files with 346 additions and 405 deletions
|
|
@ -1,6 +1,5 @@
|
|||
"""The tests for the Recorder component."""
|
||||
from datetime import datetime
|
||||
import unittest
|
||||
|
||||
import pytest
|
||||
import pytz
|
||||
|
|
@ -21,150 +20,112 @@ from homeassistant.exceptions import InvalidEntityFormatError
|
|||
from homeassistant.util import dt
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
ENGINE = None
|
||||
SESSION = None
|
||||
|
||||
def test_from_event_to_db_event():
|
||||
"""Test converting event to db event."""
|
||||
event = ha.Event("test_event", {"some_data": 15})
|
||||
assert event == Events.from_event(event).to_native()
|
||||
|
||||
|
||||
def setUpModule(): # pylint: disable=invalid-name
|
||||
"""Set up a database to use."""
|
||||
global ENGINE
|
||||
global SESSION
|
||||
|
||||
ENGINE = create_engine("sqlite://")
|
||||
Base.metadata.create_all(ENGINE)
|
||||
session_factory = sessionmaker(bind=ENGINE)
|
||||
SESSION = scoped_session(session_factory)
|
||||
def test_from_event_to_db_state():
|
||||
"""Test converting event to db state."""
|
||||
state = ha.State("sensor.temperature", "18")
|
||||
event = ha.Event(
|
||||
EVENT_STATE_CHANGED,
|
||||
{"entity_id": "sensor.temperature", "old_state": None, "new_state": state},
|
||||
context=state.context,
|
||||
)
|
||||
# We don't restore context unless we need it by joining the
|
||||
# events table on the event_id for state_changed events
|
||||
state.context = ha.Context(id=None)
|
||||
assert state == States.from_event(event).to_native()
|
||||
|
||||
|
||||
def tearDownModule(): # pylint: disable=invalid-name
|
||||
"""Close database."""
|
||||
global ENGINE
|
||||
global SESSION
|
||||
def test_from_event_to_delete_state():
|
||||
"""Test converting deleting state event to db state."""
|
||||
event = ha.Event(
|
||||
EVENT_STATE_CHANGED,
|
||||
{
|
||||
"entity_id": "sensor.temperature",
|
||||
"old_state": ha.State("sensor.temperature", "18"),
|
||||
"new_state": None,
|
||||
},
|
||||
)
|
||||
db_state = States.from_event(event)
|
||||
|
||||
ENGINE.dispose()
|
||||
ENGINE = None
|
||||
SESSION = None
|
||||
assert db_state.entity_id == "sensor.temperature"
|
||||
assert db_state.domain == "sensor"
|
||||
assert db_state.state == ""
|
||||
assert db_state.last_changed == event.time_fired
|
||||
assert db_state.last_updated == event.time_fired
|
||||
|
||||
|
||||
class TestEvents(unittest.TestCase):
|
||||
"""Test Events model."""
|
||||
def test_entity_ids():
|
||||
"""Test if entity ids helper method works."""
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
session_factory = sessionmaker(bind=engine)
|
||||
|
||||
# pylint: disable=no-self-use
|
||||
def test_from_event(self):
|
||||
"""Test converting event to db event."""
|
||||
event = ha.Event("test_event", {"some_data": 15})
|
||||
assert event == Events.from_event(event).to_native()
|
||||
session = scoped_session(session_factory)
|
||||
session.query(Events).delete()
|
||||
session.query(States).delete()
|
||||
session.query(RecorderRuns).delete()
|
||||
|
||||
run = RecorderRuns(
|
||||
start=datetime(2016, 7, 9, 11, 0, 0, tzinfo=dt.UTC),
|
||||
end=datetime(2016, 7, 9, 23, 0, 0, tzinfo=dt.UTC),
|
||||
closed_incorrect=False,
|
||||
created=datetime(2016, 7, 9, 11, 0, 0, tzinfo=dt.UTC),
|
||||
)
|
||||
|
||||
class TestStates(unittest.TestCase):
|
||||
"""Test States model."""
|
||||
session.add(run)
|
||||
session.commit()
|
||||
|
||||
# pylint: disable=no-self-use
|
||||
before_run = datetime(2016, 7, 9, 8, 0, 0, tzinfo=dt.UTC)
|
||||
in_run = datetime(2016, 7, 9, 13, 0, 0, tzinfo=dt.UTC)
|
||||
in_run2 = datetime(2016, 7, 9, 15, 0, 0, tzinfo=dt.UTC)
|
||||
in_run3 = datetime(2016, 7, 9, 18, 0, 0, tzinfo=dt.UTC)
|
||||
after_run = datetime(2016, 7, 9, 23, 30, 0, tzinfo=dt.UTC)
|
||||
|
||||
def test_from_event(self):
|
||||
"""Test converting event to db state."""
|
||||
state = ha.State("sensor.temperature", "18")
|
||||
event = ha.Event(
|
||||
EVENT_STATE_CHANGED,
|
||||
{"entity_id": "sensor.temperature", "old_state": None, "new_state": state},
|
||||
context=state.context,
|
||||
assert run.to_native() == run
|
||||
assert run.entity_ids() == []
|
||||
|
||||
session.add(
|
||||
States(
|
||||
entity_id="sensor.temperature",
|
||||
state="20",
|
||||
last_changed=before_run,
|
||||
last_updated=before_run,
|
||||
)
|
||||
# We don't restore context unless we need it by joining the
|
||||
# events table on the event_id for state_changed events
|
||||
state.context = ha.Context(id=None)
|
||||
assert state == States.from_event(event).to_native()
|
||||
|
||||
def test_from_event_to_delete_state(self):
|
||||
"""Test converting deleting state event to db state."""
|
||||
event = ha.Event(
|
||||
EVENT_STATE_CHANGED,
|
||||
{
|
||||
"entity_id": "sensor.temperature",
|
||||
"old_state": ha.State("sensor.temperature", "18"),
|
||||
"new_state": None,
|
||||
},
|
||||
)
|
||||
session.add(
|
||||
States(
|
||||
entity_id="sensor.sound",
|
||||
state="10",
|
||||
last_changed=after_run,
|
||||
last_updated=after_run,
|
||||
)
|
||||
db_state = States.from_event(event)
|
||||
)
|
||||
|
||||
assert db_state.entity_id == "sensor.temperature"
|
||||
assert db_state.domain == "sensor"
|
||||
assert db_state.state == ""
|
||||
assert db_state.last_changed == event.time_fired
|
||||
assert db_state.last_updated == event.time_fired
|
||||
|
||||
|
||||
class TestRecorderRuns(unittest.TestCase):
|
||||
"""Test recorder run model."""
|
||||
|
||||
def setUp(self): # pylint: disable=invalid-name
|
||||
"""Set up recorder runs."""
|
||||
self.session = session = SESSION()
|
||||
session.query(Events).delete()
|
||||
session.query(States).delete()
|
||||
session.query(RecorderRuns).delete()
|
||||
self.addCleanup(self.tear_down_cleanup)
|
||||
|
||||
def tear_down_cleanup(self):
|
||||
"""Clean up."""
|
||||
self.session.rollback()
|
||||
|
||||
def test_entity_ids(self):
|
||||
"""Test if entity ids helper method works."""
|
||||
run = RecorderRuns(
|
||||
start=datetime(2016, 7, 9, 11, 0, 0, tzinfo=dt.UTC),
|
||||
end=datetime(2016, 7, 9, 23, 0, 0, tzinfo=dt.UTC),
|
||||
closed_incorrect=False,
|
||||
created=datetime(2016, 7, 9, 11, 0, 0, tzinfo=dt.UTC),
|
||||
session.add(
|
||||
States(
|
||||
entity_id="sensor.humidity",
|
||||
state="76",
|
||||
last_changed=in_run,
|
||||
last_updated=in_run,
|
||||
)
|
||||
|
||||
self.session.add(run)
|
||||
self.session.commit()
|
||||
|
||||
before_run = datetime(2016, 7, 9, 8, 0, 0, tzinfo=dt.UTC)
|
||||
in_run = datetime(2016, 7, 9, 13, 0, 0, tzinfo=dt.UTC)
|
||||
in_run2 = datetime(2016, 7, 9, 15, 0, 0, tzinfo=dt.UTC)
|
||||
in_run3 = datetime(2016, 7, 9, 18, 0, 0, tzinfo=dt.UTC)
|
||||
after_run = datetime(2016, 7, 9, 23, 30, 0, tzinfo=dt.UTC)
|
||||
|
||||
assert run.to_native() == run
|
||||
assert run.entity_ids() == []
|
||||
|
||||
self.session.add(
|
||||
States(
|
||||
entity_id="sensor.temperature",
|
||||
state="20",
|
||||
last_changed=before_run,
|
||||
last_updated=before_run,
|
||||
)
|
||||
)
|
||||
self.session.add(
|
||||
States(
|
||||
entity_id="sensor.sound",
|
||||
state="10",
|
||||
last_changed=after_run,
|
||||
last_updated=after_run,
|
||||
)
|
||||
)
|
||||
session.add(
|
||||
States(
|
||||
entity_id="sensor.lux",
|
||||
state="5",
|
||||
last_changed=in_run3,
|
||||
last_updated=in_run3,
|
||||
)
|
||||
)
|
||||
|
||||
self.session.add(
|
||||
States(
|
||||
entity_id="sensor.humidity",
|
||||
state="76",
|
||||
last_changed=in_run,
|
||||
last_updated=in_run,
|
||||
)
|
||||
)
|
||||
self.session.add(
|
||||
States(
|
||||
entity_id="sensor.lux",
|
||||
state="5",
|
||||
last_changed=in_run3,
|
||||
last_updated=in_run3,
|
||||
)
|
||||
)
|
||||
|
||||
assert sorted(run.entity_ids()) == ["sensor.humidity", "sensor.lux"]
|
||||
assert run.entity_ids(in_run2) == ["sensor.humidity"]
|
||||
assert sorted(run.entity_ids()) == ["sensor.humidity", "sensor.lux"]
|
||||
assert run.entity_ids(in_run2) == ["sensor.humidity"]
|
||||
|
||||
|
||||
def test_states_from_native_invalid_entity_id():
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue