Rework Mikrotik device scanning following Unifi (#27484)
* rework device scanning, add tests * update requirements and coverage * fix description comments * update tests, fix disabled entity updates * rework device scanning, add tests * update requirements and coverage * fix description comments * update tests, fix disabled entity updates * update librouteros to 3.0.0 * fix sorting * fix sorting 2 * revert to 2.3.0 as 3.0.0 requires code update * fix requirements * apply fixes * fix tests * update hub.py and fix tests * fix test_hub_setup_failed * rebased on dev and update librouteros to 3.0.0 * fixed test_config_flow * fixed tests * fix test_config_flow
This commit is contained in:
1 parent
4f79ec0c78
commit
04cb2e9fd5
18 files changed
+1546
-341
No files matched your search
@@ -0,0 +1,133 @@
|
||||
"""Tests for the Mikrotik component."""
|
||||
from homeassistant.components import mikrotik
|
||||
|
||||
MOCK_DATA = {
|
||||
mikrotik.CONF_NAME: "Mikrotik",
|
||||
mikrotik.CONF_HOST: "0.0.0.0",
|
||||
mikrotik.CONF_USERNAME: "user",
|
||||
mikrotik.CONF_PASSWORD: "pass",
|
||||
mikrotik.CONF_PORT: 8278,
|
||||
mikrotik.CONF_VERIFY_SSL: False,
|
||||
}
|
||||
|
||||
MOCK_OPTIONS = {
|
||||
mikrotik.CONF_ARP_PING: False,
|
||||
mikrotik.const.CONF_FORCE_DHCP: False,
|
||||
mikrotik.CONF_DETECTION_TIME: mikrotik.DEFAULT_DETECTION_TIME,
|
||||
}
|
||||
|
||||
DEVICE_1_DHCP = {
|
||||
".id": "*1A",
|
||||
"address": "0.0.0.1",
|
||||
"mac-address": "00:00:00:00:00:01",
|
||||
"active-address": "0.0.0.1",
|
||||
"host-name": "Device_1",
|
||||
"comment": "Mobile",
|
||||
}
|
||||
DEVICE_2_DHCP = {
|
||||
".id": "*1B",
|
||||
"address": "0.0.0.2",
|
||||
"mac-address": "00:00:00:00:00:02",
|
||||
"active-address": "0.0.0.2",
|
||||
"host-name": "Device_2",
|
||||
"comment": "PC",
|
||||
}
|
||||
DEVICE_1_WIRELESS = {
|
||||
".id": "*264",
|
||||
"interface": "wlan1",
|
||||
"mac-address": "00:00:00:00:00:01",
|
||||
"ap": False,
|
||||
"wds": False,
|
||||
"bridge": False,
|
||||
"rx-rate": "72.2Mbps-20MHz/1S/SGI",
|
||||
"tx-rate": "72.2Mbps-20MHz/1S/SGI",
|
||||
"packets": "59542,17464",
|
||||
"bytes": "17536671,2966351",
|
||||
"frames": "59542,17472",
|
||||
"frame-bytes": "17655785,2862445",
|
||||
"hw-frames": "78935,38395",
|
||||
"hw-frame-bytes": "25636019,4063445",
|
||||
"tx-frames-timed-out": 0,
|
||||
"uptime": "5h49m36s",
|
||||
"last-activity": "170ms",
|
||||
"signal-strength": "-62@1Mbps",
|
||||
"signal-to-noise": 52,
|
||||
"signal-strength-ch0": -63,
|
||||
"signal-strength-ch1": -69,
|
||||
"strength-at-rates": "-62@1Mbps 16s330ms,-64@6Mbps 13s560ms,-65@HT20-3 52m6s30ms,-66@HT20-4 52m4s350ms,-66@HT20-5 51m58s580ms,-65@HT20-6 51m24s780ms,-65@HT20-7 5s680ms",
|
||||
"tx-ccq": 93,
|
||||
"p-throughput": 54928,
|
||||
"last-ip": "0.0.0.1",
|
||||
"802.1x-port-enabled": True,
|
||||
"authentication-type": "wpa2-psk",
|
||||
"encryption": "aes-ccm",
|
||||
"group-encryption": "aes-ccm",
|
||||
"management-protection": False,
|
||||
"wmm-enabled": True,
|
||||
"tx-rate-set": "OFDM:6-54 BW:1x SGI:1x HT:0-7",
|
||||
}
|
||||
|
||||
DEVICE_2_WIRELESS = {
|
||||
".id": "*265",
|
||||
"interface": "wlan1",
|
||||
"mac-address": "00:00:00:00:00:02",
|
||||
"ap": False,
|
||||
"wds": False,
|
||||
"bridge": False,
|
||||
"rx-rate": "72.2Mbps-20MHz/1S/SGI",
|
||||
"tx-rate": "72.2Mbps-20MHz/1S/SGI",
|
||||
"packets": "59542,17464",
|
||||
"bytes": "17536671,2966351",
|
||||
"frames": "59542,17472",
|
||||
"frame-bytes": "17655785,2862445",
|
||||
"hw-frames": "78935,38395",
|
||||
"hw-frame-bytes": "25636019,4063445",
|
||||
"tx-frames-timed-out": 0,
|
||||
"uptime": "5h49m36s",
|
||||
"last-activity": "170ms",
|
||||
"signal-strength": "-62@1Mbps",
|
||||
"signal-to-noise": 52,
|
||||
"signal-strength-ch0": -63,
|
||||
"signal-strength-ch1": -69,
|
||||
"strength-at-rates": "-62@1Mbps 16s330ms,-64@6Mbps 13s560ms,-65@HT20-3 52m6s30ms,-66@HT20-4 52m4s350ms,-66@HT20-5 51m58s580ms,-65@HT20-6 51m24s780ms,-65@HT20-7 5s680ms",
|
||||
"tx-ccq": 93,
|
||||
"p-throughput": 54928,
|
||||
"last-ip": "0.0.0.2",
|
||||
"802.1x-port-enabled": True,
|
||||
"authentication-type": "wpa2-psk",
|
||||
"encryption": "aes-ccm",
|
||||
"group-encryption": "aes-ccm",
|
||||
"management-protection": False,
|
||||
"wmm-enabled": True,
|
||||
"tx-rate-set": "OFDM:6-54 BW:1x SGI:1x HT:0-7",
|
||||
}
|
||||
DHCP_DATA = [DEVICE_1_DHCP, DEVICE_2_DHCP]
|
||||
|
||||
WIRELESS_DATA = [DEVICE_1_WIRELESS]
|
||||
|
||||
ARP_DATA = [
|
||||
{
|
||||
".id": "*1",
|
||||
"address": "0.0.0.1",
|
||||
"mac-address": "00:00:00:00:00:01",
|
||||
"interface": "bridge",
|
||||
"published": False,
|
||||
"invalid": False,
|
||||
"DHCP": True,
|
||||
"dynamic": True,
|
||||
"complete": True,
|
||||
"disabled": False,
|
||||
},
|
||||
{
|
||||
".id": "*2",
|
||||
"address": "0.0.0.2",
|
||||
"mac-address": "00:00:00:00:00:02",
|
||||
"interface": "bridge",
|
||||
"published": False,
|
||||
"invalid": False,
|
||||
"DHCP": True,
|
||||
"dynamic": True,
|
||||
"complete": True,
|
||||
"disabled": False,
|
||||
},
|
||||
]
|
||||
@@ -0,0 +1,208 @@
|
||||
"""Test Mikrotik setup process."""
|
||||
from datetime import timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
import librouteros
|
||||
import pytest
|
||||
|
||||
from homeassistant import data_entry_flow
|
||||
from homeassistant.components import mikrotik
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_USERNAME,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
DEMO_USER_INPUT = {
|
||||
CONF_NAME: "Home router",
|
||||
CONF_HOST: "0.0.0.0",
|
||||
CONF_USERNAME: "username",
|
||||
CONF_PASSWORD: "password",
|
||||
CONF_PORT: 8278,
|
||||
CONF_VERIFY_SSL: False,
|
||||
}
|
||||
|
||||
DEMO_CONFIG = {
|
||||
CONF_NAME: "Home router",
|
||||
CONF_HOST: "0.0.0.0",
|
||||
CONF_USERNAME: "username",
|
||||
CONF_PASSWORD: "password",
|
||||
CONF_PORT: 8278,
|
||||
CONF_VERIFY_SSL: False,
|
||||
mikrotik.const.CONF_FORCE_DHCP: False,
|
||||
mikrotik.CONF_ARP_PING: False,
|
||||
mikrotik.CONF_DETECTION_TIME: timedelta(seconds=30),
|
||||
}
|
||||
|
||||
DEMO_CONFIG_ENTRY = {
|
||||
CONF_NAME: "Home router",
|
||||
CONF_HOST: "0.0.0.0",
|
||||
CONF_USERNAME: "username",
|
||||
CONF_PASSWORD: "password",
|
||||
CONF_PORT: 8278,
|
||||
CONF_VERIFY_SSL: False,
|
||||
mikrotik.const.CONF_FORCE_DHCP: False,
|
||||
mikrotik.CONF_ARP_PING: False,
|
||||
mikrotik.CONF_DETECTION_TIME: 30,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(name="api")
|
||||
def mock_mikrotik_api():
|
||||
"""Mock an api."""
|
||||
with patch("librouteros.connect"):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture(name="auth_error")
|
||||
def mock_api_authentication_error():
|
||||
"""Mock an api."""
|
||||
with patch(
|
||||
"librouteros.connect",
|
||||
side_effect=librouteros.exceptions.TrapError("invalid user name or password"),
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture(name="conn_error")
|
||||
def mock_api_connection_error():
|
||||
"""Mock an api."""
|
||||
with patch(
|
||||
"librouteros.connect", side_effect=librouteros.exceptions.ConnectionClosed
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
async def test_import(hass, api):
|
||||
"""Test import step."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
mikrotik.DOMAIN, context={"source": "import"}, data=DEMO_CONFIG
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["title"] == "Home router"
|
||||
assert result["data"][CONF_NAME] == "Home router"
|
||||
assert result["data"][CONF_HOST] == "0.0.0.0"
|
||||
assert result["data"][CONF_USERNAME] == "username"
|
||||
assert result["data"][CONF_PASSWORD] == "password"
|
||||
assert result["data"][CONF_PORT] == 8278
|
||||
assert result["data"][CONF_VERIFY_SSL] is False
|
||||
|
||||
|
||||
async def test_flow_works(hass, api):
|
||||
"""Test config flow."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
mikrotik.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=DEMO_USER_INPUT
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["title"] == "Home router"
|
||||
assert result["data"][CONF_NAME] == "Home router"
|
||||
assert result["data"][CONF_HOST] == "0.0.0.0"
|
||||
assert result["data"][CONF_USERNAME] == "username"
|
||||
assert result["data"][CONF_PASSWORD] == "password"
|
||||
assert result["data"][CONF_PORT] == 8278
|
||||
|
||||
|
||||
async def test_options(hass):
|
||||
"""Test updating options."""
|
||||
entry = MockConfigEntry(domain=mikrotik.DOMAIN, data=DEMO_CONFIG_ENTRY)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "device_tracker"
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
mikrotik.CONF_DETECTION_TIME: 30,
|
||||
mikrotik.CONF_ARP_PING: True,
|
||||
mikrotik.const.CONF_FORCE_DHCP: False,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"] == {
|
||||
mikrotik.CONF_DETECTION_TIME: 30,
|
||||
mikrotik.CONF_ARP_PING: True,
|
||||
mikrotik.const.CONF_FORCE_DHCP: False,
|
||||
}
|
||||
|
||||
|
||||
async def test_host_already_configured(hass, auth_error):
|
||||
"""Test host already configured."""
|
||||
|
||||
entry = MockConfigEntry(domain=mikrotik.DOMAIN, data=DEMO_CONFIG_ENTRY)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
mikrotik.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=DEMO_USER_INPUT
|
||||
)
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_name_exists(hass, api):
|
||||
"""Test name already configured."""
|
||||
|
||||
entry = MockConfigEntry(domain=mikrotik.DOMAIN, data=DEMO_CONFIG_ENTRY)
|
||||
entry.add_to_hass(hass)
|
||||
user_input = DEMO_USER_INPUT.copy()
|
||||
user_input[CONF_HOST] = "0.0.0.1"
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
mikrotik.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=user_input
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"] == {CONF_NAME: "name_exists"}
|
||||
|
||||
|
||||
async def test_connection_error(hass, conn_error):
|
||||
"""Test error when connection is unsuccesful."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
mikrotik.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=DEMO_USER_INPUT
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_wrong_credentials(hass, auth_error):
|
||||
"""Test error when credentials are wrong."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
mikrotik.DOMAIN, context={"source": "user"}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=DEMO_USER_INPUT
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {
|
||||
CONF_USERNAME: "wrong_credentials",
|
||||
CONF_PASSWORD: "wrong_credentials",
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
"""The tests for the Mikrotik device tracker platform."""
|
||||
from datetime import timedelta
|
||||
|
||||
from homeassistant.components import mikrotik
|
||||
import homeassistant.components.device_tracker as device_tracker
|
||||
from homeassistant.helpers import entity_registry
|
||||
from homeassistant.setup import async_setup_component
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
from . import DEVICE_2_WIRELESS, DHCP_DATA, MOCK_DATA, MOCK_OPTIONS, WIRELESS_DATA
|
||||
from .test_hub import setup_mikrotik_entry
|
||||
|
||||
from tests.common import MockConfigEntry, patch
|
||||
|
||||
DEFAULT_DETECTION_TIME = timedelta(seconds=300)
|
||||
|
||||
|
||||
def mock_command(self, cmd, params=None):
|
||||
"""Mock the Mikrotik command method."""
|
||||
if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.IS_WIRELESS]:
|
||||
return True
|
||||
if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.DHCP]:
|
||||
return DHCP_DATA
|
||||
if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.WIRELESS]:
|
||||
return WIRELESS_DATA
|
||||
return {}
|
||||
|
||||
|
||||
async def test_platform_manually_configured(hass):
|
||||
"""Test that nothing happens when configuring mikrotik through device tracker platform."""
|
||||
assert (
|
||||
await async_setup_component(
|
||||
hass,
|
||||
device_tracker.DOMAIN,
|
||||
{device_tracker.DOMAIN: {"platform": "mikrotik"}},
|
||||
)
|
||||
is False
|
||||
)
|
||||
assert mikrotik.DOMAIN not in hass.data
|
||||
|
||||
|
||||
async def test_device_trackers(hass):
|
||||
"""Test device_trackers created by mikrotik."""
|
||||
|
||||
# test devices are added from wireless list only
|
||||
hub = await setup_mikrotik_entry(hass)
|
||||
|
||||
device_1 = hass.states.get("device_tracker.device_1")
|
||||
assert device_1 is not None
|
||||
assert device_1.state == "home"
|
||||
device_2 = hass.states.get("device_tracker.device_2")
|
||||
assert device_2 is None
|
||||
|
||||
with patch.object(mikrotik.hub.MikrotikData, "command", new=mock_command):
|
||||
# test device_2 is added after connecting to wireless network
|
||||
WIRELESS_DATA.append(DEVICE_2_WIRELESS)
|
||||
|
||||
await hub.async_update()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
device_2 = hass.states.get("device_tracker.device_2")
|
||||
assert device_2 is not None
|
||||
assert device_2.state == "home"
|
||||
|
||||
# test state remains home if last_seen consider_home_interval
|
||||
del WIRELESS_DATA[1] # device 2 is removed from wireless list
|
||||
hub.api.devices["00:00:00:00:00:02"]._last_seen = dt_util.utcnow() - timedelta(
|
||||
minutes=4
|
||||
)
|
||||
await hub.async_update()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
device_2 = hass.states.get("device_tracker.device_2")
|
||||
assert device_2.state != "not_home"
|
||||
|
||||
# test state changes to away if last_seen > consider_home_interval
|
||||
hub.api.devices["00:00:00:00:00:02"]._last_seen = dt_util.utcnow() - timedelta(
|
||||
minutes=5
|
||||
)
|
||||
await hub.async_update()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
device_2 = hass.states.get("device_tracker.device_2")
|
||||
assert device_2.state == "not_home"
|
||||
|
||||
|
||||
async def test_restoring_devices(hass):
|
||||
"""Test restoring existing device_tracker entities if not detected on startup."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=mikrotik.DOMAIN, data=MOCK_DATA, options=MOCK_OPTIONS
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
registry = await entity_registry.async_get_registry(hass)
|
||||
registry.async_get_or_create(
|
||||
device_tracker.DOMAIN,
|
||||
mikrotik.DOMAIN,
|
||||
"00:00:00:00:00:01",
|
||||
suggested_object_id="device_1",
|
||||
config_entry=config_entry,
|
||||
)
|
||||
registry.async_get_or_create(
|
||||
device_tracker.DOMAIN,
|
||||
mikrotik.DOMAIN,
|
||||
"00:00:00:00:00:02",
|
||||
suggested_object_id="device_2",
|
||||
config_entry=config_entry,
|
||||
)
|
||||
|
||||
await setup_mikrotik_entry(hass)
|
||||
|
||||
# test device_2 which is not in wireless list is restored
|
||||
device_1 = hass.states.get("device_tracker.device_1")
|
||||
assert device_1 is not None
|
||||
assert device_1.state == "home"
|
||||
device_2 = hass.states.get("device_tracker.device_2")
|
||||
assert device_2 is not None
|
||||
assert device_2.state == "not_home"
|
||||
@@ -0,0 +1,179 @@
|
||||
"""Test Mikrotik hub."""
|
||||
from asynctest import patch
|
||||
import librouteros
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components import mikrotik
|
||||
|
||||
from . import ARP_DATA, DHCP_DATA, MOCK_DATA, MOCK_OPTIONS, WIRELESS_DATA
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def setup_mikrotik_entry(hass, **kwargs):
|
||||
"""Set up Mikrotik intergation successfully."""
|
||||
support_wireless = kwargs.get("support_wireless", True)
|
||||
dhcp_data = kwargs.get("dhcp_data", DHCP_DATA)
|
||||
wireless_data = kwargs.get("wireless_data", WIRELESS_DATA)
|
||||
|
||||
def mock_command(self, cmd, params=None):
|
||||
if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.IS_WIRELESS]:
|
||||
return support_wireless
|
||||
if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.DHCP]:
|
||||
return dhcp_data
|
||||
if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.WIRELESS]:
|
||||
return wireless_data
|
||||
if cmd == mikrotik.const.MIKROTIK_SERVICES[mikrotik.const.ARP]:
|
||||
return ARP_DATA
|
||||
return {}
|
||||
|
||||
config_entry = MockConfigEntry(
|
||||
domain=mikrotik.DOMAIN, data=MOCK_DATA, options=MOCK_OPTIONS
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
if "force_dhcp" in kwargs:
|
||||
config_entry.options["force_dhcp"] = True
|
||||
|
||||
if "arp_ping" in kwargs:
|
||||
config_entry.options["arp_ping"] = True
|
||||
|
||||
with patch("librouteros.connect"), patch.object(
|
||||
mikrotik.hub.MikrotikData, "command", new=mock_command
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
return hass.data[mikrotik.DOMAIN][config_entry.entry_id]
|
||||
|
||||
|
||||
async def test_hub_setup_successful(hass):
|
||||
"""Successful setup of Mikrotik hub."""
|
||||
with patch(
|
||||
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setup",
|
||||
return_value=True,
|
||||
) as forward_entry_setup:
|
||||
hub = await setup_mikrotik_entry(hass)
|
||||
|
||||
assert hub.config_entry.data == {
|
||||
mikrotik.CONF_NAME: "Mikrotik",
|
||||
mikrotik.CONF_HOST: "0.0.0.0",
|
||||
mikrotik.CONF_USERNAME: "user",
|
||||
mikrotik.CONF_PASSWORD: "pass",
|
||||
mikrotik.CONF_PORT: 8278,
|
||||
mikrotik.CONF_VERIFY_SSL: False,
|
||||
}
|
||||
assert hub.config_entry.options == {
|
||||
mikrotik.hub.CONF_FORCE_DHCP: False,
|
||||
mikrotik.CONF_ARP_PING: False,
|
||||
mikrotik.CONF_DETECTION_TIME: 300,
|
||||
}
|
||||
|
||||
assert hub.api.available is True
|
||||
assert hub.signal_update == "mikrotik-update-0.0.0.0"
|
||||
assert forward_entry_setup.mock_calls[0][1] == (hub.config_entry, "device_tracker")
|
||||
|
||||
|
||||
async def test_hub_setup_failed(hass):
|
||||
"""Failed setup of Mikrotik hub."""
|
||||
|
||||
config_entry = MockConfigEntry(domain=mikrotik.DOMAIN, data=MOCK_DATA)
|
||||
config_entry.add_to_hass(hass)
|
||||
# error when connection fails
|
||||
with patch(
|
||||
"librouteros.connect", side_effect=librouteros.exceptions.ConnectionClosed
|
||||
):
|
||||
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
|
||||
assert config_entry.state == config_entries.ENTRY_STATE_SETUP_RETRY
|
||||
|
||||
# error when username or password is invalid
|
||||
config_entry = MockConfigEntry(domain=mikrotik.DOMAIN, data=MOCK_DATA)
|
||||
config_entry.add_to_hass(hass)
|
||||
with patch(
|
||||
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setup"
|
||||
) as forward_entry_setup, patch(
|
||||
"librouteros.connect",
|
||||
side_effect=librouteros.exceptions.TrapError("invalid user name or password"),
|
||||
):
|
||||
|
||||
result = await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
|
||||
assert result is False
|
||||
assert len(forward_entry_setup.mock_calls) == 0
|
||||
|
||||
|
||||
async def test_update_failed(hass):
|
||||
"""Test failing to connect during update."""
|
||||
|
||||
hub = await setup_mikrotik_entry(hass)
|
||||
|
||||
with patch.object(
|
||||
mikrotik.hub.MikrotikData, "command", side_effect=mikrotik.errors.CannotConnect
|
||||
):
|
||||
await hub.async_update()
|
||||
|
||||
assert hub.api.available is False
|
||||
|
||||
|
||||
async def test_hub_not_support_wireless(hass):
|
||||
"""Test updating hub devices when hub doesn't support wireless interfaces."""
|
||||
|
||||
# test that the devices are constructed from dhcp data
|
||||
|
||||
hub = await setup_mikrotik_entry(hass, support_wireless=False)
|
||||
|
||||
assert hub.api.devices["00:00:00:00:00:01"]._params == DHCP_DATA[0]
|
||||
assert hub.api.devices["00:00:00:00:00:01"]._wireless_params is None
|
||||
assert hub.api.devices["00:00:00:00:00:02"]._params == DHCP_DATA[1]
|
||||
assert hub.api.devices["00:00:00:00:00:02"]._wireless_params is None
|
||||
|
||||
|
||||
async def test_hub_support_wireless(hass):
|
||||
"""Test updating hub devices when hub support wireless interfaces."""
|
||||
|
||||
# test that the device list is from wireless data list
|
||||
|
||||
hub = await setup_mikrotik_entry(hass)
|
||||
|
||||
assert hub.api.support_wireless is True
|
||||
assert hub.api.devices["00:00:00:00:00:01"]._params == DHCP_DATA[0]
|
||||
assert hub.api.devices["00:00:00:00:00:01"]._wireless_params == WIRELESS_DATA[0]
|
||||
|
||||
# devices not in wireless list will not be added
|
||||
assert "00:00:00:00:00:02" not in hub.api.devices
|
||||
|
||||
|
||||
async def test_force_dhcp(hass):
|
||||
"""Test updating hub devices with forced dhcp method."""
|
||||
|
||||
# test that the devices are constructed from dhcp data
|
||||
|
||||
hub = await setup_mikrotik_entry(hass, force_dhcp=True)
|
||||
|
||||
assert hub.api.support_wireless is True
|
||||
assert hub.api.devices["00:00:00:00:00:01"]._params == DHCP_DATA[0]
|
||||
assert hub.api.devices["00:00:00:00:00:01"]._wireless_params == WIRELESS_DATA[0]
|
||||
|
||||
# devices not in wireless list are added from dhcp
|
||||
assert hub.api.devices["00:00:00:00:00:02"]._params == DHCP_DATA[1]
|
||||
assert hub.api.devices["00:00:00:00:00:02"]._wireless_params is None
|
||||
|
||||
|
||||
async def test_arp_ping(hass):
|
||||
"""Test arp ping devices to confirm they are connected."""
|
||||
|
||||
# test device show as home if arp ping returns value
|
||||
with patch.object(mikrotik.hub.MikrotikData, "do_arp_ping", return_value=True):
|
||||
hub = await setup_mikrotik_entry(hass, arp_ping=True, force_dhcp=True)
|
||||
|
||||
assert hub.api.devices["00:00:00:00:00:01"].last_seen is not None
|
||||
assert hub.api.devices["00:00:00:00:00:02"].last_seen is not None
|
||||
|
||||
# test device show as away if arp ping times out
|
||||
with patch.object(mikrotik.hub.MikrotikData, "do_arp_ping", return_value=False):
|
||||
hub = await setup_mikrotik_entry(hass, arp_ping=True, force_dhcp=True)
|
||||
|
||||
assert hub.api.devices["00:00:00:00:00:01"].last_seen is not None
|
||||
# this device is not wireless so it will show as away
|
||||
assert hub.api.devices["00:00:00:00:00:02"].last_seen is None
|
||||
@@ -0,0 +1,83 @@
|
||||
"""Test Mikrotik setup process."""
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from homeassistant.components import mikrotik
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import MOCK_DATA
|
||||
|
||||
from tests.common import MockConfigEntry, mock_coro
|
||||
|
||||
|
||||
async def test_setup_with_no_config(hass):
|
||||
"""Test that we do not discover anything or try to set up a hub."""
|
||||
assert await async_setup_component(hass, mikrotik.DOMAIN, {}) is True
|
||||
assert mikrotik.DOMAIN not in hass.data
|
||||
|
||||
|
||||
async def test_successful_config_entry(hass):
|
||||
"""Test config entry successfull setup."""
|
||||
entry = MockConfigEntry(domain=mikrotik.DOMAIN, data=MOCK_DATA,)
|
||||
entry.add_to_hass(hass)
|
||||
mock_registry = Mock()
|
||||
|
||||
with patch.object(mikrotik, "MikrotikHub") as mock_hub, patch(
|
||||
"homeassistant.helpers.device_registry.async_get_registry",
|
||||
return_value=mock_coro(mock_registry),
|
||||
):
|
||||
mock_hub.return_value.async_setup.return_value = mock_coro(True)
|
||||
mock_hub.return_value.serial_num = "12345678"
|
||||
mock_hub.return_value.model = "RB750"
|
||||
mock_hub.return_value.hostname = "mikrotik"
|
||||
mock_hub.return_value.firmware = "3.65"
|
||||
assert await mikrotik.async_setup_entry(hass, entry) is True
|
||||
|
||||
assert len(mock_hub.mock_calls) == 2
|
||||
p_hass, p_entry = mock_hub.mock_calls[0][1]
|
||||
|
||||
assert p_hass is hass
|
||||
assert p_entry is entry
|
||||
|
||||
assert len(mock_registry.mock_calls) == 1
|
||||
assert mock_registry.mock_calls[0][2] == {
|
||||
"config_entry_id": entry.entry_id,
|
||||
"connections": {("mikrotik", "12345678")},
|
||||
"manufacturer": mikrotik.ATTR_MANUFACTURER,
|
||||
"model": "RB750",
|
||||
"name": "mikrotik",
|
||||
"sw_version": "3.65",
|
||||
}
|
||||
|
||||
|
||||
async def test_hub_fail_setup(hass):
|
||||
"""Test that a failed setup will not store the hub."""
|
||||
entry = MockConfigEntry(domain=mikrotik.DOMAIN, data=MOCK_DATA,)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch.object(mikrotik, "MikrotikHub") as mock_hub:
|
||||
mock_hub.return_value.async_setup.return_value = mock_coro(False)
|
||||
assert await mikrotik.async_setup_entry(hass, entry) is False
|
||||
|
||||
assert mikrotik.DOMAIN not in hass.data
|
||||
|
||||
|
||||
async def test_unload_entry(hass):
|
||||
"""Test being able to unload an entry."""
|
||||
entry = MockConfigEntry(domain=mikrotik.DOMAIN, data=MOCK_DATA,)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch.object(mikrotik, "MikrotikHub") as mock_hub, patch(
|
||||
"homeassistant.helpers.device_registry.async_get_registry",
|
||||
return_value=mock_coro(Mock()),
|
||||
):
|
||||
mock_hub.return_value.async_setup.return_value = mock_coro(True)
|
||||
mock_hub.return_value.serial_num = "12345678"
|
||||
mock_hub.return_value.model = "RB750"
|
||||
mock_hub.return_value.hostname = "mikrotik"
|
||||
mock_hub.return_value.firmware = "3.65"
|
||||
assert await mikrotik.async_setup_entry(hass, entry) is True
|
||||
|
||||
assert len(mock_hub.return_value.mock_calls) == 1
|
||||
|
||||
assert await mikrotik.async_unload_entry(hass, entry)
|
||||
assert entry.entry_id not in hass.data[mikrotik.DOMAIN]
|
||||
Reference in new issue
Block a user