Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve UniFi tests (#45871)

pull/46063/head
Robert Svensson 2021-02-05 16:31:47 +01:00 committed by GitHub
parent 7144c5f316
commit ae2c7e4c74
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 315 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
tests/components/unifi/test_config_flow.py
View File

@ -191,7 +191,7 @@ async def test_flow_works_multiple_sites(hass, aioclient_mock):
async def test_flow_raise_already_configured(hass, aioclient_mock):
"""Test config flow aborts since a connected config entry already exists."""
await setup_unifi_integration(hass)
await setup_unifi_integration(hass, aioclient_mock)
result = await hass.config_entries.flow.async_init(
UNIFI_DOMAIN, context={"source": "user"}
@ -200,6 +200,8 @@ async def test_flow_raise_already_configured(hass, aioclient_mock):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
aioclient_mock.clear_requests()
aioclient_mock.get("https://1.2.3.4:1234", status=302)
aioclient_mock.post(
@ -340,17 +342,19 @@ async def test_flow_fails_controller_unavailable(hass, aioclient_mock):
async def test_reauth_flow_update_configuration(hass, aioclient_mock):
"""Verify reauth flow can update controller configuration."""
controller = await setup_unifi_integration(hass)
config_entry = await setup_unifi_integration(hass, aioclient_mock)
result = await hass.config_entries.flow.async_init(
UNIFI_DOMAIN,
context={"source": SOURCE_REAUTH},
data=controller.config_entry,
data=config_entry,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == SOURCE_USER
aioclient_mock.clear_requests()
aioclient_mock.get("https://1.2.3.4:1234", status=302)
aioclient_mock.post(
@ -381,15 +385,16 @@ async def test_reauth_flow_update_configuration(hass, aioclient_mock):
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful"
assert controller.host == "1.2.3.4"
assert controller.config_entry.data[CONF_CONTROLLER][CONF_USERNAME] == "new_name"
assert controller.config_entry.data[CONF_CONTROLLER][CONF_PASSWORD] == "new_pass"
assert config_entry.data[CONF_CONTROLLER][CONF_HOST] == "1.2.3.4"
assert config_entry.data[CONF_CONTROLLER][CONF_USERNAME] == "new_name"
assert config_entry.data[CONF_CONTROLLER][CONF_PASSWORD] == "new_pass"
async def test_advanced_option_flow(hass):
async def test_advanced_option_flow(hass, aioclient_mock):
"""Test advanced config flow options."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
clients_response=CLIENTS,
devices_response=DEVICES,
wlans_response=WLANS,
@ -398,7 +403,7 @@ async def test_advanced_option_flow(hass):
)
result = await hass.config_entries.options.async_init(
controller.config_entry.entry_id, context={"show_advanced_options": True}
config_entry.entry_id, context={"show_advanced_options": True}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
@ -457,10 +462,11 @@ async def test_advanced_option_flow(hass):
}
async def test_simple_option_flow(hass):
async def test_simple_option_flow(hass, aioclient_mock):
"""Test simple config flow options."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
clients_response=CLIENTS,
wlans_response=WLANS,
dpigroup_response=DPI_GROUPS,
@ -468,7 +474,7 @@ async def test_simple_option_flow(hass):
)
result = await hass.config_entries.options.async_init(
controller.config_entry.entry_id, context={"show_advanced_options": False}
config_entry.entry_id, context={"show_advanced_options": False}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM

194
tests/components/unifi/test_controller.py
View File

@ -1,5 +1,5 @@
"""Test UniFi Controller."""
from collections import deque
from copy import deepcopy
from datetime import timedelta
from unittest.mock import patch
@ -33,14 +33,18 @@ from homeassistant.const import (
CONF_PORT,
CONF_USERNAME,
CONF_VERIFY_SSL,
CONTENT_TYPE_JSON,
)
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
DEFAULT_HOST = "1.2.3.4"
DEFAULT_SITE = "site_id"
CONTROLLER_HOST = {
"hostname": "controller_host",
"ip": "1.2.3.4",
"ip": DEFAULT_HOST,
"is_wired": True,
"last_seen": 1562600145,
"mac": "10:00:00:00:00:01",
@ -54,11 +58,11 @@ CONTROLLER_HOST = {
}
CONTROLLER_DATA = {
CONF_HOST: "1.2.3.4",
CONF_HOST: DEFAULT_HOST,
CONF_USERNAME: "username",
CONF_PASSWORD: "password",
CONF_PORT: 1234,
CONF_SITE_ID: "site_id",
CONF_SITE_ID: DEFAULT_SITE,
CONF_VERIFY_SSL: False,
}
@ -67,22 +71,90 @@ ENTRY_OPTIONS = {}
CONFIGURATION = []
SITES = {"Site name": {"desc": "Site name", "name": "site_id", "role": "admin"}}
SITE = [{"desc": "Site name", "name": "site_id", "role": "admin"}]
DESCRIPTION = [{"name": "username", "site_name": "site_id", "site_role": "admin"}]
def mock_default_unifi_requests(
aioclient_mock,
host,
site_id,
sites=None,
description=None,
clients_response=None,
clients_all_response=None,
devices_response=None,
dpiapp_response=None,
dpigroup_response=None,
wlans_response=None,
):
"""Mock default UniFi requests responses."""
aioclient_mock.get(f"https://{host}:1234", status=302) # Check UniFi OS
aioclient_mock.post(
f"https://{host}:1234/api/login",
json={"data": "login successful", "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
f"https://{host}:1234/api/self/sites",
json={"data": sites or [], "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
f"https://{host}:1234/api/s/{site_id}/self",
json={"data": description or [], "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
f"https://{host}:1234/api/s/{site_id}/stat/sta",
json={"data": clients_response or [], "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
f"https://{host}:1234/api/s/{site_id}/rest/user",
json={"data": clients_all_response or [], "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
f"https://{host}:1234/api/s/{site_id}/stat/device",
json={"data": devices_response or [], "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
f"https://{host}:1234/api/s/{site_id}/rest/dpiapp",
json={"data": dpiapp_response or [], "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
f"https://{host}:1234/api/s/{site_id}/rest/dpigroup",
json={"data": dpigroup_response or [], "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
aioclient_mock.get(
f"https://{host}:1234/api/s/{site_id}/rest/wlanconf",
json={"data": wlans_response or [], "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
async def setup_unifi_integration(
hass,
aioclient_mock=None,
*,
config=ENTRY_CONFIG,
options=ENTRY_OPTIONS,
sites=SITES,
sites=SITE,
site_description=DESCRIPTION,
clients_response=None,
devices_response=None,
clients_all_response=None,
wlans_response=None,
dpigroup_response=None,
devices_response=None,
dpiapp_response=None,
dpigroup_response=None,
wlans_response=None,
known_wireless_clients=None,
controllers=None,
):
@ -102,82 +174,39 @@ async def setup_unifi_integration(
known_wireless_clients, config_entry
)
mock_client_responses = deque()
if clients_response:
mock_client_responses.append(clients_response)
if aioclient_mock:
mock_default_unifi_requests(
aioclient_mock,
host=config_entry.data[CONF_CONTROLLER][CONF_HOST],
site_id=config_entry.data[CONF_CONTROLLER][CONF_SITE_ID],
sites=sites,
description=site_description,
clients_response=clients_response,
clients_all_response=clients_all_response,
devices_response=devices_response,
dpiapp_response=dpiapp_response,
dpigroup_response=dpigroup_response,
wlans_response=wlans_response,
)
mock_device_responses = deque()
if devices_response:
mock_device_responses.append(devices_response)
mock_client_all_responses = deque()
if clients_all_response:
mock_client_all_responses.append(clients_all_response)
mock_wlans_responses = deque()
if wlans_response:
mock_wlans_responses.append(wlans_response)
mock_dpigroup_responses = deque()
if dpigroup_response:
mock_dpigroup_responses.append(dpigroup_response)
mock_dpiapp_responses = deque()
if dpiapp_response:
mock_dpiapp_responses.append(dpiapp_response)
mock_requests = []
async def mock_request(self, method, path, json=None):
mock_requests.append({"method": method, "path": path, "json": json})
if path == "/stat/sta" and mock_client_responses:
return mock_client_responses.popleft()
if path == "/stat/device" and mock_device_responses:
return mock_device_responses.popleft()
if path == "/rest/user" and mock_client_all_responses:
return mock_client_all_responses.popleft()
if path == "/rest/wlanconf" and mock_wlans_responses:
return mock_wlans_responses.popleft()
if path == "/rest/dpigroup" and mock_dpigroup_responses:
return mock_dpigroup_responses.popleft()
if path == "/rest/dpiapp" and mock_dpiapp_responses:
return mock_dpiapp_responses.popleft()
return {}
with patch("aiounifi.Controller.check_unifi_os", return_value=True), patch(
"aiounifi.Controller.login",
return_value=True,
), patch("aiounifi.Controller.sites", return_value=sites), patch(
"aiounifi.Controller.site_description", return_value=site_description
), patch(
"aiounifi.Controller.request", new=mock_request
), patch.object(
aiounifi.websocket.WSClient, "start", return_value=True
):
with patch.object(aiounifi.websocket.WSClient, "start", return_value=True):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
if config_entry.entry_id not in hass.data[UNIFI_DOMAIN]:
return None
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
controller.mock_client_responses = mock_client_responses
controller.mock_device_responses = mock_device_responses
controller.mock_client_all_responses = mock_client_all_responses
controller.mock_wlans_responses = mock_wlans_responses
controller.mock_requests = mock_requests
return controller
return config_entry
async def test_controller_setup(hass):
async def test_controller_setup(hass, aioclient_mock):
"""Successful setup."""
with patch(
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setup",
return_value=True,
) as forward_entry_setup:
controller = await setup_unifi_integration(hass)
config_entry = await setup_unifi_integration(hass, aioclient_mock)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
entry = controller.config_entry
assert len(forward_entry_setup.mock_calls) == len(SUPPORTED_PLATFORMS)
@ -187,8 +216,8 @@ async def test_controller_setup(hass):
assert controller.host == CONTROLLER_DATA[CONF_HOST]
assert controller.site == CONTROLLER_DATA[CONF_SITE_ID]
assert controller.site_name in SITES
assert controller.site_role == SITES[controller.site_name]["role"]
assert controller.site_name == SITE[0]["desc"]
assert controller.site_role == SITE[0]["role"]
assert controller.option_allow_bandwidth_sensors == DEFAULT_ALLOW_BANDWIDTH_SENSORS
assert controller.option_allow_uptime_sensors == DEFAULT_ALLOW_UPTIME_SENSORS
@ -208,9 +237,12 @@ async def test_controller_setup(hass):
assert controller.signal_heartbeat_missed == "unifi-heartbeat-missed"
async def test_controller_mac(hass):
async def test_controller_mac(hass, aioclient_mock):
"""Test that it is possible to identify controller mac."""
controller = await setup_unifi_integration(hass, clients_response=[CONTROLLER_HOST])
config_entry = await setup_unifi_integration(
hass, aioclient_mock, clients_response=[CONTROLLER_HOST]
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert controller.mac == CONTROLLER_HOST["mac"]
@ -245,9 +277,10 @@ async def test_controller_unknown_error(hass):
assert hass.data[UNIFI_DOMAIN] == {}
async def test_reset_after_successful_setup(hass):
async def test_reset_after_successful_setup(hass, aioclient_mock):
"""Calling reset when the entry has been setup."""
controller = await setup_unifi_integration(hass)
config_entry = await setup_unifi_integration(hass, aioclient_mock)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(controller.listeners) == 6
@ -258,9 +291,12 @@ async def test_reset_after_successful_setup(hass):
assert len(controller.listeners) == 0
async def test_wireless_client_event_calls_update_wireless_devices(hass):
async def test_wireless_client_event_calls_update_wireless_devices(
hass, aioclient_mock
):
"""Call update_wireless_devices method when receiving wireless client event."""
controller = await setup_unifi_integration(hass)
config_entry = await setup_unifi_integration(hass, aioclient_mock)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
with patch(
"homeassistant.components.unifi.controller.UniFiController.update_wireless_clients",

127
tests/components/unifi/test_device_tracker.py
View File

@ -25,7 +25,6 @@ from homeassistant.components.unifi.const import (
)
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.helpers import entity_registry
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from .test_controller import ENTRY_CONFIG, setup_unifi_integration
@ -151,27 +150,19 @@ EVENT_DEVICE_2_UPGRADED = {
}
async def test_platform_manually_configured(hass):
"""Test that nothing happens when configuring unifi through device tracker platform."""
assert (
await async_setup_component(
hass, TRACKER_DOMAIN, {TRACKER_DOMAIN: {"platform": UNIFI_DOMAIN}}
)
is False
)
assert UNIFI_DOMAIN not in hass.data
async def test_no_clients(hass):
async def test_no_clients(hass, aioclient_mock):
"""Test the update_clients function when no clients are found."""
await setup_unifi_integration(hass)
await setup_unifi_integration(hass, aioclient_mock)
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 0
async def test_tracked_wireless_clients(hass):
async def test_tracked_wireless_clients(hass, aioclient_mock):
"""Test the update_items function with some clients."""
controller = await setup_unifi_integration(hass, clients_response=[CLIENT_1])
config_entry = await setup_unifi_integration(
hass, aioclient_mock, clients_response=[CLIENT_1]
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 1
client_1 = hass.states.get("device_tracker.client_1")
@ -224,17 +215,19 @@ async def test_tracked_wireless_clients(hass):
assert client_1.state == "home"
async def test_tracked_clients(hass):
async def test_tracked_clients(hass, aioclient_mock):
"""Test the update_items function with some clients."""
client_4_copy = copy(CLIENT_4)
client_4_copy["last_seen"] = dt_util.as_timestamp(dt_util.utcnow())
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_SSID_FILTER: ["ssid"]},
clients_response=[CLIENT_1, CLIENT_2, CLIENT_3, CLIENT_5, client_4_copy],
known_wireless_clients=(CLIENT_4["mac"],),
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 4
client_1 = hass.states.get("device_tracker.client_1")
@ -269,12 +262,14 @@ async def test_tracked_clients(hass):
assert client_1.state == "home"
async def test_tracked_devices(hass):
async def test_tracked_devices(hass, aioclient_mock):
"""Test the update_items function with some devices."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
devices_response=[DEVICE_1, DEVICE_2],
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 2
device_1 = hass.states.get("device_tracker.device_1")
@ -338,11 +333,12 @@ async def test_tracked_devices(hass):
assert device.sw_version == EVENT_DEVICE_2_UPGRADED["version_to"]
async def test_remove_clients(hass):
async def test_remove_clients(hass, aioclient_mock):
"""Test the remove_items function with some clients."""
controller = await setup_unifi_integration(
hass, clients_response=[CLIENT_1, CLIENT_2]
config_entry = await setup_unifi_integration(
hass, aioclient_mock, clients_response=[CLIENT_1, CLIENT_2]
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 2
client_1 = hass.states.get("device_tracker.client_1")
@ -367,13 +363,15 @@ async def test_remove_clients(hass):
assert wired_client is not None
async def test_controller_state_change(hass):
async def test_controller_state_change(hass, aioclient_mock):
"""Verify entities state reflect on controller becoming unavailable."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
clients_response=[CLIENT_1],
devices_response=[DEVICE_1],
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 2
client_1 = hass.states.get("device_tracker.client_1")
@ -405,10 +403,11 @@ async def test_controller_state_change(hass):
assert device_1.state == "home"
async def test_option_track_clients(hass):
async def test_option_track_clients(hass, aioclient_mock):
"""Test the tracking of clients can be turned off."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
clients_response=[CLIENT_1, CLIENT_2],
devices_response=[DEVICE_1],
)
@ -424,7 +423,7 @@ async def test_option_track_clients(hass):
assert device_1 is not None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_TRACK_CLIENTS: False},
)
await hass.async_block_till_done()
@ -439,7 +438,7 @@ async def test_option_track_clients(hass):
assert device_1 is not None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_TRACK_CLIENTS: True},
)
await hass.async_block_till_done()
@ -454,10 +453,11 @@ async def test_option_track_clients(hass):
assert device_1 is not None
async def test_option_track_wired_clients(hass):
async def test_option_track_wired_clients(hass, aioclient_mock):
"""Test the tracking of wired clients can be turned off."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
clients_response=[CLIENT_1, CLIENT_2],
devices_response=[DEVICE_1],
)
@ -473,7 +473,7 @@ async def test_option_track_wired_clients(hass):
assert device_1 is not None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_TRACK_WIRED_CLIENTS: False},
)
await hass.async_block_till_done()
@ -488,7 +488,7 @@ async def test_option_track_wired_clients(hass):
assert device_1 is not None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_TRACK_WIRED_CLIENTS: True},
)
await hass.async_block_till_done()
@ -503,10 +503,11 @@ async def test_option_track_wired_clients(hass):
assert device_1 is not None
async def test_option_track_devices(hass):
async def test_option_track_devices(hass, aioclient_mock):
"""Test the tracking of devices can be turned off."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
clients_response=[CLIENT_1, CLIENT_2],
devices_response=[DEVICE_1],
)
@ -522,7 +523,7 @@ async def test_option_track_devices(hass):
assert device_1 is not None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_TRACK_DEVICES: False},
)
await hass.async_block_till_done()
@ -537,7 +538,7 @@ async def test_option_track_devices(hass):
assert device_1 is None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_TRACK_DEVICES: True},
)
await hass.async_block_till_done()
@ -552,7 +553,7 @@ async def test_option_track_devices(hass):
assert device_1 is not None
async def test_option_ssid_filter(hass):
async def test_option_ssid_filter(hass, aioclient_mock):
"""Test the SSID filter works.
Client 1 will travel from a supported SSID to an unsupported ssid.
@ -561,9 +562,10 @@ async def test_option_ssid_filter(hass):
client_1_copy = copy(CLIENT_1)
client_1_copy["last_seen"] = dt_util.as_timestamp(dt_util.utcnow())
controller = await setup_unifi_integration(
hass, clients_response=[client_1_copy, CLIENT_3]
config_entry = await setup_unifi_integration(
hass, aioclient_mock, clients_response=[client_1_copy, CLIENT_3]
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 2
client_1 = hass.states.get("device_tracker.client_1")
@ -574,7 +576,7 @@ async def test_option_ssid_filter(hass):
# Setting SSID filter will remove clients outside of filter
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_SSID_FILTER: ["ssid"]},
)
await hass.async_block_till_done()
@ -609,7 +611,7 @@ async def test_option_ssid_filter(hass):
# Remove SSID filter
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_SSID_FILTER: []},
)
await hass.async_block_till_done()
@ -656,7 +658,7 @@ async def test_option_ssid_filter(hass):
assert client_3.state == "not_home"
async def test_wireless_client_go_wired_issue(hass):
async def test_wireless_client_go_wired_issue(hass, aioclient_mock):
"""Test the solution to catch wireless device go wired UniFi issue.
UniFi has a known issue that when a wireless device goes away it sometimes gets marked as wired.
@ -664,7 +666,10 @@ async def test_wireless_client_go_wired_issue(hass):
client_1_client = copy(CLIENT_1)
client_1_client["last_seen"] = dt_util.as_timestamp(dt_util.utcnow())
controller = await setup_unifi_integration(hass, clients_response=[client_1_client])
config_entry = await setup_unifi_integration(
hass, aioclient_mock, clients_response=[client_1_client]
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 1
# Client is wireless
@ -717,14 +722,18 @@ async def test_wireless_client_go_wired_issue(hass):
assert client_1.attributes["is_wired"] is False
async def test_option_ignore_wired_bug(hass):
async def test_option_ignore_wired_bug(hass, aioclient_mock):
"""Test option to ignore wired bug."""
client_1_client = copy(CLIENT_1)
client_1_client["last_seen"] = dt_util.as_timestamp(dt_util.utcnow())
controller = await setup_unifi_integration(
hass, options={CONF_IGNORE_WIRED_BUG: True}, clients_response=[client_1_client]
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_IGNORE_WIRED_BUG: True},
clients_response=[client_1_client],
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 1
# Client is wireless
@ -777,7 +786,7 @@ async def test_option_ignore_wired_bug(hass):
assert client_1.attributes["is_wired"] is False
async def test_restoring_client(hass):
async def test_restoring_client(hass, aioclient_mock):
"""Test the update_items function with some clients."""
config_entry = config_entries.ConfigEntry(
version=1,
@ -809,6 +818,7 @@ async def test_restoring_client(hass):
await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_BLOCK_CLIENT: True},
clients_response=[CLIENT_2],
clients_all_response=[CLIENT_1],
@ -819,10 +829,11 @@ async def test_restoring_client(hass):
assert device_1 is not None
async def test_dont_track_clients(hass):
async def test_dont_track_clients(hass, aioclient_mock):
"""Test don't track clients config works."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_TRACK_CLIENTS: False},
clients_response=[CLIENT_1],
devices_response=[DEVICE_1],
@ -836,7 +847,7 @@ async def test_dont_track_clients(hass):
assert device_1 is not None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_TRACK_CLIENTS: True},
)
await hass.async_block_till_done()
@ -850,10 +861,11 @@ async def test_dont_track_clients(hass):
assert device_1 is not None
async def test_dont_track_devices(hass):
async def test_dont_track_devices(hass, aioclient_mock):
"""Test don't track devices config works."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_TRACK_DEVICES: False},
clients_response=[CLIENT_1],
devices_response=[DEVICE_1],
@ -867,7 +879,7 @@ async def test_dont_track_devices(hass):
assert device_1 is None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_TRACK_DEVICES: True},
)
await hass.async_block_till_done()
@ -881,10 +893,11 @@ async def test_dont_track_devices(hass):
assert device_1 is not None
async def test_dont_track_wired_clients(hass):
async def test_dont_track_wired_clients(hass, aioclient_mock):
"""Test don't track wired clients config works."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_TRACK_WIRED_CLIENTS: False},
clients_response=[CLIENT_1, CLIENT_2],
)
@ -897,7 +910,7 @@ async def test_dont_track_wired_clients(hass):
assert client_2 is None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_TRACK_WIRED_CLIENTS: True},
)
await hass.async_block_till_done()

13
tests/components/unifi/test_init.py
View File

@ -11,14 +11,14 @@ from tests.common import MockConfigEntry, mock_coro
async def test_setup_with_no_config(hass):
"""Test that we do not discover anything or try to set up a bridge."""
"""Test that we do not discover anything or try to set up a controller."""
assert await async_setup_component(hass, UNIFI_DOMAIN, {}) is True
assert UNIFI_DOMAIN not in hass.data
async def test_successful_config_entry(hass):
async def test_successful_config_entry(hass, aioclient_mock):
"""Test that configured options for a host are loaded via config entry."""
await setup_unifi_integration(hass)
await setup_unifi_integration(hass, aioclient_mock)
assert hass.data[UNIFI_DOMAIN]
@ -44,7 +44,6 @@ async def test_controller_no_mac(hass):
"site": "default",
"verify_ssl": True,
},
"poe_control": True,
},
)
entry.add_to_hass(hass)
@ -64,10 +63,10 @@ async def test_controller_no_mac(hass):
assert len(mock_registry.mock_calls) == 0
async def test_unload_entry(hass):
async def test_unload_entry(hass, aioclient_mock):
"""Test being able to unload an entry."""
controller = await setup_unifi_integration(hass)
config_entry = await setup_unifi_integration(hass, aioclient_mock)
assert hass.data[UNIFI_DOMAIN]
assert await unifi.async_unload_entry(hass, controller.config_entry)
assert await unifi.async_unload_entry(hass, config_entry)
assert not hass.data[UNIFI_DOMAIN]

35
tests/components/unifi/test_sensor.py
View File

@ -14,7 +14,6 @@ from homeassistant.components.unifi.const import (
DOMAIN as UNIFI_DOMAIN,
)
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_setup_component
from .test_controller import setup_unifi_integration
@ -50,35 +49,25 @@ CLIENTS = [
]
async def test_platform_manually_configured(hass):
"""Test that we do not discover anything or try to set up a controller."""
assert (
await async_setup_component(
hass, SENSOR_DOMAIN, {SENSOR_DOMAIN: {"platform": UNIFI_DOMAIN}}
)
is True
)
assert UNIFI_DOMAIN not in hass.data
async def test_no_clients(hass):
async def test_no_clients(hass, aioclient_mock):
"""Test the update_clients function when no clients are found."""
controller = await setup_unifi_integration(
await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_ALLOW_BANDWIDTH_SENSORS: True,
CONF_ALLOW_UPTIME_SENSORS: True,
},
)
assert len(controller.mock_requests) == 6
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 0
async def test_sensors(hass):
async def test_sensors(hass, aioclient_mock):
"""Test the update_items function with some clients."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_ALLOW_BANDWIDTH_SENSORS: True,
CONF_ALLOW_UPTIME_SENSORS: True,
@ -87,8 +76,8 @@ async def test_sensors(hass):
},
clients_response=CLIENTS,
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(controller.mock_requests) == 6
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 6
wired_client_rx = hass.states.get("sensor.wired_client_name_rx")
@ -129,7 +118,7 @@ async def test_sensors(hass):
assert wireless_client_uptime.state == "2020-09-15T14:41:00+00:00"
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={
CONF_ALLOW_BANDWIDTH_SENSORS: False,
CONF_ALLOW_UPTIME_SENSORS: False,
@ -150,7 +139,7 @@ async def test_sensors(hass):
assert wireless_client_uptime is None
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={
CONF_ALLOW_BANDWIDTH_SENSORS: True,
CONF_ALLOW_UPTIME_SENSORS: True,
@ -189,16 +178,18 @@ async def test_sensors(hass):
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 6
async def test_remove_sensors(hass):
async def test_remove_sensors(hass, aioclient_mock):
"""Test the remove_items function with some clients."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_ALLOW_BANDWIDTH_SENSORS: True,
CONF_ALLOW_UPTIME_SENSORS: True,
},
clients_response=CLIENTS,
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 6
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 2

190
tests/components/unifi/test_switch.py
View File

@ -17,7 +17,6 @@ from homeassistant.components.unifi.const import (
)
from homeassistant.components.unifi.switch import POE_SWITCH
from homeassistant.helpers import entity_registry
from homeassistant.setup import async_setup_component
from .test_controller import (
CONTROLLER_HOST,
@ -282,21 +281,11 @@ DPI_APPS = [
]
async def test_platform_manually_configured(hass):
"""Test that we do not discover anything or try to set up a controller."""
assert (
await async_setup_component(
hass, SWITCH_DOMAIN, {SWITCH_DOMAIN: {"platform": UNIFI_DOMAIN}}
)
is True
)
assert UNIFI_DOMAIN not in hass.data
async def test_no_clients(hass):
async def test_no_clients(hass, aioclient_mock):
"""Test the update_clients function when no clients are found."""
controller = await setup_unifi_integration(
await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_TRACK_CLIENTS: False,
CONF_TRACK_DEVICES: False,
@ -304,45 +293,46 @@ async def test_no_clients(hass):
},
)
assert len(controller.mock_requests) == 6
assert aioclient_mock.call_count == 10
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
async def test_controller_not_client(hass):
async def test_controller_not_client(hass, aioclient_mock):
"""Test that the controller doesn't become a switch."""
controller = await setup_unifi_integration(
await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_TRACK_CLIENTS: False, CONF_TRACK_DEVICES: False},
clients_response=[CONTROLLER_HOST],
devices_response=[DEVICE_1],
)
assert len(controller.mock_requests) == 6
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
cloudkey = hass.states.get("switch.cloud_key")
assert cloudkey is None
async def test_not_admin(hass):
async def test_not_admin(hass, aioclient_mock):
"""Test that switch platform only work on an admin account."""
description = deepcopy(DESCRIPTION)
description[0]["site_role"] = "not admin"
controller = await setup_unifi_integration(
await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_TRACK_CLIENTS: False, CONF_TRACK_DEVICES: False},
site_description=description,
clients_response=[CLIENT_1],
devices_response=[DEVICE_1],
)
assert len(controller.mock_requests) == 6
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
async def test_switches(hass):
async def test_switches(hass, aioclient_mock):
"""Test the update_items function with some clients."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_BLOCK_CLIENT: [BLOCKED["mac"], UNBLOCKED["mac"]],
CONF_TRACK_CLIENTS: False,
@ -354,8 +344,8 @@ async def test_switches(hass):
dpigroup_response=DPI_GROUPS,
dpiapp_response=DPI_APPS,
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(controller.mock_requests) == 6
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 4
switch_1 = hass.states.get("switch.poe_client_1")
@ -381,38 +371,44 @@ async def test_switches(hass):
assert dpi_switch is not None
assert dpi_switch.state == "on"
# Block and unblock client
aioclient_mock.post(
f"https://{controller.host}:1234/api/s/{controller.site}/cmd/stamgr",
)
await hass.services.async_call(
SWITCH_DOMAIN, "turn_off", {"entity_id": "switch.block_client_1"}, blocking=True
)
assert len(controller.mock_requests) == 7
assert controller.mock_requests[6] == {
"json": {"mac": "00:00:00:00:01:01", "cmd": "block-sta"},
"method": "post",
"path": "/cmd/stamgr",
assert aioclient_mock.call_count == 11
assert aioclient_mock.mock_calls[10][2] == {
"mac": "00:00:00:00:01:01",
"cmd": "block-sta",
}
await hass.services.async_call(
SWITCH_DOMAIN, "turn_on", {"entity_id": "switch.block_client_1"}, blocking=True
)
assert len(controller.mock_requests) == 8
assert controller.mock_requests[7] == {
"json": {"mac": "00:00:00:00:01:01", "cmd": "unblock-sta"},
"method": "post",
"path": "/cmd/stamgr",
assert aioclient_mock.call_count == 12
assert aioclient_mock.mock_calls[11][2] == {
"mac": "00:00:00:00:01:01",
"cmd": "unblock-sta",
}
# Enable and disable DPI
aioclient_mock.put(
f"https://{controller.host}:1234/api/s/{controller.site}/rest/dpiapp/5f976f62e3c58f018ec7e17d",
)
await hass.services.async_call(
SWITCH_DOMAIN,
"turn_off",
{"entity_id": "switch.block_media_streaming"},
blocking=True,
)
assert len(controller.mock_requests) == 9
assert controller.mock_requests[8] == {
"json": {"enabled": False},
"method": "put",
"path": "/rest/dpiapp/5f976f62e3c58f018ec7e17d",
}
assert aioclient_mock.call_count == 13
assert aioclient_mock.mock_calls[12][2] == {"enabled": False}
await hass.services.async_call(
SWITCH_DOMAIN,
@ -420,22 +416,20 @@ async def test_switches(hass):
{"entity_id": "switch.block_media_streaming"},
blocking=True,
)
assert len(controller.mock_requests) == 10
assert controller.mock_requests[9] == {
"json": {"enabled": True},
"method": "put",
"path": "/rest/dpiapp/5f976f62e3c58f018ec7e17d",
}
assert aioclient_mock.call_count == 14
assert aioclient_mock.mock_calls[13][2] == {"enabled": True}
async def test_remove_switches(hass):
async def test_remove_switches(hass, aioclient_mock):
"""Test the update_items function with some clients."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_BLOCK_CLIENT: [UNBLOCKED["mac"]]},
clients_response=[CLIENT_1, UNBLOCKED],
devices_response=[DEVICE_1],
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
poe_switch = hass.states.get("switch.poe_client_1")
@ -460,10 +454,11 @@ async def test_remove_switches(hass):
assert block_switch is None
async def test_block_switches(hass):
async def test_block_switches(hass, aioclient_mock):
"""Test the update_items function with some clients."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_BLOCK_CLIENT: [BLOCKED["mac"], UNBLOCKED["mac"]],
CONF_TRACK_CLIENTS: False,
@ -472,6 +467,7 @@ async def test_block_switches(hass):
clients_response=[UNBLOCKED],
clients_all_response=[BLOCKED],
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
@ -507,31 +503,34 @@ async def test_block_switches(hass):
assert blocked is not None
assert blocked.state == "off"
aioclient_mock.post(
f"https://{controller.host}:1234/api/s/{controller.site}/cmd/stamgr",
)
await hass.services.async_call(
SWITCH_DOMAIN, "turn_off", {"entity_id": "switch.block_client_1"}, blocking=True
)
assert len(controller.mock_requests) == 7
assert controller.mock_requests[6] == {
"json": {"mac": "00:00:00:00:01:01", "cmd": "block-sta"},
"method": "post",
"path": "/cmd/stamgr",
assert aioclient_mock.call_count == 11
assert aioclient_mock.mock_calls[10][2] == {
"mac": "00:00:00:00:01:01",
"cmd": "block-sta",
}
await hass.services.async_call(
SWITCH_DOMAIN, "turn_on", {"entity_id": "switch.block_client_1"}, blocking=True
)
assert len(controller.mock_requests) == 8
assert controller.mock_requests[7] == {
"json": {"mac": "00:00:00:00:01:01", "cmd": "unblock-sta"},
"method": "post",
"path": "/cmd/stamgr",
assert aioclient_mock.call_count == 12
assert aioclient_mock.mock_calls[11][2] == {
"mac": "00:00:00:00:01:01",
"cmd": "unblock-sta",
}
async def test_new_client_discovered_on_block_control(hass):
async def test_new_client_discovered_on_block_control(hass, aioclient_mock):
"""Test if 2nd update has a new client."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_BLOCK_CLIENT: [BLOCKED["mac"]],
CONF_TRACK_CLIENTS: False,
@ -539,8 +538,8 @@ async def test_new_client_discovered_on_block_control(hass):
CONF_DPI_RESTRICTIONS: False,
},
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(controller.mock_requests) == 6
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
blocked = hass.states.get("switch.block_client_1")
@ -567,10 +566,11 @@ async def test_new_client_discovered_on_block_control(hass):
assert blocked is not None
async def test_option_block_clients(hass):
async def test_option_block_clients(hass, aioclient_mock):
"""Test the changes to option reflects accordingly."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_BLOCK_CLIENT: [BLOCKED["mac"]]},
clients_all_response=[BLOCKED, UNBLOCKED],
)
@ -578,7 +578,7 @@ async def test_option_block_clients(hass):
# Add a second switch
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_BLOCK_CLIENT: [BLOCKED["mac"], UNBLOCKED["mac"]]},
)
await hass.async_block_till_done()
@ -586,7 +586,7 @@ async def test_option_block_clients(hass):
# Remove the second switch again
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_BLOCK_CLIENT: [BLOCKED["mac"]]},
)
await hass.async_block_till_done()
@ -594,7 +594,7 @@ async def test_option_block_clients(hass):
# Enable one and remove another one
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_BLOCK_CLIENT: [UNBLOCKED["mac"]]},
)
await hass.async_block_till_done()
@ -602,17 +602,18 @@ async def test_option_block_clients(hass):
# Remove one
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_BLOCK_CLIENT: []},
)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
async def test_option_remove_switches(hass):
async def test_option_remove_switches(hass, aioclient_mock):
"""Test removal of DPI switch when options updated."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_TRACK_CLIENTS: False,
CONF_TRACK_DEVICES: False,
@ -626,23 +627,24 @@ async def test_option_remove_switches(hass):
# Disable DPI Switches
hass.config_entries.async_update_entry(
controller.config_entry,
config_entry,
options={CONF_DPI_RESTRICTIONS: False, CONF_POE_CLIENTS: False},
)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
async def test_new_client_discovered_on_poe_control(hass):
async def test_new_client_discovered_on_poe_control(hass, aioclient_mock):
"""Test if 2nd update has a new client."""
controller = await setup_unifi_integration(
config_entry = await setup_unifi_integration(
hass,
aioclient_mock,
options={CONF_TRACK_CLIENTS: False, CONF_TRACK_DEVICES: False},
clients_response=[CLIENT_1],
devices_response=[DEVICE_1],
)
controller = hass.data[UNIFI_DOMAIN][config_entry.entry_id]
assert len(controller.mock_requests) == 6
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
controller.api.websocket._data = {
@ -665,47 +667,41 @@ async def test_new_client_discovered_on_poe_control(hass):
switch_2 = hass.states.get("switch.poe_client_2")
assert switch_2 is not None
aioclient_mock.put(
f"https://{controller.host}:1234/api/s/{controller.site}/rest/device/mock-id",
)
await hass.services.async_call(
SWITCH_DOMAIN, "turn_off", {"entity_id": "switch.poe_client_1"}, blocking=True
)
assert len(controller.mock_requests) == 7
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
assert controller.mock_requests[6] == {
"json": {
"port_overrides": [{"port_idx": 1, "portconf_id": "1a1", "poe_mode": "off"}]
},
"method": "put",
"path": "/rest/device/mock-id",
assert aioclient_mock.call_count == 11
assert aioclient_mock.mock_calls[10][2] == {
"port_overrides": [{"port_idx": 1, "portconf_id": "1a1", "poe_mode": "off"}]
}
await hass.services.async_call(
SWITCH_DOMAIN, "turn_on", {"entity_id": "switch.poe_client_1"}, blocking=True
)
assert len(controller.mock_requests) == 8
assert controller.mock_requests[7] == {
"json": {
"port_overrides": [
{"port_idx": 1, "portconf_id": "1a1", "poe_mode": "auto"}
]
},
"method": "put",
"path": "/rest/device/mock-id",
assert aioclient_mock.call_count == 12
assert aioclient_mock.mock_calls[11][2] == {
"port_overrides": [{"port_idx": 1, "portconf_id": "1a1", "poe_mode": "auto"}]
}
async def test_ignore_multiple_poe_clients_on_same_port(hass):
async def test_ignore_multiple_poe_clients_on_same_port(hass, aioclient_mock):
"""Ignore when there are multiple POE driven clients on same port.
If there is a non-UniFi switch powered by POE,
clients will be transparently marked as having POE as well.
"""
controller = await setup_unifi_integration(
await setup_unifi_integration(
hass,
aioclient_mock,
clients_response=POE_SWITCH_CLIENTS,
devices_response=[DEVICE_1],
)
assert len(controller.mock_requests) == 6
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 3
switch_1 = hass.states.get("switch.poe_client_1")
@ -714,7 +710,7 @@ async def test_ignore_multiple_poe_clients_on_same_port(hass):
assert switch_2 is None
async def test_restoring_client(hass):
async def test_restoring_client(hass, aioclient_mock):
"""Test the update_items function with some clients."""
config_entry = config_entries.ConfigEntry(
version=1,
@ -744,8 +740,9 @@ async def test_restoring_client(hass):
config_entry=config_entry,
)
controller = await setup_unifi_integration(
await setup_unifi_integration(
hass,
aioclient_mock,
options={
CONF_BLOCK_CLIENT: ["random mac"],
CONF_TRACK_CLIENTS: False,
@ -756,7 +753,6 @@ async def test_restoring_client(hass):
clients_all_response=[CLIENT_1],
)
assert len(controller.mock_requests) == 6
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
device_1 = hass.states.get("switch.client_1")