From 3f948e027a9304bd7927af1c3e3b61dd25b9c292 Mon Sep 17 00:00:00 2001 From: Julian Engelhardt Date: Wed, 27 Jan 2021 22:37:59 +0100 Subject: [PATCH] Clean tcp tests (#41673) Co-authored-by: Martin Hjelmare --- homeassistant/components/tcp/sensor.py | 5 +- tests/components/tcp/test_binary_sensor.py | 117 ++++---- tests/components/tcp/test_sensor.py | 302 +++++++-------------- 3 files changed, 170 insertions(+), 254 deletions(-) diff --git a/homeassistant/components/tcp/sensor.py b/homeassistant/components/tcp/sensor.py index 868cd9b8557..9b7e1539fb4 100644 --- a/homeassistant/components/tcp/sensor.py +++ b/homeassistant/components/tcp/sensor.py @@ -78,10 +78,7 @@ class TcpSensor(Entity): @property def name(self): """Return the name of this sensor.""" - name = self._config[CONF_NAME] - if name is not None: - return name - return super().name + return self._config[CONF_NAME] @property def state(self): diff --git a/tests/components/tcp/test_binary_sensor.py b/tests/components/tcp/test_binary_sensor.py index 2dc16ad79c7..21dd84b1892 100644 --- a/tests/components/tcp/test_binary_sensor.py +++ b/tests/components/tcp/test_binary_sensor.py @@ -1,62 +1,83 @@ """The tests for the TCP binary sensor platform.""" -import unittest -from unittest.mock import Mock, patch +from datetime import timedelta +from unittest.mock import call, patch -from homeassistant.components.tcp import binary_sensor as bin_tcp -import homeassistant.components.tcp.sensor as tcp -from homeassistant.setup import setup_component +import pytest -from tests.common import assert_setup_component, get_test_home_assistant +from homeassistant.const import STATE_OFF, STATE_ON +from homeassistant.setup import async_setup_component +from homeassistant.util.dt import utcnow + +from tests.common import assert_setup_component, async_fire_time_changed import tests.components.tcp.test_sensor as test_tcp +BINARY_SENSOR_CONFIG = test_tcp.TEST_CONFIG["sensor"] +TEST_CONFIG = {"binary_sensor": BINARY_SENSOR_CONFIG} +TEST_ENTITY = "binary_sensor.test_name" -class TestTCPBinarySensor(unittest.TestCase): - """Test the TCP Binary Sensor.""" - def setup_method(self, method): - """Set up things to be run when tests are started.""" - self.hass = get_test_home_assistant() +@pytest.fixture(name="mock_socket") +def mock_socket_fixture(): + """Mock the socket.""" + with patch( + "homeassistant.components.tcp.sensor.socket.socket" + ) as mock_socket, patch( + "homeassistant.components.tcp.sensor.select.select", + return_value=(True, False, False), + ): + # yield the return value of the socket context manager + yield mock_socket.return_value.__enter__.return_value - def teardown_method(self, method): - """Stop down everything that was started.""" - self.hass.stop() - def test_setup_platform_valid_config(self): - """Check a valid configuration.""" - with assert_setup_component(0, "binary_sensor"): - assert setup_component(self.hass, "binary_sensor", test_tcp.TEST_CONFIG) +@pytest.fixture +def now(): + """Return datetime UTC now.""" + return utcnow() - def test_setup_platform_invalid_config(self): - """Check the invalid configuration.""" - with assert_setup_component(0): - assert setup_component( - self.hass, - "binary_sensor", - {"binary_sensor": {"platform": "tcp", "porrt": 1234}}, - ) - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_setup_platform_devices(self, mock_update): - """Check the supplied config and call add_entities with sensor.""" - add_entities = Mock() - ret = bin_tcp.setup_platform(None, test_tcp.TEST_CONFIG, add_entities) - assert ret is None - assert add_entities.called - assert isinstance(add_entities.call_args[0][0][0], bin_tcp.TcpBinarySensor) +async def test_setup_platform_valid_config(hass, mock_socket): + """Check a valid configuration.""" + with assert_setup_component(1, "binary_sensor"): + assert await async_setup_component(hass, "binary_sensor", TEST_CONFIG) + await hass.async_block_till_done() - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_is_on_true(self, mock_update): - """Check the return that _state is value_on.""" - sensor = bin_tcp.TcpBinarySensor(self.hass, test_tcp.TEST_CONFIG["sensor"]) - sensor._state = test_tcp.TEST_CONFIG["sensor"][tcp.CONF_VALUE_ON] - print(sensor._state) - assert sensor.is_on - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_is_on_false(self, mock_update): - """Check the return that _state is not the same as value_on.""" - sensor = bin_tcp.TcpBinarySensor(self.hass, test_tcp.TEST_CONFIG["sensor"]) - sensor._state = "{} abc".format( - test_tcp.TEST_CONFIG["sensor"][tcp.CONF_VALUE_ON] +async def test_setup_platform_invalid_config(hass, mock_socket): + """Check the invalid configuration.""" + with assert_setup_component(0): + assert await async_setup_component( + hass, + "binary_sensor", + {"binary_sensor": {"platform": "tcp", "porrt": 1234}}, ) - assert not sensor.is_on + await hass.async_block_till_done() + + +async def test_state(hass, mock_socket, now): + """Check the state and update of the binary sensor.""" + mock_socket.recv.return_value = b"off" + assert await async_setup_component(hass, "binary_sensor", TEST_CONFIG) + await hass.async_block_till_done() + + state = hass.states.get(TEST_ENTITY) + + assert state + assert state.state == STATE_OFF + assert mock_socket.connect.called + assert mock_socket.connect.call_args == call( + (BINARY_SENSOR_CONFIG["host"], BINARY_SENSOR_CONFIG["port"]) + ) + assert mock_socket.send.called + assert mock_socket.send.call_args == call(BINARY_SENSOR_CONFIG["payload"].encode()) + assert mock_socket.recv.called + assert mock_socket.recv.call_args == call(BINARY_SENSOR_CONFIG["buffer_size"]) + + mock_socket.recv.return_value = b"on" + + async_fire_time_changed(hass, now + timedelta(seconds=45)) + await hass.async_block_till_done() + + state = hass.states.get(TEST_ENTITY) + + assert state + assert state.state == STATE_ON diff --git a/tests/components/tcp/test_sensor.py b/tests/components/tcp/test_sensor.py index 8e79d4e514d..b1efef305bf 100644 --- a/tests/components/tcp/test_sensor.py +++ b/tests/components/tcp/test_sensor.py @@ -1,16 +1,13 @@ """The tests for the TCP sensor platform.""" from copy import copy -import socket -import unittest -from unittest.mock import Mock, patch -from uuid import uuid4 +from unittest.mock import call, patch + +import pytest import homeassistant.components.tcp.sensor as tcp -from homeassistant.helpers.entity import Entity -from homeassistant.helpers.template import Template -from homeassistant.setup import setup_component +from homeassistant.setup import async_setup_component -from tests.common import assert_setup_component, get_test_home_assistant +from tests.common import assert_setup_component TEST_CONFIG = { "sensor": { @@ -21,13 +18,16 @@ TEST_CONFIG = { tcp.CONF_TIMEOUT: tcp.DEFAULT_TIMEOUT + 1, tcp.CONF_PAYLOAD: "test_payload", tcp.CONF_UNIT_OF_MEASUREMENT: "test_unit", - tcp.CONF_VALUE_TEMPLATE: Template("test_template"), + tcp.CONF_VALUE_TEMPLATE: "{{ 'test_' + value }}", tcp.CONF_VALUE_ON: "test_on", tcp.CONF_BUFFER_SIZE: tcp.DEFAULT_BUFFER_SIZE + 1, } } +SENSOR_TEST_CONFIG = TEST_CONFIG["sensor"] +TEST_ENTITY = "sensor.test_name" KEYS_AND_DEFAULTS = { + tcp.CONF_NAME: tcp.DEFAULT_NAME, tcp.CONF_TIMEOUT: tcp.DEFAULT_TIMEOUT, tcp.CONF_UNIT_OF_MEASUREMENT: None, tcp.CONF_VALUE_TEMPLATE: None, @@ -35,229 +35,127 @@ KEYS_AND_DEFAULTS = { tcp.CONF_BUFFER_SIZE: tcp.DEFAULT_BUFFER_SIZE, } +socket_test_value = "value" -class TestTCPSensor(unittest.TestCase): - """Test the TCP Sensor.""" - def setup_method(self, method): - """Set up things to be run when tests are started.""" - self.hass = get_test_home_assistant() +@pytest.fixture(name="mock_socket") +def mock_socket_fixture(mock_select): + """Mock socket.""" + with patch("homeassistant.components.tcp.sensor.socket.socket") as mock_socket: + socket_instance = mock_socket.return_value.__enter__.return_value + socket_instance.recv.return_value = socket_test_value.encode() + yield socket_instance - def teardown_method(self, method): - """Stop everything that was started.""" - self.hass.stop() - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_setup_platform_valid_config(self, mock_update): - """Check a valid configuration and call add_entities with sensor.""" - with assert_setup_component(0, "sensor"): - assert setup_component(self.hass, "sensor", TEST_CONFIG) +@pytest.fixture(name="mock_select") +def mock_select_fixture(): + """Mock select.""" + with patch( + "homeassistant.components.tcp.sensor.select.select", + return_value=(True, False, False), + ) as mock_select: + yield mock_select - add_entities = Mock() - tcp.setup_platform(None, TEST_CONFIG["sensor"], add_entities) - assert add_entities.called - assert isinstance(add_entities.call_args[0][0][0], tcp.TcpSensor) - def test_setup_platform_invalid_config(self): - """Check an invalid configuration.""" - with assert_setup_component(0): - assert setup_component( - self.hass, "sensor", {"sensor": {"platform": "tcp", "porrt": 1234}} - ) +async def test_setup_platform_valid_config(hass, mock_socket): + """Check a valid configuration and call add_entities with sensor.""" + with assert_setup_component(1, "sensor"): + assert await async_setup_component(hass, "sensor", TEST_CONFIG) + await hass.async_block_till_done() - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_name(self, mock_update): - """Return the name if set in the configuration.""" - sensor = tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - assert sensor.name == TEST_CONFIG["sensor"][tcp.CONF_NAME] - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_name_not_set(self, mock_update): - """Return the superclass name property if not set in configuration.""" - config = copy(TEST_CONFIG["sensor"]) - del config[tcp.CONF_NAME] - entity = Entity() - sensor = tcp.TcpSensor(self.hass, config) - assert sensor.name == entity.name - - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_state(self, mock_update): - """Return the contents of _state.""" - sensor = tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - uuid = str(uuid4()) - sensor._state = uuid - assert sensor.state == uuid - - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_unit_of_measurement(self, mock_update): - """Return the configured unit of measurement.""" - sensor = tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - assert ( - sensor.unit_of_measurement - == TEST_CONFIG["sensor"][tcp.CONF_UNIT_OF_MEASUREMENT] +async def test_setup_platform_invalid_config(hass, mock_socket): + """Check an invalid configuration.""" + with assert_setup_component(0): + assert await async_setup_component( + hass, "sensor", {"sensor": {"platform": "tcp", "porrt": 1234}} ) + await hass.async_block_till_done() - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_config_valid_keys(self, *args): - """Store valid keys in _config.""" - sensor = tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - del TEST_CONFIG["sensor"]["platform"] - for key in TEST_CONFIG["sensor"]: - assert key in sensor._config +async def test_state(hass, mock_socket, mock_select): + """Return the contents of _state.""" + assert await async_setup_component(hass, "sensor", TEST_CONFIG) + await hass.async_block_till_done() - def test_validate_config_valid_keys(self): - """Return True when provided with the correct keys.""" - with assert_setup_component(0, "sensor"): - assert setup_component(self.hass, "sensor", TEST_CONFIG) + state = hass.states.get(TEST_ENTITY) - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_config_invalid_keys(self, mock_update): - """Shouldn't store invalid keys in _config.""" - config = copy(TEST_CONFIG["sensor"]) - config.update({"a": "test_a", "b": "test_b", "c": "test_c"}) - sensor = tcp.TcpSensor(self.hass, config) - for invalid_key in "abc": - assert invalid_key not in sensor._config + assert state + assert state.state == "test_value" + assert ( + state.attributes["unit_of_measurement"] + == SENSOR_TEST_CONFIG[tcp.CONF_UNIT_OF_MEASUREMENT] + ) + assert mock_socket.connect.called + assert mock_socket.connect.call_args == call( + (SENSOR_TEST_CONFIG["host"], SENSOR_TEST_CONFIG["port"]) + ) + assert mock_socket.send.called + assert mock_socket.send.call_args == call(SENSOR_TEST_CONFIG["payload"].encode()) + assert mock_select.call_args == call( + [mock_socket], [], [], SENSOR_TEST_CONFIG[tcp.CONF_TIMEOUT] + ) + assert mock_socket.recv.called + assert mock_socket.recv.call_args == call(SENSOR_TEST_CONFIG["buffer_size"]) - def test_validate_config_invalid_keys(self): - """Test with invalid keys plus some extra.""" - config = copy(TEST_CONFIG["sensor"]) - config.update({"a": "test_a", "b": "test_b", "c": "test_c"}) - with assert_setup_component(0, "sensor"): - assert setup_component(self.hass, "sensor", {"tcp": config}) - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_config_uses_defaults(self, mock_update): - """Check if defaults were set.""" - config = copy(TEST_CONFIG["sensor"]) +async def test_config_uses_defaults(hass, mock_socket): + """Check if defaults were set.""" + config = copy(SENSOR_TEST_CONFIG) - for key in KEYS_AND_DEFAULTS: - del config[key] + for key in KEYS_AND_DEFAULTS: + del config[key] - with assert_setup_component(1) as result_config: - assert setup_component(self.hass, "sensor", {"sensor": config}) + with assert_setup_component(1) as result_config: + assert await async_setup_component(hass, "sensor", {"sensor": config}) + await hass.async_block_till_done() - sensor = tcp.TcpSensor(self.hass, result_config["sensor"][0]) + state = hass.states.get("sensor.tcp_sensor") - for key, default in KEYS_AND_DEFAULTS.items(): - assert sensor._config[key] == default + assert state + assert state.state == "value" - def test_validate_config_missing_defaults(self): - """Return True when defaulted keys are not provided.""" - config = copy(TEST_CONFIG["sensor"]) + for key, default in KEYS_AND_DEFAULTS.items(): + assert result_config["sensor"][0].get(key) == default - for key in KEYS_AND_DEFAULTS: - del config[key] - with assert_setup_component(0, "sensor"): - assert setup_component(self.hass, "sensor", {"tcp": config}) +@pytest.mark.parametrize("sock_attr", ["connect", "send"]) +async def test_update_socket_error(hass, mock_socket, sock_attr): + """Test socket errors during update.""" + socket_method = getattr(mock_socket, sock_attr) + socket_method.side_effect = OSError("Boom") - def test_validate_config_missing_required(self): - """Return False when required config items are missing.""" - for key in TEST_CONFIG["sensor"]: - if key in KEYS_AND_DEFAULTS: - continue - config = copy(TEST_CONFIG["sensor"]) - del config[key] - with assert_setup_component(0, "sensor"): - assert setup_component(self.hass, "sensor", {"tcp": config}) + assert await async_setup_component(hass, "sensor", TEST_CONFIG) + await hass.async_block_till_done() - @patch("homeassistant.components.tcp.sensor.TcpSensor.update") - def test_init_calls_update(self, mock_update): - """Call update() method during __init__().""" - tcp.TcpSensor(self.hass, TEST_CONFIG) - assert mock_update.called + state = hass.states.get(TEST_ENTITY) - @patch("socket.socket") - @patch("select.select", return_value=(True, False, False)) - def test_update_connects_to_host_and_port(self, mock_select, mock_socket): - """Connect to the configured host and port.""" - tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - mock_socket = mock_socket().__enter__() - assert mock_socket.connect.mock_calls[0][1] == ( - ( - TEST_CONFIG["sensor"][tcp.CONF_HOST], - TEST_CONFIG["sensor"][tcp.CONF_PORT], - ), - ) + assert state + assert state.state == "unknown" - @patch("socket.socket.connect", side_effect=socket.error()) - def test_update_returns_if_connecting_fails(self, *args): - """Return if connecting to host fails.""" - with patch("homeassistant.components.tcp.sensor.TcpSensor.update"): - sensor = tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - assert sensor.update() is None - @patch("socket.socket.connect") - @patch("socket.socket.send", side_effect=socket.error()) - def test_update_returns_if_sending_fails(self, *args): - """Return if sending fails.""" - with patch("homeassistant.components.tcp.sensor.TcpSensor.update"): - sensor = tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - assert sensor.update() is None +async def test_update_select_fails(hass, mock_socket, mock_select): + """Test select fails to return a socket for reading.""" + mock_select.return_value = (False, False, False) - @patch("socket.socket.connect") - @patch("socket.socket.send") - @patch("select.select", return_value=(False, False, False)) - def test_update_returns_if_select_fails(self, *args): - """Return if select fails to return a socket.""" - with patch("homeassistant.components.tcp.sensor.TcpSensor.update"): - sensor = tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - assert sensor.update() is None + assert await async_setup_component(hass, "sensor", TEST_CONFIG) + await hass.async_block_till_done() - @patch("socket.socket") - @patch("select.select", return_value=(True, False, False)) - def test_update_sends_payload(self, mock_select, mock_socket): - """Send the configured payload as bytes.""" - tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - mock_socket = mock_socket().__enter__() - mock_socket.send.assert_called_with( - TEST_CONFIG["sensor"][tcp.CONF_PAYLOAD].encode() - ) + state = hass.states.get(TEST_ENTITY) - @patch("socket.socket") - @patch("select.select", return_value=(True, False, False)) - def test_update_calls_select_with_timeout(self, mock_select, mock_socket): - """Provide the timeout argument to select.""" - tcp.TcpSensor(self.hass, TEST_CONFIG["sensor"]) - mock_socket = mock_socket().__enter__() - mock_select.assert_called_with( - [mock_socket], [], [], TEST_CONFIG["sensor"][tcp.CONF_TIMEOUT] - ) + assert state + assert state.state == "unknown" - @patch("socket.socket") - @patch("select.select", return_value=(True, False, False)) - def test_update_receives_packet_and_sets_as_state(self, mock_select, mock_socket): - """Test the response from the socket and set it as the state.""" - test_value = "test_value" - mock_socket = mock_socket().__enter__() - mock_socket.recv.return_value = test_value.encode() - config = copy(TEST_CONFIG["sensor"]) - del config[tcp.CONF_VALUE_TEMPLATE] - sensor = tcp.TcpSensor(self.hass, config) - assert sensor._state == test_value - @patch("socket.socket") - @patch("select.select", return_value=(True, False, False)) - def test_update_renders_value_in_template(self, mock_select, mock_socket): - """Render the value in the provided template.""" - test_value = "test_value" - mock_socket = mock_socket().__enter__() - mock_socket.recv.return_value = test_value.encode() - config = copy(TEST_CONFIG["sensor"]) - config[tcp.CONF_VALUE_TEMPLATE] = Template("{{ value }} {{ 1+1 }}") - sensor = tcp.TcpSensor(self.hass, config) - assert sensor._state == "%s 2" % test_value +async def test_update_returns_if_template_render_fails(hass, mock_socket): + """Return None if rendering the template fails.""" + config = copy(SENSOR_TEST_CONFIG) + config[tcp.CONF_VALUE_TEMPLATE] = "{{ value / 0 }}" - @patch("socket.socket") - @patch("select.select", return_value=(True, False, False)) - def test_update_returns_if_template_render_fails(self, mock_select, mock_socket): - """Return None if rendering the template fails.""" - test_value = "test_value" - mock_socket = mock_socket().__enter__() - mock_socket.recv.return_value = test_value.encode() - config = copy(TEST_CONFIG["sensor"]) - config[tcp.CONF_VALUE_TEMPLATE] = Template("{{ this won't work") - sensor = tcp.TcpSensor(self.hass, config) - assert sensor.update() is None + assert await async_setup_component(hass, "sensor", {"sensor": config}) + await hass.async_block_till_done() + + state = hass.states.get(TEST_ENTITY) + + assert state + assert state.state == "unknown"