device tracker - tomato https support

homeassistant/components/device_tracker/tomato.py

@@ -14,14 +14,20 @@
 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'
 
-_LOGGER = logging.getLogger(__name__)
+_LOGGER = logging.getLogger("{}.{}".format(__name__, "Tomato"))
 
 PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
     vol.Required(CONF_HOST): cv.string,
+    vol.Optional(CONF_PORT, default=80): 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,19 @@ 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]
 
         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<param>\w*) = (?P<value>.*);")
 
-        self.logger = logging.getLogger("{}.{}".format(__name__, "Tomato"))
         self.last_results = {"wldev": [], "dhcpd_lease": []}
 
         self.success_init = self._update_tomato_info()
@@ -74,10 +83,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,25 +107,25 @@ 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
 
         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

tests/components/device_tracker/test_tomato.py

@@ -0,0 +1,280 @@
+"""The tests for the Tomato device tracker platform."""
+from unittest import mock
+import pytest
+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):
+    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 result.req.body == "_http_id=1234567890&exec=devlist"
+
+
+@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 result.req.body == "_http_id=1234567890&exec=devlist"
+    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 result.req.body == "_http_id=0987654321&exec=devlist"
+    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 result.req.body == "_http_id=0987654321&exec=devlist"
+    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.
+
+    Representing the absolute path to a CA certificate bundle.
+    """
+    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_scan_devices(hass, mock_exception_logger):
+    """Test the setup with bad credentials.
+
+    Representing the absolute path to a CA certificate bundle.
+    """
+    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_get_device_name(hass, mock_exception_logger):
+    """Test the setup with bad credentials.
+
+    Representing the absolute path to a CA certificate bundle.
+    """
+    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'