diff --git a/homeassistant/components/device_tracker/tomato.py b/homeassistant/components/device_tracker/tomato.py
index 57e83eaeb94a9..7cebf0abdf421 100644
--- a/homeassistant/components/device_tracker/tomato.py
+++ b/homeassistant/components/device_tracker/tomato.py
@@ -14,7 +14,9 @@
import homeassistant.helpers.config_validation as cv
from homeassistant.components.device_tracker import (
DOMAIN, PLATFORM_SCHEMA, DeviceScanner)
-from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.const import (
+ CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL,
+ CONF_PASSWORD, CONF_USERNAME)
CONF_HTTP_ID = 'http_id'
@@ -22,6 +24,10 @@
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_HOST): cv.string,
+ vol.Optional(CONF_PORT, default=-1): cv.port,
+ vol.Optional(CONF_SSL, default=False): cv.boolean,
+ vol.Optional(CONF_VERIFY_SSL, default=True): vol.Any(
+ cv.boolean, cv.isfile),
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_HTTP_ID): cv.string
@@ -39,16 +45,23 @@ class TomatoDeviceScanner(DeviceScanner):
def __init__(self, config):
"""Initialize the scanner."""
host, http_id = config[CONF_HOST], config[CONF_HTTP_ID]
+ port = config[CONF_PORT]
username, password = config[CONF_USERNAME], config[CONF_PASSWORD]
+ self.ssl, self.verify_ssl = config[CONF_SSL], config[CONF_VERIFY_SSL]
+ if port == -1:
+ port = 80
+ if self.ssl:
+ port = 443
self.req = requests.Request(
- 'POST', 'http://{}/update.cgi'.format(host),
+ 'POST', 'http{}://{}:{}/update.cgi'.format(
+ "s" if self.ssl else "", host, port
+ ),
data={'_http_id': http_id, 'exec': 'devlist'},
auth=requests.auth.HTTPBasicAuth(username, password)).prepare()
self.parse_api_pattern = re.compile(r"(?P\w*) = (?P.*);")
- self.logger = logging.getLogger("{}.{}".format(__name__, "Tomato"))
self.last_results = {"wldev": [], "dhcpd_lease": []}
self.success_init = self._update_tomato_info()
@@ -74,10 +87,16 @@ def _update_tomato_info(self):
Return boolean if scanning successful.
"""
- self.logger.info("Scanning")
+ _LOGGER.info("Scanning")
try:
- response = requests.Session().send(self.req, timeout=3)
+ if self.ssl:
+ response = requests.Session().send(self.req,
+ timeout=3,
+ verify=self.verify_ssl)
+ else:
+ response = requests.Session().send(self.req, timeout=3)
+
# Calling and parsing the Tomato api here. We only need the
# wldev and dhcpd_lease values.
if response.status_code == 200:
@@ -92,7 +111,7 @@ def _update_tomato_info(self):
elif response.status_code == 401:
# Authentication error
- self.logger.exception((
+ _LOGGER.exception((
"Failed to authenticate, "
"please check your username and password"))
return False
@@ -100,17 +119,17 @@ def _update_tomato_info(self):
except requests.exceptions.ConnectionError:
# We get this if we could not connect to the router or
# an invalid http_id was supplied.
- self.logger.exception("Failed to connect to the router or "
- "invalid http_id supplied")
+ _LOGGER.exception("Failed to connect to the router or "
+ "invalid http_id supplied")
return False
except requests.exceptions.Timeout:
# We get this if we could not connect to the router or
# an invalid http_id was supplied.
- self.logger.exception("Connection to the router timed out")
+ _LOGGER.exception("Connection to the router timed out")
return False
except ValueError:
# If JSON decoder could not parse the response.
- self.logger.exception("Failed to parse response from router")
+ _LOGGER.exception("Failed to parse response from router")
return False
diff --git a/tests/components/device_tracker/test_tomato.py b/tests/components/device_tracker/test_tomato.py
new file mode 100644
index 0000000000000..cce39ce43a79d
--- /dev/null
+++ b/tests/components/device_tracker/test_tomato.py
@@ -0,0 +1,380 @@
+"""The tests for the Tomato device tracker platform."""
+from unittest import mock
+import pytest
+import requests
+import requests_mock
+import voluptuous as vol
+
+from homeassistant.components.device_tracker import DOMAIN, tomato as tomato
+from homeassistant.const import (CONF_HOST, CONF_USERNAME, CONF_PASSWORD,
+ CONF_PORT, CONF_SSL, CONF_PLATFORM,
+ CONF_VERIFY_SSL)
+
+
+def mock_session_response(*args, **kwargs):
+ """Mock data generation for session response."""
+ class MockSessionResponse:
+ def __init__(self, text, status_code):
+ self.text = text
+ self.status_code = status_code
+
+ # Username: foo
+ # Password: bar
+ if args[0].headers['Authorization'] != 'Basic Zm9vOmJhcg==':
+ return MockSessionResponse(None, 401)
+ elif "gimmie_bad_data" in args[0].body:
+ return MockSessionResponse('This shouldn\'t (wldev = be here.;', 200)
+ elif "gimmie_good_data" in args[0].body:
+ return MockSessionResponse(
+ "wldev = [ ['eth1','F4:F5:D8:AA:AA:AA',"
+ "-42,5500,1000,7043,0],['eth1','58:EF:68:00:00:00',"
+ "-42,5500,1000,7043,0]];\n"
+ "dhcpd_lease = [ ['chromecast','172.10.10.5','F4:F5:D8:AA:AA:AA',"
+ "'0 days, 16:17:08'],['wemo','172.10.10.6','58:EF:68:00:00:00',"
+ "'0 days, 12:09:08']];", 200)
+
+ return MockSessionResponse(None, 200)
+
+
+@pytest.fixture
+def mock_exception_logger():
+ """Mock pyunifi."""
+ with mock.patch('homeassistant.components.device_tracker'
+ '.tomato._LOGGER.exception') as mock_exception_logger:
+ yield mock_exception_logger
+
+
+@pytest.fixture
+def mock_session_send():
+ """Mock requests.Session().send."""
+ with mock.patch('requests.Session.send') as mock_session_send:
+ yield mock_session_send
+
+
+def test_config_missing_optional_params(hass, mock_session_send):
+ """Test the setup without optional parameters."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_USERNAME: 'foo',
+ CONF_PASSWORD: 'password',
+ tomato.CONF_HTTP_ID: '1234567890'
+ })
+ }
+ result = tomato.get_scanner(hass, config)
+ assert result.req.url == "http://tomato-router:80/update.cgi"
+ assert result.req.headers == {
+ 'Content-Length': '32',
+ 'Content-Type': 'application/x-www-form-urlencoded',
+ 'Authorization': 'Basic Zm9vOnBhc3N3b3Jk'
+ }
+ assert "_http_id=1234567890" in result.req.body
+ assert "exec=devlist" in result.req.body
+
+
+@mock.patch('os.access', return_value=True)
+@mock.patch('os.path.isfile', mock.Mock(return_value=True))
+def test_config_default_nonssl_port(hass, mock_session_send):
+ """Test the setup without a default port set without ssl enabled."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_USERNAME: 'foo',
+ CONF_PASSWORD: 'password',
+ tomato.CONF_HTTP_ID: '1234567890'
+ })
+ }
+ result = tomato.get_scanner(hass, config)
+ assert result.req.url == "http://tomato-router:80/update.cgi"
+
+
+@mock.patch('os.access', return_value=True)
+@mock.patch('os.path.isfile', mock.Mock(return_value=True))
+def test_config_default_ssl_port(hass, mock_session_send):
+ """Test the setup without a default port set with ssl enabled."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_SSL: True,
+ CONF_USERNAME: 'foo',
+ CONF_PASSWORD: 'password',
+ tomato.CONF_HTTP_ID: '1234567890'
+ })
+ }
+ result = tomato.get_scanner(hass, config)
+ assert result.req.url == "https://tomato-router:443/update.cgi"
+
+
+@mock.patch('os.access', return_value=True)
+@mock.patch('os.path.isfile', mock.Mock(return_value=True))
+def test_config_verify_ssl_but_no_ssl_enabled(hass, mock_session_send):
+ """Test the setup with a string with ssl_verify but ssl not enabled."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_PORT: 1234,
+ CONF_SSL: False,
+ CONF_VERIFY_SSL: "/tmp/tomato.crt",
+ CONF_USERNAME: 'foo',
+ CONF_PASSWORD: 'password',
+ tomato.CONF_HTTP_ID: '1234567890'
+ })
+ }
+ result = tomato.get_scanner(hass, config)
+ assert result.req.url == "http://tomato-router:1234/update.cgi"
+ assert result.req.headers == {
+ 'Content-Length': '32',
+ 'Content-Type': 'application/x-www-form-urlencoded',
+ 'Authorization': 'Basic Zm9vOnBhc3N3b3Jk'
+ }
+ assert "_http_id=1234567890" in result.req.body
+ assert "exec=devlist" in result.req.body
+ assert mock_session_send.call_count == 1
+ assert mock_session_send.mock_calls[0] == \
+ mock.call(result.req, timeout=3)
+
+
+@mock.patch('os.access', return_value=True)
+@mock.patch('os.path.isfile', mock.Mock(return_value=True))
+def test_config_valid_verify_ssl_path(hass, mock_session_send):
+ """Test the setup with a string for ssl_verify.
+
+ Representing the absolute path to a CA certificate bundle.
+ """
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_PORT: 1234,
+ CONF_SSL: True,
+ CONF_VERIFY_SSL: "/tmp/tomato.crt",
+ CONF_USERNAME: 'bar',
+ CONF_PASSWORD: 'foo',
+ tomato.CONF_HTTP_ID: '0987654321'
+ })
+ }
+ result = tomato.get_scanner(hass, config)
+ assert result.req.url == "https://tomato-router:1234/update.cgi"
+ assert result.req.headers == {
+ 'Content-Length': '32',
+ 'Content-Type': 'application/x-www-form-urlencoded',
+ 'Authorization': 'Basic YmFyOmZvbw=='
+ }
+ assert "_http_id=0987654321" in result.req.body
+ assert "exec=devlist" in result.req.body
+ assert mock_session_send.call_count == 1
+ assert mock_session_send.mock_calls[0] == \
+ mock.call(result.req, timeout=3, verify="/tmp/tomato.crt")
+
+
+def test_config_valid_verify_ssl_bool(hass, mock_session_send):
+ """Test the setup with a bool for ssl_verify."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_PORT: 1234,
+ CONF_SSL: True,
+ CONF_VERIFY_SSL: "False",
+ CONF_USERNAME: 'bar',
+ CONF_PASSWORD: 'foo',
+ tomato.CONF_HTTP_ID: '0987654321'
+ })
+ }
+ result = tomato.get_scanner(hass, config)
+ assert result.req.url == "https://tomato-router:1234/update.cgi"
+ assert result.req.headers == {
+ 'Content-Length': '32',
+ 'Content-Type': 'application/x-www-form-urlencoded',
+ 'Authorization': 'Basic YmFyOmZvbw=='
+ }
+ assert "_http_id=0987654321" in result.req.body
+ assert "exec=devlist" in result.req.body
+ assert mock_session_send.call_count == 1
+ assert mock_session_send.mock_calls[0] == \
+ mock.call(result.req, timeout=3, verify=False)
+
+
+def test_config_errors():
+ """Test for configuration errors."""
+ with pytest.raises(vol.Invalid):
+ tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ # No Host,
+ CONF_PORT: 1234,
+ CONF_SSL: True,
+ CONF_VERIFY_SSL: "False",
+ CONF_USERNAME: 'bar',
+ CONF_PASSWORD: 'foo',
+ tomato.CONF_HTTP_ID: '0987654321'
+ })
+ with pytest.raises(vol.Invalid):
+ tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_PORT: -123456789, # Bad Port
+ CONF_SSL: True,
+ CONF_VERIFY_SSL: "False",
+ CONF_USERNAME: 'bar',
+ CONF_PASSWORD: 'foo',
+ tomato.CONF_HTTP_ID: '0987654321'
+ })
+ with pytest.raises(vol.Invalid):
+ tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_PORT: 1234,
+ CONF_SSL: True,
+ CONF_VERIFY_SSL: "False",
+ # No Username
+ CONF_PASSWORD: 'foo',
+ tomato.CONF_HTTP_ID: '0987654321'
+ })
+ with pytest.raises(vol.Invalid):
+ tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_PORT: 1234,
+ CONF_SSL: True,
+ CONF_VERIFY_SSL: "False",
+ CONF_USERNAME: 'bar',
+ # No Password
+ tomato.CONF_HTTP_ID: '0987654321'
+ })
+ with pytest.raises(vol.Invalid):
+ tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_PORT: 1234,
+ CONF_SSL: True,
+ CONF_VERIFY_SSL: "False",
+ CONF_USERNAME: 'bar',
+ CONF_PASSWORD: 'foo',
+ # No HTTP_ID
+ })
+
+
+@mock.patch('requests.Session.send', side_effect=mock_session_response)
+def test_config_bad_credentials(hass, mock_exception_logger):
+ """Test the setup with bad credentials."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_USERNAME: 'i_am',
+ CONF_PASSWORD: 'an_imposter',
+ tomato.CONF_HTTP_ID: '1234'
+ })
+ }
+
+ tomato.get_scanner(hass, config)
+
+ assert mock_exception_logger.call_count == 1
+ assert mock_exception_logger.mock_calls[0] == \
+ mock.call("Failed to authenticate, "
+ "please check your username and password")
+
+
+@mock.patch('requests.Session.send', side_effect=mock_session_response)
+def test_bad_response(hass, mock_exception_logger):
+ """Test the setup with bad response from router."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_USERNAME: 'foo',
+ CONF_PASSWORD: 'bar',
+ tomato.CONF_HTTP_ID: 'gimmie_bad_data'
+ })
+ }
+
+ tomato.get_scanner(hass, config)
+
+ assert mock_exception_logger.call_count == 1
+ assert mock_exception_logger.mock_calls[0] == \
+ mock.call("Failed to parse response from router")
+
+
+@mock.patch('requests.Session.send', side_effect=mock_session_response)
+def test_scan_devices(hass, mock_exception_logger):
+ """Test scanning for new devices."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_USERNAME: 'foo',
+ CONF_PASSWORD: 'bar',
+ tomato.CONF_HTTP_ID: 'gimmie_good_data'
+ })
+ }
+
+ scanner = tomato.get_scanner(hass, config)
+ assert scanner.scan_devices() == ['F4:F5:D8:AA:AA:AA', '58:EF:68:00:00:00']
+
+
+@mock.patch('requests.Session.send', side_effect=mock_session_response)
+def test_bad_connection(hass, mock_exception_logger):
+ """Test the router with a connection error."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_USERNAME: 'foo',
+ CONF_PASSWORD: 'bar',
+ tomato.CONF_HTTP_ID: 'gimmie_good_data'
+ })
+ }
+
+ with requests_mock.Mocker() as adapter:
+ adapter.register_uri('POST', 'http://tomato-router:80/update.cgi',
+ exc=requests.exceptions.ConnectionError),
+ tomato.get_scanner(hass, config)
+ assert mock_exception_logger.call_count == 1
+ assert mock_exception_logger.mock_calls[0] == \
+ mock.call("Failed to connect to the router "
+ "or invalid http_id supplied")
+
+
+@mock.patch('requests.Session.send', side_effect=mock_session_response)
+def test_router_timeout(hass, mock_exception_logger):
+ """Test the router with a timeout error."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_USERNAME: 'foo',
+ CONF_PASSWORD: 'bar',
+ tomato.CONF_HTTP_ID: 'gimmie_good_data'
+ })
+ }
+
+ with requests_mock.Mocker() as adapter:
+ adapter.register_uri('POST', 'http://tomato-router:80/update.cgi',
+ exc=requests.exceptions.Timeout),
+ tomato.get_scanner(hass, config)
+ assert mock_exception_logger.call_count == 1
+ assert mock_exception_logger.mock_calls[0] == \
+ mock.call("Connection to the router timed out")
+
+
+@mock.patch('requests.Session.send', side_effect=mock_session_response)
+def test_get_device_name(hass, mock_exception_logger):
+ """Test getting device names."""
+ config = {
+ DOMAIN: tomato.PLATFORM_SCHEMA({
+ CONF_PLATFORM: tomato.DOMAIN,
+ CONF_HOST: 'tomato-router',
+ CONF_USERNAME: 'foo',
+ CONF_PASSWORD: 'bar',
+ tomato.CONF_HTTP_ID: 'gimmie_good_data'
+ })
+ }
+
+ scanner = tomato.get_scanner(hass, config)
+ assert scanner.get_device_name('F4:F5:D8:AA:AA:AA') == 'chromecast'
+ assert scanner.get_device_name('58:EF:68:00:00:00') == 'wemo'
+ assert scanner.get_device_name('AA:BB:CC:00:00:00') is None