Increase Shelly code coverage for Gen1 EM3 (home-assistant#156752)

Signed-off-by: David Rapan <[email protected]>

Author: davidrapan

homeassistant/components/shelly/utils.py

@@ -424,23 +424,30 @@ def get_rpc_channel_name(device: RpcDevice, key: str) -> str | None:
     if BLU_TRV_IDENTIFIER in key:
         return None
 
-    component = key.split(":")[0]
-    component_id = key.split(":")[-1]
+    _, component, component_id = get_rpc_key(key)
 
     if custom_name := get_rpc_custom_name(device, key):
         if component in (*VIRTUAL_COMPONENTS, "input", "presencezone", "script"):
             return custom_name
 
-        channels = get_rpc_number_of_channels(device, component)
-
-        return custom_name if channels == 1 else None
+        return (
+            custom_name if get_rpc_number_of_channels(device, component) == 1 else None
+        )
 
     if component in (*VIRTUAL_COMPONENTS, "input"):
         return f"{component.title()} {component_id}"
 
     return None
 
 
+def get_rpc_key_normalized(key: str) -> str:
+    """Get normalized key. Workaround for Pro EM50 and Pro 3EM."""
+    # workaround for Pro EM50
+    key = key.replace("em1data", "em1")
+    # workaround for Pro 3EM
+    return key.replace("emdata", "em")
+
+
 def get_rpc_sub_device_name(
     device: RpcDevice, key: str, emeter_phase: str | None = None
 ) -> str:
@@ -455,11 +462,7 @@ def get_rpc_sub_device_name(
         if entity_name := device.config[key].get("name"):
             return cast(str, entity_name)
 
-    key = key.replace("emdata", "em")
-    key = key.replace("em1data", "em1")
-
-    component = key.split(":")[0]
-    component_id = key.split(":")[-1]
+    _, component, component_id = get_rpc_key(get_rpc_key_normalized(key))
 
     if component in ("cct", "rgb", "rgbw"):
         return f"{device.name} {component.upper()} light {component_id}"
@@ -528,7 +531,7 @@ def get_rpc_key_instances(
 
 def get_rpc_key_ids(keys_dict: dict[str, Any], key: str) -> list[int]:
     """Return list of key ids for RPC device from a dict."""
-    return [int(k.split(":")[1]) for k in keys_dict if k.startswith(f"{key}:")]
+    return [get_rpc_key_id(k) for k in keys_dict if k.startswith(f"{key}:")]
 
 
 def get_rpc_key_by_role(keys_dict: dict[str, Any], role: str) -> str | None:
@@ -810,11 +813,10 @@ def is_rpc_exclude_from_relay(
     settings: dict[str, Any], status: dict[str, Any], channel: str
 ) -> bool:
     """Return true if rpc channel should be excludeed from switch platform."""
-    ch = int(channel.split(":")[1])
     if is_rpc_thermostat_internal_actuator(status):
         return True
 
-    return is_rpc_channel_type_light(settings, ch)
+    return is_rpc_channel_type_light(settings, get_rpc_key_id(channel))
 
 
 def get_shelly_air_lamp_life(lamp_seconds: int) -> float:
@@ -836,7 +838,7 @@ async def get_rpc_scripts_event_types(
         if script_name in ignore_scripts:
             continue
 
-        script_id = int(script.split(":")[-1])
+        script_id = get_rpc_key_id(script)
         script_events[script_id] = await get_rpc_script_event_types(device, script_id)
 
     return script_events
@@ -867,14 +869,8 @@ def get_rpc_device_info(
     if key is None:
         return DeviceInfo(connections={(CONNECTION_NETWORK_MAC, mac)})
 
-    # workaround for Pro EM50
-    key = key.replace("em1data", "em1")
-    # workaround for Pro 3EM
-    key = key.replace("emdata", "em")
-
-    key_parts = key.split(":")
-    component = key_parts[0]
-    idx = key_parts[1] if len(key_parts) > 1 else None
+    key = get_rpc_key_normalized(key)
+    has_id, component, _ = get_rpc_key(key)
 
     if emeter_phase is not None:
         return DeviceInfo(
@@ -893,7 +889,7 @@ def get_rpc_device_info(
             component not in (*All_LIGHT_TYPES, "cover", "em1", "switch")
             and get_irrigation_zone_id(device, key) is None
         )
-        or idx is None
+        or not has_id
         or get_rpc_number_of_channels(device, component) < 2
     ):
         return DeviceInfo(connections={(CONNECTION_NETWORK_MAC, mac)})

tests/components/shelly/test_event.py

@@ -1,6 +1,6 @@
 """Tests for Shelly button platform."""
 
-import copy
+from copy import deepcopy
 from unittest.mock import Mock
 
 from aioshelly.ble.const import BLE_SCRIPT_NAME
@@ -264,7 +264,7 @@ async def test_block_event_shix3_1(
     hass: HomeAssistant, mock_block_device: Mock, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     """Test block device event for SHIX3-1."""
-    blocks = copy.deepcopy(MOCK_BLOCKS)
+    blocks = deepcopy(MOCK_BLOCKS)
     blocks[0] = Mock(
         sensor_ids={
             "inputEvent": "S",

tests/components/shelly/test_sensor.py

@@ -3,7 +3,7 @@
 from copy import deepcopy
 from unittest.mock import Mock, PropertyMock
 
-from aioshelly.const import MODEL_BLU_GATEWAY_G3
+from aioshelly.const import MODEL_BLU_GATEWAY_G3, MODEL_EM3
 from aioshelly.exceptions import NotInitialized
 from freezegun.api import FrozenDateTimeFactory
 import pytest
@@ -53,7 +53,7 @@
     register_device,
     register_entity,
 )
-from .conftest import MOCK_CONFIG, MOCK_SHELLY_RPC, MOCK_STATUS_RPC
+from .conftest import MOCK_BLOCKS, MOCK_CONFIG, MOCK_SHELLY_RPC, MOCK_STATUS_RPC
 
 from tests.common import (
     async_fire_time_changed,
@@ -97,6 +97,55 @@ async def test_block_sensor(
     assert entry.unique_id == "123456789ABC-relay_0-power"
 
 
+async def test_block_sensor_em3(
+    hass: HomeAssistant, mock_block_device: Mock, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Test block sensor of EM3."""
+    monkeypatch.setitem(mock_block_device.shelly, "num_emeters", 3)
+    monkeypatch.setitem(mock_block_device.settings["device"], "type", MODEL_EM3)
+    monkeypatch.setitem(
+        mock_block_device.settings,
+        "emeters",
+        [
+            {"name": "Grid L1", "appliance_type": "General", "max_power": 0},
+            {"appliance_type": "General", "max_power": 0},
+            {"appliance_type": "General", "max_power": 0},
+        ],
+    )
+    blocks = deepcopy(MOCK_BLOCKS)
+    blocks[5] = Mock(
+        sensor_ids={"power": 20},
+        channel="0",
+        power=20,
+        description="emeter_0",
+        type="emeter",
+    )
+    blocks.append(
+        Mock(
+            sensor_ids={"power": 20},
+            channel="1",
+            power=20,
+            description="emeter_1",
+            type="emeter",
+        )
+    )
+    blocks.append(
+        Mock(
+            sensor_ids={"power": 20},
+            channel="2",
+            power=20,
+            description="emeter_2",
+            type="emeter",
+        )
+    )
+    monkeypatch.setattr(mock_block_device, "blocks", blocks)
+    await init_integration(hass, 1, model=MODEL_EM3)
+
+    assert hass.states.get(f"{SENSOR_DOMAIN}.grid_l1_power")
+    assert hass.states.get(f"{SENSOR_DOMAIN}.test_name_phase_b_power")
+    assert hass.states.get(f"{SENSOR_DOMAIN}.test_name_phase_c_power")
+
+
 async def test_energy_sensor(
     hass: HomeAssistant, mock_block_device: Mock, entity_registry: EntityRegistry
 ) -> None:

tests/components/shelly/test_utils.py

@@ -147,6 +147,7 @@ async def test_is_block_momentary_input(
         is False
     )
 
+    monkeypatch.delitem(mock_block_device.settings, "inputs")
     monkeypatch.delitem(mock_block_device.settings, "relays")
     monkeypatch.delitem(mock_block_device.settings, "rollers")
     assert (