home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Update bsblan integration (#67399)

Browse Source 
* Update bsblan integration

Update the integration to current standards

* removed unused code

update coverage

* some cleanup

* fix conflicts due upstream changes

* fix prettier json files

* fix remove comment code

* use dataclass instead of tuple

* fix spelling

* Set as class attribute

main entity doesn't need to give own name

* fix requirements
This commit is contained in:
Willem-Jan van Rootselaar
2022-10-18 12:06:51 +02:00
committed by GitHub
parent c1213857ce
commit 1fe397f7d7
16 changed files with 588 additions and 434 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -161,6 +161,8 @@ omit =
homeassistant/components/brunt/const.py
homeassistant/components/brunt/cover.py
homeassistant/components/bsblan/climate.py
homeassistant/components/bsblan/const.py
homeassistant/components/bsblan/entity.py
homeassistant/components/bt_home_hub_5/device_tracker.py
homeassistant/components/bt_smarthub/device_tracker.py
homeassistant/components/buienradar/sensor.py

51
homeassistant/components/bsblan/__init__.py
View File

@ -1,7 +1,7 @@
"""The BSB-Lan integration."""
from datetime import timedelta
import dataclasses
from bsblan import BSBLan, BSBLanConnectionError
from bsblan import BSBLAN, Device, Info, State
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@ -12,21 +12,29 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import CONF_PASSKEY, DATA_BSBLAN_CLIENT, DOMAIN
SCAN_INTERVAL = timedelta(seconds=30)
from .const import CONF_PASSKEY, DOMAIN, LOGGER, SCAN_INTERVAL
PLATFORMS = [Platform.CLIMATE]
@dataclasses.dataclass
class HomeAssistantBSBLANData:
"""BSBLan data stored in the Home Assistant data object."""
coordinator: DataUpdateCoordinator[State]
client: BSBLAN
device: Device
info: Info
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up BSB-Lan from a config entry."""
session = async_get_clientsession(hass)
bsblan = BSBLan(
bsblan = BSBLAN(
entry.data[CONF_HOST],
passkey=entry.data[CONF_PASSKEY],
port=entry.data[CONF_PORT],
@ -35,13 +43,23 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
session=session,
)
try:
await bsblan.info()
except BSBLanConnectionError as exception:
raise ConfigEntryNotReady from exception
coordinator: DataUpdateCoordinator[State] = DataUpdateCoordinator(
hass,
LOGGER,
name=f"{DOMAIN}_{entry.data[CONF_HOST]}",
update_interval=SCAN_INTERVAL,
update_method=bsblan.state,
)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = {DATA_BSBLAN_CLIENT: bsblan}
device = await bsblan.device()
info = await bsblan.info()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = HomeAssistantBSBLANData(
client=bsblan,
coordinator=coordinator,
device=device,
info=info,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@ -49,13 +67,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload BSBLan config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
"""Unload BSBLAN config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
# Cleanup
del hass.data[DOMAIN][entry.entry_id]
if not hass.data[DOMAIN]:
del hass.data[DOMAIN]
return unload_ok

193
homeassistant/components/bsblan/climate.py
View File

@ -1,11 +1,9 @@
"""BSBLAN platform to control a compatible Climate Device."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from bsblan import BSBLan, BSBLanError, Info, State
from bsblan import BSBLAN, BSBLANError, Device, Info, State
from homeassistant.components.climate import (
ATTR_HVAC_MODE,
@ -19,15 +17,18 @@ from homeassistant.components.climate import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE, TEMP_CELSIUS, TEMP_FAHRENHEIT
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import ATTR_TARGET_TEMPERATURE, DATA_BSBLAN_CLIENT, DOMAIN
_LOGGER = logging.getLogger(__name__)
from . import HomeAssistantBSBLANData
from .const import ATTR_TARGET_TEMPERATURE, DOMAIN, LOGGER
from .entity import BSBLANEntity
PARALLEL_UPDATES = 1
SCAN_INTERVAL = timedelta(seconds=20)
HVAC_MODES = [
HVACMode.AUTO,
@ -40,130 +41,122 @@ PRESET_MODES = [
PRESET_NONE,
]
HA_STATE_TO_BSBLAN = {
HVACMode.AUTO: "1",
HVACMode.HEAT: "3",
HVACMode.OFF: "0",
}
BSBLAN_TO_HA_STATE = {value: key for key, value in HA_STATE_TO_BSBLAN.items()}
HA_PRESET_TO_BSBLAN = {
PRESET_ECO: "2",
}
BSBLAN_TO_HA_PRESET = {
2: PRESET_ECO,
}
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up BSBLan device based on a config entry."""
bsblan: BSBLan = hass.data[DOMAIN][entry.entry_id][DATA_BSBLAN_CLIENT]
info = await bsblan.info()
async_add_entities([BSBLanClimate(entry.entry_id, bsblan, info)], True)
"""Set up BSBLAN device based on a config entry."""
data: HomeAssistantBSBLANData = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
[
BSBLANClimate(
data.coordinator,
data.client,
data.device,
data.info,
entry,
)
],
True,
)
class BSBLanClimate(ClimateEntity):
"""Defines a BSBLan climate device."""
class BSBLANClimate(BSBLANEntity, CoordinatorEntity, ClimateEntity):
"""Defines a BSBLAN climate device."""
coordinator: DataUpdateCoordinator[State]
_attr_has_entity_name = True
# Determine preset modes
_attr_supported_features = (
ClimateEntityFeature.TARGET_TEMPERATURE | ClimateEntityFeature.PRESET_MODE
)
_attr_hvac_modes = HVAC_MODES
_attr_preset_modes = PRESET_MODES
# Determine hvac modes
_attr_hvac_modes = HVAC_MODES
def __init__(
self,
entry_id: str,
bsblan: BSBLan,
coordinator: DataUpdateCoordinator,
client: BSBLAN,
device: Device,
info: Info,
entry: ConfigEntry,
) -> None:
"""Initialize BSBLan climate device."""
self._attr_available = True
self._store_hvac_mode: HVACMode | str | None = None
self.bsblan = bsblan
self._attr_name = self._attr_unique_id = info.device_identification
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, info.device_identification)},
manufacturer="BSBLan",
model=info.controller_variant,
name="BSBLan Device",
"""Initialize BSBLAN climate device."""
super().__init__(client, device, info, entry)
CoordinatorEntity.__init__(self, coordinator)
self._attr_unique_id = f"{format_mac(device.MAC)}-climate"
self._attr_min_temp = float(self.coordinator.data.min_temp.value)
self._attr_max_temp = float(self.coordinator.data.max_temp.value)
self._attr_temperature_unit = (
TEMP_CELSIUS
if self.coordinator.data.current_temperature.unit == "°C"
else TEMP_FAHRENHEIT
)
@property
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return float(self.coordinator.data.current_temperature.value)
@property
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return float(self.coordinator.data.target_temperature.value)
@property
def hvac_mode(self) -> str:
"""Return hvac operation ie. heat, cool mode."""
if self.coordinator.data.hvac_mode.value == PRESET_ECO:
return HVACMode.AUTO
return self.coordinator.data.hvac_mode.value
@property
def preset_mode(self) -> str | None:
"""Return the current preset mode."""
if (
self.hvac_mode == HVACMode.AUTO
and self.coordinator.data.hvac_mode.value == PRESET_ECO
):
return PRESET_ECO
return PRESET_NONE
async def async_set_hvac_mode(self, hvac_mode: str) -> None:
"""Set hvac mode."""
await self.async_set_data(hvac_mode=hvac_mode)
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set preset mode."""
_LOGGER.debug("Setting preset mode to: %s", preset_mode)
if preset_mode == PRESET_NONE:
# restore previous hvac mode
self._attr_hvac_mode = self._store_hvac_mode
else:
# Store hvac mode.
self._store_hvac_mode = self._attr_hvac_mode
# only allow preset mode when hvac mode is auto
if self.hvac_mode == HVACMode.AUTO:
await self.async_set_data(preset_mode=preset_mode)
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set HVAC mode."""
_LOGGER.debug("Setting HVAC mode to: %s", hvac_mode)
# preset should be none when hvac mode is set
self._attr_preset_mode = PRESET_NONE
await self.async_set_data(hvac_mode=hvac_mode)
else:
LOGGER.error("Can't set preset mode when hvac mode is not auto")
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperatures."""
await self.async_set_data(**kwargs)
async def async_set_data(self, **kwargs: Any) -> None:
"""Set device settings using BSBLan."""
"""Set device settings using BSBLAN."""
data = {}
if ATTR_TEMPERATURE in kwargs:
data[ATTR_TARGET_TEMPERATURE] = kwargs[ATTR_TEMPERATURE]
_LOGGER.debug("Set temperature data = %s", data)
if ATTR_HVAC_MODE in kwargs:
data[ATTR_HVAC_MODE] = HA_STATE_TO_BSBLAN[kwargs[ATTR_HVAC_MODE]]
_LOGGER.debug("Set hvac mode data = %s", data)
data[ATTR_HVAC_MODE] = kwargs[ATTR_HVAC_MODE]
if ATTR_PRESET_MODE in kwargs:
# for now we set the preset as hvac_mode as the api expect this
data[ATTR_HVAC_MODE] = HA_PRESET_TO_BSBLAN[kwargs[ATTR_PRESET_MODE]]
# If preset mode is None, set hvac to auto
if kwargs[ATTR_PRESET_MODE] == PRESET_NONE:
data[ATTR_HVAC_MODE] = HVACMode.AUTO
else:
data[ATTR_HVAC_MODE] = kwargs[ATTR_PRESET_MODE]
try:
await self.bsblan.thermostat(**data)
except BSBLanError:
_LOGGER.error("An error occurred while updating the BSBLan device")
self._attr_available = False
async def async_update(self) -> None:
"""Update BSBlan entity."""
try:
state: State = await self.bsblan.state()
except BSBLanError:
if self.available:
_LOGGER.error("An error occurred while updating the BSBLan device")
self._attr_available = False
return
self._attr_available = True
self._attr_current_temperature = float(state.current_temperature.value)
self._attr_target_temperature = float(state.target_temperature.value)
# check if preset is active else get hvac mode
_LOGGER.debug("state hvac/preset mode: %s", state.hvac_mode.value)
if state.hvac_mode.value == "2":
self._attr_preset_mode = PRESET_ECO
else:
self._attr_hvac_mode = BSBLAN_TO_HA_STATE[state.hvac_mode.value]
self._attr_preset_mode = PRESET_NONE
self._attr_temperature_unit = (
TEMP_CELSIUS
if state.current_temperature.unit == "°C"
else TEMP_FAHRENHEIT
)
await self.client.thermostat(**data)
except BSBLANError:
LOGGER.error("An error occurred while updating the BSBLAN device")
await self.coordinator.async_request_refresh()

105
homeassistant/components/bsblan/config_flow.py
View File

@ -1,27 +1,33 @@
"""Config flow for BSB-Lan integration."""
from __future__ import annotations
import logging
from typing import Any
from bsblan import BSBLan, BSBLanError, Info
from bsblan import BSBLAN, BSBLANError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import format_mac
from .const import CONF_DEVICE_IDENT, CONF_PASSKEY, DOMAIN
_LOGGER = logging.getLogger(__name__)
from .const import CONF_PASSKEY, DEFAULT_PORT, DOMAIN
class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a BSBLan config flow."""
class BSBLANFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a BSBLAN config flow."""
VERSION = 1
host: str
port: int
mac: str
passkey: str | None = None
username: str | None = None
password: str | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -29,33 +35,20 @@ class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN):
if user_input is None:
return self._show_setup_form()
self.host = user_input[CONF_HOST]
self.port = user_input[CONF_PORT]
self.passkey = user_input.get(CONF_PASSKEY)
self.username = user_input.get(CONF_USERNAME)
self.password = user_input.get(CONF_PASSWORD)
try:
info = await self._get_bsblan_info(
host=user_input[CONF_HOST],
port=user_input[CONF_PORT],
passkey=user_input.get(CONF_PASSKEY),
username=user_input.get(CONF_USERNAME),
password=user_input.get(CONF_PASSWORD),
)
except BSBLanError:
await self._get_bsblan_info()
except BSBLANError:
return self._show_setup_form({"base": "cannot_connect"})
# Check if already configured
await self.async_set_unique_id(info.device_identification)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=info.device_identification,
data={
CONF_HOST: user_input[CONF_HOST],
CONF_PORT: user_input[CONF_PORT],
CONF_PASSKEY: user_input.get(CONF_PASSKEY),
CONF_DEVICE_IDENT: info.device_identification,
CONF_USERNAME: user_input.get(CONF_USERNAME),
CONF_PASSWORD: user_input.get(CONF_PASSWORD),
},
)
return self._async_create_entry()
@callback
def _show_setup_form(self, errors: dict | None = None) -> FlowResult:
"""Show the setup form to the user."""
return self.async_show_form(
@ -63,7 +56,7 @@ class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN):
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Optional(CONF_PORT, default=80): int,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): int,
vol.Optional(CONF_PASSKEY): str,
vol.Optional(CONF_USERNAME): str,
vol.Optional(CONF_PASSWORD): str,
@ -72,23 +65,39 @@ class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN):
errors=errors or {},
)
async def _get_bsblan_info(
self,
host: str,
username: str | None,
password: str | None,
passkey: str | None,
port: int,
) -> Info:
"""Get device information from an BSBLan device."""
@callback
def _async_create_entry(self) -> FlowResult:
return self.async_create_entry(
title=format_mac(self.mac),
data={
CONF_HOST: self.host,
CONF_PORT: self.port,
CONF_PASSKEY: self.passkey,
CONF_USERNAME: self.username,
CONF_PASSWORD: self.password,
},
)
async def _get_bsblan_info(self, raise_on_progress: bool = True) -> None:
"""Get device information from an BSBLAN device."""
session = async_get_clientsession(self.hass)
_LOGGER.debug("request bsblan.info:")
bsblan = BSBLan(
host,
username=username,
password=password,
passkey=passkey,
port=port,
bsblan = BSBLAN(
host=self.host,
username=self.username,
password=self.password,
passkey=self.passkey,
port=self.port,
session=session,
)
return await bsblan.info()
device = await bsblan.device()
self.mac = device.MAC
await self.async_set_unique_id(
format_mac(self.mac), raise_on_progress=raise_on_progress
)
self._abort_if_unique_id_configured(
updates={
CONF_HOST: self.host,
CONF_PORT: self.port,
}
)

26
homeassistant/components/bsblan/const.py
View File

@ -1,23 +1,25 @@
"""Constants for the BSB-Lan integration."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Final
DOMAIN = "bsblan"
# Integration domain
DOMAIN: Final = "bsblan"
LOGGER = logging.getLogger(__package__)
SCAN_INTERVAL = timedelta(seconds=12)
# Services
DATA_BSBLAN_CLIENT: Final = "bsblan_client"
DATA_BSBLAN_TIMER: Final = "bsblan_timer"
DATA_BSBLAN_UPDATED: Final = "bsblan_updated"
ATTR_TARGET_TEMPERATURE: Final = "target_temperature"
ATTR_INSIDE_TEMPERATURE: Final = "inside_temperature"
ATTR_OUTSIDE_TEMPERATURE: Final = "outside_temperature"
ATTR_STATE_ON: Final = "on"
ATTR_STATE_OFF: Final = "off"
CONF_DEVICE_IDENT: Final = "device_identification"
CONF_CONTROLLER_FAM: Final = "controller_family"
CONF_CONTROLLER_VARI: Final = "controller_variant"
SENSOR_TYPE_TEMPERATURE: Final = "temperature"
CONF_PASSKEY: Final = "passkey"
CONF_DEVICE_IDENT: Final = "RVS21.831F/127"
DEFAULT_PORT: Final = 80

34
homeassistant/components/bsblan/entity.py Normal file
View File

@ -0,0 +1,34 @@
"""Base entity for the BSBLAN integration."""
from __future__ import annotations
from bsblan import BSBLAN, Device, Info
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.entity import DeviceInfo, Entity
from .const import DOMAIN
class BSBLANEntity(Entity):
"""Defines a BSBLAN entity."""
def __init__(
self,
client: BSBLAN,
device: Device,
info: Info,
entry: ConfigEntry,
) -> None:
"""Initialize an BSBLAN entity."""
self.client = client
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, format_mac(device.MAC))},
manufacturer="BSBLAN Inc.",
model=info.device_identification.value,
name=device.name,
sw_version=f"{device.version})",
configuration_url=f"http://{entry.data[CONF_HOST]}",
)

4
homeassistant/components/bsblan/manifest.json
View File

@ -1,9 +1,9 @@
{
"domain": "bsblan",
"name": "BSB-Lan",
"name": "BSBLAN",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/bsblan",
"requirements": ["bsblan==0.5.0"],
"requirements": ["python-bsblan==0.5.5"],
"codeowners": ["@liudger"],
"iot_class": "local_polling",
"loggers": ["bsblan"]

6
requirements_all.txt
View File

@ -465,9 +465,6 @@ brottsplatskartan==0.0.1
# homeassistant.components.brunt
brunt==1.2.0
# homeassistant.components.bsblan
bsblan==0.5.0
# homeassistant.components.bluetooth_tracker
bt_proximity==0.2.1
@ -1949,6 +1946,9 @@ pythinkingcleaner==0.0.3
# homeassistant.components.blockchain
python-blockchain-api==0.0.2
# homeassistant.components.bsblan
python-bsblan==0.5.5
# homeassistant.components.clementine
python-clementine-remote==1.0.1

6
requirements_test_all.txt
View File

@ -372,9 +372,6 @@ brother==2.0.0
# homeassistant.components.brunt
brunt==1.2.0
# homeassistant.components.bsblan
bsblan==0.5.0
# homeassistant.components.bthome
bthome-ble==1.2.2
@ -1369,6 +1366,9 @@ pytankerkoenig==0.0.6
# homeassistant.components.tautulli
pytautulli==21.11.0
# homeassistant.components.bsblan
python-bsblan==0.5.5
# homeassistant.components.ecobee
python-ecobee-api==0.2.14

87
tests/components/bsblan/__init__.py
View File

@ -1,88 +1 @@
"""Tests for the bsblan integration."""
from homeassistant.components.bsblan.const import (
CONF_DEVICE_IDENT,
CONF_PASSKEY,
DOMAIN,
)
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONTENT_TYPE_JSON,
)
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_fixture
from tests.test_util.aiohttp import AiohttpClientMocker
async def init_integration(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
skip_setup: bool = False,
) -> MockConfigEntry:
"""Set up the BSBLan integration in Home Assistant."""
aioclient_mock.post(
"http://example.local:80/1234/JQ?Parameter=6224,6225,6226",
params={"Parameter": "6224,6225,6226"},
text=load_fixture("bsblan/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="RVS21.831F/127",
data={
CONF_HOST: "example.local",
CONF_USERNAME: "nobody",
CONF_PASSWORD: "qwerty",
CONF_PASSKEY: "1234",
CONF_PORT: 80,
CONF_DEVICE_IDENT: "RVS21.831F/127",
},
)
entry.add_to_hass(hass)
if not skip_setup:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
async def init_integration_without_auth(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
skip_setup: bool = False,
) -> MockConfigEntry:
"""Set up the BSBLan integration in Home Assistant."""
aioclient_mock.post(
"http://example.local:80/1234/JQ?Parameter=6224,6225,6226",
params={"Parameter": "6224,6225,6226"},
text=load_fixture("bsblan/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="RVS21.831F/127",
data={
CONF_HOST: "example.local",
CONF_PASSKEY: "1234",
CONF_PORT: 80,
CONF_DEVICE_IDENT: "RVS21.831F/127",
},
)
entry.add_to_hass(hass)
if not skip_setup:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

79
tests/components/bsblan/conftest.py Normal file
View File

@ -0,0 +1,79 @@
"""Fixtures for BSBLAN integration tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
from bsblan import Device, Info, State
import pytest
from homeassistant.components.bsblan.const import CONF_PASSKEY, DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_fixture
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="BSBLAN Setup",
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 80,
CONF_PASSKEY: "1234",
CONF_USERNAME: "admin",
CONF_PASSWORD: "admin1234",
},
unique_id="00:80:41:19:69:90",
)
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Mock setting up a config entry."""
with patch(
"homeassistant.components.bsblan.async_setup_entry", return_value=True
) as mock_setup:
yield mock_setup
@pytest.fixture
def mock_bsblan_config_flow() -> Generator[None, MagicMock, None]:
"""Return a mocked BSBLAN client."""
with patch(
"homeassistant.components.bsblan.config_flow.BSBLAN", autospec=True
) as bsblan_mock:
bsblan = bsblan_mock.return_value
bsblan.device.return_value = Device.parse_raw(
load_fixture("device.json", DOMAIN)
)
bsblan.info.return_value = Info.parse_raw(load_fixture("info.json", DOMAIN))
yield bsblan
@pytest.fixture
def mock_bsblan(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]:
"""Return a mocked BSBLAN client."""
with patch("homeassistant.components.bsblan.BSBLAN", autospec=True) as bsblan_mock:
bsblan = bsblan_mock.return_value
bsblan.info.return_value = Info.parse_raw(load_fixture("info.json", DOMAIN))
bsblan.device.return_value = Device.parse_raw(
load_fixture("device.json", DOMAIN)
)
bsblan.state.return_value = State.parse_raw(load_fixture("state.json", DOMAIN))
yield bsblan
@pytest.fixture
async def init_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_bsblan: MagicMock
) -> MockConfigEntry:
"""Set up the bsblan integration for testing."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry

42
tests/components/bsblan/fixtures/device.json Normal file
View File

@ -0,0 +1,42 @@
{
"name": "BSB-LAN",
"version": "1.0.38-20200730234859",
"freeram": 85479,
"uptime": 969402857,
"MAC": "00:80:41:19:69:90",
"freespace": 0,
"bus": "BSB",
"buswritable": 1,
"busaddr": 66,
"busdest": 0,
"monitor": 0,
"verbose": 1,
"protectedGPIO": [
{ "pin": 0 },
{ "pin": 1 },
{ "pin": 4 },
{ "pin": 10 },
{ "pin": 11 },
{ "pin": 12 },
{ "pin": 13 },
{ "pin": 18 },
{ "pin": 19 },
{ "pin": 20 },
{ "pin": 21 },
{ "pin": 22 },
{ "pin": 23 },
{ "pin": 50 },
{ "pin": 51 },
{ "pin": 52 },
{ "pin": 53 },
{ "pin": 62 },
{ "pin": 63 },
{ "pin": 64 },
{ "pin": 65 },
{ "pin": 66 },
{ "pin": 67 },
{ "pin": 68 },
{ "pin": 69 }
],
"averages": []
}

26
tests/components/bsblan/fixtures/info.json
View File

@ -1,23 +1,29 @@
{
"6224": {
"name": "Geräte-Identifikation",
"device_identification": {
"name": "Gerte-Identifikation",
"error": 0,
"value": "RVS21.831F/127",
"unit": "",
"desc": "",
"dataType": 7
"dataType": 7,
"readonly": 0,
"unit": ""
},
"6225": {
"controller_family": {
"name": "Device family",
"error": 0,
"value": "211",
"unit": "",
"desc": "",
"dataType": 0
"dataType": 0,
"readonly": 0,
"unit": ""
},
"6226": {
"controller_variant": {
"name": "Device variant",
"error": 0,
"value": "127",
"unit": "",
"desc": "",
"dataType": 0
"dataType": 0,
"readonly": 0,
"unit": ""
}
}

101
tests/components/bsblan/fixtures/state.json Normal file
View File

@ -0,0 +1,101 @@
{
"hvac_mode": {
"name": "Operating mode",
"error": 0,
"value": "heat",
"desc": "Komfort",
"dataType": 1,
"readonly": 0,
"unit": ""
},
"target_temperature": {
"name": "Room temperature Comfort setpoint",
"error": 0,
"value": "18.5",
"desc": "",
"dataType": 0,
"readonly": 0,
"unit": "°C"
},
"target_temperature_high": {
"name": "Komfortsollwert Maximum",
"error": 0,
"value": "23.0",
"desc": "",
"dataType": 0,
"readonly": 0,
"unit": "°C"
},
"target_temperature_low": {
"name": "Room temp reduced setpoint",
"error": 0,
"value": "17.0",
"desc": "",
"dataType": 0,
"readonly": 0,
"unit": "°C"
},
"min_temp": {
"name": "Room temp frost protection setpoint",
"error": 0,
"value": "8.0",
"desc": "",
"dataType": 0,
"readonly": 0,
"unit": "°C"
},
"max_temp": {
"name": "Summer/winter changeover temp heat circuit 1",
"error": 0,
"value": "20.0",
"desc": "",
"dataType": 0,
"readonly": 0,
"unit": "°C"
},
"hvac_mode2": {
"name": "Operating mode",
"error": 0,
"value": "2",
"desc": "Reduziert",
"dataType": 1,
"readonly": 0,
"unit": ""
},
"hvac_action": {
"name": "Status heating circuit 1",
"error": 0,
"value": "122",
"desc": "Raumtemp\u2019begrenzung",
"dataType": 1,
"readonly": 1,
"unit": ""
},
"outside_temperature": {
"name": "Outside temp sensor local",
"error": 0,
"value": "6.1",
"desc": "",
"dataType": 0,
"readonly": 0,
"unit": "°C"
},
"current_temperature": {
"name": "Room temp 1 actual value",
"error": 0,
"value": "18.6",
"desc": "",
"dataType": 0,
"readonly": 1,
"unit": "°C"
},
"room1_thermostat_mode": {
"name": "Raumthermostat 1",
"error": 0,
"value": "0",
"desc": "Kein Bedarf",
"dataType": 1,
"readonly": 1,
"unit": ""
}
}

194
tests/components/bsblan/test_config_flow.py
View File

@ -1,23 +1,64 @@
"""Tests for the BSBLan device config flow."""
import aiohttp
from unittest.mock import AsyncMock, MagicMock
from bsblan import BSBLANConnectionError
from homeassistant import data_entry_flow
from homeassistant.components.bsblan import config_flow
from homeassistant.components.bsblan.const import CONF_DEVICE_IDENT, CONF_PASSKEY
from homeassistant.components.bsblan.const import CONF_PASSKEY, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONTENT_TYPE_JSON,
)
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from homeassistant.helpers.device_registry import format_mac
from . import init_integration
from tests.common import MockConfigEntry
from tests.common import load_fixture
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_full_user_flow_implementation(
hass: HomeAssistant,
mock_bsblan_config_flow: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test the full manual user flow from start to finish."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("step_id") == SOURCE_USER
assert "flow_id" in result
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "127.0.0.1",
CONF_PORT: 80,
CONF_PASSKEY: "1234",
CONF_USERNAME: "admin",
CONF_PASSWORD: "admin1234",
},
)
assert result2.get("type") == RESULT_TYPE_CREATE_ENTRY
assert result2.get("title") == format_mac("00:80:41:19:69:90")
assert result2.get("data") == {
CONF_HOST: "127.0.0.1",
CONF_PORT: 80,
CONF_PASSKEY: "1234",
CONF_USERNAME: "admin",
CONF_PASSWORD: "admin1234",
}
assert "result" in result2
assert result2["result"].unique_id == format_mac("00:80:41:19:69:90")
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_bsblan_config_flow.device.mock_calls) == 1
async def test_show_user_form(hass: HomeAssistant) -> None:
@ -28,132 +69,51 @@ async def test_show_user_form(hass: HomeAssistant) -> None:
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
async def test_connection_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
hass: HomeAssistant,
mock_bsblan_config_flow: MagicMock,
) -> None:
"""Test we show user form on BSBLan connection error."""
aioclient_mock.post(
"http://example.local:80/1234/JQ?Parameter=6224,6225,6226",
exc=aiohttp.ClientError,
)
mock_bsblan_config_flow.device.side_effect = BSBLANConnectionError
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
DOMAIN,
context={"source": SOURCE_USER},
data={
CONF_HOST: "example.local",
CONF_USERNAME: "nobody",
CONF_PASSWORD: "qwerty",
CONF_PASSKEY: "1234",
CONF_HOST: "127.0.0.1",
CONF_PORT: 80,
CONF_PASSKEY: "1234",
CONF_USERNAME: "admin",
CONF_PASSWORD: "admin1234",
},
)
assert result["errors"] == {"base": "cannot_connect"}
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("errors") == {"base": "cannot_connect"}
assert result.get("step_id") == "user"
async def test_user_device_exists_abort(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
hass: HomeAssistant,
mock_bsblan_config_flow: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test we abort zeroconf flow if BSBLan device already configured."""
await init_integration(hass, aioclient_mock)
"""Test we abort flow if BSBLAN device already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
DOMAIN,
context={"source": SOURCE_USER},
data={
CONF_HOST: "example.local",
CONF_USERNAME: "nobody",
CONF_PASSWORD: "qwerty",
CONF_HOST: "127.0.0.1",
CONF_PORT: 80,
CONF_PASSKEY: "1234",
CONF_PORT: 80,
CONF_USERNAME: "admin",
CONF_PASSWORD: "admin1234",
},
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
async def test_full_user_flow_implementation(
hass: HomeAssistant, aioclient_mock
) -> None:
"""Test the full manual user flow from start to finish."""
aioclient_mock.post(
"http://example.local:80/1234/JQ?Parameter=6224,6225,6226",
text=load_fixture("bsblan/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": SOURCE_USER},
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "example.local",
CONF_USERNAME: "nobody",
CONF_PASSWORD: "qwerty",
CONF_PASSKEY: "1234",
CONF_PORT: 80,
},
)
assert result["data"][CONF_HOST] == "example.local"
assert result["data"][CONF_USERNAME] == "nobody"
assert result["data"][CONF_PASSWORD] == "qwerty"
assert result["data"][CONF_PASSKEY] == "1234"
assert result["data"][CONF_PORT] == 80
assert result["data"][CONF_DEVICE_IDENT] == "RVS21.831F/127"
assert result["title"] == "RVS21.831F/127"
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
entries = hass.config_entries.async_entries(config_flow.DOMAIN)
assert entries[0].unique_id == "RVS21.831F/127"
async def test_full_user_flow_implementation_without_auth(
hass: HomeAssistant, aioclient_mock
) -> None:
"""Test the full manual user flow from start to finish."""
aioclient_mock.post(
"http://example2.local:80/JQ?Parameter=6224,6225,6226",
text=load_fixture("bsblan/info.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": SOURCE_USER},
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "example2.local",
CONF_PORT: 80,
},
)
assert result["data"][CONF_HOST] == "example2.local"
assert result["data"][CONF_USERNAME] is None
assert result["data"][CONF_PASSWORD] is None
assert result["data"][CONF_PASSKEY] is None
assert result["data"][CONF_PORT] == 80
assert result["data"][CONF_DEVICE_IDENT] == "RVS21.831F/127"
assert result["title"] == "RVS21.831F/127"
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
entries = hass.config_entries.async_entries(config_flow.DOMAIN)
assert entries[0].unique_id == "RVS21.831F/127"
assert result.get("type") == RESULT_TYPE_ABORT
assert result.get("reason") == "already_configured"

66
tests/components/bsblan/test_init.py
View File

@ -1,48 +1,46 @@
"""Tests for the BSBLan integration."""
import aiohttp
from unittest.mock import MagicMock
from bsblan import BSBLANConnectionError
from homeassistant.components.bsblan.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import init_integration, init_integration_without_auth
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_load_unload_config_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_bsblan: MagicMock,
) -> None:
"""Test the BSBLAN configuration entry loading/unloading."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert len(mock_bsblan.device.mock_calls) == 1
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert not hass.data.get(DOMAIN)
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_config_entry_not_ready(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_bsblan: MagicMock,
) -> None:
"""Test the BSBLan configuration entry not ready."""
aioclient_mock.post(
"http://example.local:80/1234/JQ?Parameter=6224,6225,6226",
exc=aiohttp.ClientError,
)
"""Test the bsblan configuration entry not ready."""
mock_bsblan.state.side_effect = BSBLANConnectionError
entry = await init_integration(hass, aioclient_mock)
assert entry.state is ConfigEntryState.SETUP_RETRY
async def test_unload_config_entry(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the BSBLan configuration entry unloading."""
entry = await init_integration(hass, aioclient_mock)
assert hass.data[DOMAIN]
await hass.config_entries.async_unload(entry.entry_id)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert not hass.data.get(DOMAIN)
async def test_config_entry_no_authentication(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the BSBLan configuration entry not ready."""
aioclient_mock.post(
"http://example.local:80/1234/JQ?Parameter=6224,6225,6226",
exc=aiohttp.ClientError,
)
entry = await init_integration_without_auth(hass, aioclient_mock)
assert entry.state is ConfigEntryState.SETUP_RETRY
assert len(mock_bsblan.state.mock_calls) == 1
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY