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.
270 lines
11 KiB
Python
270 lines
11 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}
|