Phase 6 / Gate 3 — Security-Review auf den fertigen Code, 4 thematische Pakete: - P1 Secrets+Injection (client/phraser/diagnostics): validate haertet URL/Entity/ Unicode-Umgehungen (// , &entity; , Cc/Cf/Zl/Zp); kein Key-Leak. - P2 XSS+externe (card/ha_entity/frontend): kein XSS; DoS-Caps (Forecast 400, Karte 60). - P3 Mutator+Startup (coordinator/__init__/const): nicht-blockierender Startup via Background-Task, LLM-/Store-Ops defensiv, weather_entity_id-Guard, build_items NaN/inf+Caps, Warn-Log ohne Secret. - P4 Flows (config_flow): test_entity exception-frei, _clean_item defensiv, finite Bandgrenzen + sanitized_options-Normalisierung, echte bool-Coercion, Typ-Log. - 24 Findings, 14 uebernommen (mit Regressionstests), 10 verworfen mit Evidenz. Suite gruen 188/188. Gate 3 BESTANDEN. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
135 lines
5.3 KiB
Python
135 lines
5.3 KiB
Python
"""Story 3.2 — schema versioning, migration no-op, options robustness (AD-8)."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
from homeassistant.components.weather import WeatherEntityFeature
|
|
from homeassistant.core import HomeAssistant, ServiceResponse, SupportsResponse
|
|
from homeassistant.util import dt as dt_util
|
|
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
|
|
|
from custom_components.what_to_wear import const
|
|
from custom_components.what_to_wear.const import (
|
|
CONF_WEATHER_ENTITY,
|
|
CONFIG_MINOR_VERSION,
|
|
CONFIG_VERSION,
|
|
DOMAIN,
|
|
)
|
|
|
|
ENTITY = "weather.home"
|
|
|
|
|
|
def _set_weather(hass) -> None:
|
|
hass.states.async_set(
|
|
ENTITY, "cloudy",
|
|
{"temperature_unit": "°C", "wind_speed_unit": "km/h", "precipitation_unit": "mm",
|
|
"supported_features": WeatherEntityFeature.FORECAST_DAILY | WeatherEntityFeature.FORECAST_HOURLY},
|
|
)
|
|
|
|
|
|
def _register_forecast(hass) -> None:
|
|
async def handler(call) -> ServiceResponse:
|
|
base = dt_util.now()
|
|
entries = [{"datetime": (base + timedelta(days=o)).replace(hour=12), "temperature": 8.0,
|
|
"templow": 5.0} for o in range(-1, 4)]
|
|
return {call.data["entity_id"]: {"forecast": entries}}
|
|
|
|
hass.services.async_register("weather", "get_forecasts", handler,
|
|
supports_response=SupportsResponse.ONLY)
|
|
|
|
|
|
# --- pure sanitizer tests (AD-8 robustness contract) ------------------------
|
|
|
|
def test_sanitized_missing_uses_default() -> None:
|
|
opts = const.sanitized_options({})
|
|
assert opts == const.default_options()
|
|
|
|
|
|
def test_sanitized_type_wrong_falls_back_and_warns(caplog) -> None:
|
|
warnings = []
|
|
bad = {
|
|
"warmth_band_limits": "oops", # not a list
|
|
"cold_sensitivity_offset": "x", # not numeric
|
|
"switchover_time": 123, # not a string
|
|
"llm_enabled": "yes", # not a bool
|
|
"rain_prob_should": 55, # valid
|
|
}
|
|
opts = const.sanitized_options(bad, warn=lambda *a: warnings.append(a))
|
|
assert opts["warmth_band_limits"] == [0.0, 8.0, 15.0, 22.0]
|
|
assert opts["cold_sensitivity_offset"] == 0
|
|
assert opts["switchover_time"] == "10:00:00"
|
|
assert opts["llm_enabled"] is False
|
|
assert opts["rain_prob_should"] == 55 # the valid one is kept
|
|
assert len(warnings) == 4 # one per bad key
|
|
|
|
|
|
def test_sanitized_non_monotone_bands_rejected() -> None:
|
|
opts = const.sanitized_options({"warmth_band_limits": [0, 20, 15, 22]})
|
|
assert opts["warmth_band_limits"] == [0.0, 8.0, 15.0, 22.0] # default
|
|
|
|
|
|
def test_sanitized_unknown_key_ignored() -> None:
|
|
opts = const.sanitized_options({"unknown_future_key": 42})
|
|
assert "unknown_future_key" not in opts # logic ignores it (storage keeps it)
|
|
|
|
|
|
# --- migration + robust load in a real entry --------------------------------
|
|
|
|
async def test_entry_carries_version(hass: HomeAssistant) -> None:
|
|
_set_weather(hass)
|
|
_register_forecast(hass)
|
|
entry = MockConfigEntry(domain=DOMAIN, data={CONF_WEATHER_ENTITY: ENTITY},
|
|
version=CONFIG_VERSION, minor_version=CONFIG_MINOR_VERSION)
|
|
entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert entry.version == CONFIG_VERSION and entry.minor_version == CONFIG_MINOR_VERSION
|
|
|
|
|
|
async def test_setup_survives_corrupt_options(hass: HomeAssistant) -> None:
|
|
_set_weather(hass)
|
|
_register_forecast(hass)
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN, data={CONF_WEATHER_ENTITY: ENTITY},
|
|
options={"warmth_band_limits": "corrupt", "rain_prob_should": "NaN", "future_key": 1},
|
|
)
|
|
entry.add_to_hass(hass)
|
|
# setup must not crash despite corrupt options (AD-8)
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
o = coordinator._options()
|
|
assert o["warmth_band_limits"] == [0.0, 8.0, 15.0, 22.0] # defaulted
|
|
assert o["rain_prob_should"] == 40 # defaulted
|
|
assert coordinator.data is not None # still produced a recommendation
|
|
|
|
|
|
async def test_build_items_rejects_nan_and_corrupt(hass: HomeAssistant) -> None:
|
|
from homeassistant.config_entries import ConfigSubentryData
|
|
|
|
from custom_components.what_to_wear.coordinator import build_items
|
|
|
|
_set_weather(hass)
|
|
_register_forecast(hass)
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN, data={CONF_WEATHER_ENTITY: ENTITY},
|
|
subentries_data=[ConfigSubentryData(
|
|
data={"name": "Weird", "category": "top", "warmth": 2,
|
|
"temp_min": float("nan"), "temp_max": float("inf")},
|
|
subentry_type="item", title="Weird", unique_id=None)],
|
|
)
|
|
entry.add_to_hass(hass)
|
|
item = build_items(entry)[0]
|
|
assert item.temp_min is None and item.temp_max is None # NaN/inf rejected
|
|
|
|
|
|
async def test_setup_survives_missing_weather_entity(hass: HomeAssistant) -> None:
|
|
# corrupt entry without the weather entity id must not crash setup (AD-2/AD-7)
|
|
entry = MockConfigEntry(domain=DOMAIN, data={}, options={})
|
|
entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
assert coordinator.data is not None
|
|
assert coordinator.data.status == "fehler_prognose"
|