evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Migrate to voluptuous (#3748)

Browse source
This commit is contained in:
Fabian Affolter 2016-10-22 06:14:35 +02:00 • committed by Paulus Schoutsen
parent b821a82417
commit fae620f3b3
4 changed files with 253 additions and 224 deletions
Download patch file Download diff file Expand all files Collapse all files

19
homeassistant/components/binary_sensor/tcp.py
View file

@ -7,21 +7,20 @@ https://home-assistant.io/components/binary_sensor.tcp/
import logging
from homeassistant.components.binary_sensor import BinarySensorDevice
from homeassistant.components.sensor.tcp import Sensor, CONF_VALUE_ON
from homeassistant.components.sensor.tcp import (
TcpSensor, CONF_VALUE_ON, PLATFORM_SCHEMA)
_LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Create the binary sensor."""
if not BinarySensor.validate_config(config):
return False
add_entities((BinarySensor(hass, config),))
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({})
class BinarySensor(BinarySensorDevice, Sensor):
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the TCP binary sensor."""
add_devices([TcpBinarySensor(hass, config)])
class TcpBinarySensor(BinarySensorDevice, TcpSensor):
"""A binary sensor which is on when its state == CONF_VALUE_ON."""
required = (CONF_VALUE_ON,)

77
homeassistant/components/sensor/tcp.py
View file

@ -8,33 +8,46 @@ import logging
import socket
import select
from homeassistant.const import CONF_NAME, CONF_HOST
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import (
CONF_NAME, CONF_HOST, CONF_PORT, CONF_PAYLOAD, CONF_TIMEOUT,
CONF_UNIT_OF_MEASUREMENT, CONF_VALUE_TEMPLATE)
from homeassistant.exceptions import TemplateError
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.template import Template
CONF_PORT = "port"
CONF_TIMEOUT = "timeout"
CONF_PAYLOAD = "payload"
CONF_UNIT = "unit"
CONF_VALUE_TEMPLATE = "value_template"
CONF_VALUE_ON = "value_on"
CONF_BUFFER_SIZE = "buffer_size"
DEFAULT_TIMEOUT = 10
DEFAULT_BUFFER_SIZE = 1024
import homeassistant.helpers.config_validation as cv
_LOGGER = logging.getLogger(__name__)
CONF_BUFFER_SIZE = 'buffer_size'
CONF_VALUE_ON = 'value_on'
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Setup the TCP Sensor."""
if not Sensor.validate_config(config):
return False
add_entities((Sensor(hass, config),))
DEFAULT_BUFFER_SIZE = 1024
DEFAULT_NAME = 'TCP Sensor'
DEFAULT_TIMEOUT = 10
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PORT): cv.port,
vol.Required(CONF_PAYLOAD): cv.string,
vol.Optional(CONF_BUFFER_SIZE, default=DEFAULT_BUFFER_SIZE):
cv.positive_int,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
vol.Optional(CONF_UNIT_OF_MEASUREMENT): cv.string,
vol.Optional(CONF_VALUE_ON): cv.string,
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
})
class Sensor(Entity):
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the TCP Sensor."""
add_devices([TcpSensor(hass, config)])
class TcpSensor(Entity):
"""Implementation of a TCP socket based sensor."""
required = tuple()
@ -49,37 +62,25 @@ class Sensor(Entity):
self._hass = hass
self._config = {
CONF_NAME: config.get(CONF_NAME),
CONF_HOST: config[CONF_HOST],
CONF_PORT: config[CONF_PORT],
CONF_TIMEOUT: config.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
CONF_PAYLOAD: config[CONF_PAYLOAD],
CONF_UNIT: config.get(CONF_UNIT),
CONF_HOST: config.get(CONF_HOST),
CONF_PORT: config.get(CONF_PORT),
CONF_TIMEOUT: config.get(CONF_TIMEOUT),
CONF_PAYLOAD: config.get(CONF_PAYLOAD),
CONF_UNIT_OF_MEASUREMENT: config.get(CONF_UNIT_OF_MEASUREMENT),
CONF_VALUE_TEMPLATE: value_template,
CONF_VALUE_ON: config.get(CONF_VALUE_ON),
CONF_BUFFER_SIZE: config.get(
CONF_BUFFER_SIZE, DEFAULT_BUFFER_SIZE),
CONF_BUFFER_SIZE: config.get(CONF_BUFFER_SIZE),
}
self._state = None
self.update()
@classmethod
def validate_config(cls, config):
"""Ensure the configuration has all of the necessary values."""
always_required = (CONF_HOST, CONF_PORT, CONF_PAYLOAD)
for key in always_required + tuple(cls.required):
if key not in config:
_LOGGER.error(
"You must provide %r to create any TCP entity.", key)
return False
return True
@property
def name(self):
"""Return the name of this sensor."""
name = self._config[CONF_NAME]
if name is not None:
return name
return super(Sensor, self).name
return super(TcpSensor, self).name
@property
def state(self):
@ -89,7 +90,7 @@ class Sensor(Entity):
@property
def unit_of_measurement(self):
"""Return the unit of measurement of this entity."""
return self._config[CONF_UNIT]
return self._config[CONF_UNIT_OF_MEASUREMENT]
def update(self):
"""Get the latest value for this sensor."""

86
tests/components/binary_sensor/test_tcp.py
View file

@ -1,59 +1,65 @@
"""The tests for the TCP binary sensor platform."""
from copy import copy
import unittest
from unittest.mock import patch, Mock
from homeassistant.components.sensor import tcp
from homeassistant.bootstrap import setup_component
from homeassistant.components.binary_sensor import tcp as bin_tcp
from tests.common import get_test_home_assistant
from homeassistant.components.sensor import tcp
from tests.common import (get_test_home_assistant, assert_setup_component)
from tests.components.sensor import test_tcp
@patch('homeassistant.components.sensor.tcp.Sensor.update')
def test_setup_platform_valid_config(mock_update):
"""Should 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, "setup_platform() should return None if successful."
assert add_entities.called
assert isinstance(add_entities.call_args[0][0][0], bin_tcp.BinarySensor)
def test_setup_platform_invalid_config():
"""Should check the supplied config and return False if it is invalid."""
config = copy(test_tcp.TEST_CONFIG)
del config[tcp.CONF_HOST]
assert bin_tcp.setup_platform(None, config, None) is False
class TestTCPBinarySensor():
class TestTCPBinarySensor(unittest.TestCase):
"""Test the TCP Binary Sensor."""
def setup_class(cls):
def setup_method(self, method):
"""Setup things to be run when tests are started."""
cls.hass = get_test_home_assistant()
self.hass = get_test_home_assistant()
def teardown_class(cls):
def teardown_method(self, method):
"""Stop down everything that was started."""
cls.hass.stop()
self.hass.stop()
def test_requires_additional_values(self):
"""Should require the additional config values specified."""
config = copy(test_tcp.TEST_CONFIG)
for key in bin_tcp.BinarySensor.required:
del config[key]
assert len(config) != len(test_tcp.TEST_CONFIG)
assert not bin_tcp.BinarySensor.validate_config(config)
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)
@patch('homeassistant.components.sensor.tcp.Sensor.update')
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.sensor.tcp.TcpSensor.update')
def test_setup_platform_devices(self, mock_update):
"""Check the supplied config and call add_devices with sensor."""
add_devices = Mock()
ret = bin_tcp.setup_platform(None, test_tcp.TEST_CONFIG, add_devices)
assert ret is None
assert add_devices.called
assert isinstance(
add_devices.call_args[0][0][0], bin_tcp.TcpBinarySensor)
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_is_on_true(self, mock_update):
"""Should return True if _state is the same as value_on."""
sensor = bin_tcp.BinarySensor(self.hass, test_tcp.TEST_CONFIG)
sensor._state = test_tcp.TEST_CONFIG[tcp.CONF_VALUE_ON]
"""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.sensor.tcp.Sensor.update')
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_is_on_false(self, mock_update):
"""Should return False if _state is not the same as value_on."""
sensor = bin_tcp.BinarySensor(self.hass, test_tcp.TEST_CONFIG)
sensor._state = "%s abc" % test_tcp.TEST_CONFIG[tcp.CONF_VALUE_ON]
"""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])
assert not sensor.is_on

289
tests/components/sensor/test_tcp.py
View file

@ -1,247 +1,270 @@
"""The tests for the TCP sensor platform."""
import socket
import unittest
from copy import copy
from uuid import uuid4
from unittest.mock import patch, Mock
from tests.common import (get_test_home_assistant, assert_setup_component)
from homeassistant.bootstrap import setup_component
from homeassistant.components.sensor import tcp
from homeassistant.helpers.entity import Entity
from tests.common import get_test_home_assistant
TEST_CONFIG = {
tcp.CONF_NAME: "test_name",
tcp.CONF_HOST: "test_host",
'sensor': {
'platform': 'tcp',
tcp.CONF_NAME: 'test_name',
tcp.CONF_HOST: 'test_host',
tcp.CONF_PORT: 12345,
tcp.CONF_TIMEOUT: tcp.DEFAULT_TIMEOUT + 1,
tcp.CONF_PAYLOAD: "test_payload",
tcp.CONF_UNIT: "test_unit",
tcp.CONF_VALUE_TEMPLATE: "test_template",
tcp.CONF_VALUE_ON: "test_on",
tcp.CONF_PAYLOAD: 'test_payload',
tcp.CONF_UNIT_OF_MEASUREMENT: 'test_unit',
tcp.CONF_VALUE_TEMPLATE: 'test_template',
tcp.CONF_VALUE_ON: 'test_on',
tcp.CONF_BUFFER_SIZE: tcp.DEFAULT_BUFFER_SIZE + 1
},
}
KEYS_AND_DEFAULTS = {
tcp.CONF_NAME: None,
tcp.CONF_TIMEOUT: tcp.DEFAULT_TIMEOUT,
tcp.CONF_UNIT: None,
tcp.CONF_UNIT_OF_MEASUREMENT: None,
tcp.CONF_VALUE_TEMPLATE: None,
tcp.CONF_VALUE_ON: None,
tcp.CONF_BUFFER_SIZE: tcp.DEFAULT_BUFFER_SIZE
}
@patch('homeassistant.components.sensor.tcp.Sensor.update')
def test_setup_platform_valid_config(mock_update):
"""Should check the supplied config and call add_entities with Sensor."""
add_entities = Mock()
ret = tcp.setup_platform(None, TEST_CONFIG, add_entities)
assert ret is None, "setup_platform() should return None if successful."
assert add_entities.called
assert isinstance(add_entities.call_args[0][0][0], tcp.Sensor)
def test_setup_platform_invalid_config():
"""Should check the supplied config and return False if it is invalid."""
config = copy(TEST_CONFIG)
del config[tcp.CONF_HOST]
assert tcp.setup_platform(None, config, None) is False
class TestTCPSensor():
class TestTCPSensor(unittest.TestCase):
"""Test the TCP Sensor."""
def setup_class(cls):
def setup_method(self, method):
"""Setup things to be run when tests are started."""
cls.hass = get_test_home_assistant()
self.hass = get_test_home_assistant()
def teardown_class(cls):
def teardown_method(self, method):
"""Stop everything that was started."""
cls.hass.stop()
self.hass.stop()
@patch('homeassistant.components.sensor.tcp.Sensor.update')
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_setup_platform_valid_config(self, mock_update):
"""Check a valid configuration and call add_devices with sensor."""
with assert_setup_component(0, 'sensor'):
assert setup_component(self.hass, 'sensor', TEST_CONFIG)
add_devices = Mock()
tcp.setup_platform(None, TEST_CONFIG['sensor'], add_devices)
assert add_devices.called
assert isinstance(add_devices.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,
}
})
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_name(self, mock_update):
"""Should return the name if set in the config."""
sensor = tcp.Sensor(self.hass, TEST_CONFIG)
assert sensor.name == TEST_CONFIG[tcp.CONF_NAME]
"""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.sensor.tcp.Sensor.update')
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_name_not_set(self, mock_update):
"""Should return the superclass name property if not set in config."""
config = copy(TEST_CONFIG)
"""Return the superclass name property if not set in configuration."""
config = copy(TEST_CONFIG['sensor'])
del config[tcp.CONF_NAME]
entity = Entity()
sensor = tcp.Sensor(self.hass, config)
sensor = tcp.TcpSensor(self.hass, config)
assert sensor.name == entity.name
@patch('homeassistant.components.sensor.tcp.Sensor.update')
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_state(self, mock_update):
"""Should return the contents of _state."""
sensor = tcp.Sensor(self.hass, TEST_CONFIG)
"""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.sensor.tcp.Sensor.update')
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_unit_of_measurement(self, mock_update):
"""Should return the configured unit of measurement."""
sensor = tcp.Sensor(self.hass, TEST_CONFIG)
assert sensor.unit_of_measurement == TEST_CONFIG[tcp.CONF_UNIT]
"""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]
@patch("homeassistant.components.sensor.tcp.Sensor.update")
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_config_valid_keys(self, *args):
"""Should store valid keys in _config."""
sensor = tcp.Sensor(self.hass, TEST_CONFIG)
for key in TEST_CONFIG:
"""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
def test_validate_config_valid_keys(self):
"""Should return True when provided with the correct keys."""
assert tcp.Sensor.validate_config(TEST_CONFIG)
"""Return True when provided with the correct keys."""
with assert_setup_component(0, 'sensor'):
assert setup_component(self.hass, 'sensor', TEST_CONFIG)
@patch("homeassistant.components.sensor.tcp.Sensor.update")
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_config_invalid_keys(self, mock_update):
"""Shouldn't store invalid keys in _config."""
config = copy(TEST_CONFIG)
config = copy(TEST_CONFIG['sensor'])
config.update({
"a": "test_a",
"b": "test_b",
"c": "test_c"
'a': 'test_a',
'b': 'test_b',
'c': 'test_c'
})
sensor = tcp.Sensor(self.hass, config)
for invalid_key in "abc":
sensor = tcp.TcpSensor(self.hass, config)
for invalid_key in 'abc':
assert invalid_key not in sensor._config
def test_validate_config_invalid_keys(self):
"""Test with invalid keys plus some extra."""
config = copy(TEST_CONFIG)
config = copy(TEST_CONFIG['sensor'])
config.update({
"a": "test_a",
"b": "test_b",
"c": "test_c"
'a': 'test_a',
'b': 'test_b',
'c': 'test_c'
})
assert tcp.Sensor.validate_config(config)
with assert_setup_component(0, 'sensor'):
assert setup_component(self.hass, 'sensor', {'tcp': config})
@patch("homeassistant.components.sensor.tcp.Sensor.update")
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_config_uses_defaults(self, mock_update):
"""Should use defaults where appropriate."""
config = copy(TEST_CONFIG)
for key in KEYS_AND_DEFAULTS.keys():
"""Check if defaults were set."""
config = copy(TEST_CONFIG['sensor'])
for key in KEYS_AND_DEFAULTS:
del config[key]
sensor = tcp.Sensor(self.hass, config)
with assert_setup_component(1) as result_config:
assert setup_component(self.hass, 'sensor', {
'sensor': config,
})
sensor = tcp.TcpSensor(self.hass, result_config['sensor'][0])
for key, default in KEYS_AND_DEFAULTS.items():
assert sensor._config[key] == default
def test_validate_config_missing_defaults(self):
"""Should return True when defaulted keys are not provided."""
config = copy(TEST_CONFIG)
for key in KEYS_AND_DEFAULTS.keys():
"""Return True when defaulted keys are not provided."""
config = copy(TEST_CONFIG['sensor'])
for key in KEYS_AND_DEFAULTS:
del config[key]
assert tcp.Sensor.validate_config(config)
with assert_setup_component(0, 'sensor'):
assert setup_component(self.hass, 'sensor', {'tcp': config})
def test_validate_config_missing_required(self):
"""Should return False when required config items are missing."""
for key in TEST_CONFIG:
"""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)
config = copy(TEST_CONFIG['sensor'])
del config[key]
assert not tcp.Sensor.validate_config(config), (
"validate_config() should have returned False since %r was not"
"provided." % key)
with assert_setup_component(0, 'sensor'):
assert setup_component(self.hass, 'sensor', {'tcp': config})
@patch("homeassistant.components.sensor.tcp.Sensor.update")
@patch('homeassistant.components.sensor.tcp.TcpSensor.update')
def test_init_calls_update(self, mock_update):
"""Should call update() method during __init__()."""
tcp.Sensor(self.hass, TEST_CONFIG)
"""Call update() method during __init__()."""
tcp.TcpSensor(self.hass, TEST_CONFIG)
assert mock_update.called
@patch("socket.socket")
@patch("select.select", return_value=(True, False, False))
@patch('socket.socket')
@patch('select.select', return_value=(True, False, False))
def test_update_connects_to_host_and_port(self, mock_select, mock_socket):
"""Should connect to the configured host and port."""
tcp.Sensor(self.hass, TEST_CONFIG)
"""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[tcp.CONF_HOST],
TEST_CONFIG[tcp.CONF_PORT]),)
TEST_CONFIG['sensor'][tcp.CONF_HOST],
TEST_CONFIG['sensor'][tcp.CONF_PORT]),)
@patch("socket.socket.connect", side_effect=socket.error())
@patch('socket.socket.connect', side_effect=socket.error())
def test_update_returns_if_connecting_fails(self, *args):
"""Should return if connecting to host fails."""
with patch("homeassistant.components.sensor.tcp.Sensor.update"):
sensor = tcp.Sensor(self.hass, TEST_CONFIG)
"""Return if connecting to host fails."""
with patch('homeassistant.components.sensor.tcp.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())
@patch('socket.socket.connect')
@patch('socket.socket.send', side_effect=socket.error())
def test_update_returns_if_sending_fails(self, *args):
"""Should return if sending fails."""
with patch("homeassistant.components.sensor.tcp.Sensor.update"):
sensor = tcp.Sensor(self.hass, TEST_CONFIG)
"""Return if sending fails."""
with patch('homeassistant.components.sensor.tcp.TcpSensor.update'):
sensor = tcp.TcpSensor(self.hass, TEST_CONFIG['sensor'])
assert sensor.update() is None
@patch("socket.socket.connect")
@patch("socket.socket.send")
@patch("select.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):
"""Should return if select fails to return a socket."""
with patch("homeassistant.components.sensor.tcp.Sensor.update"):
sensor = tcp.Sensor(self.hass, TEST_CONFIG)
"""Return if select fails to return a socket."""
with patch('homeassistant.components.sensor.tcp.TcpSensor.update'):
sensor = tcp.TcpSensor(self.hass, TEST_CONFIG['sensor'])
assert sensor.update() is None
@patch("socket.socket")
@patch("select.select", return_value=(True, False, False))
@patch('socket.socket')
@patch('select.select', return_value=(True, False, False))
def test_update_sends_payload(self, mock_select, mock_socket):
"""Should send the configured payload as bytes."""
tcp.Sensor(self.hass, TEST_CONFIG)
"""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[tcp.CONF_PAYLOAD].encode()
TEST_CONFIG['sensor'][tcp.CONF_PAYLOAD].encode()
)
@patch("socket.socket")
@patch("select.select", return_value=(True, False, False))
@patch('socket.socket')
@patch('select.select', return_value=(True, False, False))
def test_update_calls_select_with_timeout(self, mock_select, mock_socket):
"""Should provide the timeout argument to select."""
tcp.Sensor(self.hass, TEST_CONFIG)
"""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[tcp.CONF_TIMEOUT])
[mock_socket], [], [], TEST_CONFIG['sensor'][tcp.CONF_TIMEOUT])
@patch("socket.socket")
@patch("select.select", return_value=(True, False, False))
@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"
test_value = 'test_value'
mock_socket = mock_socket().__enter__()
mock_socket.recv.return_value = test_value.encode()
config = copy(TEST_CONFIG)
config = copy(TEST_CONFIG['sensor'])
del config[tcp.CONF_VALUE_TEMPLATE]
sensor = tcp.Sensor(self.hass, config)
sensor = tcp.TcpSensor(self.hass, config)
assert sensor._state == test_value
@patch("socket.socket")
@patch("select.select", return_value=(True, False, False))
@patch('socket.socket')
@patch('select.select', return_value=(True, False, False))
def test_update_renders_value_in_template(self, mock_select, mock_socket):
"""Should render the value in the provided template."""
test_value = "test_value"
"""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)
config[tcp.CONF_VALUE_TEMPLATE] = "{{ value }} {{ 1+1 }}"
sensor = tcp.Sensor(self.hass, config)
assert sensor._state == "%s 2" % test_value
config = copy(TEST_CONFIG['sensor'])
config[tcp.CONF_VALUE_TEMPLATE] = '{{ value }} {{ 1+1 }}'
sensor = tcp.TcpSensor(self.hass, config)
assert sensor._state == '%s 2' % test_value
@patch("socket.socket")
@patch("select.select", return_value=(True, False, False))
@patch('socket.socket')
@patch('select.select', return_value=(True, False, False))
def test_update_returns_if_template_render_fails(
self, mock_select, mock_socket):
"""Should return None if rendering the template fails."""
test_value = "test_value"
"""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)
config = copy(TEST_CONFIG['sensor'])
config[tcp.CONF_VALUE_TEMPLATE] = "{{ this won't work"
sensor = tcp.Sensor(self.hass, config)
sensor = tcp.TcpSensor(self.hass, config)
assert sensor.update() is None