a6511fc0b9
* remove the need to have query feature support Some InfluxDB servers don't have /query support feature but are still valid servers for storing data. Usually those servers are proxies to others timeseries databases. The change proposes to still validate the configuration but with less requirements on the server side. * `.query` call is replaced by `.write_points` * no more query call in the influxdb component. remove test * reset mock after the setup and before the test * remove unused import * reset mock stats after component setup
692 lines
26 KiB
Python
692 lines
26 KiB
Python
"""The tests for the InfluxDB component."""
|
|
import datetime
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from homeassistant.setup import setup_component
|
|
import homeassistant.components.influxdb as influxdb
|
|
from homeassistant.const import EVENT_STATE_CHANGED, STATE_OFF, STATE_ON, \
|
|
STATE_STANDBY
|
|
|
|
from tests.common import get_test_home_assistant
|
|
|
|
|
|
@mock.patch('influxdb.InfluxDBClient')
|
|
@mock.patch(
|
|
'homeassistant.components.influxdb.InfluxThread.batch_timeout',
|
|
mock.Mock(return_value=0))
|
|
class TestInfluxDB(unittest.TestCase):
|
|
"""Test the InfluxDB component."""
|
|
|
|
def setUp(self):
|
|
"""Set up things to be run when tests are started."""
|
|
self.hass = get_test_home_assistant()
|
|
self.handler_method = None
|
|
self.hass.bus.listen = mock.Mock()
|
|
|
|
def tearDown(self):
|
|
"""Clear data."""
|
|
self.hass.stop()
|
|
|
|
def test_setup_config_full(self, mock_client):
|
|
"""Test the setup with full configuration."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'port': 123,
|
|
'database': 'db',
|
|
'username': 'user',
|
|
'password': 'password',
|
|
'max_retries': 4,
|
|
'ssl': 'False',
|
|
'verify_ssl': 'False',
|
|
}
|
|
}
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
assert self.hass.bus.listen.called
|
|
assert \
|
|
EVENT_STATE_CHANGED == self.hass.bus.listen.call_args_list[0][0][0]
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
|
|
def test_setup_config_defaults(self, mock_client):
|
|
"""Test the setup with default configuration."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
}
|
|
}
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
assert self.hass.bus.listen.called
|
|
assert \
|
|
EVENT_STATE_CHANGED == self.hass.bus.listen.call_args_list[0][0][0]
|
|
|
|
def test_setup_minimal_config(self, mock_client):
|
|
"""Test the setup with minimal configuration."""
|
|
config = {
|
|
'influxdb': {}
|
|
}
|
|
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
|
|
def test_setup_missing_password(self, mock_client):
|
|
"""Test the setup with existing username and missing password."""
|
|
config = {
|
|
'influxdb': {
|
|
'username': 'user'
|
|
}
|
|
}
|
|
|
|
assert not setup_component(self.hass, influxdb.DOMAIN, config)
|
|
|
|
def _setup(self, mock_client, **kwargs):
|
|
"""Set up the client."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
'exclude': {
|
|
'entities': ['fake.blacklisted'],
|
|
'domains': ['another_fake']
|
|
}
|
|
}
|
|
}
|
|
config['influxdb'].update(kwargs)
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
self.handler_method = self.hass.bus.listen.call_args_list[0][0][1]
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener(self, mock_client):
|
|
"""Test the event listener."""
|
|
self._setup(mock_client)
|
|
|
|
# map of HA State to valid influxdb [state, value] fields
|
|
valid = {
|
|
'1': [None, 1],
|
|
'1.0': [None, 1.0],
|
|
STATE_ON: [STATE_ON, 1],
|
|
STATE_OFF: [STATE_OFF, 0],
|
|
STATE_STANDBY: [STATE_STANDBY, None],
|
|
'foo': ['foo', None]
|
|
}
|
|
for in_, out in valid.items():
|
|
attrs = {
|
|
'unit_of_measurement': 'foobars',
|
|
'longitude': '1.1',
|
|
'latitude': '2.2',
|
|
'battery_level': '99%',
|
|
'temperature': '20c',
|
|
'last_seen': 'Last seen 23 minutes ago',
|
|
'updated_at': datetime.datetime(2017, 1, 1, 0, 0),
|
|
'multi_periods': '0.120.240.2023873'
|
|
}
|
|
state = mock.MagicMock(
|
|
state=in_, domain='fake', entity_id='fake.entity-id',
|
|
object_id='entity', attributes=attrs)
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'foobars',
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': 'entity',
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'longitude': 1.1,
|
|
'latitude': 2.2,
|
|
'battery_level_str': '99%',
|
|
'battery_level': 99.0,
|
|
'temperature_str': '20c',
|
|
'temperature': 20.0,
|
|
'last_seen_str': 'Last seen 23 minutes ago',
|
|
'last_seen': 23.0,
|
|
'updated_at_str': '2017-01-01 00:00:00',
|
|
'updated_at': 20170101000000,
|
|
'multi_periods_str': '0.120.240.2023873'
|
|
},
|
|
}]
|
|
if out[0] is not None:
|
|
body[0]['fields']['state'] = out[0]
|
|
if out[1] is not None:
|
|
body[0]['fields']['value'] = out[1]
|
|
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_no_units(self, mock_client):
|
|
"""Test the event listener for missing units."""
|
|
self._setup(mock_client)
|
|
|
|
for unit in (None, ''):
|
|
if unit:
|
|
attrs = {'unit_of_measurement': unit}
|
|
else:
|
|
attrs = {}
|
|
state = mock.MagicMock(
|
|
state=1, domain='fake', entity_id='fake.entity-id',
|
|
object_id='entity', attributes=attrs)
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'fake.entity-id',
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': 'entity',
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 1,
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_inf(self, mock_client):
|
|
"""Test the event listener for missing units."""
|
|
self._setup(mock_client)
|
|
|
|
attrs = {'bignumstring': '9' * 999, 'nonumstring': 'nan'}
|
|
state = mock.MagicMock(
|
|
state=8, domain='fake', entity_id='fake.entity-id',
|
|
object_id='entity', attributes=attrs)
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'fake.entity-id',
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': 'entity',
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 8,
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_states(self, mock_client):
|
|
"""Test the event listener against ignored states."""
|
|
self._setup(mock_client)
|
|
|
|
for state_state in (1, 'unknown', '', 'unavailable'):
|
|
state = mock.MagicMock(
|
|
state=state_state, domain='fake', entity_id='fake.entity-id',
|
|
object_id='entity', attributes={})
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'fake.entity-id',
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': 'entity',
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 1,
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
if state_state == 1:
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
else:
|
|
assert not mock_client.return_value.write_points.called
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_blacklist(self, mock_client):
|
|
"""Test the event listener against a blacklist."""
|
|
self._setup(mock_client)
|
|
|
|
for entity_id in ('ok', 'blacklisted'):
|
|
state = mock.MagicMock(
|
|
state=1, domain='fake', entity_id='fake.{}'.format(entity_id),
|
|
object_id=entity_id, attributes={})
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'fake.{}'.format(entity_id),
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': entity_id,
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 1,
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
if entity_id == 'ok':
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
else:
|
|
assert not mock_client.return_value.write_points.called
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_blacklist_domain(self, mock_client):
|
|
"""Test the event listener against a blacklist."""
|
|
self._setup(mock_client)
|
|
|
|
for domain in ('ok', 'another_fake'):
|
|
state = mock.MagicMock(
|
|
state=1, domain=domain,
|
|
entity_id='{}.something'.format(domain),
|
|
object_id='something', attributes={})
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': '{}.something'.format(domain),
|
|
'tags': {
|
|
'domain': domain,
|
|
'entity_id': 'something',
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 1,
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
if domain == 'ok':
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
else:
|
|
assert not mock_client.return_value.write_points.called
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_whitelist(self, mock_client):
|
|
"""Test the event listener against a whitelist."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
'include': {
|
|
'entities': ['fake.included'],
|
|
}
|
|
}
|
|
}
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
self.handler_method = self.hass.bus.listen.call_args_list[0][0][1]
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
for entity_id in ('included', 'default'):
|
|
state = mock.MagicMock(
|
|
state=1, domain='fake', entity_id='fake.{}'.format(entity_id),
|
|
object_id=entity_id, attributes={})
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'fake.{}'.format(entity_id),
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': entity_id,
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 1,
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
if entity_id == 'included':
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
else:
|
|
assert not mock_client.return_value.write_points.called
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_whitelist_domain(self, mock_client):
|
|
"""Test the event listener against a whitelist."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
'include': {
|
|
'domains': ['fake'],
|
|
}
|
|
}
|
|
}
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
self.handler_method = self.hass.bus.listen.call_args_list[0][0][1]
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
for domain in ('fake', 'another_fake'):
|
|
state = mock.MagicMock(
|
|
state=1, domain=domain,
|
|
entity_id='{}.something'.format(domain),
|
|
object_id='something', attributes={})
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': '{}.something'.format(domain),
|
|
'tags': {
|
|
'domain': domain,
|
|
'entity_id': 'something',
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 1,
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
if domain == 'fake':
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
else:
|
|
assert not mock_client.return_value.write_points.called
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_invalid_type(self, mock_client):
|
|
"""Test the event listener when an attribute has an invalid type."""
|
|
self._setup(mock_client)
|
|
|
|
# map of HA State to valid influxdb [state, value] fields
|
|
valid = {
|
|
'1': [None, 1],
|
|
'1.0': [None, 1.0],
|
|
STATE_ON: [STATE_ON, 1],
|
|
STATE_OFF: [STATE_OFF, 0],
|
|
STATE_STANDBY: [STATE_STANDBY, None],
|
|
'foo': ['foo', None]
|
|
}
|
|
for in_, out in valid.items():
|
|
attrs = {
|
|
'unit_of_measurement': 'foobars',
|
|
'longitude': '1.1',
|
|
'latitude': '2.2',
|
|
'invalid_attribute': ['value1', 'value2']
|
|
}
|
|
state = mock.MagicMock(
|
|
state=in_, domain='fake', entity_id='fake.entity-id',
|
|
object_id='entity', attributes=attrs)
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'foobars',
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': 'entity',
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'longitude': 1.1,
|
|
'latitude': 2.2,
|
|
'invalid_attribute_str': "['value1', 'value2']"
|
|
},
|
|
}]
|
|
if out[0] is not None:
|
|
body[0]['fields']['state'] = out[0]
|
|
if out[1] is not None:
|
|
body[0]['fields']['value'] = out[1]
|
|
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_default_measurement(self, mock_client):
|
|
"""Test the event listener with a default measurement."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
'default_measurement': 'state',
|
|
'exclude': {
|
|
'entities': ['fake.blacklisted']
|
|
}
|
|
}
|
|
}
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
self.handler_method = self.hass.bus.listen.call_args_list[0][0][1]
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
for entity_id in ('ok', 'blacklisted'):
|
|
state = mock.MagicMock(
|
|
state=1, domain='fake', entity_id='fake.{}'.format(entity_id),
|
|
object_id=entity_id, attributes={})
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'state',
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': entity_id,
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 1,
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
if entity_id == 'ok':
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
else:
|
|
assert not mock_client.return_value.write_points.called
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_unit_of_measurement_field(self, mock_client):
|
|
"""Test the event listener for unit of measurement field."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
'override_measurement': 'state',
|
|
}
|
|
}
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
self.handler_method = self.hass.bus.listen.call_args_list[0][0][1]
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
attrs = {
|
|
'unit_of_measurement': 'foobars',
|
|
}
|
|
state = mock.MagicMock(
|
|
state='foo', domain='fake', entity_id='fake.entity-id',
|
|
object_id='entity', attributes=attrs)
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'state',
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': 'entity',
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'state': 'foo',
|
|
'unit_of_measurement_str': 'foobars',
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_tags_attributes(self, mock_client):
|
|
"""Test the event listener when some attributes should be tags."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
'tags_attributes': ['friendly_fake']
|
|
}
|
|
}
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
self.handler_method = self.hass.bus.listen.call_args_list[0][0][1]
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
attrs = {
|
|
'friendly_fake': 'tag_str',
|
|
'field_fake': 'field_str',
|
|
}
|
|
state = mock.MagicMock(
|
|
state=1, domain='fake',
|
|
entity_id='fake.something',
|
|
object_id='something', attributes=attrs)
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': 'fake.something',
|
|
'tags': {
|
|
'domain': 'fake',
|
|
'entity_id': 'something',
|
|
'friendly_fake': 'tag_str'
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 1,
|
|
'field_fake_str': 'field_str'
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_event_listener_component_override_measurement(self, mock_client):
|
|
"""Test the event listener with overridden measurements."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
'component_config': {
|
|
'sensor.fake_humidity': {
|
|
'override_measurement': 'humidity'
|
|
}
|
|
},
|
|
'component_config_glob': {
|
|
'binary_sensor.*motion': {
|
|
'override_measurement': 'motion'
|
|
}
|
|
},
|
|
'component_config_domain': {
|
|
'climate': {
|
|
'override_measurement': 'hvac'
|
|
}
|
|
}
|
|
}
|
|
}
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
self.handler_method = self.hass.bus.listen.call_args_list[0][0][1]
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
test_components = [
|
|
{'domain': 'sensor', 'id': 'fake_humidity', 'res': 'humidity'},
|
|
{'domain': 'binary_sensor', 'id': 'fake_motion', 'res': 'motion'},
|
|
{'domain': 'climate', 'id': 'fake_thermostat', 'res': 'hvac'},
|
|
{'domain': 'other', 'id': 'just_fake', 'res': 'other.just_fake'},
|
|
]
|
|
for comp in test_components:
|
|
state = mock.MagicMock(
|
|
state=1, domain=comp['domain'],
|
|
entity_id=comp['domain'] + '.' + comp['id'],
|
|
object_id=comp['id'], attributes={})
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
body = [{
|
|
'measurement': comp['res'],
|
|
'tags': {
|
|
'domain': comp['domain'],
|
|
'entity_id': comp['id']
|
|
},
|
|
'time': 12345,
|
|
'fields': {
|
|
'value': 1,
|
|
},
|
|
}]
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert mock_client.return_value.write_points.call_count == 1
|
|
assert mock_client.return_value.write_points.call_args == \
|
|
mock.call(body)
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
def test_scheduled_write(self, mock_client):
|
|
"""Test the event listener to retry after write failures."""
|
|
config = {
|
|
'influxdb': {
|
|
'host': 'host',
|
|
'username': 'user',
|
|
'password': 'pass',
|
|
'max_retries': 1
|
|
}
|
|
}
|
|
assert setup_component(self.hass, influxdb.DOMAIN, config)
|
|
self.handler_method = self.hass.bus.listen.call_args_list[0][0][1]
|
|
mock_client.return_value.write_points.reset_mock()
|
|
|
|
state = mock.MagicMock(
|
|
state=1, domain='fake', entity_id='entity.id', object_id='entity',
|
|
attributes={})
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
mock_client.return_value.write_points.side_effect = \
|
|
IOError('foo')
|
|
|
|
# Write fails
|
|
with mock.patch.object(influxdb.time, 'sleep') as mock_sleep:
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert mock_sleep.called
|
|
json_data = mock_client.return_value.write_points.call_args[0][0]
|
|
assert mock_client.return_value.write_points.call_count == 2
|
|
mock_client.return_value.write_points.assert_called_with(json_data)
|
|
|
|
# Write works again
|
|
mock_client.return_value.write_points.side_effect = None
|
|
with mock.patch.object(influxdb.time, 'sleep') as mock_sleep:
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert not mock_sleep.called
|
|
assert mock_client.return_value.write_points.call_count == 3
|
|
|
|
def test_queue_backlog_full(self, mock_client):
|
|
"""Test the event listener to drop old events."""
|
|
self._setup(mock_client)
|
|
|
|
state = mock.MagicMock(
|
|
state=1, domain='fake', entity_id='entity.id', object_id='entity',
|
|
attributes={})
|
|
event = mock.MagicMock(data={'new_state': state}, time_fired=12345)
|
|
|
|
monotonic_time = 0
|
|
|
|
def fast_monotonic():
|
|
"""Monotonic time that ticks fast enough to cause a timeout."""
|
|
nonlocal monotonic_time
|
|
monotonic_time += 60
|
|
return monotonic_time
|
|
|
|
with mock.patch('homeassistant.components.influxdb.time.monotonic',
|
|
new=fast_monotonic):
|
|
self.handler_method(event)
|
|
self.hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
assert mock_client.return_value.write_points.call_count == 0
|
|
|
|
mock_client.return_value.write_points.reset_mock()
|