mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 11:17:21 +00:00
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
This commit is contained in:
parent
b9154158e8
commit
2b1416c514
@ -15,12 +15,11 @@ from homeassistant.components.device_tracker import (PLATFORM_SCHEMA,
|
|||||||
ATTR_ATTRIBUTES)
|
ATTR_ATTRIBUTES)
|
||||||
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD
|
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD
|
||||||
import homeassistant.helpers.config_validation as cv
|
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__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=30)
|
|
||||||
|
|
||||||
CONF_CLIENT_ID = 'client_id'
|
CONF_CLIENT_ID = 'client_id'
|
||||||
CONF_SECRET = 'secret'
|
CONF_SECRET = 'secret'
|
||||||
CONF_DEVICES = 'devices'
|
CONF_DEVICES = 'devices'
|
||||||
@ -53,7 +52,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
|||||||
def setup_scanner(hass, config: dict, see):
|
def setup_scanner(hass, config: dict, see):
|
||||||
"""Validate the configuration and return an Automatic scanner."""
|
"""Validate the configuration and return an Automatic scanner."""
|
||||||
try:
|
try:
|
||||||
AutomaticDeviceScanner(config, see)
|
AutomaticDeviceScanner(hass, config, see)
|
||||||
except requests.HTTPError as err:
|
except requests.HTTPError as err:
|
||||||
_LOGGER.error(str(err))
|
_LOGGER.error(str(err))
|
||||||
return False
|
return False
|
||||||
@ -61,11 +60,14 @@ def setup_scanner(hass, config: dict, see):
|
|||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
# pylint: disable=too-many-instance-attributes
|
||||||
|
# pylint: disable=too-few-public-methods
|
||||||
class AutomaticDeviceScanner(object):
|
class AutomaticDeviceScanner(object):
|
||||||
"""A class representing an Automatic device."""
|
"""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."""
|
"""Initialize the automatic device scanner."""
|
||||||
|
self.hass = hass
|
||||||
self._devices = config.get(CONF_DEVICES, None)
|
self._devices = config.get(CONF_DEVICES, None)
|
||||||
self._access_token_payload = {
|
self._access_token_payload = {
|
||||||
'username': config.get(CONF_USERNAME),
|
'username': config.get(CONF_USERNAME),
|
||||||
@ -81,20 +83,10 @@ class AutomaticDeviceScanner(object):
|
|||||||
self.last_trips = {}
|
self.last_trips = {}
|
||||||
self.see = see
|
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()
|
self._update_info()
|
||||||
|
|
||||||
return [item['id'] for item in self.last_results]
|
track_utc_time_change(self.hass, self._update_info,
|
||||||
|
second=range(0, 60, 30))
|
||||||
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]
|
|
||||||
|
|
||||||
def _update_headers(self):
|
def _update_headers(self):
|
||||||
"""Get the access token from automatic."""
|
"""Get the access token from automatic."""
|
||||||
@ -114,10 +106,9 @@ class AutomaticDeviceScanner(object):
|
|||||||
'Authorization': 'Bearer {}'.format(access_token)
|
'Authorization': 'Bearer {}'.format(access_token)
|
||||||
}
|
}
|
||||||
|
|
||||||
@Throttle(MIN_TIME_BETWEEN_SCANS)
|
def _update_info(self, now=None) -> None:
|
||||||
def _update_info(self) -> None:
|
|
||||||
"""Update the device info."""
|
"""Update the device info."""
|
||||||
_LOGGER.info('Updating devices')
|
_LOGGER.debug('Updating devices %s', now)
|
||||||
self._update_headers()
|
self._update_headers()
|
||||||
|
|
||||||
response = requests.get(URL_VEHICLES, headers=self._headers)
|
response = requests.get(URL_VEHICLES, headers=self._headers)
|
||||||
|
@ -6,8 +6,9 @@ import unittest
|
|||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
from homeassistant.components.device_tracker.automatic import (
|
from homeassistant.components.device_tracker.automatic import (
|
||||||
URL_AUTHORIZE, URL_VEHICLES, URL_TRIPS, setup_scanner,
|
URL_AUTHORIZE, URL_VEHICLES, URL_TRIPS, setup_scanner)
|
||||||
AutomaticDeviceScanner)
|
|
||||||
|
from tests.common import get_test_home_assistant
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
@ -205,6 +206,7 @@ class TestAutomatic(unittest.TestCase):
|
|||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
"""Set up test data."""
|
"""Set up test data."""
|
||||||
|
self.hass = get_test_home_assistant()
|
||||||
|
|
||||||
def tearDown(self):
|
def tearDown(self):
|
||||||
"""Tear down test data."""
|
"""Tear down test data."""
|
||||||
@ -221,7 +223,7 @@ class TestAutomatic(unittest.TestCase):
|
|||||||
'secret': CLIENT_SECRET
|
'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.get', side_effect=mocked_requests)
|
||||||
@patch('requests.post', side_effect=mocked_requests)
|
@patch('requests.post', side_effect=mocked_requests)
|
||||||
@ -235,20 +237,4 @@ class TestAutomatic(unittest.TestCase):
|
|||||||
'secret': CLIENT_SECRET
|
'secret': CLIENT_SECRET
|
||||||
}
|
}
|
||||||
|
|
||||||
self.assertTrue(setup_scanner(None, config, self.see_mock))
|
self.assertTrue(setup_scanner(self.hass, 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'))
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user