buzeraandClaude Opus 5.5
9bb64e79ca
Sensor weights already existed, but only as a second step after saving
a zone with two or more sensors, with fields labelled by raw keys such
as weight__sensor.living_a, so they were easy to miss.
The zone form now says that choosing two or more sensors leads to a
sensor-weight step. That step uses numbered fields ("Sensor 1 weight",
English and Romanian) and its description lists which sensor is which
by friendly name. Editing a multi-sensor zone shows the saved weights.
Stored data is unchanged.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
264 lines
16 KiB
Python
264 lines
16 KiB
Python
"""Validate actual HA form schemas, options, zone subentries and entry migration."""
|
|
from homeassistant.config_entries import SOURCE_RECONFIGURE, SOURCE_USER
|
|
from homeassistant.data_entry_flow import FlowResultType
|
|
from homeassistant.helpers import entity_registry as er
|
|
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
|
|
|
from custom_components.heating_mpc.const import DOMAIN
|
|
from custom_components.heating_mpc.weather_sources import sources_for
|
|
|
|
MAIN = {"climate_entity": "climate.reference", "heater_entity": "switch.boiler"}
|
|
|
|
|
|
def zones_of(entry):
|
|
return {sub.unique_id: sub for sub in entry.subentries.values() if sub.subentry_type == "zone"}
|
|
|
|
|
|
async def setup_entry(hass, entry):
|
|
entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
return hass.data[DOMAIN][entry.entry_id]
|
|
|
|
|
|
async def test_config_flow_creates_one_zone_per_sensor(hass, inputs):
|
|
hass.states.async_set("sensor.bedroom", "19", {"unit_of_measurement": "°C", "friendly_name": "Bedroom"})
|
|
result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
|
|
assert result["type"] is FlowResultType.FORM
|
|
validated = result["data_schema"]({**MAIN, "room_entities": ["sensor.room", "sensor.bedroom"]})
|
|
assert validated["room_max_age_minutes"] == 30
|
|
assert validated["weather_max_age_minutes"] == 120
|
|
assert validated["room_stale_use_last_value"] is False
|
|
result = await hass.config_entries.flow.async_configure(result["flow_id"], validated)
|
|
assert result["step_id"] == "weather"
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], result["data_schema"]({"outside_temperature_sources": ["sensor.outside"]}))
|
|
assert result["step_id"] == "weather_extra"
|
|
result = await hass.config_entries.flow.async_configure(result["flow_id"], result["data_schema"]({}))
|
|
assert result["step_id"] == "weights"
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {"weight__sensor.room": 2, "weight__sensor.bedroom": 1})
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
await hass.async_block_till_done()
|
|
entry = result["result"]
|
|
assert entry.version == 3
|
|
assert "room_entities" not in entry.data and "weights" not in entry.data
|
|
zones = zones_of(entry)
|
|
assert {key: (sub.title, sub.data["comfort_weight"]) for key, sub in zones.items()} == {
|
|
"sensor.room": ("room", 2.0), "sensor.bedroom": ("Bedroom", 1.0)} # HA's name fallback
|
|
assert hass.data[DOMAIN][entry.entry_id].rooms == ["sensor.room", "sensor.bedroom"]
|
|
|
|
|
|
async def test_options_flow_keeps_settings_without_rooms(hass, entry, inputs):
|
|
await setup_entry(hass, entry)
|
|
result = await hass.config_entries.options.async_init(entry.entry_id)
|
|
assert "room_entities" not in {str(key) for key in result["data_schema"].schema}
|
|
main = result["data_schema"]({**MAIN, "room_max_age_minutes": 45, "room_stale_use_last_value": True})
|
|
result = await hass.config_entries.options.async_configure(result["flow_id"], main)
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"], result["data_schema"]({"outside_temperature_sources": ["sensor.outside"]}))
|
|
result = await hass.config_entries.options.async_configure(result["flow_id"], result["data_schema"]({}))
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
await hass.async_block_till_done()
|
|
assert entry.options["room_max_age_minutes"] == 45
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
assert coordinator.config["room_max_age_minutes"] == 45
|
|
assert coordinator.config["room_stale_use_last_value"] is True
|
|
assert coordinator.rooms == ["sensor.room"] # zones are untouched by Configure
|
|
result = await hass.config_entries.options.async_init(entry.entry_id)
|
|
assert result["data_schema"](MAIN)["room_stale_use_last_value"] is True
|
|
hass.config_entries.options.async_abort(result["flow_id"])
|
|
|
|
|
|
async def test_weather_source_may_not_be_a_zone_sensor(hass, entry, inputs):
|
|
await setup_entry(hass, entry)
|
|
result = await hass.config_entries.options.async_init(entry.entry_id)
|
|
result = await hass.config_entries.options.async_configure(result["flow_id"], result["data_schema"](MAIN))
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"], result["data_schema"]({"outside_temperature_sources": ["sensor.room"]}))
|
|
assert result["errors"] == {"base": "invalid_weather"}
|
|
|
|
|
|
async def test_v1_entry_migrates_to_zones_on_setup(hass, inputs):
|
|
entry = MockConfigEntry(domain=DOMAIN, version=1, data={
|
|
**MAIN, "room_entities": ["sensor.room"], "outside_entity": "sensor.outside"})
|
|
await setup_entry(hass, entry)
|
|
assert entry.version == 3
|
|
assert entry.data["outside_temperature_sources"] == ["sensor.outside"]
|
|
assert "outside_entity" not in entry.data and "room_entities" not in entry.data
|
|
assert set(zones_of(entry)) == {"sensor.room"}
|
|
|
|
|
|
async def test_v2_migration_keeps_weights_models_and_entities(hass, entry, inputs, hass_storage):
|
|
from custom_components.heating_mpc.physics import Learner
|
|
hass.states.async_set("sensor.room", "20", {"unit_of_measurement": "°C", "friendly_name": "Living room"})
|
|
hass_storage[f"{DOMAIN}.{entry.entry_id}"] = {"version": 1, "data": {
|
|
"schema_version": 2, "heater_entity": "switch.boiler",
|
|
"learners": {"sensor.room": Learner(count=250, on_count=125, off_count=125).as_dict()}}}
|
|
entry.add_to_hass(hass)
|
|
# A forecast entity created by 0.3 for this room.
|
|
registry = er.async_get(hass)
|
|
old = registry.async_get_or_create("sensor", DOMAIN, f"{entry.entry_id}_sensor.room_predicted_60min",
|
|
config_entry=entry, suggested_object_id="mpc_living_forecast")
|
|
hass.config_entries.async_update_entry(entry, options={"weights": {"sensor.room": 4.0}})
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
zone = zones_of(entry)["sensor.room"]
|
|
assert (zone.title, zone.data["comfort_weight"], zone.data["sensors"]) == ("Living room", 4.0, ["sensor.room"])
|
|
assert "weights" not in entry.options
|
|
assert coordinator.learners["sensor.room"].count == 250
|
|
# Same entity, now linked to its zone, so deleting the zone removes it.
|
|
migrated = registry.async_get(old.entity_id)
|
|
assert migrated.config_subentry_id == zone.subentry_id
|
|
assert hass.config_entries.async_remove_subentry(entry, zone.subentry_id)
|
|
await hass.async_block_till_done()
|
|
assert registry.async_get(old.entity_id) is None
|
|
|
|
|
|
async def test_zone_subentry_add_edit_and_validation(hass, entry, inputs):
|
|
await setup_entry(hass, entry)
|
|
for entity, value in (("sensor.living_a", "21"), ("sensor.living_b", "22")):
|
|
hass.states.async_set(entity, value, {"unit_of_measurement": "°C"})
|
|
flows = hass.config_entries.subentries
|
|
result = await flows.async_init((entry.entry_id, "zone"), context={"source": SOURCE_USER})
|
|
assert result["type"] is FlowResultType.FORM
|
|
# A sensor of another zone, or an outdoor source, is refused.
|
|
for sensors, error in ((["sensor.room"], "sensor_in_other_zone"), (["sensor.outside"], "sensor_is_weather_source")):
|
|
result = await flows.async_configure(result["flow_id"], {
|
|
"name": "Living", "sensors": sensors, "comfort_weight": 2, "target_offset": 0})
|
|
assert result["errors"] == {"base": error}
|
|
result = await flows.async_configure(result["flow_id"], {
|
|
"name": " Living ", "sensors": ["sensor.living_a", "sensor.living_b"], "comfort_weight": 2,
|
|
"target_offset": 0.5, "neighbors": ["sensor.room"]})
|
|
assert result["step_id"] == "sensor_weights"
|
|
# Numbered, translated fields; the description says which sensor is which.
|
|
assert [str(key) for key in result["data_schema"].schema] == ["sensor_weight_1", "sensor_weight_2"]
|
|
assert result["description_placeholders"]["sensors"].splitlines() == [
|
|
"1. living a (sensor.living_a)", "2. living b (sensor.living_b)"] # HA name fallback
|
|
result = await flows.async_configure(result["flow_id"], {"sensor_weight_1": 3, "sensor_weight_2": 1})
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
await hass.async_block_till_done()
|
|
living = next(sub for sub in zones_of(entry).values() if sub.title == "Living")
|
|
assert living.data["sensor_weights"] == {"sensor.living_a": 3.0, "sensor.living_b": 1.0}
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
key = living.data["key"]
|
|
assert coordinator.zones[key].neighbors == ("sensor.room",)
|
|
assert coordinator.zones["sensor.room"].neighbors == (key,) # symmetric
|
|
assert coordinator.data["temperatures"][key] == 21.25
|
|
# Edit: drop a sensor; the key (and so the learned model) is kept.
|
|
result = await flows.async_init((entry.entry_id, "zone"), context={
|
|
"source": SOURCE_RECONFIGURE, "subentry_id": living.subentry_id})
|
|
result = await flows.async_configure(result["flow_id"], {
|
|
"name": "Living", "sensors": ["sensor.living_a"], "comfort_weight": 1, "target_offset": 0})
|
|
assert result["type"] is FlowResultType.ABORT and result["reason"] == "reconfigure_successful"
|
|
# Editing a multi-sensor zone shows the saved sensor weights.
|
|
result = await flows.async_init((entry.entry_id, "zone"), context={
|
|
"source": SOURCE_RECONFIGURE, "subentry_id": living.subentry_id})
|
|
result = await flows.async_configure(result["flow_id"], {
|
|
"name": "Living", "sensors": ["sensor.living_a", "sensor.living_b"], "comfort_weight": 1, "target_offset": 0})
|
|
assert result["data_schema"]({})["sensor_weight_1"] == 1.0
|
|
flows.async_abort(result["flow_id"])
|
|
await hass.async_block_till_done()
|
|
assert entry.subentries[living.subentry_id].data["key"] == key
|
|
assert hass.data[DOMAIN][entry.entry_id].zones[key].sensors == ("sensor.living_a",)
|
|
|
|
|
|
async def test_options_flow_clears_optional_weather_sources(hass, inputs):
|
|
"""Swapping providers in the UI must allow removing an optional source list."""
|
|
entry = MockConfigEntry(domain=DOMAIN, version=2, unique_id="home_heating", data={
|
|
**MAIN, "room_entities": ["sensor.room"], "outside_temperature_sources": ["sensor.outside"],
|
|
"outside_wind_sources": ["sensor.wind"], "outside_pressure_sources": ["sensor.pressure"],
|
|
"weights": {"sensor.room": 1.0}})
|
|
await setup_entry(hass, entry)
|
|
flow = hass.config_entries.options
|
|
result = await flow.async_init(entry.entry_id)
|
|
result = await flow.async_configure(result["flow_id"], result["data_schema"](MAIN))
|
|
markers = {str(key): key for key in result["data_schema"].schema}
|
|
assert markers["outside_wind_sources"].description == {"suggested_value": ["sensor.wind"]}
|
|
# The user removes the wind provider; a cleared field may be omitted on submit.
|
|
result = await flow.async_configure(result["flow_id"], result["data_schema"](
|
|
{"outside_temperature_sources": ["sensor.outside"]}))
|
|
result = await flow.async_configure(result["flow_id"], result["data_schema"](
|
|
{"outside_pressure_sources": ["sensor.pressure"]}))
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
await hass.async_block_till_done()
|
|
assert entry.options["outside_wind_sources"] == []
|
|
assert entry.options["outside_pressure_sources"] == ["sensor.pressure"]
|
|
config = hass.data[DOMAIN][entry.entry_id].config
|
|
assert sources_for(config, "wind") == []
|
|
assert sources_for(config, "pressure") == ["sensor.pressure"]
|
|
|
|
|
|
async def test_heat_source_subentry_flow(hass, entry, inputs):
|
|
await setup_entry(hass, entry)
|
|
for sensor, value in (("sensor.ups_power", "600"), ("sensor.pc_power", "400")):
|
|
hass.states.async_set(sensor, value, {"unit_of_measurement": "W", "device_class": "power"})
|
|
flows = hass.config_entries.subentries
|
|
base = {"zone": "sensor.room", "heat_fraction": 1, "response_minutes": 15, "enabled": True}
|
|
result = await flows.async_init((entry.entry_id, "heat_source"), context={"source": SOURCE_USER})
|
|
result = await flows.async_configure(result["flow_id"], {
|
|
**base, "name": "UPS", "sensor": "sensor.ups_power", "baseline_mode": "auto"})
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
await hass.async_block_till_done()
|
|
ups = next(sub for sub in entry.subentries.values() if sub.subentry_type == "heat_source")
|
|
# The PC is behind the UPS: its power is already in the UPS reading.
|
|
result = await flows.async_init((entry.entry_id, "heat_source"), context={"source": SOURCE_USER})
|
|
pc = {**base, "name": "Gaming PC", "sensor": "sensor.pc_power", "baseline_mode": "manual",
|
|
"included_in": ups.unique_id, "response_minutes": 10}
|
|
result = await flows.async_configure(result["flow_id"], pc)
|
|
assert result["errors"] == {"base": "baseline_required"}
|
|
result = await flows.async_configure(result["flow_id"], {**pc, "baseline_w": 0})
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
await hass.async_block_till_done()
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
assert [(s.name, s.included_in) for s in coordinator.sources] == [("UPS", None), ("Gaming PC", ups.unique_id)]
|
|
assert coordinator.extended["sensor.room"].features == ["loss", "heat", "bias", "internal"]
|
|
registry = er.async_get(hass)
|
|
status = hass.states.get(registry.async_get_entity_id("sensor", DOMAIN, f"{entry.entry_id}_status"))
|
|
sources = status.attributes["heat_sources"]
|
|
assert (sources["UPS"]["power_w"], sources["UPS"]["effective_w"]) == (600, 200) # PC not counted twice
|
|
assert (sources["Gaming PC"]["included_in"], sources["Gaming PC"]["variable_w"]) == ("UPS", 400)
|
|
|
|
|
|
async def test_window_subentry_flow_and_ventilation_options(hass, entry, inputs):
|
|
await setup_entry(hass, entry)
|
|
hass.states.async_set("binary_sensor.bedroom_window", "off", {"device_class": "window"})
|
|
flows = hass.config_entries.subentries
|
|
result = await flows.async_init((entry.entry_id, "window"), context={"source": SOURCE_USER})
|
|
form = result["data_schema"]({"name": "Bedroom window", "sensor": "binary_sensor.bedroom_window",
|
|
"zone": "sensor.room"})
|
|
assert (form["azimuth"], form["opening_factor"], form["inverted"]) == (345.0, 1.0, False)
|
|
result = await flows.async_configure(result["flow_id"], {**form, "azimuth": 340})
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
await hass.async_block_till_done()
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
assert [(w.name, w.azimuth) for w in coordinator.windows] == [("Bedroom window", 340.0)]
|
|
# No wind sources yet: only the terms whose inputs exist.
|
|
assert coordinator.ventilation["sensor.room"].features == ["vent_natural", "recovery"]
|
|
result = await flows.async_init((entry.entry_id, "window"), context={"source": SOURCE_USER})
|
|
result = await flows.async_configure(result["flow_id"], {**form, "name": "Again"})
|
|
assert result["errors"] == {"base": "sensor_in_other_window"}
|
|
flows.async_abort(result["flow_id"])
|
|
# Wind, direction and gust sources plus the gust option add their terms.
|
|
for sensor, value, unit in (("sensor.wind", "4", "m/s"), ("sensor.wind_dir", "204", "°"),
|
|
("sensor.gust", "9", "m/s")):
|
|
hass.states.async_set(sensor, value, {"unit_of_measurement": unit})
|
|
options = hass.config_entries.options
|
|
result = await options.async_init(entry.entry_id)
|
|
result = await options.async_configure(result["flow_id"], result["data_schema"](
|
|
{**MAIN, "ventilation_gust": True, "ventilation_cross": True}))
|
|
result = await options.async_configure(result["flow_id"], result["data_schema"]({
|
|
"outside_temperature_sources": ["sensor.outside"], "outside_wind_sources": ["sensor.wind"],
|
|
"outside_wind_direction_sources": ["sensor.wind_dir"], "outside_wind_gust_sources": ["sensor.gust"]}))
|
|
result = await options.async_configure(result["flow_id"], result["data_schema"]({}))
|
|
await hass.async_block_till_done()
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
# Cross-ventilation needs windows in two zones, so it stays off.
|
|
assert coordinator.ventilation["sensor.room"].features == [
|
|
"vent_natural", "recovery", "vent_wind", "vent_wind_dir", "vent_gust"]
|
|
status = hass.states.get(er.async_get(hass).async_get_entity_id("sensor", DOMAIN, f"{entry.entry_id}_status"))
|
|
window = status.attributes["windows"]["Bedroom window"]
|
|
assert (window["state"], window["relative_wind_direction"]) == ("closed", -136.0)
|
|
assert {"windows", "ventilation"} <= status.state_info["unrecorded_attributes"]
|