Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add sensors platform to Watergate integration (#133015)

pull/133492/head
adam-the-hero 2024-12-18 14:52:25 +01:00 committed by GitHub
parent 2d6d313e5c
commit 943b1d9f08
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 1002 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
homeassistant/components/watergate/__init__.py
View File

@ -25,8 +25,13 @@ from .coordinator import WatergateDataCoordinator
_LOGGER = logging.getLogger(__name__)
WEBHOOK_TELEMETRY_TYPE = "telemetry"
WEBHOOK_VALVE_TYPE = "valve"
WEBHOOK_WIFI_CHANGED_TYPE = "wifi-changed"
WEBHOOK_POWER_SUPPLY_CHANGED_TYPE = "power-supply-changed"
PLATFORMS: list[Platform] = [
Platform.SENSOR,
Platform.VALVE,
]
@ -82,7 +87,6 @@ def get_webhook_handler(
async def async_webhook_handler(
hass: HomeAssistant, webhook_id: str, request: Request
) -> Response | None:
# Handle http post calls to the path.
if not request.body_exists:
return HomeAssistantView.json(
result="No Body", status_code=HTTPStatus.BAD_REQUEST
@ -96,9 +100,29 @@ def get_webhook_handler(
body_type = body.get("type")
coordinator_data = coordinator.data
if body_type == Platform.VALVE and coordinator_data:
coordinator_data.valve_state = data.state
if not (coordinator_data := coordinator.data):
pass
elif body_type == WEBHOOK_VALVE_TYPE:
coordinator_data.state.valve_state = data.state
elif body_type == WEBHOOK_TELEMETRY_TYPE:
errors = data.errors or {}
coordinator_data.telemetry.flow = (
data.flow if "flow" not in errors else None
)
coordinator_data.telemetry.pressure = (
data.pressure if "pressure" not in errors else None
)
coordinator_data.telemetry.water_temperature = (
data.temperature if "temperature" not in errors else None
)
elif body_type == WEBHOOK_WIFI_CHANGED_TYPE:
coordinator_data.networking.ip = data.ip
coordinator_data.networking.gateway = data.gateway
coordinator_data.networking.subnet = data.subnet
coordinator_data.networking.ssid = data.ssid
coordinator_data.networking.rssi = data.rssi
elif body_type == WEBHOOK_POWER_SUPPLY_CHANGED_TYPE:
coordinator_data.state.power_supply = data.supply
coordinator.async_set_updated_data(coordinator_data)

33
homeassistant/components/watergate/coordinator.py
View File

@ -1,10 +1,11 @@
"""Coordinator for Watergate API."""
from dataclasses import dataclass
from datetime import timedelta
import logging
from watergate_local_api import WatergateApiException, WatergateLocalApiClient
from watergate_local_api.models import DeviceState
from watergate_local_api.models import DeviceState, NetworkingData, TelemetryData
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -14,7 +15,16 @@ from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class WatergateDataCoordinator(DataUpdateCoordinator[DeviceState]):
@dataclass
class WatergateAgregatedRequests:
"""Class to hold aggregated requests."""
state: DeviceState
telemetry: TelemetryData
networking: NetworkingData
class WatergateDataCoordinator(DataUpdateCoordinator[WatergateAgregatedRequests]):
"""Class to manage fetching watergate data."""
def __init__(self, hass: HomeAssistant, api: WatergateLocalApiClient) -> None:
@ -27,9 +37,22 @@ class WatergateDataCoordinator(DataUpdateCoordinator[DeviceState]):
)
self.api = api
async def _async_update_data(self) -> DeviceState:
async def _async_update_data(self) -> WatergateAgregatedRequests:
try:
state = await self.api.async_get_device_state()
telemetry = await self.api.async_get_telemetry_data()
networking = await self.api.async_get_networking()
except WatergateApiException as exc:
raise UpdateFailed from exc
return state
raise UpdateFailed(f"Sonic device is unavailable: {exc}") from exc
return WatergateAgregatedRequests(state, telemetry, networking)
def async_set_updated_data(self, data: WatergateAgregatedRequests) -> None:
"""Manually update data, notify listeners and DO NOT reset refresh interval."""
self.data = data
self.logger.debug(
"Manually updated %s data",
self.name,
)
self.async_update_listeners()

10
homeassistant/components/watergate/entity.py
View File

@ -20,11 +20,13 @@ class WatergateEntity(CoordinatorEntity[WatergateDataCoordinator]):
"""Initialize the entity."""
super().__init__(coordinator)
self._api_client = coordinator.api
self._attr_unique_id = f"{coordinator.data.serial_number}.{entity_name}"
self._attr_unique_id = f"{coordinator.data.state.serial_number}.{entity_name}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.data.serial_number)},
identifiers={(DOMAIN, coordinator.data.state.serial_number)},
name="Sonic",
serial_number=coordinator.data.serial_number,
serial_number=coordinator.data.state.serial_number,
manufacturer=MANUFACTURER,
sw_version=coordinator.data.firmware_version if coordinator.data else None,
sw_version=(
coordinator.data.state.firmware_version if coordinator.data else None
),
)

1
homeassistant/components/watergate/quality_scale.yaml
View File

@ -27,6 +27,7 @@ rules:
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
config-entry-unloading: done
log-when-unavailable: todo

214
homeassistant/components/watergate/sensor.py Normal file
View File

@ -0,0 +1,214 @@
"""Support for Watergate sensors."""
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import StrEnum
import logging
from homeassistant.components.sensor import (
HomeAssistant,
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
EntityCategory,
UnitOfPressure,
UnitOfTemperature,
UnitOfTime,
UnitOfVolume,
UnitOfVolumeFlowRate,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util import dt as dt_util
from . import WatergateConfigEntry
from .coordinator import WatergateAgregatedRequests, WatergateDataCoordinator
from .entity import WatergateEntity
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
class PowerSupplyMode(StrEnum):
"""LED bar mode."""
BATTERY = "battery"
EXTERNAL = "external"
BATTERY_EXTERNAL = "battery_external"
@dataclass(kw_only=True, frozen=True)
class WatergateSensorEntityDescription(SensorEntityDescription):
"""Description for Watergate sensor entities."""
value_fn: Callable[
[WatergateAgregatedRequests],
StateType | datetime | PowerSupplyMode,
]
DESCRIPTIONS: list[WatergateSensorEntityDescription] = [
WatergateSensorEntityDescription(
value_fn=lambda data: (
data.state.water_meter.duration
if data.state and data.state.water_meter
else None
),
translation_key="water_meter_volume",
key="water_meter_volume",
native_unit_of_measurement=UnitOfVolume.LITERS,
device_class=SensorDeviceClass.WATER,
state_class=SensorStateClass.TOTAL_INCREASING,
),
WatergateSensorEntityDescription(
value_fn=lambda data: (
data.state.water_meter.duration
if data.state and data.state.water_meter
else None
),
translation_key="water_meter_duration",
key="water_meter_duration",
native_unit_of_measurement=UnitOfTime.MINUTES,
device_class=SensorDeviceClass.DURATION,
state_class=SensorStateClass.TOTAL_INCREASING,
),
WatergateSensorEntityDescription(
value_fn=lambda data: data.networking.rssi if data.networking else None,
key="rssi",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
state_class=SensorStateClass.MEASUREMENT,
),
WatergateSensorEntityDescription(
value_fn=lambda data: (
dt_util.as_utc(
dt_util.now() - timedelta(microseconds=data.networking.wifi_uptime)
)
if data.networking
else None
),
translation_key="wifi_up_since",
key="wifi_up_since",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.TIMESTAMP,
),
WatergateSensorEntityDescription(
value_fn=lambda data: (
dt_util.as_utc(
dt_util.now() - timedelta(microseconds=data.networking.mqtt_uptime)
)
if data.networking
else None
),
translation_key="mqtt_up_since",
key="mqtt_up_since",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.TIMESTAMP,
),
WatergateSensorEntityDescription(
value_fn=lambda data: (
data.telemetry.water_temperature if data.telemetry else None
),
translation_key="water_temperature",
key="water_temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
),
WatergateSensorEntityDescription(
value_fn=lambda data: data.telemetry.pressure if data.telemetry else None,
translation_key="water_pressure",
key="water_pressure",
native_unit_of_measurement=UnitOfPressure.MBAR,
device_class=SensorDeviceClass.PRESSURE,
state_class=SensorStateClass.MEASUREMENT,
),
WatergateSensorEntityDescription(
value_fn=lambda data: (
data.telemetry.flow / 1000
if data.telemetry and data.telemetry.flow is not None
else None
),
key="water_flow_rate",
native_unit_of_measurement=UnitOfVolumeFlowRate.LITERS_PER_MINUTE,
device_class=SensorDeviceClass.VOLUME_FLOW_RATE,
state_class=SensorStateClass.MEASUREMENT,
),
WatergateSensorEntityDescription(
value_fn=lambda data: (
dt_util.as_utc(dt_util.now() - timedelta(seconds=data.state.uptime))
if data.state
else None
),
translation_key="up_since",
key="up_since",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.TIMESTAMP,
),
WatergateSensorEntityDescription(
value_fn=lambda data: (
PowerSupplyMode(data.state.power_supply.replace("+", "_"))
if data.state
else None
),
translation_key="power_supply_mode",
key="power_supply_mode",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.ENUM,
options=[member.value for member in PowerSupplyMode],
),
]
async def async_setup_entry(
hass: HomeAssistant,
config_entry: WatergateConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up all entries for Watergate Platform."""
coordinator = config_entry.runtime_data
async_add_entities(
SonicSensor(coordinator, description) for description in DESCRIPTIONS
)
class SonicSensor(WatergateEntity, SensorEntity):
"""Define a Sonic Sensor entity."""
entity_description: WatergateSensorEntityDescription
def __init__(
self,
coordinator: WatergateDataCoordinator,
entity_description: WatergateSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, entity_description.key)
self.entity_description = entity_description
@property
def available(self) -> bool:
"""Return True if entity is available."""
return (
super().available
and self.entity_description.value_fn(self.coordinator.data) is not None
)
@property
def native_value(self) -> str | int | float | datetime | PowerSupplyMode | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)

33
homeassistant/components/watergate/strings.json
View File

@ -17,5 +17,38 @@
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"entity": {
"sensor": {
"water_meter_volume": {
"name": "Water meter volume"
},
"water_meter_duration": {
"name": "Water meter duration"
},
"wifi_up_since": {
"name": "Wi-Fi up since"
},
"mqtt_up_since": {
"name": "MQTT up since"
},
"water_temperature": {
"name": "Water temperature"
},
"water_pressure": {
"name": "Water pressure"
},
"up_since": {
"name": "Up since"
},
"power_supply_mode": {
"name": "Power supply mode",
"state": {
"battery": "Battery",
"external": "Mains",
"battery_external": "Battery and mains"
}
}
}
}
}

13
homeassistant/components/watergate/valve.py
View File

@ -43,7 +43,9 @@ class SonicValve(WatergateEntity, ValveEntity):
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, ENTITY_NAME)
self._valve_state = coordinator.data.valve_state if coordinator.data else None
self._valve_state = (
coordinator.data.state.valve_state if coordinator.data.state else None
)
@property
def is_closed(self) -> bool:
@ -65,7 +67,9 @@ class SonicValve(WatergateEntity, ValveEntity):
"""Handle data update."""
self._attr_available = self.coordinator.data is not None
self._valve_state = (
self.coordinator.data.valve_state if self.coordinator.data else None
self.coordinator.data.state.valve_state
if self.coordinator.data.state
else None
)
self.async_write_ha_state()
@ -80,3 +84,8 @@ class SonicValve(WatergateEntity, ValveEntity):
await self._api_client.async_set_valve_state(ValveState.CLOSED)
self._valve_state = ValveState.CLOSING
self.async_write_ha_state()
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self.coordinator.data.state is not None

8
tests/components/watergate/conftest.py
View File

@ -9,7 +9,9 @@ from homeassistant.const import CONF_IP_ADDRESS
from .const import (
DEFAULT_DEVICE_STATE,
DEFAULT_NETWORKING_STATE,
DEFAULT_SERIAL_NUMBER,
DEFAULT_TELEMETRY_STATE,
MOCK_CONFIG,
MOCK_WEBHOOK_ID,
)
@ -35,6 +37,12 @@ def mock_watergate_client() -> Generator[AsyncMock]:
mock_client_instance.async_get_device_state = AsyncMock(
return_value=DEFAULT_DEVICE_STATE
)
mock_client_instance.async_get_networking = AsyncMock(
return_value=DEFAULT_NETWORKING_STATE
)
mock_client_instance.async_get_telemetry_data = AsyncMock(
return_value=DEFAULT_TELEMETRY_STATE
)
yield mock_client_instance

19
tests/components/watergate/const.py
View File

@ -1,6 +1,7 @@
"""Constants for the Watergate tests."""
from watergate_local_api.models import DeviceState
from watergate_local_api.models import DeviceState, NetworkingData, TelemetryData
from watergate_local_api.models.water_meter import WaterMeter
from homeassistant.const import CONF_IP_ADDRESS, CONF_NAME, CONF_WEBHOOK_ID
@ -22,6 +23,20 @@ DEFAULT_DEVICE_STATE = DeviceState(
"battery",
"1.0.0",
100,
{"volume": 1.2, "duration": 100},
WaterMeter(1.2, 100),
DEFAULT_SERIAL_NUMBER,
)
DEFAULT_NETWORKING_STATE = NetworkingData(
True,
True,
"192.168.1.127",
"192.168.1.1",
"255.255.255.0",
"Sonic",
-50,
2137,
1910,
)
DEFAULT_TELEMETRY_STATE = TelemetryData(0.0, 100, 28.32, None, [])

506
tests/components/watergate/snapshots/test_sensor.ambr Normal file
View File

@ -0,0 +1,506 @@
# serializer version: 1
# name: test_sensor[sensor.sonic_mqtt_up_since-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.sonic_mqtt_up_since',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'MQTT up since',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'mqtt_up_since',
'unique_id': 'a63182948ce2896a.mqtt_up_since',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.sonic_mqtt_up_since-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'Sonic MQTT up since',
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_mqtt_up_since',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2021-01-09T11:59:59+00:00',
})
# ---
# name: test_sensor[sensor.sonic_power_supply_mode-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'battery',
'external',
'battery_external',
]),
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.sonic_power_supply_mode',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Power supply mode',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'power_supply_mode',
'unique_id': 'a63182948ce2896a.power_supply_mode',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.sonic_power_supply_mode-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Sonic Power supply mode',
'options': list([
'battery',
'external',
'battery_external',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_power_supply_mode',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'battery',
})
# ---
# name: test_sensor[sensor.sonic_signal_strength-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.sonic_signal_strength',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.SIGNAL_STRENGTH: 'signal_strength'>,
'original_icon': None,
'original_name': 'Signal strength',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'a63182948ce2896a.rssi',
'unit_of_measurement': 'dBm',
})
# ---
# name: test_sensor[sensor.sonic_signal_strength-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'signal_strength',
'friendly_name': 'Sonic Signal strength',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': 'dBm',
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_signal_strength',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '-50',
})
# ---
# name: test_sensor[sensor.sonic_up_since-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.sonic_up_since',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Up since',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'up_since',
'unique_id': 'a63182948ce2896a.up_since',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.sonic_up_since-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'Sonic Up since',
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_up_since',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2021-01-09T11:58:20+00:00',
})
# ---
# name: test_sensor[sensor.sonic_volume_flow_rate-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.sonic_volume_flow_rate',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.VOLUME_FLOW_RATE: 'volume_flow_rate'>,
'original_icon': None,
'original_name': 'Volume flow rate',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'a63182948ce2896a.water_flow_rate',
'unit_of_measurement': <UnitOfVolumeFlowRate.LITERS_PER_MINUTE: 'L/min'>,
})
# ---
# name: test_sensor[sensor.sonic_volume_flow_rate-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'volume_flow_rate',
'friendly_name': 'Sonic Volume flow rate',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfVolumeFlowRate.LITERS_PER_MINUTE: 'L/min'>,
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_volume_flow_rate',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.0',
})
# ---
# name: test_sensor[sensor.sonic_water_meter_duration-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.sonic_water_meter_duration',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Water meter duration',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'water_meter_duration',
'unique_id': 'a63182948ce2896a.water_meter_duration',
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
})
# ---
# name: test_sensor[sensor.sonic_water_meter_duration-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': 'Sonic Water meter duration',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_water_meter_duration',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---
# name: test_sensor[sensor.sonic_water_meter_volume-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.sonic_water_meter_volume',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.WATER: 'water'>,
'original_icon': None,
'original_name': 'Water meter volume',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'water_meter_volume',
'unique_id': 'a63182948ce2896a.water_meter_volume',
'unit_of_measurement': <UnitOfVolume.LITERS: 'L'>,
})
# ---
# name: test_sensor[sensor.sonic_water_meter_volume-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'water',
'friendly_name': 'Sonic Water meter volume',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfVolume.LITERS: 'L'>,
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_water_meter_volume',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---
# name: test_sensor[sensor.sonic_water_pressure-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.sonic_water_pressure',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.PRESSURE: 'pressure'>,
'original_icon': None,
'original_name': 'Water pressure',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'water_pressure',
'unique_id': 'a63182948ce2896a.water_pressure',
'unit_of_measurement': <UnitOfPressure.MBAR: 'mbar'>,
})
# ---
# name: test_sensor[sensor.sonic_water_pressure-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'pressure',
'friendly_name': 'Sonic Water pressure',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfPressure.MBAR: 'mbar'>,
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_water_pressure',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---
# name: test_sensor[sensor.sonic_water_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.sonic_water_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Water temperature',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'water_temperature',
'unique_id': 'a63182948ce2896a.water_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor[sensor.sonic_water_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'Sonic Water temperature',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_water_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '28.32',
})
# ---
# name: test_sensor[sensor.sonic_wi_fi_up_since-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.sonic_wi_fi_up_since',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Wi-Fi up since',
'platform': 'watergate',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'wifi_up_since',
'unique_id': 'a63182948ce2896a.wifi_up_since',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.sonic_wi_fi_up_since-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'Sonic Wi-Fi up since',
}),
'context': <ANY>,
'entity_id': 'sensor.sonic_wi_fi_up_since',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2021-01-09T11:59:59+00:00',
})
# ---

150
tests/components/watergate/test_sensor.py Normal file
View File

@ -0,0 +1,150 @@
"""Tests for the Watergate valve platform."""
from collections.abc import Generator
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import EntityCategory, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import init_integration
from .const import DEFAULT_NETWORKING_STATE, DEFAULT_TELEMETRY_STATE, MOCK_WEBHOOK_ID
from tests.common import AsyncMock, MockConfigEntry, patch, snapshot_platform
from tests.typing import ClientSessionGenerator
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensor(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_entry: MockConfigEntry,
mock_watergate_client: Generator[AsyncMock],
freezer: FrozenDateTimeFactory,
snapshot: SnapshotAssertion,
) -> None:
"""Test states of the sensor."""
freezer.move_to("2021-01-09 12:00:00+00:00")
with patch("homeassistant.components.watergate.PLATFORMS", [Platform.SENSOR]):
await init_integration(hass, mock_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_entry.entry_id)
async def test_diagnostics_are_disabled_by_default(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_entry: MockConfigEntry,
mock_watergate_client: Generator[AsyncMock],
) -> None:
"""Test if all diagnostic entities are disabled by default."""
with patch("homeassistant.components.watergate.PLATFORMS", [Platform.SENSOR]):
await init_integration(hass, mock_entry)
entries = [
entry
for entry in entity_registry.entities.get_entries_for_config_entry_id(
mock_entry.entry_id
)
if entry.entity_category == EntityCategory.DIAGNOSTIC
]
assert len(entries) == 5
for entry in entries:
assert entry.disabled
async def test_telemetry_webhook(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
mock_entry: MockConfigEntry,
mock_watergate_client: Generator[AsyncMock],
) -> None:
"""Test if water flow webhook is handled correctly."""
await init_integration(hass, mock_entry)
def assert_state(entity_id: str, expected_state: str):
state = hass.states.get(entity_id)
assert state.state == str(expected_state)
assert_state("sensor.sonic_volume_flow_rate", DEFAULT_TELEMETRY_STATE.flow)
assert_state("sensor.sonic_water_pressure", DEFAULT_TELEMETRY_STATE.pressure)
assert_state(
"sensor.sonic_water_temperature", DEFAULT_TELEMETRY_STATE.water_temperature
)
telemetry_change_data = {
"type": "telemetry",
"data": {"flow": 2137, "pressure": 1910, "temperature": 20},
}
client = await hass_client_no_auth()
await client.post(f"/api/webhook/{MOCK_WEBHOOK_ID}", json=telemetry_change_data)
await hass.async_block_till_done()
assert_state("sensor.sonic_volume_flow_rate", "2.137")
assert_state("sensor.sonic_water_pressure", "1910")
assert_state("sensor.sonic_water_temperature", "20")
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_wifi_webhook(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
mock_entry: MockConfigEntry,
mock_watergate_client: Generator[AsyncMock],
) -> None:
"""Test if water flow webhook is handled correctly."""
await init_integration(hass, mock_entry)
def assert_state(entity_id: str, expected_state: str):
state = hass.states.get(entity_id)
assert state.state == str(expected_state)
assert_state("sensor.sonic_signal_strength", DEFAULT_NETWORKING_STATE.rssi)
wifi_change_data = {
"type": "wifi-changed",
"data": {
"ip": "192.168.2.137",
"gateway": "192.168.2.1",
"ssid": "Sonic 2",
"rssi": -70,
"subnet": "255.255.255.0",
},
}
client = await hass_client_no_auth()
await client.post(f"/api/webhook/{MOCK_WEBHOOK_ID}", json=wifi_change_data)
await hass.async_block_till_done()
assert_state("sensor.sonic_signal_strength", "-70")
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_power_supply_webhook(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
mock_entry: MockConfigEntry,
mock_watergate_client: Generator[AsyncMock],
) -> None:
"""Test if water flow webhook is handled correctly."""
await init_integration(hass, mock_entry)
entity_id = "sensor.sonic_power_supply_mode"
registered_entity = hass.states.get(entity_id)
assert registered_entity
assert registered_entity.state == "battery"
power_supply_change_data = {
"type": "power-supply-changed",
"data": {"supply": "external"},
}
client = await hass_client_no_auth()
await client.post(f"/api/webhook/{MOCK_WEBHOOK_ID}", json=power_supply_change_data)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == "external"