d603947a14
Story 1.9 (TDD, phcc, Suite gruen 108/108) — schliesst Epic 1 ab: - sensor.py: CoordinatorEntity+SensorEntity, stabile ID sensor.what_to_wear (unique_id + self.entity_id vor add_entities), Stale-Besitz (Einmal-Timer auf last_success+6h), native_value=to_state, extra_state_attributes=to_attributes (kein llm_text), leerer Schrank=ok (AD-2/14/15). - __init__.py PLATFORMS=[SENSOR]; translations entity-Name de/en. - luna-pro-Story-Review: 2/3 Findings uebernommen, 1 verworfen (to_state-Evidenz). Epic 1 (Wetter->Outfit->Sensor) vollstaendig: 9 Stories, installierbar & sichtbar. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
"""Story 1.9 — the recommendation sensor (phcc)."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
|
|
from homeassistant.components.weather import WeatherEntityFeature
|
|
from homeassistant.core import HomeAssistant, ServiceResponse, SupportsResponse
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.util import dt as dt_util
|
|
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
|
|
|
from custom_components.what_to_wear.const import CONF_WEATHER_ENTITY, DOMAIN, SENSOR_ENTITY_ID
|
|
|
|
ENTITY = "weather.home"
|
|
|
|
|
|
def _set_weather(hass, state="cloudy") -> None:
|
|
hass.states.async_set(
|
|
ENTITY,
|
|
state,
|
|
{
|
|
"temperature_unit": "°C",
|
|
"wind_speed_unit": "km/h",
|
|
"precipitation_unit": "mm",
|
|
"supported_features": WeatherEntityFeature.FORECAST_DAILY
|
|
| WeatherEntityFeature.FORECAST_HOURLY,
|
|
},
|
|
)
|
|
|
|
|
|
def _register_forecast(hass, temp=5.0) -> None:
|
|
async def handler(call) -> ServiceResponse:
|
|
base = dt_util.now()
|
|
entries = [
|
|
{"datetime": (base + timedelta(days=off)).replace(hour=12), "temperature": temp + 3,
|
|
"templow": temp}
|
|
for off in range(-1, 4)
|
|
]
|
|
if call.data["type"] == "hourly":
|
|
entries = [
|
|
{"datetime": (base + timedelta(days=off)).replace(hour=7), "temperature": temp}
|
|
for off in range(-1, 4)
|
|
]
|
|
return {call.data["entity_id"]: {"forecast": entries}}
|
|
|
|
hass.services.async_register(
|
|
"weather", "get_forecasts", handler, supports_response=SupportsResponse.ONLY
|
|
)
|
|
|
|
|
|
def _item(name, category, warmth):
|
|
from homeassistant.config_entries import ConfigSubentryData
|
|
|
|
return ConfigSubentryData(
|
|
data={"name": name, "category": category, "warmth": warmth},
|
|
subentry_type="item",
|
|
title=name,
|
|
unique_id=None,
|
|
)
|
|
|
|
|
|
async def _setup(hass, subentries=(), temp=5.0):
|
|
_set_weather(hass)
|
|
_register_forecast(hass, temp)
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN, data={CONF_WEATHER_ENTITY: ENTITY}, options={},
|
|
subentries_data=list(subentries),
|
|
)
|
|
entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
return entry
|
|
|
|
|
|
async def test_sensor_exists_with_stable_id(hass: HomeAssistant) -> None:
|
|
await _setup(hass, [
|
|
_item("Shirt", "top", 3), _item("Jeans", "bottom", 3), _item("Boots", "shoes", 3),
|
|
])
|
|
state = hass.states.get(SENSOR_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.state # non-empty short text
|
|
assert len(state.state) <= 255
|
|
attrs = state.attributes
|
|
assert attrs["status"] == "ok"
|
|
assert attrs["target_date"]
|
|
assert "items" in attrs and "gaps" in attrs
|
|
assert "stale" in attrs
|
|
assert "llm_text" not in attrs # never in attributes (AD-20)
|
|
|
|
|
|
async def test_sensor_empty_wardrobe_names_gaps(hass: HomeAssistant) -> None:
|
|
await _setup(hass) # no items
|
|
state = hass.states.get(SENSOR_ENTITY_ID)
|
|
assert state is not None
|
|
assert state.attributes["status"] == "ok" # empty wardrobe is not an error
|
|
assert state.attributes["gaps"] # base categories named
|
|
|
|
|
|
async def test_sensor_stale_flag(hass: HomeAssistant) -> None:
|
|
entry = await _setup(hass, [_item("Shirt", "top", 3), _item("Jeans", "bottom", 3),
|
|
_item("Boots", "shoes", 3)])
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
# not stale right after a fresh compute
|
|
assert hass.states.get(SENSOR_ENTITY_ID).attributes["stale"] is False
|
|
# force the last success far into the past and re-write the state
|
|
coordinator.last_success_utc = dt_util.utcnow() - timedelta(hours=7)
|
|
async_dispatch = coordinator.async_update_listeners
|
|
async_dispatch()
|
|
await hass.async_block_till_done()
|
|
assert hass.states.get(SENSOR_ENTITY_ID).attributes["stale"] is True
|