From 6d54f686a65214d05d4cf54cc8edd64865d0e4f5 Mon Sep 17 00:00:00 2001 From: Mischa Siekmann <45062894+gnumpi@users.noreply.github.com> Date: Fri, 29 Mar 2024 13:29:14 +0100 Subject: [PATCH] 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 --- .strict-typing | 1 + CODEOWNERS | 2 + .../energenie_power_sockets/__init__.py | 44 ++++ .../energenie_power_sockets/config_flow.py | 55 +++++ .../energenie_power_sockets/const.py | 8 + .../energenie_power_sockets/manifest.json | 11 + .../energenie_power_sockets/strings.json | 27 +++ .../energenie_power_sockets/switch.py | 77 +++++++ homeassistant/generated/config_flows.py | 1 + homeassistant/generated/integrations.json | 6 + mypy.ini | 10 + requirements_all.txt | 3 + requirements_test_all.txt | 3 + .../energenie_power_sockets/__init__.py | 1 + .../energenie_power_sockets/conftest.py | 83 ++++++++ .../snapshots/test_switch.ambr | 189 ++++++++++++++++++ .../test_config_flow.py | 140 +++++++++++++ .../energenie_power_sockets/test_init.py | 64 ++++++ .../energenie_power_sockets/test_switch.py | 134 +++++++++++++ 19 files changed, 859 insertions(+) create mode 100644 homeassistant/components/energenie_power_sockets/__init__.py create mode 100644 homeassistant/components/energenie_power_sockets/config_flow.py create mode 100644 homeassistant/components/energenie_power_sockets/const.py create mode 100644 homeassistant/components/energenie_power_sockets/manifest.json create mode 100644 homeassistant/components/energenie_power_sockets/strings.json create mode 100644 homeassistant/components/energenie_power_sockets/switch.py create mode 100644 tests/components/energenie_power_sockets/__init__.py create mode 100644 tests/components/energenie_power_sockets/conftest.py create mode 100644 tests/components/energenie_power_sockets/snapshots/test_switch.ambr create mode 100644 tests/components/energenie_power_sockets/test_config_flow.py create mode 100644 tests/components/energenie_power_sockets/test_init.py create mode 100644 tests/components/energenie_power_sockets/test_switch.py diff --git a/.strict-typing b/.strict-typing index 39ff23a472e..b1d6df7c9b8 100644 --- a/.strict-typing +++ b/.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.* diff --git a/CODEOWNERS b/CODEOWNERS index 81add403413..59359e708f9 100644 --- a/CODEOWNERS +++ b/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 diff --git a/homeassistant/components/energenie_power_sockets/__init__.py b/homeassistant/components/energenie_power_sockets/__init__.py new file mode 100644 index 00000000000..12ddb0d1389 --- /dev/null +++ b/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 diff --git a/homeassistant/components/energenie_power_sockets/config_flow.py b/homeassistant/components/energenie_power_sockets/config_flow.py new file mode 100644 index 00000000000..ab39427f15a --- /dev/null +++ b/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)) diff --git a/homeassistant/components/energenie_power_sockets/const.py b/homeassistant/components/energenie_power_sockets/const.py new file mode 100644 index 00000000000..a02373815c2 --- /dev/null +++ b/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" diff --git a/homeassistant/components/energenie_power_sockets/manifest.json b/homeassistant/components/energenie_power_sockets/manifest.json new file mode 100644 index 00000000000..8a55a539e7f --- /dev/null +++ b/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"] +} diff --git a/homeassistant/components/energenie_power_sockets/strings.json b/homeassistant/components/energenie_power_sockets/strings.json new file mode 100644 index 00000000000..e193b06b25f --- /dev/null +++ b/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}" + } + } + } +} diff --git a/homeassistant/components/energenie_power_sockets/switch.py b/homeassistant/components/energenie_power_sockets/switch.py new file mode 100644 index 00000000000..1d5b9ed5197 --- /dev/null +++ b/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 diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 283cdf1a0de..acac5f8df5d 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -144,6 +144,7 @@ FLOWS = { "elvia", "emonitor", "emulated_roku", + "energenie_power_sockets", "energyzero", "enocean", "enphase_envoy", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index c30f22acc6e..631c8b1e73c 100644 --- a/homeassistant/generated/integrations.json +++ b/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", diff --git a/mypy.ini b/mypy.ini index 66af4c9c25a..159101a21b3 100644 --- a/mypy.ini +++ b/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 diff --git a/requirements_all.txt b/requirements_all.txt index 06d4ca4f6c8..aba5ae375e0 100644 --- a/requirements_all.txt +++ b/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 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 3e5f81a0767..f672c943803 100644 --- a/requirements_test_all.txt +++ b/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 diff --git a/tests/components/energenie_power_sockets/__init__.py b/tests/components/energenie_power_sockets/__init__.py new file mode 100644 index 00000000000..8397567ef82 --- /dev/null +++ b/tests/components/energenie_power_sockets/__init__.py @@ -0,0 +1 @@ +"""Tests for Energenie-Power-Sockets (EGPS) integration.""" diff --git a/tests/components/energenie_power_sockets/conftest.py b/tests/components/energenie_power_sockets/conftest.py new file mode 100644 index 00000000000..f119c0008f7 --- /dev/null +++ b/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 diff --git a/tests/components/energenie_power_sockets/snapshots/test_switch.ambr b/tests/components/energenie_power_sockets/snapshots/test_switch.ambr new file mode 100644 index 00000000000..d462d6ca6d4 --- /dev/null +++ b/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': , + 'entity_id': 'switch.mockedusbdevice_socket_0', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'off', + }) +# --- +# name: test_switch_setup[mockedusbdevice_socket_0].1 + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'switch', + 'entity_category': None, + 'entity_id': 'switch.mockedusbdevice_socket_0', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': , + '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': , + 'entity_id': 'switch.mockedusbdevice_socket_1', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'on', + }) +# --- +# name: test_switch_setup[mockedusbdevice_socket_1].1 + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'switch', + 'entity_category': None, + 'entity_id': 'switch.mockedusbdevice_socket_1', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': , + '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': , + 'entity_id': 'switch.mockedusbdevice_socket_2', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'off', + }) +# --- +# name: test_switch_setup[mockedusbdevice_socket_2].1 + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'switch', + 'entity_category': None, + 'entity_id': 'switch.mockedusbdevice_socket_2', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': , + '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': , + 'entity_id': 'switch.mockedusbdevice_socket_3', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'on', + }) +# --- +# name: test_switch_setup[mockedusbdevice_socket_3].1 + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'switch', + 'entity_category': None, + 'entity_id': 'switch.mockedusbdevice_socket_3', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': , + '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, + }) +# --- diff --git a/tests/components/energenie_power_sockets/test_config_flow.py b/tests/components/energenie_power_sockets/test_config_flow.py new file mode 100644 index 00000000000..ef433d0ef09 --- /dev/null +++ b/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" diff --git a/tests/components/energenie_power_sockets/test_init.py b/tests/components/energenie_power_sockets/test_init.py new file mode 100644 index 00000000000..a60949c34cc --- /dev/null +++ b/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 diff --git a/tests/components/energenie_power_sockets/test_switch.py b/tests/components/energenie_power_sockets/test_switch.py new file mode 100644 index 00000000000..b98a3e07f56 --- /dev/null +++ b/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()