mirror of
https://github.com/home-assistant/core.git
synced 2025-07-21 12:17:07 +00:00
Add Matter Websocket commands for node actions and diagnostics (#109127)
* bump python-matter-server to version 5.3.0 * Add all node related websocket services * remove open_commissioning_window service as it wasnt working anyways * use device id instead of node id * tests * add decorator to get node * add some tests for invalid device id * add test for unknown node * add explicit exception * adjust test * move exceptions * remove the additional config entry check for now to be picked up in follow up pR
This commit is contained in:
parent
fb04451c08
commit
68c633c317
@ -7,14 +7,13 @@ from functools import cache
|
||||
|
||||
from matter_server.client import MatterClient
|
||||
from matter_server.client.exceptions import CannotConnect, InvalidServerVersion
|
||||
from matter_server.common.errors import MatterError, NodeCommissionFailed, NodeNotExists
|
||||
import voluptuous as vol
|
||||
from matter_server.common.errors import MatterError, NodeNotExists
|
||||
|
||||
from homeassistant.components.hassio import AddonError, AddonManager, AddonState
|
||||
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
|
||||
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.core import Event, HomeAssistant, callback
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.issue_registry import (
|
||||
@ -22,7 +21,6 @@ from homeassistant.helpers.issue_registry import (
|
||||
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
|
||||
@ -117,7 +115,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
|
||||
if DOMAIN not in hass.data:
|
||||
hass.data[DOMAIN] = {}
|
||||
_async_init_services(hass)
|
||||
|
||||
# create an intermediate layer (adapter) which keeps track of the nodes
|
||||
# and discovery of platform entities from the node attributes
|
||||
@ -237,35 +234,6 @@ async def async_remove_config_entry_device(
|
||||
return True
|
||||
|
||||
|
||||
@callback
|
||||
def _async_init_services(hass: HomeAssistant) -> None:
|
||||
"""Init services."""
|
||||
|
||||
async def open_commissioning_window(call: ServiceCall) -> None:
|
||||
"""Open commissioning window on specific node."""
|
||||
node = node_from_ha_device_id(hass, call.data["device_id"])
|
||||
|
||||
if node 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.node_id)
|
||||
except NodeCommissionFailed 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)
|
||||
|
@ -5,7 +5,9 @@ from collections.abc import Callable, Coroutine
|
||||
from functools import wraps
|
||||
from typing import Any, Concatenate, ParamSpec
|
||||
|
||||
from matter_server.client.models.node import MatterNode
|
||||
from matter_server.common.errors import MatterError
|
||||
from matter_server.common.helpers.util import dataclass_to_dict
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components import websocket_api
|
||||
@ -13,12 +15,16 @@ from homeassistant.components.websocket_api import ActiveConnection
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
|
||||
from .adapter import MatterAdapter
|
||||
from .helpers import get_matter
|
||||
from .helpers import MissingNode, get_matter, node_from_ha_device_id
|
||||
|
||||
_P = ParamSpec("_P")
|
||||
|
||||
ID = "id"
|
||||
TYPE = "type"
|
||||
DEVICE_ID = "device_id"
|
||||
|
||||
|
||||
ERROR_NODE_NOT_FOUND = "node_not_found"
|
||||
|
||||
|
||||
@callback
|
||||
@ -28,6 +34,40 @@ def async_register_api(hass: HomeAssistant) -> None:
|
||||
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)
|
||||
websocket_api.async_register_command(hass, websocket_node_diagnostics)
|
||||
websocket_api.async_register_command(hass, websocket_ping_node)
|
||||
websocket_api.async_register_command(hass, websocket_open_commissioning_window)
|
||||
websocket_api.async_register_command(hass, websocket_remove_matter_fabric)
|
||||
websocket_api.async_register_command(hass, websocket_interview_node)
|
||||
|
||||
|
||||
def async_get_node(
|
||||
func: Callable[
|
||||
[HomeAssistant, ActiveConnection, dict[str, Any], MatterAdapter, MatterNode],
|
||||
Coroutine[Any, Any, None],
|
||||
],
|
||||
) -> Callable[
|
||||
[HomeAssistant, ActiveConnection, dict[str, Any], MatterAdapter],
|
||||
Coroutine[Any, Any, None],
|
||||
]:
|
||||
"""Decorate async function to get node."""
|
||||
|
||||
@wraps(func)
|
||||
async def async_get_node_func(
|
||||
hass: HomeAssistant,
|
||||
connection: ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
matter: MatterAdapter,
|
||||
) -> None:
|
||||
"""Provide user specific data and store to function."""
|
||||
node = node_from_ha_device_id(hass, msg[DEVICE_ID])
|
||||
if not node:
|
||||
raise MissingNode(
|
||||
f"Could not resolve Matter node from device id {msg[DEVICE_ID]}"
|
||||
)
|
||||
await func(hass, connection, msg, matter, node)
|
||||
|
||||
return async_get_node_func
|
||||
|
||||
|
||||
def async_get_matter_adapter(
|
||||
@ -76,6 +116,8 @@ def async_handle_failed_command(
|
||||
await func(hass, connection, msg, *args, **kwargs)
|
||||
except MatterError as err:
|
||||
connection.send_error(msg[ID], str(err.error_code), err.args[0])
|
||||
except MissingNode as err:
|
||||
connection.send_error(msg[ID], ERROR_NODE_NOT_FOUND, err.args[0])
|
||||
|
||||
return async_handle_failed_command_func
|
||||
|
||||
@ -173,3 +215,119 @@ async def websocket_set_wifi_credentials(
|
||||
ssid=msg["network_name"], credentials=msg["password"]
|
||||
)
|
||||
connection.send_result(msg[ID])
|
||||
|
||||
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required(TYPE): "matter/node_diagnostics",
|
||||
vol.Required(DEVICE_ID): str,
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
@async_handle_failed_command
|
||||
@async_get_matter_adapter
|
||||
@async_get_node
|
||||
async def websocket_node_diagnostics(
|
||||
hass: HomeAssistant,
|
||||
connection: ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
matter: MatterAdapter,
|
||||
node: MatterNode,
|
||||
) -> None:
|
||||
"""Gather diagnostics for the given node."""
|
||||
result = await matter.matter_client.node_diagnostics(node_id=node.node_id)
|
||||
connection.send_result(msg[ID], dataclass_to_dict(result))
|
||||
|
||||
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required(TYPE): "matter/ping_node",
|
||||
vol.Required(DEVICE_ID): str,
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
@async_handle_failed_command
|
||||
@async_get_matter_adapter
|
||||
@async_get_node
|
||||
async def websocket_ping_node(
|
||||
hass: HomeAssistant,
|
||||
connection: ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
matter: MatterAdapter,
|
||||
node: MatterNode,
|
||||
) -> None:
|
||||
"""Ping node on the currently known IP-adress(es)."""
|
||||
result = await matter.matter_client.ping_node(node_id=node.node_id)
|
||||
connection.send_result(msg[ID], result)
|
||||
|
||||
|
||||
@websocket_api.require_admin
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required(TYPE): "matter/open_commissioning_window",
|
||||
vol.Required(DEVICE_ID): str,
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
@async_handle_failed_command
|
||||
@async_get_matter_adapter
|
||||
@async_get_node
|
||||
async def websocket_open_commissioning_window(
|
||||
hass: HomeAssistant,
|
||||
connection: ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
matter: MatterAdapter,
|
||||
node: MatterNode,
|
||||
) -> None:
|
||||
"""Open a commissioning window to commission a device present on this controller to another."""
|
||||
result = await matter.matter_client.open_commissioning_window(node_id=node.node_id)
|
||||
connection.send_result(msg[ID], dataclass_to_dict(result))
|
||||
|
||||
|
||||
@websocket_api.require_admin
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required(TYPE): "matter/remove_matter_fabric",
|
||||
vol.Required(DEVICE_ID): str,
|
||||
vol.Required("fabric_index"): int,
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
@async_handle_failed_command
|
||||
@async_get_matter_adapter
|
||||
@async_get_node
|
||||
async def websocket_remove_matter_fabric(
|
||||
hass: HomeAssistant,
|
||||
connection: ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
matter: MatterAdapter,
|
||||
node: MatterNode,
|
||||
) -> None:
|
||||
"""Remove Matter fabric from a device."""
|
||||
await matter.matter_client.remove_matter_fabric(
|
||||
node_id=node.node_id, fabric_index=msg["fabric_index"]
|
||||
)
|
||||
connection.send_result(msg[ID])
|
||||
|
||||
|
||||
@websocket_api.require_admin
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required(TYPE): "matter/interview_node",
|
||||
vol.Required(DEVICE_ID): str,
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
@async_handle_failed_command
|
||||
@async_get_matter_adapter
|
||||
@async_get_node
|
||||
async def websocket_interview_node(
|
||||
hass: HomeAssistant,
|
||||
connection: ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
matter: MatterAdapter,
|
||||
node: MatterNode,
|
||||
) -> None:
|
||||
"""Interview a node."""
|
||||
await matter.matter_client.interview_node(node_id=node.node_id)
|
||||
connection.send_result(msg[ID])
|
||||
|
@ -6,6 +6,7 @@ from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
|
||||
from .const import DOMAIN, ID_TYPE_DEVICE_ID
|
||||
@ -17,6 +18,10 @@ if TYPE_CHECKING:
|
||||
from .adapter import MatterAdapter
|
||||
|
||||
|
||||
class MissingNode(HomeAssistantError):
|
||||
"""Exception raised when we can't find a node."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class MatterEntryData:
|
||||
"""Hold Matter data for the config entry."""
|
||||
@ -72,7 +77,7 @@ def node_from_ha_device_id(hass: HomeAssistant, ha_device_id: str) -> MatterNode
|
||||
dev_reg = dr.async_get(hass)
|
||||
device = dev_reg.async_get(ha_device_id)
|
||||
if device is None:
|
||||
raise ValueError("Invalid device ID")
|
||||
raise MissingNode(f"Invalid device ID: {ha_device_id}")
|
||||
return get_node_from_device_entry(hass, device)
|
||||
|
||||
|
||||
|
@ -1,7 +0,0 @@
|
||||
open_commissioning_window:
|
||||
fields:
|
||||
device_id:
|
||||
required: true
|
||||
selector:
|
||||
device:
|
||||
integration: matter
|
@ -1,11 +1,28 @@
|
||||
"""Test the api module."""
|
||||
from unittest.mock import MagicMock, call
|
||||
from unittest.mock import AsyncMock, MagicMock, call
|
||||
|
||||
from matter_server.client.models.node import (
|
||||
MatterFabricData,
|
||||
NetworkType,
|
||||
NodeDiagnostics,
|
||||
NodeType,
|
||||
)
|
||||
from matter_server.common.errors import InvalidCommand, NodeCommissionFailed
|
||||
from matter_server.common.helpers.util import dataclass_to_dict
|
||||
from matter_server.common.models import CommissioningParameters
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.matter.api import ID, TYPE
|
||||
from homeassistant.components.matter.api import (
|
||||
DEVICE_ID,
|
||||
ERROR_NODE_NOT_FOUND,
|
||||
ID,
|
||||
TYPE,
|
||||
)
|
||||
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
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.typing import WebSocketGenerator
|
||||
@ -177,3 +194,307 @@ async def test_set_wifi_credentials(
|
||||
assert matter_client.set_wifi_credentials.call_args == call(
|
||||
ssid="test_network", credentials="test_password"
|
||||
)
|
||||
|
||||
|
||||
# This tests needs to be adjusted to remove lingering tasks
|
||||
@pytest.mark.parametrize("expected_lingering_tasks", [True])
|
||||
async def test_node_diagnostics(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
matter_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test the node diagnostics command."""
|
||||
# setup (mock) integration with a random node fixture
|
||||
await setup_integration_with_node_fixture(
|
||||
hass,
|
||||
"onoff-light",
|
||||
matter_client,
|
||||
)
|
||||
# get the device registry entry for the mocked node
|
||||
dev_reg = dr.async_get(hass)
|
||||
entry = dev_reg.async_get_device(
|
||||
identifiers={
|
||||
(DOMAIN, "deviceid_00000000000004D2-0000000000000001-MatterNodeDevice")
|
||||
}
|
||||
)
|
||||
assert entry is not None
|
||||
|
||||
# create a mock NodeDiagnostics
|
||||
mock_diagnostics = NodeDiagnostics(
|
||||
node_id=1,
|
||||
network_type=NetworkType.WIFI,
|
||||
node_type=NodeType.END_DEVICE,
|
||||
network_name="SuperCoolWiFi",
|
||||
ip_adresses=["192.168.1.1", "fe80::260:97ff:fe02:6ea5"],
|
||||
mac_address="00:11:22:33:44:55",
|
||||
available=True,
|
||||
active_fabrics=[MatterFabricData(2, 4939, 1, vendor_name="Nabu Casa")],
|
||||
)
|
||||
matter_client.node_diagnostics = AsyncMock(return_value=mock_diagnostics)
|
||||
|
||||
# issue command on the ws api
|
||||
ws_client = await hass_ws_client(hass)
|
||||
await ws_client.send_json(
|
||||
{
|
||||
ID: 1,
|
||||
TYPE: "matter/node_diagnostics",
|
||||
DEVICE_ID: entry.id,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert msg["success"]
|
||||
assert msg["type"] == "result"
|
||||
diag_res = dataclass_to_dict(mock_diagnostics)
|
||||
# dataclass to dict allows enums which are converted to string when serializing
|
||||
diag_res["network_type"] = diag_res["network_type"].value
|
||||
diag_res["node_type"] = diag_res["node_type"].value
|
||||
assert msg["result"] == diag_res
|
||||
|
||||
# repeat test with a device id that does not have a node attached
|
||||
new_entry = dev_reg.async_get_or_create(
|
||||
config_entry_id=list(entry.config_entries)[0],
|
||||
identifiers={(DOMAIN, "MatterNodeDevice")},
|
||||
)
|
||||
await ws_client.send_json(
|
||||
{
|
||||
ID: 2,
|
||||
TYPE: "matter/node_diagnostics",
|
||||
DEVICE_ID: new_entry.id,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
|
||||
|
||||
|
||||
# This tests needs to be adjusted to remove lingering tasks
|
||||
@pytest.mark.parametrize("expected_lingering_tasks", [True])
|
||||
async def test_ping_node(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
matter_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test the ping_node command."""
|
||||
# setup (mock) integration with a random node fixture
|
||||
await setup_integration_with_node_fixture(
|
||||
hass,
|
||||
"onoff-light",
|
||||
matter_client,
|
||||
)
|
||||
# get the device registry entry for the mocked node
|
||||
dev_reg = dr.async_get(hass)
|
||||
entry = dev_reg.async_get_device(
|
||||
identifiers={
|
||||
(DOMAIN, "deviceid_00000000000004D2-0000000000000001-MatterNodeDevice")
|
||||
}
|
||||
)
|
||||
assert entry is not None
|
||||
|
||||
# create a mocked ping result
|
||||
ping_result = {"192.168.1.1": False, "fe80::260:97ff:fe02:6ea5": True}
|
||||
matter_client.ping_node = AsyncMock(return_value=ping_result)
|
||||
|
||||
# issue command on the ws api
|
||||
ws_client = await hass_ws_client(hass)
|
||||
await ws_client.send_json(
|
||||
{
|
||||
ID: 1,
|
||||
TYPE: "matter/ping_node",
|
||||
DEVICE_ID: entry.id,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert msg["success"]
|
||||
assert msg["type"] == "result"
|
||||
assert msg["result"] == ping_result
|
||||
|
||||
# repeat test with a device id that does not have a node attached
|
||||
new_entry = dev_reg.async_get_or_create(
|
||||
config_entry_id=list(entry.config_entries)[0],
|
||||
identifiers={(DOMAIN, "MatterNodeDevice")},
|
||||
)
|
||||
await ws_client.send_json(
|
||||
{
|
||||
ID: 2,
|
||||
TYPE: "matter/ping_node",
|
||||
DEVICE_ID: new_entry.id,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
|
||||
|
||||
|
||||
# This tests needs to be adjusted to remove lingering tasks
|
||||
@pytest.mark.parametrize("expected_lingering_tasks", [True])
|
||||
async def test_open_commissioning_window(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
matter_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test the open_commissioning_window command."""
|
||||
# setup (mock) integration with a random node fixture
|
||||
await setup_integration_with_node_fixture(
|
||||
hass,
|
||||
"onoff-light",
|
||||
matter_client,
|
||||
)
|
||||
# get the device registry entry for the mocked node
|
||||
dev_reg = dr.async_get(hass)
|
||||
entry = dev_reg.async_get_device(
|
||||
identifiers={
|
||||
(DOMAIN, "deviceid_00000000000004D2-0000000000000001-MatterNodeDevice")
|
||||
}
|
||||
)
|
||||
assert entry is not None
|
||||
|
||||
# create mocked CommissioningParameters
|
||||
commissioning_parameters = CommissioningParameters(
|
||||
setup_pin_code=51590642,
|
||||
setup_manual_code="36296231484",
|
||||
setup_qr_code="MT:00000CQM008-WE3G310",
|
||||
)
|
||||
matter_client.open_commissioning_window = AsyncMock(
|
||||
return_value=commissioning_parameters
|
||||
)
|
||||
|
||||
# issue command on the ws api
|
||||
ws_client = await hass_ws_client(hass)
|
||||
await ws_client.send_json(
|
||||
{
|
||||
ID: 1,
|
||||
TYPE: "matter/open_commissioning_window",
|
||||
DEVICE_ID: entry.id,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert msg["success"]
|
||||
assert msg["type"] == "result"
|
||||
assert msg["result"] == dataclass_to_dict(commissioning_parameters)
|
||||
|
||||
# repeat test with a device id that does not have a node attached
|
||||
new_entry = dev_reg.async_get_or_create(
|
||||
config_entry_id=list(entry.config_entries)[0],
|
||||
identifiers={(DOMAIN, "MatterNodeDevice")},
|
||||
)
|
||||
await ws_client.send_json(
|
||||
{
|
||||
ID: 2,
|
||||
TYPE: "matter/open_commissioning_window",
|
||||
DEVICE_ID: new_entry.id,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
|
||||
|
||||
|
||||
# This tests needs to be adjusted to remove lingering tasks
|
||||
@pytest.mark.parametrize("expected_lingering_tasks", [True])
|
||||
async def test_remove_matter_fabric(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
matter_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test the remove_matter_fabric command."""
|
||||
# setup (mock) integration with a random node fixture
|
||||
await setup_integration_with_node_fixture(
|
||||
hass,
|
||||
"onoff-light",
|
||||
matter_client,
|
||||
)
|
||||
# get the device registry entry for the mocked node
|
||||
dev_reg = dr.async_get(hass)
|
||||
entry = dev_reg.async_get_device(
|
||||
identifiers={
|
||||
(DOMAIN, "deviceid_00000000000004D2-0000000000000001-MatterNodeDevice")
|
||||
}
|
||||
)
|
||||
assert entry is not None
|
||||
|
||||
# issue command on the ws api
|
||||
ws_client = await hass_ws_client(hass)
|
||||
await ws_client.send_json(
|
||||
{
|
||||
ID: 1,
|
||||
TYPE: "matter/remove_matter_fabric",
|
||||
DEVICE_ID: entry.id,
|
||||
"fabric_index": 3,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
assert msg["success"]
|
||||
matter_client.remove_matter_fabric.assert_called_once_with(1, 3)
|
||||
|
||||
# repeat test with a device id that does not have a node attached
|
||||
new_entry = dev_reg.async_get_or_create(
|
||||
config_entry_id=list(entry.config_entries)[0],
|
||||
identifiers={(DOMAIN, "MatterNodeDevice")},
|
||||
)
|
||||
await ws_client.send_json(
|
||||
{
|
||||
ID: 2,
|
||||
TYPE: "matter/remove_matter_fabric",
|
||||
DEVICE_ID: new_entry.id,
|
||||
"fabric_index": 3,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
|
||||
|
||||
|
||||
# This tests needs to be adjusted to remove lingering tasks
|
||||
@pytest.mark.parametrize("expected_lingering_tasks", [True])
|
||||
async def test_interview_node(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
matter_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test the interview_node command."""
|
||||
# setup (mock) integration with a random node fixture
|
||||
await setup_integration_with_node_fixture(
|
||||
hass,
|
||||
"onoff-light",
|
||||
matter_client,
|
||||
)
|
||||
# get the device registry entry for the mocked node
|
||||
dev_reg = dr.async_get(hass)
|
||||
entry = dev_reg.async_get_device(
|
||||
identifiers={
|
||||
(DOMAIN, "deviceid_00000000000004D2-0000000000000001-MatterNodeDevice")
|
||||
}
|
||||
)
|
||||
assert entry is not None
|
||||
# issue command on the ws api
|
||||
ws_client = await hass_ws_client(hass)
|
||||
await ws_client.send_json(
|
||||
{ID: 1, TYPE: "matter/interview_node", DEVICE_ID: entry.id}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
assert msg["success"]
|
||||
matter_client.interview_node.assert_called_once_with(1)
|
||||
|
||||
# repeat test with a device id that does not have a node attached
|
||||
new_entry = dev_reg.async_get_or_create(
|
||||
config_entry_id=list(entry.config_entries)[0],
|
||||
identifiers={(DOMAIN, "MatterNodeDevice")},
|
||||
)
|
||||
await ws_client.send_json(
|
||||
{
|
||||
ID: 2,
|
||||
TYPE: "matter/interview_node",
|
||||
DEVICE_ID: new_entry.id,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
|
||||
|
Loading…
x
Reference in New Issue
Block a user