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

Code Quality improvements for Aussie Broadband (#65408)

Browse source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Brett Adams 2022-02-17 08:38:05 +10:00 • committed by GitHub
parent be0ef5ad6c
commit bccfaceedb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 62 additions and 229 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
homeassistant/components/aussie_broadband/__init__.py
View file

@ -15,7 +15,7 @@ from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_SERVICES, DEFAULT_UPDATE_INTERVAL, DOMAIN, SERVICE_ID
from .const import DEFAULT_UPDATE_INTERVAL, DOMAIN, SERVICE_ID
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.SENSOR]
@ -31,17 +31,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
)
try:
await client.login()
all_services = await client.get_services()
services = await client.get_services()
except AuthenticationException as exc:
raise ConfigEntryAuthFailed() from exc
except ClientError as exc:
raise ConfigEntryNotReady() from exc
# Filter the service list to those that are enabled in options
services = [
s for s in all_services if str(s["service_id"]) in entry.options[CONF_SERVICES]
]
# Create an appropriate refresh function
def update_data_factory(service_id):
async def async_update_data():
@ -71,16 +66,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"services": services,
}
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
return True
async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Reload to update options."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload the config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

99
homeassistant/components/aussie_broadband/config_flow.py
View file

@ -9,12 +9,10 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from .const import CONF_SERVICES, DOMAIN, SERVICE_ID
from .const import CONF_SERVICES, DOMAIN
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@ -39,11 +37,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
try:
await self.client.login()
return None
except AuthenticationException:
return {"base": "invalid_auth"}
except ClientError:
return {"base": "cannot_connect"}
return None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
@ -61,15 +59,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if not self.services:
return self.async_abort(reason="no_services_found")
if len(self.services) == 1:
return self.async_create_entry(
title=self.data[CONF_USERNAME],
data=self.data,
options={CONF_SERVICES: [str(self.services[0][SERVICE_ID])]},
)
# Account has more than one service, select service to add
return await self.async_step_service()
return self.async_create_entry(
title=self.data[CONF_USERNAME],
data=self.data,
)
return self.async_show_form(
step_id="user",
@ -82,37 +75,20 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_service(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the optional service selection step."""
if user_input is not None:
return self.async_create_entry(
title=self.data[CONF_USERNAME], data=self.data, options=user_input
)
async def async_step_reauth(self, user_input: dict[str, str]) -> FlowResult:
"""Handle reauth on credential failure."""
self._reauth_username = user_input[CONF_USERNAME]
service_options = {str(s[SERVICE_ID]): s["description"] for s in self.services}
return self.async_show_form(
step_id="service",
data_schema=vol.Schema(
{
vol.Required(
CONF_SERVICES, default=list(service_options.keys())
): cv.multi_select(service_options)
}
),
errors=None,
)
return await self.async_step_reauth_confirm()
async def async_step_reauth(
self, user_input: dict[str, Any] | None = None
async def async_step_reauth_confirm(
self, user_input: dict[str, str] | None = None
) -> FlowResult:
"""Handle reauth."""
"""Handle users reauth credentials."""
errors: dict[str, str] | None = None
if user_input and user_input.get(CONF_USERNAME):
self._reauth_username = user_input[CONF_USERNAME]
elif self._reauth_username and user_input and user_input.get(CONF_PASSWORD):
if user_input and self._reauth_username:
data = {
CONF_USERNAME: self._reauth_username,
CONF_PASSWORD: user_input[CONF_PASSWORD],
@ -130,7 +106,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=self._reauth_username, data=data)
return self.async_show_form(
step_id="reauth",
step_id="reauth_confirm",
description_placeholders={"username": self._reauth_username},
data_schema=vol.Schema(
{
@ -139,46 +115,3 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
),
errors=errors,
)
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> config_entries.OptionsFlow:
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Options flow for picking services."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
if self.config_entry.state != config_entries.ConfigEntryState.LOADED:
return self.async_abort(reason="unknown")
data = self.hass.data[DOMAIN][self.config_entry.entry_id]
try:
services = await data["client"].get_services()
except AuthenticationException:
return self.async_abort(reason="invalid_auth")
except ClientError:
return self.async_abort(reason="cannot_connect")
service_options = {str(s[SERVICE_ID]): s["description"] for s in services}
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Required(
CONF_SERVICES,
default=self.config_entry.options.get(CONF_SERVICES),
): cv.multi_select(service_options),
}
),
)

58
homeassistant/components/aussie_broadband/sensor.py
View file

@ -1,7 +1,9 @@
"""Support for Aussie Broadband metric sensors."""
from __future__ import annotations
from typing import Any
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, cast
from homeassistant.components.sensor import (
SensorEntity,
@ -14,27 +16,36 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, SERVICE_ID
SENSOR_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
@dataclass
class SensorValueEntityDescription(SensorEntityDescription):
"""Class describing Aussie Broadband sensor entities."""
value: Callable = lambda x: x
SENSOR_DESCRIPTIONS: tuple[SensorValueEntityDescription, ...] = (
# Internet Services sensors
SensorEntityDescription(
SensorValueEntityDescription(
key="usedMb",
name="Data Used",
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=DATA_MEGABYTES,
icon="mdi:network",
),
SensorEntityDescription(
SensorValueEntityDescription(
key="downloadedMb",
name="Downloaded",
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=DATA_MEGABYTES,
icon="mdi:download-network",
),
SensorEntityDescription(
SensorValueEntityDescription(
key="uploadedMb",
name="Uploaded",
state_class=SensorStateClass.TOTAL_INCREASING,
@ -42,46 +53,50 @@ SENSOR_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
icon="mdi:upload-network",
),
# Mobile Phone Services sensors
SensorEntityDescription(
SensorValueEntityDescription(
key="national",
name="National Calls",
state_class=SensorStateClass.TOTAL_INCREASING,
icon="mdi:phone",
value=lambda x: x.get("calls"),
),
SensorEntityDescription(
SensorValueEntityDescription(
key="mobile",
name="Mobile Calls",
state_class=SensorStateClass.TOTAL_INCREASING,
icon="mdi:phone",
value=lambda x: x.get("calls"),
),
SensorEntityDescription(
SensorValueEntityDescription(
key="international",
name="International Calls",
entity_registry_enabled_default=False,
state_class=SensorStateClass.TOTAL_INCREASING,
icon="mdi:phone-plus",
),
SensorEntityDescription(
SensorValueEntityDescription(
key="sms",
name="SMS Sent",
state_class=SensorStateClass.TOTAL_INCREASING,
icon="mdi:message-processing",
value=lambda x: x.get("calls"),
),
SensorEntityDescription(
SensorValueEntityDescription(
key="internet",
name="Data Used",
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=DATA_KILOBYTES,
icon="mdi:network",
value=lambda x: x.get("kbytes"),
),
SensorEntityDescription(
SensorValueEntityDescription(
key="voicemail",
name="Voicemail Calls",
entity_registry_enabled_default=False,
state_class=SensorStateClass.TOTAL_INCREASING,
icon="mdi:phone",
),
SensorEntityDescription(
SensorValueEntityDescription(
key="other",
name="Other Calls",
entity_registry_enabled_default=False,
@ -89,13 +104,13 @@ SENSOR_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
icon="mdi:phone",
),
# Generic sensors
SensorEntityDescription(
SensorValueEntityDescription(
key="daysTotal",
name="Billing Cycle Length",
native_unit_of_measurement=TIME_DAYS,
icon="mdi:calendar-range",
),
SensorEntityDescription(
SensorValueEntityDescription(
key="daysRemaining",
name="Billing Cycle Remaining",
native_unit_of_measurement=TIME_DAYS,
@ -122,8 +137,10 @@ async def async_setup_entry(
class AussieBroadandSensorEntity(CoordinatorEntity, SensorEntity):
"""Base class for Aussie Broadband metric sensors."""
entity_description: SensorValueEntityDescription
def __init__(
self, service: dict[str, Any], description: SensorEntityDescription
self, service: dict[str, Any], description: SensorValueEntityDescription
) -> None:
"""Initialize the sensor."""
super().__init__(service["coordinator"])
@ -134,16 +151,13 @@ class AussieBroadandSensorEntity(CoordinatorEntity, SensorEntity):
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, service[SERVICE_ID])},
manufacturer="Aussie Broadband",
configuration_url=f"https://my.aussiebroadband.com.au/#/{service['name']}/{service[SERVICE_ID]}/",
configuration_url=f"https://my.aussiebroadband.com.au/#/{service['name'].lower()}/{service[SERVICE_ID]}/",
name=service["description"],
model=service["name"],
)
@property
def native_value(self):
def native_value(self) -> StateType:
"""Return the state of the sensor."""
if self.entity_description.key == "internet":
return self.coordinator.data[self.entity_description.key].get("kbytes")
if self.entity_description.key in ("national", "mobile", "sms"):
return self.coordinator.data[self.entity_description.key].get("calls")
return self.coordinator.data[self.entity_description.key]
parent = self.coordinator.data[self.entity_description.key]
return cast(StateType, self.entity_description.value(parent))

2
homeassistant/components/aussie_broadband/strings.json
View file

@ -13,7 +13,7 @@
"services": "Services"
}
},
"reauth": {
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "Update password for {username}",
"data": {

2
homeassistant/components/aussie_broadband/translations/en.json
View file

@ -11,7 +11,7 @@
"unknown": "Unexpected error"
},
"step": {
"reauth": {
"reauth_confirm": {
"data": {
"password": "Password"
},

115
tests/components/aussie_broadband/test_config_flow.py
View file

@ -4,8 +4,8 @@ from unittest.mock import patch
from aiohttp import ClientConnectionError
from aussiebb.asyncio import AuthenticationException
from homeassistant import config_entries, setup
from homeassistant.components.aussie_broadband.const import CONF_SERVICES, DOMAIN
from homeassistant import config_entries
from homeassistant.components.aussie_broadband.const import DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
@ -14,7 +14,7 @@ from homeassistant.data_entry_flow import (
RESULT_TYPE_FORM,
)
from .common import FAKE_DATA, FAKE_SERVICES, setup_platform
from .common import FAKE_DATA, FAKE_SERVICES
TEST_USERNAME = FAKE_DATA[CONF_USERNAME]
TEST_PASSWORD = FAKE_DATA[CONF_PASSWORD]
@ -31,7 +31,7 @@ async def test_form(hass: HomeAssistant) -> None:
with patch("aussiebb.asyncio.AussieBB.__init__", return_value=None), patch(
"aussiebb.asyncio.AussieBB.login", return_value=True
), patch(
"aussiebb.asyncio.AussieBB.get_services", return_value=[FAKE_SERVICES[0]]
"aussiebb.asyncio.AussieBB.get_services", return_value=FAKE_SERVICES
), patch(
"homeassistant.components.aussie_broadband.async_setup_entry",
return_value=True,
@ -45,7 +45,6 @@ async def test_form(hass: HomeAssistant) -> None:
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == TEST_USERNAME
assert result2["data"] == FAKE_DATA
assert result2["options"] == {CONF_SERVICES: ["12345678"]}
assert len(mock_setup_entry.mock_calls) == 1
@ -117,46 +116,6 @@ async def test_no_services(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 0
async def test_form_multiple_services(hass: HomeAssistant) -> None:
"""Test the config flow with multiple services."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
with patch("aussiebb.asyncio.AussieBB.__init__", return_value=None), patch(
"aussiebb.asyncio.AussieBB.login", return_value=True
), patch("aussiebb.asyncio.AussieBB.get_services", return_value=FAKE_SERVICES):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
FAKE_DATA,
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_FORM
assert result2["step_id"] == "service"
assert result2["errors"] is None
with patch(
"homeassistant.components.aussie_broadband.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_SERVICES: [FAKE_SERVICES[1]["service_id"]]},
)
await hass.async_block_till_done()
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
assert result3["title"] == TEST_USERNAME
assert result3["data"] == FAKE_DATA
assert result3["options"] == {
CONF_SERVICES: [FAKE_SERVICES[1]["service_id"]],
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
"""Test invalid auth is handled."""
result1 = await hass.config_entries.flow.async_init(
@ -196,8 +155,6 @@ async def test_form_network_issue(hass: HomeAssistant) -> None:
async def test_reauth(hass: HomeAssistant) -> None:
"""Test reauth flow."""
await setup.async_setup_component(hass, "persistent_notification", {})
# Test reauth but the entry doesn't exist
result1 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_REAUTH}, data=FAKE_DATA
@ -229,7 +186,7 @@ async def test_reauth(hass: HomeAssistant) -> None:
context={"source": config_entries.SOURCE_REAUTH},
data=FAKE_DATA,
)
assert result5["step_id"] == "reauth"
assert result5["step_id"] == "reauth_confirm"
with patch("aussiebb.asyncio.AussieBB.__init__", return_value=None), patch(
"aussiebb.asyncio.AussieBB.login", side_effect=AuthenticationException()
@ -243,7 +200,7 @@ async def test_reauth(hass: HomeAssistant) -> None:
)
await hass.async_block_till_done()
assert result6["step_id"] == "reauth"
assert result6["step_id"] == "reauth_confirm"
# Test successful reauth
with patch("aussiebb.asyncio.AussieBB.__init__", return_value=None), patch(
@ -260,63 +217,3 @@ async def test_reauth(hass: HomeAssistant) -> None:
assert result7["type"] == "abort"
assert result7["reason"] == "reauth_successful"
async def test_options_flow(hass):
"""Test options flow."""
entry = await setup_platform(hass)
with patch("aussiebb.asyncio.AussieBB.get_services", return_value=FAKE_SERVICES):
result1 = await hass.config_entries.options.async_init(entry.entry_id)
assert result1["type"] == RESULT_TYPE_FORM
assert result1["step_id"] == "init"
result2 = await hass.config_entries.options.async_configure(
result1["flow_id"],
user_input={CONF_SERVICES: []},
)
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert entry.options == {CONF_SERVICES: []}
async def test_options_flow_auth_failure(hass):
"""Test options flow with auth failure."""
entry = await setup_platform(hass)
with patch(
"aussiebb.asyncio.AussieBB.get_services", side_effect=AuthenticationException()
):
result1 = await hass.config_entries.options.async_init(entry.entry_id)
assert result1["type"] == RESULT_TYPE_ABORT
assert result1["reason"] == "invalid_auth"
async def test_options_flow_network_failure(hass):
"""Test options flow with connectivity failure."""
entry = await setup_platform(hass)
with patch(
"aussiebb.asyncio.AussieBB.get_services", side_effect=ClientConnectionError()
):
result1 = await hass.config_entries.options.async_init(entry.entry_id)
assert result1["type"] == RESULT_TYPE_ABORT
assert result1["reason"] == "cannot_connect"
async def test_options_flow_not_loaded(hass):
"""Test the options flow aborts when the entry has unloaded due to a reauth."""
entry = await setup_platform(hass)
with patch(
"aussiebb.asyncio.AussieBB.get_services", side_effect=AuthenticationException()
):
entry.state = config_entries.ConfigEntryState.NOT_LOADED
result1 = await hass.config_entries.options.async_init(entry.entry_id)
assert result1["type"] == RESULT_TYPE_ABORT
assert result1["reason"] == "unknown"