Modernize Huawei LTE (#26675)

* Modernization rework

- config entry support, with override support from huawei_lte platform in YAML
- device tracker entity registry support
- refactor for easier addition of more features
- internal code cleanups

* Remove log level dependent subscription/data debug hack

No longer needed, because pretty much all keys from supported
categories are exposed as sensors.

Closes https://github.com/home-assistant/home-assistant/issues/23819

* Upgrade huawei-lte-api to 1.4.1

https://github.com/Salamek/huawei-lte-api/releases

* Add support for access without username and password

* Use subclass init instead of config_entries.HANDLERS

* Update huawei-lte-api to 1.4.3 (#27269)

* Convert device state attributes to snake_case

* Simplify scanner entity initialization

* Remove not needed hass reference from Router

* Return explicit None from unsupported old device tracker setup

* Mark unknown connection errors during config as such

* Drop some dead config flow code

* Run config flow sync I/O in executor

* Parametrize config flow login error tests

* Forward entry unload to platforms

* Async/sync fixups

* Improve data subscription debug logging

* Implement on the fly add of new and tracking of seen device tracker entities

* Handle device tracker entry unload cleanup in component

* Remove unnecessary _async_setup_lte, just have code in async_setup_entry

* Remove time tracker on unload

* Fix to not use same mutable default subscription set for all routers

* Pylint fixes

* Remove some redundant defensive device tracker code

* Add back explicit get_scanner None return, hush pylint

* Adjust approach to set system_options on entry create

* Enable some sensors on first add instead of disabling everything

* Fix SMS notification recipients default value

* Add option to skip new device tracker entities

* Fix SMS notification recipient option default

* Work around https://github.com/PyCQA/pylint/issues/3202

* Remove unrelated type hint additions

* Change async_add_new_entities to a regular function

* Remove option to disable polling for new device tracker entries
This commit is contained in:
Ville Skyttä
2019-10-24 19:31:49 +03:00
committed by GitHub
parent 969322e14a
commit fc09702cc3
15 changed files with 1067 additions and 276 deletions
@@ -0,0 +1,140 @@
"""Tests for the Huawei LTE config flow."""
from huawei_lte_api.enums.client import ResponseCodeEnum
from huawei_lte_api.enums.user import LoginErrorEnum, LoginStateEnum, PasswordTypeEnum
from requests_mock import ANY
from requests.exceptions import ConnectionError
import pytest
from homeassistant import data_entry_flow
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD, CONF_URL
from homeassistant.components.huawei_lte.const import DOMAIN
from homeassistant.components.huawei_lte.config_flow import ConfigFlowHandler
from tests.common import MockConfigEntry
FIXTURE_USER_INPUT = {
CONF_URL: "http://192.168.1.1/",
CONF_USERNAME: "admin",
CONF_PASSWORD: "secret",
}
async def test_show_set_form(hass):
"""Test that the setup form is served."""
flow = ConfigFlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=None)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
async def test_urlize_plain_host(hass, requests_mock):
"""Test that plain host or IP gets converted to a URL."""
requests_mock.request(ANY, ANY, exc=ConnectionError())
flow = ConfigFlowHandler()
flow.hass = hass
host = "192.168.100.1"
user_input = {**FIXTURE_USER_INPUT, CONF_URL: host}
result = await flow.async_step_user(user_input=user_input)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert user_input[CONF_URL] == f"http://{host}/"
async def test_already_configured(hass):
"""Test we reject already configured devices."""
MockConfigEntry(
domain=DOMAIN, data=FIXTURE_USER_INPUT, title="Already configured"
).add_to_hass(hass)
flow = ConfigFlowHandler()
flow.hass = hass
# Tweak URL a bit to check that doesn't fail duplicate detection
result = await flow.async_step_user(
user_input={
**FIXTURE_USER_INPUT,
CONF_URL: FIXTURE_USER_INPUT[CONF_URL].replace("http", "HTTP"),
}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_connection_error(hass, requests_mock):
"""Test we show user form on connection error."""
requests_mock.request(ANY, ANY, exc=ConnectionError())
flow = ConfigFlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=FIXTURE_USER_INPUT)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {CONF_URL: "unknown_connection_error"}
@pytest.fixture
def login_requests_mock(requests_mock):
"""Set up a requests_mock with base mocks for login tests."""
requests_mock.request(
ANY, FIXTURE_USER_INPUT[CONF_URL], text='<meta name="csrf_token" content="x"/>'
)
requests_mock.request(
ANY,
f"{FIXTURE_USER_INPUT[CONF_URL]}api/user/state-login",
text=(
f"<response><State>{LoginStateEnum.LOGGED_OUT}</State>"
f"<password_type>{PasswordTypeEnum.SHA256}</password_type></response>"
),
)
return requests_mock
@pytest.mark.parametrize(
("code", "errors"),
(
(LoginErrorEnum.USERNAME_WRONG, {CONF_USERNAME: "incorrect_username"}),
(LoginErrorEnum.PASSWORD_WRONG, {CONF_PASSWORD: "incorrect_password"}),
(
LoginErrorEnum.USERNAME_PWD_WRONG,
{CONF_USERNAME: "incorrect_username_or_password"},
),
(LoginErrorEnum.USERNAME_PWD_ORERRUN, {"base": "login_attempts_exceeded"}),
(ResponseCodeEnum.ERROR_SYSTEM_UNKNOWN, {"base": "response_error"}),
),
)
async def test_login_error(hass, login_requests_mock, code, errors):
"""Test we show user form with appropriate error on response failure."""
login_requests_mock.request(
ANY,
f"{FIXTURE_USER_INPUT[CONF_URL]}api/user/login",
text=f"<error><code>{code}</code><message/></error>",
)
flow = ConfigFlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=FIXTURE_USER_INPUT)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == errors
async def test_success(hass, login_requests_mock):
"""Test successful flow provides entry creation data."""
login_requests_mock.request(
ANY,
f"{FIXTURE_USER_INPUT[CONF_URL]}api/user/login",
text=f"<response>OK</response>",
)
flow = ConfigFlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=FIXTURE_USER_INPUT)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"][CONF_URL] == FIXTURE_USER_INPUT[CONF_URL]
assert result["data"][CONF_USERNAME] == FIXTURE_USER_INPUT[CONF_USERNAME]
assert result["data"][CONF_PASSWORD] == FIXTURE_USER_INPUT[CONF_PASSWORD]
@@ -0,0 +1,20 @@
"""Huawei LTE device tracker tests."""
import pytest
from homeassistant.components.huawei_lte import device_tracker
@pytest.mark.parametrize(
("value", "expected"),
(
("HTTP", "http"),
("ID", "id"),
("IPAddress", "ip_address"),
("HTTPResponse", "http_response"),
("foo_bar", "foo_bar"),
),
)
def test_better_snakecase(value, expected):
"""Test that better snakecase works better."""
assert device_tracker._better_snakecase(value) == expected
-48
View File
@@ -1,48 +0,0 @@
"""Huawei LTE component tests."""
from unittest.mock import Mock
import pytest
from homeassistant.components import huawei_lte
from homeassistant.components.huawei_lte.const import KEY_DEVICE_INFORMATION
@pytest.fixture(autouse=True)
def routerdata():
"""Set up a router data for testing."""
rd = huawei_lte.RouterData(Mock(), "de:ad:be:ef:00:00")
rd.device_information = {"SoftwareVersion": "1.0", "nested": {"foo": "bar"}}
return rd
async def test_routerdata_get_nonexistent_root(routerdata):
"""Test that accessing a nonexistent root element raises KeyError."""
with pytest.raises(KeyError): # NOT AttributeError
routerdata["nonexistent_root.foo"]
async def test_routerdata_get_nonexistent_leaf(routerdata):
"""Test that accessing a nonexistent leaf element raises KeyError."""
with pytest.raises(KeyError):
routerdata[f"{KEY_DEVICE_INFORMATION}.foo"]
async def test_routerdata_get_nonexistent_leaf_path(routerdata):
"""Test that accessing a nonexistent long path raises KeyError."""
with pytest.raises(KeyError):
routerdata[f"{KEY_DEVICE_INFORMATION}.long.path.foo"]
async def test_routerdata_get_simple(routerdata):
"""Test that accessing a short, simple path works."""
assert routerdata[f"{KEY_DEVICE_INFORMATION}.SoftwareVersion"] == "1.0"
async def test_routerdata_get_longer(routerdata):
"""Test that accessing a longer path works."""
assert routerdata[f"{KEY_DEVICE_INFORMATION}.nested.foo"] == "bar"
async def test_routerdata_get_dict(routerdata):
"""Test that returning an intermediate dict works."""
assert routerdata[f"{KEY_DEVICE_INFORMATION}.nested"] == {"foo": "bar"}