mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 08:47:57 +00:00
Add config flow to mystrom (#74719)
* rebase * fixed review comments * fix test * Update tests * increase test coverage * implement some review comments * Enhance device check for old FWs and add tests * Reworked exception handling * small code optimizations * fix test * Increase test coverage * Update __init__.py changed from hass.config_entries.async_setup_platforms(entry, PLATFORMS) to await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) * Update __init__.py remove spaces * Bump python-mystrom to 2.2.0 * Migrate to get_device_info from python-mystrom * Migrate to get_device_info from python-mystrom * Update __init__.py * update requirements_all.txt * update config_flow * fix tests * Update homeassistant/components/mystrom/__init__.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Implemented review changes * increase test coverage * Implemented user defined title * implemented user defined title * Additional review comments * fix test * fix linter * fix linter * Update homeassistant/components/mystrom/models.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * fix review comments * fix missing import * simplify * Update homeassistant/components/mystrom/__init__.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/mystrom/__init__.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/mystrom/config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/mystrom/config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/mystrom/config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_init.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_init.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_init.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_init.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/mystrom/light.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update homeassistant/components/mystrom/switch.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * fix review comments * fix review comments * Update tests/components/mystrom/test_config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * fix review comment * Update tests/components/mystrom/test_config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> * Update tests/components/mystrom/test_config_flow.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> --------- Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>
This commit is contained in:
Paul Frank
GitHub
parent
ebafb1f2c8
commit
c67f32b924
@ -786,6 +786,7 @@ build.json @home-assistant/supervisor
|
||||
/homeassistant/components/mysensors/ @MartinHjelmare @functionpointer
|
||||
/tests/components/mysensors/ @MartinHjelmare @functionpointer
|
||||
/homeassistant/components/mystrom/ @fabaff
|
||||
/tests/components/mystrom/ @fabaff
|
||||
/homeassistant/components/nam/ @bieniu
|
||||
/tests/components/nam/ @bieniu
|
||||
/homeassistant/components/nanoleaf/ @milanmeu
|
||||
|
||||
@ -1 +1,79 @@
|
||||
"""The mystrom component."""
|
||||
"""The myStrom integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import pymystrom
|
||||
from pymystrom.bulb import MyStromBulb
|
||||
from pymystrom.exceptions import MyStromConnectionError
|
||||
from pymystrom.switch import MyStromSwitch
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
|
||||
from .const import DOMAIN
|
||||
from .models import MyStromData
|
||||
|
||||
PLATFORMS_SWITCH = [Platform.SWITCH]
|
||||
PLATFORMS_BULB = [Platform.LIGHT]
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up myStrom from a config entry."""
|
||||
host = entry.data[CONF_HOST]
|
||||
device = None
|
||||
try:
|
||||
info = await pymystrom.get_device_info(host)
|
||||
except MyStromConnectionError as err:
|
||||
_LOGGER.error("No route to myStrom plug: %s", host)
|
||||
raise ConfigEntryNotReady() from err
|
||||
|
||||
device_type = info["type"]
|
||||
if device_type in [101, 106, 107]:
|
||||
device = MyStromSwitch(host)
|
||||
platforms = PLATFORMS_SWITCH
|
||||
elif device_type == 102:
|
||||
mac = info["mac"]
|
||||
device = MyStromBulb(host, mac)
|
||||
platforms = PLATFORMS_BULB
|
||||
if device.bulb_type not in ["rgblamp", "strip"]:
|
||||
_LOGGER.error(
|
||||
"Device %s (%s) is not a myStrom bulb nor myStrom LED Strip",
|
||||
host,
|
||||
mac,
|
||||
)
|
||||
return False
|
||||
else:
|
||||
_LOGGER.error("Unsupported myStrom device type: %s", device_type)
|
||||
return False
|
||||
|
||||
try:
|
||||
await device.get_state()
|
||||
except MyStromConnectionError as err:
|
||||
_LOGGER.error("No route to myStrom plug: %s", info["ip"])
|
||||
raise ConfigEntryNotReady() from err
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = MyStromData(
|
||||
device=device,
|
||||
info=info,
|
||||
)
|
||||
await hass.config_entries.async_forward_entry_setups(entry, platforms)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
device_type = hass.data[DOMAIN][entry.entry_id].info["type"]
|
||||
if device_type in [101, 106, 107]:
|
||||
platforms = PLATFORMS_SWITCH
|
||||
elif device_type == 102:
|
||||
platforms = PLATFORMS_BULB
|
||||
if unload_ok := await hass.config_entries.async_unload_platforms(entry, platforms):
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
|
||||
57
homeassistant/components/mystrom/config_flow.py
Normal file
57
homeassistant/components/mystrom/config_flow.py
Normal file
@ -0,0 +1,57 @@
|
||||
"""Config flow for myStrom integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import pymystrom
|
||||
from pymystrom.exceptions import MyStromConnectionError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_NAME = "myStrom Device"
|
||||
|
||||
STEP_USER_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): str,
|
||||
vol.Required(CONF_HOST): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for myStrom."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_import(self, import_config: dict[str, Any]) -> FlowResult:
|
||||
"""Handle import from config."""
|
||||
return await self.async_step_user(import_config)
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
if user_input is not None:
|
||||
try:
|
||||
info = await pymystrom.get_device_info(user_input[CONF_HOST])
|
||||
except MyStromConnectionError:
|
||||
errors["base"] = "cannot_connect"
|
||||
else:
|
||||
await self.async_set_unique_id(info["mac"])
|
||||
self._abort_if_unique_id_configured()
|
||||
data = {CONF_HOST: user_input[CONF_HOST]}
|
||||
title = user_input.get(CONF_NAME) or DEFAULT_NAME
|
||||
return self.async_create_entry(title=title, data=data)
|
||||
|
||||
schema = self.add_suggested_values_to_schema(STEP_USER_DATA_SCHEMA, user_input)
|
||||
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
|
||||
@ -1,2 +1,4 @@
|
||||
"""Constants for the myStrom integration."""
|
||||
DOMAIN = "mystrom"
|
||||
DEFAULT_NAME = "myStrom"
|
||||
MANUFACTURER = "myStrom"
|
||||
|
||||
@ -4,7 +4,6 @@ from __future__ import annotations
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pymystrom.bulb import MyStromBulb
|
||||
from pymystrom.exceptions import MyStromConnectionError
|
||||
import voluptuous as vol
|
||||
|
||||
@ -17,13 +16,17 @@ from homeassistant.components.light import (
|
||||
LightEntity,
|
||||
LightEntityFeature,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import PlatformNotReady
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from .const import DOMAIN, MANUFACTURER
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_NAME = "myStrom bulb"
|
||||
@ -40,6 +43,15 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up the myStrom entities."""
|
||||
info = hass.data[DOMAIN][entry.entry_id].info
|
||||
device = hass.data[DOMAIN][entry.entry_id].device
|
||||
async_add_entities([MyStromLight(device, entry.title, info["mac"])])
|
||||
|
||||
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
@ -47,23 +59,20 @@ async def async_setup_platform(
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the myStrom light integration."""
|
||||
host = config.get(CONF_HOST)
|
||||
mac = config.get(CONF_MAC)
|
||||
name = config.get(CONF_NAME)
|
||||
|
||||
bulb = MyStromBulb(host, mac)
|
||||
try:
|
||||
await bulb.get_state()
|
||||
if bulb.bulb_type not in ["rgblamp", "strip"]:
|
||||
_LOGGER.error(
|
||||
"Device %s (%s) is not a myStrom bulb nor myStrom LED Strip", host, mac
|
||||
)
|
||||
return
|
||||
except MyStromConnectionError as err:
|
||||
_LOGGER.warning("No route to myStrom bulb: %s", host)
|
||||
raise PlatformNotReady() from err
|
||||
|
||||
async_add_entities([MyStromLight(bulb, name, mac)], True)
|
||||
async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
"deprecated_yaml",
|
||||
breaks_in_ha_version="2023.12.0",
|
||||
is_fixable=False,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml",
|
||||
)
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class MyStromLight(LightEntity):
|
||||
@ -81,6 +90,12 @@ class MyStromLight(LightEntity):
|
||||
self._attr_available = False
|
||||
self._attr_unique_id = mac
|
||||
self._attr_hs_color = 0, 0
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, mac)},
|
||||
name=name,
|
||||
manufacturer=MANUFACTURER,
|
||||
sw_version=self._bulb.firmware,
|
||||
)
|
||||
|
||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn on the light."""
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
"domain": "mystrom",
|
||||
"name": "myStrom",
|
||||
"codeowners": ["@fabaff"],
|
||||
"config_flow": true,
|
||||
"dependencies": ["http"],
|
||||
"documentation": "https://www.home-assistant.io/integrations/mystrom",
|
||||
"iot_class": "local_polling",
|
||||
|
||||
14
homeassistant/components/mystrom/models.py
Normal file
14
homeassistant/components/mystrom/models.py
Normal file
@ -0,0 +1,14 @@
|
||||
"""Models for the mystrom integration."""
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from pymystrom.bulb import MyStromBulb
|
||||
from pymystrom.switch import MyStromSwitch
|
||||
|
||||
|
||||
@dataclass
|
||||
class MyStromData:
|
||||
"""Data class for mystrom device data."""
|
||||
|
||||
device: MyStromSwitch | MyStromBulb
|
||||
info: dict[str, Any]
|
||||
24
homeassistant/components/mystrom/strings.json
Normal file
24
homeassistant/components/mystrom/strings.json
Normal file
@ -0,0 +1,24 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"name": "[%key:common::config_flow::data::name%]",
|
||||
"host": "[%key:common::config_flow::data::host%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"deprecated_yaml": {
|
||||
"title": "The myStrom YAML configuration is being removed",
|
||||
"description": "Configuring myStrom using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the myStrom YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -5,17 +5,20 @@ import logging
|
||||
from typing import Any
|
||||
|
||||
from pymystrom.exceptions import MyStromConnectionError
|
||||
from pymystrom.switch import MyStromSwitch as _MyStromSwitch
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.switch import PLATFORM_SCHEMA, SwitchEntity
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import PlatformNotReady
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from .const import DOMAIN, MANUFACTURER
|
||||
|
||||
DEFAULT_NAME = "myStrom Switch"
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
@ -28,6 +31,14 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up the myStrom entities."""
|
||||
device = hass.data[DOMAIN][entry.entry_id].device
|
||||
async_add_entities([MyStromSwitch(device, entry.title)])
|
||||
|
||||
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
@ -35,17 +46,20 @@ async def async_setup_platform(
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the myStrom switch/plug integration."""
|
||||
name = config.get(CONF_NAME)
|
||||
host = config.get(CONF_HOST)
|
||||
|
||||
try:
|
||||
plug = _MyStromSwitch(host)
|
||||
await plug.get_state()
|
||||
except MyStromConnectionError as err:
|
||||
_LOGGER.error("No route to myStrom plug: %s", host)
|
||||
raise PlatformNotReady() from err
|
||||
|
||||
async_add_entities([MyStromSwitch(plug, name)])
|
||||
async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
"deprecated_yaml",
|
||||
breaks_in_ha_version="2023.12.0",
|
||||
is_fixable=False,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml",
|
||||
)
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class MyStromSwitch(SwitchEntity):
|
||||
@ -56,6 +70,12 @@ class MyStromSwitch(SwitchEntity):
|
||||
self.plug = plug
|
||||
self._attr_name = name
|
||||
self._attr_unique_id = self.plug.mac
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, self.plug.mac)},
|
||||
name=name,
|
||||
manufacturer=MANUFACTURER,
|
||||
sw_version=self.plug.firmware,
|
||||
)
|
||||
|
||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn the switch on."""
|
||||
|
||||
@ -285,6 +285,7 @@ FLOWS = {
|
||||
"mutesync",
|
||||
"myq",
|
||||
"mysensors",
|
||||
"mystrom",
|
||||
"nam",
|
||||
"nanoleaf",
|
||||
"neato",
|
||||
|
||||
@ -3556,7 +3556,7 @@
|
||||
"mystrom": {
|
||||
"name": "myStrom",
|
||||
"integration_type": "hub",
|
||||
"config_flow": false,
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling"
|
||||
},
|
||||
"mythicbeastsdns": {
|
||||
|
||||
@ -1543,6 +1543,9 @@ python-matter-server==3.4.1
|
||||
# homeassistant.components.xiaomi_miio
|
||||
python-miio==0.5.12
|
||||
|
||||
# homeassistant.components.mystrom
|
||||
python-mystrom==2.2.0
|
||||
|
||||
# homeassistant.components.nest
|
||||
python-nest==4.2.0
|
||||
|
||||
|
||||
1
tests/components/mystrom/__init__.py
Normal file
1
tests/components/mystrom/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
"""Tests for the myStrom integration."""
|
||||
37
tests/components/mystrom/conftest.py
Normal file
37
tests/components/mystrom/conftest.py
Normal file
@ -0,0 +1,37 @@
|
||||
"""Provide common mystrom fixtures and mocks."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.mystrom.const import DOMAIN
|
||||
from homeassistant.const import CONF_HOST
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
DEVICE_NAME = "myStrom Device"
|
||||
DEVICE_MAC = "6001940376EB"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
|
||||
"""Override async_setup_entry."""
|
||||
with patch(
|
||||
"homeassistant.components.mystrom.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_entry(hass: HomeAssistant) -> MockConfigEntry:
|
||||
"""Create and add a config entry."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id=DEVICE_MAC,
|
||||
data={CONF_HOST: "1.1.1.1"},
|
||||
title=DEVICE_NAME,
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
return config_entry
|
||||
133
tests/components/mystrom/test_config_flow.py
Normal file
133
tests/components/mystrom/test_config_flow.py
Normal file
@ -0,0 +1,133 @@
|
||||
"""Test the myStrom config flow."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from pymystrom.exceptions import MyStromConnectionError
|
||||
import pytest
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.mystrom.const import DOMAIN
|
||||
from homeassistant.const import CONF_HOST
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from .conftest import DEVICE_MAC
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
|
||||
|
||||
|
||||
async def test_form_combined(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
|
||||
"""Test we get the form."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"pymystrom.get_device_info",
|
||||
side_effect=AsyncMock(return_value={"type": 101, "mac": DEVICE_MAC}),
|
||||
):
|
||||
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"] == FlowResultType.CREATE_ENTRY
|
||||
assert result2["title"] == "myStrom Device"
|
||||
assert result2["data"] == {"host": "1.1.1.1"}
|
||||
|
||||
|
||||
async def test_form_duplicates(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test abort on duplicate."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"pymystrom.get_device_info",
|
||||
return_value={"type": 101, "mac": DEVICE_MAC},
|
||||
) as mock_session:
|
||||
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"] == FlowResultType.ABORT
|
||||
assert result2["reason"] == "already_configured"
|
||||
|
||||
mock_session.assert_called_once()
|
||||
|
||||
|
||||
async def test_step_import(hass: HomeAssistant) -> None:
|
||||
"""Test the import step."""
|
||||
conf = {
|
||||
CONF_HOST: "1.1.1.1",
|
||||
}
|
||||
with patch("pymystrom.switch.MyStromSwitch.get_state"), patch(
|
||||
"pymystrom.get_device_info",
|
||||
return_value={"type": 101, "mac": DEVICE_MAC},
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=conf
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "myStrom Device"
|
||||
assert result["data"] == {
|
||||
CONF_HOST: "1.1.1.1",
|
||||
}
|
||||
|
||||
|
||||
async def test_wong_answer_from_device(hass: HomeAssistant) -> None:
|
||||
"""Test handling of wrong answers from the device."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {}
|
||||
with patch(
|
||||
"pymystrom.get_device_info",
|
||||
side_effect=MyStromConnectionError(),
|
||||
):
|
||||
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"] == FlowResultType.FORM
|
||||
assert result2["step_id"] == "user"
|
||||
assert result2["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
with patch(
|
||||
"pymystrom.get_device_info",
|
||||
return_value={"type": 101, "mac": DEVICE_MAC},
|
||||
):
|
||||
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"] == FlowResultType.CREATE_ENTRY
|
||||
assert result2["title"] == "myStrom Device"
|
||||
assert result2["data"] == {"host": "1.1.1.1"}
|
||||
132
tests/components/mystrom/test_init.py
Normal file
132
tests/components/mystrom/test_init.py
Normal file
@ -0,0 +1,132 @@
|
||||
"""Test the myStrom init."""
|
||||
from unittest.mock import AsyncMock, PropertyMock, patch
|
||||
|
||||
from pymystrom.exceptions import MyStromConnectionError
|
||||
|
||||
from homeassistant.components.mystrom.const import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .conftest import DEVICE_MAC
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def init_integration(
|
||||
hass: HomeAssistant,
|
||||
config_entry: MockConfigEntry,
|
||||
device_type: int,
|
||||
bulb_type: str = "strip",
|
||||
) -> None:
|
||||
"""Inititialize integration for testing."""
|
||||
with patch(
|
||||
"pymystrom.get_device_info",
|
||||
side_effect=AsyncMock(return_value={"type": device_type, "mac": DEVICE_MAC}),
|
||||
), patch("pymystrom.switch.MyStromSwitch.get_state", return_value={}), patch(
|
||||
"pymystrom.bulb.MyStromBulb.get_state", return_value={}
|
||||
), patch(
|
||||
"pymystrom.bulb.MyStromBulb.bulb_type", bulb_type
|
||||
), patch(
|
||||
"pymystrom.switch.MyStromSwitch.mac",
|
||||
new_callable=PropertyMock,
|
||||
return_value=DEVICE_MAC,
|
||||
), patch(
|
||||
"pymystrom.bulb.MyStromBulb.mac",
|
||||
new_callable=PropertyMock,
|
||||
return_value=DEVICE_MAC,
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert config_entry.state == ConfigEntryState.LOADED
|
||||
|
||||
|
||||
async def test_init_switch_and_unload(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test the initialization of a myStrom switch."""
|
||||
await init_integration(hass, config_entry, 101)
|
||||
state = hass.states.get("switch.mystrom_device")
|
||||
assert state is not None
|
||||
assert config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.config_entries.async_unload(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert config_entry.state is ConfigEntryState.NOT_LOADED
|
||||
assert not hass.data.get(DOMAIN)
|
||||
|
||||
|
||||
async def test_init_bulb(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
|
||||
"""Test the initialization of a myStrom bulb."""
|
||||
await init_integration(hass, config_entry, 102)
|
||||
state = hass.states.get("light.mystrom_device")
|
||||
assert state is not None
|
||||
assert config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
|
||||
async def test_init_of_unknown_bulb(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test the initialization of a unknown myStrom bulb."""
|
||||
with patch(
|
||||
"pymystrom.get_device_info",
|
||||
side_effect=AsyncMock(return_value={"type": 102, "mac": DEVICE_MAC}),
|
||||
), patch("pymystrom.bulb.MyStromBulb.get_state", return_value={}), patch(
|
||||
"pymystrom.bulb.MyStromBulb.bulb_type", "new_type"
|
||||
), patch(
|
||||
"pymystrom.bulb.MyStromBulb.mac",
|
||||
new_callable=PropertyMock,
|
||||
return_value=DEVICE_MAC,
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert config_entry.state == ConfigEntryState.SETUP_ERROR
|
||||
|
||||
|
||||
async def test_init_of_unknown_device(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test the initialization of a unsupported myStrom device."""
|
||||
with patch(
|
||||
"pymystrom.get_device_info",
|
||||
side_effect=AsyncMock(return_value={"type": 103, "mac": DEVICE_MAC}),
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert config_entry.state == ConfigEntryState.SETUP_ERROR
|
||||
|
||||
|
||||
async def test_init_cannot_connect_because_of_device_info(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test error handling for failing get_device_info."""
|
||||
with patch(
|
||||
"pymystrom.get_device_info",
|
||||
side_effect=MyStromConnectionError(),
|
||||
), patch("pymystrom.switch.MyStromSwitch.get_state", return_value={}), patch(
|
||||
"pymystrom.bulb.MyStromBulb.get_state", return_value={}
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert config_entry.state == ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_init_cannot_connect_because_of_get_state(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test error handling for failing get_state."""
|
||||
with patch(
|
||||
"pymystrom.get_device_info",
|
||||
side_effect=AsyncMock(return_value={"type": 101, "mac": DEVICE_MAC}),
|
||||
), patch(
|
||||
"pymystrom.switch.MyStromSwitch.get_state", side_effect=MyStromConnectionError()
|
||||
), patch(
|
||||
"pymystrom.bulb.MyStromBulb.get_state", side_effect=MyStromConnectionError()
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert config_entry.state == ConfigEntryState.SETUP_ERROR
|
||||
Loading…
x
Reference in New Issue
Block a user