Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions tests/components/recorder/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
"""Common test tools."""

import pytest

from homeassistant.components.recorder.const import DATA_INSTANCE

from tests.common import get_test_home_assistant, init_recorder_component


@pytest.fixture
def hass_recorder():
Comment thread
balloob marked this conversation as resolved.
"""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()
Comment thread
balloob marked this conversation as resolved.
119 changes: 43 additions & 76 deletions tests/components/recorder/test_init.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,6 @@
"""The tests for the Recorder component."""
# pylint: disable=protected-access
from datetime import datetime, timedelta
import unittest

import pytest

from homeassistant.components.recorder import (
CONFIG_SCHEMA,
Expand All @@ -24,99 +21,69 @@
from .common import wait_recording_done

from tests.async_mock import patch
from tests.common import (
async_fire_time_changed,
get_test_home_assistant,
init_recorder_component,
)


class TestRecorder(unittest.TestCase):
"""Test the recorder module."""

def setUp(self): # pylint: disable=invalid-name
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
init_recorder_component(self.hass)
self.hass.start()
self.addCleanup(self.tear_down_cleanup)
from tests.common import async_fire_time_changed, get_test_home_assistant

def tear_down_cleanup(self):
"""Stop everything that was started."""
self.hass.stop()

def test_saving_state(self):
"""Test saving and restoring a state."""
entity_id = "test.recorder"
state = "restoring_from_db"
attributes = {"test_attr": 5, "test_attr_10": "nice"}

self.hass.states.set(entity_id, state, attributes)
def test_saving_state(hass, hass_recorder):
"""Test saving and restoring a state."""
hass = hass_recorder()

wait_recording_done(self.hass)
entity_id = "test.recorder"
state = "restoring_from_db"
attributes = {"test_attr": 5, "test_attr_10": "nice"}

with session_scope(hass=self.hass) as session:
db_states = list(session.query(States))
assert len(db_states) == 1
assert db_states[0].event_id > 0
state = db_states[0].to_native()
hass.states.set(entity_id, state, attributes)

assert state == _state_empty_context(self.hass, entity_id)
wait_recording_done(hass)

def test_saving_event(self):
"""Test saving and restoring an event."""
event_type = "EVENT_TEST"
event_data = {"test_attr": 5, "test_attr_10": "nice"}
with session_scope(hass=hass) as session:
db_states = list(session.query(States))
assert len(db_states) == 1
assert db_states[0].event_id > 0
state = db_states[0].to_native()

events = []
assert state == _state_empty_context(hass, entity_id)

@callback
def event_listener(event):
"""Record events from eventbus."""
if event.event_type == event_type:
events.append(event)

self.hass.bus.listen(MATCH_ALL, event_listener)
def test_saving_event(hass, hass_recorder):
"""Test saving and restoring an event."""
hass = hass_recorder()

self.hass.bus.fire(event_type, event_data)
event_type = "EVENT_TEST"
event_data = {"test_attr": 5, "test_attr_10": "nice"}

wait_recording_done(self.hass)
events = []

assert len(events) == 1
event = events[0]
@callback
def event_listener(event):
"""Record events from eventbus."""
if event.event_type == event_type:
events.append(event)

self.hass.data[DATA_INSTANCE].block_till_done()
hass.bus.listen(MATCH_ALL, event_listener)

with session_scope(hass=self.hass) as session:
db_events = list(session.query(Events).filter_by(event_type=event_type))
assert len(db_events) == 1
db_event = db_events[0].to_native()
hass.bus.fire(event_type, event_data)

assert event.event_type == db_event.event_type
assert event.data == db_event.data
assert event.origin == db_event.origin
wait_recording_done(hass)

# Recorder uses SQLite and stores datetimes as integer unix timestamps
assert event.time_fired.replace(microsecond=0) == db_event.time_fired.replace(
microsecond=0
)
assert len(events) == 1
event = events[0]

hass.data[DATA_INSTANCE].block_till_done()

@pytest.fixture
def hass_recorder():
"""Home Assistant fixture with in-memory recorder."""
hass = get_test_home_assistant()
with session_scope(hass=hass) as session:
db_events = list(session.query(Events).filter_by(event_type=event_type))
assert len(db_events) == 1
db_event = db_events[0].to_native()

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
assert event.event_type == db_event.event_type
assert event.data == db_event.data
assert event.origin == db_event.origin

yield setup_recorder
hass.stop()
# Recorder uses SQLite and stores datetimes as integer unix timestamps
assert event.time_fired.replace(microsecond=0) == db_event.time_fired.replace(
microsecond=0
)


def _add_entities(hass, entity_ids):
Expand Down
Loading