1114 lines
34 KiB
Python
1114 lines
34 KiB
Python
"""The tests for the InfluxDB component."""
|
|
import datetime
|
|
|
|
import pytest
|
|
|
|
import homeassistant.components.influxdb as influxdb
|
|
from homeassistant.const import (
|
|
EVENT_STATE_CHANGED,
|
|
STATE_OFF,
|
|
STATE_ON,
|
|
STATE_STANDBY,
|
|
UNIT_PERCENTAGE,
|
|
)
|
|
from homeassistant.setup import async_setup_component
|
|
|
|
from tests.async_mock import MagicMock, Mock, call, patch
|
|
|
|
BASE_V1_CONFIG = {}
|
|
BASE_V2_CONFIG = {
|
|
"api_version": influxdb.API_VERSION_2,
|
|
"organization": "org",
|
|
"token": "token",
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_batch_timeout(hass, monkeypatch):
|
|
"""Mock the event bus listener and the batch timeout for tests."""
|
|
hass.bus.listen = MagicMock()
|
|
monkeypatch.setattr(
|
|
"homeassistant.components.influxdb.InfluxThread.batch_timeout",
|
|
Mock(return_value=0),
|
|
)
|
|
|
|
|
|
@pytest.fixture(name="mock_client")
|
|
def mock_client_fixture(request):
|
|
"""Patch the InfluxDBClient object with mock for version under test."""
|
|
if request.param == influxdb.API_VERSION_2:
|
|
client_target = "homeassistant.components.influxdb.InfluxDBClientV2"
|
|
else:
|
|
client_target = "homeassistant.components.influxdb.InfluxDBClient"
|
|
|
|
with patch(client_target) as client:
|
|
yield client
|
|
|
|
|
|
@pytest.fixture(name="get_mock_call")
|
|
def get_mock_call_fixture(request):
|
|
"""Get version specific lambda to make write API call mock."""
|
|
if request.param == influxdb.API_VERSION_2:
|
|
return lambda body: call(bucket=influxdb.DEFAULT_BUCKET, record=body)
|
|
# pylint: disable=unnecessary-lambda
|
|
return lambda body: call(body)
|
|
|
|
|
|
def _get_write_api_mock_v1(mock_influx_client):
|
|
"""Return the write api mock for the V1 client."""
|
|
return mock_influx_client.return_value.write_points
|
|
|
|
|
|
def _get_write_api_mock_v2(mock_influx_client):
|
|
"""Return the write api mock for the V2 client."""
|
|
return mock_influx_client.return_value.write_api.return_value.write
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
{
|
|
"api_version": influxdb.DEFAULT_API_VERSION,
|
|
"username": "user",
|
|
"password": "password",
|
|
"verify_ssl": "False",
|
|
},
|
|
_get_write_api_mock_v1,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
{
|
|
"api_version": influxdb.API_VERSION_2,
|
|
"token": "token",
|
|
"organization": "organization",
|
|
"bucket": "bucket",
|
|
},
|
|
_get_write_api_mock_v2,
|
|
),
|
|
],
|
|
indirect=["mock_client"],
|
|
)
|
|
async def test_setup_config_full(hass, mock_client, config_ext, get_write_api):
|
|
"""Test the setup with full configuration."""
|
|
config = {
|
|
"influxdb": {
|
|
"host": "host",
|
|
"port": 123,
|
|
"database": "db",
|
|
"max_retries": 4,
|
|
"ssl": "False",
|
|
}
|
|
}
|
|
config["influxdb"].update(config_ext)
|
|
|
|
assert await async_setup_component(hass, influxdb.DOMAIN, config)
|
|
await hass.async_block_till_done()
|
|
assert hass.bus.listen.called
|
|
assert EVENT_STATE_CHANGED == hass.bus.listen.call_args_list[0][0][0]
|
|
assert get_write_api(mock_client).call_count == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api",
|
|
[
|
|
(influxdb.DEFAULT_API_VERSION, BASE_V1_CONFIG, _get_write_api_mock_v1),
|
|
(influxdb.API_VERSION_2, BASE_V2_CONFIG, _get_write_api_mock_v2),
|
|
],
|
|
indirect=["mock_client"],
|
|
)
|
|
async def test_setup_minimal_config(hass, mock_client, config_ext, get_write_api):
|
|
"""Test the setup with minimal configuration and defaults."""
|
|
config = {"influxdb": {}}
|
|
config["influxdb"].update(config_ext)
|
|
|
|
assert await async_setup_component(hass, influxdb.DOMAIN, config)
|
|
await hass.async_block_till_done()
|
|
assert hass.bus.listen.called
|
|
assert EVENT_STATE_CHANGED == hass.bus.listen.call_args_list[0][0][0]
|
|
assert get_write_api(mock_client).call_count == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api",
|
|
[
|
|
(influxdb.DEFAULT_API_VERSION, {"username": "user"}, _get_write_api_mock_v1),
|
|
(influxdb.DEFAULT_API_VERSION, {"token": "token"}, _get_write_api_mock_v1),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
{"api_version": influxdb.API_VERSION_2, "organization": "organization"},
|
|
_get_write_api_mock_v2,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
{
|
|
"api_version": influxdb.API_VERSION_2,
|
|
"token": "token",
|
|
"organization": "organization",
|
|
"username": "user",
|
|
"password": "pass",
|
|
},
|
|
_get_write_api_mock_v2,
|
|
),
|
|
],
|
|
indirect=["mock_client"],
|
|
)
|
|
async def test_invalid_config(hass, mock_client, config_ext, get_write_api):
|
|
"""Test the setup with invalid config or config options specified for wrong version."""
|
|
config = {"influxdb": {}}
|
|
config["influxdb"].update(config_ext)
|
|
|
|
assert not await async_setup_component(hass, influxdb.DOMAIN, config)
|
|
|
|
|
|
async def _setup(hass, mock_influx_client, config_ext, get_write_api):
|
|
"""Prepare client for next test and return event handler method."""
|
|
config = {
|
|
"influxdb": {
|
|
"host": "host",
|
|
"exclude": {"entities": ["fake.blacklisted"], "domains": ["another_fake"]},
|
|
}
|
|
}
|
|
config["influxdb"].update(config_ext)
|
|
assert await async_setup_component(hass, influxdb.DOMAIN, config)
|
|
await hass.async_block_till_done()
|
|
# A call is made to the write API during setup to test the connection.
|
|
# Therefore we reset the write API mock here before the test begins.
|
|
get_write_api(mock_influx_client).reset_mock()
|
|
return hass.bus.listen.call_args_list[0][0][1]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener."""
|
|
handler_method = await _setup(hass, mock_client, config_ext, get_write_api)
|
|
|
|
# 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": f"99{UNIT_PERCENTAGE}",
|
|
"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 = MagicMock(
|
|
state=in_,
|
|
domain="fake",
|
|
entity_id="fake.entity-id",
|
|
object_id="entity",
|
|
attributes=attrs,
|
|
)
|
|
event = 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": f"99{UNIT_PERCENTAGE}",
|
|
"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]
|
|
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_no_units(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener for missing units."""
|
|
handler_method = await _setup(hass, mock_client, config_ext, get_write_api)
|
|
|
|
for unit in (None, ""):
|
|
if unit:
|
|
attrs = {"unit_of_measurement": unit}
|
|
else:
|
|
attrs = {}
|
|
state = MagicMock(
|
|
state=1,
|
|
domain="fake",
|
|
entity_id="fake.entity-id",
|
|
object_id="entity",
|
|
attributes=attrs,
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": "fake.entity-id",
|
|
"tags": {"domain": "fake", "entity_id": "entity"},
|
|
"time": 12345,
|
|
"fields": {"value": 1},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_inf(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener with large or invalid numbers."""
|
|
handler_method = await _setup(hass, mock_client, config_ext, get_write_api)
|
|
|
|
attrs = {"bignumstring": "9" * 999, "nonumstring": "nan"}
|
|
state = MagicMock(
|
|
state=8,
|
|
domain="fake",
|
|
entity_id="fake.entity-id",
|
|
object_id="entity",
|
|
attributes=attrs,
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": "fake.entity-id",
|
|
"tags": {"domain": "fake", "entity_id": "entity"},
|
|
"time": 12345,
|
|
"fields": {"value": 8},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_states(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener against ignored states."""
|
|
handler_method = await _setup(hass, mock_client, config_ext, get_write_api)
|
|
|
|
for state_state in (1, "unknown", "", "unavailable"):
|
|
state = MagicMock(
|
|
state=state_state,
|
|
domain="fake",
|
|
entity_id="fake.entity-id",
|
|
object_id="entity",
|
|
attributes={},
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": "fake.entity-id",
|
|
"tags": {"domain": "fake", "entity_id": "entity"},
|
|
"time": 12345,
|
|
"fields": {"value": 1},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
if state_state == 1:
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
else:
|
|
assert not write_api.called
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_blacklist(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener against a blacklist."""
|
|
handler_method = await _setup(hass, mock_client, config_ext, get_write_api)
|
|
|
|
for entity_id in ("ok", "blacklisted"):
|
|
state = MagicMock(
|
|
state=1,
|
|
domain="fake",
|
|
entity_id=f"fake.{entity_id}",
|
|
object_id=entity_id,
|
|
attributes={},
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": f"fake.{entity_id}",
|
|
"tags": {"domain": "fake", "entity_id": entity_id},
|
|
"time": 12345,
|
|
"fields": {"value": 1},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
if entity_id == "ok":
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
else:
|
|
assert not write_api.called
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_blacklist_domain(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener against a domain blacklist."""
|
|
handler_method = await _setup(hass, mock_client, config_ext, get_write_api)
|
|
|
|
for domain in ("ok", "another_fake"):
|
|
state = MagicMock(
|
|
state=1,
|
|
domain=domain,
|
|
entity_id=f"{domain}.something",
|
|
object_id="something",
|
|
attributes={},
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": f"{domain}.something",
|
|
"tags": {"domain": domain, "entity_id": "something"},
|
|
"time": 12345,
|
|
"fields": {"value": 1},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
if domain == "ok":
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
else:
|
|
assert not write_api.called
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_whitelist(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener against a whitelist."""
|
|
config = {"include": {"entities": ["fake.included"]}}
|
|
config.update(config_ext)
|
|
handler_method = await _setup(hass, mock_client, config, get_write_api)
|
|
|
|
for entity_id in ("included", "default"):
|
|
state = MagicMock(
|
|
state=1,
|
|
domain="fake",
|
|
entity_id=f"fake.{entity_id}",
|
|
object_id=entity_id,
|
|
attributes={},
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": f"fake.{entity_id}",
|
|
"tags": {"domain": "fake", "entity_id": entity_id},
|
|
"time": 12345,
|
|
"fields": {"value": 1},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
if entity_id == "included":
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
else:
|
|
assert not write_api.called
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_whitelist_domain(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener against a domain whitelist."""
|
|
config = {"include": {"domains": ["fake"]}}
|
|
config.update(config_ext)
|
|
handler_method = await _setup(hass, mock_client, config, get_write_api)
|
|
|
|
for domain in ("fake", "another_fake"):
|
|
state = MagicMock(
|
|
state=1,
|
|
domain=domain,
|
|
entity_id=f"{domain}.something",
|
|
object_id="something",
|
|
attributes={},
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": f"{domain}.something",
|
|
"tags": {"domain": domain, "entity_id": "something"},
|
|
"time": 12345,
|
|
"fields": {"value": 1},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
if domain == "fake":
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
else:
|
|
assert not write_api.called
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_whitelist_domain_and_entities(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener against a domain and entity whitelist."""
|
|
config = {"include": {"domains": ["fake"], "entities": ["other.one"]}}
|
|
config.update(config_ext)
|
|
handler_method = await _setup(hass, mock_client, config, get_write_api)
|
|
|
|
for domain in ("fake", "another_fake"):
|
|
state = MagicMock(
|
|
state=1,
|
|
domain=domain,
|
|
entity_id=f"{domain}.something",
|
|
object_id="something",
|
|
attributes={},
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": f"{domain}.something",
|
|
"tags": {"domain": domain, "entity_id": "something"},
|
|
"time": 12345,
|
|
"fields": {"value": 1},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
if domain == "fake":
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
else:
|
|
assert not write_api.called
|
|
write_api.reset_mock()
|
|
|
|
for entity_id in ("one", "two"):
|
|
state = MagicMock(
|
|
state=1,
|
|
domain="other",
|
|
entity_id=f"other.{entity_id}",
|
|
object_id=entity_id,
|
|
attributes={},
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": f"other.{entity_id}",
|
|
"tags": {"domain": "other", "entity_id": entity_id},
|
|
"time": 12345,
|
|
"fields": {"value": 1},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
if entity_id == "one":
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
else:
|
|
assert not write_api.called
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_invalid_type(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener when an attribute has an invalid type."""
|
|
handler_method = await _setup(hass, mock_client, config_ext, get_write_api)
|
|
|
|
# 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 = MagicMock(
|
|
state=in_,
|
|
domain="fake",
|
|
entity_id="fake.entity-id",
|
|
object_id="entity",
|
|
attributes=attrs,
|
|
)
|
|
event = 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]
|
|
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_default_measurement(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener with a default measurement."""
|
|
config = {"default_measurement": "state"}
|
|
config.update(config_ext)
|
|
handler_method = await _setup(hass, mock_client, config, get_write_api)
|
|
|
|
state = MagicMock(
|
|
state=1, domain="fake", entity_id="fake.ok", object_id="ok", attributes={},
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
body = [
|
|
{
|
|
"measurement": "state",
|
|
"tags": {"domain": "fake", "entity_id": "ok"},
|
|
"time": 12345,
|
|
"fields": {"value": 1},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_unit_of_measurement_field(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener for unit of measurement field."""
|
|
config = {"override_measurement": "state"}
|
|
config.update(config_ext)
|
|
handler_method = await _setup(hass, mock_client, config, get_write_api)
|
|
|
|
attrs = {"unit_of_measurement": "foobars"}
|
|
state = MagicMock(
|
|
state="foo",
|
|
domain="fake",
|
|
entity_id="fake.entity-id",
|
|
object_id="entity",
|
|
attributes=attrs,
|
|
)
|
|
event = 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"},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_tags_attributes(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener when some attributes should be tags."""
|
|
config = {"tags_attributes": ["friendly_fake"]}
|
|
config.update(config_ext)
|
|
handler_method = await _setup(hass, mock_client, config, get_write_api)
|
|
|
|
attrs = {"friendly_fake": "tag_str", "field_fake": "field_str"}
|
|
state = MagicMock(
|
|
state=1,
|
|
domain="fake",
|
|
entity_id="fake.something",
|
|
object_id="something",
|
|
attributes=attrs,
|
|
)
|
|
event = 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"},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_component_override_measurement(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener with overridden measurements."""
|
|
config = {
|
|
"component_config": {
|
|
"sensor.fake_humidity": {"override_measurement": "humidity"}
|
|
},
|
|
"component_config_glob": {
|
|
"binary_sensor.*motion": {"override_measurement": "motion"}
|
|
},
|
|
"component_config_domain": {"climate": {"override_measurement": "hvac"}},
|
|
}
|
|
config.update(config_ext)
|
|
handler_method = await _setup(hass, mock_client, config, get_write_api)
|
|
|
|
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 = MagicMock(
|
|
state=1,
|
|
domain=comp["domain"],
|
|
entity_id=f"{comp['domain']}.{comp['id']}",
|
|
object_id=comp["id"],
|
|
attributes={},
|
|
)
|
|
event = 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},
|
|
}
|
|
]
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
write_api = get_write_api(mock_client)
|
|
assert write_api.call_count == 1
|
|
assert write_api.call_args == get_mock_call(body)
|
|
write_api.reset_mock()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_scheduled_write(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener retries after a write failure."""
|
|
config = {"max_retries": 1}
|
|
config.update(config_ext)
|
|
handler_method = await _setup(hass, mock_client, config, get_write_api)
|
|
|
|
state = MagicMock(
|
|
state=1,
|
|
domain="fake",
|
|
entity_id="entity.id",
|
|
object_id="entity",
|
|
attributes={},
|
|
)
|
|
event = MagicMock(data={"new_state": state}, time_fired=12345)
|
|
write_api = get_write_api(mock_client)
|
|
write_api.side_effect = IOError("foo")
|
|
|
|
# Write fails
|
|
with patch.object(influxdb.time, "sleep") as mock_sleep:
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert mock_sleep.called
|
|
assert write_api.call_count == 2
|
|
|
|
# Write works again
|
|
write_api.side_effect = None
|
|
with patch.object(influxdb.time, "sleep") as mock_sleep:
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
assert not mock_sleep.called
|
|
assert write_api.call_count == 3
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mock_client, config_ext, get_write_api, get_mock_call",
|
|
[
|
|
(
|
|
influxdb.DEFAULT_API_VERSION,
|
|
BASE_V1_CONFIG,
|
|
_get_write_api_mock_v1,
|
|
influxdb.DEFAULT_API_VERSION,
|
|
),
|
|
(
|
|
influxdb.API_VERSION_2,
|
|
BASE_V2_CONFIG,
|
|
_get_write_api_mock_v2,
|
|
influxdb.API_VERSION_2,
|
|
),
|
|
],
|
|
indirect=["mock_client", "get_mock_call"],
|
|
)
|
|
async def test_event_listener_backlog_full(
|
|
hass, mock_client, config_ext, get_write_api, get_mock_call
|
|
):
|
|
"""Test the event listener drops old events when backlog gets full."""
|
|
handler_method = await _setup(hass, mock_client, config_ext, get_write_api)
|
|
|
|
state = MagicMock(
|
|
state=1,
|
|
domain="fake",
|
|
entity_id="entity.id",
|
|
object_id="entity",
|
|
attributes={},
|
|
)
|
|
event = 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 patch("homeassistant.components.influxdb.time.monotonic", new=fast_monotonic):
|
|
handler_method(event)
|
|
hass.data[influxdb.DOMAIN].block_till_done()
|
|
|
|
assert get_write_api(mock_client).call_count == 0
|