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

Review AsusWRT sensors tests (#71307)

Browse source 
* Review AsusWRT sensors tests

* Fix grammar

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
ollo69 2022-12-02 11:22:16 +01:00 • committed by GitHub
parent bf7e50eb32
commit 5789790d32
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 153 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
tests/components/asuswrt/test_config_flow.py
View file

@ -12,6 +12,7 @@ from homeassistant.components.asuswrt.const import (
CONF_SSH_KEY,
CONF_TRACK_UNKNOWN,
DOMAIN,
PROTOCOL_TELNET,
)
from homeassistant.components.device_tracker import CONF_CONSIDER_HOME
from homeassistant.config_entries import SOURCE_USER
@ -34,8 +35,8 @@ SSH_KEY = "1234"
CONFIG_DATA = {
CONF_HOST: HOST,
CONF_PORT: 22,
CONF_PROTOCOL: "telnet",
CONF_PORT: 23,
CONF_PROTOCOL: PROTOCOL_TELNET,
CONF_USERNAME: "user",
CONF_PASSWORD: "pwd",
CONF_MODE: "ap",

231
tests/components/asuswrt/test_sensor.py
View file

@ -6,7 +6,11 @@ from aioasuswrt.asuswrt import Device
import pytest
from homeassistant.components import device_tracker, sensor
from homeassistant.components.asuswrt.const import CONF_INTERFACE, DOMAIN
from homeassistant.components.asuswrt.const import (
CONF_INTERFACE,
DOMAIN,
PROTOCOL_TELNET,
)
from homeassistant.components.asuswrt.router import DEFAULT_NAME
from homeassistant.components.device_tracker import CONF_CONSIDER_HOME
from homeassistant.config_entries import ConfigEntryState
@ -27,13 +31,15 @@ from homeassistant.util.dt import utcnow
from tests.common import MockConfigEntry, async_fire_time_changed
ASUSWRT_LIB = "homeassistant.components.asuswrt.router.AsusWrt"
HOST = "myrouter.asuswrt.com"
IP_ADDRESS = "192.168.1.1"
CONFIG_DATA = {
CONF_HOST: HOST,
CONF_PORT: 22,
CONF_PROTOCOL: "telnet",
CONF_PROTOCOL: PROTOCOL_TELNET,
CONF_USERNAME: "user",
CONF_PASSWORD: "pwd",
CONF_MODE: "router",
@ -50,20 +56,37 @@ MOCK_MAC_2 = "A2:B2:C2:D2:E2:F2"
MOCK_MAC_3 = "A3:B3:C3:D3:E3:F3"
MOCK_MAC_4 = "A4:B4:C4:D4:E4:F4"
SENSOR_NAMES = [
"Devices Connected",
SENSORS_DEFAULT = [
"Download Speed",
"Download",
"Upload Speed",
"Upload",
]
SENSORS_LOADAVG = [
"Load Avg (1m)",
"Load Avg (5m)",
"Load Avg (15m)",
]
SENSORS_TEMP = [
"2.4GHz Temperature",
"5GHz Temperature",
"CPU Temperature",
]
SENSORS_ALL = [*SENSORS_DEFAULT, *SENSORS_LOADAVG, *SENSORS_TEMP]
PATCH_SETUP_ENTRY = patch(
"homeassistant.components.asuswrt.async_setup_entry",
return_value=True,
)
def new_device(mac, ip, name):
"""Return a new device for specific protocol."""
return Device(mac, ip, name)
@pytest.fixture(name="mock_devices")
def mock_devices_fixture():
@ -75,11 +98,9 @@ def mock_devices_fixture():
@pytest.fixture(name="mock_available_temps")
def mock_available_temps_list():
def mock_available_temps_fixture():
"""Mock a list of available temperature sensors."""
# Only length of 3 booleans is valid. First checking the exception handling.
return [True, False]
return [True, False, True]
@pytest.fixture(name="create_device_registry_devices")
@ -103,13 +124,14 @@ def create_device_registry_devices_fixture(hass):
@pytest.fixture(name="connect")
def mock_controller_connect(mock_devices, mock_available_temps):
"""Mock a successful connection."""
with patch("homeassistant.components.asuswrt.router.AsusWrt") as service_mock:
"""Mock a successful connection with AsusWrt library."""
with patch(ASUSWRT_LIB) as service_mock:
service_mock.return_value.connection.async_connect = AsyncMock()
service_mock.return_value.is_connected = True
service_mock.return_value.connection.disconnect = Mock()
service_mock.return_value.async_get_nvram = AsyncMock(
return_value={
"label_mac": MAC_ADDR,
"model": "abcd",
"firmver": "efg",
"buildno": "123",
@ -138,8 +160,8 @@ def mock_controller_connect(mock_devices, mock_available_temps):
@pytest.fixture(name="connect_sens_fail")
def mock_controller_connect_sens_fail():
"""Mock a successful connection with sensor fail."""
with patch("homeassistant.components.asuswrt.router.AsusWrt") as service_mock:
"""Mock a successful connection using AsusWrt library with sensors failing."""
with patch(ASUSWRT_LIB) as service_mock:
service_mock.return_value.connection.async_connect = AsyncMock()
service_mock.return_value.is_connected = True
service_mock.return_value.connection.disconnect = Mock()
@ -159,14 +181,14 @@ def mock_controller_connect_sens_fail():
yield service_mock
def _setup_entry(hass, unique_id=None):
"""Create mock config entry."""
def _setup_entry(hass, config, sensors, unique_id=None):
"""Create mock config entry with enabled sensors."""
entity_reg = er.async_get(hass)
# init config entry
config_entry = MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA,
data=config,
options={CONF_CONSIDER_HOME: 60},
unique_id=unique_id,
)
@ -176,27 +198,13 @@ def _setup_entry(hass, unique_id=None):
sensor_prefix = f"{sensor.DOMAIN}.{obj_prefix}"
# Pre-enable the status sensor
for sensor_name in SENSOR_NAMES:
for sensor_name in sensors:
sensor_id = slugify(sensor_name)
entity_reg.async_get_or_create(
sensor.DOMAIN,
DOMAIN,
f"{DOMAIN} {unique_id or DEFAULT_NAME} {sensor_name}",
suggested_object_id=f"{obj_prefix}_{sensor_id}",
disabled_by=None,
)
# Create the first device tracker to test mac conversion
for mac, name in {
MOCK_MAC_1: "test",
dr.format_mac(MOCK_MAC_2): "testtwo",
MOCK_MAC_2: "testremove",
}.items():
entity_reg.async_get_or_create(
device_tracker.DOMAIN,
DOMAIN,
mac,
suggested_object_id=name,
config_entry=config_entry,
disabled_by=None,
)
@ -212,12 +220,30 @@ async def test_sensors(
hass,
connect,
mock_devices,
mock_available_temps,
create_device_registry_devices,
entry_unique_id,
):
"""Test creating an AsusWRT sensor."""
config_entry, sensor_prefix = _setup_entry(hass, entry_unique_id)
"""Test creating AsusWRT default sensors and tracker."""
config_entry, sensor_prefix = _setup_entry(
hass, CONFIG_DATA, SENSORS_DEFAULT, entry_unique_id
)
# Create the first device tracker to test mac conversion
entity_reg = er.async_get(hass)
for mac, name in {
MOCK_MAC_1: "test",
dr.format_mac(MOCK_MAC_2): "testtwo",
MOCK_MAC_2: "testremove",
}.items():
entity_reg.async_get_or_create(
device_tracker.DOMAIN,
DOMAIN,
mac,
suggested_object_id=name,
config_entry=config_entry,
disabled_by=None,
)
config_entry.add_to_hass(hass)
# initial devices setup
@ -232,17 +258,9 @@ async def test_sensors(
assert hass.states.get(f"{sensor_prefix}_download").state == "60.0"
assert hass.states.get(f"{sensor_prefix}_upload_speed").state == "80.0"
assert hass.states.get(f"{sensor_prefix}_upload").state == "50.0"
assert hass.states.get(f"{sensor_prefix}_load_avg_1m").state == "1.1"
assert hass.states.get(f"{sensor_prefix}_load_avg_5m").state == "1.2"
assert hass.states.get(f"{sensor_prefix}_load_avg_15m").state == "1.3"
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "2"
# assert temperature availability exception is handled correctly
assert not hass.states.get(f"{sensor_prefix}_2_4ghz_temperature")
assert not hass.states.get(f"{sensor_prefix}_5ghz_temperature")
assert not hass.states.get(f"{sensor_prefix}_cpu_temperature")
# remove first track device
# remove first tracked device
mock_devices.pop(MOCK_MAC_1)
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
@ -253,11 +271,11 @@ async def test_sensors(
assert hass.states.get(f"{device_tracker.DOMAIN}.testtwo").state == STATE_HOME
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "1"
# add 2 new device, one unnamed that should be ignored but counted
mock_devices[MOCK_MAC_3] = Device(MOCK_MAC_3, "192.168.1.4", "TestThree")
mock_devices[MOCK_MAC_4] = Device(MOCK_MAC_4, "192.168.1.5", None)
# add 2 new devices, one unnamed that should be ignored but counted
mock_devices[MOCK_MAC_3] = new_device(MOCK_MAC_3, "192.168.1.4", "TestThree")
mock_devices[MOCK_MAC_4] = new_device(MOCK_MAC_4, "192.168.1.5", None)
# change consider home settings to have status not home of removed track device
# change consider home settings to have status not home of removed tracked device
hass.config_entries.async_update_entry(
config_entry, options={CONF_CONSIDER_HOME: 0}
)
@ -271,23 +289,67 @@ async def test_sensors(
assert hass.states.get(f"{device_tracker.DOMAIN}.testthree").state == STATE_HOME
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "3"
# checking temperature sensors without exceptions
mock_available_temps.append(True)
await hass.config_entries.async_reload(config_entry.entry_id)
async def test_loadavg_sensors(
hass,
connect,
):
"""Test creating an AsusWRT load average sensors."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA, SENSORS_LOADAVG)
config_entry.add_to_hass(hass)
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
await hass.async_block_till_done()
# assert temperature sensor available
assert hass.states.get(f"{sensor_prefix}_load_avg_1m").state == "1.1"
assert hass.states.get(f"{sensor_prefix}_load_avg_5m").state == "1.2"
assert hass.states.get(f"{sensor_prefix}_load_avg_15m").state == "1.3"
async def test_temperature_sensors_fail(
hass,
connect,
mock_available_temps,
):
"""Test fail creating AsusWRT temperature sensors."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA, SENSORS_TEMP)
config_entry.add_to_hass(hass)
# Only length of 3 booleans is valid. Checking the exception handling.
mock_available_temps.pop(2)
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
# assert temperature availability exception is handled correctly
assert not hass.states.get(f"{sensor_prefix}_2_4ghz_temperature")
assert not hass.states.get(f"{sensor_prefix}_5ghz_temperature")
assert not hass.states.get(f"{sensor_prefix}_cpu_temperature")
async def test_temperature_sensors(
hass,
connect,
):
"""Test creating a AsusWRT temperature sensors."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA, SENSORS_TEMP)
config_entry.add_to_hass(hass)
# initial devices setup
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
await hass.async_block_till_done()
# assert temperature sensor available
assert hass.states.get(f"{sensor_prefix}_2_4ghz_temperature").state == "40.0"
assert not hass.states.get(f"{sensor_prefix}_5ghz_temperature")
assert hass.states.get(f"{sensor_prefix}_cpu_temperature").state == "71.2"
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "3"
# change an option that require integration reload
hass.config_entries.async_update_entry(
config_entry, options={CONF_CONSIDER_HOME: 60, CONF_INTERFACE: "eth1"}
)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "3"
@pytest.mark.parametrize(
@ -304,10 +366,11 @@ async def test_connect_fail(hass, side_effect):
)
config_entry.add_to_hass(hass)
with patch("homeassistant.components.asuswrt.router.AsusWrt") as asus_wrt:
with patch(ASUSWRT_LIB) as asus_wrt:
asus_wrt.return_value.connection.async_connect = AsyncMock(
side_effect=side_effect
)
asus_wrt.return_value.async_get_nvram = AsyncMock()
asus_wrt.return_value.is_connected = False
# initial setup fail
@ -316,12 +379,9 @@ async def test_connect_fail(hass, side_effect):
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_sensors_polling_fails(
hass,
connect_sens_fail,
):
async def test_sensors_polling_fails(hass, connect_sens_fail):
"""Test AsusWRT sensors are unavailable when polling fails."""
config_entry, sensor_prefix = _setup_entry(hass)
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA, SENSORS_ALL)
config_entry.add_to_hass(hass)
# initial devices setup
@ -330,21 +390,30 @@ async def test_sensors_polling_fails(
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
await hass.async_block_till_done()
assert hass.states.get(f"{sensor_prefix}_download_speed").state == STATE_UNAVAILABLE
assert hass.states.get(f"{sensor_prefix}_download").state == STATE_UNAVAILABLE
assert hass.states.get(f"{sensor_prefix}_upload_speed").state == STATE_UNAVAILABLE
assert hass.states.get(f"{sensor_prefix}_upload").state == STATE_UNAVAILABLE
assert hass.states.get(f"{sensor_prefix}_load_avg_1m").state == STATE_UNAVAILABLE
assert hass.states.get(f"{sensor_prefix}_load_avg_5m").state == STATE_UNAVAILABLE
assert hass.states.get(f"{sensor_prefix}_load_avg_15m").state == STATE_UNAVAILABLE
for sensor_name in SENSORS_ALL:
assert (
hass.states.get(f"{sensor_prefix}_{slugify(sensor_name)}").state
== STATE_UNAVAILABLE
)
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "0"
assert (
hass.states.get(f"{sensor_prefix}_2_4ghz_temperature").state
== STATE_UNAVAILABLE
)
assert (
hass.states.get(f"{sensor_prefix}_5ghz_temperature").state == STATE_UNAVAILABLE
)
assert (
hass.states.get(f"{sensor_prefix}_cpu_temperature").state == STATE_UNAVAILABLE
)
async def test_options_reload(hass, connect):
"""Test AsusWRT integration is reload changing an options that require this."""
config_entry = MockConfigEntry(domain=DOMAIN, data=CONFIG_DATA, unique_id=MAC_ADDR)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
await hass.async_block_till_done()
with PATCH_SETUP_ENTRY as setup_entry_call:
# change an option that requires integration reload
hass.config_entries.async_update_entry(
config_entry, options={CONF_INTERFACE: "eth1"}
)
await hass.async_block_till_done()
assert setup_entry_call.called
assert config_entry.state is ConfigEntryState.LOADED