mirror of
https://github.com/home-assistant/core.git
synced 2025-04-25 09:47:52 +00:00
Switch unifi_direct to external library (#105046)
* switch to external library Signed-off-by: Tobias Perschon <tobias@perschon.at> * use mac as name if no hostname is available Signed-off-by: Tobias Perschon <tobias@perschon.at> * update requirements_test_all Signed-off-by: Tobias Perschon <tobias@perschon.at> * update .coveragerc Signed-off-by: Tobias Perschon <tobias@perschon.at> * update codeowners and remove old tests Signed-off-by: Tobias Perschon <tobias@perschon.at> * reverted get_device_name to old behaviour Signed-off-by: Tobias Perschon <tobias@perschon.at> * typing and some cleanup Signed-off-by: Tobias Perschon <tobias@perschon.at> * typing fix Signed-off-by: Tobias Perschon <tobias@perschon.at> * code cleanup Signed-off-by: Tobias Perschon <tobias@perschon.at> --------- Signed-off-by: Tobias Perschon <tobias@perschon.at>
This commit is contained in:
Tobias Perschon
GitHub
parent
0d2ec6cd5c
commit
c8f9285aba
@ -1427,6 +1427,8 @@ omit =
|
||||
homeassistant/components/ukraine_alarm/__init__.py
|
||||
homeassistant/components/ukraine_alarm/binary_sensor.py
|
||||
homeassistant/components/unifiled/*
|
||||
homeassistant/components/unifi_direct/__init__.py
|
||||
homeassistant/components/unifi_direct/device_tracker.py
|
||||
homeassistant/components/upb/__init__.py
|
||||
homeassistant/components/upb/light.py
|
||||
homeassistant/components/upc_connect/*
|
||||
|
||||
@ -1389,6 +1389,7 @@ build.json @home-assistant/supervisor
|
||||
/tests/components/ukraine_alarm/ @PaulAnnekov
|
||||
/homeassistant/components/unifi/ @Kane610
|
||||
/tests/components/unifi/ @Kane610
|
||||
/homeassistant/components/unifi_direct/ @tofuSCHNITZEL
|
||||
/homeassistant/components/unifiled/ @florisvdk
|
||||
/homeassistant/components/unifiprotect/ @AngellusMortis @bdraco
|
||||
/tests/components/unifiprotect/ @AngellusMortis @bdraco
|
||||
|
||||
@ -1,10 +1,10 @@
|
||||
"""Support for Unifi AP direct access."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pexpect import exceptions, pxssh
|
||||
from unifi_ap import UniFiAP, UniFiAPConnectionException, UniFiAPDataException
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.device_tracker import (
|
||||
@ -20,9 +20,6 @@ from homeassistant.helpers.typing import ConfigType
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_SSH_PORT = 22
|
||||
UNIFI_COMMAND = 'mca-dump | tr -d "\n"'
|
||||
UNIFI_SSID_TABLE = "vap_table"
|
||||
UNIFI_CLIENT_TABLE = "sta_table"
|
||||
|
||||
PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
@ -37,104 +34,43 @@ PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend(
|
||||
def get_scanner(hass: HomeAssistant, config: ConfigType) -> UnifiDeviceScanner | None:
|
||||
"""Validate the configuration and return a Unifi direct scanner."""
|
||||
scanner = UnifiDeviceScanner(config[DOMAIN])
|
||||
if not scanner.connected:
|
||||
return None
|
||||
return scanner
|
||||
return scanner if scanner.update_clients() else None
|
||||
|
||||
|
||||
class UnifiDeviceScanner(DeviceScanner):
|
||||
"""Class which queries Unifi wireless access point."""
|
||||
|
||||
def __init__(self, config):
|
||||
def __init__(self, config: ConfigType) -> None:
|
||||
"""Initialize the scanner."""
|
||||
self.host = config[CONF_HOST]
|
||||
self.username = config[CONF_USERNAME]
|
||||
self.password = config[CONF_PASSWORD]
|
||||
self.port = config[CONF_PORT]
|
||||
self.ssh = None
|
||||
self.connected = False
|
||||
self.last_results = {}
|
||||
self._connect()
|
||||
self.clients: dict[str, dict[str, Any]] = {}
|
||||
self.ap = UniFiAP(
|
||||
target=config[CONF_HOST],
|
||||
username=config[CONF_USERNAME],
|
||||
password=config[CONF_PASSWORD],
|
||||
port=config[CONF_PORT],
|
||||
)
|
||||
|
||||
def scan_devices(self):
|
||||
def scan_devices(self) -> list[str]:
|
||||
"""Scan for new devices and return a list with found device IDs."""
|
||||
result = _response_to_json(self._get_update())
|
||||
if result:
|
||||
self.last_results = result
|
||||
return self.last_results.keys()
|
||||
self.update_clients()
|
||||
return list(self.clients)
|
||||
|
||||
def get_device_name(self, device):
|
||||
def get_device_name(self, device: str) -> str | None:
|
||||
"""Return the name of the given device or None if we don't know."""
|
||||
hostname = next(
|
||||
(
|
||||
value.get("hostname")
|
||||
for key, value in self.last_results.items()
|
||||
if key.upper() == device.upper()
|
||||
),
|
||||
None,
|
||||
)
|
||||
if hostname is not None:
|
||||
hostname = str(hostname)
|
||||
return hostname
|
||||
|
||||
def _connect(self):
|
||||
"""Connect to the Unifi AP SSH server."""
|
||||
|
||||
self.ssh = pxssh.pxssh(options={"HostKeyAlgorithms": "ssh-rsa"})
|
||||
try:
|
||||
self.ssh.login(
|
||||
self.host, self.username, password=self.password, port=self.port
|
||||
)
|
||||
self.connected = True
|
||||
except exceptions.EOF:
|
||||
_LOGGER.error("Connection refused. SSH enabled?")
|
||||
self._disconnect()
|
||||
|
||||
def _disconnect(self):
|
||||
"""Disconnect the current SSH connection."""
|
||||
try:
|
||||
self.ssh.logout()
|
||||
except Exception: # pylint: disable=broad-except
|
||||
pass
|
||||
finally:
|
||||
self.ssh = None
|
||||
|
||||
self.connected = False
|
||||
|
||||
def _get_update(self):
|
||||
try:
|
||||
if not self.connected:
|
||||
self._connect()
|
||||
# If we still aren't connected at this point
|
||||
# don't try to send anything to the AP.
|
||||
if not self.connected:
|
||||
return None
|
||||
self.ssh.sendline(UNIFI_COMMAND)
|
||||
self.ssh.prompt()
|
||||
return self.ssh.before
|
||||
except pxssh.ExceptionPxssh as err:
|
||||
_LOGGER.error("Unexpected SSH error: %s", str(err))
|
||||
self._disconnect()
|
||||
return None
|
||||
except (AssertionError, exceptions.EOF) as err:
|
||||
_LOGGER.error("Connection to AP unavailable: %s", str(err))
|
||||
self._disconnect()
|
||||
client_info = self.clients.get(device)
|
||||
if client_info:
|
||||
return client_info.get("hostname")
|
||||
return None
|
||||
|
||||
|
||||
def _response_to_json(response):
|
||||
def update_clients(self) -> bool:
|
||||
"""Update the client info from AP."""
|
||||
try:
|
||||
json_response = json.loads(str(response)[31:-1].replace("\\", ""))
|
||||
_LOGGER.debug(str(json_response))
|
||||
ssid_table = json_response.get(UNIFI_SSID_TABLE)
|
||||
active_clients = {}
|
||||
self.clients = self.ap.get_clients()
|
||||
except UniFiAPConnectionException:
|
||||
_LOGGER.error("Failed to connect to accesspoint")
|
||||
return False
|
||||
except UniFiAPDataException:
|
||||
_LOGGER.error("Failed to get proper response from accesspoint")
|
||||
return False
|
||||
|
||||
for ssid in ssid_table:
|
||||
client_table = ssid.get(UNIFI_CLIENT_TABLE)
|
||||
for client in client_table:
|
||||
active_clients[client.get("mac")] = client
|
||||
|
||||
return active_clients
|
||||
except (ValueError, TypeError):
|
||||
_LOGGER.error("Failed to decode response from AP")
|
||||
return {}
|
||||
return True
|
||||
|
||||
@ -1,9 +1,9 @@
|
||||
{
|
||||
"domain": "unifi_direct",
|
||||
"name": "UniFi AP",
|
||||
"codeowners": [],
|
||||
"codeowners": ["@tofuSCHNITZEL"],
|
||||
"documentation": "https://www.home-assistant.io/integrations/unifi_direct",
|
||||
"iot_class": "local_polling",
|
||||
"loggers": ["pexpect", "ptyprocess"],
|
||||
"requirements": ["pexpect==4.6.0"]
|
||||
"loggers": ["unifi_ap"],
|
||||
"requirements": ["unifi_ap==0.0.1"]
|
||||
}
|
||||
|
||||
@ -1478,7 +1478,6 @@ pescea==1.0.12
|
||||
# homeassistant.components.aruba
|
||||
# homeassistant.components.cisco_ios
|
||||
# homeassistant.components.pandora
|
||||
# homeassistant.components.unifi_direct
|
||||
pexpect==4.6.0
|
||||
|
||||
# homeassistant.components.modem_callerid
|
||||
@ -2698,6 +2697,9 @@ ultraheat-api==0.5.7
|
||||
# homeassistant.components.unifiprotect
|
||||
unifi-discovery==1.1.7
|
||||
|
||||
# homeassistant.components.unifi_direct
|
||||
unifi_ap==0.0.1
|
||||
|
||||
# homeassistant.components.unifiled
|
||||
unifiled==0.11
|
||||
|
||||
|
||||
@ -1142,12 +1142,6 @@ peco==0.0.29
|
||||
# homeassistant.components.escea
|
||||
pescea==1.0.12
|
||||
|
||||
# homeassistant.components.aruba
|
||||
# homeassistant.components.cisco_ios
|
||||
# homeassistant.components.pandora
|
||||
# homeassistant.components.unifi_direct
|
||||
pexpect==4.6.0
|
||||
|
||||
# homeassistant.components.modem_callerid
|
||||
phone-modem==0.1.1
|
||||
|
||||
|
||||
@ -1 +0,0 @@
|
||||
"""Tests for the unifi_direct component."""
|
||||
File diff suppressed because one or more lines are too long
@ -1,178 +0,0 @@
|
||||
"""The tests for the Unifi direct device tracker platform."""
|
||||
from datetime import timedelta
|
||||
import os
|
||||
from unittest.mock import MagicMock, call, patch
|
||||
|
||||
import pytest
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.device_tracker import (
|
||||
CONF_CONSIDER_HOME,
|
||||
CONF_NEW_DEVICE_DEFAULTS,
|
||||
CONF_TRACK_NEW,
|
||||
)
|
||||
from homeassistant.components.device_tracker.legacy import YAML_DEVICES
|
||||
from homeassistant.components.unifi_direct.device_tracker import (
|
||||
CONF_PORT,
|
||||
DOMAIN,
|
||||
PLATFORM_SCHEMA,
|
||||
UnifiDeviceScanner,
|
||||
_response_to_json,
|
||||
get_scanner,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PLATFORM, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import assert_setup_component, load_fixture, mock_component
|
||||
|
||||
scanner_path = "homeassistant.components.unifi_direct.device_tracker.UnifiDeviceScanner"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_comp(hass):
|
||||
"""Initialize components."""
|
||||
mock_component(hass, "zone")
|
||||
yaml_devices = hass.config.path(YAML_DEVICES)
|
||||
yield
|
||||
if os.path.isfile(yaml_devices):
|
||||
os.remove(yaml_devices)
|
||||
|
||||
|
||||
@patch(scanner_path, return_value=MagicMock(spec=UnifiDeviceScanner))
|
||||
async def test_get_scanner(unifi_mock, hass: HomeAssistant) -> None:
|
||||
"""Test creating an Unifi direct scanner with a password."""
|
||||
conf_dict = {
|
||||
DOMAIN: {
|
||||
CONF_PLATFORM: "unifi_direct",
|
||||
CONF_HOST: "fake_host",
|
||||
CONF_USERNAME: "fake_user",
|
||||
CONF_PASSWORD: "fake_pass",
|
||||
CONF_TRACK_NEW: True,
|
||||
CONF_CONSIDER_HOME: timedelta(seconds=180),
|
||||
CONF_NEW_DEVICE_DEFAULTS: {CONF_TRACK_NEW: True},
|
||||
}
|
||||
}
|
||||
|
||||
with assert_setup_component(1, DOMAIN):
|
||||
assert await async_setup_component(hass, DOMAIN, conf_dict)
|
||||
|
||||
conf_dict[DOMAIN][CONF_PORT] = 22
|
||||
assert unifi_mock.call_args == call(conf_dict[DOMAIN])
|
||||
|
||||
|
||||
@patch("pexpect.pxssh.pxssh")
|
||||
async def test_get_device_name(mock_ssh, hass: HomeAssistant) -> None:
|
||||
"""Testing MAC matching."""
|
||||
conf_dict = {
|
||||
DOMAIN: {
|
||||
CONF_PLATFORM: "unifi_direct",
|
||||
CONF_HOST: "fake_host",
|
||||
CONF_USERNAME: "fake_user",
|
||||
CONF_PASSWORD: "fake_pass",
|
||||
CONF_PORT: 22,
|
||||
CONF_TRACK_NEW: True,
|
||||
CONF_CONSIDER_HOME: timedelta(seconds=180),
|
||||
}
|
||||
}
|
||||
mock_ssh.return_value.before = load_fixture("data.txt", "unifi_direct")
|
||||
scanner = get_scanner(hass, conf_dict)
|
||||
devices = scanner.scan_devices()
|
||||
assert len(devices) == 23
|
||||
assert scanner.get_device_name("98:00:c6:56:34:12") == "iPhone"
|
||||
assert scanner.get_device_name("98:00:C6:56:34:12") == "iPhone"
|
||||
|
||||
|
||||
@patch("pexpect.pxssh.pxssh.logout")
|
||||
@patch("pexpect.pxssh.pxssh.login")
|
||||
async def test_failed_to_log_in(mock_login, mock_logout, hass: HomeAssistant) -> None:
|
||||
"""Testing exception at login results in False."""
|
||||
from pexpect import exceptions
|
||||
|
||||
conf_dict = {
|
||||
DOMAIN: {
|
||||
CONF_PLATFORM: "unifi_direct",
|
||||
CONF_HOST: "fake_host",
|
||||
CONF_USERNAME: "fake_user",
|
||||
CONF_PASSWORD: "fake_pass",
|
||||
CONF_PORT: 22,
|
||||
CONF_TRACK_NEW: True,
|
||||
CONF_CONSIDER_HOME: timedelta(seconds=180),
|
||||
}
|
||||
}
|
||||
|
||||
mock_login.side_effect = exceptions.EOF("Test")
|
||||
scanner = get_scanner(hass, conf_dict)
|
||||
assert not scanner
|
||||
|
||||
|
||||
@patch("pexpect.pxssh.pxssh.logout")
|
||||
@patch("pexpect.pxssh.pxssh.login", autospec=True)
|
||||
@patch("pexpect.pxssh.pxssh.prompt")
|
||||
@patch("pexpect.pxssh.pxssh.sendline")
|
||||
async def test_to_get_update(
|
||||
mock_sendline, mock_prompt, mock_login, mock_logout, hass: HomeAssistant
|
||||
) -> None:
|
||||
"""Testing exception in get_update matching."""
|
||||
conf_dict = {
|
||||
DOMAIN: {
|
||||
CONF_PLATFORM: "unifi_direct",
|
||||
CONF_HOST: "fake_host",
|
||||
CONF_USERNAME: "fake_user",
|
||||
CONF_PASSWORD: "fake_pass",
|
||||
CONF_PORT: 22,
|
||||
CONF_TRACK_NEW: True,
|
||||
CONF_CONSIDER_HOME: timedelta(seconds=180),
|
||||
}
|
||||
}
|
||||
|
||||
scanner = get_scanner(hass, conf_dict)
|
||||
# mock_sendline.side_effect = AssertionError("Test")
|
||||
mock_prompt.side_effect = AssertionError("Test")
|
||||
devices = scanner._get_update()
|
||||
assert devices is None
|
||||
|
||||
|
||||
def test_good_response_parses(hass: HomeAssistant) -> None:
|
||||
"""Test that the response form the AP parses to JSON correctly."""
|
||||
response = _response_to_json(load_fixture("data.txt", "unifi_direct"))
|
||||
assert response != {}
|
||||
|
||||
|
||||
def test_bad_response_returns_none(hass: HomeAssistant) -> None:
|
||||
"""Test that a bad response form the AP parses to JSON correctly."""
|
||||
assert _response_to_json("{(}") == {}
|
||||
|
||||
|
||||
def test_config_error() -> None:
|
||||
"""Test for configuration errors."""
|
||||
with pytest.raises(vol.Invalid):
|
||||
PLATFORM_SCHEMA(
|
||||
{
|
||||
# no username
|
||||
CONF_PASSWORD: "password",
|
||||
CONF_PLATFORM: DOMAIN,
|
||||
CONF_HOST: "myhost",
|
||||
"port": 123,
|
||||
}
|
||||
)
|
||||
with pytest.raises(vol.Invalid):
|
||||
PLATFORM_SCHEMA(
|
||||
{
|
||||
# no password
|
||||
CONF_USERNAME: "foo",
|
||||
CONF_PLATFORM: DOMAIN,
|
||||
CONF_HOST: "myhost",
|
||||
"port": 123,
|
||||
}
|
||||
)
|
||||
with pytest.raises(vol.Invalid):
|
||||
PLATFORM_SCHEMA(
|
||||
{
|
||||
CONF_PLATFORM: DOMAIN,
|
||||
CONF_USERNAME: "foo",
|
||||
CONF_PASSWORD: "password",
|
||||
CONF_HOST: "myhost",
|
||||
"port": "foo", # bad port!
|
||||
}
|
||||
)
|
||||
Loading…
x
Reference in New Issue
Block a user