Add test coverage for tplink_omada update entities (#162549)

This commit is contained in:
MarkGodwin
2026-02-16 21:17:56 +00:00
committed by GitHub
parent 9ec456d28e
commit 73fa9925c4
8 changed files with 391 additions and 32 deletions

View File

@@ -39,7 +39,7 @@ rules:
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: done
test-coverage: todo
test-coverage: done
# Gold
devices: done

View File

@@ -2,7 +2,6 @@
from collections.abc import AsyncGenerator, Generator
from functools import partial
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -14,6 +13,7 @@ from tplink_omada_client.clients import (
OmadaWirelessClient,
)
from tplink_omada_client.devices import (
OmadaFirmwareUpdate,
OmadaGateway,
OmadaListDevice,
OmadaSwitch,
@@ -25,7 +25,11 @@ from homeassistant.components.tplink_omada.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_VERIFY_SSL
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, async_load_fixture
from tests.common import (
MockConfigEntry,
async_load_json_array_fixture,
async_load_json_object_fixture,
)
@pytest.fixture
@@ -59,29 +63,44 @@ async def mock_omada_site_client(hass: HomeAssistant) -> AsyncGenerator[AsyncMoc
"""Mock Omada site client."""
site_client = MagicMock()
gateway_data = json.loads(
await async_load_fixture(hass, "gateway-TL-ER7212PC.json", DOMAIN)
gateway_data = await async_load_json_object_fixture(
hass, "gateway-TL-ER7212PC.json", DOMAIN
)
gateway = OmadaGateway(gateway_data)
site_client.get_gateway = AsyncMock(return_value=gateway)
switch1_data = json.loads(
await async_load_fixture(hass, "switch-TL-SG3210XHP-M2.json", DOMAIN)
switch1_data = await async_load_json_object_fixture(
hass, "switch-TL-SG3210XHP-M2.json", DOMAIN
)
switch1 = OmadaSwitch(switch1_data)
site_client.get_switches = AsyncMock(return_value=[switch1])
site_client.get_switch = AsyncMock(return_value=switch1)
devices_data = json.loads(await async_load_fixture(hass, "devices.json", DOMAIN))
devices_data = await async_load_json_array_fixture(hass, "devices.json", DOMAIN)
devices = [OmadaListDevice(d) for d in devices_data]
site_client.get_devices = AsyncMock(return_value=devices)
switch1_ports_data = json.loads(
await async_load_fixture(hass, "switch-ports-TL-SG3210XHP-M2.json", DOMAIN)
switch1_ports_data = await async_load_json_array_fixture(
hass, "switch-ports-TL-SG3210XHP-M2.json", DOMAIN
)
switch1_ports = [OmadaSwitchPortDetails(p) for p in switch1_ports_data]
site_client.get_switch_ports = AsyncMock(return_value=switch1_ports)
# Mock firmware update API
async def get_firmware_details(
device: OmadaListDevice,
) -> OmadaFirmwareUpdate | None:
"""Mock getting firmware details for a device."""
if device.need_upgrade:
firmware_data = await async_load_json_object_fixture(
hass, f"firmware-update-{device.mac}.json", DOMAIN
)
return OmadaFirmwareUpdate(firmware_data)
return None
site_client.get_firmware_details = AsyncMock(side_effect=get_firmware_details)
site_client.start_firmware_upgrade = AsyncMock()
async def async_empty() -> AsyncGenerator:
for c in ():
yield c
@@ -114,8 +133,8 @@ async def _get_mock_known_clients(
hass: HomeAssistant,
) -> AsyncGenerator[OmadaNetworkClient]:
"""Mock known clients of the Omada network."""
known_clients_data = json.loads(
await async_load_fixture(hass, "known-clients.json", DOMAIN)
known_clients_data = await async_load_json_array_fixture(
hass, "known-clients.json", DOMAIN
)
for c in known_clients_data:
if c["wireless"]:
@@ -128,8 +147,8 @@ async def _get_mock_connected_clients(
hass: HomeAssistant,
) -> AsyncGenerator[OmadaConnectedClient]:
"""Mock connected clients of the Omada network."""
connected_clients_data = json.loads(
await async_load_fixture(hass, "connected-clients.json", DOMAIN)
connected_clients_data = await async_load_json_array_fixture(
hass, "connected-clients.json", DOMAIN
)
for c in connected_clients_data:
if c["wireless"]:
@@ -140,8 +159,8 @@ async def _get_mock_connected_clients(
async def _get_mock_client(hass: HomeAssistant, mac: str) -> OmadaNetworkClient:
"""Mock an Omada client."""
connected_clients_data = json.loads(
await async_load_fixture(hass, "connected-clients.json", DOMAIN)
connected_clients_data = await async_load_json_array_fixture(
hass, "connected-clients.json", DOMAIN
)
for c in connected_clients_data:

View File

@@ -35,7 +35,7 @@
"memUtil": 20,
"status": 14,
"statusCategory": 1,
"needUpgrade": false,
"needUpgrade": true,
"fwDownload": false
}
]

View File

@@ -0,0 +1,5 @@
{
"curFwVer": "1.0.12 Build 20230203 Rel.36545",
"lastFwVer": "1.0.15 Build 20231101 Rel.40123",
"fwReleaseLog": "Bug fixes and performance improvements"
}

View File

@@ -0,0 +1,125 @@
# serializer version: 1
# name: test_entities[update.test_poe_switch_firmware-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'update',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'update.test_poe_switch_firmware',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Firmware',
'options': dict({
}),
'original_device_class': <UpdateDeviceClass.FIRMWARE: 'firmware'>,
'original_icon': None,
'original_name': 'Firmware',
'platform': 'tplink_omada',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <UpdateEntityFeature: 21>,
'translation_key': None,
'unique_id': '54-AF-97-00-00-01_firmware',
'unit_of_measurement': None,
})
# ---
# name: test_entities[update.test_poe_switch_firmware-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'auto_update': False,
'device_class': 'firmware',
'display_precision': 0,
'entity_picture': 'https://brands.home-assistant.io/_/tplink_omada/icon.png',
'friendly_name': 'Test PoE Switch Firmware',
'in_progress': False,
'installed_version': '1.0.12 Build 20230203 Rel.36545',
'latest_version': '1.0.15 Build 20231101 Rel.40123',
'release_summary': None,
'release_url': None,
'skipped_version': None,
'supported_features': <UpdateEntityFeature: 21>,
'title': None,
'update_percentage': None,
}),
'context': <ANY>,
'entity_id': 'update.test_poe_switch_firmware',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_entities[update.test_router_firmware-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'update',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'update.test_router_firmware',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Firmware',
'options': dict({
}),
'original_device_class': <UpdateDeviceClass.FIRMWARE: 'firmware'>,
'original_icon': None,
'original_name': 'Firmware',
'platform': 'tplink_omada',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <UpdateEntityFeature: 21>,
'translation_key': None,
'unique_id': 'AA-BB-CC-DD-EE-FF_firmware',
'unit_of_measurement': None,
})
# ---
# name: test_entities[update.test_router_firmware-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'auto_update': False,
'device_class': 'firmware',
'display_precision': 0,
'entity_picture': 'https://brands.home-assistant.io/_/tplink_omada/icon.png',
'friendly_name': 'Test Router Firmware',
'in_progress': False,
'installed_version': '1.1.1 Build 20230901 Rel.55651',
'latest_version': '1.1.1 Build 20230901 Rel.55651',
'release_summary': None,
'release_url': None,
'skipped_version': None,
'supported_features': <UpdateEntityFeature: 21>,
'title': None,
'update_percentage': None,
}),
'context': <ANY>,
'entity_id': 'update.test_router_firmware',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---

View File

@@ -18,7 +18,7 @@ from tests.common import (
Generator,
MockConfigEntry,
async_fire_time_changed,
load_json_array_fixture,
async_load_json_array_fixture,
snapshot_platform,
)
@@ -51,7 +51,7 @@ async def test_no_gateway_creates_no_port_sensors(
) -> None:
"""Test that if there is no gateway, no gateway port sensors are created."""
_remove_test_device(mock_omada_site_client, 0)
await _remove_test_device(hass, mock_omada_site_client, 0)
mock_config_entry.add_to_hass(hass)
@@ -70,7 +70,8 @@ async def test_disconnected_device_sensor_not_registered(
) -> None:
"""Test that if the gateway is not connected to the controller, gateway entities are not created."""
_set_test_device_status(
await _set_test_device_status(
hass,
mock_omada_site_client,
0,
DeviceStatus.DISCONNECTED.value,
@@ -87,7 +88,8 @@ async def test_disconnected_device_sensor_not_registered(
assert entity is None
# "Connect" the gateway
_set_test_device_status(
await _set_test_device_status(
hass,
mock_omada_site_client,
0,
DeviceStatus.CONNECTED.value,
@@ -105,13 +107,14 @@ async def test_disconnected_device_sensor_not_registered(
mock_omada_site_client.get_gateway.assert_called_once_with("AA-BB-CC-DD-EE-FF")
def _set_test_device_status(
async def _set_test_device_status(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
dev_index: int,
status: int,
status_category: int,
) -> None:
devices_data = load_json_array_fixture("devices.json", DOMAIN)
devices_data = await async_load_json_array_fixture(hass, "devices.json", DOMAIN)
devices_data[dev_index]["status"] = status
devices_data[dev_index]["statusCategory"] = status_category
devices = [OmadaListDevice(d) for d in devices_data]
@@ -120,11 +123,12 @@ def _set_test_device_status(
mock_omada_site_client.get_devices.return_value = devices
def _remove_test_device(
async def _remove_test_device(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
dev_index: int,
) -> None:
devices_data = load_json_array_fixture("devices.json", DOMAIN)
devices_data = await async_load_json_array_fixture(hass, "devices.json", DOMAIN)
del devices_data[dev_index]
devices = [OmadaListDevice(d) for d in devices_data]

View File

@@ -1,7 +1,6 @@
"""Tests for TP-Link Omada sensor entities."""
from datetime import timedelta
import json
from unittest.mock import MagicMock, patch
from freezegun.api import FrozenDateTimeFactory
@@ -18,7 +17,7 @@ from homeassistant.helpers import entity_registry as er
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
load_fixture,
async_load_json_array_fixture,
snapshot_platform,
)
@@ -63,7 +62,8 @@ async def test_device_specific_status(
assert entity is not None
assert entity.state == "connected"
_set_test_device_status(
await _set_test_device_status(
hass,
mock_omada_site_client,
DeviceStatus.ADOPT_FAILED.value,
DeviceStatusCategory.CONNECTED.value,
@@ -89,9 +89,10 @@ async def test_device_category_status(
assert entity is not None
assert entity.state == "connected"
_set_test_device_status(
await _set_test_device_status(
hass,
mock_omada_site_client,
DeviceStatus.PENDING_WIRELESS,
DeviceStatus.PENDING_WIRELESS.value,
DeviceStatusCategory.PENDING.value,
)
@@ -103,12 +104,13 @@ async def test_device_category_status(
assert entity and entity.state == "pending"
def _set_test_device_status(
async def _set_test_device_status(
hass: HomeAssistant,
mock_omada_site_client: MagicMock,
status: int,
status_category: int,
) -> None:
devices_data = json.loads(load_fixture("devices.json", DOMAIN))
devices_data = await async_load_json_array_fixture(hass, "devices.json", DOMAIN)
devices_data[1]["status"] = status
devices_data[1]["statusCategory"] = status_category
devices = [OmadaListDevice(d) for d in devices_data]

View File

@@ -0,0 +1,204 @@
"""Tests for TP-Link Omada update entities."""
from datetime import timedelta
from unittest.mock import AsyncMock, MagicMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from tplink_omada_client.devices import OmadaListDevice
from tplink_omada_client.exceptions import OmadaClientException, RequestFailed
from homeassistant.components.tplink_omada.coordinator import POLL_DEVICES
from homeassistant.components.update import (
ATTR_IN_PROGRESS,
DOMAIN as UPDATE_DOMAIN,
SERVICE_INSTALL,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_ON, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
async_load_json_array_fixture,
snapshot_platform,
)
from tests.typing import WebSocketGenerator
POLL_INTERVAL = timedelta(seconds=POLL_DEVICES)
async def _rebuild_device_list_with_update(
hass: HomeAssistant, mac: str, **overrides
) -> list[OmadaListDevice]:
"""Rebuild device list from fixture with specified overrides for a device."""
devices_data = await async_load_json_array_fixture(
hass, "devices.json", "tplink_omada"
)
for device_data in devices_data:
if device_data["mac"] == mac:
device_data.update(overrides)
return [OmadaListDevice(d) for d in devices_data]
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_omada_client: MagicMock,
) -> MockConfigEntry:
"""Set up the TP-Link Omada integration for testing."""
mock_config_entry.add_to_hass(hass)
with patch("homeassistant.components.tplink_omada.PLATFORMS", [Platform.UPDATE]):
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry
async def test_entities(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
init_integration: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test the creation of the TP-Link Omada update entities."""
await snapshot_platform(hass, entity_registry, snapshot, init_integration.entry_id)
async def test_firmware_download_in_progress(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_omada_site_client: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test update entity when firmware download is in progress."""
entity_id = "update.test_poe_switch_firmware"
freezer.tick(POLL_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Rebuild device list with fwDownload set to True for the switch
updated_devices = await _rebuild_device_list_with_update(
hass, "54-AF-97-00-00-01", fwDownload=True
)
mock_omada_site_client.get_devices.return_value = updated_devices
# Trigger coordinator update
freezer.tick(POLL_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Verify update entity shows in progress
entity = hass.states.get(entity_id)
assert entity is not None
assert entity.attributes.get(ATTR_IN_PROGRESS) is True
async def test_install_firmware_success(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_omada_site_client: MagicMock,
) -> None:
"""Test successful firmware installation."""
entity_id = "update.test_poe_switch_firmware"
# Verify update is available
entity = hass.states.get(entity_id)
assert entity is not None
assert entity.state == STATE_ON
# Call install service
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
# Verify start_firmware_upgrade was called with the correct device
mock_omada_site_client.start_firmware_upgrade.assert_awaited_once()
await_args = mock_omada_site_client.start_firmware_upgrade.await_args[0]
assert await_args[0].mac == "54-AF-97-00-00-01"
@pytest.mark.parametrize(
("exception_type", "error_message"),
[
(
RequestFailed(500, "Update rejected"),
"Firmware update request rejected",
),
(
OmadaClientException("Connection error"),
"Unable to send Firmware update request. Check the controller is online.",
),
],
)
async def test_install_firmware_exceptions(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_omada_site_client: MagicMock,
exception_type: Exception,
error_message: str,
) -> None:
"""Test firmware installation exception handling."""
entity_id = "update.test_poe_switch_firmware"
# Mock exception
mock_omada_site_client.start_firmware_upgrade = AsyncMock(
side_effect=exception_type
)
# Call install service and expect error
with pytest.raises(
HomeAssistantError,
match=error_message,
):
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
@pytest.mark.parametrize(
("entity_name", "expected_notes"),
[
("test_router", None),
("test_poe_switch", "Bug fixes and performance improvements"),
],
)
async def test_release_notes(
hass: HomeAssistant,
init_integration: MockConfigEntry,
hass_ws_client: WebSocketGenerator,
entity_name: str,
expected_notes: str | None,
) -> None:
"""Test that release notes are available via websocket."""
entity_id = f"update.{entity_name}_firmware"
# Get release notes via websocket
client = await hass_ws_client(hass)
await hass.async_block_till_done()
await client.send_json(
{
"id": 1,
"type": "update/release_notes",
"entity_id": entity_id,
}
)
result = await client.receive_json()
assert expected_notes == result["result"]