Add Integration for Energenie Power-Sockets (#113097)

* Integration for Energenie Power-Strips (EGPS)

* cleanups reocommended by reviewer

* Adds missing exception handling when trying to send a command to an unreachable device.

* fix: incorrect handling of already opened devices in pyegps api. bump to pyegps=0.2.4

* Add blank line after file docstring, and other cosmetics

* change asyncio.to_thread to async_add_executer_job

* raises HomeAssistantError EgpsException in switch services.

* switch test parameterized by entity name

* reoved unused device registry

* add translation_key and update_before_add

* bump pyegps dependency to version to 0.2.5

* combined get_device patches and put into conftest.py

* changed switch entity to use _attr_is_on and cleanups

* further cleanup

* Apply suggestions from code review

* refactor: rename egps to energenie_power_sockets

* updated test snapshot

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

.strict-typing

@@ -166,6 +166,7 @@ homeassistant.components.electric_kiwi.*
 homeassistant.components.elgato.*
 homeassistant.components.elkm1.*
 homeassistant.components.emulated_hue.*
+homeassistant.components.energenie_power_sockets.*
 homeassistant.components.energy.*
 homeassistant.components.energyzero.*
 homeassistant.components.enigma2.*

CODEOWNERS

@@ -378,6 +378,8 @@ build.json @home-assistant/supervisor
 /tests/components/emulated_hue/ @bdraco @Tho85
 /homeassistant/components/emulated_kasa/ @kbickar
 /tests/components/emulated_kasa/ @kbickar
+/homeassistant/components/energenie_power_sockets/ @gnumpi
+/tests/components/energenie_power_sockets/ @gnumpi
 /homeassistant/components/energy/ @home-assistant/core
 /tests/components/energy/ @home-assistant/core
 /homeassistant/components/energyzero/ @klaasnicolaas

homeassistant/components/energenie_power_sockets/__init__.py

@@ -0,0 +1,44 @@
+"""Energenie Power-Sockets (EGPS) integration."""
+
+from pyegps import PowerStripUSB, get_device
+from pyegps.exceptions import MissingLibrary, UsbError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
+
+from .const import CONF_DEVICE_API_ID, DOMAIN
+
+PLATFORMS = [Platform.SWITCH]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Energenie Power Sockets."""
+    try:
+        powerstrip: PowerStripUSB | None = get_device(entry.data[CONF_DEVICE_API_ID])
+
+    except (MissingLibrary, UsbError) as ex:
+        raise ConfigEntryError("Can't access usb devices.") from ex
+
+    if powerstrip is None:
+        raise ConfigEntryNotReady(
+            "Can't access Energenie Power Sockets, will retry later."
+        )
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = powerstrip
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        powerstrip = hass.data[DOMAIN].pop(entry.entry_id)
+        powerstrip.release()
+
+    if not hass.data[DOMAIN]:
+        hass.data.pop(DOMAIN)
+
+    return unload_ok

homeassistant/components/energenie_power_sockets/config_flow.py

@@ -0,0 +1,55 @@
+"""ConfigFlow for Energenie-Power-Sockets devices."""
+
+from typing import Any
+
+from pyegps import get_device, search_for_devices
+from pyegps.exceptions import MissingLibrary, UsbError
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+
+from .const import CONF_DEVICE_API_ID, DOMAIN, LOGGER
+
+
+class EGPSConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle the config flow for EGPS devices."""
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Initiate user flow."""
+
+        if user_input is not None:
+            dev_id = user_input[CONF_DEVICE_API_ID]
+            dev = await self.hass.async_add_executor_job(get_device, dev_id)
+            if dev is not None:
+                await self.async_set_unique_id(dev.device_id)
+                self._abort_if_unique_id_configured()
+
+                return self.async_create_entry(
+                    title=dev_id,
+                    data={CONF_DEVICE_API_ID: dev_id},
+                )
+            return self.async_abort(reason="device_not_found")
+
+        currently_configured = self._async_current_ids(include_ignore=True)
+        try:
+            found_devices = await self.hass.async_add_executor_job(search_for_devices)
+        except (MissingLibrary, UsbError):
+            LOGGER.exception("Unable to access USB devices")
+            return self.async_abort(reason="usb_error")
+
+        devices = [
+            d
+            for d in found_devices
+            if d.get_device_type() == "PowerStrip"
+            and d.device_id not in currently_configured
+        ]
+        LOGGER.debug("Found %d devices", len(devices))
+        if len(devices) > 0:
+            options = {d.device_id: f"{d.name} ({d.device_id})" for d in devices}
+            data_schema = {CONF_DEVICE_API_ID: vol.In(options)}
+        else:
+            return self.async_abort(reason="no_device")
+
+        return self.async_show_form(step_id="user", data_schema=vol.Schema(data_schema))

homeassistant/components/energenie_power_sockets/const.py

@@ -0,0 +1,8 @@
+"""Constants for Energenie Power Sockets."""
+
+import logging
+
+LOGGER = logging.getLogger(__package__)
+
+CONF_DEVICE_API_ID = "api-device-id"
+DOMAIN = "energenie_power_sockets"

homeassistant/components/energenie_power_sockets/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "energenie_power_sockets",
+  "name": "Energenie Power Sockets",
+  "codeowners": ["@gnumpi"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/energenie_power_sockets",
+  "integration_type": "device",
+  "iot_class": "local_polling",
+  "loggers": ["pyegps"],
+  "requirements": ["pyegps==0.2.5"]
+}

homeassistant/components/energenie_power_sockets/strings.json

@@ -0,0 +1,27 @@
+{
+  "title": "Energenie Power Sockets Integration.",
+  "config": {
+    "step": {
+      "user": {
+        "title": "Searching for Energenie-Power-Sockets Devices.",
+        "description": "Choose a discovered device.",
+        "data": {
+          "device": "[%key:common::config_flow::data::device%]"
+        }
+      }
+    },
+    "abort": {
+      "usb_error": "Couldn't access USB devices!",
+      "no_device": "Unable to discover any (new) supported device.",
+      "device_not_found": "No device was found for the given id.",
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  },
+  "entity": {
+    "switch": {
+      "socket": {
+        "name": "Socket {socket_id}"
+      }
+    }
+  }
+}

homeassistant/components/energenie_power_sockets/switch.py

@@ -0,0 +1,77 @@
+"""Switch implementation for Energenie-Power-Sockets Platform."""
+
+from typing import Any
+
+from pyegps import __version__ as PYEGPS_VERSION
+from pyegps.exceptions import EgpsException
+from pyegps.powerstrip import PowerStrip
+
+from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Add EGPS sockets for passed config_entry in HA."""
+    powerstrip: PowerStrip = hass.data[DOMAIN][config_entry.entry_id]
+
+    async_add_entities(
+        (
+            EGPowerStripSocket(powerstrip, socket)
+            for socket in range(powerstrip.numberOfSockets)
+        ),
+        update_before_add=True,
+    )
+
+
+class EGPowerStripSocket(SwitchEntity):
+    """Represents a socket of an Energenie-Socket-Strip."""
+
+    _attr_device_class = SwitchDeviceClass.OUTLET
+    _attr_has_entity_name = True
+    _attr_translation_key = "socket"
+
+    def __init__(self, dev: PowerStrip, socket: int) -> None:
+        """Initiate a new socket."""
+        self._dev = dev
+        self._socket = socket
+        self._attr_translation_placeholders = {"socket_id": str(socket)}
+
+        self._attr_unique_id = f"{dev.device_id}_{socket}"
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, dev.device_id)},
+            name=dev.name,
+            manufacturer=dev.manufacturer,
+            model=dev.name,
+            sw_version=PYEGPS_VERSION,
+        )
+
+    def turn_on(self, **kwargs: Any) -> None:
+        """Switch the socket on."""
+        try:
+            self._dev.switch_on(self._socket)
+        except EgpsException as err:
+            raise HomeAssistantError(f"Couldn't access USB device: {err}") from err
+
+    def turn_off(self, **kwargs: Any) -> None:
+        """Switch the socket off."""
+        try:
+            self._dev.switch_off(self._socket)
+        except EgpsException as err:
+            raise HomeAssistantError(f"Couldn't access USB device: {err}") from err
+
+    def update(self) -> None:
+        """Read the current state from the device."""
+        try:
+            self._attr_is_on = self._dev.get_status(self._socket)
+        except EgpsException as err:
+            raise HomeAssistantError(f"Couldn't access USB device: {err}") from err

homeassistant/generated/config_flows.py

@@ -144,6 +144,7 @@ FLOWS = {
         "elvia",
         "emonitor",
         "emulated_roku",
+        "energenie_power_sockets",
         "energyzero",
         "enocean",
         "enphase_envoy",

homeassistant/generated/integrations.json

@@ -1571,6 +1571,11 @@
       "config_flow": true,
       "iot_class": "local_push"
     },
+    "energenie_power_sockets": {
+      "integration_type": "device",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "energie_vanons": {
       "name": "Energie VanOns",
       "integration_type": "virtual",
@@ -7179,6 +7184,7 @@
     "demo",
     "derivative",
     "emulated_roku",
+    "energenie_power_sockets",
     "filesize",
     "garages_amsterdam",
     "generic",

mypy.ini

@@ -1421,6 +1421,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.energenie_power_sockets.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.energy.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_all.txt

@@ -1799,6 +1799,9 @@ pyedimax==0.2.1
 # homeassistant.components.efergy
 pyefergy==22.1.1
 
+# homeassistant.components.energenie_power_sockets
+pyegps==0.2.5
+
 # homeassistant.components.enphase_envoy
 pyenphase==1.20.1
 

requirements_test_all.txt

@@ -1398,6 +1398,9 @@ pyeconet==0.1.22
 # homeassistant.components.efergy
 pyefergy==22.1.1
 
+# homeassistant.components.energenie_power_sockets
+pyegps==0.2.5
+
 # homeassistant.components.enphase_envoy
 pyenphase==1.20.1
 

tests/components/energenie_power_sockets/__init__.py

@@ -0,0 +1 @@
+"""Tests for Energenie-Power-Sockets (EGPS) integration."""

tests/components/energenie_power_sockets/conftest.py

@@ -0,0 +1,83 @@
+"""Configure tests for Energenie-Power-Sockets."""
+
+from collections.abc import Generator
+from typing import Final
+from unittest.mock import MagicMock, patch
+
+from pyegps.fakes.powerstrip import FakePowerStrip
+import pytest
+
+from homeassistant.components.energenie_power_sockets.const import (
+    CONF_DEVICE_API_ID,
+    DOMAIN,
+)
+from homeassistant.const import CONF_NAME
+
+from tests.common import MockConfigEntry
+
+DEMO_CONFIG_DATA: Final = {
+    CONF_NAME: "Unit Test",
+    CONF_DEVICE_API_ID: "DYPS:00:11:22",
+}
+
+
+@pytest.fixture
+def demo_config_data() -> dict:
+    """Return valid user input."""
+    return {CONF_DEVICE_API_ID: DEMO_CONFIG_DATA[CONF_DEVICE_API_ID]}
+
+
+@pytest.fixture
+def valid_config_entry() -> MockConfigEntry:
+    """Return a valid egps config entry."""
+    return MockConfigEntry(
+        domain=DOMAIN,
+        data=DEMO_CONFIG_DATA,
+        unique_id=DEMO_CONFIG_DATA[CONF_DEVICE_API_ID],
+    )
+
+
+@pytest.fixture(name="pyegps_device_mock")
+def get_pyegps_device_mock() -> MagicMock:
+    """Fixture for a mocked FakePowerStrip."""
+
+    fkObj = FakePowerStrip(
+        devId=DEMO_CONFIG_DATA[CONF_DEVICE_API_ID], number_of_sockets=4
+    )
+    fkObj.release = lambda: True
+    fkObj._status = [0, 1, 0, 1]
+
+    usb_device_mock = MagicMock(wraps=fkObj)
+    usb_device_mock.get_device_type.return_value = "PowerStrip"
+    usb_device_mock.numberOfSockets = 4
+    usb_device_mock.device_id = DEMO_CONFIG_DATA[CONF_DEVICE_API_ID]
+    usb_device_mock.manufacturer = "Energenie"
+    usb_device_mock.name = "MockedUSBDevice"
+
+    return usb_device_mock
+
+
+@pytest.fixture(name="mock_get_device")
+def patch_get_device(pyegps_device_mock: MagicMock) -> Generator[MagicMock, None, None]:
+    """Fixture to patch the `get_device` api method."""
+    with (
+        patch("homeassistant.components.energenie_power_sockets.get_device") as m1,
+        patch(
+            "homeassistant.components.energenie_power_sockets.config_flow.get_device",
+            new=m1,
+        ) as mock,
+    ):
+        mock.return_value = pyegps_device_mock
+        yield mock
+
+
+@pytest.fixture(name="mock_search_for_devices")
+def patch_search_devices(
+    pyegps_device_mock: MagicMock,
+) -> Generator[MagicMock, None, None]:
+    """Fixture to patch the `search_for_devices` api method."""
+    with patch(
+        "homeassistant.components.energenie_power_sockets.config_flow.search_for_devices",
+        return_value=[pyegps_device_mock],
+    ) as mock:
+        yield mock

tests/components/energenie_power_sockets/snapshots/test_switch.ambr

@@ -0,0 +1,189 @@
+# serializer version: 1
+# name: test_switch_setup[mockedusbdevice_socket_0]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'outlet',
+      'friendly_name': 'MockedUSBDevice Socket 0',
+    }),
+    'context': <ANY>,
+    'entity_id': 'switch.mockedusbdevice_socket_0',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'off',
+  })
+# ---
+# name: test_switch_setup[mockedusbdevice_socket_0].1
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'switch',
+    'entity_category': None,
+    'entity_id': 'switch.mockedusbdevice_socket_0',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <SwitchDeviceClass.OUTLET: 'outlet'>,
+    'original_icon': None,
+    'original_name': 'Socket 0',
+    'platform': 'energenie_power_sockets',
+    'previous_unique_id': None,
+    'supported_features': 0,
+    'translation_key': 'socket',
+    'unique_id': 'DYPS:00:11:22_0',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_switch_setup[mockedusbdevice_socket_1]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'outlet',
+      'friendly_name': 'MockedUSBDevice Socket 1',
+    }),
+    'context': <ANY>,
+    'entity_id': 'switch.mockedusbdevice_socket_1',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'on',
+  })
+# ---
+# name: test_switch_setup[mockedusbdevice_socket_1].1
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'switch',
+    'entity_category': None,
+    'entity_id': 'switch.mockedusbdevice_socket_1',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <SwitchDeviceClass.OUTLET: 'outlet'>,
+    'original_icon': None,
+    'original_name': 'Socket 1',
+    'platform': 'energenie_power_sockets',
+    'previous_unique_id': None,
+    'supported_features': 0,
+    'translation_key': 'socket',
+    'unique_id': 'DYPS:00:11:22_1',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_switch_setup[mockedusbdevice_socket_2]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'outlet',
+      'friendly_name': 'MockedUSBDevice Socket 2',
+    }),
+    'context': <ANY>,
+    'entity_id': 'switch.mockedusbdevice_socket_2',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'off',
+  })
+# ---
+# name: test_switch_setup[mockedusbdevice_socket_2].1
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'switch',
+    'entity_category': None,
+    'entity_id': 'switch.mockedusbdevice_socket_2',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <SwitchDeviceClass.OUTLET: 'outlet'>,
+    'original_icon': None,
+    'original_name': 'Socket 2',
+    'platform': 'energenie_power_sockets',
+    'previous_unique_id': None,
+    'supported_features': 0,
+    'translation_key': 'socket',
+    'unique_id': 'DYPS:00:11:22_2',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_switch_setup[mockedusbdevice_socket_3]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'outlet',
+      'friendly_name': 'MockedUSBDevice Socket 3',
+    }),
+    'context': <ANY>,
+    'entity_id': 'switch.mockedusbdevice_socket_3',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'on',
+  })
+# ---
+# name: test_switch_setup[mockedusbdevice_socket_3].1
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'switch',
+    'entity_category': None,
+    'entity_id': 'switch.mockedusbdevice_socket_3',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <SwitchDeviceClass.OUTLET: 'outlet'>,
+    'original_icon': None,
+    'original_name': 'Socket 3',
+    'platform': 'energenie_power_sockets',
+    'previous_unique_id': None,
+    'supported_features': 0,
+    'translation_key': 'socket',
+    'unique_id': 'DYPS:00:11:22_3',
+    'unit_of_measurement': None,
+  })
+# ---

tests/components/energenie_power_sockets/test_config_flow.py

@@ -0,0 +1,140 @@
+"""Tests for Energenie-Power-Sockets config flow."""
+
+from unittest.mock import MagicMock
+
+from pyegps.exceptions import UsbError
+
+from homeassistant.components.energenie_power_sockets.const import (
+    CONF_DEVICE_API_ID,
+    DOMAIN,
+)
+from homeassistant.config_entries import SOURCE_USER
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from tests.common import MockConfigEntry
+
+
+async def test_user_flow(
+    hass: HomeAssistant,
+    demo_config_data: dict,
+    mock_get_device: MagicMock,
+    mock_search_for_devices: MagicMock,
+) -> None:
+    """Test configuration flow initialized by the user."""
+
+    result1 = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result1["type"] == FlowResultType.FORM
+    assert not result1["errors"]
+
+    # check with valid data
+    result2 = await hass.config_entries.flow.async_configure(
+        result1["flow_id"], user_input=demo_config_data
+    )
+    assert result2["type"] == FlowResultType.CREATE_ENTRY
+
+
+async def test_user_flow_already_exists(
+    hass: HomeAssistant,
+    valid_config_entry: MockConfigEntry,
+    mock_get_device: MagicMock,
+    mock_search_for_devices: MagicMock,
+) -> None:
+    """Test the flow when device has been already configured."""
+    valid_config_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+        data={CONF_DEVICE_API_ID: valid_config_entry.data[CONF_DEVICE_API_ID]},
+    )
+
+    await hass.async_block_till_done()
+
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "already_configured"
+
+
+async def test_user_flow_no_new_device(
+    hass: HomeAssistant,
+    valid_config_entry: MockConfigEntry,
+    mock_get_device: MagicMock,
+    mock_search_for_devices: MagicMock,
+) -> None:
+    """Test the flow when the found device has been already included."""
+    valid_config_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+        data=None,
+    )
+
+    await hass.async_block_till_done()
+
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "no_device"
+
+
+async def test_user_flow_no_device_found(
+    hass: HomeAssistant,
+    demo_config_data: dict,
+    mock_get_device: MagicMock,
+    mock_search_for_devices: MagicMock,
+) -> None:
+    """Test configuration flow when no device is found."""
+
+    mock_search_for_devices.return_value = []
+
+    result1 = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result1["type"] == FlowResultType.ABORT
+    assert result1["reason"] == "no_device"
+
+
+async def test_user_flow_device_not_found(
+    hass: HomeAssistant,
+    demo_config_data: dict,
+    mock_get_device: MagicMock,
+    mock_search_for_devices: MagicMock,
+) -> None:
+    """Test configuration flow when the given device_id does not match any found devices."""
+
+    mock_get_device.return_value = None
+
+    result1 = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result1["type"] == FlowResultType.FORM
+    assert not result1["errors"]
+
+    # check with valid data
+    result2 = await hass.config_entries.flow.async_configure(
+        result1["flow_id"], user_input=demo_config_data
+    )
+    assert result2["type"] == FlowResultType.ABORT
+    assert result2["reason"] == "device_not_found"
+
+
+async def test_user_flow_no_usb_access(
+    hass: HomeAssistant,
+    mock_get_device: MagicMock,
+    mock_search_for_devices: MagicMock,
+) -> None:
+    """Test configuration flow when USB devices can't be accessed."""
+
+    mock_get_device.return_value = None
+    mock_search_for_devices.side_effect = UsbError
+
+    result1 = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result1["type"] == FlowResultType.ABORT
+    assert result1["reason"] == "usb_error"

tests/components/energenie_power_sockets/test_init.py

@@ -0,0 +1,64 @@
+"""Tests for setting up Energenie-Power-Sockets integration."""
+
+from unittest.mock import MagicMock
+
+from pyegps.exceptions import UsbError
+
+from homeassistant.components.energenie_power_sockets.const import DOMAIN
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+async def test_load_unload_entry(
+    hass: HomeAssistant,
+    valid_config_entry: MockConfigEntry,
+    mock_get_device: MagicMock,
+) -> None:
+    """Test loading and unloading the integration."""
+    entry = valid_config_entry
+    entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert entry.state == ConfigEntryState.LOADED
+    assert entry.entry_id in hass.data[DOMAIN]
+
+    assert await hass.config_entries.async_unload(entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert entry.state is ConfigEntryState.NOT_LOADED
+    assert DOMAIN not in hass.data
+
+
+async def test_device_not_found_on_load_entry(
+    hass: HomeAssistant,
+    valid_config_entry: MockConfigEntry,
+    mock_get_device: MagicMock,
+) -> None:
+    """Test device not available on config entry setup."""
+
+    mock_get_device.return_value = None
+
+    valid_config_entry.add_to_hass(hass)
+    assert not await hass.config_entries.async_setup(valid_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert valid_config_entry.state is ConfigEntryState.SETUP_RETRY
+
+
+async def test_usb_error(
+    hass: HomeAssistant, valid_config_entry: MockConfigEntry, mock_get_device: MagicMock
+) -> None:
+    """Test no USB access on config entry setup."""
+
+    mock_get_device.side_effect = UsbError
+
+    valid_config_entry.add_to_hass(hass)
+
+    assert not await hass.config_entries.async_setup(valid_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert valid_config_entry.state is ConfigEntryState.SETUP_ERROR

tests/components/energenie_power_sockets/test_switch.py

@@ -0,0 +1,134 @@
+"""Test the switch functionality."""
+
+from unittest.mock import MagicMock
+
+from pyegps.exceptions import EgpsException
+import pytest
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.components.energenie_power_sockets.const import DOMAIN
+from homeassistant.components.homeassistant import (
+    DOMAIN as HOME_ASSISTANT_DOMAIN,
+    SERVICE_UPDATE_ENTITY,
+)
+from homeassistant.components.switch import (
+    DOMAIN as SWITCH_DOMAIN,
+    SERVICE_TURN_OFF,
+    SERVICE_TURN_ON,
+)
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.const import STATE_OFF, STATE_ON
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as er
+
+from tests.common import MockConfigEntry
+
+
+async def _test_switch_on_off(
+    hass: HomeAssistant, entity_id: str, dev: MagicMock
+) -> None:
+    """Call switch on/off service."""
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        SERVICE_TURN_ON,
+        {"entity_id": entity_id},
+        blocking=True,
+    )
+    assert hass.states.get(entity_id).state == STATE_ON
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        SERVICE_TURN_OFF,
+        {"entity_id": entity_id},
+        blocking=True,
+    )
+
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+
+async def _test_switch_on_exeception(
+    hass: HomeAssistant, entity_id: str, dev: MagicMock
+) -> None:
+    """Call switch on service with USBError side effect."""
+    dev.switch_on.side_effect = EgpsException
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            HOME_ASSISTANT_DOMAIN,
+            SERVICE_TURN_ON,
+            {"entity_id": entity_id},
+            blocking=True,
+        )
+    dev.switch_on.side_effect = None
+
+
+async def _test_switch_off_exeception(
+    hass: HomeAssistant, entity_id: str, dev: MagicMock
+) -> None:
+    """Call switch off service with USBError side effect."""
+    dev.switch_off.side_effect = EgpsException
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            SWITCH_DOMAIN,
+            SERVICE_TURN_OFF,
+            {"entity_id": entity_id},
+            blocking=True,
+        )
+    dev.switch_off.side_effect = None
+
+
+async def _test_switch_update_exception(
+    hass: HomeAssistant, entity_id: str, dev: MagicMock
+) -> None:
+    """Call switch update with USBError side effect."""
+    dev.get_status.side_effect = EgpsException
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            SWITCH_DOMAIN,
+            SERVICE_UPDATE_ENTITY,
+            {"entity_id": entity_id},
+            blocking=True,
+        )
+    dev.get_status.side_effect = None
+
+
+@pytest.mark.parametrize(
+    "entity_name",
+    [
+        "mockedusbdevice_socket_0",
+        "mockedusbdevice_socket_1",
+        "mockedusbdevice_socket_2",
+        "mockedusbdevice_socket_3",
+    ],
+)
+async def test_switch_setup(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    valid_config_entry: MockConfigEntry,
+    mock_get_device: MagicMock,
+    entity_name: str,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test setup and functionality of device switches."""
+
+    entry = valid_config_entry
+    entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert entry.state == ConfigEntryState.LOADED
+    assert entry.entry_id in hass.data[DOMAIN]
+
+    state = hass.states.get(f"switch.{entity_name}")
+    assert state == snapshot
+    assert entity_registry.async_get(state.entity_id) == snapshot
+
+    device_mock = mock_get_device.return_value
+    await _test_switch_on_off(hass, state.entity_id, device_mock)
+    await _test_switch_on_exeception(hass, state.entity_id, device_mock)
+    await _test_switch_off_exeception(hass, state.entity_id, device_mock)
+    await _test_switch_update_exception(hass, state.entity_id, device_mock)
+
+    assert await hass.config_entries.async_unload(entry.entry_id)
+    await hass.async_block_till_done()