ha-ninstrip/tests/test_update.py

"""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