Use the Last Seen attribute in unify (#8998)
* Uses the Last Seen attribute in unify * Update unifi.py fix format * Update unifi.py formatting again * update test_unifi to call CONF_CONSIDER_HOME Updated. * Update test_unifi.py * Update test_unifi.py * More unit test test * Update where consider_home comes from. * Update test_unifi.py * Update unifi.py * Update unifi.py * Update test_unifi.py * Update unifi.py * Update unifi.py * Update test_unifi.py * fix hound * Update test_unifi.py * Update test_unifi.py * Update unifi.py * Update unifi.py * Update test_unifi.py * Update unifi.py * Update unifi.py * Update test_unifi.py * Update unifi.py * Update test_unifi.py * Update test_unifi.py * Update test_unifi.py * Update unifi.py * Update unifi.py * Update unifi.py * Update unifi.py * Update test_unifi.py Fix the butcher of tests. * Update unifi.py * Update test_unifi.py * Update test_unifi.py * Update unifi.py * Update unifi.py * Update test_unifi.py * Update test_unifi.py * Update test_unifi.py * Update test_unifi.py * Update test_unifi.py * Update test_unifi.py * Update test_unifi.py * Update test_unifi.py * Update test_unifi.py * Update unifi.py * Update test_unifi.py * Update unifi.py * Update unifi.py * Update unifi.py * Update unifi.py
This commit is contained in:
rbflurry
Pascal Vizeli
parent
f7545fe85c
commit
b00d0a1253
2 changed files with 44 additions and 19 deletions
|
|
@ -5,6 +5,7 @@ For more details about this platform, please refer to the documentation at
|
|||
https://home-assistant.io/components/device_tracker.unifi/
|
||||
"""
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
import voluptuous as vol
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
|
@ -12,16 +13,19 @@ from homeassistant.components.device_tracker import (
|
|||
DOMAIN, PLATFORM_SCHEMA, DeviceScanner)
|
||||
from homeassistant.const import CONF_HOST, CONF_USERNAME, CONF_PASSWORD
|
||||
from homeassistant.const import CONF_VERIFY_SSL
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
REQUIREMENTS = ['pyunifi==2.13']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
CONF_PORT = 'port'
|
||||
CONF_SITE_ID = 'site_id'
|
||||
CONF_DETECTION_TIME = 'detection_time'
|
||||
|
||||
DEFAULT_HOST = 'localhost'
|
||||
DEFAULT_PORT = 8443
|
||||
DEFAULT_VERIFY_SSL = True
|
||||
DEFAULT_DETECTION_TIME = timedelta(seconds=300)
|
||||
|
||||
NOTIFICATION_ID = 'unifi_notification'
|
||||
NOTIFICATION_TITLE = 'Unifi Device Tracker Setup'
|
||||
|
|
@ -33,7 +37,9 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
|||
vol.Required(CONF_USERNAME): cv.string,
|
||||
vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
|
||||
vol.Optional(CONF_VERIFY_SSL, default=DEFAULT_VERIFY_SSL): vol.Any(
|
||||
cv.boolean, cv.isfile)
|
||||
cv.boolean, cv.isfile),
|
||||
vol.Optional(CONF_DETECTION_TIME, default=DEFAULT_DETECTION_TIME): vol.All(
|
||||
cv.time_period, cv.positive_timedelta)
|
||||
})
|
||||
|
||||
|
||||
|
|
@ -47,6 +53,7 @@ def get_scanner(hass, config):
|
|||
site_id = config[DOMAIN].get(CONF_SITE_ID)
|
||||
port = config[DOMAIN].get(CONF_PORT)
|
||||
verify_ssl = config[DOMAIN].get(CONF_VERIFY_SSL)
|
||||
detection_time = config[DOMAIN].get(CONF_DETECTION_TIME)
|
||||
|
||||
try:
|
||||
ctrl = Controller(host, username, password, port, version='v4',
|
||||
|
|
@ -62,14 +69,15 @@ def get_scanner(hass, config):
|
|||
notification_id=NOTIFICATION_ID)
|
||||
return False
|
||||
|
||||
return UnifiScanner(ctrl)
|
||||
return UnifiScanner(ctrl, detection_time)
|
||||
|
||||
|
||||
class UnifiScanner(DeviceScanner):
|
||||
"""Provide device_tracker support from Unifi WAP client data."""
|
||||
|
||||
def __init__(self, controller):
|
||||
def __init__(self, controller, detection_time: timedelta):
|
||||
"""Initialize the scanner."""
|
||||
self._detection_time = detection_time
|
||||
self._controller = controller
|
||||
self._update()
|
||||
|
||||
|
|
@ -82,7 +90,11 @@ class UnifiScanner(DeviceScanner):
|
|||
_LOGGER.error("Failed to scan clients: %s", ex)
|
||||
clients = []
|
||||
|
||||
self._clients = {client['mac']: client for client in clients}
|
||||
self._clients = {
|
||||
client['mac']: client
|
||||
for client in clients
|
||||
if (dt_util.utcnow() - dt_util.utc_from_timestamp(float(
|
||||
client['last_seen']))) < self._detection_time}
|
||||
|
||||
def scan_devices(self):
|
||||
"""Scan for devices."""
|
||||
|
|
@ -97,5 +109,5 @@ class UnifiScanner(DeviceScanner):
|
|||
"""
|
||||
client = self._clients.get(mac, {})
|
||||
name = client.get('name') or client.get('hostname')
|
||||
_LOGGER.debug("Device %s name %s", mac, name)
|
||||
_LOGGER.debug("Device mac %s name %s", mac, name)
|
||||
return name
|
||||
|
|
|
|||
|
|
@ -1,6 +1,9 @@
|
|||
"""The tests for the Unifi WAP device tracker platform."""
|
||||
from unittest import mock
|
||||
from pyunifi.controller import APIError
|
||||
import homeassistant.util.dt as dt_util
|
||||
from datetime import timedelta
|
||||
|
||||
|
||||
import pytest
|
||||
import voluptuous as vol
|
||||
|
|
@ -8,6 +11,7 @@ import voluptuous as vol
|
|||
from homeassistant.components.device_tracker import DOMAIN, unifi as unifi
|
||||
from homeassistant.const import (CONF_HOST, CONF_USERNAME, CONF_PASSWORD,
|
||||
CONF_PLATFORM, CONF_VERIFY_SSL)
|
||||
DEFAULT_DETECTION_TIME = timedelta(seconds=300)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -48,7 +52,8 @@ def test_config_valid_verify_ssl(hass, mock_scanner, mock_ctrl):
|
|||
version='v4', site_id='default', ssl_verify="/tmp/unifi.crt")
|
||||
|
||||
assert mock_scanner.call_count == 1
|
||||
assert mock_scanner.call_args == mock.call(mock_ctrl.return_value)
|
||||
assert mock_scanner.call_args == mock.call(mock_ctrl.return_value,
|
||||
DEFAULT_DETECTION_TIME)
|
||||
|
||||
|
||||
def test_config_minimal(hass, mock_scanner, mock_ctrl):
|
||||
|
|
@ -68,7 +73,8 @@ def test_config_minimal(hass, mock_scanner, mock_ctrl):
|
|||
version='v4', site_id='default', ssl_verify=True)
|
||||
|
||||
assert mock_scanner.call_count == 1
|
||||
assert mock_scanner.call_args == mock.call(mock_ctrl.return_value)
|
||||
assert mock_scanner.call_args == mock.call(mock_ctrl.return_value,
|
||||
DEFAULT_DETECTION_TIME)
|
||||
|
||||
|
||||
def test_config_full(hass, mock_scanner, mock_ctrl):
|
||||
|
|
@ -82,6 +88,7 @@ def test_config_full(hass, mock_scanner, mock_ctrl):
|
|||
CONF_VERIFY_SSL: False,
|
||||
'port': 123,
|
||||
'site_id': 'abcdef01',
|
||||
'detection_time': 300,
|
||||
})
|
||||
}
|
||||
result = unifi.get_scanner(hass, config)
|
||||
|
|
@ -92,7 +99,8 @@ def test_config_full(hass, mock_scanner, mock_ctrl):
|
|||
version='v4', site_id='abcdef01', ssl_verify=False)
|
||||
|
||||
assert mock_scanner.call_count == 1
|
||||
assert mock_scanner.call_args == mock.call(mock_ctrl.return_value)
|
||||
assert mock_scanner.call_args == mock.call(mock_ctrl.return_value,
|
||||
DEFAULT_DETECTION_TIME)
|
||||
|
||||
|
||||
def test_config_error():
|
||||
|
|
@ -140,11 +148,11 @@ def test_scanner_update():
|
|||
"""Test the scanner update."""
|
||||
ctrl = mock.MagicMock()
|
||||
fake_clients = [
|
||||
{'mac': '123'},
|
||||
{'mac': '234'},
|
||||
{'mac': '123', 'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
|
||||
{'mac': '234', 'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
|
||||
]
|
||||
ctrl.get_clients.return_value = fake_clients
|
||||
unifi.UnifiScanner(ctrl)
|
||||
unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME)
|
||||
assert ctrl.get_clients.call_count == 1
|
||||
assert ctrl.get_clients.call_args == mock.call()
|
||||
|
||||
|
|
@ -154,18 +162,18 @@ def test_scanner_update_error():
|
|||
ctrl = mock.MagicMock()
|
||||
ctrl.get_clients.side_effect = APIError(
|
||||
'/', 500, 'foo', {}, None)
|
||||
unifi.UnifiScanner(ctrl)
|
||||
unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME)
|
||||
|
||||
|
||||
def test_scan_devices():
|
||||
"""Test the scanning for devices."""
|
||||
ctrl = mock.MagicMock()
|
||||
fake_clients = [
|
||||
{'mac': '123'},
|
||||
{'mac': '234'},
|
||||
{'mac': '123', 'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
|
||||
{'mac': '234', 'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
|
||||
]
|
||||
ctrl.get_clients.return_value = fake_clients
|
||||
scanner = unifi.UnifiScanner(ctrl)
|
||||
scanner = unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME)
|
||||
assert set(scanner.scan_devices()) == set(['123', '234'])
|
||||
|
||||
|
||||
|
|
@ -173,12 +181,17 @@ def test_get_device_name():
|
|||
"""Test the getting of device names."""
|
||||
ctrl = mock.MagicMock()
|
||||
fake_clients = [
|
||||
{'mac': '123', 'hostname': 'foobar'},
|
||||
{'mac': '234', 'name': 'Nice Name'},
|
||||
{'mac': '456'},
|
||||
{'mac': '123',
|
||||
'hostname': 'foobar',
|
||||
'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
|
||||
{'mac': '234',
|
||||
'name': 'Nice Name',
|
||||
'last_seen': dt_util.as_timestamp(dt_util.utcnow())},
|
||||
{'mac': '456',
|
||||
'last_seen': '1504786810'},
|
||||
]
|
||||
ctrl.get_clients.return_value = fake_clients
|
||||
scanner = unifi.UnifiScanner(ctrl)
|
||||
scanner = unifi.UnifiScanner(ctrl, DEFAULT_DETECTION_TIME)
|
||||
assert scanner.get_device_name('123') == 'foobar'
|
||||
assert scanner.get_device_name('234') == 'Nice Name'
|
||||
assert scanner.get_device_name('456') is None
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue