Michael Sauer
2c77b2d325
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.
175 lines
6.4 KiB
Python
175 lines
6.4 KiB
Python
"""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
|