evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add matter integration BETA (#83064)

Browse source 
* Add matter base (#79372)

Co-authored-by: Marcel van der Veldt <m.vanderveldt@outlook.com>

* Add matter server add-on flow (#82698)

* Add matter server add-on flow

* Fix stale error argument

* Clean docstrings

* Use localhost as default address

* Add matter websocket api foundation (#82848)

* Add matter config entry add-on management (#82865)

* Use matter refactored server/client library (#83003)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Bump python-matter-server to 1.0.6 (#83059)

* Extend matter websocket api (#82948)

* Extend matter websocket api

* Finish docstring

* Fix pin type

* Adjust api after new client

* Adjust api to frontend for now

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Marcel van der Veldt 2022-12-01 20:44:56 +01:00 • committed by Paulus Schoutsen
parent 1bf5925cdf
commit 095cc77bf9
32 changed files with 4627 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.coveragerc
View file

@ -724,6 +724,9 @@ omit =
homeassistant/components/map/*
homeassistant/components/mastodon/notify.py
homeassistant/components/matrix/*
homeassistant/components/matter/__init__.py
homeassistant/components/matter/adapter.py
homeassistant/components/matter/entity.py
homeassistant/components/meater/__init__.py
homeassistant/components/meater/const.py
homeassistant/components/meater/sensor.py

1
.strict-typing
View file

@ -179,6 +179,7 @@ homeassistant.components.logger.*
homeassistant.components.lookin.*
homeassistant.components.luftdaten.*
homeassistant.components.mailbox.*
homeassistant.components.matter.*
homeassistant.components.media_player.*
homeassistant.components.media_source.*
homeassistant.components.metoffice.*

2
CODEOWNERS
View file

@ -666,6 +666,8 @@ build.json @home-assistant/supervisor
/tests/components/lyric/ @timmo001
/homeassistant/components/mastodon/ @fabaff
/homeassistant/components/matrix/ @tinloaf
/homeassistant/components/matter/ @MartinHjelmare @marcelveldt
/tests/components/matter/ @MartinHjelmare @marcelveldt
/homeassistant/components/mazda/ @bdr99
/tests/components/mazda/ @bdr99
/homeassistant/components/meater/ @Sotolotl @emontnemery

351
homeassistant/components/matter/__init__.py Normal file
View file

@ -0,0 +1,351 @@
"""The Matter integration."""
from __future__ import annotations
import asyncio
from typing import cast
import async_timeout
from matter_server.client import MatterClient
from matter_server.client.exceptions import (
CannotConnect,
FailedCommand,
InvalidServerVersion,
)
import voluptuous as vol
from homeassistant.components.hassio import AddonError, AddonManager, AddonState
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_URL, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import Event, HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.issue_registry import (
IssueSeverity,
async_create_issue,
async_delete_issue,
)
from homeassistant.helpers.service import async_register_admin_service
from .adapter import MatterAdapter
from .addon import get_addon_manager
from .api import async_register_api
from .const import CONF_INTEGRATION_CREATED_ADDON, CONF_USE_ADDON, DOMAIN, LOGGER
from .device_platform import DEVICE_PLATFORM
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Matter from a config entry."""
if use_addon := entry.data.get(CONF_USE_ADDON):
await _async_ensure_addon_running(hass, entry)
matter_client = MatterClient(entry.data[CONF_URL], async_get_clientsession(hass))
try:
await matter_client.connect()
except CannotConnect as err:
raise ConfigEntryNotReady("Failed to connect to matter server") from err
except InvalidServerVersion as err:
if use_addon:
addon_manager = _get_addon_manager(hass)
addon_manager.async_schedule_update_addon(catch_error=True)
else:
async_create_issue(
hass,
DOMAIN,
"invalid_server_version",
is_fixable=False,
severity=IssueSeverity.ERROR,
translation_key="invalid_server_version",
)
raise ConfigEntryNotReady(f"Invalid server version: {err}") from err
except Exception as err:
matter_client.logger.exception("Failed to connect to matter server")
raise ConfigEntryNotReady(
"Unknown error connecting to the Matter server"
) from err
else:
async_delete_issue(hass, DOMAIN, "invalid_server_version")
async def on_hass_stop(event: Event) -> None:
"""Handle incoming stop event from Home Assistant."""
await matter_client.disconnect()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, on_hass_stop)
)
# register websocket api
async_register_api(hass)
# launch the matter client listen task in the background
# use the init_ready event to keep track if it did initialize successfully
init_ready = asyncio.Event()
listen_task = asyncio.create_task(matter_client.start_listening(init_ready))
try:
async with async_timeout.timeout(30):
await init_ready.wait()
except asyncio.TimeoutError as err:
listen_task.cancel()
raise ConfigEntryNotReady("Matter client not ready") from err
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}
_async_init_services(hass)
# we create an intermediate layer (adapter) which keeps track of our nodes
# and discovery of platform entities from the node's attributes
matter = MatterAdapter(hass, matter_client, entry)
hass.data[DOMAIN][entry.entry_id] = matter
# forward platform setup to all platforms in the discovery schema
await hass.config_entries.async_forward_entry_setups(entry, DEVICE_PLATFORM)
# start discovering of node entities as task
asyncio.create_task(matter.setup_nodes())
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, DEVICE_PLATFORM)
if unload_ok:
matter: MatterAdapter = hass.data[DOMAIN].pop(entry.entry_id)
await matter.matter_client.disconnect()
if entry.data.get(CONF_USE_ADDON) and entry.disabled_by:
addon_manager: AddonManager = get_addon_manager(hass)
LOGGER.debug("Stopping Matter Server add-on")
try:
await addon_manager.async_stop_addon()
except AddonError as err:
LOGGER.error("Failed to stop the Matter Server add-on: %s", err)
return False
return unload_ok
async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Config entry is being removed."""
if not entry.data.get(CONF_INTEGRATION_CREATED_ADDON):
return
addon_manager: AddonManager = get_addon_manager(hass)
try:
await addon_manager.async_stop_addon()
except AddonError as err:
LOGGER.error(err)
return
try:
await addon_manager.async_create_backup()
except AddonError as err:
LOGGER.error(err)
return
try:
await addon_manager.async_uninstall_addon()
except AddonError as err:
LOGGER.error(err)
async def async_remove_config_entry_device(
hass: HomeAssistant, config_entry: ConfigEntry, device_entry: dr.DeviceEntry
) -> bool:
"""Remove a config entry from a device."""
unique_id = None
for ident in device_entry.identifiers:
if ident[0] == DOMAIN:
unique_id = ident[1]
break
if not unique_id:
return True
matter: MatterAdapter = hass.data[DOMAIN][config_entry.entry_id]
for node in await matter.matter_client.get_nodes():
if node.unique_id == unique_id:
await matter.matter_client.remove_node(node.node_id)
break
return True
@callback
def get_matter(hass: HomeAssistant) -> MatterAdapter:
"""Return MatterAdapter instance."""
# NOTE: This assumes only one Matter connection/fabric can exist.
# Shall we support connecting to multiple servers in the client or by config entries?
# In case of the config entry we need to fix this.
matter: MatterAdapter = next(iter(hass.data[DOMAIN].values()))
return matter
@callback
def _async_init_services(hass: HomeAssistant) -> None:
"""Init services."""
async def commission(call: ServiceCall) -> None:
"""Handle commissioning."""
matter_client = get_matter(hass).matter_client
try:
await matter_client.commission_with_code(call.data["code"])
except FailedCommand as err:
raise HomeAssistantError(str(err)) from err
async_register_admin_service(
hass,
DOMAIN,
"commission",
commission,
vol.Schema({"code": str}),
)
async def accept_shared_device(call: ServiceCall) -> None:
"""Accept a shared device."""
matter_client = get_matter(hass).matter_client
try:
await matter_client.commission_on_network(call.data["pin"])
except FailedCommand as err:
raise HomeAssistantError(str(err)) from err
async_register_admin_service(
hass,
DOMAIN,
"accept_shared_device",
accept_shared_device,
vol.Schema({"pin": vol.Coerce(int)}),
)
async def set_wifi(call: ServiceCall) -> None:
"""Handle set wifi creds."""
matter_client = get_matter(hass).matter_client
try:
await matter_client.set_wifi_credentials(
call.data["ssid"], call.data["password"]
)
except FailedCommand as err:
raise HomeAssistantError(str(err)) from err
async_register_admin_service(
hass,
DOMAIN,
"set_wifi",
set_wifi,
vol.Schema(
{
"ssid": str,
"password": str,
}
),
)
async def set_thread_dataset(call: ServiceCall) -> None:
"""Handle set Thread creds."""
matter_client = get_matter(hass).matter_client
thread_dataset = bytes.fromhex(call.data["dataset"])
try:
await matter_client.set_thread_operational_dataset(thread_dataset)
except FailedCommand as err:
raise HomeAssistantError(str(err)) from err
async_register_admin_service(
hass,
DOMAIN,
"set_thread",
set_thread_dataset,
vol.Schema({"dataset": str}),
)
async def _node_id_from_ha_device_id(ha_device_id: str) -> int | None:
"""Get node id from ha device id."""
dev_reg = dr.async_get(hass)
device = dev_reg.async_get(ha_device_id)
if device is None:
return None
matter_id = next(
(
identifier
for identifier in device.identifiers
if identifier[0] == DOMAIN
),
None,
)
if not matter_id:
return None
unique_id = matter_id[1]
matter_client = get_matter(hass).matter_client
# This could be more efficient
for node in await matter_client.get_nodes():
if node.unique_id == unique_id:
return cast(int, node.node_id)
return None
async def open_commissioning_window(call: ServiceCall) -> None:
"""Open commissioning window on specific node."""
node_id = await _node_id_from_ha_device_id(call.data["device_id"])
if node_id is None:
raise HomeAssistantError("This is not a Matter device")
matter_client = get_matter(hass).matter_client
# We are sending device ID .
try:
await matter_client.open_commissioning_window(node_id)
except FailedCommand as err:
raise HomeAssistantError(str(err)) from err
async_register_admin_service(
hass,
DOMAIN,
"open_commissioning_window",
open_commissioning_window,
vol.Schema({"device_id": str}),
)
async def _async_ensure_addon_running(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Ensure that Matter Server add-on is installed and running."""
addon_manager = _get_addon_manager(hass)
try:
addon_info = await addon_manager.async_get_addon_info()
except AddonError as err:
raise ConfigEntryNotReady(err) from err
addon_state = addon_info.state
if addon_state == AddonState.NOT_INSTALLED:
addon_manager.async_schedule_install_setup_addon(
addon_info.options,
catch_error=True,
)
raise ConfigEntryNotReady
if addon_state == AddonState.NOT_RUNNING:
addon_manager.async_schedule_start_addon(catch_error=True)
raise ConfigEntryNotReady
@callback
def _get_addon_manager(hass: HomeAssistant) -> AddonManager:
"""Ensure that Matter Server add-on is updated and running.
May only be used as part of async_setup_entry above.
"""
addon_manager: AddonManager = get_addon_manager(hass)
if addon_manager.task_in_progress():
raise ConfigEntryNotReady
return addon_manager

141
homeassistant/components/matter/adapter.py Normal file
View file

@ -0,0 +1,141 @@
"""Matter to Home Assistant adapter."""
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING
from chip.clusters import Objects as all_clusters
from matter_server.common.models.node_device import AbstractMatterNodeDevice
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .device_platform import DEVICE_PLATFORM
if TYPE_CHECKING:
from matter_server.client import MatterClient
from matter_server.common.models.node import MatterNode
class MatterAdapter:
"""Connect Matter into Home Assistant."""
def __init__(
self,
hass: HomeAssistant,
matter_client: MatterClient,
config_entry: ConfigEntry,
) -> None:
"""Initialize the adapter."""
self.matter_client = matter_client
self.hass = hass
self.config_entry = config_entry
self.logger = logging.getLogger(__name__)
self.platform_handlers: dict[Platform, AddEntitiesCallback] = {}
self._platforms_set_up = asyncio.Event()
def register_platform_handler(
self, platform: Platform, add_entities: AddEntitiesCallback
) -> None:
"""Register a platform handler."""
self.platform_handlers[platform] = add_entities
if len(self.platform_handlers) == len(DEVICE_PLATFORM):
self._platforms_set_up.set()
async def setup_nodes(self) -> None:
"""Set up all existing nodes."""
await self._platforms_set_up.wait()
for node in await self.matter_client.get_nodes():
await self._setup_node(node)
async def _setup_node(self, node: MatterNode) -> None:
"""Set up an node."""
self.logger.debug("Setting up entities for node %s", node.node_id)
bridge_unique_id: str | None = None
if node.aggregator_device_type_instance is not None:
node_info = node.root_device_type_instance.get_cluster(all_clusters.Basic)
self._create_device_registry(
node_info, node_info.nodeLabel or "Hub device", None
)
bridge_unique_id = node_info.uniqueID
for node_device in node.node_devices:
self._setup_node_device(node_device, bridge_unique_id)
def _create_device_registry(
self,
info: all_clusters.Basic | all_clusters.BridgedDeviceBasic,
name: str,
bridge_unique_id: str | None,
) -> None:
"""Create a device registry entry."""
dr.async_get(self.hass).async_get_or_create(
name=name,
config_entry_id=self.config_entry.entry_id,
identifiers={(DOMAIN, info.uniqueID)},
hw_version=info.hardwareVersionString,
sw_version=info.softwareVersionString,
manufacturer=info.vendorName,
model=info.productName,
via_device=(DOMAIN, bridge_unique_id) if bridge_unique_id else None,
)
def _setup_node_device(
self, node_device: AbstractMatterNodeDevice, bridge_unique_id: str | None
) -> None:
"""Set up a node device."""
node = node_device.node()
basic_info = node_device.device_info()
device_type_instances = node_device.device_type_instances()
name = basic_info.nodeLabel
if not name and device_type_instances:
name = f"{device_type_instances[0].device_type.__doc__[:-1]} {node.node_id}"
self._create_device_registry(basic_info, name, bridge_unique_id)
for instance in device_type_instances:
created = False
for platform, devices in DEVICE_PLATFORM.items():
entity_descriptions = devices.get(instance.device_type)
if entity_descriptions is None:
continue
if not isinstance(entity_descriptions, list):
entity_descriptions = [entity_descriptions]
entities = []
for entity_description in entity_descriptions:
self.logger.debug(
"Creating %s entity for %s (%s)",
platform,
instance.device_type.__name__,
hex(instance.device_type.device_type),
)
entities.append(
entity_description.entity_cls(
self.matter_client,
node_device,
instance,
entity_description,
)
)
self.platform_handlers[platform](entities)
created = True
if not created:
self.logger.warning(
"Found unsupported device %s (%s)",
type(instance).__name__,
hex(instance.device_type.device_type),
)

17
homeassistant/components/matter/addon.py Normal file
View file

@ -0,0 +1,17 @@
"""Provide add-on management."""
from __future__ import annotations
from homeassistant.components.hassio import AddonManager
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.singleton import singleton
from .const import ADDON_SLUG, DOMAIN, LOGGER
DATA_ADDON_MANAGER = f"{DOMAIN}_addon_manager"
@singleton(DATA_ADDON_MANAGER)
@callback
def get_addon_manager(hass: HomeAssistant) -> AddonManager:
"""Get the add-on manager."""
return AddonManager(hass, LOGGER, "Matter Server", ADDON_SLUG)

152
homeassistant/components/matter/api.py Normal file
View file

@ -0,0 +1,152 @@
"""Handle websocket api for Matter."""
from __future__ import annotations
from collections.abc import Callable
from functools import wraps
from typing import Any
from matter_server.client.exceptions import FailedCommand
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.components.websocket_api import ActiveConnection
from homeassistant.core import HomeAssistant, callback
from .adapter import MatterAdapter
from .const import DOMAIN
ID = "id"
TYPE = "type"
@callback
def async_register_api(hass: HomeAssistant) -> None:
"""Register all of our api endpoints."""
websocket_api.async_register_command(hass, websocket_commission)
websocket_api.async_register_command(hass, websocket_commission_on_network)
websocket_api.async_register_command(hass, websocket_set_thread_dataset)
websocket_api.async_register_command(hass, websocket_set_wifi_credentials)
def async_get_matter_adapter(func: Callable) -> Callable:
"""Decorate function to get the MatterAdapter."""
@wraps(func)
async def _get_matter(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Provide the Matter client to the function."""
matter: MatterAdapter = next(iter(hass.data[DOMAIN].values()))
await func(hass, connection, msg, matter)
return _get_matter
def async_handle_failed_command(func: Callable) -> Callable:
"""Decorate function to handle FailedCommand and send relevant error."""
@wraps(func)
async def async_handle_failed_command_func(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict[str, Any],
*args: Any,
**kwargs: Any,
) -> None:
"""Handle FailedCommand within function and send relevant error."""
try:
await func(hass, connection, msg, *args, **kwargs)
except FailedCommand as err:
connection.send_error(msg[ID], err.error_code, err.args[0])
return async_handle_failed_command_func
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required(TYPE): "matter/commission",
vol.Required("code"): str,
}
)
@websocket_api.async_response
@async_handle_failed_command
@async_get_matter_adapter
async def websocket_commission(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict[str, Any],
matter: MatterAdapter,
) -> None:
"""Add a device to the network and commission the device."""
await matter.matter_client.commission_with_code(msg["code"])
connection.send_result(msg[ID])
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required(TYPE): "matter/commission_on_network",
vol.Required("pin"): int,
}
)
@websocket_api.async_response
@async_handle_failed_command
@async_get_matter_adapter
async def websocket_commission_on_network(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict[str, Any],
matter: MatterAdapter,
) -> None:
"""Commission a device already on the network."""
await matter.matter_client.commission_on_network(msg["pin"])
connection.send_result(msg[ID])
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required(TYPE): "matter/set_thread",
vol.Required("thread_operation_dataset"): str,
}
)
@websocket_api.async_response
@async_handle_failed_command
@async_get_matter_adapter
async def websocket_set_thread_dataset(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict[str, Any],
matter: MatterAdapter,
) -> None:
"""Set thread dataset."""
await matter.matter_client.set_thread_operational_dataset(
msg["thread_operation_dataset"]
)
connection.send_result(msg[ID])
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required(TYPE): "matter/set_wifi_credentials",
vol.Required("network_name"): str,
vol.Required("password"): str,
}
)
@websocket_api.async_response
@async_handle_failed_command
@async_get_matter_adapter
async def websocket_set_wifi_credentials(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict[str, Any],
matter: MatterAdapter,
) -> None:
"""Set WiFi credentials for a device."""
await matter.matter_client.set_wifi_credentials(
ssid=msg["network_name"], credentials=msg["password"]
)
connection.send_result(msg[ID])

325
homeassistant/components/matter/config_flow.py Normal file
View file

@ -0,0 +1,325 @@
"""Config flow for Matter integration."""
from __future__ import annotations
import asyncio
from typing import Any
from matter_server.client import MatterClient
from matter_server.client.exceptions import CannotConnect, InvalidServerVersion
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.hassio import (
AddonError,
AddonInfo,
AddonManager,
AddonState,
HassioServiceInfo,
is_hassio,
)
from homeassistant.const import CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from homeassistant.helpers import aiohttp_client
from .addon import get_addon_manager
from .const import (
ADDON_SLUG,
CONF_INTEGRATION_CREATED_ADDON,
CONF_USE_ADDON,
DOMAIN,
LOGGER,
)
ADDON_SETUP_TIMEOUT = 5
ADDON_SETUP_TIMEOUT_ROUNDS = 40
DEFAULT_URL = "ws://localhost:5580/ws"
ON_SUPERVISOR_SCHEMA = vol.Schema({vol.Optional(CONF_USE_ADDON, default=True): bool})
def get_manual_schema(user_input: dict[str, Any]) -> vol.Schema:
"""Return a schema for the manual step."""
default_url = user_input.get(CONF_URL, DEFAULT_URL)
return vol.Schema({vol.Required(CONF_URL, default=default_url): str})
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> None:
"""Validate the user input allows us to connect."""
client = MatterClient(data[CONF_URL], aiohttp_client.async_get_clientsession(hass))
await client.connect()
def build_ws_address(host: str, port: int) -> str:
"""Return the websocket address."""
return f"ws://{host}:{port}/ws"
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Matter."""
VERSION = 1
def __init__(self) -> None:
"""Set up flow instance."""
self.ws_address: str | None = None
# If we install the add-on we should uninstall it on entry remove.
self.integration_created_addon = False
self.install_task: asyncio.Task | None = None
self.start_task: asyncio.Task | None = None
self.use_addon = False
async def async_step_install_addon(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Install Matter Server add-on."""
if not self.install_task:
self.install_task = self.hass.async_create_task(self._async_install_addon())
return self.async_show_progress(
step_id="install_addon", progress_action="install_addon"
)
try:
await self.install_task
except AddonError as err:
self.install_task = None
LOGGER.error(err)
return self.async_show_progress_done(next_step_id="install_failed")
self.integration_created_addon = True
self.install_task = None
return self.async_show_progress_done(next_step_id="start_addon")
async def async_step_install_failed(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Add-on installation failed."""
return self.async_abort(reason="addon_install_failed")
async def _async_install_addon(self) -> None:
"""Install the Matter Server add-on."""
addon_manager: AddonManager = get_addon_manager(self.hass)
try:
await addon_manager.async_schedule_install_addon()
finally:
# Continue the flow after show progress when the task is done.
self.hass.async_create_task(
self.hass.config_entries.flow.async_configure(flow_id=self.flow_id)
)
async def _async_get_addon_discovery_info(self) -> dict:
"""Return add-on discovery info."""
addon_manager: AddonManager = get_addon_manager(self.hass)
try:
discovery_info_config = await addon_manager.async_get_addon_discovery_info()
except AddonError as err:
LOGGER.error(err)
raise AbortFlow("addon_get_discovery_info_failed") from err
return discovery_info_config
async def async_step_start_addon(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Start Matter Server add-on."""
if not self.start_task:
self.start_task = self.hass.async_create_task(self._async_start_addon())
return self.async_show_progress(
step_id="start_addon", progress_action="start_addon"
)
try:
await self.start_task
except (CannotConnect, AddonError, AbortFlow) as err:
self.start_task = None
LOGGER.error(err)
return self.async_show_progress_done(next_step_id="start_failed")
self.start_task = None
return self.async_show_progress_done(next_step_id="finish_addon_setup")
async def async_step_start_failed(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Add-on start failed."""
return self.async_abort(reason="addon_start_failed")
async def _async_start_addon(self) -> None:
"""Start the Matter Server add-on."""
addon_manager: AddonManager = get_addon_manager(self.hass)
try:
await addon_manager.async_schedule_start_addon()
# Sleep some seconds to let the add-on start properly before connecting.
for _ in range(ADDON_SETUP_TIMEOUT_ROUNDS):
await asyncio.sleep(ADDON_SETUP_TIMEOUT)
try:
if not (ws_address := self.ws_address):
discovery_info = await self._async_get_addon_discovery_info()
ws_address = self.ws_address = build_ws_address(
discovery_info["host"], discovery_info["port"]
)
await validate_input(self.hass, {CONF_URL: ws_address})
except (AbortFlow, CannotConnect) as err:
LOGGER.debug(
"Add-on not ready yet, waiting %s seconds: %s",
ADDON_SETUP_TIMEOUT,
err,
)
else:
break
else:
raise CannotConnect("Failed to start Matter Server add-on: timeout")
finally:
# Continue the flow after show progress when the task is done.
self.hass.async_create_task(
self.hass.config_entries.flow.async_configure(flow_id=self.flow_id)
)
async def _async_get_addon_info(self) -> AddonInfo:
"""Return Matter Server add-on info."""
addon_manager: AddonManager = get_addon_manager(self.hass)
try:
addon_info: AddonInfo = await addon_manager.async_get_addon_info()
except AddonError as err:
LOGGER.error(err)
raise AbortFlow("addon_info_failed") from err
return addon_info
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if is_hassio(self.hass):
return await self.async_step_on_supervisor()
return await self.async_step_manual()
async def async_step_manual(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a manual configuration."""
if user_input is None:
return self.async_show_form(
step_id="manual", data_schema=get_manual_schema({})
)
errors = {}
try:
await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidServerVersion:
errors["base"] = "invalid_server_version"
except Exception: # pylint: disable=broad-except
LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
self.ws_address = user_input[CONF_URL]
return await self._async_create_entry_or_abort()
return self.async_show_form(
step_id="manual", data_schema=get_manual_schema(user_input), errors=errors
)
async def async_step_hassio(self, discovery_info: HassioServiceInfo) -> FlowResult:
"""Receive configuration from add-on discovery info.
This flow is triggered by the Matter Server add-on.
"""
if discovery_info.slug != ADDON_SLUG:
return self.async_abort(reason="not_matter_addon")
await self._async_handle_discovery_without_unique_id()
self.ws_address = build_ws_address(
discovery_info.config["host"], discovery_info.config["port"]
)
return await self.async_step_hassio_confirm()
async def async_step_hassio_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm the add-on discovery."""
if user_input is not None:
return await self.async_step_on_supervisor(
user_input={CONF_USE_ADDON: True}
)
return self.async_show_form(step_id="hassio_confirm")
async def async_step_on_supervisor(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle logic when on Supervisor host."""
if user_input is None:
return self.async_show_form(
step_id="on_supervisor", data_schema=ON_SUPERVISOR_SCHEMA
)
if not user_input[CONF_USE_ADDON]:
return await self.async_step_manual()
self.use_addon = True
addon_info = await self._async_get_addon_info()
if addon_info.state == AddonState.RUNNING:
return await self.async_step_finish_addon_setup()
if addon_info.state == AddonState.NOT_RUNNING:
return await self.async_step_start_addon()
return await self.async_step_install_addon()
async def async_step_finish_addon_setup(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Prepare info needed to complete the config entry."""
if not self.ws_address:
discovery_info = await self._async_get_addon_discovery_info()
ws_address = self.ws_address = build_ws_address(
discovery_info["host"], discovery_info["port"]
)
# Check that we can connect to the address.
try:
await validate_input(self.hass, {CONF_URL: ws_address})
except CannotConnect:
return self.async_abort(reason="cannot_connect")
return await self._async_create_entry_or_abort()
async def _async_create_entry_or_abort(self) -> FlowResult:
"""Return a config entry for the flow or abort if already configured."""
assert self.ws_address is not None
if existing_config_entries := self._async_current_entries():
config_entry = existing_config_entries[0]
self.hass.config_entries.async_update_entry(
config_entry,
data={
**config_entry.data,
CONF_URL: self.ws_address,
CONF_USE_ADDON: self.use_addon,
CONF_INTEGRATION_CREATED_ADDON: self.integration_created_addon,
},
title=self.ws_address,
)
await self.hass.config_entries.async_reload(config_entry.entry_id)
raise AbortFlow("reconfiguration_successful")
# Abort any other flows that may be in progress
for progress in self._async_in_progress():
self.hass.config_entries.flow.async_abort(progress["flow_id"])
return self.async_create_entry(
title=self.ws_address,
data={
CONF_URL: self.ws_address,
CONF_USE_ADDON: self.use_addon,
CONF_INTEGRATION_CREATED_ADDON: self.integration_created_addon,
},
)

10
homeassistant/components/matter/const.py Normal file
View file

@ -0,0 +1,10 @@
"""Constants for the Matter integration."""
import logging
ADDON_SLUG = "core_matter_server"
CONF_INTEGRATION_CREATED_ADDON = "integration_created_addon"
CONF_USE_ADDON = "use_addon"
DOMAIN = "matter"
LOGGER = logging.getLogger(__package__)

24
homeassistant/components/matter/device_platform.py Normal file
View file

@ -0,0 +1,24 @@
"""All mappings of Matter devices to Home Assistant platforms."""
from __future__ import annotations
from typing import TYPE_CHECKING
from homeassistant.const import Platform
from .light import DEVICE_ENTITY as LIGHT_DEVICE_ENTITY
if TYPE_CHECKING:
from matter_server.common.models.device_types import DeviceType
from .entity import MatterEntityDescriptionBaseClass
DEVICE_PLATFORM: dict[
Platform,
dict[
type[DeviceType],
MatterEntityDescriptionBaseClass | list[MatterEntityDescriptionBaseClass],
],
] = {
Platform.LIGHT: LIGHT_DEVICE_ENTITY,
}

118
homeassistant/components/matter/entity.py Normal file
View file

@ -0,0 +1,118 @@
"""Matter entity base class."""
from __future__ import annotations
from abc import abstractmethod
from collections.abc import Callable
from dataclasses import dataclass
import logging
from typing import TYPE_CHECKING, Any
from matter_server.common.models.device_type_instance import MatterDeviceTypeInstance
from matter_server.common.models.events import EventType
from matter_server.common.models.node_device import AbstractMatterNodeDevice
from homeassistant.core import callback
from homeassistant.helpers.entity import DeviceInfo, Entity, EntityDescription
from .const import DOMAIN
if TYPE_CHECKING:
from matter_server.client import MatterClient
from matter_server.common.models.node import MatterAttribute
LOGGER = logging.getLogger(__name__)
@dataclass
class MatterEntityDescription:
"""Mixin to map a matter device to a Home Assistant entity."""
entity_cls: type[MatterEntity]
subscribe_attributes: tuple
@dataclass
class MatterEntityDescriptionBaseClass(EntityDescription, MatterEntityDescription):
"""For typing a base class that inherits from both entity descriptions."""
class MatterEntity(Entity):
"""Entity class for Matter devices."""
entity_description: MatterEntityDescriptionBaseClass
_attr_should_poll = False
_attr_has_entity_name = True
def __init__(
self,
matter_client: MatterClient,
node_device: AbstractMatterNodeDevice,
device_type_instance: MatterDeviceTypeInstance,
entity_description: MatterEntityDescriptionBaseClass,
) -> None:
"""Initialize the entity."""
self.matter_client = matter_client
self._node_device = node_device
self._device_type_instance = device_type_instance
self.entity_description = entity_description
node = device_type_instance.node
self._unsubscribes: list[Callable] = []
# for fast lookups we create a mapping to the attribute paths
self._attributes_map: dict[type, str] = {}
self._attr_unique_id = f"{matter_client.server_info.compressed_fabric_id}-{node.unique_id}-{device_type_instance.endpoint}-{device_type_instance.device_type.device_type}"
@property
def device_info(self) -> DeviceInfo | None:
"""Return device info for device registry."""
return {"identifiers": {(DOMAIN, self._node_device.device_info().uniqueID)}}
async def async_added_to_hass(self) -> None:
"""Handle being added to Home Assistant."""
await super().async_added_to_hass()
# Subscribe to attribute updates.
for attr_cls in self.entity_description.subscribe_attributes:
if matter_attr := self.get_matter_attribute(attr_cls):
self._attributes_map[attr_cls] = matter_attr.path
self._unsubscribes.append(
self.matter_client.subscribe(
self._on_matter_event,
EventType.ATTRIBUTE_UPDATED,
self._device_type_instance.node.node_id,
matter_attr.path,
)
)
continue
# not sure if this can happen, but just in case log it.
LOGGER.warning("Attribute not found on device: %s", attr_cls)
# make sure to update the attributes once
self._update_from_device()
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
for unsub in self._unsubscribes:
unsub()
@callback
def _on_matter_event(self, event: EventType, data: Any = None) -> None:
"""Call on update."""
self._update_from_device()
self.async_write_ha_state()
@callback
@abstractmethod
def _update_from_device(self) -> None:
"""Update data from Matter device."""
@callback
def get_matter_attribute(self, attribute: type) -> MatterAttribute | None:
"""Lookup MatterAttribute instance on device instance by providing the attribute class."""
return next(
(
x
for x in self._device_type_instance.attributes
if x.attribute_type == attribute
),
None,
)

173
homeassistant/components/matter/light.py Normal file
View file

@ -0,0 +1,173 @@
"""Matter light."""
from __future__ import annotations
from dataclasses import dataclass
from functools import partial
from typing import TYPE_CHECKING, Any
from chip.clusters import Objects as clusters
from matter_server.common.models import device_types
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ColorMode,
LightEntity,
LightEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .entity import MatterEntity, MatterEntityDescriptionBaseClass
from .util import renormalize
if TYPE_CHECKING:
from .adapter import MatterAdapter
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Matter Light from Config Entry."""
matter: MatterAdapter = hass.data[DOMAIN][config_entry.entry_id]
matter.register_platform_handler(Platform.LIGHT, async_add_entities)
class MatterLight(MatterEntity, LightEntity):
"""Representation of a Matter light."""
entity_description: MatterLightEntityDescription
def _supports_brightness(self) -> bool:
"""Return if device supports brightness."""
return (
clusters.LevelControl.Attributes.CurrentLevel
in self.entity_description.subscribe_attributes
)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn light on."""
if ATTR_BRIGHTNESS not in kwargs or not self._supports_brightness():
await self.matter_client.send_device_command(
node_id=self._device_type_instance.node.node_id,
endpoint=self._device_type_instance.endpoint,
command=clusters.OnOff.Commands.On(),
)
return
level_control = self._device_type_instance.get_cluster(clusters.LevelControl)
level = round(
renormalize(
kwargs[ATTR_BRIGHTNESS],
(0, 255),
(level_control.minLevel, level_control.maxLevel),
)
)
await self.matter_client.send_device_command(
node_id=self._device_type_instance.node.node_id,
endpoint=self._device_type_instance.endpoint,
command=clusters.LevelControl.Commands.MoveToLevelWithOnOff(
level=level,
# It's required in TLV. We don't implement transition time yet.
transitionTime=0,
),
)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn light off."""
await self.matter_client.send_device_command(
node_id=self._device_type_instance.node.node_id,
endpoint=self._device_type_instance.endpoint,
command=clusters.OnOff.Commands.Off(),
)
@callback
def _update_from_device(self) -> None:
"""Update from device."""
if self._attr_supported_color_modes is None:
if self._supports_brightness():
self._attr_supported_color_modes = {ColorMode.BRIGHTNESS}
if attr := self.get_matter_attribute(clusters.OnOff.Attributes.OnOff):
self._attr_is_on = attr.value
if (
clusters.LevelControl.Attributes.CurrentLevel
in self.entity_description.subscribe_attributes
):
level_control = self._device_type_instance.get_cluster(
clusters.LevelControl
)
# Convert brightness to Home Assistant = 0..255
self._attr_brightness = round(
renormalize(
level_control.currentLevel,
(level_control.minLevel, level_control.maxLevel),
(0, 255),
)
)
@dataclass
class MatterLightEntityDescription(
LightEntityDescription,
MatterEntityDescriptionBaseClass,
):
"""Matter light entity description."""
# You can't set default values on inherited data classes
MatterLightEntityDescriptionFactory = partial(
MatterLightEntityDescription, entity_cls=MatterLight
)
# Mapping of a Matter Device type to Light Entity Description.
# A Matter device type (instance) can consist of multiple attributes.
# For example a Color Light which has an attribute to control brightness
# but also for color.
DEVICE_ENTITY: dict[
type[device_types.DeviceType],
MatterEntityDescriptionBaseClass | list[MatterEntityDescriptionBaseClass],
] = {
device_types.OnOffLight: MatterLightEntityDescriptionFactory(
key=device_types.OnOffLight,
subscribe_attributes=(clusters.OnOff.Attributes.OnOff,),
),
device_types.DimmableLight: MatterLightEntityDescriptionFactory(
key=device_types.DimmableLight,
subscribe_attributes=(
clusters.OnOff.Attributes.OnOff,
clusters.LevelControl.Attributes.CurrentLevel,
),
),
device_types.DimmablePlugInUnit: MatterLightEntityDescriptionFactory(
key=device_types.DimmablePlugInUnit,
subscribe_attributes=(
clusters.OnOff.Attributes.OnOff,
clusters.LevelControl.Attributes.CurrentLevel,
),
),
device_types.ColorTemperatureLight: MatterLightEntityDescriptionFactory(
key=device_types.ColorTemperatureLight,
subscribe_attributes=(
clusters.OnOff.Attributes.OnOff,
clusters.LevelControl.Attributes.CurrentLevel,
clusters.ColorControl,
),
),
device_types.ExtendedColorLight: MatterLightEntityDescriptionFactory(
key=device_types.ExtendedColorLight,
subscribe_attributes=(
clusters.OnOff.Attributes.OnOff,
clusters.LevelControl.Attributes.CurrentLevel,
clusters.ColorControl,
),
),
}

10
homeassistant/components/matter/manifest.json Normal file
View file

@ -0,0 +1,10 @@
{
"domain": "matter",
"name": "Matter (BETA)",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/matter",
"requirements": ["python-matter-server==1.0.6"],
"dependencies": ["websocket_api"],
"codeowners": ["@MartinHjelmare", "@marcelveldt"],
"iot_class": "local_push"
}

66
homeassistant/components/matter/services.yaml Normal file
View file

@ -0,0 +1,66 @@
commission:
name: Commission device
description: >
Add a new device to your Matter network.
fields:
code:
name: Pairing code
description: The pairing code for the device.
required: true
selector:
text:
accept_shared_device:
name: Accept shared device
description: >
Add a shared device to your Matter network.
fields:
pin:
name: Pin code
description: The pin code for the device.
required: true
selector:
text:
set_wifi:
name: Set Wi-Fi credentials
description: >
The Wi-Fi credentials will be sent as part of commissioning to a Matter device so it can connect to the Wi-Fi network.
fields:
ssid:
name: Network name
description: The SSID network name.
required: true
selector:
text:
password:
name: Password
description: The Wi-Fi network password.
required: true
selector:
text:
type: password
set_thread:
name: Set Thread network operational dataset
description: >
The Thread keys will be used as part of commissioning to let a Matter device join the Thread network.
Get keys by running `ot-ctl dataset active -x` on the Open Thread Border Router.
fields:
thread_operation_dataset:
name: Thread Operational Dataset
description: The Thread Operational Dataset to set.
required: true
selector:
text:
open_commissioning_window:
name: Open Commissioning Window
description: >
Allow adding one of your devices to another Matter network by opening the commissioning window for this Matter device for 60 seconds.
fields:
device_id:
name: Device
description: The Matter device to add to the other Matter network.
required: true
selector:
device:
integration: matter

47
homeassistant/components/matter/strings.json Normal file
View file

@ -0,0 +1,47 @@
{
"config": {
"flow_title": "{name}",
"step": {
"manual": {
"data": {
"url": "[%key:common::config_flow::data::url%]"
}
},
"on_supervisor": {
"title": "Select connection method",
"description": "Do you want to use the official Matter Server Supervisor add-on?\n\nIf you are already running the Matter Server in another add-on, in a custom container, natively etc., then do not select this option.",
"data": {
"use_addon": "Use the official Matter Server Supervisor add-on"
}
},
"install_addon": {
"title": "The add-on installation has started"
},
"start_addon": {
"title": "Starting add-on."
},
"hassio_confirm": {
"title": "Set up the Matter integration with the Matter Server add-on"
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_server_version": "The Matter server is not the correct version",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"addon_get_discovery_info_failed": "Failed to get Matter Server add-on discovery info.",
"addon_info_failed": "Failed to get Matter Server add-on info.",
"addon_install_failed": "Failed to install the Matter Server add-on.",
"addon_start_failed": "Failed to start the Matter Server add-on.",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"not_matter_addon": "Discovered add-on is not the official Matter Server add-on.",
"reconfiguration_successful": "Successfully reconfigured the Matter integration."
},
"progress": {
"install_addon": "Please wait while the Matter Server add-on installation finishes. This can take several minutes.",
"start_addon": "Please wait while the Matter Server add-on starts. This add-on is what powers Matter in Home Assistant. This may take some seconds."
}
}
}

47
homeassistant/components/matter/translations/en.json Normal file
View file

@ -0,0 +1,47 @@
{
"config": {
"abort": {
"addon_get_discovery_info_failed": "Failed to get Matter Server add-on discovery info.",
"addon_info_failed": "Failed to get Matter Server add-on info.",
"addon_install_failed": "Failed to install the Matter Server add-on.",
"addon_start_failed": "Failed to start the Matter Server add-on.",
"already_configured": "Device is already configured",
"already_in_progress": "Configuration flow is already in progress",
"not_matter_addon": "Discovered add-on is not the official Matter Server add-on.",
"reconfiguration_successful": "Successfully reconfigured the Matter integration."
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_server_version": "The Matter server is not the correct version",
"unknown": "Unexpected error"
},
"flow_title": "{name}",
"progress": {
"install_addon": "Please wait while the Matter Server add-on installation finishes. This can take several minutes.",
"start_addon": "Please wait while the Matter Server add-on starts. This add-on is what powers Matter in Home Assistant. This may take some seconds."
},
"step": {
"hassio_confirm": {
"title": "Set up the Matter integration with the Matter Server add-on"
},
"install_addon": {
"title": "The add-on installation has started"
},
"manual": {
"data": {
"url": "URL"
}
},
"on_supervisor": {
"data": {
"use_addon": "Use the official Matter Server Supervisor add-on"
},
"description": "Do you want to use the official Matter Server Supervisor add-on?\n\nIf you are already running the Matter Server in another add-on, in a custom container, natively etc., then do not select this option.",
"title": "Select connection method"
},
"start_addon": {
"title": "Starting add-on."
}
}
}
}

11
homeassistant/components/matter/util.py Normal file
View file

@ -0,0 +1,11 @@
"""Provide integration utilities."""
from __future__ import annotations
def renormalize(
number: float, from_range: tuple[float, float], to_range: tuple[float, float]
) -> float:
"""Change value from from_range to to_range."""
delta1 = from_range[1] - from_range[0]
delta2 = to_range[1] - to_range[0]
return (delta2 * (number - from_range[0]) / delta1) + to_range[0]

1
homeassistant/generated/config_flows.py
View file

@ -235,6 +235,7 @@ FLOWS = {
"lutron_caseta",
"lyric",
"mailgun",
"matter",
"mazda",
"meater",
"melcloud",

6
homeassistant/generated/integrations.json
View file

@ -3046,6 +3046,12 @@
"config_flow": false,
"iot_class": "cloud_push"
},
"matter": {
"name": "Matter (BETA)",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push"
},
"mazda": {
"name": "Mazda Connected Services",
"integration_type": "hub",

10
mypy.ini
View file

@ -1543,6 +1543,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.matter.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.media_player.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt
View file

@ -2029,6 +2029,9 @@ python-kasa==0.5.0
# homeassistant.components.lirc
# python-lirc==1.2.3
# homeassistant.components.matter
python-matter-server==1.0.6
# homeassistant.components.xiaomi_miio
python-miio==0.5.12

3
requirements_test_all.txt
View file

@ -1416,6 +1416,9 @@ python-juicenet==1.1.0
# homeassistant.components.tplink
python-kasa==0.5.0
# homeassistant.components.matter
python-matter-server==1.0.6
# homeassistant.components.xiaomi_miio
python-miio==0.5.12

1
tests/components/matter/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Tests for the Matter integration."""

146
tests/components/matter/common.py Normal file
View file

@ -0,0 +1,146 @@
"""Provide common test tools."""
from __future__ import annotations
import asyncio
from functools import cache
import json
import logging
from typing import TYPE_CHECKING, Any
from unittest.mock import Mock, patch
from matter_server.client import MatterClient
from matter_server.common.models.node import MatterNode
from matter_server.common.models.server_information import ServerInfo
import pytest
from tests.common import MockConfigEntry, load_fixture
if TYPE_CHECKING:
from homeassistant.core import HomeAssistant
MOCK_FABRIC_ID = 12341234
MOCK_COMPR_FABRIC_ID = 1234
# TEMP: Tests need to be fixed
pytestmark = pytest.mark.skip("all tests still WIP")
class MockClient(MatterClient):
"""Represent a mock Matter client."""
mock_client_disconnect: asyncio.Event
mock_commands: dict[type, Any] = {}
mock_sent_commands: list[dict[str, Any]] = []
def __init__(self) -> None:
"""Initialize the mock client."""
super().__init__("mock-url", None)
self.mock_commands: dict[type, Any] = {}
self.mock_sent_commands = []
self.server_info = ServerInfo(
fabric_id=MOCK_FABRIC_ID, compressed_fabric_id=MOCK_COMPR_FABRIC_ID
)
async def connect(self) -> None:
"""Connect to the Matter server."""
self.server_info = Mock(compressed_abric_d=MOCK_COMPR_FABRIC_ID)
async def listen(self, driver_ready: asyncio.Event) -> None:
"""Listen for events."""
driver_ready.set()
self.mock_client_disconnect = asyncio.Event()
await self.mock_client_disconnect.wait()
def mock_command(self, command_type: type, response: Any) -> None:
"""Mock a command."""
self.mock_commands[command_type] = response
async def async_send_command(
self,
command: str,
args: dict[str, Any],
require_schema: int | None = None,
) -> dict:
"""Send mock commands."""
if command == "device_controller.SendCommand" and (
(cmd_type := type(args.get("payload"))) in self.mock_commands
):
self.mock_sent_commands.append(args)
return self.mock_commands[cmd_type]
return await super().async_send_command(command, args, require_schema)
async def async_send_command_no_wait(
self, command: str, args: dict[str, Any], require_schema: int | None = None
) -> None:
"""Send a command without waiting for the response."""
if command == "SendCommand" and (
(cmd_type := type(args.get("payload"))) in self.mock_commands
):
self.mock_sent_commands.append(args)
return self.mock_commands[cmd_type]
return await super().async_send_command_no_wait(command, args, require_schema)
@pytest.fixture
async def mock_matter() -> Mock:
"""Mock matter fixture."""
return await get_mock_matter()
async def get_mock_matter() -> Mock:
"""Get mock Matter."""
return Mock(
adapter=Mock(logger=logging.getLogger("mock_matter")), client=MockClient()
)
@cache
def load_node_fixture(fixture: str) -> str:
"""Load a fixture."""
return load_fixture(f"matter/nodes/{fixture}.json")
def load_and_parse_node_fixture(fixture: str) -> dict[str, Any]:
"""Load and parse a node fixture."""
return json.loads(load_node_fixture(fixture))
async def setup_integration_with_node_fixture(
hass: HomeAssistant, hass_storage: dict[str, Any], node_fixture: str
) -> MatterNode:
"""Set up Matter integration with fixture as node."""
node_data = load_and_parse_node_fixture(node_fixture)
node = MatterNode(
await get_mock_matter(),
node_data,
)
config_entry = MockConfigEntry(
domain="matter", data={"url": "http://mock-matter-server-url"}
)
config_entry.add_to_hass(hass)
storage_key = f"matter_{config_entry.entry_id}"
hass_storage[storage_key] = {
"version": 1,
"minor_version": 0,
"key": storage_key,
"data": {
"compressed_fabric_id": MOCK_COMPR_FABRIC_ID,
"next_node_id": 4339,
"nodes": {str(node.node_id): node_data},
},
}
with patch(
"matter_server.client.matter.Client", return_value=node.matter.client
), patch(
"matter_server.client.model.node.MatterDeviceTypeInstance.subscribe_updates",
), patch(
"matter_server.client.model.node.MatterDeviceTypeInstance.update_attributes"
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return node

187
tests/components/matter/conftest.py Normal file
View file

@ -0,0 +1,187 @@
"""Provide common fixtures."""
from __future__ import annotations
import asyncio
from collections.abc import AsyncGenerator, Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.fixture(name="matter_client")
async def matter_client_fixture() -> AsyncGenerator[MagicMock, None]:
"""Fixture for a Matter client."""
with patch(
"homeassistant.components.matter.MatterClient", autospec=True
) as client_class:
client = client_class.return_value
async def connect() -> None:
"""Mock connect."""
await asyncio.sleep(0)
client.connected = True
async def listen(init_ready: asyncio.Event | None) -> None:
"""Mock listen."""
if init_ready is not None:
init_ready.set()
client.connect = AsyncMock(side_effect=connect)
client.start_listening = AsyncMock(side_effect=listen)
yield client
@pytest.fixture(name="integration")
async def integration_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MockConfigEntry:
"""Set up the Matter integration."""
entry = MockConfigEntry(domain="matter", data={"url": "ws://localhost:5580/ws"})
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
@pytest.fixture(name="create_backup")
def create_backup_fixture() -> Generator[AsyncMock, None, None]:
"""Mock Supervisor create backup of add-on."""
with patch(
"homeassistant.components.hassio.addon_manager.async_create_backup"
) as create_backup:
yield create_backup
@pytest.fixture(name="addon_store_info")
def addon_store_info_fixture() -> Generator[AsyncMock, None, None]:
"""Mock Supervisor add-on store info."""
with patch(
"homeassistant.components.hassio.addon_manager.async_get_addon_store_info"
) as addon_store_info:
addon_store_info.return_value = {
"installed": None,
"state": None,
"version": "1.0.0",
}
yield addon_store_info
@pytest.fixture(name="addon_info")
def addon_info_fixture() -> Generator[AsyncMock, None, None]:
"""Mock Supervisor add-on info."""
with patch(
"homeassistant.components.hassio.addon_manager.async_get_addon_info",
) as addon_info:
addon_info.return_value = {
"hostname": None,
"options": {},
"state": None,
"update_available": False,
"version": None,
}
yield addon_info
@pytest.fixture(name="addon_not_installed")
def addon_not_installed_fixture(
addon_store_info: AsyncMock, addon_info: AsyncMock
) -> AsyncMock:
"""Mock add-on not installed."""
return addon_info
@pytest.fixture(name="addon_installed")
def addon_installed_fixture(
addon_store_info: AsyncMock, addon_info: AsyncMock
) -> AsyncMock:
"""Mock add-on already installed but not running."""
addon_store_info.return_value = {
"installed": "1.0.0",
"state": "stopped",
"version": "1.0.0",
}
addon_info.return_value["hostname"] = "core-matter-server"
addon_info.return_value["state"] = "stopped"
addon_info.return_value["version"] = "1.0.0"
return addon_info
@pytest.fixture(name="addon_running")
def addon_running_fixture(
addon_store_info: AsyncMock, addon_info: AsyncMock
) -> AsyncMock:
"""Mock add-on already running."""
addon_store_info.return_value = {
"installed": "1.0.0",
"state": "started",
"version": "1.0.0",
}
addon_info.return_value["hostname"] = "core-matter-server"
addon_info.return_value["state"] = "started"
addon_info.return_value["version"] = "1.0.0"
return addon_info
@pytest.fixture(name="install_addon")
def install_addon_fixture(
addon_store_info: AsyncMock, addon_info: AsyncMock
) -> Generator[AsyncMock, None, None]:
"""Mock install add-on."""
async def install_addon_side_effect(hass: HomeAssistant, slug: str) -> None:
"""Mock install add-on."""
addon_store_info.return_value = {
"installed": "1.0.0",
"state": "stopped",
"version": "1.0.0",
}
addon_info.return_value["state"] = "stopped"
addon_info.return_value["version"] = "1.0.0"
with patch(
"homeassistant.components.hassio.addon_manager.async_install_addon"
) as install_addon:
install_addon.side_effect = install_addon_side_effect
yield install_addon
@pytest.fixture(name="start_addon")
def start_addon_fixture() -> Generator[AsyncMock, None, None]:
"""Mock start add-on."""
with patch(
"homeassistant.components.hassio.addon_manager.async_start_addon"
) as start_addon:
yield start_addon
@pytest.fixture(name="stop_addon")
def stop_addon_fixture() -> Generator[AsyncMock, None, None]:
"""Mock stop add-on."""
with patch(
"homeassistant.components.hassio.addon_manager.async_stop_addon"
) as stop_addon:
yield stop_addon
@pytest.fixture(name="uninstall_addon")
def uninstall_addon_fixture() -> Generator[AsyncMock, None, None]:
"""Mock uninstall add-on."""
with patch(
"homeassistant.components.hassio.addon_manager.async_uninstall_addon"
) as uninstall_addon:
yield uninstall_addon
@pytest.fixture(name="update_addon")
def update_addon_fixture() -> Generator[AsyncMock, None, None]:
"""Mock update add-on."""
with patch(
"homeassistant.components.hassio.addon_manager.async_update_addon"
) as update_addon:
yield update_addon

174
tests/components/matter/fixtures/nodes/fake-bridge-two-light.json Normal file
View file

@ -0,0 +1,174 @@
{
"attributes": {
"0": {
"Descriptor": {
"deviceTypeList": [
{
"type": 22,
"revision": 1,
"_type": "chip.clusters.Objects.Descriptor.Structs.DeviceTypeStruct"
},
{
"type": 14,
"revision": 1,
"_type": "chip.clusters.Objects.Descriptor.Structs.DeviceTypeStruct"
}
],
"serverList": [29, 37, 40, 48, 49, 50, 51, 60, 62, 64, 65],
"clientList": [],
"partsList": [9, 10],
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.Descriptor"
},
"Basic": {
"dataModelRevision": 0,
"vendorName": "Mock Vendor",
"vendorID": 1234,
"productName": "Mock Bridge",
"productID": 2,
"nodeLabel": "My Mock Bridge",
"location": "nl",
"hardwareVersion": 123,
"hardwareVersionString": "TEST_VERSION",
"softwareVersion": 12345,
"softwareVersionString": "123.4.5",
"manufacturingDate": null,
"partNumber": null,
"productURL": null,
"productLabel": null,
"serialNumber": null,
"localConfigDisabled": null,
"reachable": null,
"uniqueID": "mock-hub-id",
"capabilityMinima": null,
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [
0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 18, 65528, 65529, 65531, 65532,
65533
],
"featureMap": 0,
"clusterRevision": 3,
"_type": "chip.clusters.Objects.Basic"
}
},
"9": {
"OnOff": {
"onOff": true,
"globalSceneControl": true,
"onTime": 0,
"offWaitTime": 0,
"startUpOnOff": 0,
"generatedCommandList": [],
"acceptedCommandList": [0, 1, 2, 64, 65, 66],
"attributeList": [
0, 16384, 16385, 16386, 16387, 65528, 65529, 65531, 65532, 65533
],
"featureMap": 1,
"clusterRevision": 4,
"_type": "chip.clusters.Objects.OnOff"
},
"Descriptor": {
"deviceTypeList": [
{
"type": 256,
"revision": 1,
"_type": "chip.clusters.Objects.Descriptor.Structs.DeviceTypeStruct"
},
{
"type": 19,
"revision": 1,
"_type": "chip.clusters.Objects.Descriptor.Structs.DeviceTypeStruct"
}
],
"serverList": [6, 29, 57, 768, 8, 40],
"clientList": [],
"partsList": [],
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 1, 2, 3, 65528, 65529, 65531, 65533],
"featureMap": null,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.Descriptor"
},
"BridgedDeviceBasic": {
"nodeLabel": "Kitchen Ceiling",
"reachable": true,
"vendorID": 1234,
"softwareVersionString": "67.8.9",
"softwareVersion": 6789,
"vendorName": "Mock Vendor",
"productName": "Mock Light",
"uniqueID": "mock-id-kitchen-ceiling",
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [
5, 17, 2, 4, 10, 9, 1, 3, 18, 65528, 65529, 65531, 65532, 65533
],
"_type": "chip.clusters.Objects.BridgedDeviceBasic"
}
},
"10": {
"OnOff": {
"onOff": false,
"globalSceneControl": true,
"onTime": 0,
"offWaitTime": 0,
"startUpOnOff": 0,
"generatedCommandList": [],
"acceptedCommandList": [0, 1, 2, 64, 65, 66],
"attributeList": [
0, 16384, 16385, 16386, 16387, 65528, 65529, 65531, 65532, 65533
],
"featureMap": 1,
"clusterRevision": 4,
"_type": "chip.clusters.Objects.OnOff"
},
"Descriptor": {
"deviceTypeList": [
{
"type": 256,
"revision": 1,
"_type": "chip.clusters.Objects.Descriptor.Structs.DeviceTypeStruct"
},
{
"type": 19,
"revision": 1,
"_type": "chip.clusters.Objects.Descriptor.Structs.DeviceTypeStruct"
}
],
"serverList": [6, 29, 57, 768, 40],
"clientList": [],
"partsList": [],
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 1, 2, 3, 65528, 65529, 65531, 65533],
"featureMap": null,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.Descriptor"
},
"BridgedDeviceBasic": {
"nodeLabel": "Living Room Ceiling",
"reachable": true,
"vendorID": 1234,
"softwareVersionString": "1.49.1",
"softwareVersion": 19988481,
"vendorName": "Mock Vendor",
"productName": "Mock Light",
"uniqueID": "mock-id-living-room-ceiling",
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [
5, 17, 2, 4, 10, 9, 1, 3, 18, 65528, 65529, 65531, 65532, 65533
],
"_type": "chip.clusters.Objects.BridgedDeviceBasic"
}
}
},
"events": [],
"node_id": 4338
}

882
tests/components/matter/fixtures/nodes/lighting-example-app.json Normal file
View file

@ -0,0 +1,882 @@
{
"attributes": {
"0": {
"Groups": {
"nameSupport": 128,
"generatedCommandList": [0, 1, 2, 3],
"acceptedCommandList": [0, 1, 2, 3, 4, 5],
"attributeList": [0, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 4,
"_type": "chip.clusters.Objects.Groups"
},
"Descriptor": {
"deviceTypeList": [
{
"type": 22,
"revision": 1,
"_type": "chip.clusters.Objects.Descriptor.Structs.DeviceTypeStruct"
}
],
"serverList": [
4, 29, 31, 40, 42, 43, 44, 48, 49, 50, 51, 52, 53, 54, 55, 59, 60, 62,
63, 64, 65
],
"clientList": [41],
"partsList": [1],
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.Descriptor"
},
"AccessControl": {
"acl": [
{
"privilege": 5,
"authMode": 2,
"subjects": [1],
"targets": null,
"fabricIndex": 1,
"_type": "chip.clusters.Objects.AccessControl.Structs.AccessControlEntry"
}
],
"extension": [],
"subjectsPerAccessControlEntry": 4,
"targetsPerAccessControlEntry": 3,
"accessControlEntriesPerFabric": 3,
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 1, 2, 3, 4, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.AccessControl"
},
"Basic": {
"dataModelRevision": 1,
"vendorName": "Nabu Casa",
"vendorID": 65521,
"productName": "M5STAMP Lighting App",
"productID": 32768,
"nodeLabel": "My Cool Light",
"location": "XX",
"hardwareVersion": 0,
"hardwareVersionString": "v1.0",
"softwareVersion": 1,
"softwareVersionString": "55ab764bea",
"manufacturingDate": "20200101",
"partNumber": "",
"productURL": "",
"productLabel": "",
"serialNumber": "",
"localConfigDisabled": false,
"reachable": true,
"uniqueID": "BE8F70AA40DDAE41",
"capabilityMinima": {
"caseSessionsPerFabric": 3,
"subscriptionsPerFabric": 65506,
"_type": "chip.clusters.Objects.Basic.Structs.CapabilityMinimaStruct"
},
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [
0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
65528, 65529, 65531, 65532, 65533
],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.Basic"
},
"OtaSoftwareUpdateRequestor": {
"defaultOtaProviders": [],
"updatePossible": true,
"updateState": 0,
"updateStateProgress": 0,
"generatedCommandList": [],
"acceptedCommandList": [0],
"attributeList": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.OtaSoftwareUpdateRequestor"
},
"LocalizationConfiguration": {
"activeLocale": "en-US",
"supportedLocales": [
"en-US",
"de-DE",
"fr-FR",
"en-GB",
"es-ES",
"zh-CN",
"it-IT",
"ja-JP"
],
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 1, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.LocalizationConfiguration"
},
"TimeFormatLocalization": {
"hourFormat": 0,
"activeCalendarType": 0,
"supportedCalendarTypes": [0, 1, 2, 3, 4, 5, 6, 8, 9, 10, 11, 7],
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 1, 2, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.TimeFormatLocalization"
},
"GeneralCommissioning": {
"breadcrumb": 0,
"basicCommissioningInfo": {
"failSafeExpiryLengthSeconds": 60,
"_type": "chip.clusters.Objects.GeneralCommissioning.Structs.BasicCommissioningInfo"
},
"regulatoryConfig": 0,
"locationCapability": 0,
"supportsConcurrentConnection": true,
"generatedCommandList": [1, 3, 5],
"acceptedCommandList": [0, 2, 4],
"attributeList": [0, 1, 2, 3, 4, 65528, 65529, 65531, 65532, 65533],
"featureMap": 6,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.GeneralCommissioning"
},
"NetworkCommissioning": {
"maxNetworks": 1,
"networks": [
{
"networkID": {
"_type": "bytes",
"value": "TGF6eUlvVA=="
},
"connected": true,
"_type": "chip.clusters.Objects.NetworkCommissioning.Structs.NetworkInfo"
}
],
"scanMaxTimeSeconds": 10,
"connectMaxTimeSeconds": 30,
"interfaceEnabled": true,
"lastNetworkingStatus": 0,
"lastNetworkID": {
"_type": "bytes",
"value": "TGF6eUlvVA=="
},
"lastConnectErrorValue": null,
"generatedCommandList": [1, 5, 7],
"acceptedCommandList": [0, 2, 4, 6, 8],
"attributeList": [
0, 1, 2, 3, 4, 5, 6, 7, 65528, 65529, 65531, 65532, 65533
],
"featureMap": 1,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.NetworkCommissioning"
},
"DiagnosticLogs": {
"generatedCommandList": [1],
"acceptedCommandList": [0],
"attributeList": [65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.DiagnosticLogs"
},
"GeneralDiagnostics": {
"networkInterfaces": [
{
"name": "WIFI_AP_DEF",
"isOperational": true,
"offPremiseServicesReachableIPv4": null,
"offPremiseServicesReachableIPv6": null,
"hardwareAddress": {
"_type": "bytes",
"value": "AAAAAAAA"
},
"IPv4Addresses": [],
"IPv6Addresses": [],
"type": 1,
"_type": "chip.clusters.Objects.GeneralDiagnostics.Structs.NetworkInterfaceType"
},
{
"name": "WIFI_STA_DEF",
"isOperational": true,
"offPremiseServicesReachableIPv4": null,
"offPremiseServicesReachableIPv6": null,
"hardwareAddress": {
"_type": "bytes",
"value": "hPcDJ8rI"
},
"IPv4Addresses": [],
"IPv6Addresses": [],
"type": 1,
"_type": "chip.clusters.Objects.GeneralDiagnostics.Structs.NetworkInterfaceType"
}
],
"rebootCount": 12,
"upTime": 458,
"totalOperationalHours": 0,
"bootReasons": 1,
"activeHardwareFaults": [],
"activeRadioFaults": [],
"activeNetworkFaults": [],
"testEventTriggersEnabled": false,
"generatedCommandList": [],
"acceptedCommandList": [0],
"attributeList": [
0, 1, 2, 3, 4, 5, 6, 7, 8, 65528, 65529, 65531, 65532, 65533
],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.GeneralDiagnostics"
},
"SoftwareDiagnostics": {
"threadMetrics": [],
"currentHeapFree": 116140,
"currentHeapUsed": 138932,
"currentHeapHighWatermark": 153796,
"generatedCommandList": [],
"acceptedCommandList": [0],
"attributeList": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.SoftwareDiagnostics"
},
"ThreadNetworkDiagnostics": {
"TLVValue": {
"0": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"1": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"2": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"3": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"4": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"5": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"6": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"7": [],
"8": [],
"9": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"10": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"11": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"12": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"13": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"14": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"15": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"16": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"17": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"18": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"19": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"20": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"21": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"22": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"23": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"24": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"25": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"26": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"27": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"28": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"29": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"30": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"31": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"32": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"33": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"34": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"35": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"36": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"37": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"38": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"39": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"40": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"41": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"42": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"43": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"44": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"45": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"46": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"47": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"48": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"49": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"50": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"51": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"52": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"53": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"54": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"55": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"56": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"57": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"58": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"59": [],
"60": {
"TLVValue": null,
"Reason": "InteractionModelError: Failure (0x1)"
},
"61": [],
"62": [],
"65532": 15,
"65533": 1,
"65528": [],
"65529": [0],
"65531": [
0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18,
19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35,
36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52,
53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 65528, 65529, 65531, 65532,
65533
]
},
"Reason": "Failed to decode field [].channel, expected type <class 'chip.tlv.uint'>, got <class 'chip.clusters.Attribute.ValueDecodeFailure'>"
},
"WiFiNetworkDiagnostics": {
"bssid": {
"_type": "bytes",
"value": "1iH5ZUbu"
},
"securityType": 4,
"wiFiVersion": 3,
"channelNumber": 1,
"rssi": -38,
"beaconLostCount": 0,
"beaconRxCount": 0,
"packetMulticastRxCount": 0,
"packetMulticastTxCount": 0,
"packetUnicastRxCount": 0,
"packetUnicastTxCount": 0,
"currentMaxRate": 0,
"overrunCount": 0,
"generatedCommandList": [],
"acceptedCommandList": [0],
"attributeList": [
0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 65528, 65529, 65531, 65532,
65533
],
"featureMap": 3,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.WiFiNetworkDiagnostics"
},
"EthernetNetworkDiagnostics": {
"PHYRate": null,
"fullDuplex": null,
"packetRxCount": 0,
"packetTxCount": 0,
"txErrCount": 0,
"collisionCount": 0,
"overrunCount": 0,
"carrierDetect": null,
"timeSinceReset": 0,
"generatedCommandList": [],
"acceptedCommandList": [0],
"attributeList": [
0, 1, 2, 3, 4, 5, 6, 7, 8, 65528, 65529, 65531, 65532, 65533
],
"featureMap": 3,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.EthernetNetworkDiagnostics"
},
"Switch": {
"numberOfPositions": null,
"currentPosition": null,
"multiPressMax": null,
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.Switch"
},
"AdministratorCommissioning": {
"windowStatus": 0,
"adminFabricIndex": 0,
"adminVendorId": 0,
"generatedCommandList": [],
"acceptedCommandList": [0, 1, 2],
"attributeList": [0, 1, 2, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.AdministratorCommissioning"
},
"OperationalCredentials": {
"NOCs": [
{
"noc": {
"_type": "bytes",
"value": "FTABAQEkAgE3AyQTARgmBIAigScmBYAlTTo3BiQVASUR8RAYJAcBJAgBMAlBBHQsjZ/8Hpm4iqznEv0dAO03bZx8LDgqpIOpBsHeysZu8KAmI0K+p6B8FuI1h3wld1V+tIj5OHVHtrigg6Ssl043CjUBKAEYJAIBNgMEAgQBGDAEFEWrZiyeoUgEIXz4c40+Nzq9cfxHMAUUSTs2LnMMrX7nj+dns0cSq3SmK3MYMAtAoFdxyvsbLm6VekNCQ6yqJOucAcRSVY3Si4ov1alKPK9CaIPl+u5dvBWNfyEPXSLsPmzyfd2njl8WRz8e7CBiSRg="
},
"icac": {
"_type": "bytes",
"value": "FTABAQAkAgE3AyQUABgmBIAigScmBYAlTTo3BiQTARgkBwEkCAEwCUEE09c6S9xVbf3/blpXSgRAZzKXx/4KQC274cEfa2tFjdVAJYJUvM/8PMurRHEroPpA3FXpJ8/hfabkNvHGi2l8tTcKNQEpARgkAmAwBBRJOzYucwytfueP52ezRxKrdKYrczAFFBf0ohq+KHQlEVBIMgEeZCBPR72hGDALQNwd63sOjWKYhjlvDJmcPtIzljSsXlQ10vFrB5j9V9CdiZHDfy537G39fo0RJmpU63EGXYEtXVrEfSMiafshKVcY"
},
"fabricIndex": 1,
"_type": "chip.clusters.Objects.OperationalCredentials.Structs.NOCStruct"
}
],
"fabrics": [
{
"rootPublicKey": {
"_type": "bytes",
"value": "BBGg+O3i3tDVYryXkUmEXk1fnSMHN06+poGIfZODdvbZW4JvxHnrQVAxvZWIE6poLa0sKA8X8A7jmJsVFMUqLFM="
},
"vendorId": 35328,
"fabricId": 1,
"nodeId": 4337,
"label": "",
"fabricIndex": 1,
"_type": "chip.clusters.Objects.OperationalCredentials.Structs.FabricDescriptor"
}
],
"supportedFabrics": 5,
"commissionedFabrics": 1,
"trustedRootCertificates": [
{
"_type": "bytes",
"value": "FTABAQAkAgE3AyQUABgmBIAigScmBYAlTTo3BiQUABgkBwEkCAEwCUEEEaD47eLe0NVivJeRSYReTV+dIwc3Tr6mgYh9k4N29tlbgm/EeetBUDG9lYgTqmgtrSwoDxfwDuOYmxUUxSosUzcKNQEpARgkAmAwBBQX9KIavih0JRFQSDIBHmQgT0e9oTAFFBf0ohq+KHQlEVBIMgEeZCBPR72hGDALQO3xFiF2cEXl+/kk0CQfedzHJxSJiziHEjWCMjIj7SVlDVx4CpvNYHnheq+9vJFgcL8JQhAEdz6p6C3INBDL7dsY"
}
],
"currentFabricIndex": 1,
"generatedCommandList": [1, 3, 5, 8],
"acceptedCommandList": [0, 2, 4, 6, 7, 9, 10, 11],
"attributeList": [0, 1, 2, 3, 4, 5, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.OperationalCredentials"
},
"GroupKeyManagement": {
"groupKeyMap": [],
"groupTable": [],
"maxGroupsPerFabric": 3,
"maxGroupKeysPerFabric": 2,
"generatedCommandList": [2, 5],
"acceptedCommandList": [0, 1, 3, 4],
"attributeList": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.GroupKeyManagement"
},
"FixedLabel": {
"labelList": [
{
"label": "room",
"value": "bedroom 2",
"_type": "chip.clusters.Objects.FixedLabel.Structs.LabelStruct"
},
{
"label": "orientation",
"value": "North",
"_type": "chip.clusters.Objects.FixedLabel.Structs.LabelStruct"
},
{
"label": "floor",
"value": "2",
"_type": "chip.clusters.Objects.FixedLabel.Structs.LabelStruct"
},
{
"label": "direction",
"value": "up",
"_type": "chip.clusters.Objects.FixedLabel.Structs.LabelStruct"
}
],
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.FixedLabel"
},
"UserLabel": {
"labelList": [],
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.UserLabel"
}
},
"1": {
"Identify": {
"identifyTime": 0,
"identifyType": 0,
"generatedCommandList": [],
"acceptedCommandList": [0, 64],
"attributeList": [0, 1, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 4,
"_type": "chip.clusters.Objects.Identify"
},
"Groups": {
"nameSupport": 128,
"generatedCommandList": [0, 1, 2, 3],
"acceptedCommandList": [0, 1, 2, 3, 4, 5],
"attributeList": [0, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 4,
"_type": "chip.clusters.Objects.Groups"
},
"OnOff": {
"onOff": false,
"globalSceneControl": true,
"onTime": 0,
"offWaitTime": 0,
"startUpOnOff": null,
"generatedCommandList": [],
"acceptedCommandList": [0, 1, 2, 64, 65, 66],
"attributeList": [
0, 16384, 16385, 16386, 16387, 65528, 65529, 65531, 65532, 65533
],
"featureMap": 1,
"clusterRevision": 4,
"_type": "chip.clusters.Objects.OnOff"
},
"LevelControl": {
"currentLevel": 254,
"remainingTime": 0,
"minLevel": 0,
"maxLevel": 254,
"currentFrequency": 0,
"minFrequency": 0,
"maxFrequency": 0,
"options": 0,
"onOffTransitionTime": 0,
"onLevel": null,
"onTransitionTime": 0,
"offTransitionTime": 0,
"defaultMoveRate": 50,
"startUpCurrentLevel": null,
"generatedCommandList": [],
"acceptedCommandList": [0, 1, 2, 3, 4, 5, 6, 7],
"attributeList": [
0, 1, 2, 3, 4, 5, 6, 15, 16, 17, 18, 19, 20, 16384, 65528, 65529,
65531, 65532, 65533
],
"featureMap": 3,
"clusterRevision": 5,
"_type": "chip.clusters.Objects.LevelControl"
},
"Descriptor": {
"deviceTypeList": [
{
"type": 257,
"revision": 1,
"_type": "chip.clusters.Objects.Descriptor.Structs.DeviceTypeStruct"
}
],
"serverList": [3, 4, 6, 8, 29, 768, 1030],
"clientList": [],
"partsList": [],
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 1,
"_type": "chip.clusters.Objects.Descriptor"
},
"ColorControl": {
"currentHue": 0,
"currentSaturation": 0,
"remainingTime": 0,
"currentX": 24939,
"currentY": 24701,
"driftCompensation": null,
"compensationText": null,
"colorTemperatureMireds": 0,
"colorMode": 2,
"options": 0,
"numberOfPrimaries": 0,
"primary1X": null,
"primary1Y": null,
"primary1Intensity": null,
"primary2X": null,
"primary2Y": null,
"primary2Intensity": null,
"primary3X": null,
"primary3Y": null,
"primary3Intensity": null,
"primary4X": null,
"primary4Y": null,
"primary4Intensity": null,
"primary5X": null,
"primary5Y": null,
"primary5Intensity": null,
"primary6X": null,
"primary6Y": null,
"primary6Intensity": null,
"whitePointX": null,
"whitePointY": null,
"colorPointRX": null,
"colorPointRY": null,
"colorPointRIntensity": null,
"colorPointGX": null,
"colorPointGY": null,
"colorPointGIntensity": null,
"colorPointBX": null,
"colorPointBY": null,
"colorPointBIntensity": null,
"enhancedCurrentHue": 0,
"enhancedColorMode": 2,
"colorLoopActive": 0,
"colorLoopDirection": 0,
"colorLoopTime": 25,
"colorLoopStartEnhancedHue": 8960,
"colorLoopStoredEnhancedHue": 0,
"colorCapabilities": 0,
"colorTempPhysicalMinMireds": 0,
"colorTempPhysicalMaxMireds": 65279,
"coupleColorTempToLevelMinMireds": 0,
"startUpColorTemperatureMireds": 0,
"generatedCommandList": [],
"acceptedCommandList": [
0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 64, 65, 66, 67, 68, 71, 75, 76
],
"attributeList": [
0, 1, 2, 3, 4, 7, 8, 15, 16, 16384, 16385, 16386, 16387, 16388, 16389,
16390, 16394, 16395, 16396, 16397, 16400, 65528, 65529, 65531, 65532,
65533
],
"featureMap": 31,
"clusterRevision": 5,
"_type": "chip.clusters.Objects.ColorControl"
},
"OccupancySensing": {
"occupancy": 0,
"occupancySensorType": 0,
"occupancySensorTypeBitmap": 1,
"pirOccupiedToUnoccupiedDelay": null,
"pirUnoccupiedToOccupiedDelay": null,
"pirUnoccupiedToOccupiedThreshold": null,
"ultrasonicOccupiedToUnoccupiedDelay": null,
"ultrasonicUnoccupiedToOccupiedDelay": null,
"ultrasonicUnoccupiedToOccupiedThreshold": null,
"physicalContactOccupiedToUnoccupiedDelay": null,
"physicalContactUnoccupiedToOccupiedDelay": null,
"physicalContactUnoccupiedToOccupiedThreshold": null,
"generatedCommandList": [],
"acceptedCommandList": [],
"attributeList": [0, 1, 2, 65528, 65529, 65531, 65532, 65533],
"featureMap": 0,
"clusterRevision": 3,
"_type": "chip.clusters.Objects.OccupancySensing"
}
}
},
"events": [
{
"Header": {
"EndpointId": 0,
"ClusterId": 40,
"EventId": 0,
"EventNumber": 262144,
"Priority": 2,
"Timestamp": 2019,
"TimestampType": 0
},
"Status": 0,
"Data": {
"softwareVersion": 1,
"_type": "chip.clusters.Objects.Basic.Events.StartUp"
}
},
{
"Header": {
"EndpointId": 0,
"ClusterId": 51,
"EventId": 3,
"EventNumber": 262145,
"Priority": 2,
"Timestamp": 2020,
"TimestampType": 0
},
"Status": 0,
"Data": {
"bootReason": 1,
"_type": "chip.clusters.Objects.GeneralDiagnostics.Events.BootReason"
}
},
{
"Header": {
"EndpointId": 0,
"ClusterId": 54,
"EventId": 2,
"EventNumber": 262146,
"Priority": 1,
"Timestamp": 2216,
"TimestampType": 0
},
"Status": 0,
"Data": {
"connectionStatus": 0,
"_type": "chip.clusters.Objects.WiFiNetworkDiagnostics.Events.ConnectionStatus"
}
}
],
"node_id": 4337
}

78
tests/components/matter/test_adapter.py Normal file
View file

@ -0,0 +1,78 @@
"""Test the adapter."""
from __future__ import annotations
from typing import Any
import pytest
from homeassistant.components.matter.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .common import setup_integration_with_node_fixture
# TEMP: Tests need to be fixed
pytestmark = pytest.mark.skip("all tests still WIP")
async def test_device_registry_single_node_device(
hass: HomeAssistant, hass_storage: dict[str, Any]
) -> None:
"""Test bridge devices are set up correctly with via_device."""
await setup_integration_with_node_fixture(
hass, hass_storage, "lighting-example-app"
)
dev_reg = dr.async_get(hass)
entry = dev_reg.async_get_device({(DOMAIN, "BE8F70AA40DDAE41")})
assert entry is not None
assert entry.name == "My Cool Light"
assert entry.manufacturer == "Nabu Casa"
assert entry.model == "M5STAMP Lighting App"
assert entry.hw_version == "v1.0"
assert entry.sw_version == "55ab764bea"
async def test_device_registry_bridge(
hass: HomeAssistant, hass_storage: dict[str, Any]
) -> None:
"""Test bridge devices are set up correctly with via_device."""
await setup_integration_with_node_fixture(
hass, hass_storage, "fake-bridge-two-light"
)
dev_reg = dr.async_get(hass)
# Validate bridge
bridge_entry = dev_reg.async_get_device({(DOMAIN, "mock-hub-id")})
assert bridge_entry is not None
assert bridge_entry.name == "My Mock Bridge"
assert bridge_entry.manufacturer == "Mock Vendor"
assert bridge_entry.model == "Mock Bridge"
assert bridge_entry.hw_version == "TEST_VERSION"
assert bridge_entry.sw_version == "123.4.5"
# Device 1
device1_entry = dev_reg.async_get_device({(DOMAIN, "mock-id-kitchen-ceiling")})
assert device1_entry is not None
assert device1_entry.via_device_id == bridge_entry.id
assert device1_entry.name == "Kitchen Ceiling"
assert device1_entry.manufacturer == "Mock Vendor"
assert device1_entry.model == "Mock Light"
assert device1_entry.hw_version is None
assert device1_entry.sw_version == "67.8.9"
# Device 2
device2_entry = dev_reg.async_get_device({(DOMAIN, "mock-id-living-room-ceiling")})
assert device2_entry is not None
assert device2_entry.via_device_id == bridge_entry.id
assert device2_entry.name == "Living Room Ceiling"
assert device2_entry.manufacturer == "Mock Vendor"
assert device2_entry.model == "Mock Light"
assert device2_entry.hw_version is None
assert device2_entry.sw_version == "1.49.1"

179
tests/components/matter/test_api.py Normal file
View file

@ -0,0 +1,179 @@
"""Test the api module."""
from collections.abc import Awaitable, Callable
from unittest.mock import MagicMock, call
from aiohttp import ClientWebSocketResponse
from matter_server.client.exceptions import FailedCommand
from homeassistant.components.matter.api import ID, TYPE
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_commission(
hass: HomeAssistant,
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
matter_client: MagicMock,
integration: MockConfigEntry,
) -> None:
"""Test the commission command."""
ws_client = await hass_ws_client(hass)
await ws_client.send_json(
{
ID: 1,
TYPE: "matter/commission",
"code": "12345678",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
matter_client.commission_with_code.assert_called_once_with("12345678")
matter_client.commission_with_code.reset_mock()
matter_client.commission_with_code.side_effect = FailedCommand(
"test_id", "test_code", "Failed to commission"
)
await ws_client.send_json(
{
ID: 2,
TYPE: "matter/commission",
"code": "12345678",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "test_code"
matter_client.commission_with_code.assert_called_once_with("12345678")
async def test_commission_on_network(
hass: HomeAssistant,
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
matter_client: MagicMock,
integration: MockConfigEntry,
) -> None:
"""Test the commission on network command."""
ws_client = await hass_ws_client(hass)
await ws_client.send_json(
{
ID: 1,
TYPE: "matter/commission_on_network",
"pin": 1234,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
matter_client.commission_on_network.assert_called_once_with(1234)
matter_client.commission_on_network.reset_mock()
matter_client.commission_on_network.side_effect = FailedCommand(
"test_id", "test_code", "Failed to commission on network"
)
await ws_client.send_json(
{
ID: 2,
TYPE: "matter/commission_on_network",
"pin": 1234,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "test_code"
matter_client.commission_on_network.assert_called_once_with(1234)
async def test_set_thread_dataset(
hass: HomeAssistant,
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
matter_client: MagicMock,
integration: MockConfigEntry,
) -> None:
"""Test the set thread dataset command."""
ws_client = await hass_ws_client(hass)
await ws_client.send_json(
{
ID: 1,
TYPE: "matter/set_thread",
"thread_operation_dataset": "test_dataset",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
matter_client.set_thread_operational_dataset.assert_called_once_with("test_dataset")
matter_client.set_thread_operational_dataset.reset_mock()
matter_client.set_thread_operational_dataset.side_effect = FailedCommand(
"test_id", "test_code", "Failed to commission"
)
await ws_client.send_json(
{
ID: 2,
TYPE: "matter/set_thread",
"thread_operation_dataset": "test_dataset",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "test_code"
matter_client.set_thread_operational_dataset.assert_called_once_with("test_dataset")
async def test_set_wifi_credentials(
hass: HomeAssistant,
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
matter_client: MagicMock,
integration: MockConfigEntry,
) -> None:
"""Test the set WiFi credentials command."""
ws_client = await hass_ws_client(hass)
await ws_client.send_json(
{
ID: 1,
TYPE: "matter/set_wifi_credentials",
"network_name": "test_network",
"password": "test_password",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert matter_client.set_wifi_credentials.call_count == 1
assert matter_client.set_wifi_credentials.call_args == call(
ssid="test_network", credentials="test_password"
)
matter_client.set_wifi_credentials.reset_mock()
matter_client.set_wifi_credentials.side_effect = FailedCommand(
"test_id", "test_code", "Failed to commission on network"
)
await ws_client.send_json(
{
ID: 2,
TYPE: "matter/set_wifi_credentials",
"network_name": "test_network",
"password": "test_password",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "test_code"
assert matter_client.set_wifi_credentials.call_count == 1
assert matter_client.set_wifi_credentials.call_args == call(
ssid="test_network", credentials="test_password"
)

979
tests/components/matter/test_config_flow.py Normal file
View file

@ -0,0 +1,979 @@
"""Test the Matter config flow."""
from __future__ import annotations
from collections.abc import Generator
from typing import Any
from unittest.mock import DEFAULT, AsyncMock, MagicMock, call, patch
from matter_server.client.exceptions import CannotConnect, InvalidServerVersion
import pytest
from homeassistant import config_entries
from homeassistant.components.hassio import HassioAPIError, HassioServiceInfo
from homeassistant.components.matter.const import ADDON_SLUG, DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
ADDON_DISCOVERY_INFO = {
"addon": "Matter Server",
"host": "host1",
"port": 5581,
}
@pytest.fixture(name="setup_entry", autouse=True)
def setup_entry_fixture() -> Generator[AsyncMock, None, None]:
"""Mock entry setup."""
with patch(
"homeassistant.components.matter.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture(name="client_connect", autouse=True)
def client_connect_fixture() -> Generator[AsyncMock, None, None]:
"""Mock server version."""
with patch(
"homeassistant.components.matter.config_flow.MatterClient.connect"
) as client_connect:
yield client_connect
@pytest.fixture(name="supervisor")
def supervisor_fixture() -> Generator[MagicMock, None, None]:
"""Mock Supervisor."""
with patch(
"homeassistant.components.matter.config_flow.is_hassio", return_value=True
) as is_hassio:
yield is_hassio
@pytest.fixture(name="discovery_info")
def discovery_info_fixture() -> Any:
"""Return the discovery info from the supervisor."""
return DEFAULT
@pytest.fixture(name="get_addon_discovery_info", autouse=True)
def get_addon_discovery_info_fixture(
discovery_info: Any,
) -> Generator[AsyncMock, None, None]:
"""Mock get add-on discovery info."""
with patch(
"homeassistant.components.hassio.addon_manager.async_get_addon_discovery_info",
return_value=discovery_info,
) as get_addon_discovery_info:
yield get_addon_discovery_info
@pytest.fixture(name="addon_setup_time", autouse=True)
def addon_setup_time_fixture() -> Generator[int, None, None]:
"""Mock add-on setup sleep time."""
with patch(
"homeassistant.components.matter.config_flow.ADDON_SETUP_TIMEOUT", new=0
) as addon_setup_time:
yield addon_setup_time
async def test_manual_create_entry(
hass: HomeAssistant,
client_connect: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test user step create entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"url": "ws://localhost:5580/ws",
},
)
await hass.async_block_till_done()
assert client_connect.call_count == 1
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ws://localhost:5580/ws"
assert result["data"] == {
"url": "ws://localhost:5580/ws",
"integration_created_addon": False,
"use_addon": False,
}
assert setup_entry.call_count == 1
@pytest.mark.parametrize(
"error, side_effect",
[
("cannot_connect", CannotConnect(Exception("Boom"))),
("invalid_server_version", InvalidServerVersion("Invalid version")),
("unknown", Exception("Unknown boom")),
],
)
async def test_manual_errors(
hass: HomeAssistant,
client_connect: AsyncMock,
error: str,
side_effect: Exception,
) -> None:
"""Test user step cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
client_connect.side_effect = side_effect
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"url": "ws://localhost:5580/ws",
},
)
assert client_connect.call_count == 1
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": error}
async def test_manual_already_configured(
hass: HomeAssistant,
client_connect: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test manual step abort if already configured."""
entry = MockConfigEntry(
domain=DOMAIN, data={"url": "ws://host1:5581/ws"}, title="Matter"
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"url": "ws://localhost:5580/ws",
},
)
await hass.async_block_till_done()
assert client_connect.call_count == 1
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reconfiguration_successful"
assert entry.data["url"] == "ws://localhost:5580/ws"
assert entry.data["use_addon"] is False
assert entry.data["integration_created_addon"] is False
assert entry.title == "ws://localhost:5580/ws"
assert setup_entry.call_count == 1
@pytest.mark.parametrize("discovery_info", [{"config": ADDON_DISCOVERY_INFO}])
async def test_supervisor_discovery(
hass: HomeAssistant,
supervisor: MagicMock,
addon_running: AsyncMock,
addon_info: AsyncMock,
client_connect: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test flow started from Supervisor discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HASSIO},
data=HassioServiceInfo(
config=ADDON_DISCOVERY_INFO,
name="Matter Server",
slug=ADDON_SLUG,
),
)
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert addon_info.call_count == 1
assert client_connect.call_count == 0
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ws://host1:5581/ws"
assert result["data"] == {
"url": "ws://host1:5581/ws",
"use_addon": True,
"integration_created_addon": False,
}
assert setup_entry.call_count == 1
@pytest.mark.parametrize(
"discovery_info, error",
[({"config": ADDON_DISCOVERY_INFO}, HassioAPIError())],
)
async def test_supervisor_discovery_addon_info_failed(
hass: HomeAssistant,
supervisor: MagicMock,
addon_running: AsyncMock,
addon_info: AsyncMock,
error: Exception,
) -> None:
"""Test Supervisor discovery and addon info failed."""
addon_info.side_effect = error
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HASSIO},
data=HassioServiceInfo(
config=ADDON_DISCOVERY_INFO,
name="Matter Server",
slug=ADDON_SLUG,
),
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "hassio_confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert addon_info.call_count == 1
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_info_failed"
@pytest.mark.parametrize("discovery_info", [{"config": ADDON_DISCOVERY_INFO}])
async def test_clean_supervisor_discovery_on_user_create(
hass: HomeAssistant,
supervisor: MagicMock,
addon_running: AsyncMock,
addon_info: AsyncMock,
client_connect: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test discovery flow is cleaned up when a user flow is finished."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HASSIO},
data=HassioServiceInfo(
config=ADDON_DISCOVERY_INFO,
name="Matter Server",
slug=ADDON_SLUG,
),
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "hassio_confirm"
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": False}
)
assert addon_info.call_count == 0
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"url": "ws://localhost:5580/ws",
},
)
await hass.async_block_till_done()
assert len(hass.config_entries.flow.async_progress()) == 0
assert client_connect.call_count == 1
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ws://localhost:5580/ws"
assert result["data"] == {
"url": "ws://localhost:5580/ws",
"use_addon": False,
"integration_created_addon": False,
}
assert setup_entry.call_count == 1
async def test_abort_supervisor_discovery_with_existing_entry(
hass: HomeAssistant,
supervisor: MagicMock,
addon_running: AsyncMock,
addon_info: AsyncMock,
) -> None:
"""Test discovery flow is aborted if an entry already exists."""
entry = MockConfigEntry(
domain=DOMAIN,
data={"url": "ws://localhost:5580/ws"},
title="ws://localhost:5580/ws",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HASSIO},
data=HassioServiceInfo(
config=ADDON_DISCOVERY_INFO,
name="Matter Server",
slug=ADDON_SLUG,
),
)
assert addon_info.call_count == 0
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_abort_supervisor_discovery_with_existing_flow(
hass: HomeAssistant,
supervisor: MagicMock,
addon_installed: AsyncMock,
addon_info: AsyncMock,
) -> None:
"""Test hassio discovery flow is aborted when another flow is in progress."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HASSIO},
data=HassioServiceInfo(
config=ADDON_DISCOVERY_INFO,
name="Matter Server",
slug=ADDON_SLUG,
),
)
assert addon_info.call_count == 0
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_in_progress"
async def test_abort_supervisor_discovery_for_other_addon(
hass: HomeAssistant,
supervisor: MagicMock,
addon_installed: AsyncMock,
addon_info: AsyncMock,
) -> None:
"""Test hassio discovery flow is aborted for a non official add-on discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HASSIO},
data=HassioServiceInfo(
config={
"addon": "Other Matter Server",
"host": "host1",
"port": 3001,
},
name="Other Matter Server",
slug="other_addon",
),
)
assert addon_info.call_count == 0
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "not_matter_addon"
async def test_supervisor_discovery_addon_not_running(
hass: HomeAssistant,
supervisor: MagicMock,
addon_installed: AsyncMock,
addon_info: AsyncMock,
start_addon: AsyncMock,
client_connect: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test discovery with add-on already installed but not running."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HASSIO},
data=HassioServiceInfo(
config=ADDON_DISCOVERY_INFO,
name="Matter Server",
slug=ADDON_SLUG,
),
)
assert addon_info.call_count == 0
assert result["step_id"] == "hassio_confirm"
assert result["type"] == FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert addon_info.call_count == 1
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_addon"
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert start_addon.call_args == call(hass, "core_matter_server")
assert client_connect.call_count == 1
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ws://host1:5581/ws"
assert result["data"] == {
"url": "ws://host1:5581/ws",
"use_addon": True,
"integration_created_addon": False,
}
assert setup_entry.call_count == 1
async def test_supervisor_discovery_addon_not_installed(
hass: HomeAssistant,
supervisor: MagicMock,
addon_not_installed: AsyncMock,
install_addon: AsyncMock,
addon_info: AsyncMock,
addon_store_info: AsyncMock,
start_addon: AsyncMock,
client_connect: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test discovery with add-on not installed."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_HASSIO},
data=HassioServiceInfo(
config=ADDON_DISCOVERY_INFO,
name="Matter Server",
slug=ADDON_SLUG,
),
)
assert addon_info.call_count == 0
assert addon_store_info.call_count == 0
assert result["step_id"] == "hassio_confirm"
assert result["type"] == FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert addon_info.call_count == 0
assert addon_store_info.call_count == 1
assert result["step_id"] == "install_addon"
assert result["type"] == FlowResultType.SHOW_PROGRESS
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert install_addon.call_args == call(hass, "core_matter_server")
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_addon"
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert start_addon.call_args == call(hass, "core_matter_server")
assert client_connect.call_count == 1
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ws://host1:5581/ws"
assert result["data"] == {
"url": "ws://host1:5581/ws",
"use_addon": True,
"integration_created_addon": True,
}
assert setup_entry.call_count == 1
async def test_not_addon(
hass: HomeAssistant,
supervisor: MagicMock,
client_connect: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test opting out of add-on on Supervisor."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": False}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"url": "ws://localhost:5581/ws",
},
)
await hass.async_block_till_done()
assert client_connect.call_count == 1
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ws://localhost:5581/ws"
assert result["data"] == {
"url": "ws://localhost:5581/ws",
"use_addon": False,
"integration_created_addon": False,
}
assert setup_entry.call_count == 1
@pytest.mark.parametrize("discovery_info", [{"config": ADDON_DISCOVERY_INFO}])
async def test_addon_running(
hass: HomeAssistant,
supervisor: MagicMock,
addon_running: AsyncMock,
addon_info: AsyncMock,
client_connect: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test add-on already running on Supervisor."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": True}
)
await hass.async_block_till_done()
assert addon_info.call_count == 1
assert client_connect.call_count == 1
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ws://host1:5581/ws"
assert result["data"] == {
"url": "ws://host1:5581/ws",
"use_addon": True,
"integration_created_addon": False,
}
assert setup_entry.call_count == 1
@pytest.mark.parametrize(
"discovery_info, discovery_info_error, client_connect_error, addon_info_error, "
"abort_reason, discovery_info_called, client_connect_called",
[
(
{"config": ADDON_DISCOVERY_INFO},
HassioAPIError(),
None,
None,
"addon_get_discovery_info_failed",
True,
False,
),
(
{"config": ADDON_DISCOVERY_INFO},
None,
CannotConnect(Exception("Boom")),
None,
"cannot_connect",
True,
True,
),
(
None,
None,
None,
None,
"addon_get_discovery_info_failed",
True,
False,
),
(
{"config": ADDON_DISCOVERY_INFO},
None,
None,
HassioAPIError(),
"addon_info_failed",
False,
False,
),
],
)
async def test_addon_running_failures(
hass: HomeAssistant,
supervisor: MagicMock,
addon_running: AsyncMock,
addon_info: AsyncMock,
get_addon_discovery_info: AsyncMock,
client_connect: AsyncMock,
discovery_info_error: Exception | None,
client_connect_error: Exception | None,
addon_info_error: Exception | None,
abort_reason: str,
discovery_info_called: bool,
client_connect_called: bool,
) -> None:
"""Test all failures when add-on is running."""
get_addon_discovery_info.side_effect = discovery_info_error
client_connect.side_effect = client_connect_error
addon_info.side_effect = addon_info_error
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": True}
)
assert addon_info.call_count == 1
assert get_addon_discovery_info.called is discovery_info_called
assert client_connect.called is client_connect_called
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == abort_reason
@pytest.mark.parametrize("discovery_info", [{"config": ADDON_DISCOVERY_INFO}])
async def test_addon_running_already_configured(
hass: HomeAssistant,
supervisor: MagicMock,
addon_running: AsyncMock,
addon_info: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test that only one instance is allowed when add-on is running."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
"url": "ws://localhost:5580/ws",
},
title="ws://localhost:5580/ws",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": True}
)
await hass.async_block_till_done()
assert addon_info.call_count == 1
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reconfiguration_successful"
assert entry.data["url"] == "ws://host1:5581/ws"
assert entry.title == "ws://host1:5581/ws"
assert setup_entry.call_count == 1
@pytest.mark.parametrize("discovery_info", [{"config": ADDON_DISCOVERY_INFO}])
async def test_addon_installed(
hass: HomeAssistant,
supervisor: MagicMock,
addon_installed: AsyncMock,
addon_info: AsyncMock,
start_addon: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test add-on already installed but not running on Supervisor."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": True}
)
assert addon_info.call_count == 1
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_addon"
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert start_addon.call_args == call(hass, "core_matter_server")
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ws://host1:5581/ws"
assert result["data"] == {
"url": "ws://host1:5581/ws",
"use_addon": True,
"integration_created_addon": False,
}
assert setup_entry.call_count == 1
@pytest.mark.parametrize(
"discovery_info, start_addon_error, client_connect_error, "
"discovery_info_called, client_connect_called",
[
(
{"config": ADDON_DISCOVERY_INFO},
HassioAPIError(),
None,
False,
False,
),
(
{"config": ADDON_DISCOVERY_INFO},
None,
CannotConnect(Exception("Boom")),
True,
True,
),
(
None,
None,
None,
True,
False,
),
],
)
async def test_addon_installed_failures(
hass: HomeAssistant,
supervisor: MagicMock,
addon_installed: AsyncMock,
addon_info: AsyncMock,
start_addon: AsyncMock,
get_addon_discovery_info: AsyncMock,
client_connect: AsyncMock,
start_addon_error: Exception | None,
client_connect_error: Exception | None,
discovery_info_called: bool,
client_connect_called: bool,
) -> None:
"""Test add-on start failure when add-on is installed."""
start_addon.side_effect = start_addon_error
client_connect.side_effect = client_connect_error
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": True}
)
assert addon_info.call_count == 1
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_addon"
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert start_addon.call_args == call(hass, "core_matter_server")
assert get_addon_discovery_info.called is discovery_info_called
assert client_connect.called is client_connect_called
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_start_failed"
@pytest.mark.parametrize("discovery_info", [{"config": ADDON_DISCOVERY_INFO}])
async def test_addon_installed_already_configured(
hass: HomeAssistant,
supervisor: MagicMock,
addon_installed: AsyncMock,
addon_info: AsyncMock,
start_addon: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test that only one instance is allowed when add-on is installed."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
"url": "ws://localhost:5580/ws",
},
title="ws://localhost:5580/ws",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": True}
)
assert addon_info.call_count == 1
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_addon"
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert start_addon.call_args == call(hass, "core_matter_server")
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reconfiguration_successful"
assert entry.data["url"] == "ws://host1:5581/ws"
assert entry.title == "ws://host1:5581/ws"
assert setup_entry.call_count == 1
@pytest.mark.parametrize("discovery_info", [{"config": ADDON_DISCOVERY_INFO}])
async def test_addon_not_installed(
hass: HomeAssistant,
supervisor: MagicMock,
addon_not_installed: AsyncMock,
install_addon: AsyncMock,
addon_info: AsyncMock,
addon_store_info: AsyncMock,
start_addon: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test add-on not installed."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": True}
)
assert addon_info.call_count == 0
assert addon_store_info.call_count == 1
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "install_addon"
# Make sure the flow continues when the progress task is done.
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert install_addon.call_args == call(hass, "core_matter_server")
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_addon"
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert start_addon.call_args == call(hass, "core_matter_server")
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "ws://host1:5581/ws"
assert result["data"] == {
"url": "ws://host1:5581/ws",
"use_addon": True,
"integration_created_addon": True,
}
assert setup_entry.call_count == 1
async def test_addon_not_installed_failures(
hass: HomeAssistant,
supervisor: MagicMock,
addon_not_installed: AsyncMock,
addon_info: AsyncMock,
install_addon: AsyncMock,
) -> None:
"""Test add-on install failure."""
install_addon.side_effect = HassioAPIError()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": True}
)
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "install_addon"
# Make sure the flow continues when the progress task is done.
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert install_addon.call_args == call(hass, "core_matter_server")
assert addon_info.call_count == 0
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_install_failed"
@pytest.mark.parametrize("discovery_info", [{"config": ADDON_DISCOVERY_INFO}])
async def test_addon_not_installed_already_configured(
hass: HomeAssistant,
supervisor: MagicMock,
addon_not_installed: AsyncMock,
addon_info: AsyncMock,
addon_store_info: AsyncMock,
install_addon: AsyncMock,
start_addon: AsyncMock,
client_connect: AsyncMock,
setup_entry: AsyncMock,
) -> None:
"""Test that only one instance is allowed when add-on is not installed."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
"url": "ws://localhost:5580/ws",
},
title="ws://localhost:5580/ws",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "on_supervisor"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"use_addon": True}
)
assert addon_info.call_count == 0
assert addon_store_info.call_count == 1
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "install_addon"
# Make sure the flow continues when the progress task is done.
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert install_addon.call_args == call(hass, "core_matter_server")
assert result["type"] == FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_addon"
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert start_addon.call_args == call(hass, "core_matter_server")
assert client_connect.call_count == 1
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reconfiguration_successful"
assert entry.data["url"] == "ws://host1:5581/ws"
assert entry.title == "ws://host1:5581/ws"
assert setup_entry.call_count == 1

398
tests/components/matter/test_init.py Normal file
View file

@ -0,0 +1,398 @@
"""Test the Matter integration init."""
from __future__ import annotations
import asyncio
from unittest.mock import AsyncMock, MagicMock, call
from matter_server.client.exceptions import InvalidServerVersion
import pytest
from homeassistant.components.hassio import HassioAPIError
from homeassistant.components.matter.const import DOMAIN
from homeassistant.config_entries import ConfigEntryDisabler, ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from tests.common import MockConfigEntry
async def test_raise_addon_task_in_progress(
hass: HomeAssistant,
addon_not_installed: AsyncMock,
install_addon: AsyncMock,
start_addon: AsyncMock,
) -> None:
"""Test raise ConfigEntryNotReady if an add-on task is in progress."""
install_event = asyncio.Event()
install_addon_original_side_effect = install_addon.side_effect
async def install_addon_side_effect(hass: HomeAssistant, slug: str) -> None:
"""Mock install add-on."""
await install_event.wait()
await install_addon_original_side_effect(hass, slug)
install_addon.side_effect = install_addon_side_effect
entry = MockConfigEntry(
domain=DOMAIN,
title="Matter",
data={
"url": "ws://host1:5581/ws",
"use_addon": True,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await asyncio.sleep(0.05)
assert entry.state is ConfigEntryState.SETUP_RETRY
assert install_addon.call_count == 1
assert start_addon.call_count == 0
# Check that we only call install add-on once if a task is in progress.
await hass.config_entries.async_reload(entry.entry_id)
await asyncio.sleep(0.05)
assert entry.state is ConfigEntryState.SETUP_RETRY
assert install_addon.call_count == 1
assert start_addon.call_count == 0
install_event.set()
await hass.async_block_till_done()
assert install_addon.call_count == 1
assert start_addon.call_count == 1
async def test_start_addon(
hass: HomeAssistant,
addon_installed: AsyncMock,
addon_info: AsyncMock,
install_addon: AsyncMock,
start_addon: AsyncMock,
) -> None:
"""Test start the Matter Server add-on during entry setup."""
entry = MockConfigEntry(
domain=DOMAIN,
title="Matter",
data={
"url": "ws://host1:5581/ws",
"use_addon": True,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY
assert addon_info.call_count == 1
assert install_addon.call_count == 0
assert start_addon.call_count == 1
assert start_addon.call_args == call(hass, "core_matter_server")
async def test_install_addon(
hass: HomeAssistant,
addon_not_installed: AsyncMock,
addon_store_info: AsyncMock,
install_addon: AsyncMock,
start_addon: AsyncMock,
) -> None:
"""Test install and start the Matter add-on during entry setup."""
entry = MockConfigEntry(
domain=DOMAIN,
title="Matter",
data={
"url": "ws://host1:5581/ws",
"use_addon": True,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY
assert addon_store_info.call_count == 2
assert install_addon.call_count == 1
assert install_addon.call_args == call(hass, "core_matter_server")
assert start_addon.call_count == 1
assert start_addon.call_args == call(hass, "core_matter_server")
async def test_addon_info_failure(
hass: HomeAssistant,
addon_installed: AsyncMock,
addon_info: AsyncMock,
install_addon: AsyncMock,
start_addon: AsyncMock,
) -> None:
"""Test failure to get add-on info for Matter add-on during entry setup."""
addon_info.side_effect = HassioAPIError("Boom")
entry = MockConfigEntry(
domain=DOMAIN,
title="Matter",
data={
"url": "ws://host1:5581/ws",
"use_addon": True,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY
assert addon_info.call_count == 1
assert install_addon.call_count == 0
assert start_addon.call_count == 0
@pytest.mark.parametrize(
"addon_version, update_available, update_calls, backup_calls, "
"update_addon_side_effect, create_backup_side_effect",
[
("1.0.0", True, 1, 1, None, None),
("1.0.0", False, 0, 0, None, None),
("1.0.0", True, 1, 1, HassioAPIError("Boom"), None),
("1.0.0", True, 0, 1, None, HassioAPIError("Boom")),
],
)
async def test_update_addon(
hass: HomeAssistant,
addon_installed: AsyncMock,
addon_running: AsyncMock,
addon_info: AsyncMock,
install_addon: AsyncMock,
start_addon: AsyncMock,
create_backup: AsyncMock,
update_addon: AsyncMock,
matter_client: MagicMock,
addon_version: str,
update_available: bool,
update_calls: int,
backup_calls: int,
update_addon_side_effect: Exception | None,
create_backup_side_effect: Exception | None,
):
"""Test update the Matter add-on during entry setup."""
addon_info.return_value["version"] = addon_version
addon_info.return_value["update_available"] = update_available
create_backup.side_effect = create_backup_side_effect
update_addon.side_effect = update_addon_side_effect
matter_client.connect.side_effect = InvalidServerVersion("Invalid version")
entry = MockConfigEntry(
domain=DOMAIN,
title="Matter",
data={
"url": "ws://host1:5581/ws",
"use_addon": True,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_RETRY
assert create_backup.call_count == backup_calls
assert update_addon.call_count == update_calls
async def test_issue_registry_invalid_version(
hass: HomeAssistant,
matter_client: MagicMock,
) -> None:
"""Test issue registry for invalid version."""
original_connect_side_effect = matter_client.connect.side_effect
matter_client.connect.side_effect = InvalidServerVersion("Invalid version")
entry = MockConfigEntry(
domain=DOMAIN,
title="Matter",
data={
"url": "ws://host1:5581/ws",
"use_addon": False,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
issue_reg = ir.async_get(hass)
entry_state = entry.state
assert entry_state is ConfigEntryState.SETUP_RETRY
assert issue_reg.async_get_issue(DOMAIN, "invalid_server_version")
matter_client.connect.side_effect = original_connect_side_effect
await hass.config_entries.async_reload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert not issue_reg.async_get_issue(DOMAIN, "invalid_server_version")
@pytest.mark.parametrize(
"stop_addon_side_effect, entry_state",
[
(None, ConfigEntryState.NOT_LOADED),
(HassioAPIError("Boom"), ConfigEntryState.LOADED),
],
)
async def test_stop_addon(
hass,
matter_client: MagicMock,
addon_installed: AsyncMock,
addon_running: AsyncMock,
addon_info: AsyncMock,
stop_addon: AsyncMock,
stop_addon_side_effect: Exception | None,
entry_state: ConfigEntryState,
):
"""Test stop the Matter add-on on entry unload if entry is disabled."""
stop_addon.side_effect = stop_addon_side_effect
entry = MockConfigEntry(
domain=DOMAIN,
title="Matter",
data={
"url": "ws://host1:5581/ws",
"use_addon": True,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.LOADED
assert addon_info.call_count == 1
addon_info.reset_mock()
await hass.config_entries.async_set_disabled_by(
entry.entry_id, ConfigEntryDisabler.USER
)
await hass.async_block_till_done()
assert entry.state == entry_state
assert stop_addon.call_count == 1
assert stop_addon.call_args == call(hass, "core_matter_server")
async def test_remove_entry(
hass: HomeAssistant,
addon_installed: AsyncMock,
stop_addon: AsyncMock,
create_backup: AsyncMock,
uninstall_addon: AsyncMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test remove the config entry."""
# test successful remove without created add-on
entry = MockConfigEntry(
domain=DOMAIN,
title="Matter",
data={"integration_created_addon": False},
)
entry.add_to_hass(hass)
assert entry.state is ConfigEntryState.NOT_LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
await hass.config_entries.async_remove(entry.entry_id)
assert entry.state is ConfigEntryState.NOT_LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 0
# test successful remove with created add-on
entry = MockConfigEntry(
domain=DOMAIN,
title="Matter",
data={"integration_created_addon": True},
)
entry.add_to_hass(hass)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
await hass.config_entries.async_remove(entry.entry_id)
assert stop_addon.call_count == 1
assert stop_addon.call_args == call(hass, "core_matter_server")
assert create_backup.call_count == 1
assert create_backup.call_args == call(
hass,
{"name": "addon_core_matter_server_1.0.0", "addons": ["core_matter_server"]},
partial=True,
)
assert uninstall_addon.call_count == 1
assert uninstall_addon.call_args == call(hass, "core_matter_server")
assert entry.state is ConfigEntryState.NOT_LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 0
stop_addon.reset_mock()
create_backup.reset_mock()
uninstall_addon.reset_mock()
# test add-on stop failure
entry.add_to_hass(hass)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
stop_addon.side_effect = HassioAPIError()
await hass.config_entries.async_remove(entry.entry_id)
assert stop_addon.call_count == 1
assert stop_addon.call_args == call(hass, "core_matter_server")
assert create_backup.call_count == 0
assert uninstall_addon.call_count == 0
assert entry.state is ConfigEntryState.NOT_LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 0
assert "Failed to stop the Matter Server add-on" in caplog.text
stop_addon.side_effect = None
stop_addon.reset_mock()
create_backup.reset_mock()
uninstall_addon.reset_mock()
# test create backup failure
entry.add_to_hass(hass)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
create_backup.side_effect = HassioAPIError()
await hass.config_entries.async_remove(entry.entry_id)
assert stop_addon.call_count == 1
assert stop_addon.call_args == call(hass, "core_matter_server")
assert create_backup.call_count == 1
assert create_backup.call_args == call(
hass,
{"name": "addon_core_matter_server_1.0.0", "addons": ["core_matter_server"]},
partial=True,
)
assert uninstall_addon.call_count == 0
assert entry.state is ConfigEntryState.NOT_LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 0
assert "Failed to create a backup of the Matter Server add-on" in caplog.text
create_backup.side_effect = None
stop_addon.reset_mock()
create_backup.reset_mock()
uninstall_addon.reset_mock()
# test add-on uninstall failure
entry.add_to_hass(hass)
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
uninstall_addon.side_effect = HassioAPIError()
await hass.config_entries.async_remove(entry.entry_id)
assert stop_addon.call_count == 1
assert stop_addon.call_args == call(hass, "core_matter_server")
assert create_backup.call_count == 1
assert create_backup.call_args == call(
hass,
{"name": "addon_core_matter_server_1.0.0", "addons": ["core_matter_server"]},
partial=True,
)
assert uninstall_addon.call_count == 1
assert uninstall_addon.call_args == call(hass, "core_matter_server")
assert entry.state is ConfigEntryState.NOT_LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 0
assert "Failed to uninstall the Matter Server add-on" in caplog.text

82
tests/components/matter/test_light.py Normal file
View file

@ -0,0 +1,82 @@
"""Test Matter lights."""
from typing import Any
from chip.clusters import Objects as clusters
from matter_server.common.models.node import MatterNode
import pytest
from homeassistant.core import HomeAssistant
from .common import setup_integration_with_node_fixture
# TEMP: Tests need to be fixed
pytestmark = pytest.mark.skip("all tests still WIP")
@pytest.fixture(name="light_node")
async def light_node_fixture(
hass: HomeAssistant, hass_storage: dict[str, Any]
) -> MatterNode:
"""Fixture for a light node."""
return await setup_integration_with_node_fixture(
hass, hass_storage, "lighting-example-app"
)
async def test_turn_on(hass: HomeAssistant, light_node: MatterNode) -> None:
"""Test turning on a light."""
light_node.matter.client.mock_command(clusters.OnOff.Commands.On, None)
await hass.services.async_call(
"light",
"turn_on",
{
"entity_id": "light.my_cool_light",
},
blocking=True,
)
assert len(light_node.matter.client.mock_sent_commands) == 1
args = light_node.matter.client.mock_sent_commands[0]
assert args["nodeid"] == light_node.node_id
assert args["endpoint"] == 1
light_node.matter.client.mock_command(
clusters.LevelControl.Commands.MoveToLevelWithOnOff, None
)
await hass.services.async_call(
"light",
"turn_on",
{
"entity_id": "light.my_cool_light",
"brightness": 128,
},
blocking=True,
)
assert len(light_node.matter.client.mock_sent_commands) == 2
args = light_node.matter.client.mock_sent_commands[1]
assert args["nodeid"] == light_node.node_id
assert args["endpoint"] == 1
assert args["payload"].level == 127
assert args["payload"].transitionTime == 0
async def test_turn_off(hass: HomeAssistant, light_node: MatterNode) -> None:
"""Test turning off a light."""
light_node.matter.client.mock_command(clusters.OnOff.Commands.Off, None)
await hass.services.async_call(
"light",
"turn_off",
{
"entity_id": "light.my_cool_light",
},
blocking=True,
)
assert len(light_node.matter.client.mock_sent_commands) == 1
args = light_node.matter.client.mock_sent_commands[0]
assert args["nodeid"] == light_node.node_id
assert args["endpoint"] == 1