Rewrite statsd unittest tests to pytest style test functions (#41223)

This commit is contained in:
Julian Engelhardt 2020-10-06 11:24:16 +02:00 committed by GitHub
parent 014fe48096
commit 14273e9e48
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -1,5 +1,4 @@
"""The tests for the StatsD feeder.""" """The tests for the StatsD feeder."""
import unittest
from unittest import mock from unittest import mock
import pytest import pytest
@ -8,133 +7,128 @@ import voluptuous as vol
import homeassistant.components.statsd as statsd import homeassistant.components.statsd as statsd
from homeassistant.const import EVENT_STATE_CHANGED, STATE_OFF, STATE_ON from homeassistant.const import EVENT_STATE_CHANGED, STATE_OFF, STATE_ON
import homeassistant.core as ha import homeassistant.core as ha
from homeassistant.setup import setup_component from homeassistant.setup import async_setup_component
from tests.common import get_test_home_assistant from tests.async_mock import MagicMock, patch
class TestStatsd(unittest.TestCase): @pytest.fixture
"""Test the StatsD component.""" def mock_client():
"""Pytest fixture for statsd library."""
with patch("statsd.StatsClient") as mock_client:
yield mock_client.return_value
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): def test_invalid_config():
"""Stop everything that was started.""" """Test configuration with defaults."""
self.hass.stop() config = {"statsd": {"host1": "host1"}}
def test_invalid_config(self): with pytest.raises(vol.Invalid):
"""Test configuration with defaults.""" statsd.CONFIG_SCHEMA(None)
config = {"statsd": {"host1": "host1"}} with pytest.raises(vol.Invalid):
statsd.CONFIG_SCHEMA(config)
with pytest.raises(vol.Invalid):
statsd.CONFIG_SCHEMA(None)
with pytest.raises(vol.Invalid):
statsd.CONFIG_SCHEMA(config)
@mock.patch("statsd.StatsClient") async def test_statsd_setup_full(hass):
def test_statsd_setup_full(self, mock_connection): """Test setup with all data."""
"""Test setup with all data.""" config = {"statsd": {"host": "host", "port": 123, "rate": 1, "prefix": "foo"}}
config = {"statsd": {"host": "host", "port": 123, "rate": 1, "prefix": "foo"}} hass.bus.listen = MagicMock()
self.hass.bus.listen = mock.MagicMock() with patch("statsd.StatsClient") as mock_init:
assert setup_component(self.hass, statsd.DOMAIN, config) assert await async_setup_component(hass, statsd.DOMAIN, config)
assert mock_connection.call_count == 1
assert mock_connection.call_args == mock.call( assert mock_init.call_count == 1
host="host", port=123, prefix="foo" assert mock_init.call_args == mock.call(host="host", port=123, prefix="foo")
assert hass.bus.listen.called
assert EVENT_STATE_CHANGED == hass.bus.listen.call_args_list[0][0][0]
async def test_statsd_setup_defaults(hass):
"""Test setup with defaults."""
config = {"statsd": {"host": "host"}}
config["statsd"][statsd.CONF_PORT] = statsd.DEFAULT_PORT
config["statsd"][statsd.CONF_PREFIX] = statsd.DEFAULT_PREFIX
hass.bus.listen = MagicMock()
with patch("statsd.StatsClient") as mock_init:
assert await async_setup_component(hass, statsd.DOMAIN, config)
assert mock_init.call_count == 1
assert mock_init.call_args == mock.call(host="host", port=8125, prefix="hass")
assert hass.bus.listen.called
async def test_event_listener_defaults(hass, mock_client):
"""Test event listener."""
config = {"statsd": {"host": "host", "value_mapping": {"custom": 3}}}
config["statsd"][statsd.CONF_RATE] = statsd.DEFAULT_RATE
hass.bus.listen = MagicMock()
await async_setup_component(hass, statsd.DOMAIN, config)
assert hass.bus.listen.called
handler_method = hass.bus.listen.call_args_list[0][0][1]
valid = {"1": 1, "1.0": 1.0, "custom": 3, STATE_ON: 1, STATE_OFF: 0}
for in_, out in valid.items():
state = MagicMock(state=in_, attributes={"attribute key": 3.2})
handler_method(MagicMock(data={"new_state": state}))
mock_client.gauge.assert_has_calls(
[mock.call(state.entity_id, out, statsd.DEFAULT_RATE)]
) )
assert self.hass.bus.listen.called mock_client.gauge.reset_mock()
assert EVENT_STATE_CHANGED == self.hass.bus.listen.call_args_list[0][0][0]
@mock.patch("statsd.StatsClient") assert mock_client.incr.call_count == 1
def test_statsd_setup_defaults(self, mock_connection): assert mock_client.incr.call_args == mock.call(
"""Test setup with defaults.""" state.entity_id, rate=statsd.DEFAULT_RATE
config = {"statsd": {"host": "host"}}
config["statsd"][statsd.CONF_PORT] = statsd.DEFAULT_PORT
config["statsd"][statsd.CONF_PREFIX] = statsd.DEFAULT_PREFIX
self.hass.bus.listen = mock.MagicMock()
assert setup_component(self.hass, statsd.DOMAIN, config)
assert mock_connection.call_count == 1
assert mock_connection.call_args == mock.call(
host="host", port=8125, prefix="hass"
) )
assert self.hass.bus.listen.called mock_client.incr.reset_mock()
@mock.patch("statsd.StatsClient") for invalid in ("foo", "", object):
def test_event_listener_defaults(self, mock_client): handler_method(
"""Test event listener.""" MagicMock(data={"new_state": ha.State("domain.test", invalid, {})})
config = {"statsd": {"host": "host", "value_mapping": {"custom": 3}}} )
assert not mock_client.gauge.called
assert mock_client.incr.called
config["statsd"][statsd.CONF_RATE] = statsd.DEFAULT_RATE
self.hass.bus.listen = mock.MagicMock() async def test_event_listener_attr_details(hass, mock_client):
setup_component(self.hass, statsd.DOMAIN, config) """Test event listener."""
assert self.hass.bus.listen.called config = {"statsd": {"host": "host", "log_attributes": True}}
handler_method = self.hass.bus.listen.call_args_list[0][0][1]
valid = {"1": 1, "1.0": 1.0, "custom": 3, STATE_ON: 1, STATE_OFF: 0} config["statsd"][statsd.CONF_RATE] = statsd.DEFAULT_RATE
for in_, out in valid.items():
state = mock.MagicMock(state=in_, attributes={"attribute key": 3.2})
handler_method(mock.MagicMock(data={"new_state": state}))
mock_client.return_value.gauge.assert_has_calls(
[mock.call(state.entity_id, out, statsd.DEFAULT_RATE)]
)
mock_client.return_value.gauge.reset_mock() hass.bus.listen = MagicMock()
await async_setup_component(hass, statsd.DOMAIN, config)
assert hass.bus.listen.called
handler_method = hass.bus.listen.call_args_list[0][0][1]
assert mock_client.return_value.incr.call_count == 1 valid = {"1": 1, "1.0": 1.0, STATE_ON: 1, STATE_OFF: 0}
assert mock_client.return_value.incr.call_args == mock.call( for in_, out in valid.items():
state.entity_id, rate=statsd.DEFAULT_RATE state = MagicMock(state=in_, attributes={"attribute key": 3.2})
) handler_method(MagicMock(data={"new_state": state}))
mock_client.return_value.incr.reset_mock() mock_client.gauge.assert_has_calls(
[
mock.call("%s.state" % state.entity_id, out, statsd.DEFAULT_RATE),
mock.call(
"%s.attribute_key" % state.entity_id, 3.2, statsd.DEFAULT_RATE
),
]
)
for invalid in ("foo", "", object): mock_client.gauge.reset_mock()
handler_method(
mock.MagicMock(data={"new_state": ha.State("domain.test", invalid, {})})
)
assert not mock_client.return_value.gauge.called
assert mock_client.return_value.incr.called
@mock.patch("statsd.StatsClient") assert mock_client.incr.call_count == 1
def test_event_listener_attr_details(self, mock_client): assert mock_client.incr.call_args == mock.call(
"""Test event listener.""" state.entity_id, rate=statsd.DEFAULT_RATE
config = {"statsd": {"host": "host", "log_attributes": True}} )
mock_client.incr.reset_mock()
config["statsd"][statsd.CONF_RATE] = statsd.DEFAULT_RATE for invalid in ("foo", "", object):
handler_method(
self.hass.bus.listen = mock.MagicMock() MagicMock(data={"new_state": ha.State("domain.test", invalid, {})})
setup_component(self.hass, statsd.DOMAIN, config) )
assert self.hass.bus.listen.called assert not mock_client.gauge.called
handler_method = self.hass.bus.listen.call_args_list[0][0][1] assert mock_client.incr.called
valid = {"1": 1, "1.0": 1.0, STATE_ON: 1, STATE_OFF: 0}
for in_, out in valid.items():
state = mock.MagicMock(state=in_, attributes={"attribute key": 3.2})
handler_method(mock.MagicMock(data={"new_state": state}))
mock_client.return_value.gauge.assert_has_calls(
[
mock.call("%s.state" % state.entity_id, out, statsd.DEFAULT_RATE),
mock.call(
"%s.attribute_key" % state.entity_id, 3.2, statsd.DEFAULT_RATE
),
]
)
mock_client.return_value.gauge.reset_mock()
assert mock_client.return_value.incr.call_count == 1
assert mock_client.return_value.incr.call_args == mock.call(
state.entity_id, rate=statsd.DEFAULT_RATE
)
mock_client.return_value.incr.reset_mock()
for invalid in ("foo", "", object):
handler_method(
mock.MagicMock(data={"new_state": ha.State("domain.test", invalid, {})})
)
assert not mock_client.return_value.gauge.called
assert mock_client.return_value.incr.called