diff --git a/CODEOWNERS b/CODEOWNERS index 87b7aafd1df..156a66b174f 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -577,6 +577,8 @@ build.json @home-assistant/supervisor /homeassistant/components/keenetic_ndms2/ @foxel /tests/components/keenetic_ndms2/ @foxel /homeassistant/components/kef/ @basnijholt +/homeassistant/components/kegtron/ @Ernst79 +/tests/components/kegtron/ @Ernst79 /homeassistant/components/keyboard_remote/ @bendavid @lanrat /homeassistant/components/kmtronic/ @dgomes /tests/components/kmtronic/ @dgomes diff --git a/homeassistant/components/kegtron/__init__.py b/homeassistant/components/kegtron/__init__.py new file mode 100644 index 00000000000..7a1669bdcd4 --- /dev/null +++ b/homeassistant/components/kegtron/__init__.py @@ -0,0 +1,49 @@ +"""The Kegtron integration.""" +from __future__ import annotations + +import logging + +from kegtron_ble import KegtronBluetoothDeviceData + +from homeassistant.components.bluetooth import BluetoothScanningMode +from homeassistant.components.bluetooth.passive_update_processor import ( + PassiveBluetoothProcessorCoordinator, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant + +from .const import DOMAIN + +PLATFORMS: list[Platform] = [Platform.SENSOR] + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up Kegtron BLE device from a config entry.""" + address = entry.unique_id + assert address is not None + data = KegtronBluetoothDeviceData() + coordinator = hass.data.setdefault(DOMAIN, {})[ + entry.entry_id + ] = PassiveBluetoothProcessorCoordinator( + hass, + _LOGGER, + address=address, + mode=BluetoothScanningMode.PASSIVE, + update_method=data.update, + ) + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + entry.async_on_unload( + coordinator.async_start() + ) # only start after all platforms have had a chance to subscribe + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload a config entry.""" + if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS): + hass.data[DOMAIN].pop(entry.entry_id) + + return unload_ok diff --git a/homeassistant/components/kegtron/config_flow.py b/homeassistant/components/kegtron/config_flow.py new file mode 100644 index 00000000000..cc0457af87b --- /dev/null +++ b/homeassistant/components/kegtron/config_flow.py @@ -0,0 +1,94 @@ +"""Config flow for Kegtron ble integration.""" +from __future__ import annotations + +from typing import Any + +from kegtron_ble import KegtronBluetoothDeviceData as DeviceData +import voluptuous as vol + +from homeassistant.components.bluetooth import ( + BluetoothServiceInfoBleak, + async_discovered_service_info, +) +from homeassistant.config_entries import ConfigFlow +from homeassistant.const import CONF_ADDRESS +from homeassistant.data_entry_flow import FlowResult + +from .const import DOMAIN + + +class KegtronConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for kegtron.""" + + VERSION = 1 + + def __init__(self) -> None: + """Initialize the config flow.""" + self._discovery_info: BluetoothServiceInfoBleak | None = None + self._discovered_device: DeviceData | None = None + self._discovered_devices: dict[str, str] = {} + + async def async_step_bluetooth( + self, discovery_info: BluetoothServiceInfoBleak + ) -> FlowResult: + """Handle the bluetooth discovery step.""" + await self.async_set_unique_id(discovery_info.address) + self._abort_if_unique_id_configured() + device = DeviceData() + if not device.supported(discovery_info): + return self.async_abort(reason="not_supported") + self._discovery_info = discovery_info + self._discovered_device = device + return await self.async_step_bluetooth_confirm() + + async def async_step_bluetooth_confirm( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Confirm discovery.""" + assert self._discovered_device is not None + device = self._discovered_device + assert self._discovery_info is not None + discovery_info = self._discovery_info + title = device.title or device.get_device_name() or discovery_info.name + if user_input is not None: + return self.async_create_entry(title=title, data={}) + + self._set_confirm_only() + placeholders = {"name": title} + self.context["title_placeholders"] = placeholders + return self.async_show_form( + step_id="bluetooth_confirm", description_placeholders=placeholders + ) + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Handle the user step to pick discovered device.""" + if user_input is not None: + address = user_input[CONF_ADDRESS] + await self.async_set_unique_id(address, raise_on_progress=False) + self._abort_if_unique_id_configured() + return self.async_create_entry( + title=self._discovered_devices[address], data={} + ) + + current_addresses = self._async_current_ids() + for discovery_info in async_discovered_service_info(self.hass, False): + address = discovery_info.address + if address in current_addresses or address in self._discovered_devices: + continue + device = DeviceData() + if device.supported(discovery_info): + self._discovered_devices[address] = ( + device.title or device.get_device_name() or discovery_info.name + ) + + if not self._discovered_devices: + return self.async_abort(reason="no_devices_found") + + return self.async_show_form( + step_id="user", + data_schema=vol.Schema( + {vol.Required(CONF_ADDRESS): vol.In(self._discovered_devices)} + ), + ) diff --git a/homeassistant/components/kegtron/const.py b/homeassistant/components/kegtron/const.py new file mode 100644 index 00000000000..884ea8f635c --- /dev/null +++ b/homeassistant/components/kegtron/const.py @@ -0,0 +1,3 @@ +"""Constants for the Kegtron integration.""" + +DOMAIN = "kegtron" diff --git a/homeassistant/components/kegtron/device.py b/homeassistant/components/kegtron/device.py new file mode 100644 index 00000000000..b97aed76b7d --- /dev/null +++ b/homeassistant/components/kegtron/device.py @@ -0,0 +1,35 @@ +"""Support for Kegtron devices.""" +from __future__ import annotations + +import logging + +from kegtron_ble import DeviceKey, SensorDeviceInfo + +from homeassistant.components.bluetooth.passive_update_processor import ( + PassiveBluetoothEntityKey, +) +from homeassistant.const import ATTR_MANUFACTURER, ATTR_MODEL, ATTR_NAME +from homeassistant.helpers.entity import DeviceInfo + +_LOGGER = logging.getLogger(__name__) + + +def device_key_to_bluetooth_entity_key( + device_key: DeviceKey, +) -> PassiveBluetoothEntityKey: + """Convert a device key to an entity key.""" + return PassiveBluetoothEntityKey(device_key.key, device_key.device_id) + + +def sensor_device_info_to_hass( + sensor_device_info: SensorDeviceInfo, +) -> DeviceInfo: + """Convert a sensor device info to a sensor device info.""" + hass_device_info = DeviceInfo({}) + if sensor_device_info.name is not None: + hass_device_info[ATTR_NAME] = sensor_device_info.name + if sensor_device_info.manufacturer is not None: + hass_device_info[ATTR_MANUFACTURER] = sensor_device_info.manufacturer + if sensor_device_info.model is not None: + hass_device_info[ATTR_MODEL] = sensor_device_info.model + return hass_device_info diff --git a/homeassistant/components/kegtron/manifest.json b/homeassistant/components/kegtron/manifest.json new file mode 100644 index 00000000000..c3205736226 --- /dev/null +++ b/homeassistant/components/kegtron/manifest.json @@ -0,0 +1,16 @@ +{ + "domain": "kegtron", + "name": "Kegtron", + "config_flow": true, + "documentation": "https://www.home-assistant.io/integrations/kegtron", + "bluetooth": [ + { + "connectable": false, + "manufacturer_id": 65535 + } + ], + "requirements": ["kegtron-ble==0.4.0"], + "dependencies": ["bluetooth"], + "codeowners": ["@Ernst79"], + "iot_class": "local_push" +} diff --git a/homeassistant/components/kegtron/sensor.py b/homeassistant/components/kegtron/sensor.py new file mode 100644 index 00000000000..f52a7c79634 --- /dev/null +++ b/homeassistant/components/kegtron/sensor.py @@ -0,0 +1,135 @@ +"""Support for Kegtron sensors.""" +from __future__ import annotations + +from typing import Optional, Union + +from kegtron_ble import ( + SensorDeviceClass as KegtronSensorDeviceClass, + SensorUpdate, + Units, +) + +from homeassistant import config_entries +from homeassistant.components.bluetooth.passive_update_processor import ( + PassiveBluetoothDataProcessor, + PassiveBluetoothDataUpdate, + PassiveBluetoothProcessorCoordinator, + PassiveBluetoothProcessorEntity, +) +from homeassistant.components.sensor import ( + SensorDeviceClass, + SensorEntity, + SensorEntityDescription, + SensorStateClass, +) +from homeassistant.const import SIGNAL_STRENGTH_DECIBELS_MILLIWATT, VOLUME_LITERS +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity import EntityCategory +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from .const import DOMAIN +from .device import device_key_to_bluetooth_entity_key, sensor_device_info_to_hass + +SENSOR_DESCRIPTIONS = { + KegtronSensorDeviceClass.PORT_COUNT: SensorEntityDescription( + key=KegtronSensorDeviceClass.PORT_COUNT, + icon="mdi:water-pump", + ), + KegtronSensorDeviceClass.KEG_SIZE: SensorEntityDescription( + key=KegtronSensorDeviceClass.KEG_SIZE, + icon="mdi:keg", + native_unit_of_measurement=VOLUME_LITERS, + state_class=SensorStateClass.MEASUREMENT, + ), + KegtronSensorDeviceClass.KEG_TYPE: SensorEntityDescription( + key=KegtronSensorDeviceClass.KEG_TYPE, + icon="mdi:keg", + ), + KegtronSensorDeviceClass.VOLUME_START: SensorEntityDescription( + key=KegtronSensorDeviceClass.VOLUME_START, + icon="mdi:keg", + native_unit_of_measurement=VOLUME_LITERS, + state_class=SensorStateClass.MEASUREMENT, + ), + KegtronSensorDeviceClass.VOLUME_DISPENSED: SensorEntityDescription( + key=KegtronSensorDeviceClass.VOLUME_DISPENSED, + icon="mdi:keg", + native_unit_of_measurement=VOLUME_LITERS, + state_class=SensorStateClass.TOTAL, + ), + KegtronSensorDeviceClass.PORT_STATE: SensorEntityDescription( + key=KegtronSensorDeviceClass.PORT_STATE, + icon="mdi:water-pump", + ), + KegtronSensorDeviceClass.PORT_NAME: SensorEntityDescription( + key=KegtronSensorDeviceClass.PORT_NAME, + icon="mdi:water-pump", + ), + KegtronSensorDeviceClass.SIGNAL_STRENGTH: SensorEntityDescription( + key=f"{KegtronSensorDeviceClass.SIGNAL_STRENGTH}_{Units.SIGNAL_STRENGTH_DECIBELS_MILLIWATT}", + device_class=SensorDeviceClass.SIGNAL_STRENGTH, + native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, + state_class=SensorStateClass.MEASUREMENT, + entity_category=EntityCategory.DIAGNOSTIC, + entity_registry_enabled_default=False, + ), +} + + +def sensor_update_to_bluetooth_data_update( + sensor_update: SensorUpdate, +) -> PassiveBluetoothDataUpdate: + """Convert a sensor update to a bluetooth data update.""" + return PassiveBluetoothDataUpdate( + devices={ + device_id: sensor_device_info_to_hass(device_info) + for device_id, device_info in sensor_update.devices.items() + }, + entity_descriptions={ + device_key_to_bluetooth_entity_key(device_key): SENSOR_DESCRIPTIONS[ + description.device_class + ] + for device_key, description in sensor_update.entity_descriptions.items() + if description.device_class + }, + entity_data={ + device_key_to_bluetooth_entity_key(device_key): sensor_values.native_value + for device_key, sensor_values in sensor_update.entity_values.items() + }, + entity_names={ + device_key_to_bluetooth_entity_key(device_key): sensor_values.name + for device_key, sensor_values in sensor_update.entity_values.items() + }, + ) + + +async def async_setup_entry( + hass: HomeAssistant, + entry: config_entries.ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up the Kegtron BLE sensors.""" + coordinator: PassiveBluetoothProcessorCoordinator = hass.data[DOMAIN][ + entry.entry_id + ] + processor = PassiveBluetoothDataProcessor(sensor_update_to_bluetooth_data_update) + entry.async_on_unload( + processor.async_add_entities_listener( + KegtronBluetoothSensorEntity, async_add_entities + ) + ) + entry.async_on_unload(coordinator.async_register_processor(processor)) + + +class KegtronBluetoothSensorEntity( + PassiveBluetoothProcessorEntity[ + PassiveBluetoothDataProcessor[Optional[Union[float, int]]] + ], + SensorEntity, +): + """Representation of a Kegtron sensor.""" + + @property + def native_value(self) -> int | float | None: + """Return the native value.""" + return self.processor.entity_data.get(self.entity_key) diff --git a/homeassistant/components/kegtron/strings.json b/homeassistant/components/kegtron/strings.json new file mode 100644 index 00000000000..a045d84771e --- /dev/null +++ b/homeassistant/components/kegtron/strings.json @@ -0,0 +1,22 @@ +{ + "config": { + "flow_title": "[%key:component::bluetooth::config::flow_title%]", + "step": { + "user": { + "description": "[%key:component::bluetooth::config::step::user::description%]", + "data": { + "address": "[%key:component::bluetooth::config::step::user::data::address%]" + } + }, + "bluetooth_confirm": { + "description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]" + } + }, + "abort": { + "not_supported": "Device not supported", + "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]", + "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]", + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]" + } + } +} diff --git a/homeassistant/components/kegtron/translations/en.json b/homeassistant/components/kegtron/translations/en.json new file mode 100644 index 00000000000..ebd9760c161 --- /dev/null +++ b/homeassistant/components/kegtron/translations/en.json @@ -0,0 +1,22 @@ +{ + "config": { + "abort": { + "already_configured": "Device is already configured", + "already_in_progress": "Configuration flow is already in progress", + "no_devices_found": "No devices found on the network", + "not_supported": "Device not supported" + }, + "flow_title": "{name}", + "step": { + "bluetooth_confirm": { + "description": "Do you want to setup {name}?" + }, + "user": { + "data": { + "address": "Device" + }, + "description": "Choose a device to setup" + } + } + } +} \ No newline at end of file diff --git a/homeassistant/generated/bluetooth.py b/homeassistant/generated/bluetooth.py index ed98af896ff..3ba5883a70c 100644 --- a/homeassistant/generated/bluetooth.py +++ b/homeassistant/generated/bluetooth.py @@ -148,6 +148,11 @@ BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [ "local_name": "tps", "connectable": False, }, + { + "domain": "kegtron", + "connectable": False, + "manufacturer_id": 65535, + }, { "domain": "led_ble", "local_name": "LEDnet*", diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 08ec809b0f1..993503c678a 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -189,6 +189,7 @@ FLOWS = { "justnimbus", "kaleidescape", "keenetic_ndms2", + "kegtron", "kmtronic", "knx", "kodi", diff --git a/requirements_all.txt b/requirements_all.txt index 4af838b99bb..7433451ee2a 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -956,6 +956,9 @@ kaiterra-async-client==1.0.0 # homeassistant.components.keba keba-kecontact==1.1.0 +# homeassistant.components.kegtron +kegtron-ble==0.4.0 + # homeassistant.components.kiwi kiwiki-client==0.1.1 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 2cd54e4f37f..66ea321b882 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -697,6 +697,9 @@ jsonpath==0.82 # homeassistant.components.justnimbus justnimbus==0.6.0 +# homeassistant.components.kegtron +kegtron-ble==0.4.0 + # homeassistant.components.konnected konnected==1.2.0 diff --git a/tests/components/kegtron/__init__.py b/tests/components/kegtron/__init__.py new file mode 100644 index 00000000000..0b35dc8b959 --- /dev/null +++ b/tests/components/kegtron/__init__.py @@ -0,0 +1,49 @@ +"""Tests for the Kegtron integration.""" + +from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo + +NOT_KEGTRON_SERVICE_INFO = BluetoothServiceInfo( + name="Not it", + address="61DE521B-F0BF-9F44-64D4-75BBE1738105", + rssi=-63, + manufacturer_data={3234: b"\x00\x01"}, + service_data={}, + service_uuids=[], + source="local", +) + +KEGTRON_KT100_SERVICE_INFO = BluetoothServiceInfo( + name="D0:CF:5E:5C:9B:75", + manufacturer_data={ + 65535: b"I\xef\x13\x88\x02\xe2\x01Single Port\x00\x00\x00\x00\x00\x00\x00\x00\x00" + }, + address="D0:CF:5E:5C:9B:75", + rssi=-82, + service_data={}, + service_uuids=[], + source="local", +) + +KEGTRON_KT200_PORT_1_SERVICE_INFO = BluetoothServiceInfo( + name="D0:CF:5E:5C:9B:75", + manufacturer_data={ + 65535: b"#P\xc3P2\xc8APort 1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00" + }, + address="D0:CF:5E:5C:9B:75", + rssi=-82, + service_uuids=[], + service_data={}, + source="local", +) + +KEGTRON_KT200_PORT_2_SERVICE_INFO = BluetoothServiceInfo( + name="D0:CF:5E:5C:9B:75", + manufacturer_data={ + 65535: b"\xe62:\x98\x02\xe2Q2nd Port\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00" + }, + address="D0:CF:5E:5C:9B:75", + rssi=-82, + service_uuids=[], + service_data={}, + source="local", +) diff --git a/tests/components/kegtron/conftest.py b/tests/components/kegtron/conftest.py new file mode 100644 index 00000000000..472cadddada --- /dev/null +++ b/tests/components/kegtron/conftest.py @@ -0,0 +1,8 @@ +"""Kegtron session fixtures.""" + +import pytest + + +@pytest.fixture(autouse=True) +def mock_bluetooth(enable_bluetooth): + """Auto mock bluetooth.""" diff --git a/tests/components/kegtron/test_config_flow.py b/tests/components/kegtron/test_config_flow.py new file mode 100644 index 00000000000..c7015ae6346 --- /dev/null +++ b/tests/components/kegtron/test_config_flow.py @@ -0,0 +1,196 @@ +"""Test the Kegtron config flow.""" + +from unittest.mock import patch + +from homeassistant import config_entries +from homeassistant.components.kegtron.const import DOMAIN +from homeassistant.data_entry_flow import FlowResultType + +from . import ( + KEGTRON_KT100_SERVICE_INFO, + KEGTRON_KT200_PORT_2_SERVICE_INFO, + NOT_KEGTRON_SERVICE_INFO, +) + +from tests.common import MockConfigEntry + + +async def test_async_step_bluetooth_valid_device(hass): + """Test discovery via bluetooth with a valid device.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=KEGTRON_KT100_SERVICE_INFO, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "bluetooth_confirm" + with patch("homeassistant.components.kegtron.async_setup_entry", return_value=True): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == "Kegtron KT-100 9B75" + assert result2["data"] == {} + assert result2["result"].unique_id == "D0:CF:5E:5C:9B:75" + + +async def test_async_step_bluetooth_not_kegtron(hass): + """Test discovery via bluetooth not kegtron.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=NOT_KEGTRON_SERVICE_INFO, + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "not_supported" + + +async def test_async_step_user_no_devices_found(hass): + """Test setup from service info cache with no devices found.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "no_devices_found" + + +async def test_async_step_user_with_found_devices(hass): + """Test setup from service info cache with devices found.""" + with patch( + "homeassistant.components.kegtron.config_flow.async_discovered_service_info", + return_value=[KEGTRON_KT200_PORT_2_SERVICE_INFO], + ): + 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" + with patch("homeassistant.components.kegtron.async_setup_entry", return_value=True): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"address": "D0:CF:5E:5C:9B:75"}, + ) + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == "Kegtron KT-200 9B75" + assert result2["data"] == {} + assert result2["result"].unique_id == "D0:CF:5E:5C:9B:75" + + +async def test_async_step_user_device_added_between_steps(hass): + """Test the device gets added via another flow between steps.""" + with patch( + "homeassistant.components.kegtron.config_flow.async_discovered_service_info", + return_value=[KEGTRON_KT100_SERVICE_INFO], + ): + 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" + + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="D0:CF:5E:5C:9B:75", + ) + entry.add_to_hass(hass) + + with patch("homeassistant.components.kegtron.async_setup_entry", return_value=True): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"address": "D0:CF:5E:5C:9B:75"}, + ) + assert result2["type"] == FlowResultType.ABORT + assert result2["reason"] == "already_configured" + + +async def test_async_step_user_with_found_devices_already_setup(hass): + """Test setup from service info cache with devices found.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="D0:CF:5E:5C:9B:75", + ) + entry.add_to_hass(hass) + + with patch( + "homeassistant.components.kegtron.config_flow.async_discovered_service_info", + return_value=[KEGTRON_KT100_SERVICE_INFO], + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "no_devices_found" + + +async def test_async_step_bluetooth_devices_already_setup(hass): + """Test we can't start a flow if there is already a config entry.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="D0:CF:5E:5C:9B:75", + ) + entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=KEGTRON_KT100_SERVICE_INFO, + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_async_step_bluetooth_already_in_progress(hass): + """Test we can't start a flow for the same device twice.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=KEGTRON_KT100_SERVICE_INFO, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "bluetooth_confirm" + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=KEGTRON_KT100_SERVICE_INFO, + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "already_in_progress" + + +async def test_async_step_user_takes_precedence_over_discovery(hass): + """Test manual setup takes precedence over discovery.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=KEGTRON_KT100_SERVICE_INFO, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "bluetooth_confirm" + + with patch( + "homeassistant.components.kegtron.config_flow.async_discovered_service_info", + return_value=[KEGTRON_KT100_SERVICE_INFO], + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + assert result["type"] == FlowResultType.FORM + + with patch("homeassistant.components.kegtron.async_setup_entry", return_value=True): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"address": "D0:CF:5E:5C:9B:75"}, + ) + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == "Kegtron KT-100 9B75" + assert result2["data"] == {} + assert result2["result"].unique_id == "D0:CF:5E:5C:9B:75" + + # Verify the original one was aborted + assert not hass.config_entries.flow.async_progress(DOMAIN) diff --git a/tests/components/kegtron/test_sensor.py b/tests/components/kegtron/test_sensor.py new file mode 100644 index 00000000000..7c289300922 --- /dev/null +++ b/tests/components/kegtron/test_sensor.py @@ -0,0 +1,248 @@ +"""Test the Kegtron sensors.""" + +from homeassistant.components.kegtron.const import DOMAIN +from homeassistant.components.sensor import ATTR_STATE_CLASS +from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_UNIT_OF_MEASUREMENT + +from . import ( + KEGTRON_KT100_SERVICE_INFO, + KEGTRON_KT200_PORT_1_SERVICE_INFO, + KEGTRON_KT200_PORT_2_SERVICE_INFO, +) + +from tests.common import MockConfigEntry +from tests.components.bluetooth import inject_bluetooth_service_info + + +async def test_sensors_kt100(hass): + """Test setting up creates the sensors for Kegtron KT-100.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="D0:CF:5E:5C:9B:75", + ) + entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + assert len(hass.states.async_all("sensor")) == 0 + + inject_bluetooth_service_info( + hass, + KEGTRON_KT100_SERVICE_INFO, + ) + await hass.async_block_till_done() + assert len(hass.states.async_all("sensor")) == 7 + + port_count_sensor = hass.states.get("sensor.kegtron_kt_100_9b75_port_count") + port_count_sensor_attrs = port_count_sensor.attributes + assert port_count_sensor.state == "Single port device" + assert ( + port_count_sensor_attrs[ATTR_FRIENDLY_NAME] == "Kegtron KT-100 9B75 Port Count" + ) + + keg_size_sensor = hass.states.get("sensor.kegtron_kt_100_9b75_keg_size") + keg_size_sensor_attrs = keg_size_sensor.attributes + assert keg_size_sensor.state == "18.927" + assert keg_size_sensor_attrs[ATTR_FRIENDLY_NAME] == "Kegtron KT-100 9B75 Keg Size" + assert keg_size_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "L" + assert keg_size_sensor_attrs[ATTR_STATE_CLASS] == "measurement" + + keg_type_sensor = hass.states.get("sensor.kegtron_kt_100_9b75_keg_type") + keg_type_sensor_attrs = keg_type_sensor.attributes + assert keg_type_sensor.state == "Corny (5.0 gal)" + assert keg_type_sensor_attrs[ATTR_FRIENDLY_NAME] == "Kegtron KT-100 9B75 Keg Type" + + volume_start_sensor = hass.states.get("sensor.kegtron_kt_100_9b75_volume_start") + volume_start_sensor_attrs = volume_start_sensor.attributes + assert volume_start_sensor.state == "5.0" + assert ( + volume_start_sensor_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-100 9B75 Volume Start" + ) + assert volume_start_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "L" + assert volume_start_sensor_attrs[ATTR_STATE_CLASS] == "measurement" + + volume_dispensed_sensor = hass.states.get( + "sensor.kegtron_kt_100_9b75_volume_dispensed" + ) + volume_dispensed_attrs = volume_dispensed_sensor.attributes + assert volume_dispensed_sensor.state == "0.738" + assert ( + volume_dispensed_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-100 9B75 Volume Dispensed" + ) + assert volume_dispensed_attrs[ATTR_UNIT_OF_MEASUREMENT] == "L" + assert volume_dispensed_attrs[ATTR_STATE_CLASS] == "total" + + port_state_sensor = hass.states.get("sensor.kegtron_kt_100_9b75_port_state") + port_state_sensor_attrs = port_state_sensor.attributes + assert port_state_sensor.state == "Configured" + assert ( + port_state_sensor_attrs[ATTR_FRIENDLY_NAME] == "Kegtron KT-100 9B75 Port State" + ) + + port_name_sensor = hass.states.get("sensor.kegtron_kt_100_9b75_port_name") + port_name_attrs = port_name_sensor.attributes + assert port_name_sensor.state == "Single Port" + assert port_name_attrs[ATTR_FRIENDLY_NAME] == "Kegtron KT-100 9B75 Port Name" + + assert await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done() + + +async def test_sensors_kt200(hass): + """Test setting up creates the sensors for Kegtron KT-200.""" + entry = MockConfigEntry( + domain=DOMAIN, + unique_id="D0:CF:5E:5C:9B:75", + ) + entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + assert len(hass.states.async_all("sensor")) == 0 + + # Kegtron KT-200 has two ports that are reported separately, start with port 2 + inject_bluetooth_service_info( + hass, + KEGTRON_KT200_PORT_2_SERVICE_INFO, + ) + await hass.async_block_till_done() + assert len(hass.states.async_all("sensor")) == 7 + + port_count_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_port_count") + port_count_sensor_attrs = port_count_sensor.attributes + assert port_count_sensor.state == "Dual port device" + assert ( + port_count_sensor_attrs[ATTR_FRIENDLY_NAME] == "Kegtron KT-200 9B75 Port Count" + ) + + keg_size_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_keg_size_port_2") + keg_size_sensor_attrs = keg_size_sensor.attributes + assert keg_size_sensor.state == "58.93" + assert ( + keg_size_sensor_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Keg Size Port 2" + ) + assert keg_size_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "L" + assert keg_size_sensor_attrs[ATTR_STATE_CLASS] == "measurement" + + keg_type_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_keg_type_port_2") + keg_type_sensor_attrs = keg_type_sensor.attributes + assert keg_type_sensor.state == "Other (58.93 L)" + assert ( + keg_type_sensor_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Keg Type Port 2" + ) + + volume_start_sensor = hass.states.get( + "sensor.kegtron_kt_200_9b75_volume_start_port_2" + ) + volume_start_sensor_attrs = volume_start_sensor.attributes + assert volume_start_sensor.state == "15.0" + assert ( + volume_start_sensor_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Volume Start Port 2" + ) + assert volume_start_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "L" + assert volume_start_sensor_attrs[ATTR_STATE_CLASS] == "measurement" + + volume_dispensed_sensor = hass.states.get( + "sensor.kegtron_kt_200_9b75_volume_dispensed_port_2" + ) + volume_dispensed_attrs = volume_dispensed_sensor.attributes + assert volume_dispensed_sensor.state == "0.738" + assert ( + volume_dispensed_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Volume Dispensed Port 2" + ) + assert volume_dispensed_attrs[ATTR_UNIT_OF_MEASUREMENT] == "L" + assert volume_dispensed_attrs[ATTR_STATE_CLASS] == "total" + + port_state_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_port_state_port_2") + port_state_sensor_attrs = port_state_sensor.attributes + assert port_state_sensor.state == "Configured" + assert ( + port_state_sensor_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Port State Port 2" + ) + + port_name_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_port_name_port_2") + port_name_attrs = port_name_sensor.attributes + assert port_name_sensor.state == "2nd Port" + assert port_name_attrs[ATTR_FRIENDLY_NAME] == "Kegtron KT-200 9B75 Port Name Port 2" + + # Followed by a BLE advertisement of port 1 + inject_bluetooth_service_info( + hass, + KEGTRON_KT200_PORT_1_SERVICE_INFO, + ) + await hass.async_block_till_done() + assert len(hass.states.async_all("sensor")) == 13 + + port_count_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_port_count") + port_count_sensor_attrs = port_count_sensor.attributes + assert port_count_sensor.state == "Dual port device" + assert ( + port_count_sensor_attrs[ATTR_FRIENDLY_NAME] == "Kegtron KT-200 9B75 Port Count" + ) + + keg_size_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_keg_size_port_1") + keg_size_sensor_attrs = keg_size_sensor.attributes + assert keg_size_sensor.state == "9.04" + assert ( + keg_size_sensor_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Keg Size Port 1" + ) + assert keg_size_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "L" + assert keg_size_sensor_attrs[ATTR_STATE_CLASS] == "measurement" + + keg_type_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_keg_type_port_1") + keg_type_sensor_attrs = keg_type_sensor.attributes + assert keg_type_sensor.state == "Other (9.04 L)" + assert ( + keg_type_sensor_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Keg Type Port 1" + ) + + volume_start_sensor = hass.states.get( + "sensor.kegtron_kt_200_9b75_volume_start_port_1" + ) + volume_start_sensor_attrs = volume_start_sensor.attributes + assert volume_start_sensor.state == "50.0" + assert ( + volume_start_sensor_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Volume Start Port 1" + ) + assert volume_start_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "L" + assert volume_start_sensor_attrs[ATTR_STATE_CLASS] == "measurement" + + volume_dispensed_sensor = hass.states.get( + "sensor.kegtron_kt_200_9b75_volume_dispensed_port_1" + ) + volume_dispensed_attrs = volume_dispensed_sensor.attributes + assert volume_dispensed_sensor.state == "13.0" + assert ( + volume_dispensed_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Volume Dispensed Port 1" + ) + assert volume_dispensed_attrs[ATTR_UNIT_OF_MEASUREMENT] == "L" + assert volume_dispensed_attrs[ATTR_STATE_CLASS] == "total" + + port_state_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_port_state_port_1") + port_state_sensor_attrs = port_state_sensor.attributes + assert port_state_sensor.state == "Configured" + assert ( + port_state_sensor_attrs[ATTR_FRIENDLY_NAME] + == "Kegtron KT-200 9B75 Port State Port 1" + ) + + port_name_sensor = hass.states.get("sensor.kegtron_kt_200_9b75_port_name_port_1") + port_name_attrs = port_name_sensor.attributes + assert port_name_sensor.state == "Port 1" + assert port_name_attrs[ATTR_FRIENDLY_NAME] == "Kegtron KT-200 9B75 Port Name Port 1" + + assert await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done()