Michael Sauer
9eb7adcdd7
Michaels Befund: die Integration ist nicht dasselbe wie die von ninlight. Verglichen und die Luecken geschlossen - alle Felder gibt es in der Firmware schon, sie waren nur nicht bedienbar: * number: fade_in, fade_out, transition_default in Millisekunden, mit den Grenzen der Firmware (10 s, 10 s, 60 s). * select: Einschaltverhalten (last, on, off). * switch: bei Aenderung einschalten. * light: Uebergangszeit beim Ein- und Ausschalten. HA rechnet in Sekunden, das Geraet in Millisekunden; laengeres als 60 s wird gekappt, sonst wiese das Geraet die ganze Anfrage mit 400 ab. Ohne Angabe wird kein Feld geschickt, sonst ueberschriebe jeder Schaltbefehl die Vorgabe. * sensor: letzter Update-Fehler und Startzeit (Zeitstempel statt einer Laufzeit, die jede Abfrage weiterlaeuft), beide ab Werk aus. * diagnostics.py: Diagnosedaten zum Herunterladen, ohne Token, Adressen, MAC und SSID. * config_flow: Neu-Konfigurieren von Adresse und Token, mit Gegenprobe der MAC - ein Tippfehler in der Adresse soll nicht still den falschen Eintrag umbiegen. * Hinweis in den Reparaturen, wenn ein Geraet noch auf dem oeffentlichen Werks-Token steht. Nicht uebernommen: die Auswahl sacn_layout. ninstrip ist immer rgb, ein Strip hat keinen Masterkanal - eine Auswahl mit einer Option waere nur Zierde. 241 Tests, 97 %. Die Kappung der Uebergangszeit per Mutation gegengeprueft.
299 lines
12 KiB
Python
299 lines
12 KiB
Python
"""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}
|
|
|
|
|
|
async def test_neu_konfigurieren_aendert_adresse_und_token(
|
|
hass: HomeAssistant, mock_client: MagicMock
|
|
) -> None:
|
|
eintrag = bestehender_eintrag(hass)
|
|
ergebnis = await eintrag.start_reconfigure_flow(hass)
|
|
assert ergebnis["step_id"] == "reconfigure"
|
|
ergebnis = await hass.config_entries.flow.async_configure(
|
|
ergebnis["flow_id"], {CONF_HOST: " 192.168.2.99 ", CONF_TOKEN: NEUES_TOKEN}
|
|
)
|
|
assert ergebnis["type"] is FlowResultType.ABORT
|
|
assert ergebnis["reason"] == "reconfigure_successful"
|
|
assert eintrag.data == {CONF_HOST: "192.168.2.99", CONF_TOKEN: NEUES_TOKEN}
|
|
|
|
|
|
async def test_neu_konfigurieren_lehnt_ein_anderes_geraet_ab(
|
|
hass: HomeAssistant, mock_client: MagicMock, strip_info: dict[str, Any]
|
|
) -> None:
|
|
"""Sonst biegt ein Tippfehler in der Adresse still den falschen Eintrag um."""
|
|
eintrag = bestehender_eintrag(hass)
|
|
strip_info["mac"] = "b0:81:84:49:ff:ff"
|
|
ergebnis = await eintrag.start_reconfigure_flow(hass)
|
|
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"] == "wrong_device"
|
|
assert eintrag.data[CONF_HOST] == HOST_K1Y
|