core/tests/components/zha/test_sensor.py

"""Test ZHA sensor."""

from collections.abc import Callable
from datetime import timedelta
import math
from typing import Any
from unittest.mock import MagicMock, patch

import pytest
from zhaquirks.danfoss import thermostat as danfoss_thermostat
import zigpy.profiles.zha
from zigpy.quirks import CustomCluster
from zigpy.quirks.v2 import CustomDeviceV2, add_to_registry_v2
from zigpy.quirks.v2.homeassistant import UnitOfMass
import zigpy.types as t
from zigpy.zcl.clusters import general, homeautomation, hvac, measurement, smartenergy
from zigpy.zcl.clusters.hvac import Thermostat
from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster

from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.components.zha.core import ZHADevice
from homeassistant.components.zha.core.const import ZHA_CLUSTER_HANDLER_READS_PER_REQ
import homeassistant.config as config_util
from homeassistant.const import (
    ATTR_DEVICE_CLASS,
    ATTR_UNIT_OF_MEASUREMENT,
    CONF_UNIT_SYSTEM,
    LIGHT_LUX,
    PERCENTAGE,
    STATE_UNAVAILABLE,
    STATE_UNKNOWN,
    Platform,
    UnitOfApparentPower,
    UnitOfElectricCurrent,
    UnitOfElectricPotential,
    UnitOfEnergy,
    UnitOfPower,
    UnitOfPressure,
    UnitOfTemperature,
    UnitOfVolume,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er, restore_state
from homeassistant.helpers.entity_component import async_update_entity
from homeassistant.util import dt as dt_util

from .common import (
    async_enable_traffic,
    async_test_rejoin,
    find_entity_id,
    find_entity_ids,
    send_attribute_report,
    send_attributes_report,
)
from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE

from tests.common import (
    MockConfigEntry,
    async_fire_time_changed,
    async_mock_load_restore_state_from_storage,
)

ENTITY_ID_PREFIX = "sensor.fakemanufacturer_fakemodel_{}"


@pytest.fixture(autouse=True)
def sensor_platform_only():
    """Only set up the sensor and required base platforms to speed up tests."""
    with patch(
        "homeassistant.components.zha.PLATFORMS",
        (
            Platform.DEVICE_TRACKER,
            Platform.SENSOR,
        ),
    ):
        yield


@pytest.fixture
async def elec_measurement_zigpy_dev(hass: HomeAssistant, zigpy_device_mock):
    """Electric Measurement zigpy device."""

    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [
                    general.Basic.cluster_id,
                    homeautomation.ElectricalMeasurement.cluster_id,
                ],
                SIG_EP_OUTPUT: [],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.SIMPLE_SENSOR,
                SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
            }
        },
    )
    zigpy_device.node_desc.mac_capability_flags |= 0b_0000_0100
    zigpy_device.endpoints[1].electrical_measurement.PLUGGED_ATTR_READS = {
        "ac_current_divisor": 10,
        "ac_current_multiplier": 1,
        "ac_power_divisor": 10,
        "ac_power_multiplier": 1,
        "ac_voltage_divisor": 10,
        "ac_voltage_multiplier": 1,
        "measurement_type": 8,
        "power_divisor": 10,
        "power_multiplier": 1,
    }
    return zigpy_device


@pytest.fixture
async def elec_measurement_zha_dev(elec_measurement_zigpy_dev, zha_device_joined):
    """Electric Measurement ZHA device."""

    zha_dev = await zha_device_joined(elec_measurement_zigpy_dev)
    zha_dev.available = True
    return zha_dev


async def async_test_humidity(hass: HomeAssistant, cluster, entity_id):
    """Test humidity sensor."""
    await send_attributes_report(hass, cluster, {1: 1, 0: 1000, 2: 100})
    assert_state(hass, entity_id, "10.0", PERCENTAGE)


async def async_test_temperature(hass: HomeAssistant, cluster, entity_id):
    """Test temperature sensor."""
    await send_attributes_report(hass, cluster, {1: 1, 0: 2900, 2: 100})
    assert_state(hass, entity_id, "29.0", UnitOfTemperature.CELSIUS)


async def async_test_pressure(hass: HomeAssistant, cluster, entity_id):
    """Test pressure sensor."""
    await send_attributes_report(hass, cluster, {1: 1, 0: 1000, 2: 10000})
    assert_state(hass, entity_id, "1000", UnitOfPressure.HPA)

    await send_attributes_report(hass, cluster, {0: 1000, 20: -1, 16: 10000})
    assert_state(hass, entity_id, "1000", UnitOfPressure.HPA)


async def async_test_illuminance(hass: HomeAssistant, cluster, entity_id):
    """Test illuminance sensor."""
    await send_attributes_report(hass, cluster, {1: 1, 0: 10, 2: 20})
    assert_state(hass, entity_id, "1", LIGHT_LUX)

    await send_attributes_report(hass, cluster, {1: 0, 0: 0, 2: 20})
    assert_state(hass, entity_id, "0", LIGHT_LUX)

    await send_attributes_report(hass, cluster, {1: 0, 0: 0xFFFF, 2: 20})
    assert_state(hass, entity_id, "unknown", LIGHT_LUX)


async def async_test_metering(hass: HomeAssistant, cluster, entity_id):
    """Test Smart Energy metering sensor."""
    await send_attributes_report(hass, cluster, {1025: 1, 1024: 12345, 1026: 100})
    assert_state(hass, entity_id, "12345.0", None)
    assert hass.states.get(entity_id).attributes["status"] == "NO_ALARMS"
    assert hass.states.get(entity_id).attributes["device_type"] == "Electric Metering"

    await send_attributes_report(hass, cluster, {1024: 12346, "status": 64 + 8})
    assert_state(hass, entity_id, "12346.0", None)
    assert hass.states.get(entity_id).attributes["status"] in (
        "SERVICE_DISCONNECT|POWER_FAILURE",
        "POWER_FAILURE|SERVICE_DISCONNECT",
    )

    await send_attributes_report(
        hass, cluster, {"status": 64 + 8, "metering_device_type": 1}
    )
    assert hass.states.get(entity_id).attributes["status"] in (
        "SERVICE_DISCONNECT|NOT_DEFINED",
        "NOT_DEFINED|SERVICE_DISCONNECT",
    )

    await send_attributes_report(
        hass, cluster, {"status": 64 + 8, "metering_device_type": 2}
    )
    assert hass.states.get(entity_id).attributes["status"] in (
        "SERVICE_DISCONNECT|PIPE_EMPTY",
        "PIPE_EMPTY|SERVICE_DISCONNECT",
    )

    await send_attributes_report(
        hass, cluster, {"status": 64 + 8, "metering_device_type": 5}
    )
    assert hass.states.get(entity_id).attributes["status"] in (
        "SERVICE_DISCONNECT|TEMPERATURE_SENSOR",
        "TEMPERATURE_SENSOR|SERVICE_DISCONNECT",
    )

    # Status for other meter types
    await send_attributes_report(
        hass, cluster, {"status": 32, "metering_device_type": 4}
    )
    assert hass.states.get(entity_id).attributes["status"] in ("<bitmap8.32: 32>", "32")


async def async_test_smart_energy_summation_delivered(
    hass: HomeAssistant, cluster, entity_id
):
    """Test SmartEnergy Summation delivered sensor."""

    await send_attributes_report(
        hass, cluster, {1025: 1, "current_summ_delivered": 12321, 1026: 100}
    )
    assert_state(hass, entity_id, "12.321", UnitOfEnergy.KILO_WATT_HOUR)
    assert hass.states.get(entity_id).attributes["status"] == "NO_ALARMS"
    assert hass.states.get(entity_id).attributes["device_type"] == "Electric Metering"
    assert (
        hass.states.get(entity_id).attributes[ATTR_DEVICE_CLASS]
        == SensorDeviceClass.ENERGY
    )


async def async_test_smart_energy_summation_received(
    hass: HomeAssistant, cluster, entity_id
):
    """Test SmartEnergy Summation received sensor."""

    await send_attributes_report(
        hass, cluster, {1025: 1, "current_summ_received": 12321, 1026: 100}
    )
    assert_state(hass, entity_id, "12.321", UnitOfEnergy.KILO_WATT_HOUR)
    assert hass.states.get(entity_id).attributes["status"] == "NO_ALARMS"
    assert hass.states.get(entity_id).attributes["device_type"] == "Electric Metering"
    assert (
        hass.states.get(entity_id).attributes[ATTR_DEVICE_CLASS]
        == SensorDeviceClass.ENERGY
    )


async def async_test_electrical_measurement(hass: HomeAssistant, cluster, entity_id):
    """Test electrical measurement sensor."""
    # update divisor cached value
    await send_attributes_report(hass, cluster, {"ac_power_divisor": 1})
    await send_attributes_report(hass, cluster, {0: 1, 1291: 100, 10: 1000})
    assert_state(hass, entity_id, "100", UnitOfPower.WATT)

    await send_attributes_report(hass, cluster, {0: 1, 1291: 99, 10: 1000})
    assert_state(hass, entity_id, "99", UnitOfPower.WATT)

    await send_attributes_report(hass, cluster, {"ac_power_divisor": 10})
    await send_attributes_report(hass, cluster, {0: 1, 1291: 1000, 10: 5000})
    assert_state(hass, entity_id, "100", UnitOfPower.WATT)

    await send_attributes_report(hass, cluster, {0: 1, 1291: 99, 10: 5000})
    assert_state(hass, entity_id, "9.9", UnitOfPower.WATT)

    assert "active_power_max" not in hass.states.get(entity_id).attributes
    await send_attributes_report(hass, cluster, {0: 1, 0x050D: 88, 10: 5000})
    assert hass.states.get(entity_id).attributes["active_power_max"] == "8.8"


async def async_test_em_apparent_power(hass: HomeAssistant, cluster, entity_id):
    """Test electrical measurement Apparent Power sensor."""
    # update divisor cached value
    await send_attributes_report(hass, cluster, {"ac_power_divisor": 1})
    await send_attributes_report(hass, cluster, {0: 1, 0x050F: 100, 10: 1000})
    assert_state(hass, entity_id, "100", UnitOfApparentPower.VOLT_AMPERE)

    await send_attributes_report(hass, cluster, {0: 1, 0x050F: 99, 10: 1000})
    assert_state(hass, entity_id, "99", UnitOfApparentPower.VOLT_AMPERE)

    await send_attributes_report(hass, cluster, {"ac_power_divisor": 10})
    await send_attributes_report(hass, cluster, {0: 1, 0x050F: 1000, 10: 5000})
    assert_state(hass, entity_id, "100", UnitOfApparentPower.VOLT_AMPERE)

    await send_attributes_report(hass, cluster, {0: 1, 0x050F: 99, 10: 5000})
    assert_state(hass, entity_id, "9.9", UnitOfApparentPower.VOLT_AMPERE)


async def async_test_em_power_factor(hass: HomeAssistant, cluster, entity_id):
    """Test electrical measurement Power Factor sensor."""
    # update divisor cached value
    await send_attributes_report(hass, cluster, {"ac_power_divisor": 1})
    await send_attributes_report(hass, cluster, {0: 1, 0x0510: 100, 10: 1000})
    assert_state(hass, entity_id, "100", PERCENTAGE)

    await send_attributes_report(hass, cluster, {0: 1, 0x0510: 99, 10: 1000})
    assert_state(hass, entity_id, "99", PERCENTAGE)

    await send_attributes_report(hass, cluster, {"ac_power_divisor": 10})
    await send_attributes_report(hass, cluster, {0: 1, 0x0510: 100, 10: 5000})
    assert_state(hass, entity_id, "100", PERCENTAGE)

    await send_attributes_report(hass, cluster, {0: 1, 0x0510: 99, 10: 5000})
    assert_state(hass, entity_id, "99", PERCENTAGE)


async def async_test_em_rms_current(hass: HomeAssistant, cluster, entity_id):
    """Test electrical measurement RMS Current sensor."""

    await send_attributes_report(hass, cluster, {0: 1, 0x0508: 1234, 10: 1000})
    assert_state(hass, entity_id, "1.2", UnitOfElectricCurrent.AMPERE)

    await send_attributes_report(hass, cluster, {"ac_current_divisor": 10})
    await send_attributes_report(hass, cluster, {0: 1, 0x0508: 236, 10: 1000})
    assert_state(hass, entity_id, "23.6", UnitOfElectricCurrent.AMPERE)

    await send_attributes_report(hass, cluster, {0: 1, 0x0508: 1236, 10: 1000})
    assert_state(hass, entity_id, "124", UnitOfElectricCurrent.AMPERE)

    assert "rms_current_max" not in hass.states.get(entity_id).attributes
    await send_attributes_report(hass, cluster, {0: 1, 0x050A: 88, 10: 5000})
    assert hass.states.get(entity_id).attributes["rms_current_max"] == "8.8"


async def async_test_em_rms_voltage(hass: HomeAssistant, cluster, entity_id):
    """Test electrical measurement RMS Voltage sensor."""

    await send_attributes_report(hass, cluster, {0: 1, 0x0505: 1234, 10: 1000})
    assert_state(hass, entity_id, "123", UnitOfElectricPotential.VOLT)

    await send_attributes_report(hass, cluster, {0: 1, 0x0505: 234, 10: 1000})
    assert_state(hass, entity_id, "23.4", UnitOfElectricPotential.VOLT)

    await send_attributes_report(hass, cluster, {"ac_voltage_divisor": 100})
    await send_attributes_report(hass, cluster, {0: 1, 0x0505: 2236, 10: 1000})
    assert_state(hass, entity_id, "22.4", UnitOfElectricPotential.VOLT)

    assert "rms_voltage_max" not in hass.states.get(entity_id).attributes
    await send_attributes_report(hass, cluster, {0: 1, 0x0507: 888, 10: 5000})
    assert hass.states.get(entity_id).attributes["rms_voltage_max"] == "8.9"


async def async_test_powerconfiguration(hass: HomeAssistant, cluster, entity_id):
    """Test powerconfiguration/battery sensor."""
    await send_attributes_report(hass, cluster, {33: 98})
    assert_state(hass, entity_id, "49", "%")
    assert hass.states.get(entity_id).attributes["battery_voltage"] == 2.9
    assert hass.states.get(entity_id).attributes["battery_quantity"] == 3
    assert hass.states.get(entity_id).attributes["battery_size"] == "AAA"
    await send_attributes_report(hass, cluster, {32: 20})
    assert hass.states.get(entity_id).attributes["battery_voltage"] == 2.0


async def async_test_powerconfiguration2(hass: HomeAssistant, cluster, entity_id):
    """Test powerconfiguration/battery sensor."""
    await send_attributes_report(hass, cluster, {33: -1})
    assert_state(hass, entity_id, STATE_UNKNOWN, "%")

    await send_attributes_report(hass, cluster, {33: 255})
    assert_state(hass, entity_id, STATE_UNKNOWN, "%")

    await send_attributes_report(hass, cluster, {33: 98})
    assert_state(hass, entity_id, "49", "%")


async def async_test_device_temperature(hass: HomeAssistant, cluster, entity_id):
    """Test temperature sensor."""
    await send_attributes_report(hass, cluster, {0: 2900})
    assert_state(hass, entity_id, "29.0", UnitOfTemperature.CELSIUS)


async def async_test_setpoint_change_source(hass, cluster, entity_id):
    """Test the translation of numerical state into enum text."""
    await send_attributes_report(
        hass, cluster, {Thermostat.AttributeDefs.setpoint_change_source.id: 0x01}
    )
    hass_state = hass.states.get(entity_id)
    assert hass_state.state == "Schedule"


async def async_test_pi_heating_demand(hass, cluster, entity_id):
    """Test pi heating demand is correctly returned."""
    await send_attributes_report(
        hass, cluster, {Thermostat.AttributeDefs.pi_heating_demand.id: 1}
    )
    assert_state(hass, entity_id, "1", "%")


@pytest.mark.parametrize(
    (
        "cluster_id",
        "entity_suffix",
        "test_func",
        "report_count",
        "read_plug",
        "unsupported_attrs",
        "initial_sensor_state",
    ),
    [
        (
            measurement.RelativeHumidity.cluster_id,
            "humidity",
            async_test_humidity,
            1,
            None,
            None,
            STATE_UNKNOWN,
        ),
        (
            measurement.TemperatureMeasurement.cluster_id,
            "temperature",
            async_test_temperature,
            1,
            None,
            None,
            STATE_UNKNOWN,
        ),
        (
            measurement.PressureMeasurement.cluster_id,
            "pressure",
            async_test_pressure,
            1,
            None,
            None,
            STATE_UNKNOWN,
        ),
        (
            measurement.IlluminanceMeasurement.cluster_id,
            "illuminance",
            async_test_illuminance,
            1,
            None,
            None,
            STATE_UNKNOWN,
        ),
        (
            smartenergy.Metering.cluster_id,
            "instantaneous_demand",
            async_test_metering,
            10,
            {
                "demand_formatting": 0xF9,
                "divisor": 1,
                "metering_device_type": 0x00,
                "multiplier": 1,
                "status": 0x00,
            },
            {"current_summ_delivered", "current_summ_received"},
            STATE_UNKNOWN,
        ),
        (
            smartenergy.Metering.cluster_id,
            "summation_delivered",
            async_test_smart_energy_summation_delivered,
            10,
            {
                "demand_formatting": 0xF9,
                "divisor": 1000,
                "metering_device_type": 0x00,
                "multiplier": 1,
                "status": 0x00,
                "summation_formatting": 0b1_0111_010,
                "unit_of_measure": 0x00,
            },
            {"instaneneous_demand", "current_summ_received"},
            STATE_UNKNOWN,
        ),
        (
            smartenergy.Metering.cluster_id,
            "summation_received",
            async_test_smart_energy_summation_received,
            10,
            {
                "demand_formatting": 0xF9,
                "divisor": 1000,
                "metering_device_type": 0x00,
                "multiplier": 1,
                "status": 0x00,
                "summation_formatting": 0b1_0111_010,
                "unit_of_measure": 0x00,
                "current_summ_received": 0,
            },
            {"instaneneous_demand", "current_summ_delivered"},
            "0.0",
        ),
        (
            homeautomation.ElectricalMeasurement.cluster_id,
            "power",
            async_test_electrical_measurement,
            7,
            {"ac_power_divisor": 1000, "ac_power_multiplier": 1},
            {"apparent_power", "rms_current", "rms_voltage"},
            STATE_UNKNOWN,
        ),
        (
            homeautomation.ElectricalMeasurement.cluster_id,
            "apparent_power",
            async_test_em_apparent_power,
            7,
            {"ac_power_divisor": 1000, "ac_power_multiplier": 1},
            {"active_power", "rms_current", "rms_voltage"},
            STATE_UNKNOWN,
        ),
        (
            homeautomation.ElectricalMeasurement.cluster_id,
            "power_factor",
            async_test_em_power_factor,
            7,
            {"ac_power_divisor": 1000, "ac_power_multiplier": 1},
            {"active_power", "apparent_power", "rms_current", "rms_voltage"},
            STATE_UNKNOWN,
        ),
        (
            homeautomation.ElectricalMeasurement.cluster_id,
            "current",
            async_test_em_rms_current,
            7,
            {"ac_current_divisor": 1000, "ac_current_multiplier": 1},
            {"active_power", "apparent_power", "rms_voltage"},
            STATE_UNKNOWN,
        ),
        (
            homeautomation.ElectricalMeasurement.cluster_id,
            "voltage",
            async_test_em_rms_voltage,
            7,
            {"ac_voltage_divisor": 10, "ac_voltage_multiplier": 1},
            {"active_power", "apparent_power", "rms_current"},
            STATE_UNKNOWN,
        ),
        (
            general.PowerConfiguration.cluster_id,
            "battery",
            async_test_powerconfiguration,
            2,
            {
                "battery_size": 4,  # AAA
                "battery_voltage": 29,
                "battery_quantity": 3,
            },
            None,
            STATE_UNKNOWN,
        ),
        (
            general.PowerConfiguration.cluster_id,
            "battery",
            async_test_powerconfiguration2,
            2,
            {
                "battery_size": 4,  # AAA
                "battery_voltage": 29,
                "battery_quantity": 3,
            },
            None,
            STATE_UNKNOWN,
        ),
        (
            general.DeviceTemperature.cluster_id,
            "device_temperature",
            async_test_device_temperature,
            1,
            None,
            None,
            STATE_UNKNOWN,
        ),
        (
            hvac.Thermostat.cluster_id,
            "setpoint_change_source",
            async_test_setpoint_change_source,
            10,
            None,
            None,
            STATE_UNKNOWN,
        ),
        (
            hvac.Thermostat.cluster_id,
            "pi_heating_demand",
            async_test_pi_heating_demand,
            10,
            None,
            None,
            STATE_UNKNOWN,
        ),
    ],
)
async def test_sensor(
    hass: HomeAssistant,
    zigpy_device_mock,
    zha_device_joined_restored,
    cluster_id,
    entity_suffix,
    test_func,
    report_count,
    read_plug,
    unsupported_attrs,
    initial_sensor_state,
) -> None:
    """Test ZHA sensor platform."""

    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [cluster_id, general.Basic.cluster_id],
                SIG_EP_OUTPUT: [],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
            }
        }
    )
    cluster = zigpy_device.endpoints[1].in_clusters[cluster_id]
    if unsupported_attrs:
        for attr in unsupported_attrs:
            cluster.add_unsupported_attribute(attr)
    if cluster_id in (
        smartenergy.Metering.cluster_id,
        homeautomation.ElectricalMeasurement.cluster_id,
    ):
        # this one is mains powered
        zigpy_device.node_desc.mac_capability_flags |= 0b_0000_0100
    cluster.PLUGGED_ATTR_READS = read_plug
    zha_device = await zha_device_joined_restored(zigpy_device)
    entity_id = ENTITY_ID_PREFIX.format(entity_suffix)

    await async_enable_traffic(hass, [zha_device], enabled=False)
    await hass.async_block_till_done()
    # ensure the sensor entity was created
    assert hass.states.get(entity_id).state == STATE_UNAVAILABLE

    # allow traffic to flow through the gateway and devices
    await async_enable_traffic(hass, [zha_device])

    # test that the sensor now have their correct initial state (mostly unknown)
    assert hass.states.get(entity_id).state == initial_sensor_state

    # test sensor associated logic
    await test_func(hass, cluster, entity_id)

    # test rejoin
    await async_test_rejoin(hass, zigpy_device, [cluster], (report_count,))


def assert_state(hass: HomeAssistant, entity_id, state, unit_of_measurement):
    """Check that the state is what is expected.

    This is used to ensure that the logic in each sensor class handled the
    attribute report it received correctly.
    """
    hass_state = hass.states.get(entity_id)
    assert hass_state.state == state
    assert hass_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == unit_of_measurement


@pytest.fixture
def hass_ms(hass: HomeAssistant) -> Callable[[str], HomeAssistant]:
    """Hass instance with measurement system."""

    async def _hass_ms(meas_sys: str) -> HomeAssistant:
        await config_util.async_process_ha_core_config(
            hass, {CONF_UNIT_SYSTEM: meas_sys}
        )
        await hass.async_block_till_done()
        return hass

    return _hass_ms


@pytest.fixture
def core_rs(hass_storage: dict[str, Any]):
    """Core.restore_state fixture."""

    def _storage(entity_id, uom, state):
        now = dt_util.utcnow().isoformat()

        hass_storage[restore_state.STORAGE_KEY] = {
            "version": restore_state.STORAGE_VERSION,
            "key": restore_state.STORAGE_KEY,
            "data": [
                {
                    "state": {
                        "entity_id": entity_id,
                        "state": str(state),
                        "attributes": {ATTR_UNIT_OF_MEASUREMENT: uom},
                        "last_changed": now,
                        "last_updated": now,
                        "context": {
                            "id": "3c2243ff5f30447eb12e7348cfd5b8ff",
                            "user_id": None,
                        },
                    },
                    "last_seen": now,
                }
            ],
        }

    return _storage


@pytest.mark.parametrize(
    ("uom", "raw_temp", "expected", "restore"),
    [
        (UnitOfTemperature.CELSIUS, 2900, 29, False),
        (UnitOfTemperature.CELSIUS, 2900, 29, True),
        (UnitOfTemperature.FAHRENHEIT, 2900, 84, False),
        (UnitOfTemperature.FAHRENHEIT, 2900, 84, True),
    ],
)
async def test_temp_uom(
    hass: HomeAssistant,
    uom: UnitOfTemperature,
    raw_temp: int,
    expected: int,
    restore: bool,
    hass_ms: Callable[[str], HomeAssistant],
    core_rs,
    zigpy_device_mock,
    zha_device_restored,
) -> None:
    """Test ZHA temperature sensor unit of measurement."""

    entity_id = "sensor.fake1026_fakemodel1026_004f3202_temperature"
    if restore:
        core_rs(entity_id, uom, state=(expected - 2))
        await async_mock_load_restore_state_from_storage(hass)

    hass = await hass_ms("metric" if uom == UnitOfTemperature.CELSIUS else "imperial")

    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [
                    measurement.TemperatureMeasurement.cluster_id,
                    general.Basic.cluster_id,
                ],
                SIG_EP_OUTPUT: [],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
            }
        }
    )
    cluster = zigpy_device.endpoints[1].temperature
    zha_device = await zha_device_restored(zigpy_device)
    entity_id = find_entity_id(Platform.SENSOR, zha_device, hass)

    if not restore:
        await async_enable_traffic(hass, [zha_device], enabled=False)
        assert hass.states.get(entity_id).state == STATE_UNAVAILABLE

    # allow traffic to flow through the gateway and devices
    await async_enable_traffic(hass, [zha_device])

    # test that the sensors now have a state of unknown
    if not restore:
        assert hass.states.get(entity_id).state == STATE_UNKNOWN

    await send_attribute_report(hass, cluster, 0, raw_temp)
    await hass.async_block_till_done()
    state = hass.states.get(entity_id)
    assert state is not None
    assert round(float(state.state)) == expected
    assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == uom


@patch(
    "zigpy.zcl.ClusterPersistingListener",
    MagicMock(),
)
async def test_electrical_measurement_init(
    hass: HomeAssistant,
    zigpy_device_mock,
    zha_device_joined,
) -> None:
    """Test proper initialization of the electrical measurement cluster."""

    cluster_id = homeautomation.ElectricalMeasurement.cluster_id
    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [cluster_id, general.Basic.cluster_id],
                SIG_EP_OUTPUT: [],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
            }
        }
    )
    cluster = zigpy_device.endpoints[1].in_clusters[cluster_id]
    zha_device = await zha_device_joined(zigpy_device)
    entity_id = "sensor.fakemanufacturer_fakemodel_power"

    # allow traffic to flow through the gateway and devices
    await async_enable_traffic(hass, [zha_device])

    # test that the sensor now have a state of unknown
    assert hass.states.get(entity_id).state == STATE_UNKNOWN

    await send_attributes_report(hass, cluster, {0: 1, 1291: 100, 10: 1000})
    assert int(hass.states.get(entity_id).state) == 100

    cluster_handler = zha_device._endpoints[1].all_cluster_handlers["1:0x0b04"]
    assert cluster_handler.ac_power_divisor == 1
    assert cluster_handler.ac_power_multiplier == 1

    # update power divisor
    await send_attributes_report(hass, cluster, {0: 1, 1291: 20, 0x0403: 5, 10: 1000})
    assert cluster_handler.ac_power_divisor == 5
    assert cluster_handler.ac_power_multiplier == 1
    assert hass.states.get(entity_id).state == "4.0"

    await send_attributes_report(hass, cluster, {0: 1, 1291: 30, 0x0605: 10, 10: 1000})
    assert cluster_handler.ac_power_divisor == 10
    assert cluster_handler.ac_power_multiplier == 1
    assert hass.states.get(entity_id).state == "3.0"

    # update power multiplier
    await send_attributes_report(hass, cluster, {0: 1, 1291: 20, 0x0402: 6, 10: 1000})
    assert cluster_handler.ac_power_divisor == 10
    assert cluster_handler.ac_power_multiplier == 6
    assert hass.states.get(entity_id).state == "12.0"

    await send_attributes_report(hass, cluster, {0: 1, 1291: 30, 0x0604: 20, 10: 1000})
    assert cluster_handler.ac_power_divisor == 10
    assert cluster_handler.ac_power_multiplier == 20
    assert hass.states.get(entity_id).state == "60.0"


@pytest.mark.parametrize(
    ("cluster_id", "unsupported_attributes", "entity_ids", "missing_entity_ids"),
    [
        (
            homeautomation.ElectricalMeasurement.cluster_id,
            {"apparent_power", "rms_voltage", "rms_current"},
            {
                "power",
                "ac_frequency",
                "power_factor",
            },
            {
                "apparent_power",
                "voltage",
                "current",
            },
        ),
        (
            homeautomation.ElectricalMeasurement.cluster_id,
            {"apparent_power", "rms_current", "ac_frequency", "power_factor"},
            {"voltage", "power"},
            {
                "apparent_power",
                "current",
                "ac_frequency",
                "power_factor",
            },
        ),
        (
            homeautomation.ElectricalMeasurement.cluster_id,
            set(),
            {
                "voltage",
                "power",
                "apparent_power",
                "current",
                "ac_frequency",
                "power_factor",
            },
            set(),
        ),
        (
            smartenergy.Metering.cluster_id,
            {
                "instantaneous_demand",
            },
            {
                "summation_delivered",
            },
            {
                "instantaneous_demand",
            },
        ),
        (
            smartenergy.Metering.cluster_id,
            {"instantaneous_demand", "current_summ_delivered"},
            {},
            {
                "instantaneous_demand",
                "summation_delivered",
            },
        ),
        (
            smartenergy.Metering.cluster_id,
            {},
            {
                "instantaneous_demand",
                "summation_delivered",
            },
            {},
        ),
    ],
)
async def test_unsupported_attributes_sensor(
    hass: HomeAssistant,
    zigpy_device_mock,
    zha_device_joined_restored,
    cluster_id,
    unsupported_attributes,
    entity_ids,
    missing_entity_ids,
) -> None:
    """Test ZHA sensor platform."""

    entity_ids = {ENTITY_ID_PREFIX.format(e) for e in entity_ids}
    missing_entity_ids = {ENTITY_ID_PREFIX.format(e) for e in missing_entity_ids}

    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [cluster_id, general.Basic.cluster_id],
                SIG_EP_OUTPUT: [],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
            }
        }
    )
    cluster = zigpy_device.endpoints[1].in_clusters[cluster_id]
    if cluster_id == smartenergy.Metering.cluster_id:
        # this one is mains powered
        zigpy_device.node_desc.mac_capability_flags |= 0b_0000_0100
    for attr in unsupported_attributes:
        cluster.add_unsupported_attribute(attr)
    zha_device = await zha_device_joined_restored(zigpy_device)

    await async_enable_traffic(hass, [zha_device], enabled=False)
    await hass.async_block_till_done()
    present_entity_ids = set(find_entity_ids(Platform.SENSOR, zha_device, hass))
    assert present_entity_ids == entity_ids
    assert missing_entity_ids not in present_entity_ids


@pytest.mark.parametrize(
    ("raw_uom", "raw_value", "expected_state", "expected_uom"),
    [
        (
            1,
            12320,
            "1.23",
            UnitOfVolume.CUBIC_METERS,
        ),
        (
            1,
            1232000,
            "123.2",
            UnitOfVolume.CUBIC_METERS,
        ),
        (
            3,
            2340,
            "0.65",
            UnitOfVolume.CUBIC_METERS,
        ),
        (
            3,
            2360,
            "0.68",
            UnitOfVolume.CUBIC_METERS,
        ),
        (
            8,
            23660,
            "2.37",
            UnitOfPressure.KPA,
        ),
        (
            0,
            9366,
            "0.937",
            UnitOfEnergy.KILO_WATT_HOUR,
        ),
        (
            0,
            999,
            "0.1",
            UnitOfEnergy.KILO_WATT_HOUR,
        ),
        (
            0,
            10091,
            "1.009",
            UnitOfEnergy.KILO_WATT_HOUR,
        ),
        (
            0,
            10099,
            "1.01",
            UnitOfEnergy.KILO_WATT_HOUR,
        ),
        (
            0,
            100999,
            "10.1",
            UnitOfEnergy.KILO_WATT_HOUR,
        ),
        (
            0,
            100023,
            "10.002",
            UnitOfEnergy.KILO_WATT_HOUR,
        ),
        (
            0,
            102456,
            "10.246",
            UnitOfEnergy.KILO_WATT_HOUR,
        ),
        (
            5,
            102456,
            "10.25",
            "IMP gal",
        ),
        (
            7,
            50124,
            "5.01",
            UnitOfVolume.LITERS,
        ),
    ],
)
async def test_se_summation_uom(
    hass: HomeAssistant,
    zigpy_device_mock,
    zha_device_joined,
    raw_uom,
    raw_value,
    expected_state,
    expected_uom,
) -> None:
    """Test ZHA smart energy summation."""

    entity_id = ENTITY_ID_PREFIX.format("summation_delivered")
    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [
                    smartenergy.Metering.cluster_id,
                    general.Basic.cluster_id,
                ],
                SIG_EP_OUTPUT: [],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.SIMPLE_SENSOR,
            }
        }
    )
    zigpy_device.node_desc.mac_capability_flags |= 0b_0000_0100

    cluster = zigpy_device.endpoints[1].in_clusters[smartenergy.Metering.cluster_id]
    for attr in ("instanteneous_demand",):
        cluster.add_unsupported_attribute(attr)
    cluster.PLUGGED_ATTR_READS = {
        "current_summ_delivered": raw_value,
        "demand_formatting": 0xF9,
        "divisor": 10000,
        "metering_device_type": 0x00,
        "multiplier": 1,
        "status": 0x00,
        "summation_formatting": 0b1_0111_010,
        "unit_of_measure": raw_uom,
    }
    await zha_device_joined(zigpy_device)

    assert_state(hass, entity_id, expected_state, expected_uom)


@pytest.mark.parametrize(
    ("raw_measurement_type", "expected_type"),
    [
        (1, "ACTIVE_MEASUREMENT"),
        (8, "PHASE_A_MEASUREMENT"),
        (9, "ACTIVE_MEASUREMENT, PHASE_A_MEASUREMENT"),
        (
            15,
            (
                "ACTIVE_MEASUREMENT, REACTIVE_MEASUREMENT, APPARENT_MEASUREMENT,"
                " PHASE_A_MEASUREMENT"
            ),
        ),
    ],
)
async def test_elec_measurement_sensor_type(
    hass: HomeAssistant,
    elec_measurement_zigpy_dev,
    raw_measurement_type,
    expected_type,
    zha_device_joined,
) -> None:
    """Test ZHA electrical measurement sensor type."""

    entity_id = ENTITY_ID_PREFIX.format("power")
    zigpy_dev = elec_measurement_zigpy_dev
    zigpy_dev.endpoints[1].electrical_measurement.PLUGGED_ATTR_READS[
        "measurement_type"
    ] = raw_measurement_type

    await zha_device_joined(zigpy_dev)

    state = hass.states.get(entity_id)
    assert state is not None
    assert state.attributes["measurement_type"] == expected_type


async def test_elec_measurement_sensor_polling(
    hass: HomeAssistant,
    elec_measurement_zigpy_dev,
    zha_device_joined_restored,
) -> None:
    """Test ZHA electrical measurement sensor polling."""

    entity_id = ENTITY_ID_PREFIX.format("power")
    zigpy_dev = elec_measurement_zigpy_dev
    zigpy_dev.endpoints[1].electrical_measurement.PLUGGED_ATTR_READS["active_power"] = (
        20
    )

    await zha_device_joined_restored(zigpy_dev)

    # test that the sensor has an initial state of 2.0
    state = hass.states.get(entity_id)
    assert state.state == "2.0"

    # update the value for the power reading
    zigpy_dev.endpoints[1].electrical_measurement.PLUGGED_ATTR_READS["active_power"] = (
        60
    )

    # ensure the state is still 2.0
    state = hass.states.get(entity_id)
    assert state.state == "2.0"

    # let the polling happen
    future = dt_util.utcnow() + timedelta(seconds=90)
    async_fire_time_changed(hass, future)
    await hass.async_block_till_done(wait_background_tasks=True)

    # ensure the state has been updated to 6.0
    state = hass.states.get(entity_id)
    assert state.state == "6.0"


@pytest.mark.parametrize(
    "supported_attributes",
    [
        set(),
        {
            "active_power",
            "active_power_max",
            "rms_current",
            "rms_current_max",
            "rms_voltage",
            "rms_voltage_max",
        },
        {
            "active_power",
        },
        {
            "active_power",
            "active_power_max",
        },
        {
            "rms_current",
            "rms_current_max",
        },
        {
            "rms_voltage",
            "rms_voltage_max",
        },
    ],
)
async def test_elec_measurement_skip_unsupported_attribute(
    hass: HomeAssistant,
    elec_measurement_zha_dev,
    supported_attributes,
) -> None:
    """Test ZHA electrical measurement skipping update of unsupported attributes."""

    entity_id = ENTITY_ID_PREFIX.format("power")
    zha_dev = elec_measurement_zha_dev

    cluster = zha_dev.device.endpoints[1].electrical_measurement

    all_attrs = {
        "active_power",
        "active_power_max",
        "apparent_power",
        "rms_current",
        "rms_current_max",
        "rms_voltage",
        "rms_voltage_max",
        "power_factor",
        "ac_frequency",
        "ac_frequency_max",
    }
    for attr in all_attrs - supported_attributes:
        cluster.add_unsupported_attribute(attr)
    cluster.read_attributes.reset_mock()

    await async_update_entity(hass, entity_id)
    await hass.async_block_till_done()
    assert cluster.read_attributes.call_count == math.ceil(
        len(supported_attributes) / ZHA_CLUSTER_HANDLER_READS_PER_REQ
    )
    read_attrs = {
        a for call in cluster.read_attributes.call_args_list for a in call[0][0]
    }
    assert read_attrs == supported_attributes


class OppleCluster(CustomCluster, ManufacturerSpecificCluster):
    """Aqara manufacturer specific cluster."""

    cluster_id = 0xFCC0
    ep_attribute = "opple_cluster"
    attributes = {
        0x010C: ("last_feeding_size", t.uint16_t, True),
    }

    def __init__(self, *args, **kwargs) -> None:
        """Initialize."""
        super().__init__(*args, **kwargs)
        # populate cache to create config entity
        self._attr_cache.update({0x010C: 10})


(
    add_to_registry_v2("Fake_Manufacturer_sensor", "Fake_Model_sensor")
    .replaces(OppleCluster)
    .sensor(
        "last_feeding_size",
        OppleCluster.cluster_id,
        divisor=1,
        multiplier=1,
        unit=UnitOfMass.GRAMS,
    )
)


@pytest.fixture
async def zigpy_device_aqara_sensor_v2(
    hass: HomeAssistant, zigpy_device_mock, zha_device_joined_restored
):
    """Device tracker zigpy Aqara motion sensor device."""

    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [
                    general.Basic.cluster_id,
                    OppleCluster.cluster_id,
                ],
                SIG_EP_OUTPUT: [],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.OCCUPANCY_SENSOR,
            }
        },
        manufacturer="Fake_Manufacturer_sensor",
        model="Fake_Model_sensor",
    )

    zha_device = await zha_device_joined_restored(zigpy_device)
    return zha_device, zigpy_device.endpoints[1].opple_cluster


async def test_last_feeding_size_sensor_v2(
    hass: HomeAssistant, zigpy_device_aqara_sensor_v2
) -> None:
    """Test quirks defined sensor."""

    zha_device, cluster = zigpy_device_aqara_sensor_v2
    assert isinstance(zha_device.device, CustomDeviceV2)
    entity_id = find_entity_id(
        Platform.SENSOR, zha_device, hass, qualifier="last_feeding_size"
    )
    assert entity_id is not None

    await send_attributes_report(hass, cluster, {0x010C: 1})
    assert_state(hass, entity_id, "1.0", UnitOfMass.GRAMS.value)

    await send_attributes_report(hass, cluster, {0x010C: 5})
    assert_state(hass, entity_id, "5.0", UnitOfMass.GRAMS.value)


@pytest.fixture
async def coordinator(hass: HomeAssistant, zigpy_device_mock, zha_device_joined):
    """Test ZHA fan platform."""

    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [general.Groups.cluster_id],
                SIG_EP_OUTPUT: [],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.CONTROL_BRIDGE,
                SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
            }
        },
        ieee="00:15:8d:00:02:32:4f:32",
        nwk=0x0000,
        node_descriptor=b"\xf8\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff",
    )
    zha_device = await zha_device_joined(zigpy_device)
    zha_device.available = True
    return zha_device


async def test_device_counter_sensors(
    hass: HomeAssistant,
    coordinator: ZHADevice,
    entity_registry: er.EntityRegistry,
    config_entry: MockConfigEntry,
) -> None:
    """Test quirks defined sensor."""

    entity_id = "sensor.coordinator_manufacturer_coordinator_model_counter_1"
    state = hass.states.get(entity_id)
    assert state is None

    # Enable the entity.
    entity_registry.async_update_entity(entity_id, disabled_by=None)
    await hass.config_entries.async_reload(config_entry.entry_id)
    await hass.async_block_till_done()

    state = hass.states.get(entity_id)
    assert state is not None
    assert state.state == "1"

    # simulate counter increment on application
    coordinator.device.application.state.counters["ezsp_counters"][
        "counter_1"
    ].increment()

    next_update = dt_util.utcnow() + timedelta(seconds=60)
    async_fire_time_changed(hass, next_update)
    await hass.async_block_till_done()

    state = hass.states.get(entity_id)
    assert state is not None
    assert state.state == "2"


@pytest.fixture
async def zigpy_device_danfoss_thermostat(
    hass: HomeAssistant, zigpy_device_mock, zha_device_joined_restored
):
    """Device tracker zigpy danfoss thermostat device."""

    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [
                    general.Basic.cluster_id,
                    general.PowerConfiguration.cluster_id,
                    general.Identify.cluster_id,
                    general.Time.cluster_id,
                    general.PollControl.cluster_id,
                    Thermostat.cluster_id,
                    hvac.UserInterface.cluster_id,
                    homeautomation.Diagnostic.cluster_id,
                ],
                SIG_EP_OUTPUT: [general.Basic.cluster_id, general.Ota.cluster_id],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.THERMOSTAT,
            }
        },
        manufacturer="Danfoss",
        model="eTRV0100",
    )

    zha_device = await zha_device_joined_restored(zigpy_device)
    return zha_device, zigpy_device


async def test_danfoss_thermostat_sw_error(
    hass: HomeAssistant, zigpy_device_danfoss_thermostat
) -> None:
    """Test quirks defined thermostat."""

    zha_device, zigpy_device = zigpy_device_danfoss_thermostat

    entity_id = find_entity_id(
        Platform.SENSOR, zha_device, hass, qualifier="software_error"
    )
    assert entity_id is not None

    cluster = zigpy_device.endpoints[1].diagnostic

    await send_attributes_report(
        hass,
        cluster,
        {
            danfoss_thermostat.DanfossDiagnosticCluster.AttributeDefs.sw_error_code.id: 0x0001
        },
    )

    hass_state = hass.states.get(entity_id)
    assert hass_state.state == "something"
    assert hass_state.attributes["Top_pcb_sensor_error"]