Add Home Assistant integration scaffold
This commit is contained in:
@@ -1 +1,75 @@
|
||||
"""Paquet de l'intégration Arkteos."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from .client import ArkteosClient
|
||||
from .const import CONF_HOST, CONF_PORT, DOMAIN, PLATFORMS
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
|
||||
def _get_client(hass: HomeAssistant, entry: ConfigEntry) -> ArkteosClient | None:
|
||||
"""Retourne le client partagé, depuis runtime_data ou le repli structuré."""
|
||||
|
||||
client = getattr(entry, "runtime_data", None)
|
||||
if isinstance(client, ArkteosClient):
|
||||
return client
|
||||
return hass.data.get(DOMAIN, {}).get(entry.entry_id)
|
||||
|
||||
|
||||
def _store_client(hass: HomeAssistant, entry: ConfigEntry, client: ArkteosClient) -> None:
|
||||
"""Stocke le client une fois par config entry."""
|
||||
|
||||
if hasattr(entry, "runtime_data"):
|
||||
entry.runtime_data = client
|
||||
return
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = client
|
||||
|
||||
|
||||
def _remove_client(hass: HomeAssistant, entry: ConfigEntry) -> None:
|
||||
"""Supprime les données d'exécution de la config entry."""
|
||||
|
||||
if hasattr(entry, "runtime_data"):
|
||||
entry.runtime_data = None
|
||||
domain_data = hass.data.get(DOMAIN)
|
||||
if domain_data is not None:
|
||||
domain_data.pop(entry.entry_id, None)
|
||||
if not domain_data:
|
||||
hass.data.pop(DOMAIN, None)
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Crée et partage le client Arkteos pour cette config entry."""
|
||||
|
||||
if _get_client(hass, entry) is not None:
|
||||
return True
|
||||
|
||||
client = ArkteosClient(entry.data[CONF_HOST], entry.data[CONF_PORT])
|
||||
_store_client(hass, entry, client)
|
||||
await client.start()
|
||||
try:
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
except Exception:
|
||||
await client.stop()
|
||||
_remove_client(hass, entry)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Décharge les plateformes et arrête le client partagé."""
|
||||
|
||||
client = _get_client(hass, entry)
|
||||
if client is None:
|
||||
return True
|
||||
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
if not unload_ok:
|
||||
return False
|
||||
await client.stop()
|
||||
_remove_client(hass, entry)
|
||||
return True
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
"""Binary sensor de disponibilité du proxy Arkteos."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.binary_sensor import BinarySensorDeviceClass, BinarySensorEntity
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import _get_client
|
||||
from .const import DOMAIN
|
||||
from .entity import ArkteosEntity
|
||||
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities: AddEntitiesCallback) -> None:
|
||||
"""Ajoute l'unique entité de connexion de cette config entry."""
|
||||
|
||||
client = _get_client(hass, entry)
|
||||
if client is None:
|
||||
return
|
||||
async_add_entities([ArkteosConnectionBinarySensor(client)])
|
||||
|
||||
|
||||
class ArkteosConnectionBinarySensor(ArkteosEntity, BinarySensorEntity):
|
||||
"""Expose la disponibilité du flux lu depuis le proxy."""
|
||||
|
||||
_attr_translation_key = "connection"
|
||||
_attr_unique_id = "arkteos_zuran4_connection"
|
||||
_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
|
||||
_attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, "arkteos_zuran4")},
|
||||
manufacturer="Arkteos",
|
||||
model="Zuran 4",
|
||||
name="PAC Arkteos Zuran 4",
|
||||
)
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool:
|
||||
"""Reflète la disponibilité du client sans effectuer de lecture réseau."""
|
||||
|
||||
return self._client.available
|
||||
@@ -0,0 +1,145 @@
|
||||
"""Flux de configuration par interface utilisateur de l'intégration Arkteos."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .client import ArkteosClient
|
||||
from .const import CONF_HOST, CONF_PORT, DEFAULT_PORT, DOMAIN
|
||||
|
||||
|
||||
VALIDATION_TIMEOUT = 10.0
|
||||
|
||||
|
||||
class CannotConnect(Exception):
|
||||
"""La connexion au proxy a échoué."""
|
||||
|
||||
|
||||
class ValidationTimeout(Exception):
|
||||
"""Aucune trame valide n'a été reçue dans le délai imparti."""
|
||||
|
||||
|
||||
class ArkteosConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Gère la création et la reconfiguration des config entries Arkteos."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_user(self, user_input: dict[str, Any] | None = None) -> FlowResult:
|
||||
"""Affiche et traite le formulaire initial."""
|
||||
|
||||
if user_input is not None:
|
||||
data, errors = await self._async_validate_input(user_input)
|
||||
if not errors:
|
||||
if self._entry_exists(data[CONF_HOST], data[CONF_PORT]):
|
||||
return self.async_abort(reason="already_configured")
|
||||
await self.async_set_unique_id(self._unique_id(data))
|
||||
self._abort_if_unique_id_configured()
|
||||
try:
|
||||
await self._async_validate_proxy(data)
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except ValidationTimeout:
|
||||
errors["base"] = "timeout"
|
||||
except Exception:
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return self.async_create_entry(title=self._title(data), data=data)
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=self._schema(data if "data" in locals() else None), errors=errors
|
||||
)
|
||||
|
||||
return self.async_show_form(step_id="user", data_schema=self._schema())
|
||||
|
||||
async def async_step_reconfigure(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Permet la modification de l'hôte et du port d'une entrée existante."""
|
||||
|
||||
entry = self._get_reconfigure_entry()
|
||||
if user_input is not None:
|
||||
data, errors = await self._async_validate_input(user_input)
|
||||
if not errors:
|
||||
if self._entry_exists(data[CONF_HOST], data[CONF_PORT], exclude_entry_id=entry.entry_id):
|
||||
errors["base"] = "already_configured"
|
||||
else:
|
||||
try:
|
||||
await self._async_validate_proxy(data)
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except ValidationTimeout:
|
||||
errors["base"] = "timeout"
|
||||
except Exception:
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return self.async_update_reload_and_abort(
|
||||
entry, data_updates=data, title=self._title(data)
|
||||
)
|
||||
return self.async_show_form(
|
||||
step_id="reconfigure", data_schema=self._schema(data if "data" in locals() else entry.data), errors=errors
|
||||
)
|
||||
|
||||
return self.async_show_form(step_id="reconfigure", data_schema=self._schema(entry.data))
|
||||
|
||||
async def _async_validate_input(
|
||||
self, user_input: dict[str, Any]
|
||||
) -> tuple[dict[str, Any], dict[str, str]]:
|
||||
errors: dict[str, str] = {}
|
||||
host = str(user_input.get(CONF_HOST, "")).strip()
|
||||
if not host:
|
||||
errors[CONF_HOST] = "invalid_host"
|
||||
try:
|
||||
port = int(user_input.get(CONF_PORT, DEFAULT_PORT))
|
||||
except (TypeError, ValueError):
|
||||
errors[CONF_PORT] = "invalid_port"
|
||||
port = DEFAULT_PORT
|
||||
if not 1 <= port <= 65535:
|
||||
errors[CONF_PORT] = "invalid_port"
|
||||
return {CONF_HOST: host, CONF_PORT: port}, errors
|
||||
|
||||
async def _async_validate_proxy(self, data: dict[str, Any]) -> None:
|
||||
client = ArkteosClient(data[CONF_HOST], data[CONF_PORT])
|
||||
try:
|
||||
await client.start()
|
||||
if not await client.wait_until_connected(VALIDATION_TIMEOUT):
|
||||
raise CannotConnect
|
||||
deadline = asyncio.get_running_loop().time() + VALIDATION_TIMEOUT
|
||||
while client.last_valid_frame is None:
|
||||
if asyncio.get_running_loop().time() >= deadline:
|
||||
raise ValidationTimeout
|
||||
if client.last_error is not None and not client.connected:
|
||||
raise CannotConnect
|
||||
await asyncio.sleep(0.05)
|
||||
finally:
|
||||
await client.stop()
|
||||
|
||||
def _entry_exists(self, host: str, port: int, exclude_entry_id: str | None = None) -> bool:
|
||||
return any(
|
||||
entry.entry_id != exclude_entry_id
|
||||
and entry.data.get(CONF_HOST) == host
|
||||
and entry.data.get(CONF_PORT) == port
|
||||
for entry in self._async_current_entries()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _title(data: dict[str, Any]) -> str:
|
||||
return f"Arkteos {data[CONF_HOST]}:{data[CONF_PORT]}"
|
||||
|
||||
@staticmethod
|
||||
def _unique_id(data: dict[str, Any]) -> str:
|
||||
return f"{data[CONF_HOST]}:{data[CONF_PORT]}"
|
||||
|
||||
@staticmethod
|
||||
def _schema(defaults: dict[str, Any] | None = None) -> vol.Schema:
|
||||
defaults = defaults or {}
|
||||
return vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST, default=defaults.get(CONF_HOST, "")): str,
|
||||
vol.Required(CONF_PORT, default=defaults.get(CONF_PORT, DEFAULT_PORT)): vol.Coerce(int),
|
||||
}
|
||||
)
|
||||
@@ -0,0 +1,11 @@
|
||||
"""Constantes de l'intégration Arkteos."""
|
||||
|
||||
from homeassistant.const import Platform
|
||||
|
||||
|
||||
DOMAIN = "arkteos"
|
||||
DEFAULT_PORT = 9641
|
||||
CONF_HOST = "host"
|
||||
CONF_PORT = "port"
|
||||
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR]
|
||||
RUNTIME_DATA = "runtime_data"
|
||||
@@ -0,0 +1,37 @@
|
||||
"""Entités Home Assistant partageant le client Arkteos."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
from .client import ArkteosClient
|
||||
|
||||
|
||||
class ArkteosEntity(Entity):
|
||||
"""Base d'entité sans lecture réseau dans ses propriétés."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(self, client: ArkteosClient) -> None:
|
||||
self._client = client
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""Reflète la disponibilité déterminée par le client partagé."""
|
||||
|
||||
return self._client.available
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Abonne l'entité aux changements de disponibilité."""
|
||||
|
||||
await super().async_added_to_hass()
|
||||
self._client.add_availability_callback(self._handle_availability)
|
||||
|
||||
async def async_will_remove_from_hass(self) -> None:
|
||||
"""Retire le callback avant la destruction de l'entité."""
|
||||
|
||||
self._client.remove_availability_callback(self._handle_availability)
|
||||
await super().async_will_remove_from_hass()
|
||||
|
||||
def _handle_availability(self, _available: bool) -> None:
|
||||
self.async_write_ha_state()
|
||||
@@ -0,0 +1,10 @@
|
||||
{
|
||||
"domain": "arkteos",
|
||||
"name": "Arkteos",
|
||||
"config_flow": true,
|
||||
"iot_class": "local_push",
|
||||
"version": "0.1.0",
|
||||
"requirements": [],
|
||||
"documentation": "REPLACE_WITH_GITEA_REPOSITORY_URL",
|
||||
"codeowners": []
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
{
|
||||
"title": "Arkteos",
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Configurer Arkteos",
|
||||
"description": "Renseignez le proxy TCP Arkteos.",
|
||||
"data": {
|
||||
"host": "Hôte du proxy",
|
||||
"port": "Port du proxy"
|
||||
}
|
||||
},
|
||||
"reconfigure": {
|
||||
"title": "Reconfigurer Arkteos",
|
||||
"description": "Modifiez l'hôte ou le port du proxy.",
|
||||
"data": {
|
||||
"host": "Hôte du proxy",
|
||||
"port": "Port du proxy"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Impossible de se connecter au proxy.",
|
||||
"invalid_host": "L'hôte est invalide.",
|
||||
"invalid_port": "Le port doit être compris entre 1 et 65535.",
|
||||
"timeout": "Aucune trame valide n'a été reçue avant expiration du délai.",
|
||||
"unknown": "Erreur inattendue."
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "Ce proxy est déjà configuré.",
|
||||
"reconfigure_successful": "Arkteos a été reconfiguré."
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"binary_sensor": {
|
||||
"connection": {
|
||||
"name": "PAC connectée"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
{
|
||||
"title": "Arkteos",
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Configurer Arkteos",
|
||||
"description": "Renseignez le proxy TCP Arkteos.",
|
||||
"data": {
|
||||
"host": "Hôte du proxy",
|
||||
"port": "Port du proxy"
|
||||
}
|
||||
},
|
||||
"reconfigure": {
|
||||
"title": "Reconfigurer Arkteos",
|
||||
"description": "Modifiez l'hôte ou le port du proxy.",
|
||||
"data": {
|
||||
"host": "Hôte du proxy",
|
||||
"port": "Port du proxy"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Impossible de se connecter au proxy.",
|
||||
"invalid_host": "L'hôte est invalide.",
|
||||
"invalid_port": "Le port doit être compris entre 1 et 65535.",
|
||||
"timeout": "Aucune trame valide n'a été reçue avant expiration du délai.",
|
||||
"unknown": "Erreur inattendue."
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "Ce proxy est déjà configuré.",
|
||||
"reconfigure_successful": "Arkteos a été reconfiguré."
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"binary_sensor": {
|
||||
"connection": {
|
||||
"name": "PAC connectée"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
[pytest]
|
||||
asyncio_mode = auto
|
||||
asyncio_default_fixture_loop_scope = function
|
||||
@@ -0,0 +1 @@
|
||||
pytest-homeassistant-custom-component==0.13.205
|
||||
@@ -0,0 +1,69 @@
|
||||
"""Tests de l'unique binary sensor Arkteos."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import Mock
|
||||
|
||||
from custom_components.arkteos.binary_sensor import ArkteosConnectionBinarySensor
|
||||
from custom_components.arkteos.const import DOMAIN
|
||||
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self, available: bool = False) -> None:
|
||||
self.available = available
|
||||
self.callbacks: list[object] = []
|
||||
|
||||
def add_availability_callback(self, callback) -> None:
|
||||
self.callbacks.append(callback)
|
||||
|
||||
def remove_availability_callback(self, callback) -> None:
|
||||
self.callbacks.remove(callback)
|
||||
|
||||
|
||||
def test_binary_sensor_is_created() -> None:
|
||||
assert ArkteosConnectionBinarySensor(FakeClient()) is not None
|
||||
|
||||
|
||||
def test_unique_id_is_stable() -> None:
|
||||
assert ArkteosConnectionBinarySensor(FakeClient()).unique_id == "arkteos_zuran4_connection"
|
||||
|
||||
|
||||
def test_device_info_is_correct() -> None:
|
||||
info = ArkteosConnectionBinarySensor(FakeClient()).device_info
|
||||
assert info["identifiers"] == {(DOMAIN, "arkteos_zuran4")}
|
||||
assert info["manufacturer"] == "Arkteos"
|
||||
assert info["model"] == "Zuran 4"
|
||||
|
||||
|
||||
def test_is_on_when_client_is_available() -> None:
|
||||
assert ArkteosConnectionBinarySensor(FakeClient(True)).is_on
|
||||
|
||||
|
||||
def test_is_off_when_client_is_unavailable() -> None:
|
||||
assert not ArkteosConnectionBinarySensor(FakeClient(False)).is_on
|
||||
|
||||
|
||||
def test_availability_callback_writes_state() -> None:
|
||||
client = FakeClient()
|
||||
entity = ArkteosConnectionBinarySensor(client)
|
||||
entity.async_write_ha_state = Mock()
|
||||
entity._handle_availability(True)
|
||||
entity.async_write_ha_state.assert_called_once()
|
||||
|
||||
|
||||
def test_callback_is_removed_before_entity_removal() -> None:
|
||||
client = FakeClient()
|
||||
entity = ArkteosConnectionBinarySensor(client)
|
||||
client.add_availability_callback(entity._handle_availability)
|
||||
client.remove_availability_callback(entity._handle_availability)
|
||||
assert client.callbacks == []
|
||||
|
||||
|
||||
def test_no_polling_method_is_declared() -> None:
|
||||
entity = ArkteosConnectionBinarySensor(FakeClient())
|
||||
assert not hasattr(entity, "async_update")
|
||||
|
||||
|
||||
def test_binary_sensor_is_read_only() -> None:
|
||||
source = ArkteosConnectionBinarySensor.__module__
|
||||
assert source == "custom_components.arkteos.binary_sensor"
|
||||
@@ -0,0 +1,199 @@
|
||||
"""Tests hors ligne du flux de configuration Arkteos."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import pytest
|
||||
from homeassistant.config_entries import SOURCE_RECONFIGURE, SOURCE_USER
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.arkteos.const import CONF_HOST, CONF_PORT, DEFAULT_PORT, DOMAIN
|
||||
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("enable_custom_integrations")
|
||||
|
||||
|
||||
class FakeClient:
|
||||
"""Client factice pour valider le flux sans ouvrir de socket."""
|
||||
|
||||
connected_result = True
|
||||
valid_frame = True
|
||||
instances: list["FakeClient"] = []
|
||||
|
||||
def __init__(self, _host: str, _port: int) -> None:
|
||||
self.connected = False
|
||||
self.last_valid_frame = None
|
||||
self.last_error = None
|
||||
self.started = False
|
||||
self.stopped = False
|
||||
self.__class__.instances.append(self)
|
||||
|
||||
async def start(self) -> None:
|
||||
self.started = True
|
||||
self.connected = self.__class__.connected_result
|
||||
if self.__class__.valid_frame:
|
||||
self.last_valid_frame = object()
|
||||
|
||||
async def stop(self) -> None:
|
||||
self.stopped = True
|
||||
|
||||
async def wait_until_connected(self, _timeout: float | None = None) -> bool:
|
||||
return self.__class__.connected_result
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_client(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from custom_components.arkteos import config_flow
|
||||
|
||||
FakeClient.connected_result = True
|
||||
FakeClient.valid_frame = True
|
||||
FakeClient.instances.clear()
|
||||
monkeypatch.setattr(config_flow, "ArkteosClient", FakeClient)
|
||||
|
||||
|
||||
async def _start_user_flow(hass):
|
||||
return await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER})
|
||||
|
||||
|
||||
async def test_user_form_is_shown(hass) -> None:
|
||||
result = await _start_user_flow(hass)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
|
||||
async def test_default_port_is_9641(hass) -> None:
|
||||
result = await _start_user_flow(hass)
|
||||
assert result["data_schema"]({CONF_HOST: "proxy.local"})[CONF_PORT] == DEFAULT_PORT
|
||||
|
||||
|
||||
async def test_empty_host_is_refused(hass) -> None:
|
||||
result = await _start_user_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "", CONF_PORT: DEFAULT_PORT}
|
||||
)
|
||||
assert result["errors"][CONF_HOST] == "invalid_host"
|
||||
|
||||
|
||||
async def test_port_zero_is_refused(hass) -> None:
|
||||
result = await _start_user_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: 0}
|
||||
)
|
||||
assert result["errors"][CONF_PORT] == "invalid_port"
|
||||
|
||||
|
||||
async def test_port_above_range_is_refused(hass) -> None:
|
||||
result = await _start_user_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: 65536}
|
||||
)
|
||||
assert result["errors"][CONF_PORT] == "invalid_port"
|
||||
|
||||
|
||||
async def test_cannot_connect_is_reported(hass) -> None:
|
||||
FakeClient.connected_result = False
|
||||
result = await _start_user_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT}
|
||||
)
|
||||
assert result["errors"]["base"] == "cannot_connect"
|
||||
assert FakeClient.instances[-1].stopped
|
||||
|
||||
|
||||
async def test_validation_timeout_is_reported(hass, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from custom_components.arkteos import config_flow
|
||||
|
||||
FakeClient.valid_frame = False
|
||||
monkeypatch.setattr(config_flow, "VALIDATION_TIMEOUT", 0.001)
|
||||
result = await _start_user_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT}
|
||||
)
|
||||
assert result["errors"]["base"] == "timeout"
|
||||
assert FakeClient.instances[-1].stopped
|
||||
|
||||
|
||||
async def test_connection_and_valid_frame_create_entry(hass) -> None:
|
||||
result = await _start_user_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT}
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "Arkteos proxy.local:9641"
|
||||
|
||||
|
||||
async def test_created_entry_contains_host_and_port(hass) -> None:
|
||||
result = await _start_user_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: 9999}
|
||||
)
|
||||
assert result["data"] == {CONF_HOST: "proxy.local", CONF_PORT: 9999}
|
||||
|
||||
|
||||
async def test_duplicate_host_and_port_are_refused(hass) -> None:
|
||||
entry = MockConfigEntry(domain=DOMAIN, data={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT})
|
||||
entry.add_to_hass(hass)
|
||||
result = await _start_user_flow(hass)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "proxy.local", CONF_PORT: DEFAULT_PORT}
|
||||
)
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_reconfigure_succeeds(hass, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from custom_components.arkteos import config_flow
|
||||
|
||||
entry = MockConfigEntry(domain=DOMAIN, data={CONF_HOST: "old.local", CONF_PORT: DEFAULT_PORT})
|
||||
entry.add_to_hass(hass)
|
||||
update_entry = Mock(wraps=hass.config_entries.async_update_entry)
|
||||
reload_entry = AsyncMock(return_value=True)
|
||||
|
||||
def create_fake_client(host: str, port: int) -> FakeClient:
|
||||
client = FakeClient(host, port)
|
||||
client.wait_until_connected = AsyncMock(wraps=client.wait_until_connected)
|
||||
return client
|
||||
|
||||
client_factory = Mock(side_effect=create_fake_client)
|
||||
monkeypatch.setattr(hass.config_entries, "async_update_entry", update_entry)
|
||||
monkeypatch.setattr(hass.config_entries, "async_reload", reload_entry)
|
||||
monkeypatch.setattr(config_flow, "ArkteosClient", client_factory)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_RECONFIGURE, "entry_id": entry.entry_id}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "new.local", CONF_PORT: 9999}
|
||||
)
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "reconfigure_successful"
|
||||
assert update_entry.call_count == 1
|
||||
_, update_kwargs = update_entry.call_args
|
||||
assert update_kwargs["entry"] is entry
|
||||
assert update_kwargs["data"] == {CONF_HOST: "new.local", CONF_PORT: 9999}
|
||||
assert update_kwargs["title"] == "Arkteos new.local:9999"
|
||||
reload_entry.assert_awaited_once_with(entry.entry_id)
|
||||
client_factory.assert_called_once_with("new.local", 9999)
|
||||
assert len(FakeClient.instances) == 1
|
||||
client = FakeClient.instances[0]
|
||||
assert client.started
|
||||
client.wait_until_connected.assert_awaited_once()
|
||||
assert client.last_valid_frame is not None
|
||||
assert client.stopped
|
||||
assert not any(
|
||||
task.get_name() == "arkteos-client" and not task.done()
|
||||
for task in __import__("asyncio").all_tasks()
|
||||
)
|
||||
|
||||
|
||||
async def test_invalid_reconfigure_is_refused(hass) -> None:
|
||||
entry = MockConfigEntry(domain=DOMAIN, data={CONF_HOST: "old.local", CONF_PORT: DEFAULT_PORT})
|
||||
entry.add_to_hass(hass)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_RECONFIGURE, "entry_id": entry.entry_id}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={CONF_HOST: "", CONF_PORT: DEFAULT_PORT}
|
||||
)
|
||||
assert result["errors"][CONF_HOST] == "invalid_host"
|
||||
@@ -0,0 +1,126 @@
|
||||
"""Tests du cycle de vie de la config entry Arkteos."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.arkteos.const import CONF_HOST, CONF_PORT, DOMAIN
|
||||
|
||||
|
||||
class FakeClient:
|
||||
instances: list["FakeClient"] = []
|
||||
|
||||
def __init__(self, _host: str, _port: int) -> None:
|
||||
self.started = False
|
||||
self.stopped = False
|
||||
self.available = False
|
||||
self._callbacks: list[object] = []
|
||||
self.__class__.instances.append(self)
|
||||
|
||||
async def start(self) -> None:
|
||||
self.started = True
|
||||
|
||||
async def stop(self) -> None:
|
||||
self.stopped = True
|
||||
|
||||
def add_availability_callback(self, callback) -> None:
|
||||
self._callbacks.append(callback)
|
||||
|
||||
def remove_availability_callback(self, callback) -> None:
|
||||
self._callbacks.remove(callback)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_client(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
import custom_components.arkteos as integration
|
||||
|
||||
FakeClient.instances.clear()
|
||||
monkeypatch.setattr(integration, "ArkteosClient", FakeClient)
|
||||
|
||||
|
||||
def _entry() -> MockConfigEntry:
|
||||
return MockConfigEntry(domain=DOMAIN, data={CONF_HOST: "proxy.local", CONF_PORT: 9641})
|
||||
|
||||
|
||||
async def test_setup_creates_one_client(hass) -> None:
|
||||
import custom_components.arkteos as integration
|
||||
|
||||
entry = _entry()
|
||||
hass.config_entries.async_forward_entry_setups = AsyncMock()
|
||||
assert await integration.async_setup_entry(hass, entry)
|
||||
assert len(FakeClient.instances) == 1
|
||||
|
||||
|
||||
async def test_setup_starts_client(hass) -> None:
|
||||
import custom_components.arkteos as integration
|
||||
|
||||
hass.config_entries.async_forward_entry_setups = AsyncMock()
|
||||
assert await integration.async_setup_entry(hass, _entry())
|
||||
assert FakeClient.instances[0].started
|
||||
|
||||
|
||||
async def test_unload_stops_client(hass) -> None:
|
||||
import custom_components.arkteos as integration
|
||||
|
||||
entry = _entry()
|
||||
hass.config_entries.async_forward_entry_setups = AsyncMock()
|
||||
hass.config_entries.async_unload_platforms = AsyncMock(return_value=True)
|
||||
await integration.async_setup_entry(hass, entry)
|
||||
assert await integration.async_unload_entry(hass, entry)
|
||||
assert FakeClient.instances[0].stopped
|
||||
|
||||
|
||||
async def test_unload_removes_runtime_data(hass) -> None:
|
||||
import custom_components.arkteos as integration
|
||||
|
||||
entry = _entry()
|
||||
hass.config_entries.async_forward_entry_setups = AsyncMock()
|
||||
hass.config_entries.async_unload_platforms = AsyncMock(return_value=True)
|
||||
await integration.async_setup_entry(hass, entry)
|
||||
await integration.async_unload_entry(hass, entry)
|
||||
assert integration._get_client(hass, entry) is None
|
||||
|
||||
|
||||
async def test_platform_failure_stops_client(hass) -> None:
|
||||
import custom_components.arkteos as integration
|
||||
|
||||
hass.config_entries.async_forward_entry_setups = AsyncMock(side_effect=RuntimeError("platform"))
|
||||
assert not await integration.async_setup_entry(hass, _entry())
|
||||
assert FakeClient.instances[0].stopped
|
||||
|
||||
|
||||
async def test_double_setup_keeps_single_client(hass) -> None:
|
||||
import custom_components.arkteos as integration
|
||||
|
||||
entry = _entry()
|
||||
hass.config_entries.async_forward_entry_setups = AsyncMock()
|
||||
await integration.async_setup_entry(hass, entry)
|
||||
await integration.async_setup_entry(hass, entry)
|
||||
assert len(FakeClient.instances) == 1
|
||||
|
||||
|
||||
async def test_double_unload_is_safe(hass) -> None:
|
||||
import custom_components.arkteos as integration
|
||||
|
||||
entry = _entry()
|
||||
hass.config_entries.async_forward_entry_setups = AsyncMock()
|
||||
hass.config_entries.async_unload_platforms = AsyncMock(return_value=True)
|
||||
await integration.async_setup_entry(hass, entry)
|
||||
assert await integration.async_unload_entry(hass, entry)
|
||||
assert await integration.async_unload_entry(hass, entry)
|
||||
|
||||
|
||||
def test_no_polling_platform_is_declared() -> None:
|
||||
from custom_components.arkteos.const import PLATFORMS
|
||||
|
||||
assert len(PLATFORMS) == 1
|
||||
|
||||
|
||||
def test_no_mqtt_dependency_in_manifest() -> None:
|
||||
from pathlib import Path
|
||||
|
||||
manifest = (Path(__file__).parents[1] / "custom_components" / "arkteos" / "manifest.json").read_text()
|
||||
assert "mqtt" not in manifest.lower()
|
||||
Reference in New Issue
Block a user