Add BleBox integration (#32664)

Co-Authored-By: J. Nick Koston <nick@koston.org>
This commit is contained in:
gadgetmobile 2020-05-05 11:29:58 +02:00 committed by GitHub
parent 86d410d863
commit 8e071b69f4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 1176 additions and 0 deletions

View file

@ -53,6 +53,7 @@ homeassistant/components/azure_service_bus/* @hfurubotten
homeassistant/components/beewi_smartclim/* @alemuro
homeassistant/components/bitcoin/* @fabaff
homeassistant/components/bizkaibus/* @UgaitzEtxebarria
homeassistant/components/blebox/* @gadgetmobile
homeassistant/components/blink/* @fronzbot
homeassistant/components/bmp280/* @belidzs
homeassistant/components/bmw_connected_drive/* @gerard33

View file

@ -0,0 +1,121 @@
"""The BleBox devices integration."""
import asyncio
import logging
from blebox_uniapi.error import Error
from blebox_uniapi.products import Products
from blebox_uniapi.session import ApiHost
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity import Entity
from .const import DEFAULT_SETUP_TIMEOUT, DOMAIN, PRODUCT
_LOGGER = logging.getLogger(__name__)
PLATFORMS = ["cover"]
PARALLEL_UPDATES = 0
async def async_setup(hass: HomeAssistant, config: dict):
"""Set up the BleBox devices component."""
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up BleBox devices from a config entry."""
websession = async_get_clientsession(hass)
host = entry.data[CONF_HOST]
port = entry.data[CONF_PORT]
timeout = DEFAULT_SETUP_TIMEOUT
api_host = ApiHost(host, port, timeout, websession, hass.loop)
try:
product = await Products.async_from_host(api_host)
except Error as ex:
_LOGGER.error("Identify failed at %s:%d (%s)", api_host.host, api_host.port, ex)
raise ConfigEntryNotReady from ex
domain = hass.data.setdefault(DOMAIN, {})
domain_entry = domain.setdefault(entry.entry_id, {})
product = domain_entry.setdefault(PRODUCT, product)
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, platform)
for platform in PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
@callback
def create_blebox_entities(product, async_add, entity_klass, entity_type):
"""Create entities from a BleBox product's features."""
entities = []
for feature in product.features[entity_type]:
entities.append(entity_klass(feature))
async_add(entities, True)
class BleBoxEntity(Entity):
"""Implements a common class for entities representing a BleBox feature."""
def __init__(self, feature):
"""Initialize a BleBox entity."""
self._feature = feature
@property
def name(self):
"""Return the internal entity name."""
return self._feature.full_name
@property
def unique_id(self):
"""Return a unique id."""
return self._feature.unique_id
async def async_update(self):
"""Update the entity state."""
try:
await self._feature.async_update()
except Error as ex:
_LOGGER.error("Updating '%s' failed: %s", self.name, ex)
@property
def device_info(self):
"""Return device information for this entity."""
product = self._feature.product
return {
"identifiers": {(DOMAIN, product.unique_id)},
"name": product.name,
"manufacturer": product.brand,
"model": product.model,
"sw_version": product.firmware_version,
}

View file

@ -0,0 +1,128 @@
"""Config flow for BleBox devices integration."""
import logging
from blebox_uniapi.error import Error, UnsupportedBoxVersion
from blebox_uniapi.products import Products
from blebox_uniapi.session import ApiHost
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import (
ADDRESS_ALREADY_CONFIGURED,
CANNOT_CONNECT,
DEFAULT_HOST,
DEFAULT_PORT,
DEFAULT_SETUP_TIMEOUT,
DOMAIN,
UNKNOWN,
UNSUPPORTED_VERSION,
)
_LOGGER = logging.getLogger(__name__)
def host_port(data):
"""Return a list with host and port."""
return (data[CONF_HOST], data[CONF_PORT])
def create_schema(previous_input=None):
"""Create a schema with given values as default."""
if previous_input is not None:
host, port = host_port(previous_input)
else:
host = DEFAULT_HOST
port = DEFAULT_PORT
return vol.Schema(
{
vol.Required(CONF_HOST, default=host): str,
vol.Required(CONF_PORT, default=port): int,
}
)
LOG_MSG = {
UNSUPPORTED_VERSION: "Outdated firmware",
CANNOT_CONNECT: "Failed to identify device",
UNKNOWN: "Unknown error while identifying device",
}
class BleBoxConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for BleBox devices."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
"""Initialize the BleBox config flow."""
self.device_config = {}
def handle_step_exception(
self, step, exception, schema, host, port, message_id, log_fn
):
"""Handle step exceptions."""
log_fn("%s at %s:%d (%s)", LOG_MSG[message_id], host, port, exception)
return self.async_show_form(
step_id="user",
data_schema=schema,
errors={"base": message_id},
description_placeholders={"address": f"{host}:{port}"},
)
async def async_step_user(self, user_input=None):
"""Handle initial user-triggered config step."""
hass = self.hass
schema = create_schema(user_input)
if user_input is None:
return self.async_show_form(
step_id="user",
data_schema=schema,
errors={},
description_placeholders={},
)
addr = host_port(user_input)
for entry in hass.config_entries.async_entries(DOMAIN):
if addr == host_port(entry.data):
host, port = addr
return self.async_abort(
reason=ADDRESS_ALREADY_CONFIGURED,
description_placeholders={"address": f"{host}:{port}"},
)
websession = async_get_clientsession(hass)
api_host = ApiHost(*addr, DEFAULT_SETUP_TIMEOUT, websession, hass.loop, _LOGGER)
try:
product = await Products.async_from_host(api_host)
except UnsupportedBoxVersion as ex:
return self.handle_step_exception(
"user", ex, schema, *addr, UNSUPPORTED_VERSION, _LOGGER.debug
)
except Error as ex:
return self.handle_step_exception(
"user", ex, schema, *addr, CANNOT_CONNECT, _LOGGER.warning
)
except RuntimeError as ex:
return self.handle_step_exception(
"user", ex, schema, *addr, UNKNOWN, _LOGGER.error
)
# Check if configured but IP changed since
await self.async_set_unique_id(product.unique_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(title=product.name, data=user_input)

View file

@ -0,0 +1,45 @@
"""Constants for the BleBox devices integration."""
from homeassistant.components.cover import (
DEVICE_CLASS_DOOR,
DEVICE_CLASS_GATE,
DEVICE_CLASS_SHUTTER,
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_OPENING,
)
DOMAIN = "blebox"
PRODUCT = "product"
DEFAULT_SETUP_TIMEOUT = 3
# translation strings
ADDRESS_ALREADY_CONFIGURED = "address_already_configured"
CANNOT_CONNECT = "cannot_connect"
UNSUPPORTED_VERSION = "unsupported_version"
UNKNOWN = "unknown"
BLEBOX_TO_HASS_DEVICE_CLASSES = {
"shutter": DEVICE_CLASS_SHUTTER,
"gatebox": DEVICE_CLASS_DOOR,
"gate": DEVICE_CLASS_GATE,
}
BLEBOX_TO_HASS_COVER_STATES = {
None: None,
0: STATE_CLOSING, # moving down
1: STATE_OPENING, # moving up
2: STATE_OPEN, # manually stopped
3: STATE_CLOSED, # lower limit
4: STATE_OPEN, # upper limit / open
# gateController
5: STATE_OPEN, # overload
6: STATE_OPEN, # motor failure
# 7 is not used
8: STATE_OPEN, # safety stop
}
DEFAULT_HOST = "192.168.0.2"
DEFAULT_PORT = 80

View file

@ -0,0 +1,97 @@
"""BleBox cover entity."""
from homeassistant.components.cover import (
ATTR_POSITION,
STATE_CLOSED,
STATE_CLOSING,
STATE_OPENING,
SUPPORT_CLOSE,
SUPPORT_OPEN,
SUPPORT_SET_POSITION,
SUPPORT_STOP,
CoverEntity,
)
from . import BleBoxEntity, create_blebox_entities
from .const import (
BLEBOX_TO_HASS_COVER_STATES,
BLEBOX_TO_HASS_DEVICE_CLASSES,
DOMAIN,
PRODUCT,
)
async def async_setup_entry(hass, config_entry, async_add):
"""Set up a BleBox entry."""
product = hass.data[DOMAIN][config_entry.entry_id][PRODUCT]
create_blebox_entities(product, async_add, BleBoxCoverEntity, "covers")
return True
class BleBoxCoverEntity(BleBoxEntity, CoverEntity):
"""Representation of a BleBox cover feature."""
@property
def state(self):
"""Return the equivalent HA cover state."""
return BLEBOX_TO_HASS_COVER_STATES[self._feature.state]
@property
def device_class(self):
"""Return the device class."""
return BLEBOX_TO_HASS_DEVICE_CLASSES[self._feature.device_class]
@property
def supported_features(self):
"""Return the supported cover features."""
position = SUPPORT_SET_POSITION if self._feature.is_slider else 0
stop = SUPPORT_STOP if self._feature.has_stop else 0
return position | stop | SUPPORT_OPEN | SUPPORT_CLOSE
@property
def current_cover_position(self):
"""Return the current cover position."""
position = self._feature.current
if position == -1: # possible for shutterBox
return None
return None if position is None else 100 - position
@property
def is_opening(self):
"""Return whether cover is opening."""
return self._is_state(STATE_OPENING)
@property
def is_closing(self):
"""Return whether cover is closing."""
return self._is_state(STATE_CLOSING)
@property
def is_closed(self):
"""Return whether cover is closed."""
return self._is_state(STATE_CLOSED)
async def async_open_cover(self, **kwargs):
"""Open the cover position."""
await self._feature.async_open()
async def async_close_cover(self, **kwargs):
"""Close the cover position."""
await self._feature.async_close()
async def async_set_cover_position(self, **kwargs):
"""Set the cover position."""
position = kwargs[ATTR_POSITION]
await self._feature.async_set_position(100 - position)
async def async_stop_cover(self, **kwargs):
"""Stop the cover."""
await self._feature.async_stop()
def _is_state(self, state_name):
value = self.state
return None if value is None else value == state_name

View file

@ -0,0 +1,8 @@
{
"domain": "blebox",
"name": "BleBox devices",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/blebox",
"requirements": ["blebox_uniapi==1.3.2"],
"codeowners": [ "@gadgetmobile" ]
}

View file

@ -0,0 +1,24 @@
{
"config": {
"abort": {
"already_configured": "This BleBox device is already configured.",
"address_already_configured": "A BleBox device is already configured at {address}."
},
"error": {
"cannot_connect": "Unable to connect to the BleBox device. (Check the logs for errors.)",
"unsupported_version": "BleBox device has outdated firmware. Please upgrade it first.",
"unknown": "Unknown error while connecting to the BleBox device. (Check the logs for errors.)"
},
"flow_title": "BleBox device: {name} ({host})",
"step": {
"user": {
"description": "Set up your BleBox to integrate with Home Assistant.",
"data": {
"host": "IP address",
"port": "Port"
},
"title": "Set up your BleBox device"
}
}
}
}

View file

@ -0,0 +1,24 @@
{
"config": {
"abort": {
"already_configured": "This BleBox device is already configured.",
"address_already_configured": "A BleBox device is already configured at {address}."
},
"error": {
"cannot_connect": "Unable to connect to the BleBox device. (Check the logs for errors.)",
"unsupported_version": "BleBox device has outdated firmware. Please upgrade it first.",
"unknown": "Unknown error while connecting to the BleBox device. (Check the logs for errors.)"
},
"flow_title": "BleBox device: {name} ({host})",
"step": {
"user": {
"description": "Set up your BleBox to integrate with Home Assistant.",
"data": {
"host": "IP address",
"port": "Port"
},
"title": "Set up your BleBox device"
}
}
}
}

View file

@ -16,6 +16,7 @@ FLOWS = [
"atag",
"august",
"axis",
"blebox",
"braviatv",
"brother",
"cast",

View file

@ -335,6 +335,9 @@ bimmer_connected==0.7.5
# homeassistant.components.bizkaibus
bizkaibus==0.1.1
# homeassistant.components.blebox
blebox_uniapi==1.3.2
# homeassistant.components.blink
blinkpy==0.14.3

View file

@ -143,6 +143,9 @@ base36==0.1.1
# homeassistant.components.zha
bellows-homeassistant==0.15.2
# homeassistant.components.blebox
blebox_uniapi==1.3.2
# homeassistant.components.bom
bomradarloop==0.1.4

View file

@ -0,0 +1 @@
"""Tests for the blebox component."""

View file

@ -0,0 +1,87 @@
"""PyTest fixtures and test helpers."""
from unittest import mock
import blebox_uniapi
import pytest
from homeassistant.components.blebox.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.setup import async_setup_component
from tests.async_mock import AsyncMock, PropertyMock, patch
from tests.common import MockConfigEntry
def patch_product_identify(path=None, **kwargs):
"""Patch the blebox_uniapi Products class."""
if path is None:
path = "homeassistant.components.blebox.Products"
patcher = patch(path, mock.DEFAULT, blebox_uniapi.products.Products, True, True)
products_class = patcher.start()
products_class.async_from_host = AsyncMock(**kwargs)
return products_class
def setup_product_mock(category, feature_mocks, path=None):
"""Mock a product returning the given features."""
product_mock = mock.create_autospec(
blebox_uniapi.box.Box, True, True, features=None
)
type(product_mock).features = PropertyMock(return_value={category: feature_mocks})
for feature in feature_mocks:
type(feature).product = PropertyMock(return_value=product_mock)
patch_product_identify(path, return_value=product_mock)
return product_mock
def mock_only_feature(spec, **kwargs):
"""Mock just the feature, without the product setup."""
return mock.create_autospec(spec, True, True, **kwargs)
def mock_feature(category, spec, **kwargs):
"""Mock a feature along with whole product setup."""
feature_mock = mock_only_feature(spec, **kwargs)
feature_mock.async_update = AsyncMock()
product = setup_product_mock(category, [feature_mock])
type(feature_mock.product).name = PropertyMock(return_value="Some name")
type(feature_mock.product).type = PropertyMock(return_value="some type")
type(feature_mock.product).model = PropertyMock(return_value="some model")
type(feature_mock.product).brand = PropertyMock(return_value="BleBox")
type(feature_mock.product).firmware_version = PropertyMock(return_value="1.23")
type(feature_mock.product).unique_id = PropertyMock(return_value="abcd0123ef5678")
type(feature_mock).product = PropertyMock(return_value=product)
return feature_mock
def mock_config(ip_address="172.100.123.4"):
"""Return a Mock of the HA entity config."""
return MockConfigEntry(domain=DOMAIN, data={CONF_HOST: ip_address, CONF_PORT: 80})
@pytest.fixture(name="config")
def config_fixture():
"""Create hass config fixture."""
return {DOMAIN: {CONF_HOST: "172.100.123.4", CONF_PORT: 80}}
@pytest.fixture
def wrapper(request):
"""Return an entity wrapper from given fixture name."""
return request.getfixturevalue(request.param)
async def async_setup_entity(hass, config, entity_id):
"""Return a configured entity with the given entity_id."""
config_entry = mock_config()
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
entity_registry = await hass.helpers.entity_registry.async_get_registry()
return entity_registry.async_get(entity_id)

View file

@ -0,0 +1,192 @@
"""Test Home Assistant config flow for BleBox devices."""
import blebox_uniapi
import pytest
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.blebox import config_flow
from homeassistant.setup import async_setup_component
from .conftest import mock_config, mock_only_feature, setup_product_mock
from tests.async_mock import DEFAULT, AsyncMock, PropertyMock, patch
def create_valid_feature_mock(path="homeassistant.components.blebox.Products"):
"""Return a valid, complete BleBox feature mock."""
feature = mock_only_feature(
blebox_uniapi.cover.Cover,
unique_id="BleBox-gateBox-1afe34db9437-0.position",
full_name="gateBox-0.position",
device_class="gate",
state=0,
async_update=AsyncMock(),
current=None,
)
product = setup_product_mock("covers", [feature], path)
type(product).name = PropertyMock(return_value="My gate controller")
type(product).model = PropertyMock(return_value="gateController")
type(product).type = PropertyMock(return_value="gateBox")
type(product).brand = PropertyMock(return_value="BleBox")
type(product).firmware_version = PropertyMock(return_value="1.23")
type(product).unique_id = PropertyMock(return_value="abcd0123ef5678")
return feature
@pytest.fixture
def valid_feature_mock():
"""Return a valid, complete BleBox feature mock."""
return create_valid_feature_mock()
@pytest.fixture
def flow_feature_mock():
"""Return a mocked user flow feature."""
return create_valid_feature_mock(
"homeassistant.components.blebox.config_flow.Products"
)
async def test_flow_works(hass, flow_feature_mock):
"""Test that config flow works."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={config_flow.CONF_HOST: "172.2.3.4", config_flow.CONF_PORT: 80},
)
assert result["type"] == "create_entry"
assert result["title"] == "My gate controller"
assert result["data"] == {
config_flow.CONF_HOST: "172.2.3.4",
config_flow.CONF_PORT: 80,
}
@pytest.fixture
def product_class_mock():
"""Return a mocked feature."""
path = "homeassistant.components.blebox.config_flow.Products"
patcher = patch(path, DEFAULT, blebox_uniapi.products.Products, True, True)
yield patcher
async def test_flow_with_connection_failure(hass, product_class_mock):
"""Test that config flow works."""
with product_class_mock as products_class:
products_class.async_from_host = AsyncMock(
side_effect=blebox_uniapi.error.ConnectionError
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={config_flow.CONF_HOST: "172.2.3.4", config_flow.CONF_PORT: 80},
)
assert result["errors"] == {"base": "cannot_connect"}
async def test_flow_with_api_failure(hass, product_class_mock):
"""Test that config flow works."""
with product_class_mock as products_class:
products_class.async_from_host = AsyncMock(
side_effect=blebox_uniapi.error.Error
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={config_flow.CONF_HOST: "172.2.3.4", config_flow.CONF_PORT: 80},
)
assert result["errors"] == {"base": "cannot_connect"}
async def test_flow_with_unknown_failure(hass, product_class_mock):
"""Test that config flow works."""
with product_class_mock as products_class:
products_class.async_from_host = AsyncMock(side_effect=RuntimeError)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={config_flow.CONF_HOST: "172.2.3.4", config_flow.CONF_PORT: 80},
)
assert result["errors"] == {"base": "unknown"}
async def test_flow_with_unsupported_version(hass, product_class_mock):
"""Test that config flow works."""
with product_class_mock as products_class:
products_class.async_from_host = AsyncMock(
side_effect=blebox_uniapi.error.UnsupportedBoxVersion
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={config_flow.CONF_HOST: "172.2.3.4", config_flow.CONF_PORT: 80},
)
assert result["errors"] == {"base": "unsupported_version"}
async def test_async_setup(hass):
"""Test async_setup (for coverage)."""
assert await async_setup_component(hass, "blebox", {"host": "172.2.3.4"})
await hass.async_block_till_done()
async def test_already_configured(hass, valid_feature_mock):
"""Test that same device cannot be added twice."""
config = mock_config("172.2.3.4")
config.add_to_hass(hass)
await hass.config_entries.async_setup(config.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={config_flow.CONF_HOST: "172.2.3.4", config_flow.CONF_PORT: 80},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "address_already_configured"
async def test_async_setup_entry(hass, valid_feature_mock):
"""Test async_setup_entry (for coverage)."""
config = mock_config()
config.add_to_hass(hass)
assert await hass.config_entries.async_setup(config.entry_id)
await hass.async_block_till_done()
assert hass.config_entries.async_entries() == [config]
assert config.state == config_entries.ENTRY_STATE_LOADED
async def test_async_remove_entry(hass, valid_feature_mock):
"""Test async_setup_entry (for coverage)."""
config = mock_config()
config.add_to_hass(hass)
assert await hass.config_entries.async_setup(config.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_remove(config.entry_id)
await hass.async_block_till_done()
assert hass.config_entries.async_entries() == []
assert config.state == config_entries.ENTRY_STATE_NOT_LOADED

View file

@ -0,0 +1,381 @@
"""BleBox cover entities tests."""
import blebox_uniapi
import pytest
from homeassistant.components.cover import (
ATTR_CURRENT_POSITION,
ATTR_POSITION,
DEVICE_CLASS_DOOR,
DEVICE_CLASS_GATE,
DEVICE_CLASS_SHUTTER,
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_OPENING,
SUPPORT_CLOSE,
SUPPORT_OPEN,
SUPPORT_SET_POSITION,
SUPPORT_STOP,
)
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_SUPPORTED_FEATURES,
SERVICE_CLOSE_COVER,
SERVICE_OPEN_COVER,
SERVICE_SET_COVER_POSITION,
SERVICE_STOP_COVER,
STATE_UNKNOWN,
)
from .conftest import async_setup_entity, mock_feature
from tests.async_mock import ANY, AsyncMock, PropertyMock, call, patch
ALL_COVER_FIXTURES = ["gatecontroller", "shutterbox", "gatebox"]
FIXTURES_SUPPORTING_STOP = ["gatecontroller", "shutterbox"]
@pytest.fixture(name="shutterbox")
def shutterbox_fixture():
"""Return a shutterBox fixture."""
feature = mock_feature(
"covers",
blebox_uniapi.cover.Cover,
unique_id="BleBox-shutterBox-2bee34e750b8-position",
full_name="shutterBox-position",
device_class="shutter",
current=None,
state=None,
has_stop=True,
is_slider=True,
)
product = feature.product
type(product).name = PropertyMock(return_value="My shutter")
type(product).model = PropertyMock(return_value="shutterBox")
return (feature, "cover.shutterbox_position")
@pytest.fixture(name="gatebox")
def gatebox_fixture():
"""Return a gateBox fixture."""
feature = mock_feature(
"covers",
blebox_uniapi.cover.Cover,
unique_id="BleBox-gateBox-1afe34db9437-position",
device_class="gatebox",
full_name="gateBox-position",
current=None,
state=None,
has_stop=False,
is_slider=False,
)
product = feature.product
type(product).name = PropertyMock(return_value="My gatebox")
type(product).model = PropertyMock(return_value="gateBox")
return (feature, "cover.gatebox_position")
@pytest.fixture(name="gatecontroller")
def gate_fixture():
"""Return a gateController fixture."""
feature = mock_feature(
"covers",
blebox_uniapi.cover.Cover,
unique_id="BleBox-gateController-2bee34e750b8-position",
full_name="gateController-position",
device_class="gate",
current=None,
state=None,
has_stop=True,
is_slider=True,
)
product = feature.product
type(product).name = PropertyMock(return_value="My gate controller")
type(product).model = PropertyMock(return_value="gateController")
return (feature, "cover.gatecontroller_position")
async def test_init_gatecontroller(gatecontroller, hass, config):
"""Test gateController default state."""
_, entity_id = gatecontroller
entry = await async_setup_entity(hass, config, entity_id)
assert entry.unique_id == "BleBox-gateController-2bee34e750b8-position"
state = hass.states.get(entity_id)
assert state.name == "gateController-position"
assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_GATE
supported_features = state.attributes[ATTR_SUPPORTED_FEATURES]
assert supported_features & SUPPORT_OPEN
assert supported_features & SUPPORT_CLOSE
assert supported_features & SUPPORT_STOP
assert supported_features & SUPPORT_SET_POSITION
assert ATTR_CURRENT_POSITION not in state.attributes
assert state.state == STATE_UNKNOWN
device_registry = await hass.helpers.device_registry.async_get_registry()
device = device_registry.async_get(entry.device_id)
assert device.name == "My gate controller"
assert device.identifiers == {("blebox", "abcd0123ef5678")}
assert device.manufacturer == "BleBox"
assert device.model == "gateController"
assert device.sw_version == "1.23"
async def test_init_shutterbox(shutterbox, hass, config):
"""Test gateBox default state."""
_, entity_id = shutterbox
entry = await async_setup_entity(hass, config, entity_id)
assert entry.unique_id == "BleBox-shutterBox-2bee34e750b8-position"
state = hass.states.get(entity_id)
assert state.name == "shutterBox-position"
assert entry.device_class == DEVICE_CLASS_SHUTTER
supported_features = state.attributes[ATTR_SUPPORTED_FEATURES]
assert supported_features & SUPPORT_OPEN
assert supported_features & SUPPORT_CLOSE
assert supported_features & SUPPORT_STOP
assert supported_features & SUPPORT_SET_POSITION
assert ATTR_CURRENT_POSITION not in state.attributes
assert state.state == STATE_UNKNOWN
device_registry = await hass.helpers.device_registry.async_get_registry()
device = device_registry.async_get(entry.device_id)
assert device.name == "My shutter"
assert device.identifiers == {("blebox", "abcd0123ef5678")}
assert device.manufacturer == "BleBox"
assert device.model == "shutterBox"
assert device.sw_version == "1.23"
async def test_init_gatebox(gatebox, hass, config):
"""Test cover default state."""
_, entity_id = gatebox
entry = await async_setup_entity(hass, config, entity_id)
assert entry.unique_id == "BleBox-gateBox-1afe34db9437-position"
state = hass.states.get(entity_id)
assert state.name == "gateBox-position"
assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_DOOR
supported_features = state.attributes[ATTR_SUPPORTED_FEATURES]
assert supported_features & SUPPORT_OPEN
assert supported_features & SUPPORT_CLOSE
# Not available during init since requires fetching state to detect
assert not supported_features & SUPPORT_STOP
assert not supported_features & SUPPORT_SET_POSITION
assert ATTR_CURRENT_POSITION not in state.attributes
assert state.state == STATE_UNKNOWN
device_registry = await hass.helpers.device_registry.async_get_registry()
device = device_registry.async_get(entry.device_id)
assert device.name == "My gatebox"
assert device.identifiers == {("blebox", "abcd0123ef5678")}
assert device.manufacturer == "BleBox"
assert device.model == "gateBox"
assert device.sw_version == "1.23"
@pytest.mark.parametrize("wrapper", ALL_COVER_FIXTURES, indirect=["wrapper"])
async def test_open(wrapper, hass, config):
"""Test cover opening."""
feature_mock, entity_id = wrapper
def initial_update():
feature_mock.state = 3 # manually stopped
def open_gate():
feature_mock.state = 1 # opening
feature_mock.async_update = AsyncMock(side_effect=initial_update)
feature_mock.async_open = AsyncMock(side_effect=open_gate)
await async_setup_entity(hass, config, entity_id)
assert hass.states.get(entity_id).state == STATE_CLOSED
feature_mock.async_update = AsyncMock()
await hass.services.async_call(
"cover", SERVICE_OPEN_COVER, {"entity_id": entity_id}, blocking=True,
)
assert hass.states.get(entity_id).state == STATE_OPENING
@pytest.mark.parametrize("wrapper", ALL_COVER_FIXTURES, indirect=["wrapper"])
async def test_close(wrapper, hass, config):
"""Test cover closing."""
feature_mock, entity_id = wrapper
def initial_update():
feature_mock.state = 4 # open
def close():
feature_mock.state = 0 # closing
feature_mock.async_update = AsyncMock(side_effect=initial_update)
feature_mock.async_close = AsyncMock(side_effect=close)
await async_setup_entity(hass, config, entity_id)
assert hass.states.get(entity_id).state == STATE_OPEN
feature_mock.async_update = AsyncMock()
await hass.services.async_call(
"cover", SERVICE_CLOSE_COVER, {"entity_id": entity_id}, blocking=True
)
assert hass.states.get(entity_id).state == STATE_CLOSING
def opening_to_stop_feature_mock(feature_mock):
"""Return an mocked feature which can be updated and stopped."""
def initial_update():
feature_mock.state = 1 # opening
def stop():
feature_mock.state = 2 # manually stopped
feature_mock.async_update = AsyncMock(side_effect=initial_update)
feature_mock.async_stop = AsyncMock(side_effect=stop)
@pytest.mark.parametrize("wrapper", FIXTURES_SUPPORTING_STOP, indirect=["wrapper"])
async def test_stop(wrapper, hass, config):
"""Test cover stopping."""
feature_mock, entity_id = wrapper
opening_to_stop_feature_mock(feature_mock)
await async_setup_entity(hass, config, entity_id)
assert hass.states.get(entity_id).state == STATE_OPENING
feature_mock.async_update = AsyncMock()
await hass.services.async_call(
"cover", SERVICE_STOP_COVER, {"entity_id": entity_id}, blocking=True
)
assert hass.states.get(entity_id).state == STATE_OPEN
@pytest.mark.parametrize("wrapper", ALL_COVER_FIXTURES, indirect=["wrapper"])
async def test_update(wrapper, hass, config):
"""Test cover updating."""
feature_mock, entity_id = wrapper
def initial_update():
feature_mock.current = 29 # inverted
feature_mock.state = 2 # manually stopped
feature_mock.async_update = AsyncMock(side_effect=initial_update)
await async_setup_entity(hass, config, entity_id)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_CURRENT_POSITION] == 71 # 100 - 29
assert state.state == STATE_OPEN
@pytest.mark.parametrize(
"wrapper", ["gatecontroller", "shutterbox"], indirect=["wrapper"]
)
async def test_set_position(wrapper, hass, config):
"""Test cover position setting."""
feature_mock, entity_id = wrapper
def initial_update():
feature_mock.state = 3 # closed
def set_position(position):
assert position == 99 # inverted
feature_mock.state = 1 # opening
# feature_mock.current = position
feature_mock.async_update = AsyncMock(side_effect=initial_update)
feature_mock.async_set_position = AsyncMock(side_effect=set_position)
await async_setup_entity(hass, config, entity_id)
assert hass.states.get(entity_id).state == STATE_CLOSED
feature_mock.async_update = AsyncMock()
await hass.services.async_call(
"cover",
SERVICE_SET_COVER_POSITION,
{"entity_id": entity_id, ATTR_POSITION: 1},
blocking=True,
) # almost closed
assert hass.states.get(entity_id).state == STATE_OPENING
async def test_unknown_position(shutterbox, hass, config):
"""Test cover position setting."""
feature_mock, entity_id = shutterbox
def initial_update():
feature_mock.state = 4 # opening
feature_mock.current = -1
feature_mock.async_update = AsyncMock(side_effect=initial_update)
await async_setup_entity(hass, config, entity_id)
state = hass.states.get(entity_id)
assert state.state == STATE_OPEN
assert ATTR_CURRENT_POSITION not in state.attributes
async def test_with_stop(gatebox, hass, config):
"""Test stop capability is available."""
feature_mock, entity_id = gatebox
opening_to_stop_feature_mock(feature_mock)
feature_mock.has_stop = True
await async_setup_entity(hass, config, entity_id)
state = hass.states.get(entity_id)
supported_features = state.attributes[ATTR_SUPPORTED_FEATURES]
assert supported_features & SUPPORT_STOP
async def test_with_no_stop(gatebox, hass, config):
"""Test stop capability is not available."""
feature_mock, entity_id = gatebox
opening_to_stop_feature_mock(feature_mock)
feature_mock.has_stop = False
await async_setup_entity(hass, config, entity_id)
state = hass.states.get(entity_id)
supported_features = state.attributes[ATTR_SUPPORTED_FEATURES]
assert not supported_features & SUPPORT_STOP
@pytest.mark.parametrize("wrapper", ALL_COVER_FIXTURES, indirect=["wrapper"])
async def test_update_failure(wrapper, hass, config):
"""Test that update failures are logged."""
feature_mock, entity_id = wrapper
feature_mock.async_update = AsyncMock(side_effect=blebox_uniapi.error.ClientError)
name = feature_mock.full_name
with patch("homeassistant.components.blebox._LOGGER.error") as error:
await async_setup_entity(hass, config, entity_id)
error.assert_has_calls([call("Updating '%s' failed: %s", name, ANY)])
assert isinstance(error.call_args[0][2], blebox_uniapi.error.ClientError)

View file

@ -0,0 +1,60 @@
"""BleBox devices setup tests."""
import logging
import blebox_uniapi
from homeassistant.components.blebox.const import DOMAIN
from homeassistant.config_entries import ENTRY_STATE_NOT_LOADED, ENTRY_STATE_SETUP_RETRY
from .conftest import mock_config, patch_product_identify
async def test_setup_failure(hass, caplog):
"""Test that setup failure is handled and logged."""
patch_product_identify(None, side_effect=blebox_uniapi.error.ClientError)
entry = mock_config()
entry.add_to_hass(hass)
caplog.set_level(logging.ERROR)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert "Identify failed at 172.100.123.4:80 ()" in caplog.text
assert entry.state == ENTRY_STATE_SETUP_RETRY
async def test_setup_failure_on_connection(hass, caplog):
"""Test that setup failure is handled and logged."""
patch_product_identify(None, side_effect=blebox_uniapi.error.ConnectionError)
entry = mock_config()
entry.add_to_hass(hass)
caplog.set_level(logging.ERROR)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert "Identify failed at 172.100.123.4:80 ()" in caplog.text
assert entry.state == ENTRY_STATE_SETUP_RETRY
async def test_unload_config_entry(hass):
"""Test that unloading works properly."""
patch_product_identify(None)
entry = mock_config()
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.data[DOMAIN]
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert not hass.data.get(DOMAIN)
assert entry.state == ENTRY_STATE_NOT_LOADED