"""The tests for the Prometheus exporter.""" from dataclasses import dataclass import datetime from http import HTTPStatus from typing import Any, Self from unittest import mock from freezegun import freeze_time import prometheus_client from prometheus_client.utils import floatToGoString import pytest from homeassistant.components import ( alarm_control_panel, binary_sensor, climate, counter, cover, device_tracker, fan, humidifier, input_boolean, input_number, light, lock, number, person, prometheus, sensor, switch, update, ) from homeassistant.components.alarm_control_panel import AlarmControlPanelState from homeassistant.components.climate import ( ATTR_CURRENT_TEMPERATURE, ATTR_FAN_MODE, ATTR_FAN_MODES, ATTR_HUMIDITY, ATTR_HVAC_ACTION, ATTR_HVAC_MODES, ATTR_TARGET_TEMP_HIGH, ATTR_TARGET_TEMP_LOW, ) from homeassistant.components.fan import ( ATTR_DIRECTION, ATTR_OSCILLATING, ATTR_PERCENTAGE, ATTR_PRESET_MODE, ATTR_PRESET_MODES, DIRECTION_FORWARD, DIRECTION_REVERSE, ) from homeassistant.components.humidifier import ATTR_AVAILABLE_MODES from homeassistant.components.lock import LockState from homeassistant.components.sensor import SensorDeviceClass from homeassistant.const import ( ATTR_BATTERY_LEVEL, ATTR_DEVICE_CLASS, ATTR_FRIENDLY_NAME, ATTR_MODE, ATTR_TEMPERATURE, ATTR_UNIT_OF_MEASUREMENT, CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, CONTENT_TYPE_TEXT_PLAIN, DEGREE, PERCENTAGE, STATE_CLOSED, STATE_CLOSING, STATE_HOME, STATE_NOT_HOME, STATE_OFF, STATE_ON, STATE_OPEN, STATE_OPENING, STATE_UNAVAILABLE, STATE_UNKNOWN, UnitOfEnergy, UnitOfTemperature, ) from homeassistant.core import HomeAssistant from homeassistant.helpers import entity_registry as er from homeassistant.setup import async_setup_component from homeassistant.util import dt as dt_util from tests.typing import ClientSessionGenerator PROMETHEUS_PATH = "homeassistant.components.prometheus" class EntityMetric: """Represents a Prometheus metric for a Home Assistant entity.""" metric_name: str labels: dict[str, str] @classmethod def required_labels(cls) -> list[str]: """List of all required labels for a Prometheus metric.""" return [ "domain", "friendly_name", "entity", ] def __init__(self, metric_name: str, **kwargs: Any) -> None: """Create a new EntityMetric based on metric name and labels.""" self.metric_name = metric_name self.labels = kwargs # Labels that are required for all entities. for labelname in self.required_labels(): assert labelname in self.labels assert self.labels[labelname] != "" def withValue(self, value: float) -> Self: """Return a metric with value.""" return EntityMetricWithValue(self, value) @property def _metric_name_string(self) -> str: """Return a full metric name as a string.""" labels = ",".join( f'{key}="{value}"' for key, value in sorted(self.labels.items()) ) return f"{self.metric_name}{{{labels}}}" def _in_metrics(self, metrics: list[str]) -> bool: """Report whether this metric exists in the provided Prometheus output.""" return any(line.startswith(self._metric_name_string) for line in metrics) def assert_in_metrics(self, metrics: list[str]) -> None: """Assert that this metric exists in the provided Prometheus output.""" assert self._in_metrics(metrics) def assert_not_in_metrics(self, metrics: list[str]) -> None: """Assert that this metric does not exist in Prometheus output.""" assert not self._in_metrics(metrics) class EntityMetricWithValue(EntityMetric): """Represents a Prometheus metric with a value.""" value: float def __init__(self, metric: EntityMetric, value: float) -> None: """Create a new metric with a value based on a metric.""" super().__init__(metric.metric_name, **metric.labels) self.value = value @property def _metric_string(self) -> str: """Return a full metric string.""" value = floatToGoString(self.value) return f"{self._metric_name_string} {value}" def assert_in_metrics(self, metrics: list[str]) -> None: """Assert that this metric exists in the provided Prometheus output.""" assert self._metric_string in metrics @dataclass class FilterTest: """Class for capturing a filter test.""" id: str should_pass: bool def test_entity_metric_generates_metric_name_string_without_value() -> None: """Test using EntityMetric to format a simple metric string without any value.""" domain = "sensor" object_id = "outside_temperature" entity_metric = EntityMetric( metric_name="homeassistant_sensor_temperature_celsius", domain=domain, friendly_name="Outside Temperature", entity=f"{domain}.{object_id}", ) assert entity_metric._metric_name_string == ( "homeassistant_sensor_temperature_celsius{" 'domain="sensor",' 'entity="sensor.outside_temperature",' 'friendly_name="Outside Temperature"}' ) def test_entity_metric_generates_metric_string_with_value() -> None: """Test using EntityMetric to format a simple metric string but with a metric value included.""" domain = "sensor" object_id = "outside_temperature" entity_metric = EntityMetric( metric_name="homeassistant_sensor_temperature_celsius", domain=domain, friendly_name="Outside Temperature", entity=f"{domain}.{object_id}", ).withValue(17.2) assert entity_metric._metric_string == ( "homeassistant_sensor_temperature_celsius{" 'domain="sensor",' 'entity="sensor.outside_temperature",' 'friendly_name="Outside Temperature"}' " 17.2" ) def test_entity_metric_raises_exception_without_required_labels() -> None: """Test using EntityMetric to raise exception when required labels are missing.""" domain = "sensor" object_id = "outside_temperature" test_kwargs = { "metric_name": "homeassistant_sensor_temperature_celsius", "domain": domain, "friendly_name": "Outside Temperature", "entity": f"{domain}.{object_id}", } assert len(EntityMetric.required_labels()) > 0 for labelname in EntityMetric.required_labels(): label_kwargs = dict(test_kwargs) # Delete the required label and ensure we get an exception del label_kwargs[labelname] with pytest.raises(AssertionError): EntityMetric(**label_kwargs) def test_entity_metric_raises_exception_if_required_label_is_empty_string() -> None: """Test using EntityMetric to raise exception when required label value is empty string.""" domain = "sensor" object_id = "outside_temperature" test_kwargs = { "metric_name": "homeassistant_sensor_temperature_celsius", "domain": domain, "friendly_name": "Outside Temperature", "entity": f"{domain}.{object_id}", } assert len(EntityMetric.required_labels()) > 0 for labelname in EntityMetric.required_labels(): label_kwargs = dict(test_kwargs) # Replace the required label with "" and ensure we get an exception label_kwargs[labelname] = "" with pytest.raises(AssertionError): EntityMetric(**label_kwargs) def test_entity_metric_generates_alphabetically_ordered_labels() -> None: """Test using EntityMetric to format a simple metric string with labels alphabetically ordered.""" domain = "sensor" object_id = "outside_temperature" static_metric_string = ( "homeassistant_sensor_temperature_celsius{" 'domain="sensor",' 'entity="sensor.outside_temperature",' 'friendly_name="Outside Temperature",' 'zed_label="foo"' "}" " 17.2" ) ordered_entity_metric = EntityMetric( metric_name="homeassistant_sensor_temperature_celsius", domain=domain, entity=f"{domain}.{object_id}", friendly_name="Outside Temperature", zed_label="foo", ).withValue(17.2) assert ordered_entity_metric._metric_string == static_metric_string unordered_entity_metric = EntityMetric( metric_name="homeassistant_sensor_temperature_celsius", zed_label="foo", entity=f"{domain}.{object_id}", friendly_name="Outside Temperature", domain=domain, ).withValue(17.2) assert unordered_entity_metric._metric_string == static_metric_string def test_entity_metric_generates_metric_string_with_non_required_labels() -> None: """Test using EntityMetric to format a simple metric string but with extra labels and values included.""" mode_entity_metric = EntityMetric( metric_name="climate_preset_mode", domain="climate", friendly_name="Ecobee", entity="climate.ecobee", mode="away", ).withValue(1) assert mode_entity_metric._metric_string == ( "climate_preset_mode{" 'domain="climate",' 'entity="climate.ecobee",' 'friendly_name="Ecobee",' 'mode="away"' "}" " 1.0" ) action_entity_metric = EntityMetric( metric_name="climate_action", domain="climate", friendly_name="HeatPump", entity="climate.heatpump", action="heating", ).withValue(1) assert action_entity_metric._metric_string == ( "climate_action{" 'action="heating",' 'domain="climate",' 'entity="climate.heatpump",' 'friendly_name="HeatPump"' "}" " 1.0" ) state_entity_metric = EntityMetric( metric_name="cover_state", domain="cover", friendly_name="Curtain", entity="cover.curtain", state="open", ).withValue(1) assert state_entity_metric._metric_string == ( "cover_state{" 'domain="cover",' 'entity="cover.curtain",' 'friendly_name="Curtain",' 'state="open"' "}" " 1.0" ) foo_entity_metric = EntityMetric( metric_name="homeassistant_sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", foo="bar", ).withValue(17.2) assert foo_entity_metric._metric_string == ( "homeassistant_sensor_temperature_celsius{" 'domain="sensor",' 'entity="sensor.outside_temperature",' 'foo="bar",' 'friendly_name="Outside Temperature"' "}" " 17.2" ) def test_entity_metric_assert_helpers() -> None: """Test using EntityMetric for both assert_in_metrics and assert_not_in_metrics.""" temp_metric = ( "homeassistant_sensor_temperature_celsius{" 'domain="sensor",' 'entity="sensor.outside_temperature",' 'foo="bar",' 'friendly_name="Outside Temperature"' "}" ) climate_metric = ( "climate_preset_mode{" 'domain="climate",' 'entity="climate.ecobee",' 'friendly_name="Ecobee",' 'mode="away"' "}" ) excluded_cover_metric = ( "cover_state{" 'domain="cover",' 'entity="cover.curtain",' 'friendly_name="Curtain",' 'state="open"' "}" ) metrics = [ temp_metric, climate_metric, ] # First make sure the excluded metric is not present assert excluded_cover_metric not in metrics # now check for actual metrics temp_entity_metric = EntityMetric( metric_name="homeassistant_sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", foo="bar", ) assert temp_entity_metric._metric_name_string == temp_metric temp_entity_metric.assert_in_metrics(metrics) climate_entity_metric = EntityMetric( metric_name="climate_preset_mode", domain="climate", friendly_name="Ecobee", entity="climate.ecobee", mode="away", ) assert climate_entity_metric._metric_name_string == climate_metric climate_entity_metric.assert_in_metrics(metrics) excluded_cover_entity_metric = EntityMetric( metric_name="cover_state", domain="cover", friendly_name="Curtain", entity="cover.curtain", state="open", ) assert excluded_cover_entity_metric._metric_name_string == excluded_cover_metric excluded_cover_entity_metric.assert_not_in_metrics(metrics) def test_entity_metric_with_value_assert_helpers() -> None: """Test using EntityMetricWithValue helpers, which is only assert_in_metrics.""" temp_metric = ( "homeassistant_sensor_temperature_celsius{" 'domain="sensor",' 'entity="sensor.outside_temperature",' 'foo="bar",' 'friendly_name="Outside Temperature"' "}" " 17.2" ) climate_metric = ( "climate_preset_mode{" 'domain="climate",' 'entity="climate.ecobee",' 'friendly_name="Ecobee",' 'mode="away"' "}" " 1.0" ) metrics = [ temp_metric, climate_metric, ] temp_entity_metric = EntityMetric( metric_name="homeassistant_sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", foo="bar", ).withValue(17.2) assert temp_entity_metric._metric_string == temp_metric temp_entity_metric.assert_in_metrics(metrics) climate_entity_metric = EntityMetric( metric_name="climate_preset_mode", domain="climate", friendly_name="Ecobee", entity="climate.ecobee", mode="away", ).withValue(1) assert climate_entity_metric._metric_string == climate_metric climate_entity_metric.assert_in_metrics(metrics) @pytest.fixture(name="client") async def setup_prometheus_client( hass: HomeAssistant, hass_client: ClientSessionGenerator, namespace: str, ): """Initialize an hass_client with Prometheus component.""" # Reset registry prometheus_client.REGISTRY = prometheus_client.CollectorRegistry(auto_describe=True) prometheus_client.ProcessCollector(registry=prometheus_client.REGISTRY) prometheus_client.PlatformCollector(registry=prometheus_client.REGISTRY) prometheus_client.GCCollector(registry=prometheus_client.REGISTRY) config = {} if namespace is not None: config[prometheus.CONF_PROM_NAMESPACE] = namespace assert await async_setup_component( hass, prometheus.DOMAIN, {prometheus.DOMAIN: config} ) await hass.async_block_till_done() return await hass_client() async def generate_latest_metrics(client): """Generate the latest metrics and transform the body.""" resp = await client.get(prometheus.API_ENDPOINT) assert resp.status == HTTPStatus.OK assert resp.headers["content-type"] == CONTENT_TYPE_TEXT_PLAIN body = await resp.text() body = body.split("\n") assert len(body) > 3 return body @pytest.mark.parametrize("namespace", [""]) async def test_setup_enumeration( hass: HomeAssistant, hass_client: ClientSessionGenerator, entity_registry: er.EntityRegistry, namespace: str, ) -> None: """Test that setup enumerates existing states/entities.""" # The order of when things are created must be carefully controlled in # this test, so we don't use fixtures. sensor_1 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_1", unit_of_measurement=UnitOfTemperature.CELSIUS, original_device_class=SensorDeviceClass.TEMPERATURE, suggested_object_id="outside_temperature", original_name="Outside Temperature", ) state = 12.3 set_state_with_entry(hass, sensor_1, state, {}) assert await async_setup_component(hass, prometheus.DOMAIN, {prometheus.DOMAIN: {}}) client = await hass_client() body = await generate_latest_metrics(client) EntityMetric( metric_name="homeassistant_sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(state).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_view_empty_namespace( client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics view.""" body = await generate_latest_metrics(client) assert "# HELP python_info Python platform information" in body assert ( "# HELP python_gc_objects_collected_total Objects collected during gc" in body ) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Radio Energy", entity="sensor.radio_energy", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="last_updated_time_seconds", domain="sensor", friendly_name="Radio Energy", entity="sensor.radio_energy", ).withValue(86400.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [None]) async def test_view_default_namespace( client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics view.""" body = await generate_latest_metrics(client) assert "# HELP python_info Python platform information" in body assert ( "# HELP python_gc_objects_collected_total Objects collected during gc" in body ) EntityMetric( metric_name="homeassistant_sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(15.6).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_sensor_unit( client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for sensors with a unit.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="sensor_unit_kwh", domain="sensor", friendly_name="Television Energy", entity="sensor.television_energy", ).withValue(74.0).assert_in_metrics(body) EntityMetric( metric_name="sensor_unit_sek_per_kwh", domain="sensor", friendly_name="Electricity price", entity="sensor.electricity_price", ).withValue(0.123).assert_in_metrics(body) EntityMetric( metric_name="sensor_unit_u0xb0", domain="sensor", friendly_name="Wind Direction", entity="sensor.wind_direction", ).withValue(25.0).assert_in_metrics(body) EntityMetric( metric_name="sensor_unit_u0xb5g_per_mu0xb3", domain="sensor", friendly_name="SPS30 PM <1µm Weight concentration", entity="sensor.sps30_pm_1um_weight_concentration", ).withValue(3.7069).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_sensor_without_unit( client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for sensors without a unit.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="sensor_state", domain="sensor", friendly_name="Trend Gradient", entity="sensor.trend_gradient", ).withValue(0.002).assert_in_metrics(body) EntityMetric( metric_name="sensor_state", domain="sensor", friendly_name="Text", entity="sensor.text", ).assert_not_in_metrics(body) EntityMetric( metric_name="sensor_unit_text", domain="sensor", friendly_name="Text Unit", entity="sensor.text_unit", ).assert_not_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_sensor_device_class( client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for sensor with a device_class.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Fahrenheit", entity="sensor.fahrenheit", ).withValue(10.0).assert_in_metrics(body) EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(15.6).assert_in_metrics(body) EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="sensor_power_kwh", domain="sensor", friendly_name="Radio Energy", entity="sensor.radio_energy", ).withValue(14.0).assert_in_metrics(body) EntityMetric( metric_name="sensor_timestamp_seconds", domain="sensor", friendly_name="Timestamp", entity="sensor.timestamp", ).withValue(1.691445808136036e09).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_input_number( client: ClientSessionGenerator, input_number_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for input_number.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="input_number_state", domain="input_number", friendly_name="Threshold", entity="input_number.threshold", ).withValue(5.2).assert_in_metrics(body) EntityMetric( metric_name="input_number_state", domain="input_number", friendly_name="None", entity="input_number.brightness", ).withValue(60.0).assert_in_metrics(body) EntityMetric( metric_name="input_number_state_celsius", domain="input_number", friendly_name="Target temperature", entity="input_number.target_temperature", ).withValue(22.7).assert_in_metrics(body) EntityMetric( metric_name="input_number_state_celsius", domain="input_number", friendly_name="Converted temperature", entity="input_number.converted_temperature", ).withValue(100).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_number( client: ClientSessionGenerator, number_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for number.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="number_state", domain="number", friendly_name="Threshold", entity="number.threshold", ).withValue(5.2).assert_in_metrics(body) EntityMetric( metric_name="number_state", domain="number", friendly_name="None", entity="number.brightness", ).withValue(60.0).assert_in_metrics(body) EntityMetric( metric_name="number_state_celsius", domain="number", friendly_name="Target temperature", entity="number.target_temperature", ).withValue(22.7).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_battery( client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for battery.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="battery_level_percent", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(12.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_climate( client: ClientSessionGenerator, climate_entities: dict[str, er.RegistryEntry | dict[str, Any]], ) -> None: """Test prometheus metrics for climate entities.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="climate_current_temperature_celsius", domain="climate", friendly_name="HeatPump", entity="climate.heatpump", ).withValue(25.0).assert_in_metrics(body) EntityMetric( metric_name="climate_target_temperature_celsius", domain="climate", friendly_name="HeatPump", entity="climate.heatpump", ).withValue(20.0).assert_in_metrics(body) EntityMetric( metric_name="climate_target_temperature_low_celsius", domain="climate", friendly_name="Ecobee", entity="climate.ecobee", ).withValue(21.0).assert_in_metrics(body) EntityMetric( metric_name="climate_target_temperature_high_celsius", domain="climate", friendly_name="Ecobee", entity="climate.ecobee", ).withValue(24.0).assert_in_metrics(body) EntityMetric( metric_name="climate_target_temperature_celsius", domain="climate", friendly_name="Fritz!DECT", entity="climate.fritzdect", ).withValue(0.0).assert_in_metrics(body) EntityMetric( metric_name="climate_preset_mode", domain="climate", friendly_name="Ecobee", entity="climate.ecobee", mode="away", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="climate_fan_mode", domain="climate", friendly_name="Ecobee", entity="climate.ecobee", mode="auto", ).withValue(1).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_humidifier( client: ClientSessionGenerator, humidifier_entities: dict[str, er.RegistryEntry | dict[str, Any]], ) -> None: """Test prometheus metrics for humidifier entities.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="humidifier_target_humidity_percent", domain="humidifier", friendly_name="Humidifier", entity="humidifier.humidifier", ).withValue(68.0).assert_in_metrics(body) EntityMetric( metric_name="humidifier_state", domain="humidifier", friendly_name="Dehumidifier", entity="humidifier.dehumidifier", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="humidifier_mode", domain="humidifier", friendly_name="Hygrostat", entity="humidifier.hygrostat", mode="home", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="humidifier_mode", domain="humidifier", friendly_name="Hygrostat", entity="humidifier.hygrostat", mode="eco", ).withValue(0.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_attributes( client: ClientSessionGenerator, switch_entities: dict[str, er.RegistryEntry | dict[str, Any]], ) -> None: """Test prometheus metrics for entity attributes.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="switch_state", domain="switch", friendly_name="Boolean", entity="switch.boolean", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="switch_attr_boolean", domain="switch", friendly_name="Boolean", entity="switch.boolean", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="switch_state", domain="switch", friendly_name="Number", entity="switch.number", ).withValue(0.0).assert_in_metrics(body) EntityMetric( metric_name="switch_attr_number", domain="switch", friendly_name="Number", entity="switch.number", ).withValue(10.2).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_binary_sensor( client: ClientSessionGenerator, binary_sensor_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for binary_sensor.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="binary_sensor_state", domain="binary_sensor", friendly_name="Door", entity="binary_sensor.door", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="binary_sensor_state", domain="binary_sensor", friendly_name="Window", entity="binary_sensor.window", ).withValue(0.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_input_boolean( client: ClientSessionGenerator, input_boolean_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for input_boolean.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="input_boolean_state", domain="input_boolean", friendly_name="Test", entity="input_boolean.test", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="input_boolean_state", domain="input_boolean", friendly_name="Helper", entity="input_boolean.helper", ).withValue(0.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_light( client: ClientSessionGenerator, light_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for lights.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="light_brightness_percent", domain="light", friendly_name="Desk", entity="light.desk", ).withValue(100.0).assert_in_metrics(body) EntityMetric( metric_name="light_brightness_percent", domain="light", friendly_name="Wall", entity="light.wall", ).withValue(0.0).assert_in_metrics(body) EntityMetric( metric_name="light_brightness_percent", domain="light", friendly_name="TV", entity="light.tv", ).withValue(100.0).assert_in_metrics(body) EntityMetric( metric_name="light_brightness_percent", domain="light", friendly_name="PC", entity="light.pc", ).withValue(70.58823529411765).assert_in_metrics(body) EntityMetric( metric_name="light_brightness_percent", domain="light", friendly_name="Hallway", entity="light.hallway", ).withValue(100.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_lock( client: ClientSessionGenerator, lock_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for lock.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="lock_state", domain="lock", friendly_name="Front Door", entity="lock.front_door", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="lock_state", domain="lock", friendly_name="Kitchen Door", entity="lock.kitchen_door", ).withValue(0.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_fan( client: ClientSessionGenerator, fan_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for fan.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="fan_state", domain="fan", friendly_name="Fan 1", entity="fan.fan_1", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="fan_speed_percent", domain="fan", friendly_name="Fan 1", entity="fan.fan_1", ).withValue(33.0).assert_in_metrics(body) EntityMetric( metric_name="fan_is_oscillating", domain="fan", friendly_name="Fan 1", entity="fan.fan_1", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="fan_direction_reversed", domain="fan", friendly_name="Fan 1", entity="fan.fan_1", ).withValue(0.0).assert_in_metrics(body) EntityMetric( metric_name="fan_preset_mode", domain="fan", friendly_name="Fan 1", entity="fan.fan_1", mode="LO", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="fan_direction_reversed", domain="fan", friendly_name="Reverse Fan", entity="fan.fan_2", ).withValue(1).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_alarm_control_panel( client: ClientSessionGenerator, alarm_control_panel_entities: dict[str, er.RegistryEntry], ) -> None: """Test prometheus metrics for alarm control panel.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="alarm_control_panel_state", domain="alarm_control_panel", friendly_name="Alarm Control Panel 1", entity="alarm_control_panel.alarm_control_panel_1", state="armed_away", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="alarm_control_panel_state", domain="alarm_control_panel", friendly_name="Alarm Control Panel 1", entity="alarm_control_panel.alarm_control_panel_1", state="disarmed", ).withValue(0.0).assert_in_metrics(body) EntityMetric( metric_name="alarm_control_panel_state", domain="alarm_control_panel", friendly_name="Alarm Control Panel 2", entity="alarm_control_panel.alarm_control_panel_2", state="armed_home", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="alarm_control_panel_state", domain="alarm_control_panel", friendly_name="Alarm Control Panel 2", entity="alarm_control_panel.alarm_control_panel_2", state="armed_away", ).withValue(0.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_cover( client: ClientSessionGenerator, cover_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for cover.""" data = {**cover_entities} body = await generate_latest_metrics(client) open_covers = ["cover_open", "cover_position", "cover_tilt_position"] for testcover in data: EntityMetric( metric_name="cover_state", domain="cover", friendly_name=cover_entities[testcover].original_name, entity=cover_entities[testcover].entity_id, state="open", ).withValue( 1.0 if cover_entities[testcover].unique_id in open_covers else 0.0 ).assert_in_metrics(body) EntityMetric( metric_name="cover_state", domain="cover", friendly_name=cover_entities[testcover].original_name, entity=cover_entities[testcover].entity_id, state="closed", ).withValue( 1.0 if cover_entities[testcover].unique_id == "cover_closed" else 0.0 ).assert_in_metrics(body) EntityMetric( metric_name="cover_state", domain="cover", friendly_name=cover_entities[testcover].original_name, entity=cover_entities[testcover].entity_id, state="opening", ).withValue( 1.0 if cover_entities[testcover].unique_id == "cover_opening" else 0.0 ).assert_in_metrics(body) EntityMetric( metric_name="cover_state", domain="cover", friendly_name=cover_entities[testcover].original_name, entity=cover_entities[testcover].entity_id, state="closing", ).withValue( 1.0 if cover_entities[testcover].unique_id == "cover_closing" else 0.0 ).assert_in_metrics(body) if testcover == "cover_position": EntityMetric( metric_name="cover_position", domain="cover", friendly_name=cover_entities[testcover].original_name, entity=cover_entities[testcover].entity_id, ).withValue(50.0).assert_in_metrics(body) if testcover == "cover_tilt_position": EntityMetric( metric_name="cover_tilt_position", domain="cover", friendly_name=cover_entities[testcover].original_name, entity=cover_entities[testcover].entity_id, ).withValue(50.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_device_tracker( client: ClientSessionGenerator, device_tracker_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for device_tracker.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="device_tracker_state", domain="device_tracker", friendly_name="Phone", entity="device_tracker.phone", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="device_tracker_state", domain="device_tracker", friendly_name="Watch", entity="device_tracker.watch", ).withValue(0.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_person( client: ClientSessionGenerator, person_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for person.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="person_state", domain="person", friendly_name="Bob", entity="person.bob", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="person_state", domain="person", friendly_name="Alice", entity="person.alice", ).withValue(0.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_counter( client: ClientSessionGenerator, counter_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for counter.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="counter_value", domain="counter", friendly_name="None", entity="counter.counter", ).withValue(2.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_update( client: ClientSessionGenerator, update_entities: dict[str, er.RegistryEntry] ) -> None: """Test prometheus metrics for update.""" body = await generate_latest_metrics(client) EntityMetric( metric_name="update_state", domain="update", friendly_name="Firmware", entity="update.firmware", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="update_state", domain="update", friendly_name="Addon", entity="update.addon", ).withValue(0.0).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_renaming_entity_name( hass: HomeAssistant, entity_registry: er.EntityRegistry, client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry], climate_entities: dict[str, er.RegistryEntry | dict[str, Any]], ) -> None: """Test renaming entity name.""" data = {**sensor_entities, **climate_entities} body = await generate_latest_metrics(client) EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(15.6).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="climate_action", domain="climate", friendly_name="HeatPump", entity="climate.heatpump", action="heating", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="climate_action", domain="climate", friendly_name="HeatPump", entity="climate.heatpump", action="cooling", ).withValue(0.0).assert_in_metrics(body) assert "sensor.outside_temperature" in entity_registry.entities assert "climate.heatpump" in entity_registry.entities entity_registry.async_update_entity( entity_id=data["sensor_1"].entity_id, name="Outside Temperature Renamed", ) set_state_with_entry( hass, data["sensor_1"], 15.6, {ATTR_FRIENDLY_NAME: "Outside Temperature Renamed"}, ) entity_registry.async_update_entity( entity_id=data["climate_1"].entity_id, name="HeatPump Renamed", ) data["climate_1_attributes"] = { **data["climate_1_attributes"], ATTR_FRIENDLY_NAME: "HeatPump Renamed", } set_state_with_entry( hass, data["climate_1"], climate.HVACAction.HEATING, data["climate_1_attributes"], ) await hass.async_block_till_done() body = await generate_latest_metrics(client) # Check if old metrics deleted body_line = "\n".join(body) assert 'friendly_name="Outside Temperature"' not in body_line assert 'friendly_name="HeatPump"' not in body_line # Check if new metrics created EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature Renamed", entity="sensor.outside_temperature", ).withValue(15.6).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Temperature Renamed", entity="sensor.outside_temperature", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="climate_action", domain="climate", friendly_name="HeatPump Renamed", entity="climate.heatpump", action="heating", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="climate_action", domain="climate", friendly_name="HeatPump Renamed", entity="climate.heatpump", action="cooling", ).withValue(0.0).assert_in_metrics(body) # Keep other sensors EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_renaming_entity_id( hass: HomeAssistant, entity_registry: er.EntityRegistry, client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry], climate_entities: dict[str, er.RegistryEntry | dict[str, Any]], ) -> None: """Test renaming entity id.""" data = {**sensor_entities, **climate_entities} body = await generate_latest_metrics(client) EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(15.6).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) assert "sensor.outside_temperature" in entity_registry.entities assert "climate.heatpump" in entity_registry.entities entity_registry.async_update_entity( entity_id="sensor.outside_temperature", new_entity_id="sensor.outside_temperature_renamed", ) set_state_with_entry( hass, data["sensor_1"], 15.6, None, "sensor.outside_temperature_renamed" ) await hass.async_block_till_done() body = await generate_latest_metrics(client) # Check if old metrics deleted body_line = "\n".join(body) assert 'entity="sensor.outside_temperature"' not in body_line # Check if new metrics created EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature_renamed", ).withValue(15.6).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature_renamed", ).withValue(1).assert_in_metrics(body) # Keep other sensors EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_deleting_entity( hass: HomeAssistant, entity_registry: er.EntityRegistry, client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry], climate_entities: dict[str, er.RegistryEntry | dict[str, Any]], ) -> None: """Test deleting a entity.""" data = {**sensor_entities, **climate_entities} body = await generate_latest_metrics(client) EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(15.6).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="climate_action", domain="climate", friendly_name="HeatPump", entity="climate.heatpump", action="heating", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="climate_action", domain="climate", friendly_name="HeatPump", entity="climate.heatpump", action="cooling", ).withValue(0.0).assert_in_metrics(body) assert "sensor.outside_temperature" in entity_registry.entities assert "climate.heatpump" in entity_registry.entities entity_registry.async_remove(data["sensor_1"].entity_id) entity_registry.async_remove(data["climate_1"].entity_id) await hass.async_block_till_done() body = await generate_latest_metrics(client) # Check if old metrics deleted body_line = "\n".join(body) assert 'entity="sensor.outside_temperature"' not in body_line assert 'friendly_name="Outside Temperature"' not in body_line assert 'entity="climate.heatpump"' not in body_line assert 'friendly_name="HeatPump"' not in body_line # Keep other sensors EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) async def test_disabling_entity( hass: HomeAssistant, entity_registry: er.EntityRegistry, client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry], climate_entities: dict[str, er.RegistryEntry | dict[str, Any]], ) -> None: """Test disabling a entity.""" data = {**sensor_entities, **climate_entities} await hass.async_block_till_done() body = await generate_latest_metrics(client) EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(15.6).assert_in_metrics(body) EntityMetric( metric_name="state_change_total", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="state_change_created", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).assert_in_metrics(body) EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="climate_action", domain="climate", friendly_name="HeatPump", entity="climate.heatpump", action="heating", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="climate_action", domain="climate", friendly_name="HeatPump", entity="climate.heatpump", action="cooling", ).withValue(0.0).assert_in_metrics(body) assert "sensor.outside_temperature" in entity_registry.entities assert "climate.heatpump" in entity_registry.entities entity_registry.async_update_entity( entity_id=data["sensor_1"].entity_id, disabled_by=er.RegistryEntryDisabler.USER, ) entity_registry.async_update_entity( entity_id="climate.heatpump", disabled_by=er.RegistryEntryDisabler.USER, ) await hass.async_block_till_done() body = await generate_latest_metrics(client) # Check if old metrics deleted body_line = "\n".join(body) assert 'entity="sensor.outside_temperature"' not in body_line assert 'friendly_name="Outside Temperature"' not in body_line assert 'entity="climate.heatpump"' not in body_line assert 'friendly_name="HeatPump"' not in body_line # Keep other sensors EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) @pytest.mark.parametrize("namespace", [""]) @pytest.mark.parametrize("unavailable_state", [STATE_UNAVAILABLE, STATE_UNKNOWN]) async def test_entity_becomes_unavailable( hass: HomeAssistant, entity_registry: er.EntityRegistry, client: ClientSessionGenerator, sensor_entities: dict[str, er.RegistryEntry], unavailable_state: str, ) -> None: """Test an entity that becomes unavailable/unknown is no longer exported.""" data = {**sensor_entities} await hass.async_block_till_done() body = await generate_latest_metrics(client) EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(15.6).assert_in_metrics(body) EntityMetric( metric_name="state_change_total", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="last_updated_time_seconds", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).assert_in_metrics(body) EntityMetric( metric_name="battery_level_percent", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(12.0).assert_in_metrics(body) EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="state_change_total", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) # Make sensor_1 unavailable/unknown. set_state_with_entry( hass, data["sensor_1"], unavailable_state, data["sensor_1_attributes"] ) await hass.async_block_till_done() body = await generate_latest_metrics(client) # Check that the availability changed on sensor_1 and the metric with the value is gone. EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).assert_not_in_metrics(body) EntityMetric( metric_name="battery_level_percent", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).assert_not_in_metrics(body) EntityMetric( metric_name="state_change_total", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(2.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(0.0).assert_in_metrics(body) EntityMetric( metric_name="last_updated_time_seconds", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).assert_in_metrics(body) # The other sensor should be unchanged. EntityMetric( metric_name="sensor_humidity_percent", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(54.0).assert_in_metrics(body) EntityMetric( metric_name="state_change_total", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Humidity", entity="sensor.outside_humidity", ).withValue(1).assert_in_metrics(body) # Bring sensor_1 back and check that it returned. set_state_with_entry(hass, data["sensor_1"], 201.0, data["sensor_1_attributes"]) await hass.async_block_till_done() body = await generate_latest_metrics(client) EntityMetric( metric_name="sensor_temperature_celsius", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(201.0).assert_in_metrics(body) EntityMetric( metric_name="battery_level_percent", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(12.0).assert_in_metrics(body) EntityMetric( metric_name="state_change_total", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(3.0).assert_in_metrics(body) EntityMetric( metric_name="entity_available", domain="sensor", friendly_name="Outside Temperature", entity="sensor.outside_temperature", ).withValue(1).assert_in_metrics(body) @pytest.fixture(name="sensor_entities") async def sensor_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate sensor entities.""" data = {} sensor_1 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_1", unit_of_measurement=UnitOfTemperature.CELSIUS, original_device_class=SensorDeviceClass.TEMPERATURE, suggested_object_id="outside_temperature", original_name="Outside Temperature", ) sensor_1_attributes = {ATTR_BATTERY_LEVEL: 12} set_state_with_entry(hass, sensor_1, 15.6, sensor_1_attributes) data["sensor_1"] = sensor_1 data["sensor_1_attributes"] = sensor_1_attributes sensor_2 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_2", unit_of_measurement=PERCENTAGE, original_device_class=SensorDeviceClass.HUMIDITY, suggested_object_id="outside_humidity", original_name="Outside Humidity", ) set_state_with_entry(hass, sensor_2, 54.0) data["sensor_2"] = sensor_2 sensor_3 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_3", unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR, original_device_class=SensorDeviceClass.POWER, suggested_object_id="radio_energy", original_name="Radio Energy", ) with freeze_time(datetime.datetime(1970, 1, 2, tzinfo=dt_util.UTC)): set_state_with_entry(hass, sensor_3, 14) data["sensor_3"] = sensor_3 sensor_4 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_4", unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR, suggested_object_id="television_energy", original_name="Television Energy", ) set_state_with_entry(hass, sensor_4, 74) data["sensor_4"] = sensor_4 sensor_5 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_5", unit_of_measurement=f"SEK/{UnitOfEnergy.KILO_WATT_HOUR}", suggested_object_id="electricity_price", original_name="Electricity price", ) set_state_with_entry(hass, sensor_5, 0.123) data["sensor_5"] = sensor_5 sensor_6 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_6", unit_of_measurement=DEGREE, suggested_object_id="wind_direction", original_name="Wind Direction", ) set_state_with_entry(hass, sensor_6, 25) data["sensor_6"] = sensor_6 sensor_7 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_7", unit_of_measurement=CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, suggested_object_id="sps30_pm_1um_weight_concentration", original_name="SPS30 PM <1µm Weight concentration", ) set_state_with_entry(hass, sensor_7, 3.7069) data["sensor_7"] = sensor_7 sensor_8 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_8", suggested_object_id="trend_gradient", original_name="Trend Gradient", ) set_state_with_entry(hass, sensor_8, 0.002) data["sensor_8"] = sensor_8 sensor_9 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_9", suggested_object_id="text", original_name="Text", ) set_state_with_entry(hass, sensor_9, "should_not_work") data["sensor_9"] = sensor_9 sensor_10 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_10", unit_of_measurement="Text", suggested_object_id="text_unit", original_name="Text Unit", ) set_state_with_entry(hass, sensor_10, "should_not_work") data["sensor_10"] = sensor_10 sensor_11 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_11", unit_of_measurement=UnitOfTemperature.FAHRENHEIT, original_device_class=SensorDeviceClass.TEMPERATURE, suggested_object_id="fahrenheit", original_name="Fahrenheit", ) set_state_with_entry(hass, sensor_11, 50) data["sensor_11"] = sensor_11 sensor_12 = entity_registry.async_get_or_create( domain=sensor.DOMAIN, platform="test", unique_id="sensor_12", original_device_class=SensorDeviceClass.TIMESTAMP, suggested_object_id="Timestamp", original_name="Timestamp", ) set_state_with_entry(hass, sensor_12, "2023-08-07T15:03:28.136036-0700") data["sensor_12"] = sensor_12 await hass.async_block_till_done() return data @pytest.fixture(name="climate_entities") async def climate_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry | dict[str, Any]]: """Simulate climate entities.""" data = {} climate_1 = entity_registry.async_get_or_create( domain=climate.DOMAIN, platform="test", unique_id="climate_1", unit_of_measurement=UnitOfTemperature.CELSIUS, suggested_object_id="heatpump", original_name="HeatPump", ) climate_1_attributes = { ATTR_TEMPERATURE: 20, ATTR_CURRENT_TEMPERATURE: 25, ATTR_HVAC_ACTION: climate.HVACAction.HEATING, } set_state_with_entry( hass, climate_1, climate.HVACAction.HEATING, climate_1_attributes ) data["climate_1"] = climate_1 data["climate_1_attributes"] = climate_1_attributes climate_2 = entity_registry.async_get_or_create( domain=climate.DOMAIN, platform="test", unique_id="climate_2", unit_of_measurement=UnitOfTemperature.CELSIUS, suggested_object_id="ecobee", original_name="Ecobee", ) climate_2_attributes = { ATTR_TEMPERATURE: 21, ATTR_CURRENT_TEMPERATURE: 22, ATTR_TARGET_TEMP_LOW: 21, ATTR_TARGET_TEMP_HIGH: 24, ATTR_HVAC_ACTION: climate.HVACAction.COOLING, ATTR_HVAC_MODES: ["off", "heat", "cool", "heat_cool"], ATTR_PRESET_MODE: "away", ATTR_PRESET_MODES: ["away", "home", "sleep"], ATTR_FAN_MODE: "auto", ATTR_FAN_MODES: ["auto", "on"], } set_state_with_entry( hass, climate_2, climate.HVACAction.HEATING, climate_2_attributes ) data["climate_2"] = climate_2 data["climate_2_attributes"] = climate_2_attributes climate_3 = entity_registry.async_get_or_create( domain=climate.DOMAIN, platform="test", unique_id="climate_3", unit_of_measurement=UnitOfTemperature.CELSIUS, suggested_object_id="fritzdect", original_name="Fritz!DECT", ) climate_3_attributes = { ATTR_TEMPERATURE: 0, ATTR_CURRENT_TEMPERATURE: 22, ATTR_HVAC_ACTION: climate.HVACAction.OFF, } set_state_with_entry(hass, climate_3, climate.HVACAction.OFF, climate_3_attributes) data["climate_3"] = climate_3 data["climate_3_attributes"] = climate_3_attributes await hass.async_block_till_done() return data @pytest.fixture(name="humidifier_entities") async def humidifier_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry | dict[str, Any]]: """Simulate humidifier entities.""" data = {} humidifier_1 = entity_registry.async_get_or_create( domain=humidifier.DOMAIN, platform="test", unique_id="humidifier_1", original_device_class=humidifier.HumidifierDeviceClass.HUMIDIFIER, suggested_object_id="humidifier", original_name="Humidifier", ) humidifier_1_attributes = { ATTR_HUMIDITY: 68, } set_state_with_entry(hass, humidifier_1, STATE_ON, humidifier_1_attributes) data["humidifier_1"] = humidifier_1 data["humidifier_1_attributes"] = humidifier_1_attributes humidifier_2 = entity_registry.async_get_or_create( domain=humidifier.DOMAIN, platform="test", unique_id="humidifier_2", original_device_class=humidifier.HumidifierDeviceClass.DEHUMIDIFIER, suggested_object_id="dehumidifier", original_name="Dehumidifier", ) humidifier_2_attributes = { ATTR_HUMIDITY: 54, } set_state_with_entry(hass, humidifier_2, STATE_ON, humidifier_2_attributes) data["humidifier_2"] = humidifier_2 data["humidifier_2_attributes"] = humidifier_2_attributes humidifier_3 = entity_registry.async_get_or_create( domain=humidifier.DOMAIN, platform="test", unique_id="humidifier_3", suggested_object_id="hygrostat", original_name="Hygrostat", ) humidifier_3_attributes = { ATTR_HUMIDITY: 50, ATTR_MODE: "home", ATTR_AVAILABLE_MODES: ["home", "eco"], } set_state_with_entry(hass, humidifier_3, STATE_ON, humidifier_3_attributes) data["humidifier_3"] = humidifier_3 data["humidifier_3_attributes"] = humidifier_3_attributes await hass.async_block_till_done() return data @pytest.fixture(name="lock_entities") async def lock_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate lock entities.""" data = {} lock_1 = entity_registry.async_get_or_create( domain=lock.DOMAIN, platform="test", unique_id="lock_1", suggested_object_id="front_door", original_name="Front Door", ) set_state_with_entry(hass, lock_1, LockState.LOCKED) data["lock_1"] = lock_1 lock_2 = entity_registry.async_get_or_create( domain=lock.DOMAIN, platform="test", unique_id="lock_2", suggested_object_id="kitchen_door", original_name="Kitchen Door", ) set_state_with_entry(hass, lock_2, LockState.UNLOCKED) data["lock_2"] = lock_2 await hass.async_block_till_done() return data @pytest.fixture(name="cover_entities") async def cover_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate cover entities.""" data = {} cover_open = entity_registry.async_get_or_create( domain=cover.DOMAIN, platform="test", unique_id="cover_open", suggested_object_id="open_shade", original_name="Open Shade", ) set_state_with_entry(hass, cover_open, STATE_OPEN) data["cover_open"] = cover_open cover_closed = entity_registry.async_get_or_create( domain=cover.DOMAIN, platform="test", unique_id="cover_closed", suggested_object_id="closed_shade", original_name="Closed Shade", ) set_state_with_entry(hass, cover_closed, STATE_CLOSED) data["cover_closed"] = cover_closed cover_closing = entity_registry.async_get_or_create( domain=cover.DOMAIN, platform="test", unique_id="cover_closing", suggested_object_id="closing_shade", original_name="Closing Shade", ) set_state_with_entry(hass, cover_closing, STATE_CLOSING) data["cover_closing"] = cover_closing cover_opening = entity_registry.async_get_or_create( domain=cover.DOMAIN, platform="test", unique_id="cover_opening", suggested_object_id="opening_shade", original_name="Opening Shade", ) set_state_with_entry(hass, cover_opening, STATE_OPENING) data["cover_opening"] = cover_opening cover_position = entity_registry.async_get_or_create( domain=cover.DOMAIN, platform="test", unique_id="cover_position", suggested_object_id="position_shade", original_name="Position Shade", ) cover_position_attributes = {cover.ATTR_CURRENT_POSITION: 50} set_state_with_entry(hass, cover_position, STATE_OPEN, cover_position_attributes) data["cover_position"] = cover_position cover_tilt_position = entity_registry.async_get_or_create( domain=cover.DOMAIN, platform="test", unique_id="cover_tilt_position", suggested_object_id="tilt_position_shade", original_name="Tilt Position Shade", ) cover_tilt_position_attributes = {cover.ATTR_CURRENT_TILT_POSITION: 50} set_state_with_entry( hass, cover_tilt_position, STATE_OPEN, cover_tilt_position_attributes ) data["cover_tilt_position"] = cover_tilt_position await hass.async_block_till_done() return data @pytest.fixture(name="input_number_entities") async def input_number_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate input_number entities.""" data = {} input_number_1 = entity_registry.async_get_or_create( domain=input_number.DOMAIN, platform="test", unique_id="input_number_1", suggested_object_id="threshold", original_name="Threshold", ) set_state_with_entry(hass, input_number_1, 5.2) data["input_number_1"] = input_number_1 input_number_2 = entity_registry.async_get_or_create( domain=input_number.DOMAIN, platform="test", unique_id="input_number_2", suggested_object_id="brightness", ) set_state_with_entry(hass, input_number_2, 60) data["input_number_2"] = input_number_2 input_number_3 = entity_registry.async_get_or_create( domain=input_number.DOMAIN, platform="test", unique_id="input_number_3", suggested_object_id="target_temperature", original_name="Target temperature", unit_of_measurement=UnitOfTemperature.CELSIUS, ) set_state_with_entry(hass, input_number_3, 22.7) data["input_number_3"] = input_number_3 input_number_4 = entity_registry.async_get_or_create( domain=input_number.DOMAIN, platform="test", unique_id="input_number_4", suggested_object_id="converted_temperature", original_name="Converted temperature", unit_of_measurement=UnitOfTemperature.FAHRENHEIT, ) set_state_with_entry(hass, input_number_4, 212) data["input_number_4"] = input_number_4 await hass.async_block_till_done() return data @pytest.fixture(name="number_entities") async def number_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate number entities.""" data = {} number_1 = entity_registry.async_get_or_create( domain=number.DOMAIN, platform="test", unique_id="number_1", suggested_object_id="threshold", original_name="Threshold", ) set_state_with_entry(hass, number_1, 5.2) data["number_1"] = number_1 number_2 = entity_registry.async_get_or_create( domain=number.DOMAIN, platform="test", unique_id="number_2", suggested_object_id="brightness", ) set_state_with_entry(hass, number_2, 60) data["number_2"] = number_2 number_3 = entity_registry.async_get_or_create( domain=number.DOMAIN, platform="test", unique_id="number_3", suggested_object_id="target_temperature", original_name="Target temperature", unit_of_measurement=UnitOfTemperature.CELSIUS, ) set_state_with_entry(hass, number_3, 22.7) data["number_3"] = number_3 await hass.async_block_till_done() return data @pytest.fixture(name="input_boolean_entities") async def input_boolean_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate input_boolean entities.""" data = {} input_boolean_1 = entity_registry.async_get_or_create( domain=input_boolean.DOMAIN, platform="test", unique_id="input_boolean_1", suggested_object_id="test", original_name="Test", ) set_state_with_entry(hass, input_boolean_1, STATE_ON) data["input_boolean_1"] = input_boolean_1 input_boolean_2 = entity_registry.async_get_or_create( domain=input_boolean.DOMAIN, platform="test", unique_id="input_boolean_2", suggested_object_id="helper", original_name="Helper", ) set_state_with_entry(hass, input_boolean_2, STATE_OFF) data["input_boolean_2"] = input_boolean_2 await hass.async_block_till_done() return data @pytest.fixture(name="binary_sensor_entities") async def binary_sensor_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate binary_sensor entities.""" data = {} binary_sensor_1 = entity_registry.async_get_or_create( domain=binary_sensor.DOMAIN, platform="test", unique_id="binary_sensor_1", suggested_object_id="door", original_name="Door", ) set_state_with_entry(hass, binary_sensor_1, STATE_ON) data["binary_sensor_1"] = binary_sensor_1 binary_sensor_2 = entity_registry.async_get_or_create( domain=binary_sensor.DOMAIN, platform="test", unique_id="binary_sensor_2", suggested_object_id="window", original_name="Window", ) set_state_with_entry(hass, binary_sensor_2, STATE_OFF) data["binary_sensor_2"] = binary_sensor_2 await hass.async_block_till_done() return data @pytest.fixture(name="light_entities") async def light_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate light entities.""" data = {} light_1 = entity_registry.async_get_or_create( domain=light.DOMAIN, platform="test", unique_id="light_1", suggested_object_id="desk", original_name="Desk", ) set_state_with_entry(hass, light_1, STATE_ON) data["light_1"] = light_1 light_2 = entity_registry.async_get_or_create( domain=light.DOMAIN, platform="test", unique_id="light_2", suggested_object_id="wall", original_name="Wall", ) set_state_with_entry(hass, light_2, STATE_OFF) data["light_2"] = light_2 light_3 = entity_registry.async_get_or_create( domain=light.DOMAIN, platform="test", unique_id="light_3", suggested_object_id="tv", original_name="TV", ) light_3_attributes = {light.ATTR_BRIGHTNESS: 255} set_state_with_entry(hass, light_3, STATE_ON, light_3_attributes) data["light_3"] = light_3 data["light_3_attributes"] = light_3_attributes light_4 = entity_registry.async_get_or_create( domain=light.DOMAIN, platform="test", unique_id="light_4", suggested_object_id="pc", original_name="PC", ) light_4_attributes = {light.ATTR_BRIGHTNESS: 180} set_state_with_entry(hass, light_4, STATE_ON, light_4_attributes) data["light_4"] = light_4 data["light_4_attributes"] = light_4_attributes light_5 = entity_registry.async_get_or_create( domain=light.DOMAIN, platform="test", unique_id="light_5", suggested_object_id="hallway", original_name="Hallway", ) # Light is on, but brightness is unset; expect metrics to report # brightness of 100%. light_5_attributes = {light.ATTR_BRIGHTNESS: None} set_state_with_entry(hass, light_5, STATE_ON, light_5_attributes) data["light_5"] = light_5 data["light_5_attributes"] = light_5_attributes await hass.async_block_till_done() return data @pytest.fixture(name="switch_entities") async def switch_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry | dict[str, Any]]: """Simulate switch entities.""" data = {} switch_1 = entity_registry.async_get_or_create( domain=switch.DOMAIN, platform="test", unique_id="switch_1", suggested_object_id="boolean", original_name="Boolean", ) switch_1_attributes = {"boolean": True} set_state_with_entry(hass, switch_1, STATE_ON, switch_1_attributes) data["switch_1"] = switch_1 data["switch_1_attributes"] = switch_1_attributes switch_2 = entity_registry.async_get_or_create( domain=switch.DOMAIN, platform="test", unique_id="switch_2", suggested_object_id="number", original_name="Number", ) switch_2_attributes = {"Number": 10.2} set_state_with_entry(hass, switch_2, STATE_OFF, switch_2_attributes) data["switch_2"] = switch_2 data["switch_2_attributes"] = switch_2_attributes await hass.async_block_till_done() return data @pytest.fixture(name="fan_entities") async def fan_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate fan entities.""" data = {} fan_1 = entity_registry.async_get_or_create( domain=fan.DOMAIN, platform="test", unique_id="fan_1", suggested_object_id="fan_1", original_name="Fan 1", ) fan_1_attributes = { ATTR_DIRECTION: DIRECTION_FORWARD, ATTR_OSCILLATING: True, ATTR_PERCENTAGE: 33, ATTR_PRESET_MODE: "LO", ATTR_PRESET_MODES: ["LO", "OFF", "HI"], } set_state_with_entry(hass, fan_1, STATE_ON, fan_1_attributes) data["fan_1"] = fan_1 data["fan_1_attributes"] = fan_1_attributes fan_2 = entity_registry.async_get_or_create( domain=fan.DOMAIN, platform="test", unique_id="fan_2", suggested_object_id="fan_2", original_name="Reverse Fan", ) fan_2_attributes = {ATTR_DIRECTION: DIRECTION_REVERSE} set_state_with_entry(hass, fan_2, STATE_ON, fan_2_attributes) data["fan_2"] = fan_2 data["fan_2_attributes"] = fan_2_attributes await hass.async_block_till_done() return data @pytest.fixture(name="alarm_control_panel_entities") async def alarm_control_panel_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate alarm control panel entities.""" data = {} alarm_control_panel_1 = entity_registry.async_get_or_create( domain=alarm_control_panel.DOMAIN, platform="test", unique_id="alarm_control_panel_1", suggested_object_id="alarm_control_panel_1", original_name="Alarm Control Panel 1", ) set_state_with_entry(hass, alarm_control_panel_1, AlarmControlPanelState.ARMED_AWAY) data["alarm_control_panel_1"] = alarm_control_panel_1 alarm_control_panel_2 = entity_registry.async_get_or_create( domain=alarm_control_panel.DOMAIN, platform="test", unique_id="alarm_control_panel_2", suggested_object_id="alarm_control_panel_2", original_name="Alarm Control Panel 2", ) set_state_with_entry(hass, alarm_control_panel_2, AlarmControlPanelState.ARMED_HOME) data["alarm_control_panel_2"] = alarm_control_panel_2 await hass.async_block_till_done() return data @pytest.fixture(name="person_entities") async def person_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate person entities.""" data = {} person_1 = entity_registry.async_get_or_create( domain=person.DOMAIN, platform="test", unique_id="person_1", suggested_object_id="bob", original_name="Bob", ) set_state_with_entry(hass, person_1, STATE_HOME) data["person_1"] = person_1 person_2 = entity_registry.async_get_or_create( domain=person.DOMAIN, platform="test", unique_id="person_2", suggested_object_id="alice", original_name="Alice", ) set_state_with_entry(hass, person_2, STATE_NOT_HOME) data["person_2"] = person_2 await hass.async_block_till_done() return data @pytest.fixture(name="device_tracker_entities") async def device_tracker_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate device_tracker entities.""" data = {} device_tracker_1 = entity_registry.async_get_or_create( domain=device_tracker.DOMAIN, platform="test", unique_id="device_tracker_1", suggested_object_id="phone", original_name="Phone", ) set_state_with_entry(hass, device_tracker_1, STATE_HOME) data["device_tracker_1"] = device_tracker_1 device_tracker_2 = entity_registry.async_get_or_create( domain=device_tracker.DOMAIN, platform="test", unique_id="device_tracker_2", suggested_object_id="watch", original_name="Watch", ) set_state_with_entry(hass, device_tracker_2, STATE_NOT_HOME) data["device_tracker_2"] = device_tracker_2 await hass.async_block_till_done() return data @pytest.fixture(name="counter_entities") async def counter_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate counter entities.""" data = {} counter_1 = entity_registry.async_get_or_create( domain=counter.DOMAIN, platform="test", unique_id="counter_1", suggested_object_id="counter", ) set_state_with_entry(hass, counter_1, 2) data["counter_1"] = counter_1 await hass.async_block_till_done() return data @pytest.fixture(name="update_entities") async def update_fixture( hass: HomeAssistant, entity_registry: er.EntityRegistry ) -> dict[str, er.RegistryEntry]: """Simulate update entities.""" data = {} update_1 = entity_registry.async_get_or_create( domain=update.DOMAIN, platform="test", unique_id="update_1", suggested_object_id="firmware", original_name="Firmware", ) set_state_with_entry(hass, update_1, STATE_ON) data["update_1"] = update_1 update_2 = entity_registry.async_get_or_create( domain=update.DOMAIN, platform="test", unique_id="update_2", suggested_object_id="addon", original_name="Addon", ) set_state_with_entry(hass, update_2, STATE_OFF) data["update_2"] = update_2 await hass.async_block_till_done() return data def set_state_with_entry( hass: HomeAssistant, entry: er.RegistryEntry, state, additional_attributes=None, new_entity_id=None, ): """Set the state of an entity with an Entity Registry entry.""" attributes = {} if entry.original_name: attributes[ATTR_FRIENDLY_NAME] = entry.original_name if entry.unit_of_measurement: attributes[ATTR_UNIT_OF_MEASUREMENT] = entry.unit_of_measurement if entry.original_device_class: attributes[ATTR_DEVICE_CLASS] = entry.original_device_class if additional_attributes: attributes = {**attributes, **additional_attributes} hass.states.async_set( entity_id=new_entity_id if new_entity_id else entry.entity_id, new_state=state, attributes=attributes, ) @pytest.fixture(name="mock_client") def mock_client_fixture(): """Mock the prometheus client.""" with mock.patch(f"{PROMETHEUS_PATH}.prometheus_client") as client: counter_client = mock.MagicMock() client.Counter = mock.MagicMock(return_value=counter_client) setattr(counter_client, "labels", mock.MagicMock(return_value=mock.MagicMock())) yield counter_client async def test_minimal_config(hass: HomeAssistant, mock_client: mock.MagicMock) -> None: """Test the minimal config and defaults of component.""" config = {prometheus.DOMAIN: {}} assert await async_setup_component(hass, prometheus.DOMAIN, config) await hass.async_block_till_done() async def test_full_config(hass: HomeAssistant, mock_client: mock.MagicMock) -> None: """Test the full config of component.""" config = { prometheus.DOMAIN: { "namespace": "ns", "default_metric": "m", "override_metric": "m", "requires_auth": False, "component_config": {"fake.test": {"override_metric": "km"}}, "component_config_glob": {"fake.time_*": {"override_metric": "h"}}, "component_config_domain": {"climate": {"override_metric": "°C"}}, "filter": { "include_domains": ["climate"], "include_entity_globs": ["fake.time_*"], "include_entities": ["fake.test"], "exclude_domains": ["script"], "exclude_entity_globs": ["climate.excluded_*"], "exclude_entities": ["fake.time_excluded"], }, } } assert await async_setup_component(hass, prometheus.DOMAIN, config) await hass.async_block_till_done() async def _setup(hass: HomeAssistant, filter_config): """Shared set up for filtering tests.""" config = {prometheus.DOMAIN: {"filter": filter_config}} assert await async_setup_component(hass, prometheus.DOMAIN, config) await hass.async_block_till_done() async def test_allowlist(hass: HomeAssistant, mock_client: mock.MagicMock) -> None: """Test an allowlist only config.""" await _setup( hass, { "include_domains": ["fake"], "include_entity_globs": ["test.included_*"], "include_entities": ["not_real.included"], }, ) tests = [ FilterTest("climate.excluded", False), FilterTest("fake.included", True), FilterTest("test.excluded_test", False), FilterTest("test.included_test", True), FilterTest("not_real.included", True), FilterTest("not_real.excluded", False), ] for test in tests: hass.states.async_set(test.id, "not blank") await hass.async_block_till_done() was_called = mock_client.labels.call_count == 1 assert test.should_pass == was_called mock_client.labels.reset_mock() async def test_denylist(hass: HomeAssistant, mock_client: mock.MagicMock) -> None: """Test a denylist only config.""" await _setup( hass, { "exclude_domains": ["fake"], "exclude_entity_globs": ["test.excluded_*"], "exclude_entities": ["not_real.excluded"], }, ) tests = [ FilterTest("fake.excluded", False), FilterTest("light.included", True), FilterTest("test.excluded_test", False), FilterTest("test.included_test", True), FilterTest("not_real.included", True), FilterTest("not_real.excluded", False), ] for test in tests: hass.states.async_set(test.id, "not blank") await hass.async_block_till_done() was_called = mock_client.labels.call_count == 1 assert test.should_pass == was_called mock_client.labels.reset_mock() async def test_filtered_denylist( hass: HomeAssistant, mock_client: mock.MagicMock ) -> None: """Test a denylist config with a filtering allowlist.""" await _setup( hass, { "include_entities": ["fake.included", "test.excluded_test"], "exclude_domains": ["fake"], "exclude_entity_globs": ["*.excluded_*"], "exclude_entities": ["not_real.excluded"], }, ) tests = [ FilterTest("fake.excluded", False), FilterTest("fake.included", True), FilterTest("alt_fake.excluded_test", False), FilterTest("test.excluded_test", True), FilterTest("not_real.excluded", False), FilterTest("not_real.included", True), ] for test in tests: hass.states.async_set(test.id, "not blank") await hass.async_block_till_done() was_called = mock_client.labels.call_count == 1 assert test.should_pass == was_called mock_client.labels.reset_mock()