Rework devolo Home Network tests (#74472)

This commit is contained in:
Guido Schmitz 2022-10-03 18:21:45 +02:00 committed by GitHub
parent 55b036ec5e
commit 3e3f7ea995
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 172 additions and 136 deletions

View File

@ -1,16 +1,9 @@
"""Tests for the devolo Home Network integration."""
import dataclasses
from typing import Any
from devolo_plc_api.device_api.deviceapi import DeviceApi
from devolo_plc_api.plcnet_api.plcnetapi import PlcNetApi
from homeassistant.components.devolo_home_network.const import DOMAIN
from homeassistant.const import CONF_IP_ADDRESS
from homeassistant.core import HomeAssistant
from .const import DISCOVERY_INFO, IP
from .const import IP
from tests.common import MockConfigEntry
@ -24,12 +17,3 @@ def configure_integration(hass: HomeAssistant) -> MockConfigEntry:
entry.add_to_hass(hass)
return entry
async def async_connect(self, session_instance: Any = None):
"""Give a mocked device the needed properties."""
self.plcnet = PlcNetApi(IP, None, dataclasses.asdict(DISCOVERY_INFO))
self.device = DeviceApi(IP, None, dataclasses.asdict(DISCOVERY_INFO))
self.mac = DISCOVERY_INFO.properties["PlcMacAddress"]
self.product = DISCOVERY_INFO.properties["Product"]
self.serial_number = DISCOVERY_INFO.properties["SN"]

View File

@ -1,29 +1,22 @@
"""Fixtures for tests."""
from unittest.mock import AsyncMock, patch
from itertools import cycle
from unittest.mock import patch
import pytest
from . import async_connect
from .const import CONNECTED_STATIONS, DISCOVERY_INFO, NEIGHBOR_ACCESS_POINTS, PLCNET
from .const import DISCOVERY_INFO, IP
from .mock import MockDevice
@pytest.fixture()
def mock_device():
"""Mock connecting to a devolo home network device."""
with patch("devolo_plc_api.device.Device.async_connect", async_connect), patch(
"devolo_plc_api.device.Device.async_disconnect"
), patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_connected_station",
new=AsyncMock(return_value=CONNECTED_STATIONS),
), patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_neighbor_access_points",
new=AsyncMock(return_value=NEIGHBOR_ACCESS_POINTS),
), patch(
"devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.async_get_network_overview",
new=AsyncMock(return_value=PLCNET),
device = MockDevice(ip=IP)
with patch(
"homeassistant.components.devolo_home_network.Device",
side_effect=cycle([device]),
):
yield
yield device
@pytest.fixture(name="info")

View File

@ -0,0 +1,57 @@
"""Mock of a devolo Home Network device."""
from __future__ import annotations
import dataclasses
from typing import Any
from unittest.mock import AsyncMock
from devolo_plc_api.device import Device
from devolo_plc_api.device_api.deviceapi import DeviceApi
from devolo_plc_api.plcnet_api.plcnetapi import PlcNetApi
import httpx
from zeroconf import Zeroconf
from zeroconf.asyncio import AsyncZeroconf
from .const import (
CONNECTED_STATIONS,
DISCOVERY_INFO,
IP,
NEIGHBOR_ACCESS_POINTS,
PLCNET,
)
class MockDevice(Device):
"""Mock of a devolo Home Network device."""
def __init__(
self,
ip: str,
plcnetapi: dict[str, Any] | None = None,
deviceapi: dict[str, Any] | None = None,
zeroconf_instance: AsyncZeroconf | Zeroconf | None = None,
) -> None:
"""Bring mock in a well defined state."""
super().__init__(ip, plcnetapi, deviceapi, zeroconf_instance)
self.reset()
async def async_connect(
self, session_instance: httpx.AsyncClient | None = None
) -> None:
"""Give a mocked device the needed properties."""
self.mac = DISCOVERY_INFO.properties["PlcMacAddress"]
self.product = DISCOVERY_INFO.properties["Product"]
self.serial_number = DISCOVERY_INFO.properties["SN"]
def reset(self):
"""Reset mock to starting point."""
self.async_disconnect = AsyncMock()
self.device = DeviceApi(IP, None, dataclasses.asdict(DISCOVERY_INFO))
self.device.async_get_wifi_connected_station = AsyncMock(
return_value=CONNECTED_STATIONS
)
self.device.async_get_wifi_neighbor_access_points = AsyncMock(
return_value=NEIGHBOR_ACCESS_POINTS
)
self.plcnet = PlcNetApi(IP, None, dataclasses.asdict(DISCOVERY_INFO))
self.plcnet.async_get_network_overview = AsyncMock(return_value=PLCNET)

View File

@ -1,5 +1,5 @@
"""Tests for the devolo Home Network sensors."""
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock
from devolo_plc_api.exceptions.device import DeviceUnavailable
import pytest
@ -22,6 +22,7 @@ from homeassistant.util import dt
from . import configure_integration
from .const import PLCNET_ATTACHED
from .mock import MockDevice
from tests.common import async_fire_time_changed
@ -39,8 +40,8 @@ async def test_binary_sensor_setup(hass: HomeAssistant):
await hass.config_entries.async_unload(entry.entry_id)
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "mock_device")
async def test_update_attached_to_router(hass: HomeAssistant):
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_update_attached_to_router(hass: HomeAssistant, mock_device: MockDevice):
"""Test state change of a attached_to_router binary sensor device."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
@ -59,27 +60,25 @@ async def test_update_attached_to_router(hass: HomeAssistant):
assert er.async_get(state_key).entity_category == EntityCategory.DIAGNOSTIC
# Emulate device failure
with patch(
"devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.async_get_network_overview",
side_effect=DeviceUnavailable,
):
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
mock_device.plcnet.async_get_network_overview = AsyncMock(
side_effect=DeviceUnavailable
)
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
# Emulate state change
with patch(
"devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.async_get_network_overview",
new=AsyncMock(return_value=PLCNET_ATTACHED),
):
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
mock_device.plcnet.async_get_network_overview = AsyncMock(
return_value=PLCNET_ATTACHED
)
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_ON
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_ON
await hass.config_entries.async_unload(entry.entry_id)

View File

@ -19,6 +19,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .const import DISCOVERY_INFO, DISCOVERY_INFO_WRONG_DEVICE, IP
from .mock import MockDevice
async def test_form(hass: HomeAssistant, info: dict[str, Any]):
@ -167,10 +168,12 @@ async def test_abort_if_configued(hass: HomeAssistant):
assert result3["reason"] == "already_configured"
@pytest.mark.usefixtures("mock_device")
@pytest.mark.usefixtures("mock_zeroconf")
async def test_validate_input(hass: HomeAssistant):
"""Test input validation."""
info = await config_flow.validate_input(hass, {CONF_IP_ADDRESS: IP})
assert SERIAL_NUMBER in info
assert TITLE in info
with patch(
"homeassistant.components.devolo_home_network.config_flow.Device",
new=MockDevice,
):
info = await config_flow.validate_input(hass, {CONF_IP_ADDRESS: IP})
assert SERIAL_NUMBER in info
assert TITLE in info

View File

@ -1,8 +1,7 @@
"""Tests for the devolo Home Network device tracker."""
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock
from devolo_plc_api.exceptions.device import DeviceUnavailable
import pytest
from homeassistant.components.device_tracker import DOMAIN as PLATFORM
from homeassistant.components.devolo_home_network.const import (
@ -23,6 +22,7 @@ from homeassistant.util import dt
from . import configure_integration
from .const import CONNECTED_STATIONS, DISCOVERY_INFO, NO_CONNECTED_STATIONS
from .mock import MockDevice
from tests.common import async_fire_time_changed
@ -30,8 +30,7 @@ STATION = CONNECTED_STATIONS["connected_stations"][0]
SERIAL = DISCOVERY_INFO.properties["SN"]
@pytest.mark.usefixtures("mock_device")
async def test_device_tracker(hass: HomeAssistant):
async def test_device_tracker(hass: HomeAssistant, mock_device: MockDevice):
"""Test device tracker states."""
state_key = f"{PLATFORM}.{DOMAIN}_{SERIAL}_{STATION['mac_address'].lower().replace(':', '_')}"
entry = configure_integration(hass)
@ -57,34 +56,31 @@ async def test_device_tracker(hass: HomeAssistant):
)
# Emulate state change
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_connected_station",
new=AsyncMock(return_value=NO_CONNECTED_STATIONS),
):
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
mock_device.device.async_get_wifi_connected_station = AsyncMock(
return_value=NO_CONNECTED_STATIONS
)
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_NOT_HOME
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_NOT_HOME
# Emulate device failure
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_connected_station",
side_effect=DeviceUnavailable,
):
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
mock_device.device.async_get_wifi_connected_station = AsyncMock(
side_effect=DeviceUnavailable
)
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
await hass.config_entries.async_unload(entry.entry_id)
@pytest.mark.usefixtures("mock_device")
async def test_restoring_clients(hass: HomeAssistant):
async def test_restoring_clients(hass: HomeAssistant, mock_device: MockDevice):
"""Test restoring existing device_tracker entities."""
state_key = f"{PLATFORM}.{DOMAIN}_{SERIAL}_{STATION['mac_address'].lower().replace(':', '_')}"
entry = configure_integration(hass)
@ -96,12 +92,13 @@ async def test_restoring_clients(hass: HomeAssistant):
config_entry=entry,
)
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_connected_station",
new=AsyncMock(return_value=NO_CONNECTED_STATIONS),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_NOT_HOME
mock_device.device.async_get_wifi_connected_station = AsyncMock(
return_value=NO_CONNECTED_STATIONS
)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_NOT_HOME

View File

@ -9,6 +9,7 @@ from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from . import configure_integration
from .mock import MockDevice
@pytest.mark.usefixtures("mock_device")
@ -44,15 +45,11 @@ async def test_unload_entry(hass: HomeAssistant):
assert entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.usefixtures("mock_device")
async def test_hass_stop(hass: HomeAssistant):
async def test_hass_stop(hass: HomeAssistant, mock_device: MockDevice):
"""Test homeassistant stop event."""
entry = configure_integration(hass)
with patch(
"homeassistant.components.devolo_home_network.Device.async_disconnect"
) as async_disconnect:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
async_disconnect.assert_called_once()
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
mock_device.async_disconnect.assert_called_once()

View File

@ -1,5 +1,5 @@
"""Tests for the devolo Home Network sensors."""
from unittest.mock import patch
from unittest.mock import AsyncMock
from devolo_plc_api.exceptions.device import DeviceUnavailable
import pytest
@ -16,6 +16,7 @@ from homeassistant.helpers.entity import EntityCategory
from homeassistant.util import dt
from . import configure_integration
from .mock import MockDevice
from tests.common import async_fire_time_changed
@ -35,8 +36,9 @@ async def test_sensor_setup(hass: HomeAssistant):
await hass.config_entries.async_unload(entry.entry_id)
@pytest.mark.usefixtures("mock_device")
async def test_update_connected_wifi_clients(hass: HomeAssistant):
async def test_update_connected_wifi_clients(
hass: HomeAssistant, mock_device: MockDevice
):
"""Test state change of a connected_wifi_clients sensor device."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
@ -53,18 +55,18 @@ async def test_update_connected_wifi_clients(hass: HomeAssistant):
assert state.attributes["state_class"] == SensorStateClass.MEASUREMENT
# Emulate device failure
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_connected_station",
side_effect=DeviceUnavailable,
):
async_fire_time_changed(hass, dt.utcnow() + SHORT_UPDATE_INTERVAL)
await hass.async_block_till_done()
mock_device.device.async_get_wifi_connected_station = AsyncMock(
side_effect=DeviceUnavailable
)
async_fire_time_changed(hass, dt.utcnow() + SHORT_UPDATE_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
# Emulate state change
mock_device.reset()
async_fire_time_changed(hass, dt.utcnow() + SHORT_UPDATE_INTERVAL)
await hass.async_block_till_done()
@ -75,8 +77,10 @@ async def test_update_connected_wifi_clients(hass: HomeAssistant):
await hass.config_entries.async_unload(entry.entry_id)
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "mock_device")
async def test_update_neighboring_wifi_networks(hass: HomeAssistant):
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_update_neighboring_wifi_networks(
hass: HomeAssistant, mock_device: MockDevice
):
"""Test state change of a neighboring_wifi_networks sensor device."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
@ -95,18 +99,18 @@ async def test_update_neighboring_wifi_networks(hass: HomeAssistant):
assert er.async_get(state_key).entity_category is EntityCategory.DIAGNOSTIC
# Emulate device failure
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_neighbor_access_points",
side_effect=DeviceUnavailable,
):
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
mock_device.device.async_get_wifi_neighbor_access_points = AsyncMock(
side_effect=DeviceUnavailable
)
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
# Emulate state change
mock_device.reset()
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
@ -117,8 +121,10 @@ async def test_update_neighboring_wifi_networks(hass: HomeAssistant):
await hass.config_entries.async_unload(entry.entry_id)
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "mock_device")
async def test_update_connected_plc_devices(hass: HomeAssistant):
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_update_connected_plc_devices(
hass: HomeAssistant, mock_device: MockDevice
):
"""Test state change of a connected_plc_devices sensor device."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
@ -136,18 +142,18 @@ async def test_update_connected_plc_devices(hass: HomeAssistant):
assert er.async_get(state_key).entity_category is EntityCategory.DIAGNOSTIC
# Emulate device failure
with patch(
"devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.async_get_network_overview",
side_effect=DeviceUnavailable,
):
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
mock_device.plcnet.async_get_network_overview = AsyncMock(
side_effect=DeviceUnavailable
)
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
# Emulate state change
mock_device.reset()
async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()