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

Improve AsusWRT integration tests (#102810)

Browse source
This commit is contained in:
ollo69 2023-11-09 09:18:32 +01:00 • committed by GitHub
parent 78add0f51d
commit 143e114528
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 424 additions and 369 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
tests/components/asuswrt/common.py Normal file
View file

@ -0,0 +1,53 @@
"""Test code shared between test files."""
from aioasuswrt.asuswrt import Device as LegacyDevice
from homeassistant.components.asuswrt.const import (
CONF_SSH_KEY,
MODE_ROUTER,
PROTOCOL_SSH,
PROTOCOL_TELNET,
)
from homeassistant.const import (
CONF_HOST,
CONF_MODE,
CONF_PASSWORD,
CONF_PORT,
CONF_PROTOCOL,
CONF_USERNAME,
)
ASUSWRT_BASE = "homeassistant.components.asuswrt"
HOST = "myrouter.asuswrt.com"
ROUTER_MAC_ADDR = "a1:b2:c3:d4:e5:f6"
CONFIG_DATA_TELNET = {
CONF_HOST: HOST,
CONF_PORT: 23,
CONF_PROTOCOL: PROTOCOL_TELNET,
CONF_USERNAME: "user",
CONF_PASSWORD: "pwd",
CONF_MODE: MODE_ROUTER,
}
CONFIG_DATA_SSH = {
CONF_HOST: HOST,
CONF_PORT: 22,
CONF_PROTOCOL: PROTOCOL_SSH,
CONF_USERNAME: "user",
CONF_SSH_KEY: "aaaaa",
CONF_MODE: MODE_ROUTER,
}
MOCK_MACS = [
"A1:B1:C1:D1:E1:F1",
"A2:B2:C2:D2:E2:F2",
"A3:B3:C3:D3:E3:F3",
"A4:B4:C4:D4:E4:F4",
]
def new_device(mac, ip, name):
"""Return a new device for specific protocol."""
return LegacyDevice(mac, ip, name)

83
tests/components/asuswrt/conftest.py Normal file
View file

@ -0,0 +1,83 @@
"""Fixtures for Asuswrt component."""
from unittest.mock import Mock, patch
from aioasuswrt.asuswrt import AsusWrt as AsusWrtLegacy
from aioasuswrt.connection import TelnetConnection
import pytest
from .common import ASUSWRT_BASE, MOCK_MACS, ROUTER_MAC_ADDR, new_device
ASUSWRT_LEGACY_LIB = f"{ASUSWRT_BASE}.bridge.AsusWrtLegacy"
MOCK_BYTES_TOTAL = [60000000000, 50000000000]
MOCK_CURRENT_TRANSFER_RATES = [20000000, 10000000]
MOCK_LOAD_AVG = [1.1, 1.2, 1.3]
MOCK_TEMPERATURES = {"2.4GHz": 40.2, "CPU": 71.2}
@pytest.fixture(name="patch_setup_entry")
def mock_controller_patch_setup_entry():
"""Mock setting up a config entry."""
with patch(
f"{ASUSWRT_BASE}.async_setup_entry", return_value=True
) as setup_entry_mock:
yield setup_entry_mock
@pytest.fixture(name="mock_devices_legacy")
def mock_devices_legacy_fixture():
"""Mock a list of devices."""
return {
MOCK_MACS[0]: new_device(MOCK_MACS[0], "192.168.1.2", "Test"),
MOCK_MACS[1]: new_device(MOCK_MACS[1], "192.168.1.3", "TestTwo"),
}
@pytest.fixture(name="mock_available_temps")
def mock_available_temps_fixture():
"""Mock a list of available temperature sensors."""
return [True, False, True]
@pytest.fixture(name="connect_legacy")
def mock_controller_connect_legacy(mock_devices_legacy, mock_available_temps):
"""Mock a successful connection with legacy library."""
with patch(ASUSWRT_LEGACY_LIB, spec=AsusWrtLegacy) as service_mock:
service_mock.return_value.connection = Mock(spec=TelnetConnection)
service_mock.return_value.is_connected = True
service_mock.return_value.async_get_nvram.return_value = {
"label_mac": ROUTER_MAC_ADDR,
"model": "abcd",
"firmver": "efg",
"buildno": "123",
}
service_mock.return_value.async_get_connected_devices.return_value = (
mock_devices_legacy
)
service_mock.return_value.async_get_bytes_total.return_value = MOCK_BYTES_TOTAL
service_mock.return_value.async_get_current_transfer_rates.return_value = (
MOCK_CURRENT_TRANSFER_RATES
)
service_mock.return_value.async_get_loadavg.return_value = MOCK_LOAD_AVG
service_mock.return_value.async_get_temperature.return_value = MOCK_TEMPERATURES
service_mock.return_value.async_find_temperature_commands.return_value = (
mock_available_temps
)
yield service_mock
@pytest.fixture(name="connect_legacy_sens_fail")
def mock_controller_connect_legacy_sens_fail(connect_legacy):
"""Mock a successful connection using legacy library with sensors fail."""
connect_legacy.return_value.async_get_nvram.side_effect = OSError
connect_legacy.return_value.async_get_connected_devices.side_effect = OSError
connect_legacy.return_value.async_get_bytes_total.side_effect = OSError
connect_legacy.return_value.async_get_current_transfer_rates.side_effect = OSError
connect_legacy.return_value.async_get_loadavg.side_effect = OSError
connect_legacy.return_value.async_get_temperature.side_effect = OSError
connect_legacy.return_value.async_find_temperature_commands.return_value = [
True,
True,
True,
]

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

@ -1,6 +1,6 @@
"""Tests for the AsusWrt config flow."""
from socket import gaierror
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import patch
import pytest
@ -12,105 +12,75 @@ from homeassistant.components.asuswrt.const import (
CONF_SSH_KEY,
CONF_TRACK_UNKNOWN,
DOMAIN,
PROTOCOL_TELNET,
MODE_AP,
)
from homeassistant.components.device_tracker import CONF_CONSIDER_HOME
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import (
CONF_HOST,
CONF_MODE,
CONF_PASSWORD,
CONF_PORT,
CONF_PROTOCOL,
CONF_USERNAME,
)
from homeassistant.const import CONF_HOST, CONF_MODE, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from .common import ASUSWRT_BASE, CONFIG_DATA_TELNET, HOST, ROUTER_MAC_ADDR
from tests.common import MockConfigEntry
HOST = "myrouter.asuswrt.com"
IP_ADDRESS = "192.168.1.1"
MAC_ADDR = "a1:b1:c1:d1:e1:f1"
SSH_KEY = "1234"
CONFIG_DATA = {
CONF_HOST: HOST,
CONF_PORT: 23,
CONF_PROTOCOL: PROTOCOL_TELNET,
CONF_USERNAME: "user",
CONF_PASSWORD: "pwd",
CONF_MODE: "ap",
}
PATCH_GET_HOST = patch(
"homeassistant.components.asuswrt.config_flow.socket.gethostbyname",
return_value=IP_ADDRESS,
)
PATCH_SETUP_ENTRY = patch(
"homeassistant.components.asuswrt.async_setup_entry",
return_value=True,
)
@pytest.fixture(name="patch_get_host", autouse=True)
def mock_controller_patch_get_host():
"""Mock call to socket gethostbyname function."""
with patch(
f"{ASUSWRT_BASE}.config_flow.socket.gethostbyname", return_value="192.168.1.1"
) as get_host_mock:
yield get_host_mock
@pytest.fixture(name="mock_unique_id")
def mock_unique_id_fixture():
"""Mock returned unique id."""
return {}
@pytest.fixture(name="patch_is_file", autouse=True)
def mock_controller_patch_is_file():
"""Mock call to os path.isfile function."""
with patch(
f"{ASUSWRT_BASE}.config_flow.os.path.isfile", return_value=True
) as is_file_mock:
yield is_file_mock
@pytest.fixture(name="connect")
def mock_controller_connect(mock_unique_id):
"""Mock a successful connection."""
with patch("homeassistant.components.asuswrt.bridge.AsusWrtLegacy") 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=mock_unique_id
)
yield service_mock
@pytest.mark.usefixtures("connect")
@pytest.mark.parametrize(
"unique_id",
[{}, {"label_mac": MAC_ADDR}],
)
async def test_user(hass: HomeAssistant, mock_unique_id, unique_id) -> None:
@pytest.mark.parametrize("unique_id", [{}, {"label_mac": ROUTER_MAC_ADDR}])
async def test_user(
hass: HomeAssistant, connect_legacy, patch_setup_entry, unique_id
) -> None:
"""Test user config."""
mock_unique_id.update(unique_id)
flow_result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER, "show_advanced_options": True}
)
assert flow_result["type"] == data_entry_flow.FlowResultType.FORM
assert flow_result["step_id"] == "user"
connect_legacy.return_value.async_get_nvram.return_value = unique_id
# test with all provided
with PATCH_GET_HOST, PATCH_SETUP_ENTRY as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
flow_result["flow_id"],
user_input=CONFIG_DATA,
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
flow_result["flow_id"],
user_input=CONFIG_DATA_TELNET,
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == CONFIG_DATA
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == CONFIG_DATA_TELNET
assert len(mock_setup_entry.mock_calls) == 1
assert len(patch_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("config", "error"),
[
({CONF_PASSWORD: None}, "pwd_or_ssh"),
({CONF_SSH_KEY: SSH_KEY}, "pwd_and_ssh"),
({}, "pwd_or_ssh"),
({CONF_PASSWORD: "pwd", CONF_SSH_KEY: SSH_KEY}, "pwd_and_ssh"),
],
)
async def test_error_wrong_password_ssh(hass: HomeAssistant, config, error) -> None:
"""Test we abort for wrong password and ssh file combination."""
config_data = CONFIG_DATA.copy()
config_data = {k: v for k, v in CONFIG_DATA_TELNET.items() if k != CONF_PASSWORD}
config_data.update(config)
result = await hass.config_entries.flow.async_init(
DOMAIN,
@ -122,102 +92,94 @@ async def test_error_wrong_password_ssh(hass: HomeAssistant, config, error) -> N
assert result["errors"] == {"base": error}
async def test_error_invalid_ssh(hass: HomeAssistant) -> None:
async def test_error_invalid_ssh(hass: HomeAssistant, patch_is_file) -> None:
"""Test we abort if invalid ssh file is provided."""
config_data = CONFIG_DATA.copy()
config_data.pop(CONF_PASSWORD)
config_data = {k: v for k, v in CONFIG_DATA_TELNET.items() if k != CONF_PASSWORD}
config_data[CONF_SSH_KEY] = SSH_KEY
with patch(
"homeassistant.components.asuswrt.config_flow.os.path.isfile",
return_value=False,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
data=config_data,
)
patch_is_file.return_value = False
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
data=config_data,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "ssh_not_file"}
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "ssh_not_file"}
async def test_error_invalid_host(hass: HomeAssistant) -> None:
async def test_error_invalid_host(hass: HomeAssistant, patch_get_host) -> None:
"""Test we abort if host name is invalid."""
with patch(
"homeassistant.components.asuswrt.config_flow.socket.gethostbyname",
side_effect=gaierror,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=CONFIG_DATA,
)
patch_get_host.side_effect = gaierror
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=CONFIG_DATA_TELNET,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "invalid_host"}
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": "invalid_host"}
async def test_abort_if_not_unique_id_setup(hass: HomeAssistant) -> None:
"""Test we abort if component without uniqueid is already setup."""
MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA,
data=CONFIG_DATA_TELNET,
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=CONFIG_DATA,
data=CONFIG_DATA_TELNET,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "no_unique_id"
@pytest.mark.usefixtures("connect")
async def test_update_uniqueid_exist(hass: HomeAssistant, mock_unique_id) -> None:
async def test_update_uniqueid_exist(
hass: HomeAssistant, connect_legacy, patch_setup_entry
) -> None:
"""Test we update entry if uniqueid is already configured."""
mock_unique_id.update({"label_mac": MAC_ADDR})
existing_entry = MockConfigEntry(
domain=DOMAIN,
data={**CONFIG_DATA, CONF_HOST: "10.10.10.10"},
unique_id=MAC_ADDR,
data={**CONFIG_DATA_TELNET, CONF_HOST: "10.10.10.10"},
unique_id=ROUTER_MAC_ADDR,
)
existing_entry.add_to_hass(hass)
# test with all provided
with PATCH_GET_HOST, PATCH_SETUP_ENTRY:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
data=CONFIG_DATA,
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
data=CONFIG_DATA_TELNET,
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == CONFIG_DATA
prev_entry = hass.config_entries.async_get_entry(existing_entry.entry_id)
assert not prev_entry
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == CONFIG_DATA_TELNET
prev_entry = hass.config_entries.async_get_entry(existing_entry.entry_id)
assert not prev_entry
@pytest.mark.usefixtures("connect")
async def test_abort_invalid_unique_id(hass: HomeAssistant) -> None:
async def test_abort_invalid_unique_id(hass: HomeAssistant, connect_legacy) -> None:
"""Test we abort if uniqueid not available."""
MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA,
unique_id=MAC_ADDR,
data=CONFIG_DATA_TELNET,
unique_id=ROUTER_MAC_ADDR,
).add_to_hass(hass)
with PATCH_GET_HOST:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=CONFIG_DATA,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "invalid_unique_id"
connect_legacy.return_value.async_get_nvram.return_value = {}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=CONFIG_DATA_TELNET,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "invalid_unique_id"
@pytest.mark.parametrize(
@ -228,95 +190,93 @@ async def test_abort_invalid_unique_id(hass: HomeAssistant) -> None:
(None, "cannot_connect"),
],
)
async def test_on_connect_failed(hass: HomeAssistant, side_effect, error) -> None:
async def test_on_connect_failed(
hass: HomeAssistant, connect_legacy, side_effect, error
) -> None:
"""Test when we have errors connecting the router."""
flow_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER, "show_advanced_options": True},
)
with PATCH_GET_HOST, patch(
"homeassistant.components.asuswrt.bridge.AsusWrtLegacy"
) as asus_wrt:
asus_wrt.return_value.connection.async_connect = AsyncMock(
side_effect=side_effect
)
asus_wrt.return_value.async_get_nvram = AsyncMock(return_value={})
asus_wrt.return_value.is_connected = False
connect_legacy.return_value.is_connected = False
connect_legacy.return_value.connection.async_connect.side_effect = side_effect
result = await hass.config_entries.flow.async_configure(
flow_result["flow_id"], user_input=CONFIG_DATA
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": error}
result = await hass.config_entries.flow.async_configure(
flow_result["flow_id"], user_input=CONFIG_DATA_TELNET
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {"base": error}
async def test_options_flow_ap(hass: HomeAssistant) -> None:
async def test_options_flow_ap(hass: HomeAssistant, patch_setup_entry) -> None:
"""Test config flow options for ap mode."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA,
data={**CONFIG_DATA_TELNET, CONF_MODE: MODE_AP},
options={CONF_REQUIRE_IP: True},
)
config_entry.add_to_hass(hass)
with PATCH_SETUP_ENTRY:
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert CONF_REQUIRE_IP in result["data_schema"].schema
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert CONF_REQUIRE_IP in result["data_schema"].schema
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_CONSIDER_HOME: 20,
CONF_TRACK_UNKNOWN: True,
CONF_INTERFACE: "aaa",
CONF_DNSMASQ: "bbb",
CONF_REQUIRE_IP: False,
},
)
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_CONSIDER_HOME: 20,
CONF_TRACK_UNKNOWN: True,
CONF_INTERFACE: "aaa",
CONF_DNSMASQ: "bbb",
CONF_REQUIRE_IP: False,
},
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert config_entry.options[CONF_CONSIDER_HOME] == 20
assert config_entry.options[CONF_TRACK_UNKNOWN] is True
assert config_entry.options[CONF_INTERFACE] == "aaa"
assert config_entry.options[CONF_DNSMASQ] == "bbb"
assert config_entry.options[CONF_REQUIRE_IP] is False
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert config_entry.options == {
CONF_CONSIDER_HOME: 20,
CONF_TRACK_UNKNOWN: True,
CONF_INTERFACE: "aaa",
CONF_DNSMASQ: "bbb",
CONF_REQUIRE_IP: False,
}
async def test_options_flow_router(hass: HomeAssistant) -> None:
async def test_options_flow_router(hass: HomeAssistant, patch_setup_entry) -> None:
"""Test config flow options for router mode."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={**CONFIG_DATA, CONF_MODE: "router"},
data=CONFIG_DATA_TELNET,
)
config_entry.add_to_hass(hass)
with PATCH_SETUP_ENTRY:
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert CONF_REQUIRE_IP not in result["data_schema"].schema
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert CONF_REQUIRE_IP not in result["data_schema"].schema
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_CONSIDER_HOME: 20,
CONF_TRACK_UNKNOWN: True,
CONF_INTERFACE: "aaa",
CONF_DNSMASQ: "bbb",
},
)
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_CONSIDER_HOME: 20,
CONF_TRACK_UNKNOWN: True,
CONF_INTERFACE: "aaa",
CONF_DNSMASQ: "bbb",
},
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert config_entry.options[CONF_CONSIDER_HOME] == 20
assert config_entry.options[CONF_TRACK_UNKNOWN] is True
assert config_entry.options[CONF_INTERFACE] == "aaa"
assert config_entry.options[CONF_DNSMASQ] == "bbb"
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert config_entry.options == {
CONF_CONSIDER_HOME: 20,
CONF_TRACK_UNKNOWN: True,
CONF_INTERFACE: "aaa",
CONF_DNSMASQ: "bbb",
}

41
tests/components/asuswrt/test_diagnostics.py Normal file
View file

@ -0,0 +1,41 @@
"""Tests for the diagnostics data provided by the AsusWRT integration."""
from homeassistant.components.asuswrt.const import DOMAIN
from homeassistant.components.asuswrt.diagnostics import TO_REDACT
from homeassistant.components.device_tracker import CONF_CONSIDER_HOME
from homeassistant.components.diagnostics import async_redact_data
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from .common import CONFIG_DATA_TELNET, ROUTER_MAC_ADDR
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
async def test_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
connect_legacy,
) -> None:
"""Test diagnostics."""
mock_config_entry = MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA_TELNET,
options={CONF_CONSIDER_HOME: 60},
unique_id=ROUTER_MAC_ADDR,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state == ConfigEntryState.LOADED
entry_dict = async_redact_data(mock_config_entry.as_dict(), TO_REDACT)
result = await get_diagnostics_for_config_entry(
hass, hass_client, mock_config_entry
)
assert result["entry"] == entry_dict

30
tests/components/asuswrt/test_init.py Normal file
View file

@ -0,0 +1,30 @@
"""Tests for the AsusWrt integration."""
from homeassistant.components.asuswrt.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from .common import CONFIG_DATA_TELNET, ROUTER_MAC_ADDR
from tests.common import MockConfigEntry
async def test_disconnect_on_stop(hass: HomeAssistant, connect_legacy) -> None:
"""Test we close the connection with the router when Home Assistants stops."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA_TELNET,
unique_id=ROUTER_MAC_ADDR,
)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
assert connect_legacy.return_value.connection.disconnect.call_count == 1
assert config_entry.state is ConfigEntryState.LOADED

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

@ -1,16 +1,12 @@
"""Tests for the AsusWrt sensor."""
from datetime import timedelta
from unittest.mock import AsyncMock, Mock, patch
from aioasuswrt.asuswrt import Device
import pytest
from homeassistant.components import device_tracker, sensor
from homeassistant.components.asuswrt.const import (
CONF_INTERFACE,
DOMAIN,
MODE_ROUTER,
PROTOCOL_TELNET,
SENSORS_BYTES,
SENSORS_LOAD_AVG,
SENSORS_RATES,
@ -18,76 +14,19 @@ from homeassistant.components.asuswrt.const import (
)
from homeassistant.components.device_tracker import CONF_CONSIDER_HOME
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
CONF_HOST,
CONF_MODE,
CONF_PASSWORD,
CONF_PORT,
CONF_PROTOCOL,
CONF_USERNAME,
STATE_HOME,
STATE_NOT_HOME,
STATE_UNAVAILABLE,
)
from homeassistant.const import STATE_HOME, STATE_NOT_HOME, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.util import slugify
from homeassistant.util.dt import utcnow
from .common import CONFIG_DATA_TELNET, HOST, MOCK_MACS, ROUTER_MAC_ADDR, new_device
from tests.common import MockConfigEntry, async_fire_time_changed
ASUSWRT_LIB = "homeassistant.components.asuswrt.bridge.AsusWrtLegacy"
HOST = "myrouter.asuswrt.com"
IP_ADDRESS = "192.168.1.1"
CONFIG_DATA = {
CONF_HOST: HOST,
CONF_PORT: 22,
CONF_PROTOCOL: PROTOCOL_TELNET,
CONF_USERNAME: "user",
CONF_PASSWORD: "pwd",
CONF_MODE: MODE_ROUTER,
}
MAC_ADDR = "a1:b2:c3:d4:e5:f6"
MOCK_BYTES_TOTAL = [60000000000, 50000000000]
MOCK_CURRENT_TRANSFER_RATES = [20000000, 10000000]
MOCK_LOAD_AVG = [1.1, 1.2, 1.3]
MOCK_TEMPERATURES = {"2.4GHz": 40.0, "5.0GHz": 0.0, "CPU": 71.2}
MOCK_MAC_1 = "A1:B1:C1:D1:E1:F1"
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"
SENSORS_DEFAULT = [*SENSORS_BYTES, *SENSORS_RATES]
SENSORS_ALL = [*SENSORS_DEFAULT, *SENSORS_LOAD_AVG, *SENSORS_TEMPERATURES]
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():
"""Mock a list of devices."""
return {
MOCK_MAC_1: Device(MOCK_MAC_1, "192.168.1.2", "Test"),
MOCK_MAC_2: Device(MOCK_MAC_2, "192.168.1.3", "TestTwo"),
}
@pytest.fixture(name="mock_available_temps")
def mock_available_temps_fixture():
"""Mock a list of available temperature sensors."""
return [True, False, True]
SENSORS_ALL_LEGACY = [*SENSORS_DEFAULT, *SENSORS_LOAD_AVG, *SENSORS_TEMPERATURES]
@pytest.fixture(name="create_device_registry_devices")
@ -97,12 +36,7 @@ def create_device_registry_devices_fixture(hass: HomeAssistant):
config_entry = MockConfigEntry(domain="something_else")
config_entry.add_to_hass(hass)
for idx, device in enumerate(
(
MOCK_MAC_3,
MOCK_MAC_4,
)
):
for idx, device in enumerate((MOCK_MACS[2], MOCK_MACS[3])):
dev_reg.async_get_or_create(
name=f"Device {idx}",
config_entry_id=config_entry.entry_id,
@ -110,65 +44,6 @@ def create_device_registry_devices_fixture(hass: HomeAssistant):
)
@pytest.fixture(name="connect")
def mock_controller_connect(mock_devices, mock_available_temps):
"""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",
}
)
service_mock.return_value.async_get_connected_devices = AsyncMock(
return_value=mock_devices
)
service_mock.return_value.async_get_bytes_total = AsyncMock(
return_value=MOCK_BYTES_TOTAL
)
service_mock.return_value.async_get_current_transfer_rates = AsyncMock(
return_value=MOCK_CURRENT_TRANSFER_RATES
)
service_mock.return_value.async_get_loadavg = AsyncMock(
return_value=MOCK_LOAD_AVG
)
service_mock.return_value.async_get_temperature = AsyncMock(
return_value=MOCK_TEMPERATURES
)
service_mock.return_value.async_find_temperature_commands = AsyncMock(
return_value=mock_available_temps
)
yield service_mock
@pytest.fixture(name="connect_sens_fail")
def mock_controller_connect_sens_fail():
"""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()
service_mock.return_value.async_get_nvram = AsyncMock(side_effect=OSError)
service_mock.return_value.async_get_connected_devices = AsyncMock(
side_effect=OSError
)
service_mock.return_value.async_get_bytes_total = AsyncMock(side_effect=OSError)
service_mock.return_value.async_get_current_transfer_rates = AsyncMock(
side_effect=OSError
)
service_mock.return_value.async_get_loadavg = AsyncMock(side_effect=OSError)
service_mock.return_value.async_get_temperature = AsyncMock(side_effect=OSError)
service_mock.return_value.async_find_temperature_commands = AsyncMock(
return_value=[True, True, True]
)
yield service_mock
def _setup_entry(hass: HomeAssistant, config, sensors, unique_id=None):
"""Create mock config entry with enabled sensors."""
entity_reg = er.async_get(hass)
@ -201,28 +76,23 @@ def _setup_entry(hass: HomeAssistant, config, sensors, unique_id=None):
return config_entry, sensor_prefix
@pytest.mark.parametrize(
"entry_unique_id",
[None, MAC_ADDR],
)
async def test_sensors(
async def _test_sensors(
hass: HomeAssistant,
connect,
mock_devices,
create_device_registry_devices,
config,
entry_unique_id,
) -> None:
"""Test creating AsusWRT default sensors and tracker."""
config_entry, sensor_prefix = _setup_entry(
hass, CONFIG_DATA, SENSORS_DEFAULT, entry_unique_id
hass, config, 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",
MOCK_MACS[0]: "test",
dr.format_mac(MOCK_MACS[1]): "testtwo",
MOCK_MACS[1]: "testremove",
}.items():
entity_reg.async_get_or_create(
device_tracker.DOMAIN,
@ -250,7 +120,7 @@ async def test_sensors(
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "2"
# remove first tracked device
mock_devices.pop(MOCK_MAC_1)
mock_devices.pop(MOCK_MACS[0])
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
await hass.async_block_till_done()
@ -261,8 +131,8 @@ async def test_sensors(
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "1"
# 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)
mock_devices[MOCK_MACS[2]] = new_device(MOCK_MACS[2], "192.168.1.4", "TestThree")
mock_devices[MOCK_MACS[3]] = new_device(MOCK_MACS[3], "192.168.1.5", None)
# change consider home settings to have status not home of removed tracked device
hass.config_entries.async_update_entry(
@ -279,12 +149,26 @@ async def test_sensors(
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "3"
async def test_loadavg_sensors(
@pytest.mark.parametrize(
"entry_unique_id",
[None, ROUTER_MAC_ADDR],
)
async def test_sensors(
hass: HomeAssistant,
connect,
connect_legacy,
mock_devices_legacy,
create_device_registry_devices,
entry_unique_id,
) -> None:
"""Test creating AsusWRT default sensors and tracker with legacy protocol."""
await _test_sensors(hass, mock_devices_legacy, CONFIG_DATA_TELNET, entry_unique_id)
async def test_loadavg_sensors(hass: HomeAssistant, connect_legacy) -> None:
"""Test creating an AsusWRT load average sensors."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA, SENSORS_LOAD_AVG)
config_entry, sensor_prefix = _setup_entry(
hass, CONFIG_DATA_TELNET, SENSORS_LOAD_AVG
)
config_entry.add_to_hass(hass)
# initial devices setup
@ -299,12 +183,11 @@ async def test_loadavg_sensors(
assert hass.states.get(f"{sensor_prefix}_sensor_load_avg15").state == "1.3"
async def test_temperature_sensors(
hass: HomeAssistant,
connect,
) -> None:
async def test_temperature_sensors(hass: HomeAssistant, connect_legacy) -> None:
"""Test creating a AsusWRT temperature sensors."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA, SENSORS_TEMPERATURES)
config_entry, sensor_prefix = _setup_entry(
hass, CONFIG_DATA_TELNET, SENSORS_TEMPERATURES
)
config_entry.add_to_hass(hass)
# initial devices setup
@ -314,7 +197,7 @@ async def test_temperature_sensors(
await hass.async_block_till_done()
# assert temperature sensor available
assert hass.states.get(f"{sensor_prefix}_2_4ghz").state == "40.0"
assert hass.states.get(f"{sensor_prefix}_2_4ghz").state == "40.2"
assert not hass.states.get(f"{sensor_prefix}_5_0ghz")
assert hass.states.get(f"{sensor_prefix}_cpu").state == "71.2"
@ -323,32 +206,32 @@ async def test_temperature_sensors(
"side_effect",
[OSError, None],
)
async def test_connect_fail(hass: HomeAssistant, side_effect) -> None:
async def test_connect_fail(hass: HomeAssistant, connect_legacy, side_effect) -> None:
"""Test AsusWRT connect fail."""
# init config entry
config_entry = MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA,
data=CONFIG_DATA_TELNET,
)
config_entry.add_to_hass(hass)
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
connect_legacy.return_value.connection.async_connect.side_effect = side_effect
connect_legacy.return_value.is_connected = False
# initial setup fail
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
# initial setup fail
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_sensors_polling_fails(hass: HomeAssistant, connect_sens_fail) -> None:
async def test_sensors_polling_fails(
hass: HomeAssistant, connect_legacy_sens_fail
) -> None:
"""Test AsusWRT sensors are unavailable when polling fails."""
config_entry, sensor_prefix = _setup_entry(hass, CONFIG_DATA, SENSORS_ALL)
config_entry, sensor_prefix = _setup_entry(
hass, CONFIG_DATA_TELNET, SENSORS_ALL_LEGACY
)
config_entry.add_to_hass(hass)
# initial devices setup
@ -357,7 +240,7 @@ async def test_sensors_polling_fails(hass: HomeAssistant, connect_sens_fail) ->
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
await hass.async_block_till_done()
for sensor_name in SENSORS_ALL:
for sensor_name in SENSORS_ALL_LEGACY:
assert (
hass.states.get(f"{sensor_prefix}_{slugify(sensor_name)}").state
== STATE_UNAVAILABLE
@ -365,33 +248,38 @@ async def test_sensors_polling_fails(hass: HomeAssistant, connect_sens_fail) ->
assert hass.states.get(f"{sensor_prefix}_devices_connected").state == "0"
async def test_options_reload(hass: HomeAssistant, connect) -> None:
async def test_options_reload(hass: HomeAssistant, connect_legacy) -> None:
"""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 = MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA_TELNET,
unique_id=ROUTER_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()
assert connect_legacy.return_value.connection.async_connect.call_count == 1
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()
# 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
assert config_entry.state is ConfigEntryState.LOADED
assert connect_legacy.return_value.connection.async_connect.call_count == 2
async def test_unique_id_migration(hass: HomeAssistant, connect) -> None:
async def test_unique_id_migration(hass: HomeAssistant, connect_legacy) -> None:
"""Test AsusWRT entities unique id format migration."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA,
unique_id=MAC_ADDR,
data=CONFIG_DATA_TELNET,
unique_id=ROUTER_MAC_ADDR,
)
config_entry.add_to_hass(hass)
@ -400,7 +288,7 @@ async def test_unique_id_migration(hass: HomeAssistant, connect) -> None:
entity_reg.async_get_or_create(
sensor.DOMAIN,
DOMAIN,
f"{DOMAIN} {MAC_ADDR} Upload",
f"{DOMAIN} {ROUTER_MAC_ADDR} Upload",
suggested_object_id=obj_entity_id,
config_entry=config_entry,
disabled_by=None,
@ -411,4 +299,4 @@ async def test_unique_id_migration(hass: HomeAssistant, connect) -> None:
migr_entity = entity_reg.async_get(f"{sensor.DOMAIN}.{obj_entity_id}")
assert migr_entity is not None
assert migr_entity.unique_id == slugify(f"{MAC_ADDR}_sensor_tx_bytes")
assert migr_entity.unique_id == slugify(f"{ROUTER_MAC_ADDR}_sensor_tx_bytes")