mirror of
https://github.com/home-assistant/core.git
synced 2025-04-24 09:17:53 +00:00
device tracker - tomato https support (#11566)
* initial https support * adding tests * lint errors * missing docstring * fixing non-deterministic params * fixing non-deterministic params * Updating docstrings & added missing tests * revert _LOGGER * updating default port to reflect ssl/nonssl * fixing docstrings for tests
This commit is contained in:
Gregory Dosh
Sean Dague
parent
b43b542667
commit
d65ac7421d
@ -14,7 +14,9 @@ import voluptuous as vol
|
||||
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 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
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<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 +87,16 @@ class TomatoDeviceScanner(DeviceScanner):
|
||||
|
||||
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 @@ class TomatoDeviceScanner(DeviceScanner):
|
||||
|
||||
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 @@ class TomatoDeviceScanner(DeviceScanner):
|
||||
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
|
||||
|
||||
380
tests/components/device_tracker/test_tomato.py
Normal file
380
tests/components/device_tracker/test_tomato.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user