Automatic polling (#3360)

* Test updating automatic

* Scan interval

* Schedule scan every time delta

* Pass around has

* Recursive issue

* Method invocation

* Oops

* Set up poll

* Default argument value

* Unused import

* Semicolon

* Fix tests

* Linting

* Unneeded throttle as it's handled by time event

* Use track time change event listener

* Disable lint rule

* Attribute removed - removing test

* Debug instead of info

* Unused import
pull/3471/head
Teagan Glenn 2016-09-12 20:59:34 -06:00 committed by Paulus Schoutsen
parent 5fd93e8d80
commit 380993f2ca
2 changed files with 17 additions and 40 deletions

View File

@ -15,12 +15,11 @@ from homeassistant.components.device_tracker import (PLATFORM_SCHEMA,
ATTR_ATTRIBUTES)
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD
import homeassistant.helpers.config_validation as cv
from homeassistant.util import Throttle, datetime as dt_util
from homeassistant.helpers.event import track_utc_time_change
from homeassistant.util import datetime as dt_util
_LOGGER = logging.getLogger(__name__)
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=30)
CONF_CLIENT_ID = 'client_id'
CONF_SECRET = 'secret'
CONF_DEVICES = 'devices'
@ -53,7 +52,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
def setup_scanner(hass, config: dict, see):
"""Validate the configuration and return an Automatic scanner."""
try:
AutomaticDeviceScanner(config, see)
AutomaticDeviceScanner(hass, config, see)
except requests.HTTPError as err:
_LOGGER.error(str(err))
return False
@ -61,11 +60,14 @@ def setup_scanner(hass, config: dict, see):
return True
# pylint: disable=too-many-instance-attributes
# pylint: disable=too-few-public-methods
class AutomaticDeviceScanner(object):
"""A class representing an Automatic device."""
def __init__(self, config: dict, see) -> None:
def __init__(self, hass, config: dict, see) -> None:
"""Initialize the automatic device scanner."""
self.hass = hass
self._devices = config.get(CONF_DEVICES, None)
self._access_token_payload = {
'username': config.get(CONF_USERNAME),
@ -81,20 +83,10 @@ class AutomaticDeviceScanner(object):
self.last_trips = {}
self.see = see
self.scan_devices()
def scan_devices(self):
"""Scan for new devices and return a list with found device IDs."""
self._update_info()
return [item['id'] for item in self.last_results]
def get_device_name(self, device):
"""Get the device name from id."""
vehicle = [item['display_name'] for item in self.last_results
if item['id'] == device]
return vehicle[0]
track_utc_time_change(self.hass, self._update_info,
second=range(0, 60, 30))
def _update_headers(self):
"""Get the access token from automatic."""
@ -114,10 +106,9 @@ class AutomaticDeviceScanner(object):
'Authorization': 'Bearer {}'.format(access_token)
}
@Throttle(MIN_TIME_BETWEEN_SCANS)
def _update_info(self) -> None:
def _update_info(self, now=None) -> None:
"""Update the device info."""
_LOGGER.info('Updating devices')
_LOGGER.debug('Updating devices %s', now)
self._update_headers()
response = requests.get(URL_VEHICLES, headers=self._headers)

View File

@ -6,8 +6,9 @@ import unittest
from unittest.mock import patch
from homeassistant.components.device_tracker.automatic import (
URL_AUTHORIZE, URL_VEHICLES, URL_TRIPS, setup_scanner,
AutomaticDeviceScanner)
URL_AUTHORIZE, URL_VEHICLES, URL_TRIPS, setup_scanner)
from tests.common import get_test_home_assistant
_LOGGER = logging.getLogger(__name__)
@ -205,6 +206,7 @@ class TestAutomatic(unittest.TestCase):
def setUp(self):
"""Set up test data."""
self.hass = get_test_home_assistant()
def tearDown(self):
"""Tear down test data."""
@ -221,7 +223,7 @@ class TestAutomatic(unittest.TestCase):
'secret': CLIENT_SECRET
}
self.assertFalse(setup_scanner(None, config, self.see_mock))
self.assertFalse(setup_scanner(self.hass, config, self.see_mock))
@patch('requests.get', side_effect=mocked_requests)
@patch('requests.post', side_effect=mocked_requests)
@ -235,20 +237,4 @@ class TestAutomatic(unittest.TestCase):
'secret': CLIENT_SECRET
}
self.assertTrue(setup_scanner(None, config, self.see_mock))
@patch('requests.get', side_effect=mocked_requests)
@patch('requests.post', side_effect=mocked_requests)
def test_device_attributes(self, mock_get, mock_post):
"""Test device attributes are set on load."""
config = {
'platform': 'automatic',
'username': VALID_USERNAME,
'password': PASSWORD,
'client_id': CLIENT_ID,
'secret': CLIENT_SECRET
}
scanner = AutomaticDeviceScanner(config, self.see_mock)
self.assertEqual(DISPLAY_NAME, scanner.get_device_name('vid'))
self.assertTrue(setup_scanner(self.hass, config, self.see_mock))