From ccc31d32861150530aa41dc98942a5ab72772617 Mon Sep 17 00:00:00 2001 From: Michael Sauer Date: Wed, 16 Sep 2026 16:45:24 +0200 Subject: [PATCH] feat(update): Firmware-Update aus dem Ordner mit Einmal-URLs und WLAN-Sicherung --- custom_components/ninlight/__init__.py | 32 +- custom_components/ninlight/diagnostics.py | 29 +- custom_components/ninlight/http.py | 74 +++++ custom_components/ninlight/shared.py | 75 +++++ custom_components/ninlight/update.py | 232 +++++++++++++++ tests/conftest.py | 9 + tests/test_diagnostics.py | 14 + tests/test_http.py | 84 ++++++ tests/test_update.py | 339 ++++++++++++++++++++++ 9 files changed, 878 insertions(+), 10 deletions(-) create mode 100644 custom_components/ninlight/http.py create mode 100644 custom_components/ninlight/shared.py create mode 100644 custom_components/ninlight/update.py create mode 100644 tests/test_http.py create mode 100644 tests/test_update.py diff --git a/custom_components/ninlight/__init__.py b/custom_components/ninlight/__init__.py index 88d40d3..a15d9de 100644 --- a/custom_components/ninlight/__init__.py +++ b/custom_components/ninlight/__init__.py @@ -4,14 +4,25 @@ from __future__ import annotations from homeassistant.const import CONF_HOST, CONF_TOKEN, Platform from homeassistant.core import HomeAssistant, callback -from homeassistant.helpers import device_registry as dr -from homeassistant.helpers import issue_registry as ir +from homeassistant.helpers import ( + config_validation as cv, +) +from homeassistant.helpers import ( + device_registry as dr, +) +from homeassistant.helpers import ( + issue_registry as ir, +) from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.device_registry import format_mac +from homeassistant.helpers.typing import ConfigType from .api import NinlightClient from .const import DEFAULT_TOKEN, DOMAIN from .coordinator import NinlightConfigEntry, NinlightCoordinator +from .shared import async_get_shared, async_start_catalog, async_stop_catalog + +CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) PLATFORMS: list[Platform] = [ Platform.BUTTON, @@ -20,9 +31,15 @@ PLATFORMS: list[Platform] = [ Platform.SELECT, Platform.SENSOR, Platform.SWITCH, + Platform.UPDATE, ] +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: + async_get_shared(hass) + return True + + async def async_setup_entry(hass: HomeAssistant, entry: NinlightConfigEntry) -> bool: client = NinlightClient( async_get_clientsession(hass), entry.data[CONF_HOST], entry.data[CONF_TOKEN] @@ -43,16 +60,25 @@ async def async_setup_entry(hass: HomeAssistant, entry: NinlightConfigEntry) -> registry.async_update_device(device.id, sw_version=info["fw_version"]) entry.async_on_unload(coordinator.async_add_listener(_sync_firmware_version)) + await async_start_catalog(hass, async_get_shared(hass)) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) return True async def async_unload_entry(hass: HomeAssistant, entry: NinlightConfigEntry) -> bool: - return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + unloaded = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + if unloaded and not [ + other + for other in hass.config_entries.async_loaded_entries(DOMAIN) + if other.entry_id != entry.entry_id + ]: + async_stop_catalog(async_get_shared(hass)) + return unloaded async def async_remove_entry(hass: HomeAssistant, entry: NinlightConfigEntry) -> None: ir.async_delete_issue(hass, DOMAIN, f"default_token_{entry.entry_id}") + ir.async_delete_issue(hass, DOMAIN, f"wifi_guard_{entry.entry_id}") @callback diff --git a/custom_components/ninlight/diagnostics.py b/custom_components/ninlight/diagnostics.py index 5c9478b..fd07536 100644 --- a/custom_components/ninlight/diagnostics.py +++ b/custom_components/ninlight/diagnostics.py @@ -9,6 +9,7 @@ from homeassistant.const import CONF_HOST, CONF_TOKEN from homeassistant.core import HomeAssistant from .coordinator import NinlightConfigEntry +from .shared import DATA_SHARED TO_REDACT = {CONF_TOKEN, CONF_HOST, "ssid", "mac", "ip", "sta_ip"} @@ -17,11 +18,25 @@ async def async_get_config_entry_diagnostics( hass: HomeAssistant, entry: NinlightConfigEntry ) -> dict[str, Any]: coordinator = entry.runtime_data - return async_redact_data( - { - "entry": {"data": dict(entry.data), "options": dict(entry.options)}, - "state": coordinator.data.state, - "info": coordinator.data.info, - }, - TO_REDACT, + device_type = coordinator.data.info.get("device_type") + shared = hass.data.get(DATA_SHARED) + images = ( + [] + if shared is None + else [ + {"file": image.path.name, "device_type": image.device_type, "version": image.version} + for image in shared.catalog.images + if image.device_type == device_type + ] ) + return { + **async_redact_data( + { + "entry": {"data": dict(entry.data), "options": dict(entry.options)}, + "state": coordinator.data.state, + "info": coordinator.data.info, + }, + TO_REDACT, + ), + "firmware_images": images, + } diff --git a/custom_components/ninlight/http.py b/custom_components/ninlight/http.py new file mode 100644 index 0000000..2813c05 --- /dev/null +++ b/custom_components/ninlight/http.py @@ -0,0 +1,74 @@ +"""Serve firmware images to lamps under one-time URLs.""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +import secrets +from typing import Any + +from aiohttp import web +from homeassistant.components.http import HomeAssistantView +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.network import NoURLAvailableError, get_url + +from .const import CONF_LAMP_BASE_URL, DOMAIN, FIRMWARE_URL_PREFIX + + +class OtaRegistry: + """Random keys, each bound to exactly one file, valid only while its update runs.""" + + def __init__(self) -> None: + self._paths: dict[str, Path] = {} + + def create(self, path: Path) -> str: + key = secrets.token_urlsafe(32) + self._paths[key] = path + return key + + def revoke(self, key: str) -> None: + self._paths.pop(key, None) + + def get(self, key: str) -> Path | None: + return self._paths.get(key) + + def __len__(self) -> int: + return len(self._paths) + + +class NinlightFirmwareView(HomeAssistantView): + """No HA login: the lamp cannot authenticate. The unguessable key is the protection.""" + + url = FIRMWARE_URL_PREFIX + "{key}" + name = "api:ninlight:firmware" + requires_auth = False + + def __init__(self, hass: HomeAssistant, registry: OtaRegistry) -> None: + self._hass = hass + self._registry = registry + + async def get(self, request: web.Request, key: str) -> web.Response: + path = self._registry.get(key) + if path is None: + return web.Response(status=404) + try: + body = await self._hass.async_add_executor_job(path.read_bytes) + except OSError: + return web.Response(status=404) + return web.Response(body=body, content_type="application/octet-stream") + + +def lamp_base_url(hass: HomeAssistant, options: Mapping[str, Any]) -> str: + """The http:// address lamps use to reach Home Assistant.""" + base = options.get(CONF_LAMP_BASE_URL) + if not base: + try: + base = get_url(hass, allow_external=False, allow_cloud=False, prefer_external=False) + except NoURLAvailableError as err: + raise HomeAssistantError( + translation_domain=DOMAIN, translation_key="no_http_url" + ) from err + if not str(base).startswith("http://"): + raise HomeAssistantError(translation_domain=DOMAIN, translation_key="no_http_url") + return str(base) diff --git a/custom_components/ninlight/shared.py b/custom_components/ninlight/shared.py new file mode 100644 index 0000000..b0cd4c7 --- /dev/null +++ b/custom_components/ninlight/shared.py @@ -0,0 +1,75 @@ +"""State shared by all ninlight entries: the firmware catalog and one-time download keys.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path + +from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback +from homeassistant.helpers.dispatcher import async_dispatcher_send +from homeassistant.helpers.event import async_track_time_interval +from homeassistant.util.hass_dict import HassKey + +from .const import CATALOG_SCAN_INTERVAL, DOMAIN, FIRMWARE_DIR_NAME, SIGNAL_CATALOG_UPDATED +from .firmware import FirmwareCatalog +from .http import NinlightFirmwareView, OtaRegistry + + +def firmware_directory(hass: HomeAssistant) -> Path: + return Path(hass.config.path(FIRMWARE_DIR_NAME)) + + +@dataclass +class NinlightShared: + """One instance per Home Assistant.""" + + catalog: FirmwareCatalog + registry: OtaRegistry + scan_lock: asyncio.Lock = field(default_factory=asyncio.Lock) + unsub_scan: CALLBACK_TYPE | None = None + + +DATA_SHARED: HassKey[NinlightShared] = HassKey(DOMAIN) + + +@callback +def async_get_shared(hass: HomeAssistant) -> NinlightShared: + """Create on first use and register the download view exactly once.""" + shared = hass.data.get(DATA_SHARED) + if shared is None: + shared = NinlightShared(FirmwareCatalog(firmware_directory(hass)), OtaRegistry()) + hass.data[DATA_SHARED] = shared + hass.http.register_view(NinlightFirmwareView(hass, shared.registry)) + return shared + + +async def async_scan_catalog(hass: HomeAssistant, shared: NinlightShared) -> None: + async with shared.scan_lock: + await hass.async_add_executor_job(shared.catalog.scan) + async_dispatcher_send(hass, SIGNAL_CATALOG_UPDATED) + + +async def async_start_catalog(hass: HomeAssistant, shared: NinlightShared) -> None: + if shared.unsub_scan is not None: + return + + async def _scheduled_scan(_now: datetime) -> None: + await async_scan_catalog(hass, shared) + + shared.unsub_scan = async_track_time_interval( + hass, + _scheduled_scan, + CATALOG_SCAN_INTERVAL, + name="ninlight firmware catalog", + cancel_on_shutdown=True, + ) + await async_scan_catalog(hass, shared) + + +@callback +def async_stop_catalog(shared: NinlightShared) -> None: + if shared.unsub_scan is not None: + shared.unsub_scan() + shared.unsub_scan = None diff --git a/custom_components/ninlight/update.py b/custom_components/ninlight/update.py new file mode 100644 index 0000000..782db24 --- /dev/null +++ b/custom_components/ninlight/update.py @@ -0,0 +1,232 @@ +"""Firmware update entity for ninlight lamps, including the Wi-Fi guard (spec §7).""" + +from __future__ import annotations + +import asyncio +import time +from typing import Any + +from homeassistant.components.update import ( + UpdateDeviceClass, + UpdateEntity, + UpdateEntityFeature, +) +from homeassistant.core import HomeAssistant, callback +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import issue_registry as ir +from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from .api import NinlightError +from .const import ( + DOMAIN, + MAX_LAMP_URL_LENGTH, + OTA_POLL_INTERVAL_S, + OTA_TIMEOUT_S, + SIGNAL_CATALOG_UPDATED, +) +from .coordinator import NinlightConfigEntry, NinlightCoordinator +from .entity import NinlightEntity +from .firmware import FirmwareImage +from .firmware import is_newer as _firmware_is_newer +from .http import lamp_base_url +from .shared import DATA_SHARED +from .urls import firmware_url + +PARALLEL_UPDATES = 0 # several lamps may update at once; each lamp serializes via its lock + + +def depends_on_compiled_wifi(settings: dict[str, Any]) -> bool: + """Only an explicit 'stored' makes a lamp independent of compiled-in credentials.""" + wifi = settings.get("wifi") + return not (isinstance(wifi, dict) and wifi.get("source") == "stored") + + +def _ssid(info: dict[str, Any]) -> str: + wifi = info.get("wifi") + return str(wifi.get("ssid") or "") if isinstance(wifi, dict) else "" + + +def _error(key: str, **placeholders: str) -> HomeAssistantError: + return HomeAssistantError( + translation_domain=DOMAIN, + translation_key=key, + translation_placeholders=placeholders or None, + ) + + +async def async_setup_entry( + hass: HomeAssistant, + entry: NinlightConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + async_add_entities([NinlightUpdateEntity(entry.runtime_data)]) + + +class NinlightUpdateEntity(NinlightEntity, UpdateEntity): + """Offers the newest suitable image from the firmware folder.""" + + _attr_device_class = UpdateDeviceClass.FIRMWARE + _attr_supported_features = UpdateEntityFeature.INSTALL | UpdateEntityFeature.PROGRESS + + def __init__(self, coordinator: NinlightCoordinator) -> None: + super().__init__(coordinator, "firmware") + self._offer: FirmwareImage | None = None + self._evaluated_for: tuple[str, str, str] | None = None + self._eval_lock = asyncio.Lock() + self._attr_in_progress = False + + @property + def installed_version(self) -> str | None: + return self.coordinator.data.info.get("fw_version") + + @property + def latest_version(self) -> str | None: + return self._offer.version if self._offer is not None else self.installed_version + + def version_is_newer(self, latest_version: str, installed_version: str) -> bool: + """Use our own `git describe` comparison (spec §2): HA's AwesomeVersion misparses these. + + HA 2026.9's default comparator treats "0.3.8-47-g71a57" and "0.3.8-36-gca9bb" as equal + version cores and falls back to a string compare, which returns False here even though + commit 47 is newer than commit 36 - the STATE_ON promise this entity makes would silently + break. `firmware.is_newer` already implements the correct `git describe` ordering used to + build `self._offer` in the first place, so state and offer never disagree. + """ + return _firmware_is_newer(latest_version, installed_version) + + async def async_added_to_hass(self) -> None: + await super().async_added_to_hass() + self.async_on_remove( + async_dispatcher_connect(self.hass, SIGNAL_CATALOG_UPDATED, self._schedule_evaluation) + ) + self._schedule_evaluation() + + @callback + def _schedule_evaluation(self) -> None: + self.coordinator.config_entry.async_create_task( + self.hass, self._async_evaluate(), name="ninlight firmware evaluation" + ) + + @callback + def _handle_coordinator_update(self) -> None: + if not self._attr_in_progress and self._evaluation_key() != self._evaluated_for: + self._schedule_evaluation() + super()._handle_coordinator_update() + + def _evaluation_key(self) -> tuple[str, str, str]: + info = self.coordinator.data.info + return (str(info.get("device_type")), str(info.get("fw_version")), _ssid(info)) + + async def _async_evaluate(self) -> None: + async with self._eval_lock: + catalog = self.hass.data[DATA_SHARED].catalog + key = self._evaluation_key() + device_type, installed, ssid = key + offer: FirmwareImage | None = None + blocked: FirmwareImage | None = None + newer = catalog.newer_images(device_type, installed) + if newer: + try: + settings = await self.coordinator.async_call(lambda c: c.get_settings()) + except HomeAssistantError: + return # keep the previous offer; the next catalog scan retries + dependent = depends_on_compiled_wifi(settings) + for image in newer: # newest first + if not dependent or await self.hass.async_add_executor_job( + catalog.contains_ssid, image, ssid + ): + offer = image + break + if blocked is None: + blocked = image + self._offer = offer + self._evaluated_for = key + self._update_wifi_issue(blocked, ssid) + self.async_write_ha_state() + + @callback + def _update_wifi_issue(self, blocked: FirmwareImage | None, ssid: str) -> None: + entry = self.coordinator.config_entry + issue_id = f"wifi_guard_{entry.entry_id}" + if blocked is None: + ir.async_delete_issue(self.hass, DOMAIN, issue_id) + return + ir.async_create_issue( + self.hass, + DOMAIN, + issue_id, + is_fixable=False, + severity=ir.IssueSeverity.WARNING, + translation_key="wifi_guard", + translation_placeholders={ + "name": entry.title, + "version": blocked.version, + "device_type": blocked.device_type, + "ssid": ssid or "?", + }, + ) + + async def async_install(self, version: str | None, backup: bool, **kwargs: Any) -> None: + shared = self.hass.data[DATA_SHARED] + image = self._offer + if image is None: + raise _error("no_update") + base = lamp_base_url(self.hass, self.coordinator.config_entry.options) + if not await self.hass.async_add_executor_job(shared.catalog.verify, image): + raise _error("image_invalid") + + # Re-check right before sending: settings may have changed since the offer. + info = await self.coordinator.async_call(lambda c: c.get_info()) + settings = await self.coordinator.async_call(lambda c: c.get_settings()) + if info.get("device_type") != image.device_type: + raise _error("image_invalid") + ssid = _ssid(info) + if depends_on_compiled_wifi(settings) and not await self.hass.async_add_executor_job( + shared.catalog.contains_ssid, image, ssid + ): + raise _error("wifi_guard", ssid=ssid or "?") + + key = shared.registry.create(image.path) + url = firmware_url(base, key) + if len(url) > MAX_LAMP_URL_LENGTH: + shared.registry.revoke(key) + raise _error("lamp_url_too_long") + + self._attr_in_progress = True + self.coordinator.ota_in_progress = True + self.async_write_ha_state() + try: + await self.coordinator.async_call(lambda c: c.fw_update(url)) + await self._async_wait_for(image.version, info.get("uptime")) + finally: + shared.registry.revoke(key) + self._attr_in_progress = False + self.coordinator.ota_in_progress = False + await self.coordinator.async_refresh() + self.async_write_ha_state() + + async def _async_wait_for(self, target: str, start_uptime: Any) -> None: + """Wait for the lamp to report the target version, an error, or an old-version reboot. + + The firmware clears last_ota_error when an update starts (nl_ota.c), so any value seen + here belongs to this attempt. + """ + deadline = time.monotonic() + OTA_TIMEOUT_S + while True: + await asyncio.sleep(OTA_POLL_INTERVAL_S) + if time.monotonic() >= deadline: + raise _error("update_timeout") + try: + async with self.coordinator.lock: + info = await self.coordinator.client.get_info() + except NinlightError: + continue # flashing or rebooting + version = str(info.get("fw_version")) + if version == target: + return + if error := info.get("last_ota_error"): + raise _error("update_failed", reason=str(error)) + uptime = info.get("uptime") + if isinstance(uptime, int) and isinstance(start_uptime, int) and uptime < start_uptime: + raise _error("update_failed", reason=f"rebooted with {version}") diff --git a/tests/conftest.py b/tests/conftest.py index 639bc27..50cae5a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,6 +4,7 @@ from __future__ import annotations from collections.abc import Generator from copy import deepcopy +from pathlib import Path from typing import Any from unittest.mock import AsyncMock, MagicMock, patch @@ -101,3 +102,11 @@ def mock_setup_entry() -> Generator[AsyncMock]: """Flow tests must not set the entry up for real.""" with patch("custom_components.ninlight.async_setup_entry", return_value=True) as mock: yield mock + + +@pytest.fixture(autouse=True) +def firmware_dir(tmp_path: Path) -> Generator[Path]: + """Every Home Assistant test gets its own, initially empty firmware folder.""" + directory = tmp_path / "ninlight_firmware" + with patch("custom_components.ninlight.shared.firmware_directory", return_value=directory): + yield directory diff --git a/tests/test_diagnostics.py b/tests/test_diagnostics.py index 1e33db8..18a0d42 100644 --- a/tests/test_diagnostics.py +++ b/tests/test_diagnostics.py @@ -10,6 +10,8 @@ from pytest_homeassistant_custom_component.components.diagnostics import ( ) from pytest_homeassistant_custom_component.typing import ClientSessionGenerator +from custom_components.ninlight.shared import DATA_SHARED, async_scan_catalog +from tests_unit.images import build_image from tests_unit.samples import HOST_RGB, MAC_RGB, TOKEN @@ -22,3 +24,15 @@ async def test_diagnostics_redacts_secrets( assert secret not in text, secret assert diagnostics["info"]["device_type"] == "gu10_rgb" assert diagnostics["state"]["brightness"] == 37 + + +async def test_diagnostics_lists_matching_firmware( + hass: HomeAssistant, hass_client: ClientSessionGenerator, init_integration, firmware_dir +) -> None: + (firmware_dir / "rgb.img").write_bytes(build_image("gu10_rgb", "0.3.8-47-g71a57")) + (firmware_dir / "cw.img").write_bytes(build_image("gu10_cw", "0.3.8-47-g71a57")) + await async_scan_catalog(hass, hass.data[DATA_SHARED]) + diagnostics = await get_diagnostics_for_config_entry(hass, hass_client, init_integration) + assert diagnostics["firmware_images"] == [ + {"file": "rgb.img", "device_type": "gu10_rgb", "version": "0.3.8-47-g71a57"} + ] diff --git a/tests/test_http.py b/tests/test_http.py new file mode 100644 index 0000000..2604f82 --- /dev/null +++ b/tests/test_http.py @@ -0,0 +1,84 @@ +"""One-time firmware URLs and the address lamps use to reach Home Assistant.""" + +from __future__ import annotations + +from pathlib import Path + +from homeassistant.core import HomeAssistant +from homeassistant.core_config import async_process_ha_core_config +from homeassistant.exceptions import HomeAssistantError +import pytest +from pytest_homeassistant_custom_component.typing import ClientSessionGenerator + +from custom_components.ninlight.const import CONF_LAMP_BASE_URL +from custom_components.ninlight.http import OtaRegistry, lamp_base_url +from custom_components.ninlight.shared import DATA_SHARED +from tests_unit.images import build_image + + +def test_registry_keys_are_unique_and_43_characters() -> None: + registry = OtaRegistry() + keys = {registry.create(Path("/tmp/a.img")) for _ in range(50)} + assert len(keys) == 50 + assert all(len(key) == 43 for key in keys) + assert len(registry) == 50 + + +async def test_view_serves_only_registered_keys( + hass: HomeAssistant, + hass_client_no_auth: ClientSessionGenerator, + init_integration, + firmware_dir: Path, +) -> None: + path = firmware_dir / "a.img" + path.write_bytes(build_image()) + registry = hass.data[DATA_SHARED].registry + other = firmware_dir / "b.img" + other.write_bytes(build_image(version="0.3.9-rc1")) + key = registry.create(path) + client = await hass_client_no_auth() + + response = await client.get(f"/api/ninlight/firmware/{key}") + assert response.status == 200 + assert await response.read() == path.read_bytes() + + assert (await client.get("/api/ninlight/firmware/" + "x" * 43)).status == 404 + registry.revoke(key) + assert (await client.get(f"/api/ninlight/firmware/{key}")).status == 404 + + +async def test_view_returns_404_when_file_vanished( + hass: HomeAssistant, + hass_client_no_auth: ClientSessionGenerator, + init_integration, + firmware_dir: Path, +) -> None: + key = hass.data[DATA_SHARED].registry.create(firmware_dir / "gone.img") + client = await hass_client_no_auth() + assert (await client.get(f"/api/ninlight/firmware/{key}")).status == 404 + + +async def test_base_url_prefers_option(hass: HomeAssistant) -> None: + await async_process_ha_core_config(hass, {"internal_url": "http://192.168.1.60:8123"}) + assert ( + lamp_base_url(hass, {CONF_LAMP_BASE_URL: "http://192.168.1.50:8123"}) + == "http://192.168.1.50:8123" + ) + + +async def test_base_url_falls_back_to_internal_http_url(hass: HomeAssistant) -> None: + await async_process_ha_core_config(hass, {"internal_url": "http://192.168.1.60:8123"}) + assert lamp_base_url(hass, {}) == "http://192.168.1.60:8123" + + +async def test_base_url_rejects_https(hass: HomeAssistant) -> None: + await async_process_ha_core_config(hass, {"internal_url": "https://ha.local:8123"}) + with pytest.raises(HomeAssistantError) as raised: + lamp_base_url(hass, {}) + assert raised.value.translation_key == "no_http_url" + + +async def test_base_url_without_any_url(hass: HomeAssistant) -> None: + with pytest.raises(HomeAssistantError) as raised: + lamp_base_url(hass, {}) + assert raised.value.translation_key == "no_http_url" diff --git a/tests/test_update.py b/tests/test_update.py new file mode 100644 index 0000000..e952030 --- /dev/null +++ b/tests/test_update.py @@ -0,0 +1,339 @@ +"""Firmware update entity: offer, Wi-Fi guard, install flow.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock +from urllib.parse import urlsplit + +from homeassistant.components.update import ( + ATTR_IN_PROGRESS, + ATTR_INSTALLED_VERSION, + ATTR_LATEST_VERSION, + SERVICE_INSTALL, +) +from homeassistant.components.update import ( + DOMAIN as UPDATE_DOMAIN, +) +from homeassistant.const import ATTR_ENTITY_ID, CONF_HOST, CONF_TOKEN, STATE_OFF, STATE_ON +from homeassistant.core import HomeAssistant +from homeassistant.core_config import async_process_ha_core_config +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import issue_registry as ir +import pytest +from pytest_homeassistant_custom_component.common import MockConfigEntry +from pytest_homeassistant_custom_component.typing import ClientSessionGenerator + +from custom_components.ninlight import update as update_module +from custom_components.ninlight.api import NinlightBusyError, NinlightConnectionError +from custom_components.ninlight.const import CONF_LAMP_BASE_URL, DOMAIN +from custom_components.ninlight.shared import DATA_SHARED, async_scan_catalog +from custom_components.ninlight.update import depends_on_compiled_wifi +from tests.helpers import entity_id_for +from tests_unit.images import CONTENT_WITH_SSID, CONTENT_WITHOUT_SSID, build_image +from tests_unit.samples import HOST_RGB, MAC_RGB, SETTINGS_COMPILED, SETTINGS_STORED, TOKEN + +BASE = "http://192.168.1.50:8123" +INSTALLED = "0.3.8-36-gca9bb" +NEWER = "0.3.8-47-g71a57" +NEWEST = "0.3.9-rc1" + + +@pytest.fixture +def mock_config_entry() -> MockConfigEntry: + return MockConfigEntry( + domain=DOMAIN, + title="Klyqa GU10 RGB CW/WW 06b8", + unique_id=MAC_RGB, + data={CONF_HOST: HOST_RGB, CONF_TOKEN: TOKEN}, + options={CONF_LAMP_BASE_URL: BASE}, + ) + + +@pytest.fixture(autouse=True) +def fast_polling(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(update_module, "OTA_POLL_INTERVAL_S", 0) + + +def put(directory: Path, name: str, **kwargs: Any) -> Path: + path = directory / name + path.write_bytes(build_image(**kwargs)) + return path + + +async def rescan(hass: HomeAssistant) -> None: + await async_scan_catalog(hass, hass.data[DATA_SHARED]) + await hass.async_block_till_done() + + +def wifi_issue(hass: HomeAssistant, entry: MockConfigEntry) -> ir.IssueEntry | None: + return ir.async_get(hass).async_get_issue(DOMAIN, f"wifi_guard_{entry.entry_id}") + + +async def install(hass: HomeAssistant) -> None: + await hass.services.async_call( + UPDATE_DOMAIN, + SERVICE_INSTALL, + {ATTR_ENTITY_ID: entity_id_for(hass, "update", "firmware")}, + blocking=True, + ) + + +def test_only_stored_wifi_is_independent() -> None: + assert not depends_on_compiled_wifi(SETTINGS_STORED) + assert depends_on_compiled_wifi(SETTINGS_COMPILED) + assert depends_on_compiled_wifi({"wifi": {"ssid": ""}}) + assert depends_on_compiled_wifi({"wifi": {"source": "flash"}}) + assert depends_on_compiled_wifi({}) + + +async def test_no_images_no_update(hass: HomeAssistant, init_integration) -> None: + state = hass.states.get(entity_id_for(hass, "update", "firmware")) + assert state.state == STATE_OFF + assert state.attributes[ATTR_INSTALLED_VERSION] == INSTALLED + assert state.attributes[ATTR_LATEST_VERSION] == INSTALLED + + +async def test_newer_image_with_credentials_is_offered( + hass: HomeAssistant, init_integration, firmware_dir: Path +) -> None: + put(firmware_dir, "a.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + state = hass.states.get(entity_id_for(hass, "update", "firmware")) + assert state.state == STATE_ON + assert state.attributes[ATTR_LATEST_VERSION] == NEWER + assert wifi_issue(hass, init_integration) is None + + +@pytest.mark.parametrize( + "image", + [ + { + "device_type": "gu10_cw", + "version": "0.4.0", + }, # other type that only differs in the suffix + {"device_type": "gu10_rgb", "version": "0.3.8-36-gffff0"}, # same version, other hash + {"device_type": "gu10_rgb", "version": "0.3.8"}, # older + ], +) +async def test_unsuitable_images_are_not_offered( + hass: HomeAssistant, init_integration, firmware_dir: Path, image: dict[str, str] +) -> None: + put(firmware_dir, "a.img", content=CONTENT_WITH_SSID, **image) + await rescan(hass) + assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_OFF + + +async def test_wifi_guard_blocks_image_without_credentials( + hass: HomeAssistant, init_integration, firmware_dir: Path +) -> None: + nocreds = put(firmware_dir, "nocreds.img", version=NEWEST, content=CONTENT_WITHOUT_SSID) + await rescan(hass) + entity_id = entity_id_for(hass, "update", "firmware") + assert hass.states.get(entity_id).state == STATE_OFF + issue = wifi_issue(hass, init_integration) + assert issue is not None + assert issue.translation_placeholders == { + "name": "Klyqa GU10 RGB CW/WW 06b8", + "version": NEWEST, + "device_type": "gu10_rgb", + "ssid": "ninIOT", + } + + put(firmware_dir, "creds.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + assert hass.states.get(entity_id).attributes[ATTR_LATEST_VERSION] == NEWER + assert wifi_issue(hass, init_integration) is not None # the newer, blocked one is still there + + nocreds.unlink() + await rescan(hass) + assert hass.states.get(entity_id).attributes[ATTR_LATEST_VERSION] == NEWER + assert wifi_issue(hass, init_integration) is None + + +async def test_lamp_with_stored_wifi_gets_image_without_credentials( + hass: HomeAssistant, init_integration, firmware_dir: Path, lamp_settings: dict[str, Any] +) -> None: + lamp_settings.clear() + lamp_settings.update(SETTINGS_STORED) + put(firmware_dir, "nocreds.img", version=NEWEST, content=CONTENT_WITHOUT_SSID) + await rescan(hass) + assert ( + hass.states.get(entity_id_for(hass, "update", "firmware")).attributes[ATTR_LATEST_VERSION] + == NEWEST + ) + assert wifi_issue(hass, init_integration) is None + + +async def test_lamp_without_ssid_gets_nothing( + hass: HomeAssistant, init_integration, firmware_dir: Path, lamp_info: dict[str, Any] +) -> None: + lamp_info["wifi"]["ssid"] = "" + await init_integration.runtime_data.async_refresh() + put(firmware_dir, "creds.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_OFF + + +async def test_unreadable_settings_keep_previous_offer( + hass: HomeAssistant, init_integration, firmware_dir: Path, mock_client: MagicMock +) -> None: + mock_client.get_settings.side_effect = NinlightConnectionError("down") + put(firmware_dir, "creds.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_OFF + + +async def test_install_serves_image_once_and_waits_for_new_version( + hass: HomeAssistant, + hass_client_no_auth: ClientSessionGenerator, + init_integration, + mock_client: MagicMock, + lamp_info: dict[str, Any], + firmware_dir: Path, +) -> None: + path = put(firmware_dir, "a.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + client = await hass_client_no_auth() + served: dict[str, Any] = {} + + async def lamp_downloads(url: str) -> None: + served["url"] = url + response = await client.get(urlsplit(url).path) + served["status"] = response.status + served["body"] = await response.read() + served["ota_flag"] = init_integration.runtime_data.ota_in_progress + lamp_info["fw_version"] = NEWER + lamp_info["uptime"] = 4 + + mock_client.fw_update.side_effect = lamp_downloads + await install(hass) + await hass.async_block_till_done() + + assert served["url"].startswith(f"{BASE}/api/ninlight/firmware/") + assert (served["status"], served["body"]) == (200, path.read_bytes()) + assert served["ota_flag"] is True + assert (await client.get(urlsplit(served["url"]).path)).status == 404 + assert len(hass.data[DATA_SHARED].registry) == 0 + state = hass.states.get(entity_id_for(hass, "update", "firmware")) + assert state.attributes[ATTR_INSTALLED_VERSION] == NEWER + assert state.attributes[ATTR_IN_PROGRESS] is False + assert state.state == STATE_OFF + assert init_integration.runtime_data.ota_in_progress is False + + +@pytest.mark.parametrize( + ("lamp_reaction", "key", "reason_part"), + [ + ({"last_ota_error": "hash"}, "update_failed", "hash"), + ({"uptime": 3}, "update_failed", INSTALLED), # rebooted, still the old firmware + ], +) +async def test_install_reports_lamp_failures( + hass: HomeAssistant, + init_integration, + mock_client: MagicMock, + lamp_info: dict[str, Any], + firmware_dir: Path, + lamp_reaction: dict[str, Any], + key: str, + reason_part: str, +) -> None: + put(firmware_dir, "a.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + + async def lamp_fails(url: str) -> None: + lamp_info.update(lamp_reaction) + + mock_client.fw_update.side_effect = lamp_fails + with pytest.raises(HomeAssistantError) as raised: + await install(hass) + assert raised.value.translation_key == key + assert reason_part in raised.value.translation_placeholders["reason"] + assert len(hass.data[DATA_SHARED].registry) == 0 + assert init_integration.runtime_data.ota_in_progress is False + + +async def test_install_busy_lamp( + hass: HomeAssistant, init_integration, mock_client: MagicMock, firmware_dir: Path +) -> None: + put(firmware_dir, "a.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + mock_client.fw_update.side_effect = NinlightBusyError(409, "an update is already running") + with pytest.raises(HomeAssistantError) as raised: + await install(hass) + assert raised.value.translation_key == "update_busy" + assert len(hass.data[DATA_SHARED].registry) == 0 + + +async def test_install_times_out( + hass: HomeAssistant, init_integration, firmware_dir: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(update_module, "OTA_TIMEOUT_S", 0) + put(firmware_dir, "a.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + with pytest.raises(HomeAssistantError) as raised: + await install(hass) + assert raised.value.translation_key == "update_timeout" + + +async def test_install_needs_http_address( + hass: HomeAssistant, mock_client: MagicMock, firmware_dir: Path +) -> None: + await async_process_ha_core_config(hass, {"internal_url": "https://ha.local:8123"}) + entry = MockConfigEntry( + domain=DOMAIN, unique_id=MAC_RGB, data={CONF_HOST: HOST_RGB, CONF_TOKEN: TOKEN} + ) + entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(entry.entry_id) + put(firmware_dir, "a.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + with pytest.raises(HomeAssistantError) as raised: + await install(hass) + assert raised.value.translation_key == "no_http_url" + mock_client.fw_update.assert_not_awaited() + + +async def test_install_rechecks_wifi_right_before_sending( + hass: HomeAssistant, + init_integration, + mock_client: MagicMock, + firmware_dir: Path, + lamp_settings: dict[str, Any], +) -> None: + lamp_settings.clear() + lamp_settings.update(SETTINGS_STORED) + put(firmware_dir, "nocreds.img", version=NEWEST, content=CONTENT_WITHOUT_SSID) + await rescan(hass) + assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_ON + + lamp_settings["wifi"]["source"] = "compiled" # the lamp forgot its stored Wi-Fi meanwhile + with pytest.raises(HomeAssistantError) as raised: + await install(hass) + assert raised.value.translation_key == "wifi_guard" + assert raised.value.translation_placeholders == {"ssid": "ninIOT"} + mock_client.fw_update.assert_not_awaited() + + +async def test_install_rejects_changed_file( + hass: HomeAssistant, init_integration, mock_client: MagicMock, firmware_dir: Path +) -> None: + path = put(firmware_dir, "a.img", version=NEWER, content=CONTENT_WITH_SSID) + await rescan(hass) + path.write_bytes(build_image(version=NEWEST, content=CONTENT_WITH_SSID)) + with pytest.raises(HomeAssistantError) as raised: + await install(hass) + assert raised.value.translation_key == "image_invalid" + mock_client.fw_update.assert_not_awaited() + + +async def test_removing_the_entry_clears_the_wifi_issue( + hass: HomeAssistant, init_integration, firmware_dir: Path +) -> None: + put(firmware_dir, "nocreds.img", version=NEWEST, content=CONTENT_WITHOUT_SSID) + await rescan(hass) + assert wifi_issue(hass, init_integration) is not None + assert await hass.config_entries.async_remove(init_integration.entry_id) + await hass.async_block_till_done() + assert wifi_issue(hass, init_integration) is None