feat: update-Entitaet fuer die Firmware und das Testgeruest

Aufgebaut wie ha-ninlight: die Abbilder liegen unter ninstrip_firmware im
Konfigurationsverzeichnis, Home Assistant liefert sie selbst unter einer
einmaligen Adresse aus, und zwar genau die Bytes, die vorher geprueft
wurden. Dazu firmware.py (Anhaenger, Versionsreihenfolge, WLAN-Pruefung,
Katalog), http.py, urls.py und shared.py.

Drei Unterschiede zu ninlight, alle aus der Firmware heraus:

* Verglichen wird device_type - die Platine -, nicht product. Letzteres ist
  bei allen Platinen 'ninstrip'; die Abbilder unterscheiden sich aber hart,
  die K1Y invertiert das Datensignal und die KL-RGBC3 nicht. Firmware ohne
  dieses Feld bekommt gar kein Angebot.
* Die WLAN-Pruefung sucht die SSID unverpackt: ein ESP32-Abbild ist nicht
  gzip-verpackt wie das des W600. An einem mit sdkconfig.local gebauten
  Abbild gegengeprueft - die SSID steht genau einmal NUL-gerahmt drin.
* Der Fehlerschluessel heisst device_url_too_long statt lamp_url_too_long.

Dazu das Testgeruest, das es hier bisher gar nicht gab: 204 Tests in
tests_unit/ (ohne HA-Instanz) und tests/ (mit), Abdeckung 97 %, dazu
scripts/check.sh als Ersatz fuer die fehlende CI. Die Nutzdaten in
tests_unit/samples.py sind an der Stehlampe abgefragt.

Beim Gegenpruefen per Mutation fiel auf, dass die Platinenpruefung
unmittelbar vor dem Senden ungeprueft war - der Test dazu ist jetzt da und
faellt, wenn man die Pruefung entfernt.
This commit is contained in:
Michael Sauer
2026-09-22 23:10:07 +02:00
parent d0a9487805
commit 2c77b2d325
44 changed files with 3600 additions and 43 deletions
+2
View File
@@ -2,3 +2,5 @@ __pycache__/
*.pyc
.venv/
.DS_Store
.coverage
.pytest_cache/
+59 -2
View File
@@ -54,6 +54,7 @@ Ohne mDNS geht es auch von Hand über Adresse und Token.
| `number` | sACN-Universe und -Startslot |
| `select` | Verhalten nach Stromende (`release` / `hold`) |
| `binary_sensor` | sACN aktiv, Multicast beigetreten (letzterer standardmäßig aus) |
| `update` | Firmware-Aktualisierung aus dem Abbild-Ordner |
Die Diagnose-Sensoren sind standardmäßig abgeschaltet; die
Leistungsbegrenzung ist an, weil sie verrät, ob die Strombegrenzung gerade
@@ -66,8 +67,43 @@ eine Farbe, die stimmen würde. Die Segmente stellt man über Szenen.
## Noch nicht drin
- Einzelne Segmente als eigene Entitäten
- Firmware-Aktualisierung über eine `update`-Entität (die Firmware kann OTA
mit Rückfallebene, die Integration nutzt es noch nicht)
## Firmware-Aktualisierung
Aufbau wie ha-ninlight. Die Abbilder liegen im Ordner **`ninstrip_firmware`**
unterhalb des Konfigurationsverzeichnisses; `tools/release_images.sh` der
Firmware baut sie. Alle fünf Minuten wird der Ordner durchgesehen.
Home Assistant liefert das Abbild selbst aus, unter einer **einmaligen
Adresse**, die nur gilt, solange die Aktualisierung läuft. Das Gerät kann sich
bei Home Assistant nicht anmelden — der unratbare Schlüssel ist der Schutz.
Ausgeliefert werden genau die Bytes, die vorher geprüft wurden; wird die Datei
mittendrin überschrieben, ändert das nichts mehr an dem, was ankommt.
Drei Dinge entscheiden, ob ein Abbild angeboten wird:
- **Die Platine muss passen.** Verglichen wird `device_type` aus
`system/info` mit dem Typ im NINL-Anhänger des Abbildes
(`ninstrip-rgbc3`, `ninstrip-k1y`). `product` ist bei allen Platinen
`ninstrip` und taugt dafür nicht: ein K1Y-Abbild invertiert das
Datensignal, ein KL-RGBC3-Abbild nicht — die falsche Wahl lässt den Strip
dunkel. Firmware vor `v0.1.0` meldet kein `device_type` und bekommt
deshalb gar kein Angebot.
- **Die Version muss vergleichbar und neuer sein.** Gerechnet wird nach
`git describe`, nicht mit der Versionslogik von Home Assistant: die hält
`0.1.0-12-g71a57` und `0.1.0-47-g71a57` für denselben Versionskern.
Firmware vor dem Tag `v0.1.0` meldete nur einen Hash — daran lässt sich
keine Reihenfolge ablesen, also gibt es auch kein Angebot.
- **Die WLAN-Zugangsdaten müssen drin sein**, solange das Gerät sein WLAN
nur aus der Firmware kennt (`wifi.source` = `compiled`). Sonst verlöre es
nach dem Update seine Netzverbindung. Fehlen sie, erscheint ein Hinweis in
den Reparaturen statt eines Angebots. Wer das Gerät über das Onboarding ins
WLAN gebracht hat (`stored`), bekommt das Update ohne diesen Vorbehalt.
Die Geräte laden nur über **schlichtes http**. Erreichen sie die interne
Adresse von Home Assistant nicht so, trägt man in den Einstellungen der
Integration eine eigene ein (`http://…`, höchstens so lang, dass die ganze
Firmware-Adresse unter 255 Zeichen bleibt — mehr nimmt das Gerät nicht an).
## sACN / E1.31
@@ -97,3 +133,24 @@ geht dann trotzdem.
Die Integration wird in der gemeinsamen Dev-Instanz `ha-dev` per Mount
eingebunden, der Quellcode bleibt in diesem Repo.
### Tests
```
uv venv -p /opt/homebrew/opt/python@3.14/bin/python3.14 .venv
uv pip install --python .venv/bin/python -r requirements_dev.txt
./scripts/check.sh
```
`scripts/check.sh` ersetzt die CI — das Repo liegt in einem privaten Gitea
ohne Actions — und läuft über ruff, die Tests samt Abdeckung (mindestens 95 %)
und die Übersetzungen.
Die Tests liegen in zwei Ordnern: **`tests_unit/`** kommt ohne
Home-Assistant-Instanz aus (API-Client gegen einen echten lokalen Server,
Anhänger, Versionsreihenfolge, Adressen), **`tests/`** fährt eine
Home-Assistant-Instanz hoch. Die Nutzdaten in `tests_unit/samples.py` sind an
echten Geräten abgefragt, keine erfundenen.
Die Python-Version ist nicht beliebig: Home Assistant 2026.9 verlangt
mindestens 3.14.2, und auf PyPI liegt diese HA-Version nur für uv bereit.
+29 -2
View File
@@ -4,11 +4,17 @@ from __future__ import annotations
from homeassistant.const import CONF_HOST, CONF_TOKEN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .api import NinstripClient
from .const import DEFAULT_PORT
from .const import DEFAULT_PORT, DOMAIN
from .coordinator import NinstripConfigEntry, NinstripCoordinator
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.LIGHT,
@@ -18,9 +24,16 @@ PLATFORMS: list[Platform] = [
Platform.SWITCH,
Platform.NUMBER,
Platform.BINARY_SENSOR,
Platform.UPDATE,
]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Die Ausgabe fuer die Abbilder haengt an http und wird einmal angemeldet."""
async_get_shared(hass)
return True
async def async_setup_entry(hass: HomeAssistant, entry: NinstripConfigEntry) -> bool:
"""Einen Strip einrichten."""
client = NinstripClient(
@@ -33,13 +46,27 @@ async def async_setup_entry(hass: HomeAssistant, entry: NinstripConfigEntry) ->
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await async_start_catalog(hass, async_get_shared(hass))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(_async_options_updated))
return True
async def async_unload_entry(hass: HomeAssistant, entry: NinstripConfigEntry) -> bool:
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
entladen = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
# Der Katalog gehoert allen Eintraegen gemeinsam; er laeuft weiter,
# solange noch einer da ist.
if entladen and not [
anderer
for anderer in hass.config_entries.async_loaded_entries(DOMAIN)
if anderer.entry_id != entry.entry_id
]:
async_stop_catalog(async_get_shared(hass))
return entladen
async def async_remove_entry(hass: HomeAssistant, entry: NinstripConfigEntry) -> None:
ir.async_delete_issue(hass, DOMAIN, f"wifi_guard_{entry.entry_id}")
async def _async_options_updated(hass: HomeAssistant, entry: NinstripConfigEntry) -> None:
+2 -6
View File
@@ -106,9 +106,7 @@ class NinstripClient:
async def save_preset(self, index: int, name: str | None = None) -> Any:
"""Sichert die laufende Ansicht auf einem Szenenplatz."""
return await self._request(
"PUT", f"device/presets/{index}", {"name": name} if name else {}
)
return await self._request("PUT", f"device/presets/{index}", {"name": name} if name else {})
async def apply_preset(self, index: int) -> Any:
return await self._request("PUT", f"device/presets/{index}/apply")
@@ -121,9 +119,7 @@ class NinstripClient:
await self._request("PUT", "system/command", {"command": {"type": "reboot"}})
async def fw_update(self, url: str) -> None:
await self._request(
"PUT", "system/command", {"command": {"type": "fw_update", "url": url}}
)
await self._request("PUT", "system/command", {"command": {"type": "fw_update", "url": url}})
# --- Unterbau ------------------------------------------------------
+1 -3
View File
@@ -19,9 +19,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
coordinator = entry.runtime_data
async_add_entities(
[NinstripSacnActive(coordinator), NinstripSacnMulticast(coordinator)]
)
async_add_entities([NinstripSacnActive(coordinator), NinstripSacnMulticast(coordinator)])
class _NinstripSacnInfo(NinstripEntity, BinarySensorEntity):
+30 -23
View File
@@ -21,6 +21,7 @@ import voluptuous as vol
from .api import NinstripAuthError, NinstripClient, NinstripError, NinstripResponseError
from .const import (
CONF_DEVICE_BASE_URL,
CONF_SCAN_INTERVAL,
DEFAULT_SCAN_INTERVAL,
DEFAULT_TOKEN,
@@ -29,6 +30,7 @@ from .const import (
MIN_SCAN_INTERVAL,
)
from .entity import device_name
from .urls import validate_device_base_url
_LOGGER = logging.getLogger(__name__)
@@ -78,18 +80,12 @@ class NinstripConfigFlow(ConfigFlow, domain=DOMAIN):
title=device_name(info), data={CONF_HOST: host, CONF_TOKEN: token}
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
errors: dict[str, str] = {}
if user_input is not None:
info, errors = await self._async_probe(
user_input[CONF_HOST], user_input[CONF_TOKEN]
)
info, errors = await self._async_probe(user_input[CONF_HOST], user_input[CONF_TOKEN])
if info is not None:
return await self._async_finish(
user_input[CONF_HOST], user_input[CONF_TOKEN], info
)
return await self._async_finish(user_input[CONF_HOST], user_input[CONF_TOKEN], info)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
@@ -101,9 +97,7 @@ class NinstripConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_zeroconf(
self, discovery_info: ZeroconfServiceInfo
) -> ConfigFlowResult:
async def async_step_zeroconf(self, discovery_info: ZeroconfServiceInfo) -> ConfigFlowResult:
"""Gefunden ueber _ninstrip._tcp."""
self._host = discovery_info.host
props = discovery_info.properties or {}
@@ -131,9 +125,7 @@ class NinstripConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> ConfigFlowResult:
self._host = entry_data[CONF_HOST]
return await self.async_step_reauth_confirm()
@@ -158,21 +150,36 @@ class NinstripConfigFlow(ConfigFlow, domain=DOMAIN):
class NinstripOptionsFlow(OptionsFlowWithReload):
"""Bisher nur die Abfragehaeufigkeit."""
"""Abfragehaeufigkeit und die Adresse, unter der Geraete Home Assistant erreichen."""
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
async def async_step_init(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
errors: dict[str, str] = {}
if user_input is not None:
return self.async_create_entry(data=user_input)
current = self.config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
base = str(user_input.get(CONF_DEVICE_BASE_URL) or "").strip()
fehler = validate_device_base_url(base) if base else None
if fehler:
errors[CONF_DEVICE_BASE_URL] = fehler
else:
daten: dict[str, Any] = {CONF_SCAN_INTERVAL: user_input[CONF_SCAN_INTERVAL]}
if base:
daten[CONF_DEVICE_BASE_URL] = base.rstrip("/")
return self.async_create_entry(data=daten)
current = self.config_entry.options
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Required(CONF_SCAN_INTERVAL, default=current): vol.All(
vol.Required(
CONF_SCAN_INTERVAL,
default=current.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL),
): vol.All(
vol.Coerce(int), vol.Range(min=MIN_SCAN_INTERVAL, max=MAX_SCAN_INTERVAL)
)
),
vol.Optional(
CONF_DEVICE_BASE_URL,
description={"suggested_value": current.get(CONF_DEVICE_BASE_URL, "")},
): str,
}
),
errors=errors,
)
+23
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
from datetime import timedelta
from typing import Final
DOMAIN: Final = "ninstrip"
@@ -12,6 +13,9 @@ DOMAIN: Final = "ninstrip"
DEFAULT_TOKEN: Final = "aabbccddeeff0011223344"
CONF_SCAN_INTERVAL: Final = "scan_interval"
# Nur noetig, wenn die Geraete die interne Adresse von Home Assistant nicht
# ueber schlichtes http erreichen.
CONF_DEVICE_BASE_URL: Final = "device_base_url"
DEFAULT_PORT: Final = 3333
DEFAULT_SCAN_INTERVAL: Final = 30
@@ -28,3 +32,22 @@ MAX_SEGMENTS: Final = 16
MAX_PIXELS: Final = 512
# Acht Szenenplaetze (nl_presets.h).
MAX_PRESETS: Final = 8
# --- Firmware-Aktualisierung -----------------------------------------
# Ordner unterhalb des HA-Konfigurationsverzeichnisses, in dem die Abbilder
# liegen (tools/release_images.sh der Firmware legt sie dort ab).
FIRMWARE_DIR_NAME: Final = "ninstrip_firmware"
CATALOG_SCAN_INTERVAL: Final = timedelta(minutes=5)
SIGNAL_CATALOG_UPDATED: Final = f"{DOMAIN}_catalog_updated"
FIRMWARE_URL_PREFIX: Final = "/api/ninstrip/firmware/"
# len(secrets.token_urlsafe(32))
FIRMWARE_KEY_LENGTH: Final = 43
# Laengere fw_update-Adressen weist das Geraet ab (nl_api_system.c: url[256]).
MAX_DEVICE_URL_LENGTH: Final = 255
OTA_POLL_INTERVAL_S: Final = 5.0
# Der ESP32 laedt rund 1 MB und startet danach neu; ein totes Gegenueber kann
# aber minutenlang im TCP haengen. Grosszuegig wie bei ninlight.
OTA_TIMEOUT_S: Final = 12 * 60.0
+222
View File
@@ -0,0 +1,222 @@
"""OTA-Abbilder fuer ninstrip: Anhaenger, Versionsreihenfolge, WLAN-Pruefung, Katalog.
Bewusst frei von Home-Assistant-Importen. Blockierende Funktionen sind
benannt; die Aufrufer schicken sie in einen Executor.
Der Anhaenger ist derselbe wie bei ninlight (512 Byte, Magie "NINL") - genau
deshalb kann die Migrationsstation von dort auch ninstrip-Abbilder ausliefern.
Ein Unterschied bleibt: ninlight laeuft auf dem W600 und traegt seine
Nutzdaten gzip-verpackt, der ESP32 nicht. Die WLAN-Pruefung sucht hier also
im Klartext (siehe `contains_ssid`).
"""
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import logging
from pathlib import Path
import re
import threading
_LOGGER = logging.getLogger(__name__)
TRAILER_SIZE = 512
TRAILER_MAGIC = b"NINL"
TRAILER_FORMAT = 1
# tools/mkimage.py: device[16] und fw_version[16], jeweils NUL-abgeschlossen.
VERSION_FIELD_MAX = 15
_VERSION_RE = re.compile(
r"^(?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)"
r"(?:-rc(?P<rc>\d+))?"
r"(?:-(?P<commits>\d+)(?P<hash>-g[0-9a-f]*)?)?"
r"(?:-d(?:i(?:r(?:t(?:y)?)?)?)?)?$"
)
class InvalidImageError(ValueError):
"""Die Datei ist kein gueltiges ninstrip-OTA-Abbild."""
@dataclass(frozen=True, slots=True)
class FirmwareVersion:
"""Eine gelesene `git describe`-Version. Der Hash zaehlt bewusst nicht mit."""
base: tuple[int, int, int]
rc: int | None
commits: int
@property
def sort_key(self) -> tuple[int, int, int, int, int, int]:
# Ein Release Candidate kommt vor der Freigabe derselben Grundversion.
rc_rank, rc_number = (0, self.rc) if self.rc is not None else (1, 0)
return (*self.base, rc_rank, rc_number, self.commits)
def parse_version(text: str) -> FirmwareVersion | None:
match = _VERSION_RE.match(text)
if match is None:
return None
if len(text) == VERSION_FIELD_MAX and text[-1].isdigit() and match["hash"] is None:
# Die Kuerzung kann mitten in der Commit- oder rc-Zahl gelandet sein:
# dann ist die Version nicht vergleichbar.
return None
return FirmwareVersion(
(int(match["major"]), int(match["minor"]), int(match["patch"])),
int(match["rc"]) if match["rc"] is not None else None,
int(match["commits"]) if match["commits"] is not None else 0,
)
def is_newer(candidate: str, installed: str) -> bool:
new, old = parse_version(candidate), parse_version(installed)
if new is None or old is None:
return False
return new.sort_key > old.sort_key
def parse_trailer(data: bytes) -> tuple[str, str]:
"""(device_type, version) eines gueltigen Abbildes, sonst InvalidImageError."""
if len(data) <= TRAILER_SIZE:
raise InvalidImageError("zu kurz fuer einen Anhaenger")
payload, trailer = data[:-TRAILER_SIZE], data[-TRAILER_SIZE:]
if trailer[0:4] != TRAILER_MAGIC:
raise InvalidImageError("kein NINL-Anhaenger")
if trailer[4] != TRAILER_FORMAT:
raise InvalidImageError(f"unbekanntes Anhaengerformat {trailer[4]}")
if trailer[37:69] != hashlib.sha256(payload).digest():
raise InvalidImageError("sha256 passt nicht")
device_type = _text_field(trailer[5:21])
version = _text_field(trailer[21:37])
if not device_type or not version:
raise InvalidImageError("Geraetetyp oder Version leer")
return device_type, version
def _text_field(raw: bytes) -> str:
return raw.split(b"\x00", 1)[0].decode("ascii", errors="replace")
def contains_ssid(data: bytes, ssid: str) -> bool:
"""Wahr, wenn das Abbild `ssid` als NUL-gerahmte C-Zeichenkette traegt.
Anders als bei ninlight wird nichts ausgepackt: ein ESP32-Abbild liegt
unverpackt vor, die einkompilierte SSID steht als Zeichenkette drin. Am
Geraet gegengeprueft - in einem mit sdkconfig.local gebauten Abbild kommt
sie genau einmal NUL-gerahmt vor.
"""
if not ssid or len(data) <= TRAILER_SIZE:
return False
return b"\x00" + ssid.encode() + b"\x00" in data[:-TRAILER_SIZE]
@dataclass(frozen=True, slots=True)
class FirmwareImage:
"""Ein gueltiges Abbild im Firmware-Ordner."""
path: Path
device_type: str
version: str
size: int
mtime_ns: int
def _sort_key(image: FirmwareImage) -> tuple[int, ...]:
version = parse_version(image.version)
return version.sort_key if version is not None else ()
class FirmwareCatalog:
"""Alle gueltigen Abbilder eines Ordners. Nicht threadsicher: Aufrufer ordnen die Laeufe."""
def __init__(self, directory: Path) -> None:
self.directory = directory
self.images: tuple[FirmwareImage, ...] = ()
self._parsed: dict[tuple[str, int, int], FirmwareImage | None] = {}
self._ssid_hits: dict[tuple[str, int, int, str], bool] = {}
# scan() laeuft in einem Executor-Thread, waehrend contains_ssid() in
# einem anderen laufen kann (etwa die WLAN-Gegenprobe einer laufenden
# Aktualisierung gegen einen turnusmaessigen Durchlauf): beide lesen
# und aendern _ssid_hits.
self._ssid_hits_lock = threading.Lock()
def scan(self) -> tuple[FirmwareImage, ...]:
"""Blockierend. Ungueltige Dateien werden je (Pfad, mtime, Groesse) einmal gemeldet."""
self.directory.mkdir(parents=True, exist_ok=True)
found: list[FirmwareImage] = []
seen: set[tuple[str, int, int]] = set()
for path in sorted(self.directory.glob("*.img")):
try:
if not path.is_file():
continue
stat = path.stat()
except OSError:
continue
key = (str(path), stat.st_mtime_ns, stat.st_size)
seen.add(key)
if key not in self._parsed:
try:
device_type, version = parse_trailer(path.read_bytes())
except (OSError, InvalidImageError) as err:
_LOGGER.warning("Uebergehe Firmware-Datei %s: %s", path.name, err)
self._parsed[key] = None
else:
self._parsed[key] = FirmwareImage(
path, device_type, version, stat.st_size, stat.st_mtime_ns
)
image = self._parsed[key]
if image is not None:
found.append(image)
self._parsed = {k: v for k, v in self._parsed.items() if k in seen}
with self._ssid_hits_lock:
self._ssid_hits = {k: v for k, v in self._ssid_hits.items() if k[:3] in seen}
self.images = tuple(found)
return self.images
def newer_images(self, device_type: str, installed: str) -> list[FirmwareImage]:
"""Abbilder genau dieser Platine, die neuer sind als `installed` - neuestes zuerst."""
candidates = [
image
for image in self.images
if image.device_type == device_type and is_newer(image.version, installed)
]
return sorted(candidates, key=_sort_key, reverse=True)
def verify_and_read(self, image: FirmwareImage) -> bytes | None:
"""Blockierend. Liefert genau die geprueften Bytes, sonst None.
Wer mit dem Inhalt weiterarbeitet (ausliefern, nach der SSID
durchsuchen), muss diese Bytes nehmen und die Datei nicht erneut
lesen - dazwischen kann sie sich aendern.
"""
try:
stat = image.path.stat()
if (stat.st_mtime_ns, stat.st_size) != (image.mtime_ns, image.size):
return None
data = image.path.read_bytes()
except OSError:
return None
try:
if parse_trailer(data) != (image.device_type, image.version):
return None
except InvalidImageError:
return None
return data
def contains_ssid(self, image: FirmwareImage, ssid: str) -> bool:
"""Blockierend, gemerkt je (Datei, SSID)."""
key = (str(image.path), image.mtime_ns, image.size, ssid)
with self._ssid_hits_lock:
cached = key in self._ssid_hits
hit = self._ssid_hits.get(key)
if cached:
return bool(hit)
try:
data = image.path.read_bytes()
except OSError:
return False
hit = contains_ssid(data, ssid)
with self._ssid_hits_lock:
self._ssid_hits[key] = hit
return hit
+76
View File
@@ -0,0 +1,76 @@
"""Liefert OTA-Abbilder an die Geraete aus, unter einmaligen Adressen."""
from __future__ import annotations
from collections.abc import Mapping
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_DEVICE_BASE_URL, DOMAIN, FIRMWARE_URL_PREFIX
class OtaRegistry:
"""Zufaellige Schluessel, jeder an genau die geprueften Bytes eines Abbildes
gebunden und nur gueltig, solange dessen Aktualisierung laeuft.
Bytes statt eines Pfades: die Datei koennte zwischen Pruefung und Abruf
ersetzt werden - die WLAN-Sicherung wuerde dann fuer eine Datei gelten,
die das Geraet nie bekommt.
"""
def __init__(self) -> None:
self._images: dict[str, bytes] = {}
def create(self, data: bytes) -> str:
key = secrets.token_urlsafe(32)
self._images[key] = data
return key
def revoke(self, key: str) -> None:
self._images.pop(key, None)
def get(self, key: str) -> bytes | None:
return self._images.get(key)
def __len__(self) -> int:
return len(self._images)
class NinstripFirmwareView(HomeAssistantView):
"""Ohne HA-Anmeldung: das Geraet kann sich nicht anmelden. Der unratbare
Schluessel ist der Schutz."""
url = FIRMWARE_URL_PREFIX + "{key}"
name = "api:ninstrip: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:
body = self._registry.get(key)
if body is None:
return web.Response(status=404)
return web.Response(body=body, content_type="application/octet-stream")
def device_base_url(hass: HomeAssistant, options: Mapping[str, Any]) -> str:
"""Die http://-Adresse, unter der Geraete Home Assistant erreichen."""
base = options.get(CONF_DEVICE_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)
+1
View File
@@ -3,6 +3,7 @@
"name": "ninstrip",
"codeowners": ["@ninharp"],
"config_flow": true,
"dependencies": ["http"],
"documentation": "https://gitea.zukunpht.de/ninharp/ha-ninstrip",
"integration_type": "device",
"iot_class": "local_polling",
+1 -3
View File
@@ -23,9 +23,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
coordinator = entry.runtime_data
async_add_entities(
[NinstripSacnUniverse(coordinator), NinstripSacnStartSlot(coordinator)]
)
async_add_entities([NinstripSacnUniverse(coordinator), NinstripSacnStartSlot(coordinator)])
class NinstripSacnUniverse(NinstripSacnEntity, NumberEntity):
+3 -1
View File
@@ -69,7 +69,9 @@ class NinstripPresetSelect(NinstripEntity, SelectEntity):
name = str(p.get("name") or f"Szene {p['id'] + 1}")
if name == option:
index = int(p["id"])
await self.coordinator.async_call(lambda c: c.apply_preset(index))
# index als Vorgabewert binden: sonst zeigt das Lambda auf die
# Schleifenvariable statt auf den Wert von jetzt.
await self.coordinator.async_call(lambda c, index=index: c.apply_preset(index))
await self.coordinator.async_after_change()
return
+75
View File
@@ -0,0 +1,75 @@
"""Was sich alle ninstrip-Eintraege teilen: der Abbild-Katalog und die einmaligen Schluessel."""
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 NinstripFirmwareView, OtaRegistry
def firmware_directory(hass: HomeAssistant) -> Path:
return Path(hass.config.path(FIRMWARE_DIR_NAME))
@dataclass
class NinstripShared:
"""Einmal je Home Assistant."""
catalog: FirmwareCatalog
registry: OtaRegistry
scan_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
unsub_scan: CALLBACK_TYPE | None = None
DATA_SHARED: HassKey[NinstripShared] = HassKey(DOMAIN)
@callback
def async_get_shared(hass: HomeAssistant) -> NinstripShared:
"""Beim ersten Zugriff anlegen und die Ausgabe genau einmal anmelden."""
shared = hass.data.get(DATA_SHARED)
if shared is None:
shared = NinstripShared(FirmwareCatalog(firmware_directory(hass)), OtaRegistry())
hass.data[DATA_SHARED] = shared
hass.http.register_view(NinstripFirmwareView(hass, shared.registry))
return shared
async def async_scan_catalog(hass: HomeAssistant, shared: NinstripShared) -> 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: NinstripShared) -> None:
if shared.unsub_scan is not None:
return
async def _turnus(_now: datetime) -> None:
await async_scan_catalog(hass, shared)
shared.unsub_scan = async_track_time_interval(
hass,
_turnus,
CATALOG_SCAN_INTERVAL,
name="ninstrip firmware catalog",
cancel_on_shutdown=True,
)
await async_scan_catalog(hass, shared)
@callback
def async_stop_catalog(shared: NinstripShared) -> None:
if shared.unsub_scan is not None:
shared.unsub_scan()
shared.unsub_scan = None
+44 -1
View File
@@ -40,9 +40,17 @@
"init": {
"title": "Einstellungen",
"data": {
"scan_interval": "Abfrageabstand (Sekunden)"
"scan_interval": "Abfrageabstand (Sekunden)",
"device_base_url": "Adresse, unter der die Geräte Home Assistant erreichen"
},
"data_description": {
"device_base_url": "Nur für Firmware-Updates nötig, wenn die Geräte die interne Adresse von Home Assistant nicht über schlichtes http erreichen, z. B. http://192.168.1.50:8123"
}
}
},
"error": {
"invalid_base_url": "Es muss eine schlichte http://-Adresse mit Host und optionalem Port sein, ohne Pfad.",
"base_url_too_long": "Adresse zu lang – das Gerät nimmt für die ganze Firmware-Adresse höchstens 255 Zeichen."
}
},
"entity": {
@@ -97,6 +105,11 @@
"sacn_multicast": {
"name": "sACN-Multicast beigetreten"
}
},
"update": {
"firmware": {
"name": "Firmware"
}
}
},
"exceptions": {
@@ -111,6 +124,36 @@
},
"cannot_connect": {
"message": "Das Gerät ist nicht erreichbar."
},
"update_busy": {
"message": "Auf dem Gerät läuft bereits eine Aktualisierung."
},
"no_update": {
"message": "Es steht keine neuere Firmware bereit."
},
"no_http_url": {
"message": "Die Geräte erreichen Home Assistant nur über schlichtes http. Die Adresse steht in den Einstellungen der Integration."
},
"device_url_too_long": {
"message": "Die Firmware-Adresse ist länger als die 255 Zeichen, die das Gerät annimmt."
},
"image_invalid": {
"message": "Die Firmware-Datei hat sich geändert oder ist nicht mehr gültig."
},
"wifi_guard": {
"message": "Diese Firmware enthält die WLAN-Zugangsdaten für {ssid} nicht; das Gerät verlöre seine Netzverbindung."
},
"update_failed": {
"message": "Die Aktualisierung ist fehlgeschlagen: {reason}"
},
"update_timeout": {
"message": "Das Gerät hat sich innerhalb von 12 Minuten nicht mit der neuen Firmware gemeldet."
}
},
"issues": {
"wifi_guard": {
"title": "Firmware für {name} wird nicht angeboten",
"description": "Im Firmware-Ordner liegt {version} für {device_type}, aber ohne WLAN-Zugangsdaten für {ssid}. Das Gerät kennt sein WLAN nur aus der Firmware und verlöre nach dieser Aktualisierung seine Netzverbindung. Deshalb wird sie nicht angeboten. Wer das Gerät stattdessen über das Onboarding ins WLAN gebracht hat, bekommt das Update ohne diesen Vorbehalt."
}
}
}
@@ -40,9 +40,17 @@
"init": {
"title": "Einstellungen",
"data": {
"scan_interval": "Abfrageabstand (Sekunden)"
"scan_interval": "Abfrageabstand (Sekunden)",
"device_base_url": "Adresse, unter der die Geräte Home Assistant erreichen"
},
"data_description": {
"device_base_url": "Nur für Firmware-Updates nötig, wenn die Geräte die interne Adresse von Home Assistant nicht über schlichtes http erreichen, z. B. http://192.168.1.50:8123"
}
}
},
"error": {
"invalid_base_url": "Es muss eine schlichte http://-Adresse mit Host und optionalem Port sein, ohne Pfad.",
"base_url_too_long": "Adresse zu lang – das Gerät nimmt für die ganze Firmware-Adresse höchstens 255 Zeichen."
}
},
"entity": {
@@ -97,6 +105,11 @@
"sacn_multicast": {
"name": "sACN-Multicast beigetreten"
}
},
"update": {
"firmware": {
"name": "Firmware"
}
}
},
"exceptions": {
@@ -111,6 +124,36 @@
},
"cannot_connect": {
"message": "Das Gerät ist nicht erreichbar."
},
"update_busy": {
"message": "Auf dem Gerät läuft bereits eine Aktualisierung."
},
"no_update": {
"message": "Es steht keine neuere Firmware bereit."
},
"no_http_url": {
"message": "Die Geräte erreichen Home Assistant nur über schlichtes http. Die Adresse steht in den Einstellungen der Integration."
},
"device_url_too_long": {
"message": "Die Firmware-Adresse ist länger als die 255 Zeichen, die das Gerät annimmt."
},
"image_invalid": {
"message": "Die Firmware-Datei hat sich geändert oder ist nicht mehr gültig."
},
"wifi_guard": {
"message": "Diese Firmware enthält die WLAN-Zugangsdaten für {ssid} nicht; das Gerät verlöre seine Netzverbindung."
},
"update_failed": {
"message": "Die Aktualisierung ist fehlgeschlagen: {reason}"
},
"update_timeout": {
"message": "Das Gerät hat sich innerhalb von 12 Minuten nicht mit der neuen Firmware gemeldet."
}
},
"issues": {
"wifi_guard": {
"title": "Firmware für {name} wird nicht angeboten",
"description": "Im Firmware-Ordner liegt {version} für {device_type}, aber ohne WLAN-Zugangsdaten für {ssid}. Das Gerät kennt sein WLAN nur aus der Firmware und verlöre nach dieser Aktualisierung seine Netzverbindung. Deshalb wird sie nicht angeboten. Wer das Gerät stattdessen über das Onboarding ins WLAN gebracht hat, bekommt das Update ohne diesen Vorbehalt."
}
}
}
@@ -40,9 +40,17 @@
"init": {
"title": "Settings",
"data": {
"scan_interval": "Polling interval (seconds)"
"scan_interval": "Polling interval (seconds)",
"device_base_url": "Address devices use to reach Home Assistant"
},
"data_description": {
"device_base_url": "Only needed for firmware updates if devices cannot reach Home Assistant's internal URL over plain http, e.g. http://192.168.1.50:8123"
}
}
},
"error": {
"invalid_base_url": "Must be a plain http:// address with host and optional port, without path.",
"base_url_too_long": "Address too long – the device accepts at most 255 characters for the whole firmware URL."
}
},
"entity": {
@@ -97,6 +105,11 @@
"sacn_multicast": {
"name": "sACN multicast joined"
}
},
"update": {
"firmware": {
"name": "Firmware"
}
}
},
"exceptions": {
@@ -111,6 +124,36 @@
},
"cannot_connect": {
"message": "The device is unreachable."
},
"update_busy": {
"message": "An update is already running on the device."
},
"no_update": {
"message": "No firmware update is available."
},
"no_http_url": {
"message": "Devices can only reach Home Assistant over plain http. Set the address in the integration options."
},
"device_url_too_long": {
"message": "The firmware URL is longer than the 255 characters the device accepts."
},
"image_invalid": {
"message": "The firmware file changed or is no longer valid."
},
"wifi_guard": {
"message": "This firmware does not contain the Wi-Fi credentials for {ssid}; the device would lose its network connection."
},
"update_failed": {
"message": "The update failed: {reason}"
},
"update_timeout": {
"message": "The device did not report the new firmware within 12 minutes."
}
},
"issues": {
"wifi_guard": {
"title": "Firmware for {name} not offered",
"description": "The firmware folder contains {version} for {device_type}, but without Wi-Fi credentials for {ssid}. The device only knows its Wi-Fi from its firmware and would lose its network connection after this update, so the update is not offered. A device that was put on the network through onboarding instead gets the update without this reservation."
}
}
}
+278
View File
@@ -0,0 +1,278 @@
"""Firmware-Aktualisierung fuer ninstrip, samt WLAN-Sicherung.
Aufgebaut wie ha-ninlight. Zwei Unterschiede, beide aus der Firmware heraus:
* Die Platine steht in `device_type` (system/info), nicht in `product` - das
ist bei allen Platinen gleich. Firmware ohne dieses Feld bekommt gar kein
Angebot: ohne die Platine zu kennen, koennte ein K1Y-Abbild auf einer
KL-RGBC3 landen, und die invertiert das Datensignal nicht.
* Die WLAN-Pruefung sucht die SSID unverpackt im Abbild (siehe `firmware`).
"""
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 NinstripError
from .const import (
DOMAIN,
MAX_DEVICE_URL_LENGTH,
OTA_POLL_INTERVAL_S,
OTA_TIMEOUT_S,
SIGNAL_CATALOG_UPDATED,
)
from .coordinator import NinstripConfigEntry, NinstripCoordinator
from .entity import NinstripEntity
from .firmware import FirmwareImage
from .firmware import contains_ssid as _enthaelt_ssid
from .firmware import is_newer as _ist_neuer
from .http import device_base_url
from .shared import DATA_SHARED
from .urls import firmware_url
PARALLEL_UPDATES = 0 # mehrere Geraete duerfen gleichzeitig; jedes fuer sich ist gesperrt
def haengt_an_kompiliertem_wlan(settings: dict[str, Any]) -> bool:
"""Nur ein ausdrueckliches 'stored' macht ein Geraet vom Abbild unabhaengig."""
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 _fehler(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: NinstripConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
async_add_entities([NinstripUpdateEntity(entry.runtime_data)])
class NinstripUpdateEntity(NinstripEntity, UpdateEntity):
"""Bietet das neueste passende Abbild aus dem Firmware-Ordner an."""
_attr_device_class = UpdateDeviceClass.FIRMWARE
_attr_supported_features = UpdateEntityFeature.INSTALL | UpdateEntityFeature.PROGRESS
def __init__(self, coordinator: NinstripCoordinator) -> 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._install_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:
"""Der eigene `git describe`-Vergleich statt HAs AwesomeVersion.
HA haelt "0.1.0-47-g71a57" und "0.1.0-36-gca9bb" fuer denselben
Versionskern und vergleicht dann die Zeichenketten - das faellt genau
dort falsch aus, wo diese Firmware lebt. `firmware.is_newer` hat die
Reihenfolge, nach der `self._offer` ueberhaupt erst entstanden ist;
so widersprechen sich Zustand und Angebot nie.
"""
return _ist_neuer(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._pruefung_anstossen)
)
self._pruefung_anstossen()
@callback
def _pruefung_anstossen(self) -> None:
# Nie pruefen, waehrend eine Aktualisierung laeuft: das fragte ein
# gerade flashendes Geraet nach seinen Einstellungen und koennte
# Angebot und Hinweis unter der laufenden Aktualisierung wegziehen.
# `_async_install_locked` stoesst selbst genau eine Pruefung an, sobald
# die Merker wieder zurueckgesetzt sind.
if self._attr_in_progress:
return
self.coordinator.config_entry.async_create_task(
self.hass, self._async_pruefen(), name="ninstrip firmware evaluation"
)
@callback
def _handle_coordinator_update(self) -> None:
if not self._attr_in_progress and self._pruefschluessel() != self._evaluated_for:
self._pruefung_anstossen()
super()._handle_coordinator_update()
def _pruefschluessel(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_pruefen(self) -> None:
async with self._eval_lock:
catalog = self.hass.data[DATA_SHARED].catalog
key = self._pruefschluessel()
device_type, installed, ssid = key
offer: FirmwareImage | None = None
blocked: FirmwareImage | None = None
# Firmware vor v0.1.0 meldet keinen device_type. Dann bleibt offen,
# fuer welche Platine sie gebaut wurde - also kein Angebot.
newer = (
catalog.newer_images(device_type, installed)
if self.coordinator.data.info.get("device_type")
else []
)
if newer:
try:
settings = await self.coordinator.async_call(lambda c: c.get_settings())
except HomeAssistantError:
return # das bisherige Angebot bleibt; der naechste Durchlauf versucht es
abhaengig = haengt_an_kompiliertem_wlan(settings)
for image in newer: # neuestes zuerst
if not abhaengig 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._wlan_hinweis(blocked, ssid)
self.async_write_ha_state()
@callback
def _wlan_hinweis(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:
# Nur der Aufruf, der diese Sperre haelt, darf unten in_progress und
# ota_in_progress anfassen: ohne diese Pruefung koennte ein zweiter
# Aufruf bis zu den Gegenproben durchlaufen - HAs eigene Sperre greift
# erst, wenn _attr_in_progress schon gesetzt ist - und sein `finally`
# wuerde die Merker zuruecksetzen, waehrend der erste noch flasht.
if self._install_lock.locked():
raise _fehler("update_busy")
async with self._install_lock:
await self._async_install_locked()
async def _async_install_locked(self) -> None:
shared = self.hass.data[DATA_SHARED]
image = self._offer
if image is None:
raise _fehler("no_update")
base = device_base_url(self.hass, self.coordinator.config_entry.options)
# Die Bytes hier EINMAL lesen und genau diese pruefen: die Ausgabe
# liefert dieselbe Kopie aus dem Speicher aus. Was danach mit der Datei
# passiert - etwa ein Bau, der sie mittendrin ueberschreibt - aendert
# nichts mehr an dem, was beim Geraet ankommt.
data = await self.hass.async_add_executor_job(shared.catalog.verify_and_read, image)
if data is None:
raise _fehler("image_invalid")
# Unmittelbar vor dem Senden noch einmal nachsehen: die Firmware des
# Geraets kann sich seit der Pruefung geaendert haben (etwa ein altes
# Angebot, das eine fehlgeschlagene Pruefung ueberlebt hat) und darf
# dann nicht mehr eingespielt werden.
info = await self.coordinator.async_call(lambda c: c.get_info())
if not _ist_neuer(image.version, str(info.get("fw_version"))):
raise _fehler("no_update")
settings = await self.coordinator.async_call(lambda c: c.get_settings())
if info.get("device_type") != image.device_type:
raise _fehler("image_invalid")
ssid = _ssid(info)
if haengt_an_kompiliertem_wlan(settings) and not await self.hass.async_add_executor_job(
_enthaelt_ssid, data, ssid
):
raise _fehler("wifi_guard", ssid=ssid or "?")
key = shared.registry.create(data)
url = firmware_url(base, key)
if len(url) > MAX_DEVICE_URL_LENGTH:
shared.registry.revoke(key)
raise _fehler("device_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_warten_auf(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()
self._pruefung_anstossen()
async def _async_warten_auf(self, target: str, start_uptime: Any) -> None:
"""Wartet auf die Zielversion, einen Fehler oder einen Neustart mit der alten.
Die Firmware loescht last_ota_error, wenn eine Aktualisierung beginnt
(nl_ota.c) - was hier auftaucht, gehoert also zu diesem Versuch.
"""
deadline = time.monotonic() + OTA_TIMEOUT_S
while True:
await asyncio.sleep(OTA_POLL_INTERVAL_S)
if time.monotonic() >= deadline:
raise _fehler("update_timeout")
try:
async with self.coordinator.lock:
info = await self.coordinator.client.get_info()
except NinstripError:
continue # flasht oder startet gerade neu
version = str(info.get("fw_version"))
if version == target:
return
if error := info.get("last_ota_error"):
raise _fehler("update_failed", reason=str(error))
uptime = info.get("uptime")
if isinstance(uptime, int) and isinstance(start_uptime, int) and uptime < start_uptime:
raise _fehler("update_failed", reason=f"neu gestartet mit {version}")
+34
View File
@@ -0,0 +1,34 @@
"""Die Adresse, unter der Geraete Home Assistant erreichen. Frei von HA-Importen."""
from __future__ import annotations
from urllib.parse import urlsplit
from .const import FIRMWARE_KEY_LENGTH, FIRMWARE_URL_PREFIX, MAX_DEVICE_URL_LENGTH
def firmware_url(base: str, key: str) -> str:
return f"{base.rstrip('/')}{FIRMWARE_URL_PREFIX}{key}"
def validate_device_base_url(url: str) -> str | None:
"""None, wenn brauchbar, sonst der Fehlerschluessel fuer das Formular."""
try:
parts = urlsplit(url)
port_ok = parts.port is None or parts.port > 0
except ValueError:
return "invalid_base_url"
if (
parts.scheme != "http"
or not parts.hostname
or not port_ok
or parts.username
or parts.password
or parts.query
or parts.fragment
or parts.path not in ("", "/")
):
return "invalid_base_url"
if len(firmware_url(url, "k" * FIRMWARE_KEY_LENGTH)) > MAX_DEVICE_URL_LENGTH:
return "base_url_too_long"
return None
+27
View File
@@ -0,0 +1,27 @@
[tool.pytest.ini_options]
testpaths = ["tests_unit", "tests"]
asyncio_mode = "auto"
addopts = "-p no:cacheprovider"
[tool.coverage.run]
source = ["custom_components/ninstrip"]
branch = true
[tool.coverage.report]
fail_under = 95
show_missing = true
[tool.ruff]
target-version = "py314"
line-length = 100
exclude = ["docs/"]
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM", "RUF", "ASYNC"]
# E501: line length is the formatter's job; long literals in test tables stay on one line.
# RUF012: HA entities declare _attr_* lists as plain class attributes by convention.
ignore = ["E501", "RUF012"]
[tool.ruff.lint.isort]
known-first-party = ["custom_components.ninstrip", "tests", "tests_unit"]
force-sort-within-sections = true
+4
View File
@@ -0,0 +1,4 @@
homeassistant==2026.9.0
pytest-homeassistant-custom-component==0.13.363
pytest-cov
ruff>=0.9
+9
View File
@@ -0,0 +1,9 @@
#!/usr/bin/env bash
# Ersatz fuer CI: das Repo liegt in einem privaten Gitea ohne Actions.
set -euo pipefail
cd "$(dirname "$0")/.."
.venv/bin/ruff check .
.venv/bin/ruff format --check .
.venv/bin/pytest -q --cov --cov-report=term-missing
.venv/bin/python scripts/check_translations.py
echo "check.sh: alles gruen"
+40
View File
@@ -0,0 +1,40 @@
"""strings.json und translations/de.json muessen gleich sein; en.json braucht dieselben Schluessel.
Anders herum als bei ha-ninlight: dort steht in strings.json Englisch, hier
Deutsch - die Oberflaeche dieser Integration ist deutsch, und strings.json ist
die Quelle, aus der Home Assistant die Vorgabetexte nimmt.
"""
from __future__ import annotations
import json
from pathlib import Path
import sys
BASE = Path(__file__).parents[1] / "custom_components" / "ninstrip"
def keys(data: dict, prefix: str = "") -> set[str]:
out: set[str] = set()
for key, value in data.items():
out |= keys(value, f"{prefix}{key}.") if isinstance(value, dict) else {f"{prefix}{key}"}
return out
def main() -> int:
strings = json.loads((BASE / "strings.json").read_text(encoding="utf-8"))
de = json.loads((BASE / "translations" / "de.json").read_text(encoding="utf-8"))
en = json.loads((BASE / "translations" / "en.json").read_text(encoding="utf-8"))
if strings != de:
print("translations/de.json weicht von strings.json ab")
return 1
diff = sorted(keys(strings) ^ keys(en))
if diff:
print("en.json und strings.json haben unterschiedliche Schluessel:", diff)
return 1
print("Uebersetzungen vollstaendig")
return 0
if __name__ == "__main__":
sys.exit(main())
View File
+131
View File
@@ -0,0 +1,131 @@
"""Fixtures fuer die Tests, die gegen eine Home-Assistant-Instanz laufen."""
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
from homeassistant.const import CONF_HOST, CONF_TOKEN
from homeassistant.core import HomeAssistant
import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.ninstrip.api import NinstripClient
from custom_components.ninstrip.const import DOMAIN
from tests_unit.samples import (
HOST_K1Y,
INFO_K1Y,
MAC_K1Y,
PRESETS_K1Y,
SETTINGS_COMPILED,
STATE_K1Y,
TOKEN,
)
CLIENT_PATCH_TARGETS = [
"custom_components.ninstrip.NinstripClient",
"custom_components.ninstrip.config_flow.NinstripClient",
]
@pytest.fixture(autouse=True)
def auto_enable_custom_integrations(enable_custom_integrations: None) -> Generator[None]:
"""custom_components/ in jedem Home-Assistant-Test laden."""
yield
@pytest.fixture
def strip_state() -> dict[str, Any]:
return deepcopy(STATE_K1Y)
@pytest.fixture
def strip_info() -> dict[str, Any]:
return deepcopy(INFO_K1Y)
@pytest.fixture
def strip_settings() -> dict[str, Any]:
return deepcopy(SETTINGS_COMPILED)
@pytest.fixture
def strip_presets() -> dict[str, Any]:
return deepcopy(PRESETS_K1Y)
@pytest.fixture
def mock_client(
strip_state: dict[str, Any],
strip_info: dict[str, Any],
strip_settings: dict[str, Any],
strip_presets: dict[str, Any],
) -> Generator[MagicMock]:
"""Ein Geraet, dessen Antworten den veraenderbaren strip_*-Fixtures folgen."""
client = MagicMock(spec=NinstripClient)
client.host = HOST_K1Y
client.get_state = AsyncMock(side_effect=lambda: deepcopy(strip_state))
client.get_info = AsyncMock(side_effect=lambda: deepcopy(strip_info))
client.get_settings = AsyncMock(side_effect=lambda: deepcopy(strip_settings))
client.get_presets = AsyncMock(side_effect=lambda: deepcopy(strip_presets))
async def set_state(**fields: Any) -> dict[str, Any]:
strip_state.update(fields)
return deepcopy(strip_state)
client.set_state = AsyncMock(side_effect=set_state)
async def set_sacn(**fields: Any) -> None:
# Wie das Geraet: es merkt sich, was es annimmt, und der Aufrufer
# liest den Block danach zurueck.
strip_settings.setdefault("sacn", {}).update(fields)
client.set_sacn = AsyncMock(side_effect=set_sacn)
client.apply_preset = AsyncMock(return_value=None)
client.save_preset = AsyncMock(return_value=None)
client.reboot = AsyncMock(return_value=None)
client.fw_update = AsyncMock(return_value=None)
patches = [patch(target, return_value=client) for target in CLIENT_PATCH_TARGETS]
for p in patches:
p.start()
yield client
for p in patches:
p.stop()
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
return MockConfigEntry(
domain=DOMAIN,
title="ninstrip ad4c",
unique_id=MAC_K1Y,
data={CONF_HOST: HOST_K1Y, CONF_TOKEN: TOKEN},
)
@pytest.fixture
async def init_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_client: MagicMock
) -> MockConfigEntry:
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Tests des Einrichtungsablaufs duerfen den Eintrag nicht wirklich aufsetzen."""
with patch("custom_components.ninstrip.async_setup_entry", return_value=True) as mock:
yield mock
@pytest.fixture(autouse=True)
def firmware_dir(tmp_path: Path) -> Generator[Path]:
"""Jeder Test bekommt seinen eigenen, anfangs leeren Firmware-Ordner."""
verzeichnis = tmp_path / "ninstrip_firmware"
with patch("custom_components.ninstrip.shared.firmware_directory", return_value=verzeichnis):
yield verzeichnis
+27
View File
@@ -0,0 +1,27 @@
"""Helfer fuer die Home-Assistant-Tests."""
from __future__ import annotations
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from custom_components.ninstrip.const import DOMAIN
from tests_unit.samples import MAC_K1Y
def entity_id_for(hass: HomeAssistant, platform: str, key: str, mac: str = MAC_K1Y) -> str:
"""Entitaeten ueber ihre unique_id suchen, statt Namen zu raten."""
entity_id = er.async_get(hass).async_get_entity_id(platform, DOMAIN, f"{mac}_{key}")
assert entity_id is not None, f"keine {platform}-Entitaet mit dem Schluessel {key}"
return entity_id
async def abfragen(hass: HomeAssistant, freezer) -> None:
"""Einen Abfragezyklus vorspulen (Vorgabe: 30 s)."""
from datetime import timedelta
from pytest_homeassistant_custom_component.common import async_fire_time_changed
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
+23
View File
@@ -0,0 +1,23 @@
"""Neustart-Knopf."""
from __future__ import annotations
from unittest.mock import MagicMock
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry
from tests.helpers import entity_id_for
async def test_knopf_startet_das_geraet_neu(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
await hass.services.async_call(
"button",
"press",
{ATTR_ENTITY_ID: entity_id_for(hass, "button", "restart")},
blocking=True,
)
mock_client.reboot.assert_awaited_once()
+269
View File
@@ -0,0 +1,269 @@
"""Einrichtungsablauf: von Hand, ueber mDNS, erneute Anmeldung und Einstellungen."""
from __future__ import annotations
from ipaddress import ip_address
from typing import Any
from unittest.mock import MagicMock
from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
from homeassistant.const import CONF_HOST, CONF_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.ninstrip.api import (
NinstripAuthError,
NinstripConnectionError,
NinstripResponseError,
)
from custom_components.ninstrip.const import (
CONF_DEVICE_BASE_URL,
CONF_SCAN_INTERVAL,
DEFAULT_TOKEN,
DOMAIN,
)
from tests_unit.samples import HOST_K1Y, MAC_K1Y, TOKEN
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
NEUES_TOKEN = "b71e0c9d44a2f6e3"
def zeroconf_info(ip: str = HOST_K1Y, **properties: Any) -> ZeroconfServiceInfo:
props = {"mac": MAC_K1Y, "name": "Stehlampe"} | properties
return ZeroconfServiceInfo(
ip_address=ip_address(ip),
ip_addresses=[ip_address(ip)],
port=3333,
hostname="ninstrip-ad4c.local.",
type="_ninstrip._tcp.local.",
name="ninstrip-ad4c._ninstrip._tcp.local.",
properties={k: v for k, v in props.items() if v is not None},
)
def bestehender_eintrag(hass: HomeAssistant, host: str = HOST_K1Y) -> MockConfigEntry:
eintrag = MockConfigEntry(
domain=DOMAIN, title="Strip", unique_id=MAC_K1Y, data={CONF_HOST: host, CONF_TOKEN: TOKEN}
)
eintrag.add_to_hass(hass)
return eintrag
async def test_eintrag_von_hand(hass: HomeAssistant, mock_client: MagicMock) -> None:
ergebnis = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
assert ergebnis["type"] is FlowResultType.FORM
assert ergebnis["step_id"] == "user"
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_HOST: HOST_K1Y, CONF_TOKEN: TOKEN}
)
assert ergebnis["type"] is FlowResultType.CREATE_ENTRY
assert ergebnis["data"] == {CONF_HOST: HOST_K1Y, CONF_TOKEN: TOKEN}
assert ergebnis["result"].unique_id == MAC_K1Y
# system/info antwortet auch ohne Token - geprueft wird an system/settings.
mock_client.get_settings.assert_awaited_once()
async def test_das_werks_token_ist_vorbelegt(hass: HomeAssistant, mock_client: MagicMock) -> None:
ergebnis = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
schluessel = next(k for k in ergebnis["data_schema"].schema if k == CONF_TOKEN)
assert schluessel.default() == DEFAULT_TOKEN
async def test_der_titel_kommt_vom_geraetenamen(
hass: HomeAssistant, mock_client: MagicMock, strip_info: dict[str, Any]
) -> None:
strip_info["name"] = "Stehlampe"
ergebnis = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_HOST: HOST_K1Y, CONF_TOKEN: TOKEN}
)
assert ergebnis["title"] == "Stehlampe"
async def test_ohne_namen_zaehlen_produkt_und_mac(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
ergebnis = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_HOST: HOST_K1Y, CONF_TOKEN: TOKEN}
)
assert ergebnis["title"] == "ninstrip ad4c"
@pytest.mark.parametrize(
("methode", "fehler", "erwartet"),
[
("get_settings", NinstripAuthError("401"), "invalid_auth"),
("get_settings", NinstripConnectionError("weg"), "cannot_connect"),
("get_info", NinstripResponseError("kein JSON"), "not_ninstrip"),
("get_settings", RuntimeError("peng"), "unknown"),
],
)
async def test_fehler_und_zweiter_versuch(
hass: HomeAssistant,
mock_client: MagicMock,
methode: str,
fehler: Exception,
erwartet: str,
) -> None:
mock = getattr(mock_client, methode)
urspruenglich = mock.side_effect
mock.side_effect = fehler
ergebnis = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_HOST: HOST_K1Y, CONF_TOKEN: TOKEN}
)
assert ergebnis["type"] is FlowResultType.FORM
assert ergebnis["errors"] == {"base": erwartet}
mock.side_effect = urspruenglich
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_HOST: HOST_K1Y, CONF_TOKEN: TOKEN}
)
assert ergebnis["type"] is FlowResultType.CREATE_ENTRY
async def test_dasselbe_geraet_unter_neuer_adresse_schreibt_den_host_um(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
eintrag = bestehender_eintrag(hass)
ergebnis = await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_HOST: "192.168.2.99", CONF_TOKEN: TOKEN}
)
assert ergebnis["type"] is FlowResultType.ABORT
assert ergebnis["reason"] == "already_configured"
assert eintrag.data[CONF_HOST] == "192.168.2.99"
async def test_mdns_fragt_nur_nach_dem_token(hass: HomeAssistant, mock_client: MagicMock) -> None:
ergebnis = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=zeroconf_info()
)
assert ergebnis["type"] is FlowResultType.FORM
assert ergebnis["step_id"] == "discovery_confirm"
assert ergebnis["description_placeholders"] == {"name": "Stehlampe", "host": HOST_K1Y}
assert set(ergebnis["data_schema"].schema) == {CONF_TOKEN}
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_TOKEN: TOKEN}
)
assert ergebnis["type"] is FlowResultType.CREATE_ENTRY
assert ergebnis["data"] == {CONF_HOST: HOST_K1Y, CONF_TOKEN: TOKEN}
async def test_mdns_ohne_namen_nimmt_den_dienstnamen(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
ergebnis = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=zeroconf_info(name=None)
)
assert ergebnis["description_placeholders"]["name"] == "ninstrip-ad4c"
async def test_mdns_eines_bekannten_geraets_schreibt_den_host_um(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
eintrag = bestehender_eintrag(hass, "192.168.2.5")
ergebnis = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=zeroconf_info()
)
assert ergebnis["type"] is FlowResultType.ABORT
assert ergebnis["reason"] == "already_configured"
assert eintrag.data[CONF_HOST] == HOST_K1Y
async def test_mdns_ohne_mac_faellt_auf_die_kennung_aus_info_zurueck(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
"""Ohne MAC in den mDNS-Eigenschaften entscheidet erst system/info."""
ergebnis = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=zeroconf_info(mac=None)
)
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_TOKEN: TOKEN}
)
assert ergebnis["type"] is FlowResultType.CREATE_ENTRY
assert ergebnis["result"].unique_id == MAC_K1Y
async def test_erneute_anmeldung_ersetzt_das_token(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
eintrag = bestehender_eintrag(hass)
ergebnis = await eintrag.start_reauth_flow(hass)
assert ergebnis["step_id"] == "reauth_confirm"
# HA legt den Eintragstitel als "name" dazu.
assert ergebnis["description_placeholders"]["host"] == HOST_K1Y
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_TOKEN: NEUES_TOKEN}
)
assert ergebnis["type"] is FlowResultType.ABORT
assert ergebnis["reason"] == "reauth_successful"
assert eintrag.data[CONF_TOKEN] == NEUES_TOKEN
async def test_erneute_anmeldung_mit_falschem_token_bleibt_im_formular(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
eintrag = bestehender_eintrag(hass)
mock_client.get_settings.side_effect = NinstripAuthError("401")
ergebnis = await eintrag.start_reauth_flow(hass)
ergebnis = await hass.config_entries.flow.async_configure(
ergebnis["flow_id"], {CONF_TOKEN: "falsch"}
)
assert ergebnis["type"] is FlowResultType.FORM
assert ergebnis["errors"] == {"base": "invalid_auth"}
assert eintrag.data[CONF_TOKEN] == TOKEN
async def test_einstellungen_abfrageabstand_und_adresse(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
eintrag = bestehender_eintrag(hass)
ergebnis = await hass.config_entries.options.async_init(eintrag.entry_id)
ergebnis = await hass.config_entries.options.async_configure(
ergebnis["flow_id"],
{CONF_SCAN_INTERVAL: 45, CONF_DEVICE_BASE_URL: "http://192.168.1.50:8123/"},
)
assert ergebnis["type"] is FlowResultType.CREATE_ENTRY
assert eintrag.options == {
CONF_SCAN_INTERVAL: 45,
# Der abschliessende Schraegstrich faellt weg, sonst entstuende eine
# Adresse mit zwei Schraegstrichen.
CONF_DEVICE_BASE_URL: "http://192.168.1.50:8123",
}
async def test_einstellungen_ohne_adresse_lassen_sie_weg(
hass: HomeAssistant, mock_client: MagicMock
) -> None:
eintrag = bestehender_eintrag(hass)
ergebnis = await hass.config_entries.options.async_init(eintrag.entry_id)
ergebnis = await hass.config_entries.options.async_configure(
ergebnis["flow_id"], {CONF_SCAN_INTERVAL: 30}
)
assert ergebnis["type"] is FlowResultType.CREATE_ENTRY
assert CONF_DEVICE_BASE_URL not in eintrag.options
@pytest.mark.parametrize(
("adresse", "fehler"),
[
("https://ha.local:8123", "invalid_base_url"),
("http://ha.local:8123/pfad", "invalid_base_url"),
("http://" + "a" * 183, "base_url_too_long"),
],
)
async def test_einstellungen_weisen_unbrauchbare_adressen_ab(
hass: HomeAssistant, mock_client: MagicMock, adresse: str, fehler: str
) -> None:
eintrag = bestehender_eintrag(hass)
ergebnis = await hass.config_entries.options.async_init(eintrag.entry_id)
ergebnis = await hass.config_entries.options.async_configure(
ergebnis["flow_id"], {CONF_SCAN_INTERVAL: 30, CONF_DEVICE_BASE_URL: adresse}
)
assert ergebnis["type"] is FlowResultType.FORM
assert ergebnis["errors"] == {CONF_DEVICE_BASE_URL: fehler}
+174
View File
@@ -0,0 +1,174 @@
"""Verhalten des Koordinators, das an keiner einzelnen Plattform haengt."""
from __future__ import annotations
from copy import deepcopy
from datetime import timedelta
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_ON,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
import pytest
from pytest_homeassistant_custom_component.common import async_fire_time_changed
from custom_components.ninstrip.api import (
NinstripBusyError,
NinstripConnectionError,
NinstripRequestError,
)
from tests.helpers import entity_id_for
from tests_unit.samples import STATE_K1Y
@pytest.mark.parametrize(
("error", "key", "placeholders"),
[
(
NinstripRequestError(400, "brightness"),
"request_failed",
{"message": "brightness"},
),
(NinstripBusyError(409, "Update laeuft schon"), "device_busy", None),
(NinstripConnectionError("reset"), "cannot_connect", None),
],
)
async def test_schreibfehler_werden_uebersetzt(
hass: HomeAssistant, init_integration, mock_client: MagicMock, error, key, placeholders
) -> None:
mock_client.set_state.side_effect = error
with pytest.raises(HomeAssistantError) as raised:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id_for(hass, "light", "light")},
blocking=True,
)
assert raised.value.translation_key == key
assert raised.value.translation_placeholders == placeholders
async def test_erste_zwei_fehlversuche_lassen_das_geraet_verfuegbar(
hass: HomeAssistant, init_integration, mock_client: MagicMock, freezer: FrozenDateTimeFactory
) -> None:
entity_id = entity_id_for(hass, "light", "light")
mock_client.get_state.side_effect = NinstripConnectionError("down")
reads = mock_client.get_state.await_count
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_client.get_state.await_count == reads + 1
assert hass.states.get(entity_id).state == STATE_ON
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_client.get_state.await_count == reads + 2
assert hass.states.get(entity_id).state == STATE_ON
async def test_dritter_fehlversuch_in_folge_macht_es_unverfuegbar(
hass: HomeAssistant, init_integration, mock_client: MagicMock, freezer: FrozenDateTimeFactory
) -> None:
entity_id = entity_id_for(hass, "light", "light")
mock_client.get_state.side_effect = NinstripConnectionError("down")
for _ in range(3):
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_erfolgreiche_abfrage_setzt_den_zaehler_zurueck(
hass: HomeAssistant, init_integration, mock_client: MagicMock, freezer: FrozenDateTimeFactory
) -> None:
entity_id = entity_id_for(hass, "light", "light")
mock_client.get_state.side_effect = NinstripConnectionError("down")
# Zwei Fehlversuche werden geschluckt.
for _ in range(2):
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
# Ein erfolgreicher Durchlauf setzt den Zaehler zurueck.
mock_client.get_state.side_effect = lambda: deepcopy(STATE_K1Y)
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
# Zwei weitere Fehlversuche in Folge muessen wieder geschluckt werden.
mock_client.get_state.side_effect = NinstripConnectionError("down")
for _ in range(2):
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
async def test_einschalten_erreicht_das_geraet_trotz_geschluckter_fehlversuche(
hass: HomeAssistant, init_integration, mock_client: MagicMock, freezer: FrozenDateTimeFactory
) -> None:
entity_id = entity_id_for(hass, "light", "light")
mock_client.get_state.side_effect = NinstripConnectionError("down")
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_client.set_state.assert_awaited_with(on=True)
async def test_erfolgreiches_schreiben_macht_es_wieder_verfuegbar(
hass: HomeAssistant, init_integration, mock_client: MagicMock, freezer: FrozenDateTimeFactory
) -> None:
entity_id = entity_id_for(hass, "light", "light")
mock_client.get_state.side_effect = NinstripConnectionError("down")
for _ in range(3):
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
await init_integration.runtime_data.async_set_state(on=True)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
# Das Schreiben hat den Zaehler zurueckgesetzt: ein Fehlversuch bleibt folgenlos.
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
async def test_abfragen_pausiert_waehrend_einer_aktualisierung(
hass: HomeAssistant, init_integration, mock_client: MagicMock, freezer: FrozenDateTimeFactory
) -> None:
init_integration.runtime_data.ota_in_progress = True
mock_client.get_state.side_effect = NinstripConnectionError("rebooting")
reads = mock_client.get_state.await_count
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_client.get_state.await_count == reads
assert hass.states.get(entity_id_for(hass, "light", "light")).state == STATE_ON
+60
View File
@@ -0,0 +1,60 @@
"""Geraetename und Modellbeschreibung, auch wenn das Geraet wenig meldet."""
from __future__ import annotations
from typing import Any
from freezegun.api import FrozenDateTimeFactory
from homeassistant.const import ATTR_ENTITY_ID, STATE_ON
from homeassistant.core import HomeAssistant
import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.ninstrip.entity import device_name, strip_model
from tests.helpers import abfragen, entity_id_for
def test_eigener_name_hat_vorrang() -> None:
assert device_name({"name": " Stehlampe ", "mac": "b0:81:84:49:ad:4c"}) == "Stehlampe"
def test_ohne_namen_zaehlen_produkt_und_die_letzten_vier_mac_stellen() -> None:
assert device_name({"name": "", "mac": "b0:81:84:49:ad:4c"}) == "ninstrip ad4c"
assert device_name({"mac": "b0:81:84:49:ad:4c", "product": "ninstrip"}) == "ninstrip ad4c"
@pytest.mark.parametrize(
("info", "erwartet"),
[
({"strip": {"chipset": "ws2805", "pixel_count": 40}}, "ws2805, 40 Pixel"),
({"strip": {"chipset": "ws2805"}}, "ws2805"),
({"strip": {}}, None),
({}, None), # Geraet im Onboarding: noch kein strip-Block
],
)
def test_modellbeschreibung(info: dict[str, Any], erwartet: str | None) -> None:
assert strip_model(info) == erwartet
async def test_sacn_schalter_laesst_sich_auch_einschalten(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_settings: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
strip_settings["sacn"]["enabled"] = False
await abfragen(hass, freezer)
entity_id = entity_id_for(hass, "switch", "sacn_enabled")
await hass.services.async_call("switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True)
assert hass.states.get(entity_id).state == STATE_ON
async def test_helligkeit_ohne_wert_bleibt_leer(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_state: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
strip_state["brightness"] = None
await abfragen(hass, freezer)
assert hass.states.get(entity_id_for(hass, "light", "light")).attributes["brightness"] is None
+79
View File
@@ -0,0 +1,79 @@
"""Einmalige Firmware-Adressen und die Adresse, unter der Geraete HA erreichen."""
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.ninstrip.const import CONF_DEVICE_BASE_URL
from custom_components.ninstrip.http import OtaRegistry, device_base_url
from custom_components.ninstrip.shared import DATA_SHARED
from tests_unit.images import build_image
def test_schluessel_sind_einmalig_und_43_zeichen_lang() -> None:
registry = OtaRegistry()
schluessel = {registry.create(b"nutzdaten") for _ in range(50)}
assert len(schluessel) == 50
assert all(len(k) == 43 for k in schluessel)
assert len(registry) == 50
async def test_die_ausgabe_liefert_nur_die_eigenen_bytes(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
init_integration,
firmware_dir: Path,
) -> None:
registry = hass.data[DATA_SHARED].registry
meins = build_image()
anderes = build_image(version="0.1.1-rc1")
schluessel = registry.create(meins)
anderer = registry.create(anderes)
client = await hass_client_no_auth()
antwort = await client.get(f"/api/ninstrip/firmware/{schluessel}")
assert antwort.status == 200
assert await antwort.read() == meins
andere_antwort = await client.get(f"/api/ninstrip/firmware/{anderer}")
assert andere_antwort.status == 200
assert await andere_antwort.read() == anderes
assert (await client.get("/api/ninstrip/firmware/" + "x" * 43)).status == 404
registry.revoke(schluessel)
assert (await client.get(f"/api/ninstrip/firmware/{schluessel}")).status == 404
# Ein zurueckgezogener Schluessel zieht keinen anderen mit.
assert (await client.get(f"/api/ninstrip/firmware/{anderer}")).status == 200
async def test_die_eingestellte_adresse_hat_vorrang(hass: HomeAssistant) -> None:
await async_process_ha_core_config(hass, {"internal_url": "http://192.168.1.60:8123"})
assert (
device_base_url(hass, {CONF_DEVICE_BASE_URL: "http://192.168.1.50:8123"})
== "http://192.168.1.50:8123"
)
async def test_sonst_gilt_die_interne_http_adresse(hass: HomeAssistant) -> None:
await async_process_ha_core_config(hass, {"internal_url": "http://192.168.1.60:8123"})
assert device_base_url(hass, {}) == "http://192.168.1.60:8123"
async def test_https_wird_abgelehnt(hass: HomeAssistant) -> None:
"""Die Firmware laedt nur ueber schlichtes http (esp_https_ota mit ALLOW_HTTP)."""
await async_process_ha_core_config(hass, {"internal_url": "https://ha.local:8123"})
with pytest.raises(HomeAssistantError) as raised:
device_base_url(hass, {})
assert raised.value.translation_key == "no_http_url"
async def test_ohne_jede_adresse(hass: HomeAssistant) -> None:
with pytest.raises(HomeAssistantError) as raised:
device_base_url(hass, {})
assert raised.value.translation_key == "no_http_url"
+59
View File
@@ -0,0 +1,59 @@
"""Einrichten, Entladen und das Geraet in der Registrierung."""
from __future__ import annotations
from unittest.mock import MagicMock
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.ninstrip.api import NinstripAuthError, NinstripConnectionError
from custom_components.ninstrip.const import DOMAIN
from tests_unit.samples import MAC_K1Y
async def test_einrichten_und_entladen(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
assert init_integration.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(init_integration.entry_id)
await hass.async_block_till_done()
assert init_integration.state is ConfigEntryState.NOT_LOADED
async def test_geraet_traegt_mac_modell_und_version(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
geraet = dr.async_get(hass).async_get_device_by_identifier(
(DOMAIN, MAC_K1Y), init_integration.entry_id
)
assert geraet is not None
assert geraet.manufacturer == "ninstrip"
# Das Modell beschreibt den Strip, nicht das Produkt - das steht in model_id.
assert geraet.model == "sk6812_rgbw, 60 Pixel"
assert geraet.model_id == "ninstrip"
assert geraet.sw_version == "0.1.0-12-g71a57"
assert (dr.CONNECTION_NETWORK_MAC, MAC_K1Y) in geraet.connections
async def test_unerreichbares_geraet_wird_spaeter_erneut_versucht(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_client: MagicMock
) -> None:
mock_client.get_state.side_effect = NinstripConnectionError("weg")
mock_config_entry.add_to_hass(hass)
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_abgelehntes_token_verlangt_neue_anmeldung(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_client: MagicMock
) -> None:
mock_client.get_state.side_effect = NinstripAuthError("abgelehnt")
mock_config_entry.add_to_hass(hass)
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
assert any(
flow["context"]["source"] == "reauth" for flow in hass.config_entries.flow.async_progress()
)
+125
View File
@@ -0,0 +1,125 @@
"""Hauptschalter und Helligkeit."""
from __future__ import annotations
from datetime import timedelta
from typing import Any
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.light import ATTR_BRIGHTNESS, ColorMode
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry, async_fire_time_changed
from tests.helpers import entity_id_for
def light_id(hass: HomeAssistant) -> str:
return entity_id_for(hass, "light", "light")
async def abfragen(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> None:
"""Einen Abfragezyklus vorspulen."""
freezer.tick(timedelta(seconds=31))
async_fire_time_changed(hass)
await hass.async_block_till_done()
async def test_zustand_und_helligkeit_aus_prozent(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
state = hass.states.get(light_id(hass))
assert state is not None
assert state.state == STATE_ON
# Das Geraet rechnet in Prozent: 100 % sind 255.
assert state.attributes[ATTR_BRIGHTNESS] == 255
assert state.attributes["color_mode"] == ColorMode.BRIGHTNESS
assert state.attributes["pixel_count"] == 60
assert state.attributes["segment_count"] == 1
assert state.attributes["preset"] == 2
@pytest.mark.parametrize(
("prozent", "erwartet"), [(0, 0), (1, 3), (37, 94), (50, 128), (99, 252), (100, 255)]
)
async def test_prozent_werden_auf_0_bis_255_umgerechnet(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_state: dict[str, Any],
freezer: FrozenDateTimeFactory,
prozent: int,
erwartet: int,
) -> None:
strip_state["brightness"] = prozent
await abfragen(hass, freezer)
assert hass.states.get(light_id(hass)).attributes[ATTR_BRIGHTNESS] == erwartet
@pytest.mark.parametrize("wert", [-5, 150])
async def test_unsinnige_helligkeit_wird_gekappt(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_state: dict[str, Any],
freezer: FrozenDateTimeFactory,
wert: int,
) -> None:
"""Ein Geraet, das ausserhalb von 0..100 antwortet, darf HA nicht sprengen."""
strip_state["brightness"] = wert
await abfragen(hass, freezer)
assert hass.states.get(light_id(hass)).attributes[ATTR_BRIGHTNESS] in (0, 255)
async def test_einschalten_ohne_helligkeit_schickt_nur_on(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
await hass.services.async_call(
"light", "turn_on", {ATTR_ENTITY_ID: light_id(hass)}, blocking=True
)
mock_client.set_state.assert_awaited_once_with(on=True)
@pytest.mark.parametrize(("ha_wert", "prozent"), [(255, 100), (128, 50), (3, 1), (1, 0)])
async def test_einschalten_mit_helligkeit_rechnet_in_prozent(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
ha_wert: int,
prozent: int,
) -> None:
await hass.services.async_call(
"light",
"turn_on",
{ATTR_ENTITY_ID: light_id(hass), ATTR_BRIGHTNESS: ha_wert},
blocking=True,
)
mock_client.set_state.assert_awaited_once_with(on=True, brightness=prozent)
async def test_ausschalten(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
await hass.services.async_call(
"light", "turn_off", {ATTR_ENTITY_ID: light_id(hass)}, blocking=True
)
mock_client.set_state.assert_awaited_once_with(on=False)
assert hass.states.get(light_id(hass)).state == STATE_OFF
async def test_die_antwort_des_geraets_zaehlt_nicht_die_anfrage(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
"""Das Geraet darf die Helligkeit abweichend beantworten - dann gilt seine."""
async def antwortet_anders(**fields: Any) -> dict[str, Any]:
return {**mock_client.get_state.side_effect(), **fields, "brightness": 20}
mock_client.set_state.side_effect = antwortet_anders
await hass.services.async_call(
"light",
"turn_on",
{ATTR_ENTITY_ID: light_id(hass), ATTR_BRIGHTNESS: 255},
blocking=True,
)
assert hass.states.get(light_id(hass)).attributes[ATTR_BRIGHTNESS] == 51
+160
View File
@@ -0,0 +1,160 @@
"""sACN: Schalter, Universe, Startslot, Verhalten nach dem Stromende, Merker."""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import entity_registry as er
import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry
from tests.helpers import abfragen, entity_id_for
async def test_schalter_zeigt_den_eingestellten_zustand(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
assert hass.states.get(entity_id_for(hass, "switch", "sacn_enabled")).state == STATE_ON
async def test_schalter_schreibt_und_liest_zurueck(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
entity_id = entity_id_for(hass, "switch", "sacn_enabled")
await hass.services.async_call("switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True)
mock_client.set_sacn.assert_awaited_once_with(enabled=False)
# Nicht geraten, sondern zurueckgelesen: das Geraet antwortet nur mit "ok".
assert mock_client.get_settings.await_count >= 2
assert hass.states.get(entity_id).state == STATE_OFF
async def test_universe_und_startslot(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
universe = entity_id_for(hass, "number", "sacn_universe")
slot = entity_id_for(hass, "number", "sacn_start_slot")
assert hass.states.get(universe).state == "1.0"
assert hass.states.get(slot).state == "1.0"
await hass.services.async_call(
"number", "set_value", {ATTR_ENTITY_ID: universe, "value": 7}, blocking=True
)
mock_client.set_sacn.assert_awaited_with(universe=7)
await hass.services.async_call(
"number", "set_value", {ATTR_ENTITY_ID: slot, "value": 4}, blocking=True
)
mock_client.set_sacn.assert_awaited_with(start_slot=4)
async def test_startslot_endet_bei_510(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
"""Hinter 510 passt bei drei Kanaeln kein ganzes Pixel mehr ins Universe."""
slot = entity_id_for(hass, "number", "sacn_start_slot")
assert hass.states.get(slot).attributes["max"] == 510
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
"number", "set_value", {ATTR_ENTITY_ID: slot, "value": 511}, blocking=True
)
mock_client.set_sacn.assert_not_awaited()
@pytest.mark.parametrize(("kanaele", "grenze"), [(3, 510), (4, 509), (5, 508)])
async def test_die_grenze_folgt_den_kanaelen_je_pixel(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_settings: dict[str, Any],
freezer: FrozenDateTimeFactory,
kanaele: int,
grenze: int,
) -> None:
strip_settings["sacn"]["channels_per_pixel"] = kanaele
await abfragen(hass, freezer)
slot = entity_id_for(hass, "number", "sacn_start_slot")
assert hass.states.get(slot).attributes["max"] == grenze
async def test_unsinnige_kanalzahl_faellt_auf_drei_zurueck(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_settings: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
strip_settings["sacn"]["channels_per_pixel"] = "drei"
await abfragen(hass, freezer)
slot = entity_id_for(hass, "number", "sacn_start_slot")
assert hass.states.get(slot).attributes["max"] == 510
async def test_verhalten_nach_dem_stromende(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
entity_id = entity_id_for(hass, "select", "sacn_on_timeout")
state = hass.states.get(entity_id)
assert state.state == "release"
assert state.attributes["options"] == ["release", "hold"]
await hass.services.async_call(
"select", "select_option", {ATTR_ENTITY_ID: entity_id, "option": "hold"}, blocking=True
)
mock_client.set_sacn.assert_awaited_once_with(on_timeout="hold")
assert hass.states.get(entity_id).state == "hold"
async def test_unbekanntes_verhalten_zeigt_keinen_wert(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_settings: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
strip_settings["sacn"]["on_timeout"] = "irgendwas"
await abfragen(hass, freezer)
assert hass.states.get(entity_id_for(hass, "select", "sacn_on_timeout")).state == "unknown"
async def test_aktiv_merker_folgt_dem_strom(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_info: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
entity_id = entity_id_for(hass, "binary_sensor", "sacn_active")
assert hass.states.get(entity_id).state == STATE_OFF
strip_info["sacn"]["active"] = True
await abfragen(hass, freezer)
assert hass.states.get(entity_id).state == STATE_ON
async def test_multicast_merker_ist_von_haus_aus_aus(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
eintrag = er.async_get(hass).async_get(entity_id_for(hass, "binary_sensor", "sacn_multicast"))
assert eintrag.disabled_by is er.RegistryEntryDisabler.INTEGRATION
async def test_firmware_ohne_sacn_laesst_die_entitaeten_unverfuegbar(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
strip_settings: dict[str, Any],
strip_info: dict[str, Any],
) -> None:
"""Die Entitaeten entstehen trotzdem - nach einem Firmware-Update werden
sie beim naechsten Abfragen lebendig, ohne die Integration neu zu laden."""
strip_settings.pop("sacn")
strip_info.pop("sacn")
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
for platform, key in (
("switch", "sacn_enabled"),
("number", "sacn_universe"),
("number", "sacn_start_slot"),
("select", "sacn_on_timeout"),
("binary_sensor", "sacn_active"),
):
entity_id = entity_id_for(hass, platform, key)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE, entity_id
+93
View File
@@ -0,0 +1,93 @@
"""Szenenauswahl."""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.ninstrip.select import NONE_OPTION
from tests.helpers import abfragen, entity_id_for
def preset_id(hass: HomeAssistant) -> str:
return entity_id_for(hass, "select", "preset")
async def test_nur_belegte_szenen_stehen_zur_wahl(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
state = hass.states.get(preset_id(hass))
assert state.attributes["options"] == [
NONE_OPTION,
"Warmweiss",
"Kaltweiss",
"Regenbogen",
"Feuer",
]
assert state.state == "Regenbogen"
async def test_geaenderte_segmente_zeigen_keine_szene_mehr(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_state: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
strip_state["preset_modified"] = True
await abfragen(hass, freezer)
assert hass.states.get(preset_id(hass)).state == NONE_OPTION
async def test_unbekannte_szenennummer_zeigt_keine_szene(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_state: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
strip_state["preset"] = 6 # vorhanden, aber unbelegt
await abfragen(hass, freezer)
assert hass.states.get(preset_id(hass)).state == NONE_OPTION
async def test_namenlose_szene_bekommt_eine_nummer(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_presets: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
strip_presets["presets"][2]["name"] = ""
await abfragen(hass, freezer)
state = hass.states.get(preset_id(hass))
assert "Szene 3" in state.attributes["options"]
assert state.state == "Szene 3"
async def test_szene_waehlen_schaltet_sie_auf(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: preset_id(hass), "option": "Feuer"},
blocking=True,
)
mock_client.apply_preset.assert_awaited_once_with(3)
async def test_die_leere_wahl_schaltet_aus(
hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
"""Es gibt kein 'Szene abwaehlen' - das Gegenstueck ist Ausschalten."""
await hass.services.async_call(
"select",
"select_option",
{ATTR_ENTITY_ID: preset_id(hass), "option": NONE_OPTION},
blocking=True,
)
mock_client.set_state.assert_awaited_once_with(on=False)
mock_client.apply_preset.assert_not_awaited()
+55
View File
@@ -0,0 +1,55 @@
"""Diagnosesensoren."""
from __future__ import annotations
from typing import Any
from freezegun.api import FrozenDateTimeFactory
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry
from tests.helpers import abfragen, entity_id_for
@pytest.mark.parametrize(
("key", "wert"),
[("rssi", "-68"), ("uptime", "1459"), ("free_heap", "156848"), ("power_scale", "100")],
)
async def test_werte(
hass: HomeAssistant, init_integration: MockConfigEntry, key: str, wert: str
) -> None:
registry = er.async_get(hass)
entity_id = entity_id_for(hass, "sensor", key)
eintrag = registry.async_get(entity_id)
assert eintrag is not None
if eintrag.disabled:
# Die drei Diagnosesensoren sind standardmaessig aus; dann gibt es
# keinen Zustand, und mehr ist hier nicht zu pruefen.
assert key in ("rssi", "uptime", "free_heap")
return
assert hass.states.get(entity_id).state == wert
async def test_nur_die_leistungsbegrenzung_ist_von_haus_aus_an(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
registry = er.async_get(hass)
an = [
registry.async_get(entity_id_for(hass, "sensor", key)).disabled_by is None
for key in ("rssi", "uptime", "free_heap", "power_scale")
]
assert an == [False, False, False, True]
async def test_fehlender_wifi_block_ergibt_keinen_wert(
hass: HomeAssistant,
init_integration: MockConfigEntry,
strip_info: dict[str, Any],
freezer: FrozenDateTimeFactory,
) -> None:
"""Ein Geraet im Onboarding meldet keinen strip-Block mit power_scale."""
strip_info.pop("strip")
await abfragen(hass, freezer)
assert hass.states.get(entity_id_for(hass, "sensor", "power_scale")).state == "unknown"
+578
View File
@@ -0,0 +1,578 @@
"""Die Update-Entitaet: Angebot, WLAN-Sicherung, Ablauf der Aktualisierung."""
from __future__ import annotations
import asyncio
from copy import deepcopy
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.ninstrip import update as update_module
from custom_components.ninstrip.api import NinstripBusyError, NinstripConnectionError
from custom_components.ninstrip.const import CONF_DEVICE_BASE_URL, DOMAIN
from custom_components.ninstrip.shared import DATA_SHARED, async_scan_catalog
from custom_components.ninstrip.update import haengt_an_kompiliertem_wlan
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_K1Y, MAC_K1Y, SETTINGS_COMPILED, SETTINGS_STORED, TOKEN
BASE = "http://192.168.1.50:8123"
INSTALLED = "0.1.0-12-g71a57"
NEWER = "0.1.0-47-g71a57"
NEWEST = "0.1.1-rc1"
EVEN_NEWER = "0.1.1-rc2" # newer than NEWEST
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
return MockConfigEntry(
domain=DOMAIN,
title="ninstrip ad4c",
unique_id=MAC_K1Y,
data={CONF_HOST: HOST_K1Y, CONF_TOKEN: TOKEN},
options={CONF_DEVICE_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_nur_gespeichertes_wlan_ist_unabhaengig() -> None:
assert not haengt_an_kompiliertem_wlan(SETTINGS_STORED)
assert haengt_an_kompiliertem_wlan(SETTINGS_COMPILED)
assert haengt_an_kompiliertem_wlan({"wifi": {"ssid": ""}})
assert haengt_an_kompiliertem_wlan({"wifi": {"source": "flash"}})
assert haengt_an_kompiliertem_wlan({})
async def test_ohne_abbilder_kein_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_neueres_abbild_mit_zugangsdaten_wird_angeboten(
hass: HomeAssistant, init_integration, firmware_dir: Path
) -> None:
# Zugleich die Probe auf den eigenen Versionsvergleich: HA haelt
# 0.1.0-12-g71a57 und 0.1.0-47-g71a57 fuer denselben Versionskern und
# vergleicht dann Zeichenketten - damit bliebe das hier aus.
put(firmware_dir, "a.img", version=NEWER, payload=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": "ninstrip-rgbc3",
"version": "0.2.0",
}, # other type that only differs in the suffix
{"device_type": "ninstrip-k1y", "version": "0.1.0-12-gffff0"}, # same version, other hash
{"device_type": "ninstrip-k1y", "version": "0.1.0"}, # older
],
)
async def test_unpassende_abbilder_werden_nicht_angeboten(
hass: HomeAssistant, init_integration, firmware_dir: Path, image: dict[str, str]
) -> None:
put(firmware_dir, "a.img", payload=CONTENT_WITH_SSID, **image)
await rescan(hass)
assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_OFF
async def test_wlan_sicherung_haelt_abbild_ohne_zugangsdaten_zurueck(
hass: HomeAssistant, init_integration, firmware_dir: Path
) -> None:
nocreds = put(firmware_dir, "nocreds.img", version=NEWEST, payload=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": "ninstrip ad4c",
"version": NEWEST,
"device_type": "ninstrip-k1y",
"ssid": "ninIOT",
}
put(firmware_dir, "creds.img", version=NEWER, payload=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_geraet_mit_gespeichertem_wlan_bekommt_auch_abbilder_ohne_zugangsdaten(
hass: HomeAssistant, init_integration, firmware_dir: Path, strip_settings: dict[str, Any]
) -> None:
strip_settings.clear()
strip_settings.update(SETTINGS_STORED)
put(firmware_dir, "nocreds.img", version=NEWEST, payload=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_geraet_ohne_ssid_bekommt_nichts(
hass: HomeAssistant, init_integration, firmware_dir: Path, strip_info: dict[str, Any]
) -> None:
strip_info["wifi"]["ssid"] = ""
await init_integration.runtime_data.async_refresh()
put(firmware_dir, "creds.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_OFF
async def test_unlesbare_einstellungen_lassen_das_bisherige_angebot_stehen(
hass: HomeAssistant, init_integration, firmware_dir: Path, mock_client: MagicMock
) -> None:
mock_client.get_settings.side_effect = NinstripConnectionError("down")
put(firmware_dir, "creds.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_OFF
async def test_aktualisierung_liefert_das_abbild_einmal_und_wartet_auf_die_neue_version(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
init_integration,
mock_client: MagicMock,
strip_info: dict[str, Any],
firmware_dir: Path,
) -> None:
path = put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
client = await hass_client_no_auth()
served: dict[str, Any] = {}
async def geraet_laedt(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
strip_info["fw_version"] = NEWER
strip_info["uptime"] = 4
mock_client.fw_update.side_effect = geraet_laedt
await install(hass)
await hass.async_block_till_done()
assert served["url"].startswith(f"{BASE}/api/ninstrip/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
async def test_ausgeliefert_werden_genau_die_geprueften_bytes(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
init_integration,
mock_client: MagicMock,
strip_info: dict[str, Any],
firmware_dir: Path,
) -> None:
"""F1: the strip must receive exactly the bytes checked (Wi-Fi guard, type, hash) at install
time, even if the file on disk is replaced by something else afterwards - e.g. a build script
that writes the no-credentials variant over the same filename mid-update.
"""
path = put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
verified_bytes = path.read_bytes()
await rescan(hass)
client = await hass_client_no_auth()
served: dict[str, Any] = {}
async def geraet_laedt(url: str) -> None:
# Swap in a different, otherwise-valid image for the same device type right before the
# strip "fetches" it. The Wi-Fi guard already passed for `verified_bytes`; this replacement
# has no credentials at all and must never reach the strip.
path.write_bytes(build_image(version=NEWER, payload=CONTENT_WITHOUT_SSID))
served["url"] = url
response = await client.get(urlsplit(url).path)
served["status"] = response.status
served["body"] = await response.read()
strip_info["fw_version"] = NEWER
strip_info["uptime"] = 4
mock_client.fw_update.side_effect = geraet_laedt
await install(hass)
await hass.async_block_till_done()
assert served["status"] == 200
assert served["body"] == verified_bytes
assert served["body"] != path.read_bytes() # the swapped-in file, not what was served
async def test_katalogdurchlauf_waehrend_der_aktualisierung_prueft_nicht(
hass: HomeAssistant,
init_integration,
mock_client: MagicMock,
strip_info: dict[str, Any],
firmware_dir: Path,
) -> None:
"""M2: a catalog rescan landing while an install is in progress must not evaluate offers
against the (currently flashing) strip - that would mean extra requests to it and could churn
the offer/issue underneath the install.
"""
put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
calls_before_rescan: dict[str, int] = {}
async def geraet_flasht(url: str) -> None:
# a periodic catalog scan happens to land while this install is in progress
before = mock_client.get_settings.await_count
await rescan(hass)
# Measured around the rescan itself, not after the install: once the install is
# done the coordinator polls again anyway, and since the sACN settings became
# entities that poll reads system/settings too. Comparing after the fact would
# count that poll and fail for a reason this test does not care about.
calls_before_rescan["delta"] = mock_client.get_settings.await_count - before
strip_info["fw_version"] = NEWER
strip_info["uptime"] = 4
mock_client.fw_update.side_effect = geraet_flasht
await install(hass)
await hass.async_block_till_done()
assert calls_before_rescan["delta"] == 0
@pytest.mark.parametrize(
("geraet_reaktion", "key", "reason_part"),
[
({"last_ota_error": "hash"}, "update_failed", "hash"),
({"uptime": 3}, "update_failed", INSTALLED), # rebooted, still the old firmware
],
)
async def test_fehler_des_geraets_werden_gemeldet(
hass: HomeAssistant,
init_integration,
mock_client: MagicMock,
strip_info: dict[str, Any],
firmware_dir: Path,
geraet_reaktion: dict[str, Any],
key: str,
reason_part: str,
) -> None:
put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
async def geraet_scheitert(url: str) -> None:
strip_info.update(geraet_reaktion)
mock_client.fw_update.side_effect = geraet_scheitert
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_beschaeftigtes_geraet(
hass: HomeAssistant, init_integration, mock_client: MagicMock, firmware_dir: Path
) -> None:
put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
mock_client.fw_update.side_effect = NinstripBusyError(409, "Update laeuft schon")
with pytest.raises(HomeAssistantError) as raised:
await install(hass)
# "update_busy" meldet nur der eigene Riegel gegen zwei gleichzeitige
# Aufrufe; ein 409 vom Geraet kommt als "device_busy" aus dem Koordinator.
assert raised.value.translation_key == "device_busy"
assert len(hass.data[DATA_SHARED].registry) == 0
async def test_zeitueberschreitung(
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, payload=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_ohne_http_adresse_geht_nichts(
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_K1Y, data={CONF_HOST: HOST_K1Y, 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, payload=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_wlan_wird_unmittelbar_vor_dem_senden_nochmal_geprueft(
hass: HomeAssistant,
init_integration,
mock_client: MagicMock,
firmware_dir: Path,
strip_settings: dict[str, Any],
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Guards against a regression that would let this install slip past the re-check into the
# wait loop: without this it would hang for the full OTA_TIMEOUT_S (12 min) instead of failing
# fast with "wifi_guard".
monkeypatch.setattr(update_module, "OTA_TIMEOUT_S", 0)
strip_settings.clear()
strip_settings.update(SETTINGS_STORED)
put(firmware_dir, "nocreds.img", version=NEWEST, payload=CONTENT_WITHOUT_SSID)
await rescan(hass)
assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_ON
strip_settings["wifi"]["source"] = "compiled" # the strip 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_geaenderte_datei_wird_abgelehnt(
hass: HomeAssistant, init_integration, mock_client: MagicMock, firmware_dir: Path
) -> None:
path = put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
path.write_bytes(build_image(version=NEWEST, payload=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_rueckschritt_wird_abgelehnt(
hass: HomeAssistant,
init_integration,
mock_client: MagicMock,
strip_info: dict[str, Any],
firmware_dir: Path,
) -> None:
"""The strip may have been reflashed by another route since the offer was computed."""
put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
entity_id = entity_id_for(hass, "update", "firmware")
assert hass.states.get(entity_id).attributes[ATTR_LATEST_VERSION] == NEWER
# A newer image also appears on disk, and the strip is reflashed by another route to a version
# newer than our stale offer - but the follow-up evaluation (triggered by the catalog scan)
# fails at get_settings and returns early, so the stale NEWER offer survives untouched.
put(firmware_dir, "b.img", version=EVEN_NEWER, payload=CONTENT_WITH_SSID)
mock_client.get_settings.side_effect = NinstripConnectionError("down")
strip_info["fw_version"] = NEWEST
await rescan(hass)
assert hass.states.get(entity_id).attributes[ATTR_LATEST_VERSION] == NEWER
with pytest.raises(HomeAssistantError) as raised:
await install(hass)
assert raised.value.translation_key == "no_update"
mock_client.fw_update.assert_not_awaited()
async def test_zwei_gleichzeitige_aufrufe_werden_abgewiesen(
hass: HomeAssistant,
init_integration,
mock_client: MagicMock,
strip_info: dict[str, Any],
firmware_dir: Path,
) -> None:
"""A second call arriving in the pre-flag window (before in_progress is set) must be
rejected by our own lock rather than allowed to touch OTA state.
HA's own service-level "already in progress" guard only helps once `_attr_in_progress` is
already True (it is a `cached_property` that HA invalidates whenever `_attr_in_progress` is
set, so it is never stale) - by the time that guard would fire, a second concurrent call is
already blocked by HA itself with translation_key "update_in_progress", not "update_busy".
The real gap I2 describes is *before* that: image verification, `get_info` and `get_settings`
are all awaited before we ever set the flags, and HA's guard does not protect that window. This
test
blocks the first `get_info()` re-check inside that window to reach it deterministically.
"""
put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
entity_id = entity_id_for(hass, "update", "firmware")
entered = asyncio.Event()
release = asyncio.Event()
first_call = True
async def blocking_get_info() -> dict[str, Any]:
nonlocal first_call
if first_call:
first_call = False
entered.set()
await release.wait()
return deepcopy(strip_info)
async def geraet_flasht(url: str) -> None:
strip_info["fw_version"] = NEWER
strip_info["uptime"] = 4
mock_client.get_info.side_effect = blocking_get_info
mock_client.fw_update.side_effect = geraet_flasht
task1 = hass.async_create_task(install(hass), name="install #1")
async with asyncio.timeout(5):
await entered.wait()
# Install #1 is blocked inside its pre-flag get_info() re-check: the lock is held, but
# in_progress/ota_in_progress are still untouched.
assert init_integration.runtime_data.ota_in_progress is False
assert hass.states.get(entity_id).attributes[ATTR_IN_PROGRESS] is False
with pytest.raises(HomeAssistantError) as raised:
await install(hass)
assert raised.value.translation_key == "update_busy"
assert init_integration.runtime_data.ota_in_progress is False
assert hass.states.get(entity_id).attributes[ATTR_IN_PROGRESS] is False
release.set()
async with asyncio.timeout(5):
await task1
assert init_integration.runtime_data.ota_in_progress is False
assert hass.states.get(entity_id).attributes[ATTR_IN_PROGRESS] is False
assert hass.states.get(entity_id).attributes[ATTR_INSTALLED_VERSION] == NEWER
async def test_das_loeschen_des_eintrags_raeumt_den_wlan_hinweis_weg(
hass: HomeAssistant, init_integration, firmware_dir: Path
) -> None:
put(firmware_dir, "nocreds.img", version=NEWEST, payload=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
async def test_firmware_ohne_device_type_bekommt_kein_angebot(
hass: HomeAssistant, init_integration, strip_info: dict[str, Any], firmware_dir: Path
) -> None:
"""Firmware vor v0.1.0 nennt ihre Platine nicht.
Dann bleibt offen, wofuer sie gebaut wurde - und ein K1Y-Abbild auf einer
KL-RGBC3 invertiert das Datensignal nicht, der Strip bliebe dunkel. Lieber
kein Angebot als das falsche.
"""
strip_info.pop("device_type")
strip_info["fw_version"] = "0.1.0-12-g71a57"
await init_integration.runtime_data.async_refresh()
put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
put(
firmware_dir,
"b.img",
device_type="ninstrip-rgbc3",
version=NEWER,
payload=CONTENT_WITH_SSID,
)
await rescan(hass)
assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_OFF
async def test_unvergleichbare_version_bekommt_kein_angebot(
hass: HomeAssistant, init_integration, strip_info: dict[str, Any], firmware_dir: Path
) -> None:
"""Ein Geraet mit dem alten Hash-Stand: welcher neuer ist, weiss niemand."""
strip_info["fw_version"] = "652b2e2-dirty"
await init_integration.runtime_data.async_refresh()
put(firmware_dir, "a.img", version=NEWEST, payload=CONTENT_WITH_SSID)
await rescan(hass)
assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_OFF
async def test_platine_wird_unmittelbar_vor_dem_senden_nochmal_geprueft(
hass: HomeAssistant,
init_integration,
mock_client: MagicMock,
strip_info: dict[str, Any],
firmware_dir: Path,
) -> None:
"""Das Angebot galt fuer die Platine von vorhin.
Meldet das Geraet inzwischen eine andere - etwa weil an derselben Adresse
ein anderes Geraet haengt -, darf das Abbild nicht mehr hinaus: ein
K1Y-Abbild auf einer KL-RGBC3 invertiert das Datensignal nicht.
"""
put(firmware_dir, "a.img", version=NEWER, payload=CONTENT_WITH_SSID)
await rescan(hass)
assert hass.states.get(entity_id_for(hass, "update", "firmware")).state == STATE_ON
strip_info["device_type"] = "ninstrip-rgbc3"
with pytest.raises(HomeAssistantError) as raised:
await install(hass)
assert raised.value.translation_key == "image_invalid"
mock_client.fw_update.assert_not_awaited()
assert len(hass.data[DATA_SHARED].registry) == 0
View File
+86
View File
@@ -0,0 +1,86 @@
"""Ein echter lokaler aiohttp-Server spielt das Geraet; kein Home Assistant im Spiel."""
from __future__ import annotations
import asyncio
from collections.abc import AsyncGenerator, Awaitable, Callable
from dataclasses import dataclass, field
import json
from typing import Any
import aiohttp
from aiohttp import web
from aiohttp.test_utils import TestServer
import pytest
@dataclass
class Call:
"""Eine Anfrage, die das gespielte Geraet bekommen hat."""
method: str
path: str
headers: dict[str, str]
body: Any
@dataclass
class FakeStrip:
"""Vorgegebene Antworten je (METHODE, Pfad); merkt sich jeden Aufruf."""
host: str
port: int
routes: dict[tuple[str, str], tuple[int, str, float]] = field(default_factory=dict)
calls: list[Call] = field(default_factory=list)
def reply(
self,
method: str,
path: str,
status: int = 200,
payload: Any = None,
*,
raw: str | None = None,
delay: float = 0.0,
) -> None:
body = raw if raw is not None else json.dumps(payload)
self.routes[(method.upper(), path)] = (status, body, delay)
@pytest.fixture(autouse=True)
def _enable_sockets(socket_enabled: None) -> None:
"""pytest-homeassistant-custom-component sperrt Sockets; hier wird localhost gebraucht."""
@pytest.fixture
async def strip(aiohttp_server: Callable[..., Awaitable[TestServer]]) -> FakeStrip:
holder: list[FakeStrip] = []
async def handler(request: web.Request) -> web.Response:
fake = holder[0]
text = await request.text()
try:
body: Any = json.loads(text) if text else None
except ValueError:
body = text
fake.calls.append(Call(request.method, request.path, dict(request.headers), body))
route = fake.routes.get((request.method, request.path))
if route is None:
return web.Response(status=404, text="no route")
status, text_out, delay = route
if delay:
await asyncio.sleep(delay)
return web.Response(status=status, text=text_out)
app = web.Application()
app.router.add_route("*", "/{tail:.*}", handler)
server = await aiohttp_server(app)
fake = FakeStrip(server.host, server.port)
holder.append(fake)
return fake
@pytest.fixture
async def session() -> AsyncGenerator[aiohttp.ClientSession]:
async with aiohttp.ClientSession() as client_session:
yield client_session
+33
View File
@@ -0,0 +1,33 @@
"""Kuenstliche ninstrip-OTA-Abbilder: Nutzdaten + 512-Byte-NINL-Anhaenger.
Der Aufbau folgt ninstrip-firmware/tools/mkimage.py. Die Tests bauen sich ihre
Abbilder selbst, statt echte Firmware-Dateien mitzuschleppen.
Anders als bei ninlight liegen die Nutzdaten unverpackt vor: ein ESP32-Abbild
ist nicht gzip-verpackt, die einkompilierte SSID steht als Zeichenkette drin.
"""
from __future__ import annotations
import hashlib
CONTENT_WITH_SSID = b"\x00boot\x00ninIOT\x00ninstrip\x00"
CONTENT_WITHOUT_SSID = b"\x00boot\x00ninstrip\x00"
def build_image(
device_type: str = "ninstrip-k1y",
version: str = "0.1.0-47-g71a57",
*,
payload: bytes = CONTENT_WITH_SSID,
magic: bytes = b"NINL",
fmt: int = 1,
corrupt_hash: bool = False,
) -> bytes:
trailer = bytearray(b"\xff" * 512)
trailer[0:4] = magic
trailer[4] = fmt
trailer[5:21] = device_type.encode().ljust(16, b"\x00")
trailer[21:37] = version.encode().ljust(16, b"\x00")
trailer[37:69] = bytes(32) if corrupt_hash else hashlib.sha256(payload).digest()
return payload + bytes(trailer)
+97
View File
@@ -0,0 +1,97 @@
"""Echte Nutzdaten zweier Geraete. Bewusst keine Grenzwerte - die stehen in den Tests selbst."""
from __future__ import annotations
from typing import Any
TOKEN = "a201883d13a570dd3cf1f3b64b5e1df7"
HOST_K1Y = "192.168.2.238"
MAC_K1Y = "b0:81:84:49:ad:4c"
# Abgefragt an der Stehlampe (KL-LSC5C3WV2-K1Y, SK6812-RGBW, 60 Pixel).
STATE_K1Y: dict[str, Any] = {
"on": True,
"brightness": 100,
"power_on_behaviour": "last",
"fade_in": 500,
"fade_out": 500,
"transition_default": 300,
"wake_on_change": False,
"preset": 2,
"preset_modified": False,
"segment_count": 1,
"pixel_count": 60,
}
INFO_K1Y: dict[str, Any] = {
"product": "ninstrip",
"device_type": "ninstrip-k1y",
"fw_version": "0.1.0-12-g71a57",
"build_date": "2026-09-22",
"mac": MAC_K1Y,
"name": "",
"wifi": {"mode": "sta", "ssid": "ninIOT", "rssi": -68, "ip": HOST_K1Y, "error": None},
"sacn": {"enabled": True, "active": False, "universe": 1, "universes": 1, "multicast": True},
"uptime": 1459,
"free_heap": 156848,
"strip": {
"chipset": "sk6812_rgbw",
"color_order": "grb",
"data_pin": 19,
"pixel_count": 60,
"leds_per_pixel": 1,
"has_white": True,
"inverted": True,
"power_scale": 100,
"max_milliamps": 1800,
},
"render": {"fps": 50, "render_us": 900, "show_us": 1800, "dropped_frames": 0},
"last_ota_error": None,
}
# wifi.source "compiled": das Geraet kennt sein WLAN nur aus der Firmware.
SETTINGS_COMPILED: dict[str, Any] = {
"name": "",
"wifi": {"ssid": "ninIOT", "source": "compiled"},
"auth_all": False,
"sacn": {
"enabled": True,
"universe": 1,
"start_slot": 1,
"on_timeout": "release",
"channels_per_pixel": 3,
},
"strip": {
"chipset": "sk6812_rgbw",
"color_order": "grb",
"data_pin": 19,
"pixel_count": 60,
"leds_per_pixel": 1,
"max_milliamps": 1800,
"supply_volts": 12,
"inverted": True,
},
}
PRESETS_K1Y: dict[str, Any] = {
"presets": [
{"id": 0, "name": "Warmweiss", "used": True, "segment_count": 1},
{"id": 1, "name": "Kaltweiss", "used": True, "segment_count": 1},
{"id": 2, "name": "Regenbogen", "used": True, "segment_count": 1},
{"id": 3, "name": "Feuer", "used": True, "segment_count": 1},
{"id": 4, "name": "", "used": False, "segment_count": 0},
{"id": 5, "name": "", "used": False, "segment_count": 0},
{"id": 6, "name": "", "used": False, "segment_count": 0},
{"id": 7, "name": "", "used": False, "segment_count": 0},
],
"active": 2,
"modified": False,
}
# wifi.source "stored": ueber das Onboarding ins WLAN gebracht, also vom
# Abbild unabhaengig.
SETTINGS_STORED: dict[str, Any] = {
**SETTINGS_COMPILED,
"wifi": {"ssid": "ninIOT", "source": "stored"},
}
+179
View File
@@ -0,0 +1,179 @@
"""HTTP-Client gegen einen echten lokalen Server."""
from __future__ import annotations
import aiohttp
import pytest
from custom_components.ninstrip.api import (
NinstripAuthError,
NinstripBusyError,
NinstripClient,
NinstripConnectionError,
NinstripRequestError,
NinstripResponseError,
)
from tests_unit.conftest import FakeStrip
from tests_unit.samples import INFO_K1Y, PRESETS_K1Y, SETTINGS_COMPILED, STATE_K1Y, TOKEN
def client_for(strip: FakeStrip, session: aiohttp.ClientSession, **kwargs: float) -> NinstripClient:
return NinstripClient(session, strip.host, TOKEN, port=strip.port, **kwargs)
async def test_zustand_schickt_das_token_roh(strip: FakeStrip, session) -> None:
strip.reply("GET", "/api/v1/device/state", payload=STATE_K1Y)
zustand = await client_for(strip, session).get_state()
assert zustand == STATE_K1Y
assert strip.calls[-1].headers["Authorization"] == TOKEN # kein "Bearer"
async def test_set_state_schickt_felder_und_gibt_die_antwort_zurueck(
strip: FakeStrip, session
) -> None:
antwort = {**STATE_K1Y, "brightness": 55} # die Antwort weicht von der Anfrage ab
strip.reply("POST", "/api/v1/device/state", payload=antwort)
ergebnis = await client_for(strip, session).set_state(brightness=78, transition=2500)
assert ergebnis == antwort
assert strip.calls[-1].body == {"brightness": 78, "transition": 2500}
async def test_unbekannte_felder_bleiben_erhalten(strip: FakeStrip, session) -> None:
strip.reply("GET", "/api/v1/device/state", payload={**STATE_K1Y, "kuenftig": 7})
assert (await client_for(strip, session).get_state())["kuenftig"] == 7
async def test_info_settings_und_presets(strip: FakeStrip, session) -> None:
strip.reply("GET", "/api/v1/system/info", payload=INFO_K1Y)
strip.reply("GET", "/api/v1/system/settings", payload=SETTINGS_COMPILED)
strip.reply("GET", "/api/v1/device/presets", payload=PRESETS_K1Y)
client = client_for(strip, session)
assert await client.get_info() == INFO_K1Y
assert await client.get_settings() == SETTINGS_COMPILED
assert await client.get_presets() == PRESETS_K1Y
async def test_sacn_wird_als_settings_typ_geschrieben(strip: FakeStrip, session) -> None:
strip.reply("PUT", "/api/v1/system/settings", payload={"status": "ok"})
await client_for(strip, session).set_sacn(universe=7, start_slot=4)
assert strip.calls[-1].body == {"type": "sacn", "universe": 7, "start_slot": 4}
async def test_szenen_anwenden_und_sichern(strip: FakeStrip, session) -> None:
strip.reply("PUT", "/api/v1/device/presets/3/apply", payload={"status": "ok"})
strip.reply("PUT", "/api/v1/device/presets/3", payload={"status": "ok"})
client = client_for(strip, session)
await client.apply_preset(3)
assert strip.calls[-1].path == "/api/v1/device/presets/3/apply"
await client.save_preset(3, "Abend")
assert strip.calls[-1].body == {"name": "Abend"}
await client.save_preset(3)
assert strip.calls[-1].body == {}
async def test_befehle_liegen_verschachtelt(strip: FakeStrip, session) -> None:
strip.reply("PUT", "/api/v1/system/command", payload={"ok": True})
client = client_for(strip, session)
await client.reboot()
assert strip.calls[-1].body == {"command": {"type": "reboot"}}
strip.reply("PUT", "/api/v1/system/command", status=202, payload={"status": "updating"})
await client.fw_update("http://192.168.1.50:8123/api/ninstrip/firmware/k3y")
assert strip.calls[-1].body == {
"command": {
"type": "fw_update",
"url": "http://192.168.1.50:8123/api/ninstrip/firmware/k3y",
}
}
async def test_401_ist_ein_auth_fehler(strip: FakeStrip, session) -> None:
strip.reply("GET", "/api/v1/system/settings", status=401, payload={"error": "Token fehlt"})
with pytest.raises(NinstripAuthError):
await client_for(strip, session).get_settings()
async def test_409_ist_beschaeftigt_mit_meldung(strip: FakeStrip, session) -> None:
strip.reply(
"PUT", "/api/v1/system/command", status=409, payload={"error": "Update laeuft schon"}
)
with pytest.raises(NinstripBusyError) as err:
await client_for(strip, session).fw_update("http://192.168.1.50:8123/x")
assert err.value.status == 409
assert err.value.message == "Update laeuft schon"
async def test_400_ist_ein_anfragefehler_kein_beschaeftigt(strip: FakeStrip, session) -> None:
strip.reply("POST", "/api/v1/device/state", status=400, payload={"error": "brightness"})
with pytest.raises(NinstripRequestError) as err:
await client_for(strip, session).set_state(brightness=0)
assert not isinstance(err.value, NinstripBusyError)
assert (err.value.status, err.value.message) == (400, "brightness")
async def test_fehlertext_ohne_json_kommt_roh(strip: FakeStrip, session) -> None:
strip.reply("POST", "/api/v1/device/state", status=500, raw="peng")
with pytest.raises(NinstripRequestError) as err:
await client_for(strip, session).set_state(on=True)
assert err.value.message == "peng"
async def test_json_fehlertext_ohne_error_feld_kommt_roh(strip: FakeStrip, session) -> None:
strip.reply("POST", "/api/v1/device/state", status=500, payload={"detail": "zu hell"})
with pytest.raises(NinstripRequestError) as err:
await client_for(strip, session).set_state(on=True)
assert "detail" in err.value.message
async def test_erfolg_ohne_json_ist_ein_antwortfehler(strip: FakeStrip, session) -> None:
strip.reply("GET", "/api/v1/device/state", raw="<html>")
with pytest.raises(NinstripResponseError):
await client_for(strip, session).get_state()
async def test_json_array_als_erfolg_ist_ein_antwortfehler(strip: FakeStrip, session) -> None:
strip.reply("GET", "/api/v1/device/state", raw="[1, 2, 3]")
with pytest.raises(NinstripResponseError):
await client_for(strip, session).get_state()
@pytest.mark.parametrize("fehlt", ["on", "brightness", "segment_count", "pixel_count"])
async def test_zustand_ohne_pflichtfeld_ist_ein_antwortfehler(
strip: FakeStrip, session, fehlt: str
) -> None:
strip.reply(
"GET", "/api/v1/device/state", payload={k: v for k, v in STATE_K1Y.items() if k != fehlt}
)
with pytest.raises(NinstripResponseError):
await client_for(strip, session).get_state()
@pytest.mark.parametrize("fehlt", ["mac", "product", "fw_version"])
async def test_info_ohne_pflichtfeld_ist_ein_antwortfehler(
strip: FakeStrip, session, fehlt: str
) -> None:
strip.reply(
"GET", "/api/v1/system/info", payload={k: v for k, v in INFO_K1Y.items() if k != fehlt}
)
with pytest.raises(NinstripResponseError):
await client_for(strip, session).get_info()
async def test_zeitueberschreitung_ist_ein_verbindungsfehler(strip: FakeStrip, session) -> None:
strip.reply("GET", "/api/v1/device/state", payload=STATE_K1Y, delay=0.5)
with pytest.raises(NinstripConnectionError):
await client_for(strip, session, timeout=0.05).get_state()
async def test_abgewiesene_verbindung_ist_ein_verbindungsfehler(session) -> None:
client = NinstripClient(session, "127.0.0.1", TOKEN, port=1)
with pytest.raises(NinstripConnectionError):
await client.get_state()
async def test_host_gibt_die_eingestellte_adresse(session) -> None:
assert NinstripClient(session, "192.168.2.238", TOKEN, port=3333).host == "192.168.2.238"
async def test_befehl_ohne_antwortkoerper_geht_durch(strip: FakeStrip, session) -> None:
strip.reply("PUT", "/api/v1/system/command", status=204, raw="")
await client_for(strip, session).reboot()
+252
View File
@@ -0,0 +1,252 @@
"""Anhaenger, Versionsreihenfolge, WLAN-Pruefung und der Abbild-Katalog."""
from __future__ import annotations
from pathlib import Path
import pytest
from custom_components.ninstrip import firmware
from custom_components.ninstrip.firmware import (
FirmwareCatalog,
FirmwareVersion,
InvalidImageError,
contains_ssid,
is_newer,
parse_trailer,
parse_version,
)
from tests_unit.images import CONTENT_WITH_SSID, CONTENT_WITHOUT_SSID, build_image
# Streng aufsteigend. Geprueft werden alle Nachbarpaare und alle Paare ueber
# die ganze Liste.
ORDER = [
"0.1.0",
"0.1.0-36-gca9bb",
"0.1.0-47-g71a57",
"0.1.1-rc1",
"0.1.1-rc1-4-gab12",
"0.1.1-rc2",
"0.1.1",
"0.2.0-rc1",
"0.10.0",
]
@pytest.mark.parametrize(("aelter", "neuer"), list(zip(ORDER, ORDER[1:], strict=False))) # noqa: RUF007
def test_nachbarversionen(aelter: str, neuer: str) -> None:
assert is_newer(neuer, aelter)
assert not is_newer(aelter, neuer)
def test_reihenfolge_gilt_ueber_die_ganze_liste() -> None:
for i, aelter in enumerate(ORDER):
for neuer in ORDER[i + 1 :]:
assert is_newer(neuer, aelter), (neuer, aelter)
@pytest.mark.parametrize(
("text", "erwartet"),
[
("0.1.0-36-gca9bb", FirmwareVersion((0, 1, 0), None, 36)),
("0.1.0-38-gd3cc9", FirmwareVersion((0, 1, 0), None, 38)), # genau 15 Zeichen, mit Hash
("0.1.0-36-g", FirmwareVersion((0, 1, 0), None, 36)), # Hash weggekuerzt
("0.1.0-36", FirmwareVersion((0, 1, 0), None, 36)),
("0.1.0-3-gab-dir", FirmwareVersion((0, 1, 0), None, 3)), # "-dirty" auf "-dir" gekuerzt
("0.1.0-dirty", FirmwareVersion((0, 1, 0), None, 0)),
("0.2.0-rc1-12-ga", FirmwareVersion((0, 2, 0), 1, 12)), # 15 Zeichen, Hash auf ein Zeichen
("0.1.1-rc2", FirmwareVersion((0, 1, 1), 2, 0)),
],
)
def test_version_lesen(text: str, erwartet: FirmwareVersion) -> None:
assert parse_version(text) == erwartet
@pytest.mark.parametrize(
"text",
[
"",
# Genau das meldete die Firmware vor dem Tag v0.1.0: git describe
# --always ohne Tags. Daran laesst sich keine Reihenfolge ablesen.
"652b2e2-dirty",
"v0.1.0", # das fuehrende "v" schneidet cmake/version.cmake ab
"0.1",
"0.1.0-",
"0.10.10-rc10-12", # 15 Zeichen, endet auf Ziffern ohne "-g": Zahl kann gekuerzt sein
"0.1.0-36-gXYZ",
"0.1.0 ",
],
)
def test_unlesbare_versionen(text: str) -> None:
assert parse_version(text) is None
def test_gleiche_commitzahl_mit_anderem_hash_ist_nicht_neuer() -> None:
assert not is_newer("0.1.0-36-gffff0", "0.1.0-36-gca9bb")
assert not is_newer("0.1.0-36-gca9bb", "0.1.0-36-gffff0")
def test_eine_unlesbare_seite_ist_nie_neuer() -> None:
assert not is_newer("652b2e2", "0.1.0")
assert not is_newer("0.1.1", "652b2e2")
def test_anhaenger_liest_nul_aufgefuellte_felder() -> None:
assert parse_trailer(build_image("ninstrip-rgbc3", "0.1.0-38-gd3cc9")) == (
"ninstrip-rgbc3",
"0.1.0-38-gd3cc9",
)
@pytest.mark.parametrize(
("data", "grund"),
[
(b"\xff" * 512, "kurz"),
(build_image(magic=b"NINX"), "NINL"),
(build_image(fmt=2), "format"),
(build_image(corrupt_hash=True), "sha256"),
],
)
def test_ungueltige_abbilder(data: bytes, grund: str) -> None:
with pytest.raises(InvalidImageError, match=grund):
parse_trailer(data)
def test_ssid_wird_nul_gerahmt_gefunden() -> None:
assert contains_ssid(build_image(payload=CONTENT_WITH_SSID), "ninIOT")
def test_ssid_fehlt_im_abbild_ohne_zugangsdaten() -> None:
assert not contains_ssid(build_image(payload=CONTENT_WITHOUT_SSID), "ninIOT")
@pytest.mark.parametrize("payload", [b"\x00xninIOT\x00", b"\x00ninIOTx\x00", b"ninIOT"])
def test_ssid_muss_nul_gerahmt_sein(payload: bytes) -> None:
assert not contains_ssid(build_image(payload=payload), "ninIOT")
def test_eine_andere_ssid_passt_nicht() -> None:
assert not contains_ssid(build_image(payload=CONTENT_WITH_SSID), "ninIOT-gast")
def test_leere_ssid_passt_nie() -> None:
assert not contains_ssid(build_image(payload=b"\x00\x00\x00"), "")
def test_die_ssid_im_anhaenger_zaehlt_nicht() -> None:
"""Sonst koennte der Anhaenger selbst die WLAN-Pruefung bestehen lassen."""
abbild = build_image(payload=CONTENT_WITHOUT_SSID)
manipuliert = bytearray(abbild)
manipuliert[-200:-190] = b"\x00ninIOT\x00\x00\x00"
assert not contains_ssid(bytes(manipuliert), "ninIOT")
def _schreiben(verzeichnis: Path, name: str, data: bytes) -> Path:
pfad = verzeichnis / name
pfad.write_bytes(data)
return pfad
def test_durchlauf_legt_das_verzeichnis_an_und_nimmt_nur_gueltiges(tmp_path: Path) -> None:
verzeichnis = tmp_path / "ninstrip_firmware"
katalog = FirmwareCatalog(verzeichnis)
assert katalog.scan() == ()
assert verzeichnis.is_dir()
_schreiben(verzeichnis, "a.img", build_image("ninstrip-k1y", "0.1.0-47-g71a57"))
_schreiben(verzeichnis, "kaputt.img", build_image(corrupt_hash=True))
_schreiben(verzeichnis, "notizen.txt", build_image())
abbilder = katalog.scan()
assert [(i.path.name, i.device_type, i.version) for i in abbilder] == [
("a.img", "ninstrip-k1y", "0.1.0-47-g71a57")
]
assert katalog.images == abbilder
def test_ungueltige_datei_wird_einmal_gemeldet(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
katalog = FirmwareCatalog(tmp_path)
_schreiben(tmp_path, "kaputt.img", build_image(corrupt_hash=True))
katalog.scan()
katalog.scan()
assert sum("kaputt.img" in r.getMessage() for r in caplog.records) == 1
def test_erneuter_durchlauf_sieht_neue_und_geloeschte_dateien(tmp_path: Path) -> None:
katalog = FirmwareCatalog(tmp_path)
alt = _schreiben(tmp_path, "a.img", build_image(version="0.1.0-47-g71a57"))
katalog.scan()
_schreiben(tmp_path, "b.img", build_image(version="0.1.1-rc1"))
alt.unlink()
assert [i.version for i in katalog.scan()] == ["0.1.1-rc1"]
def test_neuere_abbilder_nur_exakt_dieser_platine_neuestes_zuerst(tmp_path: Path) -> None:
katalog = FirmwareCatalog(tmp_path)
_schreiben(tmp_path, "1.img", build_image("ninstrip-k1y", "0.1.0-47-g71a57"))
_schreiben(tmp_path, "2.img", build_image("ninstrip-k1y", "0.1.1-rc1"))
# Neuer, aber fuer die andere Platine: die KL-RGBC3 invertiert das
# Datensignal nicht, die K1Y schon.
_schreiben(tmp_path, "3.img", build_image("ninstrip-rgbc3", "0.2.0"))
_schreiben(tmp_path, "4.img", build_image("ninstrip", "0.5.0")) # Praefix des Typs
_schreiben(tmp_path, "5.img", build_image("ninstrip-k1y", "0.1.0-36-gffff0")) # gleich alt
_schreiben(tmp_path, "6.img", build_image("ninstrip-k1y", "0.1.0")) # aelter
katalog.scan()
neuer = katalog.newer_images("ninstrip-k1y", "0.1.0-36-gca9bb")
assert [(i.device_type, i.version) for i in neuer] == [
("ninstrip-k1y", "0.1.1-rc1"),
("ninstrip-k1y", "0.1.0-47-g71a57"),
]
def test_verify_and_read_liefert_genau_die_geprueften_bytes(tmp_path: Path) -> None:
katalog = FirmwareCatalog(tmp_path)
daten = build_image(version="0.1.0-47-g71a57")
pfad = _schreiben(tmp_path, "a.img", daten)
(abbild,) = katalog.scan()
assert katalog.verify_and_read(abbild) == daten
pfad.write_bytes(build_image(version="0.1.1-rc1"))
assert katalog.verify_and_read(abbild) is None
pfad.unlink()
assert katalog.verify_and_read(abbild) is None
def test_verify_and_read_weist_gleich_grosse_aber_andere_datei_ab(tmp_path: Path) -> None:
"""Gleiche Groesse und gleiche mtime, anderer Inhalt: der Anhaenger entscheidet."""
katalog = FirmwareCatalog(tmp_path)
pfad = _schreiben(tmp_path, "a.img", build_image("ninstrip-k1y", "0.1.0-47-g71a57"))
(abbild,) = katalog.scan()
stat = pfad.stat()
pfad.write_bytes(build_image("ninstrip-k1y", "0.1.0-47-g71a58"))
import os
os.utime(pfad, ns=(stat.st_atime_ns, stat.st_mtime_ns))
assert katalog.verify_and_read(abbild) is None
def test_ssid_pruefung_wird_je_abbild_und_ssid_gemerkt(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
katalog = FirmwareCatalog(tmp_path)
_schreiben(tmp_path, "a.img", build_image(payload=CONTENT_WITH_SSID))
(abbild,) = katalog.scan()
aufrufe: list[str] = []
echt = firmware.contains_ssid
def spion(data: bytes, ssid: str) -> bool:
aufrufe.append(ssid)
return echt(data, ssid)
monkeypatch.setattr(firmware, "contains_ssid", spion)
assert katalog.contains_ssid(abbild, "ninIOT")
assert katalog.contains_ssid(abbild, "ninIOT")
assert not katalog.contains_ssid(abbild, "andere")
assert aufrufe == ["ninIOT", "andere"]
def test_ssid_pruefung_einer_geloeschten_datei_ist_falsch(tmp_path: Path) -> None:
katalog = FirmwareCatalog(tmp_path)
pfad = _schreiben(tmp_path, "a.img", build_image(payload=CONTENT_WITH_SSID))
(abbild,) = katalog.scan()
pfad.unlink()
assert not katalog.contains_ssid(abbild, "ninIOT")
+26
View File
@@ -0,0 +1,26 @@
"""The manifest must agree with the constants the code relies on."""
from __future__ import annotations
import json
from pathlib import Path
from custom_components.ninstrip.const import DOMAIN
MANIFEST = Path(__file__).parents[1] / "custom_components" / "ninstrip" / "manifest.json"
def test_manifest_matches_constants() -> None:
manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
assert manifest["domain"] == DOMAIN
assert manifest["zeroconf"] == [{"type": "_ninstrip._tcp.local."}]
assert manifest["iot_class"] == "local_polling"
assert manifest["config_flow"] is True
assert manifest["requirements"] == []
assert manifest["dependencies"] == ["http"]
def test_manifest_points_to_gitea() -> None:
manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
assert manifest["documentation"] == "https://gitea.zukunpht.de/ninharp/ha-ninstrip"
assert manifest["issue_tracker"] == "https://gitea.zukunpht.de/ninharp/ha-ninstrip/issues"
+46
View File
@@ -0,0 +1,46 @@
"""Pruefung der Adresse, unter der Geraete Home Assistant erreichen."""
from __future__ import annotations
import pytest
from custom_components.ninstrip.urls import firmware_url, validate_device_base_url
@pytest.mark.parametrize(
"url", ["http://192.168.1.50:8123", "http://ha.local:8123/", "http://192.168.1.50"]
)
def test_gueltige_adressen(url: str) -> None:
assert validate_device_base_url(url) is None
@pytest.mark.parametrize(
"url",
[
"https://192.168.1.50:8123",
"http://",
"ftp://ha.local",
"http://user:pass@ha.local:8123",
"http://ha.local:8123/ha",
"http://ha.local:8123?x=1",
"http://ha.local:8123#frag",
"http://ha.local:99999",
"192.168.1.50:8123",
],
)
def test_ungueltige_adressen(url: str) -> None:
assert validate_device_base_url(url) == "invalid_base_url"
def test_laengengrenze_sind_die_255_zeichen_des_geraets() -> None:
laengste = "http://" + "a" * 182
assert len(firmware_url(laengste, "k" * 43)) == 255
assert validate_device_base_url(laengste) is None
assert validate_device_base_url(laengste + "a") == "base_url_too_long"
def test_adresse_wird_ohne_doppelten_schraegstrich_zusammengesetzt() -> None:
assert (
firmware_url("http://ha.local:8123/", "abc")
== "http://ha.local:8123/api/ninstrip/firmware/abc"
)