evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Rework UniFi tests to not use runtime data (#119202)

Browse source
This commit is contained in:
Robert Svensson 2024-06-09 15:56:26 +02:00 • committed by GitHub
parent 34f20fce36
commit c9911e4dd4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 196 additions and 180 deletions
Download patch file Download diff file Expand all files Collapse all files

6
homeassistant/components/unifi/config_flow.py
View file

@ -21,6 +21,7 @@ import voluptuous as vol
from homeassistant.components import ssdp
from homeassistant.config_entries import (
ConfigEntry,
ConfigEntryState,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
@ -163,7 +164,10 @@ class UnifiFlowHandler(ConfigFlow, domain=UNIFI_DOMAIN):
config_entry = self.reauth_config_entry
abort_reason = "reauth_successful"
if config_entry:
if (
config_entry is not None
and config_entry.state is not ConfigEntryState.NOT_LOADED
):
hub = config_entry.runtime_data
if hub and hub.available:

6
tests/common.py
View file

@ -39,7 +39,7 @@ from homeassistant.components.device_automation import ( # noqa: F401
_async_get_device_automation_capabilities as async_get_device_automation_capabilities,
)
from homeassistant.config import async_process_component_config
from homeassistant.config_entries import ConfigEntry, ConfigFlow, _DataT
from homeassistant.config_entries import ConfigEntry, ConfigFlow
from homeassistant.const import (
DEVICE_DEFAULT_NAME,
EVENT_HOMEASSISTANT_CLOSE,
@ -972,11 +972,9 @@ class MockToggleEntity(entity.ToggleEntity):
return None
class MockConfigEntry(config_entries.ConfigEntry[_DataT]):
class MockConfigEntry(config_entries.ConfigEntry):
"""Helper for creating config entries that adds some defaults."""
runtime_data: _DataT
def __init__(
self,
*,

64
tests/components/unifi/conftest.py
View file

@ -7,10 +7,12 @@ from collections.abc import Callable
from datetime import timedelta
from types import MappingProxyType
from typing import Any
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
from aiounifi.models.message import MessageKey
import orjson
import pytest
from typing_extensions import Generator
from homeassistant.components.unifi import STORAGE_KEY, STORAGE_VERSION
from homeassistant.components.unifi.const import CONF_SITE_ID, DOMAIN as UNIFI_DOMAIN
@ -307,31 +309,33 @@ class WebsocketStateManager(asyncio.Event):
Prepares disconnect and reconnect flows.
"""
def __init__(self, hass: HomeAssistant, aioclient_mock: AiohttpClientMocker):
def __init__(
self, hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Store hass object and initialize asyncio.Event."""
self.hass = hass
self.aioclient_mock = aioclient_mock
super().__init__()
async def disconnect(self):
async def waiter(self, input: Callable[[bytes], None]) -> None:
"""Consume message_handler new_data callback."""
await self.wait()
async def disconnect(self) -> None:
"""Mark future as done to make 'await self.api.start_websocket' return."""
self.set()
await self.hass.async_block_till_done()
async def reconnect(self, fail=False):
async def reconnect(self, fail: bool = False) -> None:
"""Set up new future to make 'await self.api.start_websocket' block.
Mock api calls done by 'await self.api.login'.
Fail will make 'await self.api.start_websocket' return immediately.
"""
hub = self.hass.config_entries.async_get_entry(
DEFAULT_CONFIG_ENTRY_ID
).runtime_data
self.aioclient_mock.get(
f"https://{hub.config.host}:1234", status=302
) # Check UniFi OS
# Check UniFi OS
self.aioclient_mock.get(f"https://{DEFAULT_HOST}:1234", status=302)
self.aioclient_mock.post(
f"https://{hub.config.host}:1234/api/login",
f"https://{DEFAULT_HOST}:1234/api/login",
json={"data": "login successful", "meta": {"rc": "ok"}},
headers={"content-type": CONTENT_TYPE_JSON},
)
@ -343,36 +347,42 @@ class WebsocketStateManager(asyncio.Event):
await self.hass.async_block_till_done()
@pytest.fixture(autouse=True)
def websocket_mock(hass: HomeAssistant, aioclient_mock: AiohttpClientMocker):
"""Mock 'await self.api.start_websocket' in 'UniFiController.start_websocket'."""
@pytest.fixture(autouse=True, name="_mock_websocket")
def fixture_aiounifi_websocket_method() -> Generator[AsyncMock]:
"""Mock aiounifi websocket context manager."""
with patch("aiounifi.controller.Connectivity.websocket") as ws_mock:
yield ws_mock
@pytest.fixture(autouse=True, name="mock_websocket_state")
def fixture_aiounifi_websocket_state(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, _mock_websocket: AsyncMock
) -> WebsocketStateManager:
"""Provide a state manager for UniFi websocket."""
websocket_state_manager = WebsocketStateManager(hass, aioclient_mock)
with patch("aiounifi.Controller.start_websocket") as ws_mock:
ws_mock.side_effect = websocket_state_manager.wait
yield websocket_state_manager
_mock_websocket.side_effect = websocket_state_manager.waiter
return websocket_state_manager
@pytest.fixture(autouse=True)
def mock_unifi_websocket(hass):
@pytest.fixture(name="mock_websocket_message")
def fixture_aiounifi_websocket_message(_mock_websocket: AsyncMock):
"""No real websocket allowed."""
def make_websocket_call(
*,
message: MessageKey | None = None,
data: list[dict] | dict | None = None,
):
) -> None:
"""Generate a websocket call."""
hub = hass.config_entries.async_get_entry(DEFAULT_CONFIG_ENTRY_ID).runtime_data
message_handler = _mock_websocket.call_args[0][0]
if data and not message:
hub.api.messages.handler(data)
message_handler(orjson.dumps(data))
elif data and message:
if not isinstance(data, list):
data = [data]
hub.api.messages.handler(
{
"meta": {"message": message.value},
"data": data,
}
message_handler(
orjson.dumps({"meta": {"message": message.value}, "data": data})
)
else:
raise NotImplementedError

18
tests/components/unifi/test_button.py
View file

@ -123,7 +123,7 @@ async def _test_button_entity(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker,
websocket_mock,
mock_websocket_state,
config_entry: ConfigEntry,
entity_count: int,
entity_id: str,
@ -164,11 +164,11 @@ async def _test_button_entity(
# Availability signalling
# Controller disconnects
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
# Controller reconnects
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
@ -218,8 +218,8 @@ async def test_device_button_entities(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker,
config_entry_setup,
websocket_mock,
config_entry_setup: ConfigEntry,
mock_websocket_state,
entity_count: int,
entity_id: str,
unique_id: str,
@ -233,7 +233,7 @@ async def test_device_button_entities(
hass,
entity_registry,
aioclient_mock,
websocket_mock,
mock_websocket_state,
config_entry_setup,
entity_count,
entity_id,
@ -279,8 +279,8 @@ async def test_wlan_button_entities(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker,
config_entry_setup,
websocket_mock,
config_entry_setup: ConfigEntry,
mock_websocket_state,
entity_count: int,
entity_id: str,
unique_id: str,
@ -308,7 +308,7 @@ async def test_wlan_button_entities(
hass,
entity_registry,
aioclient_mock,
websocket_mock,
mock_websocket_state,
config_entry_setup,
entity_count,
entity_id,

64
tests/components/unifi/test_device_tracker.py
View file

@ -49,7 +49,7 @@ from tests.common import async_fire_time_changed
@pytest.mark.usefixtures("mock_device_registry")
async def test_tracked_wireless_clients(
hass: HomeAssistant,
mock_unifi_websocket,
mock_websocket_message,
config_entry_setup: ConfigEntry,
client_payload: list[dict[str, Any]],
) -> None:
@ -60,7 +60,7 @@ async def test_tracked_wireless_clients(
# Updated timestamp marks client as home
client = client_payload[0]
client["last_seen"] = dt_util.as_timestamp(dt_util.utcnow())
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.client").state == STATE_HOME
@ -78,7 +78,7 @@ async def test_tracked_wireless_clients(
assert hass.states.get("device_tracker.client").state == STATE_NOT_HOME
# Same timestamp doesn't explicitly mark client as away
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.client").state == STATE_HOME
@ -146,7 +146,7 @@ async def test_tracked_wireless_clients(
@pytest.mark.usefixtures("config_entry_setup")
@pytest.mark.usefixtures("mock_device_registry")
async def test_tracked_clients(
hass: HomeAssistant, mock_unifi_websocket, client_payload: list[dict[str, Any]]
hass: HomeAssistant, mock_websocket_message, client_payload: list[dict[str, Any]]
) -> None:
"""Test the update_items function with some clients."""
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 5
@ -170,7 +170,7 @@ async def test_tracked_clients(
client_1 = client_payload[0]
client_1["last_seen"] = dt_util.as_timestamp(dt_util.utcnow())
mock_unifi_websocket(message=MessageKey.CLIENT, data=client_1)
mock_websocket_message(message=MessageKey.CLIENT, data=client_1)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.client_1").state == STATE_HOME
@ -196,7 +196,7 @@ async def test_tracked_clients(
async def test_tracked_wireless_clients_event_source(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_unifi_websocket,
mock_websocket_message,
config_entry_setup: ConfigEntry,
client_payload: list[dict[str, Any]],
) -> None:
@ -226,7 +226,7 @@ async def test_tracked_wireless_clients_event_source(
),
"_id": "5ea331fa30c49e00f90ddc1a",
}
mock_unifi_websocket(message=MessageKey.EVENT, data=event)
mock_websocket_message(message=MessageKey.EVENT, data=event)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.client").state == STATE_HOME
@ -249,7 +249,7 @@ async def test_tracked_wireless_clients_event_source(
),
"_id": "5ea32ff730c49e00f90dca1a",
}
mock_unifi_websocket(message=MessageKey.EVENT, data=event)
mock_websocket_message(message=MessageKey.EVENT, data=event)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.client").state == STATE_HOME
@ -275,7 +275,7 @@ async def test_tracked_wireless_clients_event_source(
# New data
client["last_seen"] = dt_util.as_timestamp(dt_util.utcnow())
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.client").state == STATE_HOME
@ -298,7 +298,7 @@ async def test_tracked_wireless_clients_event_source(
),
"_id": "5ea32ff730c49e00f90dca1a",
}
mock_unifi_websocket(message=MessageKey.EVENT, data=event)
mock_websocket_message(message=MessageKey.EVENT, data=event)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.client").state == STATE_HOME
@ -361,7 +361,7 @@ async def test_tracked_wireless_clients_event_source(
async def test_tracked_devices(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_unifi_websocket,
mock_websocket_message,
device_payload: list[dict[str, Any]],
) -> None:
"""Test the update_items function with some devices."""
@ -375,7 +375,7 @@ async def test_tracked_devices(
device_2 = device_payload[1]
device_2["state"] = 1
device_2["next_interval"] = 50
mock_unifi_websocket(message=MessageKey.DEVICE, data=[device_1, device_2])
mock_websocket_message(message=MessageKey.DEVICE, data=[device_1, device_2])
await hass.async_block_till_done()
assert hass.states.get("device_tracker.device_1").state == STATE_HOME
@ -392,7 +392,7 @@ async def test_tracked_devices(
# Disabled device is unavailable
device_1["disabled"] = True
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.device_1").state == STATE_UNAVAILABLE
@ -422,7 +422,7 @@ async def test_tracked_devices(
@pytest.mark.usefixtures("config_entry_setup")
@pytest.mark.usefixtures("mock_device_registry")
async def test_remove_clients(
hass: HomeAssistant, mock_unifi_websocket, client_payload: list[dict[str, Any]]
hass: HomeAssistant, mock_websocket_message, client_payload: list[dict[str, Any]]
) -> None:
"""Test the remove_items function with some clients."""
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 2
@ -430,7 +430,7 @@ async def test_remove_clients(
assert hass.states.get("device_tracker.client_2")
# Remove client
mock_unifi_websocket(message=MessageKey.CLIENT_REMOVED, data=client_payload[0])
mock_websocket_message(message=MessageKey.CLIENT_REMOVED, data=client_payload[0])
await hass.async_block_till_done()
await hass.async_block_till_done()
@ -479,19 +479,19 @@ async def test_remove_clients(
)
@pytest.mark.usefixtures("config_entry_setup")
@pytest.mark.usefixtures("mock_device_registry")
async def test_hub_state_change(hass: HomeAssistant, websocket_mock) -> None:
async def test_hub_state_change(hass: HomeAssistant, mock_websocket_state) -> None:
"""Verify entities state reflect on hub connection becoming unavailable."""
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 2
assert hass.states.get("device_tracker.client").state == STATE_NOT_HOME
assert hass.states.get("device_tracker.device").state == STATE_HOME
# Controller unavailable
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get("device_tracker.client").state == STATE_UNAVAILABLE
assert hass.states.get("device_tracker.device").state == STATE_UNAVAILABLE
# Controller available
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get("device_tracker.client").state == STATE_NOT_HOME
assert hass.states.get("device_tracker.device").state == STATE_HOME
@ -707,7 +707,7 @@ async def test_option_track_devices(
@pytest.mark.usefixtures("mock_device_registry")
async def test_option_ssid_filter(
hass: HomeAssistant,
mock_unifi_websocket,
mock_websocket_message,
config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]],
) -> None:
@ -753,12 +753,12 @@ async def test_option_ssid_filter(
# Roams to SSID outside of filter
client = client_payload[0]
client["essid"] = "other_ssid"
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
# Data update while SSID filter is in effect shouldn't create the client
client_on_ssid2 = client_payload[1]
client_on_ssid2["last_seen"] = dt_util.as_timestamp(dt_util.utcnow())
mock_unifi_websocket(message=MessageKey.CLIENT, data=client_on_ssid2)
mock_websocket_message(message=MessageKey.CLIENT, data=client_on_ssid2)
await hass.async_block_till_done()
new_time = dt_util.utcnow() + timedelta(
@ -782,7 +782,7 @@ async def test_option_ssid_filter(
client["last_seen"] += 1
client_on_ssid2["last_seen"] += 1
mock_unifi_websocket(message=MessageKey.CLIENT, data=[client, client_on_ssid2])
mock_websocket_message(message=MessageKey.CLIENT, data=[client, client_on_ssid2])
await hass.async_block_till_done()
assert hass.states.get("device_tracker.client").state == STATE_HOME
@ -801,7 +801,7 @@ async def test_option_ssid_filter(
assert hass.states.get("device_tracker.client").state == STATE_NOT_HOME
client_on_ssid2["last_seen"] += 1
mock_unifi_websocket(message=MessageKey.CLIENT, data=client_on_ssid2)
mock_websocket_message(message=MessageKey.CLIENT, data=client_on_ssid2)
await hass.async_block_till_done()
# Client won't go away until after next update
@ -809,7 +809,7 @@ async def test_option_ssid_filter(
# Trigger update to get client marked as away
client_on_ssid2["last_seen"] += 1
mock_unifi_websocket(message=MessageKey.CLIENT, data=client_on_ssid2)
mock_websocket_message(message=MessageKey.CLIENT, data=client_on_ssid2)
await hass.async_block_till_done()
new_time += timedelta(
@ -825,7 +825,7 @@ async def test_option_ssid_filter(
@pytest.mark.usefixtures("mock_device_registry")
async def test_wireless_client_go_wired_issue(
hass: HomeAssistant,
mock_unifi_websocket,
mock_websocket_message,
config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]],
) -> None:
@ -855,7 +855,7 @@ async def test_wireless_client_go_wired_issue(
client = client_payload[0]
client["last_seen"] = dt_util.as_timestamp(dt_util.utcnow())
client["is_wired"] = True
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
await hass.async_block_till_done()
# Wired bug fix keeps client marked as wireless
@ -876,7 +876,7 @@ async def test_wireless_client_go_wired_issue(
# Try to mark client as connected
client["last_seen"] += 1
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
await hass.async_block_till_done()
# Make sure it don't go online again until wired bug disappears
@ -886,7 +886,7 @@ async def test_wireless_client_go_wired_issue(
# Make client wireless
client["last_seen"] += 1
client["is_wired"] = False
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
await hass.async_block_till_done()
# Client is no longer affected by wired bug and can be marked online
@ -898,7 +898,7 @@ async def test_wireless_client_go_wired_issue(
@pytest.mark.usefixtures("mock_device_registry")
async def test_option_ignore_wired_bug(
hass: HomeAssistant,
mock_unifi_websocket,
mock_websocket_message,
config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]],
) -> None:
@ -925,7 +925,7 @@ async def test_option_ignore_wired_bug(
# Trigger wired bug
client = client_payload[0]
client["is_wired"] = True
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
await hass.async_block_till_done()
# Wired bug in effect
@ -946,7 +946,7 @@ async def test_option_ignore_wired_bug(
# Mark client as connected again
client["last_seen"] += 1
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
await hass.async_block_till_done()
# Ignoring wired bug allows client to go home again even while affected
@ -956,7 +956,7 @@ async def test_option_ignore_wired_bug(
# Make client wireless
client["last_seen"] += 1
client["is_wired"] = False
mock_unifi_websocket(message=MessageKey.CLIENT, data=client)
mock_websocket_message(message=MessageKey.CLIENT, data=client)
await hass.async_block_till_done()
# Client is wireless and still connected

22
tests/components/unifi/test_hub.py
View file

@ -80,7 +80,7 @@ async def test_reset_fails(
async def test_connection_state_signalling(
hass: HomeAssistant,
mock_device_registry,
websocket_mock,
mock_websocket_state,
config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]],
) -> None:
@ -99,17 +99,19 @@ async def test_connection_state_signalling(
# Controller is connected
assert hass.states.get("device_tracker.client").state == "home"
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
# Controller is disconnected
assert hass.states.get("device_tracker.client").state == "unavailable"
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
# Controller is once again connected
assert hass.states.get("device_tracker.client").state == "home"
async def test_reconnect_mechanism(
aioclient_mock: AiohttpClientMocker, websocket_mock, config_entry_setup: ConfigEntry
aioclient_mock: AiohttpClientMocker,
mock_websocket_state,
config_entry_setup: ConfigEntry,
) -> None:
"""Verify reconnect prints only on first reconnection try."""
aioclient_mock.clear_requests()
@ -118,13 +120,13 @@ async def test_reconnect_mechanism(
status=HTTPStatus.BAD_GATEWAY,
)
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert aioclient_mock.call_count == 0
await websocket_mock.reconnect(fail=True)
await mock_websocket_state.reconnect(fail=True)
assert aioclient_mock.call_count == 1
await websocket_mock.reconnect(fail=True)
await mock_websocket_state.reconnect(fail=True)
assert aioclient_mock.call_count == 2
@ -138,7 +140,7 @@ async def test_reconnect_mechanism(
],
)
@pytest.mark.usefixtures("config_entry_setup")
async def test_reconnect_mechanism_exceptions(websocket_mock, exception) -> None:
async def test_reconnect_mechanism_exceptions(mock_websocket_state, exception) -> None:
"""Verify async_reconnect calls expected methods."""
with (
patch("aiounifi.Controller.login", side_effect=exception),
@ -146,9 +148,9 @@ async def test_reconnect_mechanism_exceptions(websocket_mock, exception) -> None
"homeassistant.components.unifi.hub.hub.UnifiWebsocket.reconnect"
) as mock_reconnect,
):
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
mock_reconnect.assert_called_once()

16
tests/components/unifi/test_image.py
View file

@ -63,8 +63,8 @@ async def test_wlan_qr_code(
entity_registry: er.EntityRegistry,
hass_client: ClientSessionGenerator,
snapshot: SnapshotAssertion,
mock_unifi_websocket,
websocket_mock,
mock_websocket_message,
mock_websocket_state,
) -> None:
"""Test the update_clients function when no clients are found."""
assert len(hass.states.async_entity_ids(IMAGE_DOMAIN)) == 0
@ -96,7 +96,7 @@ async def test_wlan_qr_code(
assert body == snapshot
# Update state object - same password - no change to state
mock_unifi_websocket(message=MessageKey.WLAN_CONF_UPDATED, data=WLAN)
mock_websocket_message(message=MessageKey.WLAN_CONF_UPDATED, data=WLAN)
await hass.async_block_till_done()
image_state_2 = hass.states.get("image.ssid_1_qr_code")
assert image_state_1.state == image_state_2.state
@ -104,7 +104,7 @@ async def test_wlan_qr_code(
# Update state object - changed password - new state
data = deepcopy(WLAN)
data["x_passphrase"] = "new password"
mock_unifi_websocket(message=MessageKey.WLAN_CONF_UPDATED, data=data)
mock_websocket_message(message=MessageKey.WLAN_CONF_UPDATED, data=data)
await hass.async_block_till_done()
image_state_3 = hass.states.get("image.ssid_1_qr_code")
assert image_state_1.state != image_state_3.state
@ -119,22 +119,22 @@ async def test_wlan_qr_code(
# Availability signalling
# Controller disconnects
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get("image.ssid_1_qr_code").state == STATE_UNAVAILABLE
# Controller reconnects
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get("image.ssid_1_qr_code").state != STATE_UNAVAILABLE
# WLAN gets disabled
wlan_1 = deepcopy(WLAN)
wlan_1["enabled"] = False
mock_unifi_websocket(message=MessageKey.WLAN_CONF_UPDATED, data=wlan_1)
mock_websocket_message(message=MessageKey.WLAN_CONF_UPDATED, data=wlan_1)
await hass.async_block_till_done()
assert hass.states.get("image.ssid_1_qr_code").state == STATE_UNAVAILABLE
# WLAN gets re-enabled
wlan_1["enabled"] = True
mock_unifi_websocket(message=MessageKey.WLAN_CONF_UPDATED, data=wlan_1)
mock_websocket_message(message=MessageKey.WLAN_CONF_UPDATED, data=wlan_1)
await hass.async_block_till_done()
assert hass.states.get("image.ssid_1_qr_code").state != STATE_UNAVAILABLE

4
tests/components/unifi/test_init.py
View file

@ -176,7 +176,7 @@ async def test_remove_config_entry_device(
config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]],
device_payload: list[dict[str, Any]],
mock_unifi_websocket,
mock_websocket_message,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Verify removing a device manually."""
@ -206,7 +206,7 @@ async def test_remove_config_entry_device(
)
# Remove a client from Unifi API
mock_unifi_websocket(message=MessageKey.CLIENT_REMOVED, data=[client_payload[1]])
mock_websocket_message(message=MessageKey.CLIENT_REMOVED, data=[client_payload[1]])
await hass.async_block_till_done()
# Try to remove an inactive client from UI: allowed

78
tests/components/unifi/test_sensor.py
View file

@ -355,7 +355,7 @@ async def test_no_clients(hass: HomeAssistant) -> None:
)
async def test_bandwidth_sensors(
hass: HomeAssistant,
mock_unifi_websocket,
mock_websocket_message,
config_entry_options: MappingProxyType[str, Any],
config_entry_setup: ConfigEntry,
client_payload: list[dict[str, Any]],
@ -391,7 +391,7 @@ async def test_bandwidth_sensors(
wireless_client["rx_bytes-r"] = 3456000000
wireless_client["tx_bytes-r"] = 7891000000
mock_unifi_websocket(message=MessageKey.CLIENT, data=wireless_client)
mock_websocket_message(message=MessageKey.CLIENT, data=wireless_client)
await hass.async_block_till_done()
assert hass.states.get("sensor.wireless_client_rx").state == "3456.0"
@ -402,7 +402,7 @@ async def test_bandwidth_sensors(
new_time = dt_util.utcnow()
wireless_client["last_seen"] = dt_util.as_timestamp(new_time)
mock_unifi_websocket(message=MessageKey.CLIENT, data=wireless_client)
mock_websocket_message(message=MessageKey.CLIENT, data=wireless_client)
await hass.async_block_till_done()
with freeze_time(new_time):
@ -490,7 +490,7 @@ async def test_uptime_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
mock_unifi_websocket,
mock_websocket_message,
config_entry_options: MappingProxyType[str, Any],
config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]],
@ -516,7 +516,7 @@ async def test_uptime_sensors(
uptime_client["uptime"] = event_uptime
now = datetime(2021, 1, 1, 1, 1, 4, tzinfo=dt_util.UTC)
with patch("homeassistant.util.dt.now", return_value=now):
mock_unifi_websocket(message=MessageKey.CLIENT, data=uptime_client)
mock_websocket_message(message=MessageKey.CLIENT, data=uptime_client)
await hass.async_block_till_done()
assert hass.states.get("sensor.client1_uptime").state == "2021-01-01T01:00:00+00:00"
@ -526,7 +526,7 @@ async def test_uptime_sensors(
uptime_client["uptime"] = new_uptime
now = datetime(2021, 2, 1, 1, 1, 0, tzinfo=dt_util.UTC)
with patch("homeassistant.util.dt.now", return_value=now):
mock_unifi_websocket(message=MessageKey.CLIENT, data=uptime_client)
mock_websocket_message(message=MessageKey.CLIENT, data=uptime_client)
await hass.async_block_till_done()
assert hass.states.get("sensor.client1_uptime").state == "2021-02-01T01:00:00+00:00"
@ -583,7 +583,7 @@ async def test_uptime_sensors(
@pytest.mark.usefixtures("config_entry_setup")
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_remove_sensors(
hass: HomeAssistant, mock_unifi_websocket, client_payload: list[dict[str, Any]]
hass: HomeAssistant, mock_websocket_message, client_payload: list[dict[str, Any]]
) -> None:
"""Verify removing of clients work as expected."""
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 6
@ -595,7 +595,7 @@ async def test_remove_sensors(
assert hass.states.get("sensor.wireless_client_uptime")
# Remove wired client
mock_unifi_websocket(message=MessageKey.CLIENT_REMOVED, data=client_payload[0])
mock_websocket_message(message=MessageKey.CLIENT_REMOVED, data=client_payload[0])
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 3
@ -612,8 +612,8 @@ async def test_remove_sensors(
async def test_poe_port_switches(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_unifi_websocket,
websocket_mock,
mock_websocket_message,
mock_websocket_state,
) -> None:
"""Test the update_items function with some clients."""
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 2
@ -642,34 +642,34 @@ async def test_poe_port_switches(
# Update state object
device_1 = deepcopy(DEVICE_1)
device_1["port_table"][0]["poe_power"] = "5.12"
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get("sensor.mock_name_port_1_poe_power").state == "5.12"
# PoE is disabled
device_1 = deepcopy(DEVICE_1)
device_1["port_table"][0]["poe_mode"] = "off"
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get("sensor.mock_name_port_1_poe_power").state == "0"
# Availability signalling
# Controller disconnects
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert (
hass.states.get("sensor.mock_name_port_1_poe_power").state == STATE_UNAVAILABLE
)
# Controller reconnects
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert (
hass.states.get("sensor.mock_name_port_1_poe_power").state != STATE_UNAVAILABLE
)
# Device gets disabled
device_1["disabled"] = True
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert (
hass.states.get("sensor.mock_name_port_1_poe_power").state == STATE_UNAVAILABLE
@ -677,7 +677,7 @@ async def test_poe_port_switches(
# Device gets re-enabled
device_1["disabled"] = False
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get("sensor.mock_name_port_1_poe_power")
@ -686,8 +686,8 @@ async def test_poe_port_switches(
async def test_wlan_client_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_unifi_websocket,
websocket_mock,
mock_websocket_message,
mock_websocket_state,
config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]],
) -> None:
@ -730,10 +730,10 @@ async def test_wlan_client_sensors(
# Verify state update - increasing number
wireless_client_1 = client_payload[0]
wireless_client_1["essid"] = "SSID 1"
mock_unifi_websocket(message=MessageKey.CLIENT, data=wireless_client_1)
mock_websocket_message(message=MessageKey.CLIENT, data=wireless_client_1)
wireless_client_2 = client_payload[1]
wireless_client_2["essid"] = "SSID 1"
mock_unifi_websocket(message=MessageKey.CLIENT, data=wireless_client_2)
mock_websocket_message(message=MessageKey.CLIENT, data=wireless_client_2)
await hass.async_block_till_done()
ssid_1 = hass.states.get("sensor.ssid_1")
@ -748,7 +748,7 @@ async def test_wlan_client_sensors(
# Verify state update - decreasing number
wireless_client_1["essid"] = "SSID"
mock_unifi_websocket(message=MessageKey.CLIENT, data=wireless_client_1)
mock_websocket_message(message=MessageKey.CLIENT, data=wireless_client_1)
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done()
@ -759,7 +759,7 @@ async def test_wlan_client_sensors(
# Verify state update - decreasing number
wireless_client_2["last_seen"] = 0
mock_unifi_websocket(message=MessageKey.CLIENT, data=wireless_client_2)
mock_websocket_message(message=MessageKey.CLIENT, data=wireless_client_2)
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done()
@ -770,23 +770,23 @@ async def test_wlan_client_sensors(
# Availability signalling
# Controller disconnects
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get("sensor.ssid_1").state == STATE_UNAVAILABLE
# Controller reconnects
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get("sensor.ssid_1").state == "0"
# WLAN gets disabled
wlan_1 = deepcopy(WLAN)
wlan_1["enabled"] = False
mock_unifi_websocket(message=MessageKey.WLAN_CONF_UPDATED, data=wlan_1)
mock_websocket_message(message=MessageKey.WLAN_CONF_UPDATED, data=wlan_1)
await hass.async_block_till_done()
assert hass.states.get("sensor.ssid_1").state == STATE_UNAVAILABLE
# WLAN gets re-enabled
wlan_1["enabled"] = True
mock_unifi_websocket(message=MessageKey.WLAN_CONF_UPDATED, data=wlan_1)
mock_websocket_message(message=MessageKey.WLAN_CONF_UPDATED, data=wlan_1)
await hass.async_block_till_done()
assert hass.states.get("sensor.ssid_1").state == "0"
@ -828,7 +828,7 @@ async def test_wlan_client_sensors(
async def test_outlet_power_readings(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_unifi_websocket,
mock_websocket_message,
device_payload: list[dict[str, Any]],
entity_id: str,
expected_unique_id: str,
@ -852,7 +852,7 @@ async def test_outlet_power_readings(
updated_device_data = deepcopy(device_payload[0])
updated_device_data.update(changed_data)
mock_unifi_websocket(message=MessageKey.DEVICE, data=updated_device_data)
mock_websocket_message(message=MessageKey.DEVICE, data=updated_device_data)
await hass.async_block_till_done()
sensor_data = hass.states.get(f"sensor.{entity_id}")
@ -887,7 +887,7 @@ async def test_outlet_power_readings(
async def test_device_uptime(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_unifi_websocket,
mock_websocket_message,
config_entry_factory: Callable[[], ConfigEntry],
device_payload: list[dict[str, Any]],
) -> None:
@ -909,7 +909,7 @@ async def test_device_uptime(
device["uptime"] = 64
now = datetime(2021, 1, 1, 1, 1, 4, tzinfo=dt_util.UTC)
with patch("homeassistant.util.dt.now", return_value=now):
mock_unifi_websocket(message=MessageKey.DEVICE, data=device)
mock_websocket_message(message=MessageKey.DEVICE, data=device)
assert hass.states.get("sensor.device_uptime").state == "2021-01-01T01:00:00+00:00"
@ -919,7 +919,7 @@ async def test_device_uptime(
device["uptime"] = 60
now = datetime(2021, 2, 1, 1, 1, 0, tzinfo=dt_util.UTC)
with patch("homeassistant.util.dt.now", return_value=now):
mock_unifi_websocket(message=MessageKey.DEVICE, data=device)
mock_websocket_message(message=MessageKey.DEVICE, data=device)
assert hass.states.get("sensor.device_uptime").state == "2021-02-01T01:00:00+00:00"
@ -955,7 +955,7 @@ async def test_device_uptime(
async def test_device_temperature(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_unifi_websocket,
mock_websocket_message,
device_payload: list[dict[str, Any]],
) -> None:
"""Verify that temperature sensors are working as expected."""
@ -969,7 +969,7 @@ async def test_device_temperature(
# Verify new event change temperature
device = device_payload[0]
device["general_temperature"] = 60
mock_unifi_websocket(message=MessageKey.DEVICE, data=device)
mock_websocket_message(message=MessageKey.DEVICE, data=device)
assert hass.states.get("sensor.device_temperature").state == "60"
@ -1004,7 +1004,7 @@ async def test_device_temperature(
async def test_device_state(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_unifi_websocket,
mock_websocket_message,
device_payload: list[dict[str, Any]],
) -> None:
"""Verify that state sensors are working as expected."""
@ -1017,7 +1017,7 @@ async def test_device_state(
device = device_payload[0]
for i in list(map(int, DeviceState)):
device["state"] = i
mock_unifi_websocket(message=MessageKey.DEVICE, data=device)
mock_websocket_message(message=MessageKey.DEVICE, data=device)
assert hass.states.get("sensor.device_state").state == DEVICE_STATES[i]
@ -1041,7 +1041,7 @@ async def test_device_state(
async def test_device_system_stats(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_unifi_websocket,
mock_websocket_message,
device_payload: list[dict[str, Any]],
) -> None:
"""Verify that device stats sensors are working as expected."""
@ -1065,7 +1065,7 @@ async def test_device_system_stats(
# Verify new event change system-stats
device = device_payload[0]
device["system-stats"] = {"cpu": 7.7, "mem": 33.3, "uptime": 7316}
mock_unifi_websocket(message=MessageKey.DEVICE, data=device)
mock_websocket_message(message=MessageKey.DEVICE, data=device)
assert hass.states.get("sensor.device_cpu_utilization").state == "7.7"
assert hass.states.get("sensor.device_memory_utilization").state == "33.3"
@ -1138,7 +1138,7 @@ async def test_device_system_stats(
async def test_bandwidth_port_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_unifi_websocket,
mock_websocket_message,
config_entry_setup: ConfigEntry,
config_entry_options: MappingProxyType[str, Any],
device_payload,
@ -1206,7 +1206,7 @@ async def test_bandwidth_port_sensors(
device_1["port_table"][0]["rx_bytes-r"] = 3456000000
device_1["port_table"][0]["tx_bytes-r"] = 7891000000
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get("sensor.mock_name_port_1_rx").state == "27648.00000"

86
tests/components/unifi/test_switch.py
View file

@ -886,14 +886,14 @@ async def test_switches(
@pytest.mark.parametrize("dpi_app_payload", [DPI_APPS])
@pytest.mark.parametrize("dpi_group_payload", [DPI_GROUPS])
@pytest.mark.usefixtures("config_entry_setup")
async def test_remove_switches(hass: HomeAssistant, mock_unifi_websocket) -> None:
async def test_remove_switches(hass: HomeAssistant, mock_websocket_message) -> None:
"""Test the update_items function with some clients."""
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
assert hass.states.get("switch.block_client_2") is not None
assert hass.states.get("switch.block_media_streaming") is not None
mock_unifi_websocket(message=MessageKey.CLIENT_REMOVED, data=[UNBLOCKED])
mock_websocket_message(message=MessageKey.CLIENT_REMOVED, data=[UNBLOCKED])
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
@ -901,7 +901,7 @@ async def test_remove_switches(hass: HomeAssistant, mock_unifi_websocket) -> Non
assert hass.states.get("switch.block_client_2") is None
assert hass.states.get("switch.block_media_streaming") is not None
mock_unifi_websocket(data=DPI_GROUP_REMOVED_EVENT)
mock_websocket_message(data=DPI_GROUP_REMOVED_EVENT)
await hass.async_block_till_done()
assert hass.states.get("switch.block_media_streaming") is None
@ -923,7 +923,7 @@ async def test_remove_switches(hass: HomeAssistant, mock_unifi_websocket) -> Non
async def test_block_switches(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket,
mock_websocket_message,
config_entry_setup,
) -> None:
"""Test the update_items function with some clients."""
@ -939,7 +939,9 @@ async def test_block_switches(
assert unblocked is not None
assert unblocked.state == "on"
mock_unifi_websocket(message=MessageKey.EVENT, data=EVENT_BLOCKED_CLIENT_UNBLOCKED)
mock_websocket_message(
message=MessageKey.EVENT, data=EVENT_BLOCKED_CLIENT_UNBLOCKED
)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
@ -947,7 +949,7 @@ async def test_block_switches(
assert blocked is not None
assert blocked.state == "on"
mock_unifi_websocket(message=MessageKey.EVENT, data=EVENT_BLOCKED_CLIENT_BLOCKED)
mock_websocket_message(message=MessageKey.EVENT, data=EVENT_BLOCKED_CLIENT_BLOCKED)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 2
@ -984,7 +986,7 @@ async def test_block_switches(
@pytest.mark.parametrize("dpi_group_payload", [DPI_GROUPS])
@pytest.mark.usefixtures("config_entry_setup")
async def test_dpi_switches(
hass: HomeAssistant, mock_unifi_websocket, websocket_mock
hass: HomeAssistant, mock_websocket_message, mock_websocket_state
) -> None:
"""Test the update_items function with some clients."""
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
@ -994,7 +996,7 @@ async def test_dpi_switches(
assert dpi_switch.state == STATE_ON
assert dpi_switch.attributes["icon"] == "mdi:network"
mock_unifi_websocket(data=DPI_APP_DISABLED_EVENT)
mock_websocket_message(data=DPI_APP_DISABLED_EVENT)
await hass.async_block_till_done()
assert hass.states.get("switch.block_media_streaming").state == STATE_OFF
@ -1002,15 +1004,15 @@ async def test_dpi_switches(
# Availability signalling
# Controller disconnects
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get("switch.block_media_streaming").state == STATE_UNAVAILABLE
# Controller reconnects
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get("switch.block_media_streaming").state == STATE_OFF
# Remove app
mock_unifi_websocket(data=DPI_GROUP_REMOVE_APP)
mock_websocket_message(data=DPI_GROUP_REMOVE_APP)
await hass.async_block_till_done()
assert hass.states.get("switch.block_media_streaming") is None
@ -1021,7 +1023,7 @@ async def test_dpi_switches(
@pytest.mark.parametrize("dpi_group_payload", [DPI_GROUPS])
@pytest.mark.usefixtures("config_entry_setup")
async def test_dpi_switches_add_second_app(
hass: HomeAssistant, mock_unifi_websocket
hass: HomeAssistant, mock_websocket_message
) -> None:
"""Test the update_items function with some clients."""
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
@ -1036,7 +1038,7 @@ async def test_dpi_switches_add_second_app(
"site_id": "name",
"_id": "61783e89c1773a18c0c61f00",
}
mock_unifi_websocket(message=MessageKey.DPI_APP_ADDED, data=second_app_event)
mock_websocket_message(message=MessageKey.DPI_APP_ADDED, data=second_app_event)
await hass.async_block_till_done()
assert hass.states.get("switch.block_media_streaming").state == STATE_ON
@ -1047,7 +1049,7 @@ async def test_dpi_switches_add_second_app(
"site_id": "name",
"dpiapp_ids": ["5f976f62e3c58f018ec7e17d", "61783e89c1773a18c0c61f00"],
}
mock_unifi_websocket(
mock_websocket_message(
message=MessageKey.DPI_GROUP_UPDATED, data=add_second_app_to_group
)
await hass.async_block_till_done()
@ -1063,7 +1065,7 @@ async def test_dpi_switches_add_second_app(
"site_id": "name",
"_id": "61783e89c1773a18c0c61f00",
}
mock_unifi_websocket(
mock_websocket_message(
message=MessageKey.DPI_APP_UPDATED, data=second_app_event_enabled
)
await hass.async_block_till_done()
@ -1082,10 +1084,10 @@ async def test_dpi_switches_add_second_app(
async def test_outlet_switches(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket,
mock_websocket_message,
config_entry_setup,
device_payload,
websocket_mock,
mock_websocket_state,
entity_id: str,
outlet_index: int,
expected_switches: int,
@ -1104,7 +1106,7 @@ async def test_outlet_switches(
# Update state object
device_1 = deepcopy(device_payload[0])
device_1["outlet_table"][outlet_index - 1]["relay_state"] = False
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get(f"switch.{entity_id}").state == STATE_OFF
@ -1149,22 +1151,22 @@ async def test_outlet_switches(
# Availability signalling
# Controller disconnects
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get(f"switch.{entity_id}").state == STATE_UNAVAILABLE
# Controller reconnects
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get(f"switch.{entity_id}").state == STATE_OFF
# Device gets disabled
device_1["disabled"] = True
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get(f"switch.{entity_id}").state == STATE_UNAVAILABLE
# Device gets re-enabled
device_1["disabled"] = False
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get(f"switch.{entity_id}").state == STATE_OFF
@ -1191,13 +1193,13 @@ async def test_outlet_switches(
)
@pytest.mark.usefixtures("config_entry_setup")
async def test_new_client_discovered_on_block_control(
hass: HomeAssistant, mock_unifi_websocket
hass: HomeAssistant, mock_websocket_message
) -> None:
"""Test if 2nd update has a new client."""
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 0
assert hass.states.get("switch.block_client_1") is None
mock_unifi_websocket(message=MessageKey.CLIENT, data=BLOCKED)
mock_websocket_message(message=MessageKey.CLIENT, data=BLOCKED)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(SWITCH_DOMAIN)) == 1
@ -1281,8 +1283,8 @@ async def test_poe_port_switches(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket,
websocket_mock,
mock_websocket_message,
mock_websocket_state,
config_entry_setup,
device_payload,
) -> None:
@ -1319,7 +1321,7 @@ async def test_poe_port_switches(
# Update state object
device_1 = deepcopy(device_payload[0])
device_1["port_table"][0]["poe_mode"] = "off"
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get("switch.mock_name_port_1_poe").state == STATE_OFF
@ -1369,22 +1371,22 @@ async def test_poe_port_switches(
# Availability signalling
# Controller disconnects
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get("switch.mock_name_port_1_poe").state == STATE_UNAVAILABLE
# Controller reconnects
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get("switch.mock_name_port_1_poe").state == STATE_OFF
# Device gets disabled
device_1["disabled"] = True
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get("switch.mock_name_port_1_poe").state == STATE_UNAVAILABLE
# Device gets re-enabled
device_1["disabled"] = False
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
assert hass.states.get("switch.mock_name_port_1_poe").state == STATE_OFF
@ -1394,8 +1396,8 @@ async def test_wlan_switches(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket,
websocket_mock,
mock_websocket_message,
mock_websocket_state,
config_entry_setup,
wlan_payload,
) -> None:
@ -1417,7 +1419,7 @@ async def test_wlan_switches(
# Update state object
wlan = deepcopy(wlan_payload[0])
wlan["enabled"] = False
mock_unifi_websocket(message=MessageKey.WLAN_CONF_UPDATED, data=wlan)
mock_websocket_message(message=MessageKey.WLAN_CONF_UPDATED, data=wlan)
await hass.async_block_till_done()
assert hass.states.get("switch.ssid_1").state == STATE_OFF
@ -1450,11 +1452,11 @@ async def test_wlan_switches(
# Availability signalling
# Controller disconnects
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get("switch.ssid_1").state == STATE_UNAVAILABLE
# Controller reconnects
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get("switch.ssid_1").state == STATE_OFF
@ -1481,8 +1483,8 @@ async def test_port_forwarding_switches(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
aioclient_mock: AiohttpClientMocker,
mock_unifi_websocket,
websocket_mock,
mock_websocket_message,
mock_websocket_state,
config_entry_setup,
port_forward_payload,
) -> None:
@ -1503,7 +1505,7 @@ async def test_port_forwarding_switches(
# Update state object
data = port_forward_payload[0].copy()
data["enabled"] = False
mock_unifi_websocket(message=MessageKey.PORT_FORWARD_UPDATED, data=data)
mock_websocket_message(message=MessageKey.PORT_FORWARD_UPDATED, data=data)
await hass.async_block_till_done()
assert hass.states.get("switch.unifi_network_plex").state == STATE_OFF
@ -1538,15 +1540,15 @@ async def test_port_forwarding_switches(
# Availability signalling
# Controller disconnects
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get("switch.unifi_network_plex").state == STATE_UNAVAILABLE
# Controller reconnects
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get("switch.unifi_network_plex").state == STATE_OFF
# Remove entity on deleted message
mock_unifi_websocket(
mock_websocket_message(
message=MessageKey.PORT_FORWARD_DELETED, data=port_forward_payload[0]
)
await hass.async_block_till_done()

12
tests/components/unifi/test_update.py
View file

@ -61,7 +61,7 @@ DEVICE_2 = {
@pytest.mark.parametrize("device_payload", [[DEVICE_1, DEVICE_2]])
@pytest.mark.usefixtures("config_entry_setup")
async def test_device_updates(hass: HomeAssistant, mock_unifi_websocket) -> None:
async def test_device_updates(hass: HomeAssistant, mock_websocket_message) -> None:
"""Test the update_items function with some devices."""
assert len(hass.states.async_entity_ids(UPDATE_DOMAIN)) == 2
@ -95,7 +95,7 @@ async def test_device_updates(hass: HomeAssistant, mock_unifi_websocket) -> None
device_1 = deepcopy(DEVICE_1)
device_1["state"] = 4
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
device_1_state = hass.states.get("update.device_1")
@ -110,7 +110,7 @@ async def test_device_updates(hass: HomeAssistant, mock_unifi_websocket) -> None
device_1["version"] = "4.3.17.11279"
device_1["upgradable"] = False
del device_1["upgrade_to_firmware"]
mock_unifi_websocket(message=MessageKey.DEVICE, data=device_1)
mock_websocket_message(message=MessageKey.DEVICE, data=device_1)
await hass.async_block_till_done()
device_1_state = hass.states.get("update.device_1")
@ -173,15 +173,15 @@ async def test_install(
@pytest.mark.parametrize("device_payload", [[DEVICE_1]])
@pytest.mark.usefixtures("config_entry_setup")
async def test_hub_state_change(hass: HomeAssistant, websocket_mock) -> None:
async def test_hub_state_change(hass: HomeAssistant, mock_websocket_state) -> None:
"""Verify entities state reflect on hub becoming unavailable."""
assert len(hass.states.async_entity_ids(UPDATE_DOMAIN)) == 1
assert hass.states.get("update.device_1").state == STATE_ON
# Controller unavailable
await websocket_mock.disconnect()
await mock_websocket_state.disconnect()
assert hass.states.get("update.device_1").state == STATE_UNAVAILABLE
# Controller available
await websocket_mock.reconnect()
await mock_websocket_state.reconnect()
assert hass.states.get("update.device_1").state == STATE_ON