Shelly code quality (#86733)

This commit is contained in:
Shay Levy 2023-01-27 10:47:05 +02:00 committed by GitHub
parent e4a78420b8
commit ae6bc96002
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 76 additions and 306 deletions

View File

@ -20,7 +20,7 @@ from homeassistant.util import slugify
from .const import SHELLY_GAS_MODELS
from .coordinator import ShellyBlockCoordinator, ShellyRpcCoordinator, get_entry_data
from .utils import get_block_device_name, get_device_entry_gen, get_rpc_device_name
from .utils import get_device_entry_gen
_ShellyCoordinatorT = TypeVar(
"_ShellyCoordinatorT", bound=ShellyBlockCoordinator | ShellyRpcCoordinator
@ -121,12 +121,7 @@ class ShellyButton(
super().__init__(coordinator)
self.entity_description = description
if isinstance(coordinator, ShellyRpcCoordinator):
device_name = get_rpc_device_name(coordinator.device)
else:
device_name = get_block_device_name(coordinator.device)
self._attr_name = f"{device_name} {description.name}"
self._attr_name = f"{coordinator.device.name} {description.name}"
self._attr_unique_id = slugify(self._attr_name)
self._attr_device_info = DeviceInfo(
connections={(CONNECTION_NETWORK_MAC, coordinator.mac)}

View File

@ -33,13 +33,11 @@ from .const import (
)
from .coordinator import async_reconnect_soon, get_entry_data
from .utils import (
get_block_device_name,
get_block_device_sleep_period,
get_coap_context,
get_info_auth,
get_info_gen,
get_model_name,
get_rpc_device_name,
get_rpc_device_sleep_period,
get_ws_context,
mac_address_from_name,
@ -80,7 +78,7 @@ async def validate_input(
assert rpc_device.shelly
return {
"title": get_rpc_device_name(rpc_device),
"title": rpc_device.name,
CONF_SLEEP_PERIOD: get_rpc_device_sleep_period(rpc_device.config),
"model": rpc_device.shelly.get("model"),
"gen": 2,
@ -95,7 +93,7 @@ async def validate_input(
)
block_device.shutdown()
return {
"title": get_block_device_name(block_device),
"title": block_device.name,
CONF_SLEEP_PERIOD: get_block_device_sleep_period(block_device.settings),
"model": block_device.model,
"gen": 1,

View File

@ -51,7 +51,7 @@ from .const import (
UPDATE_PERIOD_MULTIPLIER,
BLEScannerMode,
)
from .utils import device_update_info, get_device_name, get_rpc_device_wakeup_period
from .utils import device_update_info, get_rpc_device_wakeup_period
_DeviceT = TypeVar("_DeviceT", bound="BlockDevice|RpcDevice")
@ -86,7 +86,7 @@ class ShellyCoordinatorBase(DataUpdateCoordinator[None], Generic[_DeviceT]):
self.entry = entry
self.device = device
self.device_id: str | None = None
device_name = get_device_name(device) if device.initialized else entry.title
device_name = device.name if device.initialized else entry.title
interval_td = timedelta(seconds=update_interval)
super().__init__(hass, LOGGER, name=device_name, update_interval=interval_td)

View File

@ -21,7 +21,7 @@ from .coordinator import (
get_block_coordinator_by_device_id,
get_rpc_coordinator_by_device_id,
)
from .utils import get_block_device_name, get_rpc_entity_name
from .utils import get_rpc_entity_name
@callback
@ -48,8 +48,7 @@ def async_describe_events(
elif click_type in BLOCK_INPUTS_EVENTS_TYPES:
block_coordinator = get_block_coordinator_by_device_id(hass, device_id)
if block_coordinator and block_coordinator.device.initialized:
device_name = get_block_device_name(block_coordinator.device)
input_name = f"{device_name} channel {channel}"
input_name = f"{block_coordinator.device.name} channel {channel}"
return {
LOGBOOK_ENTRY_NAME: "Shelly",

View File

@ -19,10 +19,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import slugify
from .const import CONF_SLEEP_PERIOD
from .coordinator import ShellyBlockCoordinator, ShellyRpcCoordinator, get_entry_data
from .coordinator import ShellyBlockCoordinator, ShellyRpcCoordinator
from .entity import (
RestEntityDescription,
RpcEntityDescription,
@ -31,12 +30,7 @@ from .entity import (
async_setup_entry_rest,
async_setup_entry_rpc,
)
from .utils import (
async_remove_shelly_entity,
get_block_device_name,
get_device_entry_gen,
get_rpc_device_name,
)
from .utils import get_device_entry_gen
LOGGER = logging.getLogger(__name__)
@ -123,28 +117,10 @@ async def async_setup_entry(
) -> None:
"""Set up update entities for Shelly component."""
if get_device_entry_gen(config_entry) == 2:
# Remove legacy update binary sensor & buttons, remove in 2023.2.0
rpc_coordinator = get_entry_data(hass)[config_entry.entry_id].rpc
assert rpc_coordinator
mac = rpc_coordinator.mac
async_remove_shelly_entity(hass, "binary_sensor", f"{mac}-sys-fwupdate")
device_name = slugify(get_rpc_device_name(rpc_coordinator.device))
async_remove_shelly_entity(hass, "button", f"{device_name}_ota_update")
async_remove_shelly_entity(hass, "button", f"{device_name}_ota_update_beta")
return async_setup_entry_rpc(
hass, config_entry, async_add_entities, RPC_UPDATES, RpcUpdateEntity
)
# Remove legacy update binary sensor & buttons, remove in 2023.2.0
block_coordinator = get_entry_data(hass)[config_entry.entry_id].block
assert block_coordinator
mac = block_coordinator.mac
async_remove_shelly_entity(hass, "binary_sensor", f"{mac}-fwupdate")
device_name = slugify(get_block_device_name(block_coordinator.device))
async_remove_shelly_entity(hass, "button", f"{device_name}_ota_update")
async_remove_shelly_entity(hass, "button", f"{device_name}_ota_update_beta")
if not config_entry.data[CONF_SLEEP_PERIOD]:
async_setup_entry_rest(
hass,

View File

@ -49,24 +49,6 @@ def async_remove_shelly_entity(
entity_reg.async_remove(entity_id)
def get_block_device_name(device: BlockDevice) -> str:
"""Get Block device name."""
return cast(str, device.settings["name"] or device.settings["device"]["hostname"])
def get_rpc_device_name(device: RpcDevice) -> str:
"""Get RPC device name."""
return cast(str, device.config["sys"]["device"].get("name") or device.hostname)
def get_device_name(device: BlockDevice | RpcDevice) -> str:
"""Get device name."""
if isinstance(device, BlockDevice):
return get_block_device_name(device)
return get_rpc_device_name(device)
def get_number_of_channels(device: BlockDevice, block: Block) -> int:
"""Get number of channels for block type."""
assert isinstance(device.shelly, dict)
@ -105,7 +87,7 @@ def get_block_entity_name(
def get_block_channel_name(device: BlockDevice, block: Block | None) -> str:
"""Get name based on device and channel name."""
entity_name = get_block_device_name(device)
entity_name = device.name
if (
not block
@ -306,7 +288,7 @@ def get_rpc_channel_name(device: RpcDevice, key: str) -> str:
"""Get name based on device and channel name."""
if device.config.get("switch:0"):
key = key.replace("input", "switch")
device_name = get_rpc_device_name(device)
device_name = device.name
entity_name: str | None = None
if key in device.config:
entity_name = device.config[key].get("name", device_name)

View File

@ -1,7 +1,7 @@
"""Test configuration for Shelly."""
from __future__ import annotations
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import AsyncMock, Mock, PropertyMock, patch
from aioshelly.block_device import BlockDevice
from aioshelly.rpc_device import RpcDevice, UpdateType
@ -245,7 +245,9 @@ async def mock_block_device():
status=MOCK_STATUS_COAP,
firmware_version="some fw string",
initialized=True,
model="SHSW-1",
)
type(device).name = PropertyMock(return_value="Test name")
block_device_mock.return_value = device
block_device_mock.return_value.mock_update = Mock(side_effect=update)
@ -254,7 +256,7 @@ async def mock_block_device():
def _mock_rpc_device(version: str | None = None):
"""Mock rpc (Gen2, Websocket) device."""
return Mock(
device = Mock(
spec=RpcDevice,
config=MOCK_CONFIG,
event={},
@ -265,6 +267,8 @@ def _mock_rpc_device(version: str | None = None):
firmware_version="some fw string",
initialized=True,
)
type(device).name = PropertyMock(return_value="Test name")
return device
@pytest.fixture

View File

@ -2,7 +2,7 @@
from __future__ import annotations
from dataclasses import replace
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import AsyncMock, patch
from aioshelly.exceptions import (
DeviceConnectionError,
@ -55,8 +55,14 @@ MOCK_CONFIG = {
}
@pytest.mark.parametrize("gen", [1, 2])
async def test_form(hass, gen):
@pytest.mark.parametrize(
"gen, model",
[
(1, "SHSW-1"),
(2, "SNSW-002P16EU"),
],
)
async def test_form(hass, mock_block_device, mock_rpc_device, gen, model):
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
@ -68,23 +74,6 @@ async def test_form(hass, gen):
with patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False, "gen": gen},
), patch(
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=MOCK_SETTINGS,
)
),
), patch(
"aioshelly.rpc_device.RpcDevice.create",
new=AsyncMock(
return_value=Mock(
shelly={"model": "SHSW-1", "gen": gen},
config=MOCK_CONFIG,
shutdown=AsyncMock(),
)
),
), patch(
"homeassistant.components.shelly.async_setup", return_value=True
) as mock_setup, patch(
@ -101,7 +90,7 @@ async def test_form(hass, gen):
assert result2["title"] == "Test name"
assert result2["data"] == {
"host": "1.1.1.1",
"model": "SHSW-1",
"model": model,
"sleep_period": 0,
"gen": gen,
}
@ -109,64 +98,26 @@ async def test_form(hass, gen):
assert len(mock_setup_entry.mock_calls) == 1
async def test_title_without_name(hass):
"""Test we set the title to the hostname when the device doesn't have a name."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {}
settings = MOCK_SETTINGS.copy()
settings["name"] = None
settings["device"] = settings["device"].copy()
settings["device"]["hostname"] = "shelly1pm-12345"
with patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False},
), patch(
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=settings,
)
),
), patch(
"homeassistant.components.shelly.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.shelly.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"host": "1.1.1.1"},
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result2["title"] == "shelly1pm-12345"
assert result2["data"] == {
"host": "1.1.1.1",
"model": "SHSW-1",
"sleep_period": 0,
"gen": 1,
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
"test_data",
[
(1, {"username": "test user", "password": "test1 password"}, "test user"),
(2, {"password": "test2 password"}, "admin"),
(
1,
"SHSW-1",
{"username": "test user", "password": "test1 password"},
"test user",
),
(
2,
"SNSW-002P16EU",
{"password": "test2 password"},
"admin",
),
],
)
async def test_form_auth(hass, test_data):
async def test_form_auth(hass, test_data, mock_block_device, mock_rpc_device):
"""Test manual configuration if auth is required."""
gen, user_input, username = test_data
gen, model, user_input, username = test_data
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -186,23 +137,6 @@ async def test_form_auth(hass, test_data):
assert result["errors"] == {}
with patch(
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=MOCK_SETTINGS,
)
),
), patch(
"aioshelly.rpc_device.RpcDevice.create",
new=AsyncMock(
return_value=Mock(
shelly={"model": "SHSW-1", "gen": gen},
config=MOCK_CONFIG,
shutdown=AsyncMock(),
)
),
), patch(
"homeassistant.components.shelly.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.shelly.async_setup_entry",
@ -217,7 +151,7 @@ async def test_form_auth(hass, test_data):
assert result3["title"] == "Test name"
assert result3["data"] == {
"host": "1.1.1.1",
"model": "SHSW-1",
"model": model,
"sleep_period": 0,
"gen": gen,
"username": username,
@ -247,23 +181,15 @@ async def test_form_errors_get_info(hass, error):
assert result2["errors"] == {"base": base_error}
async def test_form_missing_model_key(hass):
async def test_form_missing_model_key(hass, mock_rpc_device, monkeypatch):
"""Test we handle missing Shelly model key."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
monkeypatch.setattr(mock_rpc_device, "shelly", {"gen": 2})
with patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value={"mac": "test-mac", "auth": False, "gen": "2"},
), patch(
"aioshelly.rpc_device.RpcDevice.create",
new=AsyncMock(
return_value=Mock(
shelly={"gen": 2},
config=MOCK_CONFIG,
shutdown=AsyncMock(),
)
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -274,7 +200,7 @@ async def test_form_missing_model_key(hass):
assert result2["errors"] == {"base": "firmware_not_fully_provisioned"}
async def test_form_missing_model_key_auth_enabled(hass):
async def test_form_missing_model_key_auth_enabled(hass, mock_rpc_device, monkeypatch):
"""Test we handle missing Shelly model key when auth enabled."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -294,39 +220,22 @@ async def test_form_missing_model_key_auth_enabled(hass):
assert result2["type"] == data_entry_flow.FlowResultType.FORM
assert result["errors"] == {}
with patch(
"aioshelly.rpc_device.RpcDevice.create",
new=AsyncMock(
return_value=Mock(
shelly={"gen": 2},
config=MOCK_CONFIG,
shutdown=AsyncMock(),
)
),
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"], {"password": "1234"}
)
monkeypatch.setattr(mock_rpc_device, "shelly", {"gen": 2})
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"], {"password": "1234"}
)
assert result3["type"] == data_entry_flow.FlowResultType.FORM
assert result3["errors"] == {"base": "firmware_not_fully_provisioned"}
async def test_form_missing_model_key_zeroconf(hass, caplog):
async def test_form_missing_model_key_zeroconf(
hass, mock_rpc_device, monkeypatch, caplog
):
"""Test we handle missing Shelly model key via zeroconf."""
monkeypatch.setattr(mock_rpc_device, "shelly", {"gen": 2})
with patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value={"mac": "test-mac", "auth": False, "gen": 2},
), patch(
"aioshelly.rpc_device.RpcDevice.create",
new=AsyncMock(
return_value=Mock(
shelly={"gen": 2},
config=MOCK_CONFIG,
shutdown=AsyncMock(),
)
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
@ -397,7 +306,7 @@ async def test_form_already_configured(hass):
assert entry.data["host"] == "1.1.1.1"
async def test_user_setup_ignored_device(hass):
async def test_user_setup_ignored_device(hass, mock_block_device):
"""Test user can successfully setup an ignored device."""
entry = MockConfigEntry(
@ -412,21 +321,9 @@ async def test_user_setup_ignored_device(hass):
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
settings = MOCK_SETTINGS.copy()
settings["device"]["type"] = "SHSW-1"
settings["fw"] = "20201124-092534/v1.9.0@57ac4ad8"
with patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False},
), patch(
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=settings,
)
),
), patch(
"homeassistant.components.shelly.async_setup", return_value=True
) as mock_setup, patch(
@ -538,29 +435,25 @@ async def test_form_auth_errors_test_connection_gen2(hass, error):
@pytest.mark.parametrize(
"gen, get_info",
"gen, model, get_info",
[
(1, {"mac": "test-mac", "type": "SHSW-1", "auth": False, "gen": 1}),
(2, {"mac": "test-mac", "model": "SHSW-1", "auth": False, "gen": 2}),
(
1,
"SHSW-1",
{"mac": "test-mac", "type": "SHSW-1", "auth": False, "gen": 1},
),
(
2,
"SNSW-002P16EU",
{"mac": "test-mac", "model": "SHSW-1", "auth": False, "gen": 2},
),
],
)
async def test_zeroconf(hass, gen, get_info):
async def test_zeroconf(hass, gen, model, get_info, mock_block_device, mock_rpc_device):
"""Test we get the form."""
with patch(
"homeassistant.components.shelly.config_flow.get_info", return_value=get_info
), patch(
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(return_value=Mock(model="SHSW-1", settings=MOCK_SETTINGS)),
), patch(
"aioshelly.rpc_device.RpcDevice.create",
new=AsyncMock(
return_value=Mock(
shelly={"model": "SHSW-1", "gen": gen},
config=MOCK_CONFIG,
shutdown=AsyncMock(),
)
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
@ -592,7 +485,7 @@ async def test_zeroconf(hass, gen, get_info):
assert result2["title"] == "Test name"
assert result2["data"] == {
"host": "1.1.1.1",
"model": "SHSW-1",
"model": model,
"sleep_period": 0,
"gen": gen,
}
@ -600,9 +493,13 @@ async def test_zeroconf(hass, gen, get_info):
assert len(mock_setup_entry.mock_calls) == 1
async def test_zeroconf_sleeping_device(hass):
async def test_zeroconf_sleeping_device(hass, mock_block_device, monkeypatch):
"""Test sleeping device configuration via zeroconf."""
monkeypatch.setitem(
mock_block_device.settings,
"sleep_mode",
{"period": 10, "unit": "m"},
)
with patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value={
@ -611,22 +508,6 @@ async def test_zeroconf_sleeping_device(hass):
"auth": False,
"sleep_mode": True,
},
), patch(
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings={
"name": "Test name",
"device": {
"mac": "test-mac",
"hostname": "test-host",
"type": "SHSW-1",
},
"sleep_mode": {"period": 10, "unit": "m"},
},
)
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
@ -791,7 +672,7 @@ async def test_zeroconf_cannot_connect(hass):
assert result["reason"] == "cannot_connect"
async def test_zeroconf_require_auth(hass):
async def test_zeroconf_require_auth(hass, mock_block_device):
"""Test zeroconf if auth is required."""
with patch(
@ -807,14 +688,6 @@ async def test_zeroconf_require_auth(hass):
assert result["errors"] == {}
with patch(
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=MOCK_SETTINGS,
)
),
), patch(
"homeassistant.components.shelly.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.shelly.async_setup_entry",
@ -847,7 +720,7 @@ async def test_zeroconf_require_auth(hass):
(2, {"password": "test2 password"}),
],
)
async def test_reauth_successful(hass, test_data):
async def test_reauth_successful(hass, test_data, mock_block_device, mock_rpc_device):
"""Test starting a reauthentication flow."""
gen, user_input = test_data
entry = MockConfigEntry(
@ -858,23 +731,6 @@ async def test_reauth_successful(hass, test_data):
with patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": True, "gen": gen},
), patch(
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=MOCK_SETTINGS,
)
),
), patch(
"aioshelly.rpc_device.RpcDevice.create",
new=AsyncMock(
return_value=Mock(
shelly={"model": "SHSW-1", "gen": gen},
config=MOCK_CONFIG,
shutdown=AsyncMock(),
)
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,

View File

@ -4,8 +4,6 @@ from unittest.mock import AsyncMock
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
import pytest
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN
from homeassistant.components.shelly.const import DOMAIN
from homeassistant.components.update import (
ATTR_IN_PROGRESS,
@ -23,44 +21,6 @@ from homeassistant.helpers.entity_registry import async_get
from . import MOCK_MAC, init_integration, mock_rest_update
@pytest.mark.parametrize(
"gen, domain, unique_id, object_id",
[
(1, BINARY_SENSOR_DOMAIN, f"{MOCK_MAC}-fwupdate", "firmware_update"),
(1, BUTTON_DOMAIN, "test_name_ota_update", "ota_update"),
(1, BUTTON_DOMAIN, "test_name_ota_update_beta", "ota_update_beta"),
(2, BINARY_SENSOR_DOMAIN, f"{MOCK_MAC}-sys-fwupdate", "firmware_update"),
(2, BUTTON_DOMAIN, "test_name_ota_update", "ota_update"),
(2, BUTTON_DOMAIN, "test_name_ota_update_beta", "ota_update_beta"),
],
)
async def test_remove_legacy_entities(
hass: HomeAssistant,
gen,
domain,
unique_id,
object_id,
mock_block_device,
mock_rpc_device,
):
"""Test removes legacy update entities."""
entity_id = f"{domain}.test_name_{object_id}"
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
domain,
DOMAIN,
unique_id,
suggested_object_id=f"test_name_{object_id}",
disabled_by=None,
)
assert entity_registry.async_get(entity_id) is not None
await init_integration(hass, gen)
assert entity_registry.async_get(entity_id) is None
async def test_block_update(hass: HomeAssistant, mock_block_device, monkeypatch):
"""Test block device update entity."""
entity_registry = async_get(hass)