"""Test the Energy sensors.""" import copy from datetime import timedelta from unittest.mock import patch import pytest from homeassistant.components.energy import data from homeassistant.components.sensor import ( ATTR_LAST_RESET, ATTR_STATE_CLASS, STATE_CLASS_MEASUREMENT, STATE_CLASS_TOTAL, STATE_CLASS_TOTAL_INCREASING, ) from homeassistant.components.sensor.recorder import compile_statistics from homeassistant.const import ( ATTR_DEVICE_CLASS, ATTR_UNIT_OF_MEASUREMENT, DEVICE_CLASS_MONETARY, ENERGY_KILO_WATT_HOUR, ENERGY_MEGA_WATT_HOUR, ENERGY_WATT_HOUR, STATE_UNKNOWN, VOLUME_CUBIC_METERS, ) from homeassistant.setup import async_setup_component import homeassistant.util.dt as dt_util from tests.common import async_init_recorder_component from tests.components.recorder.common import async_wait_recording_done_without_instance async def setup_integration(hass): """Set up the integration.""" assert await async_setup_component( hass, "energy", {"recorder": {"db_url": "sqlite://"}} ) await hass.async_block_till_done() def get_statistics_for_entity(statistics_results, entity_id): """Get statistics for a certain entity, or None if there is none.""" for statistics_result in statistics_results: if statistics_result["meta"]["statistic_id"] == entity_id: return statistics_result return None async def test_cost_sensor_no_states(hass, hass_storage) -> None: """Test sensors are created.""" energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "grid", "flow_from": [ { "stat_energy_from": "foo", "entity_energy_from": "foo", "stat_cost": None, "entity_energy_price": "bar", "number_energy_price": None, } ], "cost_adjustment_day": 0, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } await setup_integration(hass) # TODO: No states, should the cost entity refuse to setup? @pytest.mark.parametrize("initial_energy,initial_cost", [(0, "0.0"), (None, "unknown")]) @pytest.mark.parametrize( "price_entity,fixed_price", [("sensor.energy_price", None), (None, 1)] ) @pytest.mark.parametrize( "usage_sensor_entity_id,cost_sensor_entity_id,flow_type", [ ("sensor.energy_consumption", "sensor.energy_consumption_cost", "flow_from"), ( "sensor.energy_production", "sensor.energy_production_compensation", "flow_to", ), ], ) async def test_cost_sensor_price_entity_total_increasing( hass, hass_storage, hass_ws_client, initial_energy, initial_cost, price_entity, fixed_price, usage_sensor_entity_id, cost_sensor_entity_id, flow_type, ) -> None: """Test energy cost price from total_increasing type sensor entity.""" def _compile_statistics(_): return compile_statistics(hass, now, now + timedelta(seconds=1)) energy_attributes = { ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR, ATTR_STATE_CLASS: STATE_CLASS_TOTAL_INCREASING, } await async_init_recorder_component(hass) energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "grid", "flow_from": [ { "stat_energy_from": "sensor.energy_consumption", "entity_energy_from": "sensor.energy_consumption", "stat_cost": None, "entity_energy_price": price_entity, "number_energy_price": fixed_price, } ] if flow_type == "flow_from" else [], "flow_to": [ { "stat_energy_to": "sensor.energy_production", "entity_energy_to": "sensor.energy_production", "stat_compensation": None, "entity_energy_price": price_entity, "number_energy_price": fixed_price, } ] if flow_type == "flow_to" else [], "cost_adjustment_day": 0, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } now = dt_util.utcnow() last_reset_cost_sensor = now.isoformat() # Optionally initialize dependent entities if initial_energy is not None: hass.states.async_set( usage_sensor_entity_id, initial_energy, energy_attributes, ) hass.states.async_set("sensor.energy_price", "1") with patch("homeassistant.util.dt.utcnow", return_value=now): await setup_integration(hass) state = hass.states.get(cost_sensor_entity_id) assert state.state == initial_cost assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_MONETARY if initial_cost != "unknown": assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor assert state.attributes[ATTR_STATE_CLASS] == STATE_CLASS_TOTAL assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "EUR" # Optional late setup of dependent entities if initial_energy is None: with patch("homeassistant.util.dt.utcnow", return_value=now): hass.states.async_set( usage_sensor_entity_id, "0", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "0.0" assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_MONETARY assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor assert state.attributes[ATTR_STATE_CLASS] == STATE_CLASS_TOTAL assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "EUR" # # Unique ID temp disabled # # entity_registry = er.async_get(hass) # # entry = entity_registry.async_get(cost_sensor_entity_id) # # assert entry.unique_id == "energy_energy_consumption cost" # Energy use bumped to 10 kWh hass.states.async_set( usage_sensor_entity_id, "10", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "10.0" # 0 EUR + (10-0) kWh * 1 EUR/kWh = 10 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Nothing happens when price changes if price_entity is not None: hass.states.async_set(price_entity, "2") await hass.async_block_till_done() else: energy_data = copy.deepcopy(energy_data) energy_data["energy_sources"][0][flow_type][0]["number_energy_price"] = 2 client = await hass_ws_client(hass) await client.send_json({"id": 5, "type": "energy/save_prefs", **energy_data}) msg = await client.receive_json() assert msg["success"] state = hass.states.get(cost_sensor_entity_id) assert state.state == "10.0" # 10 EUR + (10-10) kWh * 2 EUR/kWh = 10 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Additional consumption is using the new price hass.states.async_set( usage_sensor_entity_id, "14.5", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "19.0" # 10 EUR + (14.5-10) kWh * 2 EUR/kWh = 19 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Check generated statistics await async_wait_recording_done_without_instance(hass) all_statistics = await hass.loop.run_in_executor(None, _compile_statistics, hass) statistics = get_statistics_for_entity(all_statistics, cost_sensor_entity_id) assert statistics["stat"]["sum"] == 19.0 # Energy sensor has a small dip, no reset should be detected hass.states.async_set( usage_sensor_entity_id, "14", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "18.0" # 19 EUR + (14-14.5) kWh * 2 EUR/kWh = 18 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Energy sensor is reset, with initial state at 4kWh, 0 kWh is used as zero-point hass.states.async_set( usage_sensor_entity_id, "4", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "8.0" # 0 EUR + (4-0) kWh * 2 EUR/kWh = 8 EUR assert state.attributes[ATTR_LAST_RESET] != last_reset_cost_sensor last_reset_cost_sensor = state.attributes[ATTR_LAST_RESET] # Energy use bumped to 10 kWh hass.states.async_set( usage_sensor_entity_id, "10", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "20.0" # 8 EUR + (10-4) kWh * 2 EUR/kWh = 20 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Check generated statistics await async_wait_recording_done_without_instance(hass) all_statistics = await hass.loop.run_in_executor(None, _compile_statistics, hass) statistics = get_statistics_for_entity(all_statistics, cost_sensor_entity_id) assert statistics["stat"]["sum"] == 38.0 @pytest.mark.parametrize("initial_energy,initial_cost", [(0, "0.0"), (None, "unknown")]) @pytest.mark.parametrize( "price_entity,fixed_price", [("sensor.energy_price", None), (None, 1)] ) @pytest.mark.parametrize( "usage_sensor_entity_id,cost_sensor_entity_id,flow_type", [ ("sensor.energy_consumption", "sensor.energy_consumption_cost", "flow_from"), ( "sensor.energy_production", "sensor.energy_production_compensation", "flow_to", ), ], ) @pytest.mark.parametrize("energy_state_class", ["total", "measurement"]) async def test_cost_sensor_price_entity_total( hass, hass_storage, hass_ws_client, initial_energy, initial_cost, price_entity, fixed_price, usage_sensor_entity_id, cost_sensor_entity_id, flow_type, energy_state_class, ) -> None: """Test energy cost price from total type sensor entity.""" def _compile_statistics(_): return compile_statistics(hass, now, now + timedelta(seconds=1)) energy_attributes = { ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR, ATTR_STATE_CLASS: energy_state_class, } await async_init_recorder_component(hass) energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "grid", "flow_from": [ { "stat_energy_from": "sensor.energy_consumption", "entity_energy_from": "sensor.energy_consumption", "stat_cost": None, "entity_energy_price": price_entity, "number_energy_price": fixed_price, } ] if flow_type == "flow_from" else [], "flow_to": [ { "stat_energy_to": "sensor.energy_production", "entity_energy_to": "sensor.energy_production", "stat_compensation": None, "entity_energy_price": price_entity, "number_energy_price": fixed_price, } ] if flow_type == "flow_to" else [], "cost_adjustment_day": 0, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } now = dt_util.utcnow() last_reset = dt_util.utc_from_timestamp(0).isoformat() last_reset_cost_sensor = now.isoformat() # Optionally initialize dependent entities if initial_energy is not None: hass.states.async_set( usage_sensor_entity_id, initial_energy, {**energy_attributes, **{"last_reset": last_reset}}, ) hass.states.async_set("sensor.energy_price", "1") with patch("homeassistant.util.dt.utcnow", return_value=now): await setup_integration(hass) state = hass.states.get(cost_sensor_entity_id) assert state.state == initial_cost assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_MONETARY if initial_cost != "unknown": assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor assert state.attributes[ATTR_STATE_CLASS] == STATE_CLASS_TOTAL assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "EUR" # Optional late setup of dependent entities if initial_energy is None: with patch("homeassistant.util.dt.utcnow", return_value=now): hass.states.async_set( usage_sensor_entity_id, "0", {**energy_attributes, **{"last_reset": last_reset}}, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "0.0" assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_MONETARY assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor assert state.attributes[ATTR_STATE_CLASS] == STATE_CLASS_TOTAL assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "EUR" # # Unique ID temp disabled # # entity_registry = er.async_get(hass) # # entry = entity_registry.async_get(cost_sensor_entity_id) # # assert entry.unique_id == "energy_energy_consumption cost" # Energy use bumped to 10 kWh hass.states.async_set( usage_sensor_entity_id, "10", {**energy_attributes, **{"last_reset": last_reset}}, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "10.0" # 0 EUR + (10-0) kWh * 1 EUR/kWh = 10 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Nothing happens when price changes if price_entity is not None: hass.states.async_set(price_entity, "2") await hass.async_block_till_done() else: energy_data = copy.deepcopy(energy_data) energy_data["energy_sources"][0][flow_type][0]["number_energy_price"] = 2 client = await hass_ws_client(hass) await client.send_json({"id": 5, "type": "energy/save_prefs", **energy_data}) msg = await client.receive_json() assert msg["success"] state = hass.states.get(cost_sensor_entity_id) assert state.state == "10.0" # 10 EUR + (10-10) kWh * 2 EUR/kWh = 10 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Additional consumption is using the new price hass.states.async_set( usage_sensor_entity_id, "14.5", {**energy_attributes, **{"last_reset": last_reset}}, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "19.0" # 10 EUR + (14.5-10) kWh * 2 EUR/kWh = 19 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Check generated statistics await async_wait_recording_done_without_instance(hass) all_statistics = await hass.loop.run_in_executor(None, _compile_statistics, hass) statistics = get_statistics_for_entity(all_statistics, cost_sensor_entity_id) assert statistics["stat"]["sum"] == 19.0 # Energy sensor has a small dip hass.states.async_set( usage_sensor_entity_id, "14", {**energy_attributes, **{"last_reset": last_reset}}, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "18.0" # 19 EUR + (14-14.5) kWh * 2 EUR/kWh = 18 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Energy sensor is reset, with initial state at 4kWh, 0 kWh is used as zero-point last_reset = (now + timedelta(seconds=1)).isoformat() hass.states.async_set( usage_sensor_entity_id, "4", {**energy_attributes, **{"last_reset": last_reset}}, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "8.0" # 0 EUR + (4-0) kWh * 2 EUR/kWh = 8 EUR assert state.attributes[ATTR_LAST_RESET] != last_reset_cost_sensor last_reset_cost_sensor = state.attributes[ATTR_LAST_RESET] # Energy use bumped to 10 kWh hass.states.async_set( usage_sensor_entity_id, "10", {**energy_attributes, **{"last_reset": last_reset}}, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "20.0" # 8 EUR + (10-4) kWh * 2 EUR/kWh = 20 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Check generated statistics await async_wait_recording_done_without_instance(hass) all_statistics = await hass.loop.run_in_executor(None, _compile_statistics, hass) statistics = get_statistics_for_entity(all_statistics, cost_sensor_entity_id) assert statistics["stat"]["sum"] == 38.0 @pytest.mark.parametrize("initial_energy,initial_cost", [(0, "0.0"), (None, "unknown")]) @pytest.mark.parametrize( "price_entity,fixed_price", [("sensor.energy_price", None), (None, 1)] ) @pytest.mark.parametrize( "usage_sensor_entity_id,cost_sensor_entity_id,flow_type", [ ("sensor.energy_consumption", "sensor.energy_consumption_cost", "flow_from"), ( "sensor.energy_production", "sensor.energy_production_compensation", "flow_to", ), ], ) @pytest.mark.parametrize("energy_state_class", ["total"]) async def test_cost_sensor_price_entity_total_no_reset( hass, hass_storage, hass_ws_client, initial_energy, initial_cost, price_entity, fixed_price, usage_sensor_entity_id, cost_sensor_entity_id, flow_type, energy_state_class, ) -> None: """Test energy cost price from total type sensor entity with no last_reset.""" def _compile_statistics(_): return compile_statistics(hass, now, now + timedelta(seconds=1)) energy_attributes = { ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR, ATTR_STATE_CLASS: energy_state_class, } await async_init_recorder_component(hass) energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "grid", "flow_from": [ { "stat_energy_from": "sensor.energy_consumption", "entity_energy_from": "sensor.energy_consumption", "stat_cost": None, "entity_energy_price": price_entity, "number_energy_price": fixed_price, } ] if flow_type == "flow_from" else [], "flow_to": [ { "stat_energy_to": "sensor.energy_production", "entity_energy_to": "sensor.energy_production", "stat_compensation": None, "entity_energy_price": price_entity, "number_energy_price": fixed_price, } ] if flow_type == "flow_to" else [], "cost_adjustment_day": 0, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } now = dt_util.utcnow() last_reset_cost_sensor = now.isoformat() # Optionally initialize dependent entities if initial_energy is not None: hass.states.async_set( usage_sensor_entity_id, initial_energy, energy_attributes, ) hass.states.async_set("sensor.energy_price", "1") with patch("homeassistant.util.dt.utcnow", return_value=now): await setup_integration(hass) state = hass.states.get(cost_sensor_entity_id) assert state.state == initial_cost assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_MONETARY if initial_cost != "unknown": assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor assert state.attributes[ATTR_STATE_CLASS] == STATE_CLASS_TOTAL assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "EUR" # Optional late setup of dependent entities if initial_energy is None: with patch("homeassistant.util.dt.utcnow", return_value=now): hass.states.async_set( usage_sensor_entity_id, "0", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "0.0" assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_MONETARY assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor assert state.attributes[ATTR_STATE_CLASS] == STATE_CLASS_TOTAL assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "EUR" # # Unique ID temp disabled # # entity_registry = er.async_get(hass) # # entry = entity_registry.async_get(cost_sensor_entity_id) # # assert entry.unique_id == "energy_energy_consumption cost" # Energy use bumped to 10 kWh hass.states.async_set( usage_sensor_entity_id, "10", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "10.0" # 0 EUR + (10-0) kWh * 1 EUR/kWh = 10 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Nothing happens when price changes if price_entity is not None: hass.states.async_set(price_entity, "2") await hass.async_block_till_done() else: energy_data = copy.deepcopy(energy_data) energy_data["energy_sources"][0][flow_type][0]["number_energy_price"] = 2 client = await hass_ws_client(hass) await client.send_json({"id": 5, "type": "energy/save_prefs", **energy_data}) msg = await client.receive_json() assert msg["success"] state = hass.states.get(cost_sensor_entity_id) assert state.state == "10.0" # 10 EUR + (10-10) kWh * 2 EUR/kWh = 10 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Additional consumption is using the new price hass.states.async_set( usage_sensor_entity_id, "14.5", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "19.0" # 10 EUR + (14.5-10) kWh * 2 EUR/kWh = 19 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Check generated statistics await async_wait_recording_done_without_instance(hass) all_statistics = await hass.loop.run_in_executor(None, _compile_statistics, hass) statistics = get_statistics_for_entity(all_statistics, cost_sensor_entity_id) assert statistics["stat"]["sum"] == 19.0 # Energy sensor has a small dip hass.states.async_set( usage_sensor_entity_id, "14", energy_attributes, ) await hass.async_block_till_done() state = hass.states.get(cost_sensor_entity_id) assert state.state == "18.0" # 19 EUR + (14-14.5) kWh * 2 EUR/kWh = 18 EUR assert state.attributes[ATTR_LAST_RESET] == last_reset_cost_sensor # Check generated statistics await async_wait_recording_done_without_instance(hass) all_statistics = await hass.loop.run_in_executor(None, _compile_statistics, hass) statistics = get_statistics_for_entity(all_statistics, cost_sensor_entity_id) assert statistics["stat"]["sum"] == 18.0 @pytest.mark.parametrize( "energy_unit,factor", [ (ENERGY_WATT_HOUR, 1000), (ENERGY_KILO_WATT_HOUR, 1), (ENERGY_MEGA_WATT_HOUR, 0.001), ], ) async def test_cost_sensor_handle_energy_units( hass, hass_storage, energy_unit, factor ) -> None: """Test energy cost price from sensor entity.""" energy_attributes = { ATTR_UNIT_OF_MEASUREMENT: energy_unit, ATTR_STATE_CLASS: STATE_CLASS_TOTAL_INCREASING, } energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "grid", "flow_from": [ { "stat_energy_from": "sensor.energy_consumption", "entity_energy_from": "sensor.energy_consumption", "stat_cost": None, "entity_energy_price": None, "number_energy_price": 0.5, } ], "flow_to": [], "cost_adjustment_day": 0, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } now = dt_util.utcnow() # Initial state: 10kWh hass.states.async_set( "sensor.energy_consumption", 10 * factor, energy_attributes, ) with patch("homeassistant.util.dt.utcnow", return_value=now): await setup_integration(hass) state = hass.states.get("sensor.energy_consumption_cost") assert state.state == "0.0" # Energy use bumped by 10 kWh hass.states.async_set( "sensor.energy_consumption", 20 * factor, energy_attributes, ) await hass.async_block_till_done() state = hass.states.get("sensor.energy_consumption_cost") assert state.state == "5.0" @pytest.mark.parametrize( "price_unit,factor", [ (f"EUR/{ENERGY_WATT_HOUR}", 0.001), (f"EUR/{ENERGY_KILO_WATT_HOUR}", 1), (f"EUR/{ENERGY_MEGA_WATT_HOUR}", 1000), ], ) async def test_cost_sensor_handle_price_units( hass, hass_storage, price_unit, factor ) -> None: """Test energy cost price from sensor entity.""" energy_attributes = { ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR, ATTR_STATE_CLASS: STATE_CLASS_TOTAL_INCREASING, } price_attributes = { ATTR_UNIT_OF_MEASUREMENT: price_unit, ATTR_STATE_CLASS: STATE_CLASS_MEASUREMENT, } energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "grid", "flow_from": [ { "stat_energy_from": "sensor.energy_consumption", "entity_energy_from": "sensor.energy_consumption", "stat_cost": None, "entity_energy_price": "sensor.energy_price", "number_energy_price": None, } ], "flow_to": [], "cost_adjustment_day": 0, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } now = dt_util.utcnow() # Initial state: 10kWh hass.states.async_set("sensor.energy_price", "2", price_attributes) hass.states.async_set( "sensor.energy_consumption", 10 * factor, energy_attributes, ) with patch("homeassistant.util.dt.utcnow", return_value=now): await setup_integration(hass) state = hass.states.get("sensor.energy_consumption_cost") assert state.state == "0.0" # Energy use bumped by 10 kWh hass.states.async_set( "sensor.energy_consumption", 20 * factor, energy_attributes, ) await hass.async_block_till_done() state = hass.states.get("sensor.energy_consumption_cost") assert state.state == "20.0" async def test_cost_sensor_handle_gas(hass, hass_storage) -> None: """Test gas cost price from sensor entity.""" energy_attributes = { ATTR_UNIT_OF_MEASUREMENT: VOLUME_CUBIC_METERS, ATTR_STATE_CLASS: STATE_CLASS_TOTAL_INCREASING, } energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "gas", "stat_energy_from": "sensor.gas_consumption", "entity_energy_from": "sensor.gas_consumption", "stat_cost": None, "entity_energy_price": None, "number_energy_price": 0.5, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } now = dt_util.utcnow() hass.states.async_set( "sensor.gas_consumption", 100, energy_attributes, ) with patch("homeassistant.util.dt.utcnow", return_value=now): await setup_integration(hass) state = hass.states.get("sensor.gas_consumption_cost") assert state.state == "0.0" # gas use bumped to 10 kWh hass.states.async_set( "sensor.gas_consumption", 200, energy_attributes, ) await hass.async_block_till_done() state = hass.states.get("sensor.gas_consumption_cost") assert state.state == "50.0" async def test_cost_sensor_handle_gas_kwh(hass, hass_storage) -> None: """Test gas cost price from sensor entity.""" energy_attributes = { ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR, ATTR_STATE_CLASS: STATE_CLASS_TOTAL_INCREASING, } energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "gas", "stat_energy_from": "sensor.gas_consumption", "entity_energy_from": "sensor.gas_consumption", "stat_cost": None, "entity_energy_price": None, "number_energy_price": 0.5, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } now = dt_util.utcnow() hass.states.async_set( "sensor.gas_consumption", 100, energy_attributes, ) with patch("homeassistant.util.dt.utcnow", return_value=now): await setup_integration(hass) state = hass.states.get("sensor.gas_consumption_cost") assert state.state == "0.0" # gas use bumped to 10 kWh hass.states.async_set( "sensor.gas_consumption", 200, energy_attributes, ) await hass.async_block_till_done() state = hass.states.get("sensor.gas_consumption_cost") assert state.state == "50.0" @pytest.mark.parametrize("state_class", [None]) async def test_cost_sensor_wrong_state_class( hass, hass_storage, caplog, state_class ) -> None: """Test energy sensor rejects sensor with wrong state_class.""" energy_attributes = { ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR, ATTR_STATE_CLASS: state_class, } energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "grid", "flow_from": [ { "stat_energy_from": "sensor.energy_consumption", "entity_energy_from": "sensor.energy_consumption", "stat_cost": None, "entity_energy_price": None, "number_energy_price": 0.5, } ], "flow_to": [], "cost_adjustment_day": 0, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } now = dt_util.utcnow() hass.states.async_set( "sensor.energy_consumption", 10000, energy_attributes, ) with patch("homeassistant.util.dt.utcnow", return_value=now): await setup_integration(hass) state = hass.states.get("sensor.energy_consumption_cost") assert state.state == STATE_UNKNOWN assert ( f"Found unexpected state_class {state_class} for sensor.energy_consumption" in caplog.text ) # Energy use bumped to 10 kWh hass.states.async_set( "sensor.energy_consumption", 20000, energy_attributes, ) await hass.async_block_till_done() state = hass.states.get("sensor.energy_consumption_cost") assert state.state == STATE_UNKNOWN @pytest.mark.parametrize("state_class", [STATE_CLASS_MEASUREMENT]) async def test_cost_sensor_state_class_measurement_no_reset( hass, hass_storage, caplog, state_class ) -> None: """Test energy sensor rejects state_class measurement with no last_reset.""" energy_attributes = { ATTR_UNIT_OF_MEASUREMENT: ENERGY_KILO_WATT_HOUR, ATTR_STATE_CLASS: state_class, } energy_data = data.EnergyManager.default_preferences() energy_data["energy_sources"].append( { "type": "grid", "flow_from": [ { "stat_energy_from": "sensor.energy_consumption", "entity_energy_from": "sensor.energy_consumption", "stat_cost": None, "entity_energy_price": None, "number_energy_price": 0.5, } ], "flow_to": [], "cost_adjustment_day": 0, } ) hass_storage[data.STORAGE_KEY] = { "version": 1, "data": energy_data, } now = dt_util.utcnow() hass.states.async_set( "sensor.energy_consumption", 10000, energy_attributes, ) with patch("homeassistant.util.dt.utcnow", return_value=now): await setup_integration(hass) state = hass.states.get("sensor.energy_consumption_cost") assert state.state == STATE_UNKNOWN # Energy use bumped to 10 kWh hass.states.async_set( "sensor.energy_consumption", 20000, energy_attributes, ) await hass.async_block_till_done() state = hass.states.get("sensor.energy_consumption_cost") assert state.state == STATE_UNKNOWN