jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-18 18:57:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use Parametrizing in Matter tests (#126759)

Browse Source 
* Overhaul matter tests

* Overhaul

* Remove matter_client where obsolete

* Move snapshots to the top

* Use usefixtures

* Add Valve
This commit is contained in:
Joost Lekkerkerker 2024-09-26 19:40:20 +02:00 committed by GitHub
parent 6c539cd2d8
commit 6e12726b11
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 392 additions and 806 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
tests/components/matter/conftest.py
View File

@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
from collections.abc import AsyncGenerator
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from matter_server.client.models.node import MatterNode
@ -86,39 +87,20 @@ async def matter_devices(
return await setup_integration_with_node_fixture(hass, request.param, matter_client)
@pytest.fixture(name="door_lock")
async def door_lock_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a door lock node."""
return await setup_integration_with_node_fixture(hass, "door_lock", matter_client)
@pytest.fixture
def attributes() -> dict[str, Any]:
"""Return common attributes for all nodes."""
return {}
@pytest.fixture(name="smoke_detector")
async def smoke_detector_fixture(
hass: HomeAssistant, matter_client: MagicMock
@pytest.fixture
async def matter_node(
hass: HomeAssistant,
matter_client: MagicMock,
node_fixture: str,
attributes: dict[str, Any],
) -> MatterNode:
"""Fixture for a smoke detector node."""
"""Fixture for a Matter node."""
return await setup_integration_with_node_fixture(
hass, "smoke_detector", matter_client
)
@pytest.fixture(name="door_lock_with_unbolt")
async def door_lock_with_unbolt_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a door lock node with unbolt feature."""
return await setup_integration_with_node_fixture(
hass, "door_lock_with_unbolt", matter_client
)
@pytest.fixture(name="eve_contact_sensor_node")
async def eve_contact_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a contact sensor node."""
return await setup_integration_with_node_fixture(
hass, "eve_contact_sensor", matter_client
hass, node_fixture, matter_client, attributes
)

57
tests/components/matter/test_adapter.py
View File

@ -12,11 +12,12 @@ from homeassistant.components.matter.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .common import create_node_from_fixture, setup_integration_with_node_fixture
from .common import create_node_from_fixture
from tests.common import MockConfigEntry
@pytest.mark.usefixtures("matter_node")
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
@ -30,17 +31,9 @@ from tests.common import MockConfigEntry
async def test_device_registry_single_node_device(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
matter_client: MagicMock,
node_fixture: str,
name: str,
) -> None:
"""Test bridge devices are set up correctly with via_device."""
await setup_integration_with_node_fixture(
hass,
node_fixture,
matter_client,
)
entry = device_registry.async_get_device(
identifiers={
(DOMAIN, "deviceid_00000000000004D2-0000000000000001-MatterNodeDevice")
@ -60,20 +53,15 @@ async def test_device_registry_single_node_device(
assert entry.serial_number == "12345678"
@pytest.mark.usefixtures("matter_node")
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["on_off_plugin_unit"])
async def test_device_registry_single_node_device_alt(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
matter_client: MagicMock,
) -> None:
"""Test additional device with different attribute values."""
await setup_integration_with_node_fixture(
hass,
"on_off_plugin_unit",
matter_client,
)
entry = device_registry.async_get_device(
identifiers={
(DOMAIN, "deviceid_00000000000004D2-0000000000000001-MatterNodeDevice")
@ -89,19 +77,14 @@ async def test_device_registry_single_node_device_alt(
assert entry.serial_number is None
@pytest.mark.usefixtures("matter_node")
@pytest.mark.skip("Waiting for a new test fixture")
@pytest.mark.parametrize("node_fixture", ["fake_bridge_two_light"])
async def test_device_registry_bridge(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
matter_client: MagicMock,
) -> None:
"""Test bridge devices are set up correctly with via_device."""
await setup_integration_with_node_fixture(
hass,
"fake_bridge_two_light",
matter_client,
)
# Validate bridge
bridge_entry = device_registry.async_get_device(
identifiers={(DOMAIN, "mock-hub-id")}
@ -141,12 +124,12 @@ async def test_device_registry_bridge(
assert device2_entry.sw_version == "1.49.1"
@pytest.mark.usefixtures("integration")
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
async def test_node_added_subscription(
hass: HomeAssistant,
matter_client: MagicMock,
integration: MagicMock,
) -> None:
"""Test subscription to new devices work."""
assert matter_client.subscribe_events.call_count == 5
@ -168,30 +151,20 @@ async def test_node_added_subscription(
assert entity_state
@pytest.mark.usefixtures("matter_node")
@pytest.mark.parametrize("node_fixture", ["air_purifier"])
async def test_device_registry_single_node_composed_device(
hass: HomeAssistant,
matter_client: MagicMock,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test that a composed device within a standalone node only creates one HA device entry."""
await setup_integration_with_node_fixture(
hass,
"air_purifier",
matter_client,
)
dev_reg = dr.async_get(hass)
assert len(dev_reg.devices) == 1
assert len(device_registry.devices) == 1
async def test_multi_endpoint_name(
hass: HomeAssistant,
matter_client: MagicMock,
) -> None:
@pytest.mark.usefixtures("matter_node")
@pytest.mark.parametrize("node_fixture", ["multi_endpoint_light"])
async def test_multi_endpoint_name(hass: HomeAssistant) -> None:
"""Test that the entity name gets postfixed if the device has multiple primary endpoints."""
await setup_integration_with_node_fixture(
hass,
"multi_endpoint_light",
matter_client,
)
entity_state = hass.states.get("light.inovelli_light_1")
assert entity_state
assert entity_state.name == "Inovelli Light (1)"
@ -200,7 +173,7 @@ async def test_multi_endpoint_name(
assert entity_state.name == "Inovelli Light (6)"
async def test_get_clean_name_() -> None:
async def test_get_clean_name() -> None:
"""Test get_clean_name helper.
Test device names that are assigned to `null`

47
tests/components/matter/test_api.py
View File

@ -23,8 +23,6 @@ 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
@ -197,8 +195,11 @@ async def test_set_wifi_credentials(
)
@pytest.mark.usefixtures("matter_node")
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
# setup (mock) integration with a random node fixture
@pytest.mark.parametrize("node_fixture", ["onoff_light"])
async def test_node_diagnostics(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@ -206,12 +207,6 @@ async def test_node_diagnostics(
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
entry = device_registry.async_get_device(
identifiers={
@ -271,8 +266,11 @@ async def test_node_diagnostics(
assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
@pytest.mark.usefixtures("matter_node")
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
# setup (mock) integration with a random node fixture
@pytest.mark.parametrize("node_fixture", ["onoff_light"])
async def test_ping_node(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@ -280,12 +278,6 @@ async def test_ping_node(
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
entry = device_registry.async_get_device(
identifiers={
@ -331,8 +323,11 @@ async def test_ping_node(
assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
@pytest.mark.usefixtures("matter_node")
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
# setup (mock) integration with a random node fixture
@pytest.mark.parametrize("node_fixture", ["onoff_light"])
async def test_open_commissioning_window(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@ -340,12 +335,6 @@ async def test_open_commissioning_window(
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
entry = device_registry.async_get_device(
identifiers={
@ -397,8 +386,11 @@ async def test_open_commissioning_window(
assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
@pytest.mark.usefixtures("matter_node")
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
# setup (mock) integration with a random node fixture
@pytest.mark.parametrize("node_fixture", ["onoff_light"])
async def test_remove_matter_fabric(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@ -406,12 +398,6 @@ async def test_remove_matter_fabric(
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
entry = device_registry.async_get_device(
identifiers={
@ -453,8 +439,11 @@ async def test_remove_matter_fabric(
assert msg["error"]["code"] == ERROR_NODE_NOT_FOUND
@pytest.mark.usefixtures("matter_node")
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
# setup (mock) integration with a random node fixture
@pytest.mark.parametrize("node_fixture", ["onoff_light"])
async def test_interview_node(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@ -462,12 +451,6 @@ async def test_interview_node(
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
entry = device_registry.async_get_device(
identifiers={

61
tests/components/matter/test_binary_sensor.py
View File

@ -16,7 +16,6 @@ from homeassistant.helpers import entity_registry as er
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
snapshot_matter_entities,
trigger_subscription_callback,
)
@ -34,31 +33,34 @@ def binary_sensor_platform() -> Generator[None]:
yield
@pytest.fixture(name="occupancy_sensor_node")
async def occupancy_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a occupancy sensor node."""
return await setup_integration_with_node_fixture(
hass, "occupancy_sensor", matter_client
)
@pytest.mark.usefixtures("matter_devices")
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
async def test_binary_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test binary sensors."""
snapshot_matter_entities(hass, entity_registry, snapshot, Platform.BINARY_SENSOR)
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["occupancy_sensor"])
async def test_occupancy_sensor(
hass: HomeAssistant,
matter_client: MagicMock,
occupancy_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test occupancy sensor."""
state = hass.states.get("binary_sensor.mock_occupancy_sensor_occupancy")
assert state
assert state.state == "on"
set_node_attribute(occupancy_sensor_node, 1, 1030, 0, 0)
set_node_attribute(matter_node, 1, 1030, 0, 0)
await trigger_subscription_callback(
hass, matter_client, data=(occupancy_sensor_node.node_id, "1/1030/0", 0)
hass, matter_client, data=(matter_node.node_id, "1/1030/0", 0)
)
state = hass.states.get("binary_sensor.mock_occupancy_sensor_occupancy")
@ -69,7 +71,7 @@ async def test_occupancy_sensor(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("eve_contact_sensor", "binary_sensor.eve_door_door"),
("leak_sensor", "binary_sensor.water_leak_detector_water_leak"),
@ -78,24 +80,19 @@ async def test_occupancy_sensor(
async def test_boolean_state_sensors(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test if binary sensors get created from devices with Boolean State cluster."""
node = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
state = hass.states.get(entity_id)
assert state
assert state.state == "on"
# invert the value
cur_attr_value = node.get_attribute_value(1, 69, 0)
set_node_attribute(node, 1, 69, 0, not cur_attr_value)
cur_attr_value = matter_node.get_attribute_value(1, 69, 0)
set_node_attribute(matter_node, 1, 69, 0, not cur_attr_value)
await trigger_subscription_callback(
hass, matter_client, data=(node.node_id, "1/69/0", not cur_attr_value)
hass, matter_client, data=(matter_node.node_id, "1/69/0", not cur_attr_value)
)
state = hass.states.get(entity_id)
@ -105,11 +102,12 @@ async def test_boolean_state_sensors(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["door_lock"])
async def test_battery_sensor(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
matter_client: MagicMock,
door_lock: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test battery sensor."""
entity_id = "binary_sensor.mock_door_lock_battery"
@ -117,24 +115,11 @@ async def test_battery_sensor(
assert state
assert state.state == "off"
set_node_attribute(door_lock, 1, 47, 14, 1)
set_node_attribute(matter_node, 1, 47, 14, 1)
await trigger_subscription_callback(
hass, matter_client, data=(door_lock.node_id, "1/47/14", 1)
hass, matter_client, data=(matter_node.node_id, "1/47/14", 1)
)
state = hass.states.get(entity_id)
assert state
assert state.state == "on"
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
async def test_binary_sensors(
hass: HomeAssistant,
matter_client: MagicMock,
matter_devices: MatterNode,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test binary sensors."""
snapshot_matter_entities(hass, entity_registry, snapshot, Platform.BINARY_SENSOR)

34
tests/components/matter/test_button.py
View File

@ -8,35 +8,14 @@ import pytest
from homeassistant.core import HomeAssistant
from .common import setup_integration_with_node_fixture
@pytest.fixture(name="powerplug_node")
async def powerplug_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Powerplug node."""
return await setup_integration_with_node_fixture(
hass, "eve_energy_plug", matter_client
)
@pytest.fixture(name="dishwasher_node")
async def dishwasher_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for an dishwasher node."""
return await setup_integration_with_node_fixture(
hass, "silabs_dishwasher", matter_client
)
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["eve_energy_plug"])
async def test_identify_button(
hass: HomeAssistant,
matter_client: MagicMock,
powerplug_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test button entity is created for a Matter Identify Cluster."""
state = hass.states.get("button.eve_energy_plug_identify")
@ -53,23 +32,24 @@ async def test_identify_button(
)
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=powerplug_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.Identify.Commands.Identify(identifyTime=15),
)
@pytest.mark.parametrize("node_fixture", ["silabs_dishwasher"])
async def test_operational_state_buttons(
hass: HomeAssistant,
matter_client: MagicMock,
dishwasher_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test if button entities are created for operational state commands."""
assert hass.states.get("button.dishwasher_pause")
assert hass.states.get("button.dishwasher_start")
assert hass.states.get("button.dishwasher_stop")
# resume may not be disocvered as its missing in the supported command list
# resume may not be discovered as it's missing in the supported command list
assert hass.states.get("button.dishwasher_resume") is None
# test press action
@ -83,7 +63,7 @@ async def test_operational_state_buttons(
)
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=dishwasher_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OperationalState.Commands.Pause(),
)

91
tests/components/matter/test_climate.py
View File

@ -10,37 +10,16 @@ import pytest
from homeassistant.components.climate import ClimateEntityFeature, HVACAction, HVACMode
from homeassistant.core import HomeAssistant
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
trigger_subscription_callback,
)
@pytest.fixture(name="thermostat")
async def thermostat_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a thermostat node."""
return await setup_integration_with_node_fixture(hass, "thermostat", matter_client)
@pytest.fixture(name="room_airconditioner")
async def room_airconditioner(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a room air conditioner node."""
return await setup_integration_with_node_fixture(
hass, "room_airconditioner", matter_client
)
from .common import set_node_attribute, trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["thermostat"])
async def test_thermostat_base(
hass: HomeAssistant,
matter_client: MagicMock,
thermostat: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test thermostat base attributes and state updates."""
# test entity attributes
@ -61,10 +40,10 @@ async def test_thermostat_base(
assert state.attributes["supported_features"] & mask == mask
# test common state updates from device
set_node_attribute(thermostat, 1, 513, 3, 1600)
set_node_attribute(thermostat, 1, 513, 4, 3000)
set_node_attribute(thermostat, 1, 513, 5, 1600)
set_node_attribute(thermostat, 1, 513, 6, 3000)
set_node_attribute(matter_node, 1, 513, 3, 1600)
set_node_attribute(matter_node, 1, 513, 4, 3000)
set_node_attribute(matter_node, 1, 513, 5, 1600)
set_node_attribute(matter_node, 1, 513, 6, 3000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
@ -78,63 +57,63 @@ async def test_thermostat_base(
]
# test system mode update from device
set_node_attribute(thermostat, 1, 513, 28, 0)
set_node_attribute(matter_node, 1, 513, 28, 0)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
assert state.state == HVACMode.OFF
# test running state update from device
set_node_attribute(thermostat, 1, 513, 41, 1)
set_node_attribute(matter_node, 1, 513, 41, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
assert state.attributes["hvac_action"] == HVACAction.HEATING
set_node_attribute(thermostat, 1, 513, 41, 8)
set_node_attribute(matter_node, 1, 513, 41, 8)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
assert state.attributes["hvac_action"] == HVACAction.HEATING
set_node_attribute(thermostat, 1, 513, 41, 2)
set_node_attribute(matter_node, 1, 513, 41, 2)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
assert state.attributes["hvac_action"] == HVACAction.COOLING
set_node_attribute(thermostat, 1, 513, 41, 16)
set_node_attribute(matter_node, 1, 513, 41, 16)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
assert state.attributes["hvac_action"] == HVACAction.COOLING
set_node_attribute(thermostat, 1, 513, 41, 4)
set_node_attribute(matter_node, 1, 513, 41, 4)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
assert state.attributes["hvac_action"] == HVACAction.FAN
set_node_attribute(thermostat, 1, 513, 41, 32)
set_node_attribute(matter_node, 1, 513, 41, 32)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
assert state.attributes["hvac_action"] == HVACAction.FAN
set_node_attribute(thermostat, 1, 513, 41, 64)
set_node_attribute(matter_node, 1, 513, 41, 64)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
assert state.attributes["hvac_action"] == HVACAction.FAN
set_node_attribute(thermostat, 1, 513, 41, 66)
set_node_attribute(matter_node, 1, 513, 41, 66)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
assert state.attributes["hvac_action"] == HVACAction.OFF
# change system mode to heat
set_node_attribute(thermostat, 1, 513, 28, 4)
set_node_attribute(matter_node, 1, 513, 28, 4)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
@ -142,7 +121,7 @@ async def test_thermostat_base(
assert state.state == HVACMode.HEAT
# change occupied heating setpoint to 20
set_node_attribute(thermostat, 1, 513, 18, 2000)
set_node_attribute(matter_node, 1, 513, 18, 2000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
@ -152,10 +131,11 @@ async def test_thermostat_base(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["thermostat"])
async def test_thermostat_service_calls(
hass: HomeAssistant,
matter_client: MagicMock,
thermostat: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test climate platform service calls."""
# test single-setpoint temperature adjustment when cool mode is active
@ -174,14 +154,14 @@ async def test_thermostat_service_calls(
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=thermostat.node_id,
node_id=matter_node.node_id,
attribute_path="1/513/17",
value=2500,
)
matter_client.write_attribute.reset_mock()
# ensure that no command is executed when the temperature is the same
set_node_attribute(thermostat, 1, 513, 17, 2500)
set_node_attribute(matter_node, 1, 513, 17, 2500)
await trigger_subscription_callback(hass, matter_client)
await hass.services.async_call(
"climate",
@ -197,7 +177,7 @@ async def test_thermostat_service_calls(
matter_client.write_attribute.reset_mock()
# test single-setpoint temperature adjustment when heat mode is active
set_node_attribute(thermostat, 1, 513, 28, 4)
set_node_attribute(matter_node, 1, 513, 28, 4)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
@ -215,14 +195,14 @@ async def test_thermostat_service_calls(
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=thermostat.node_id,
node_id=matter_node.node_id,
attribute_path="1/513/18",
value=2000,
)
matter_client.write_attribute.reset_mock()
# test dual setpoint temperature adjustments when heat_cool mode is active
set_node_attribute(thermostat, 1, 513, 28, 1)
set_node_attribute(matter_node, 1, 513, 28, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.longan_link_hvac_thermostat")
assert state
@ -241,12 +221,12 @@ async def test_thermostat_service_calls(
assert matter_client.write_attribute.call_count == 2
assert matter_client.write_attribute.call_args_list[0] == call(
node_id=thermostat.node_id,
node_id=matter_node.node_id,
attribute_path="1/513/18",
value=1000,
)
assert matter_client.write_attribute.call_args_list[1] == call(
node_id=thermostat.node_id,
node_id=matter_node.node_id,
attribute_path="1/513/17",
value=3000,
)
@ -265,7 +245,7 @@ async def test_thermostat_service_calls(
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=thermostat.node_id,
node_id=matter_node.node_id,
attribute_path=create_attribute_path_from_attribute(
endpoint_id=1,
attribute=clusters.Thermostat.Attributes.SystemMode,
@ -289,7 +269,7 @@ async def test_thermostat_service_calls(
)
assert matter_client.write_attribute.call_count == 2
assert matter_client.write_attribute.call_args_list[0] == call(
node_id=thermostat.node_id,
node_id=matter_node.node_id,
attribute_path=create_attribute_path_from_attribute(
endpoint_id=1,
attribute=clusters.Thermostat.Attributes.SystemMode,
@ -297,7 +277,7 @@ async def test_thermostat_service_calls(
value=3,
)
assert matter_client.write_attribute.call_args_list[1] == call(
node_id=thermostat.node_id,
node_id=matter_node.node_id,
attribute_path="1/513/17",
value=2200,
)
@ -306,10 +286,11 @@ async def test_thermostat_service_calls(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["room_airconditioner"])
async def test_room_airconditioner(
hass: HomeAssistant,
matter_client: MagicMock,
room_airconditioner: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test if a climate entity is created for a Room Airconditioner device."""
state = hass.states.get("climate.room_airconditioner_thermostat")
@ -324,7 +305,7 @@ async def test_room_airconditioner(
assert state.attributes["supported_features"] & mask == mask
# set mains power to ON (OnOff cluster)
set_node_attribute(room_airconditioner, 1, 6, 0, True)
set_node_attribute(matter_node, 1, 6, 0, True)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.room_airconditioner_thermostat")
@ -338,21 +319,21 @@ async def test_room_airconditioner(
HVACMode.HEAT_COOL,
]
# test fan-only hvac mode
set_node_attribute(room_airconditioner, 1, 513, 28, 7)
set_node_attribute(matter_node, 1, 513, 28, 7)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.room_airconditioner_thermostat")
assert state
assert state.state == HVACMode.FAN_ONLY
# test dry hvac mode
set_node_attribute(room_airconditioner, 1, 513, 28, 8)
set_node_attribute(matter_node, 1, 513, 28, 8)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.room_airconditioner_thermostat")
assert state
assert state.state == HVACMode.DRY
# test featuremap update
set_node_attribute(room_airconditioner, 1, 513, 65532, 1)
set_node_attribute(matter_node, 1, 513, 65532, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("climate.room_airconditioner_thermostat")
assert state.attributes["supported_features"] & ClimateEntityFeature.TURN_ON

182
tests/components/matter/test_cover.py
View File

@ -4,6 +4,7 @@ from math import floor
from unittest.mock import MagicMock, call
from chip.clusters import Objects as clusters
from matter_server.client.models.node import MatterNode
import pytest
from homeassistant.components.cover import (
@ -15,17 +16,13 @@ from homeassistant.components.cover import (
)
from homeassistant.core import HomeAssistant
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
trigger_subscription_callback,
)
from .common import set_node_attribute, trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("window_covering_lift", "cover.mock_lift_window_covering_cover"),
("window_covering_pa_lift", "cover.longan_link_wncv_da01_cover"),
@ -37,17 +34,11 @@ from .common import (
async def test_cover(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test window covering commands that always are implemented."""
window_covering = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
await hass.services.async_call(
"cover",
"close_cover",
@ -59,7 +50,7 @@ async def test_cover(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=window_covering.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.WindowCovering.Commands.DownOrClose(),
)
@ -76,7 +67,7 @@ async def test_cover(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=window_covering.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.WindowCovering.Commands.StopMotion(),
)
@ -93,7 +84,7 @@ async def test_cover(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=window_covering.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.WindowCovering.Commands.UpOrOpen(),
)
@ -103,7 +94,7 @@ async def test_cover(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("window_covering_lift", "cover.mock_lift_window_covering_cover"),
("window_covering_pa_lift", "cover.longan_link_wncv_da01_cover"),
@ -113,17 +104,10 @@ async def test_cover(
async def test_cover_lift(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test window covering devices with lift and position aware lift features."""
window_covering = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
await hass.services.async_call(
"cover",
"set_cover_position",
@ -136,20 +120,20 @@ async def test_cover_lift(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=window_covering.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.WindowCovering.Commands.GoToLiftPercentage(5000),
)
matter_client.send_device_command.reset_mock()
set_node_attribute(window_covering, 1, 258, 10, 0b001010)
set_node_attribute(matter_node, 1, 258, 10, 0b001010)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_CLOSING
set_node_attribute(window_covering, 1, 258, 10, 0b000101)
set_node_attribute(matter_node, 1, 258, 10, 0b000101)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -160,7 +144,7 @@ async def test_cover_lift(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("window_covering_lift", "cover.mock_lift_window_covering_cover"),
],
@ -168,33 +152,27 @@ async def test_cover_lift(
async def test_cover_lift_only(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test window covering devices with lift feature and without position aware lift feature."""
window_covering = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
set_node_attribute(window_covering, 1, 258, 14, None)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, None)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == "unknown"
set_node_attribute(window_covering, 1, 258, 65529, [0, 1, 2])
set_node_attribute(matter_node, 1, 258, 65529, [0, 1, 2])
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.attributes["supported_features"] & CoverEntityFeature.SET_POSITION == 0
set_node_attribute(window_covering, 1, 258, 65529, [0, 1, 2, 5])
set_node_attribute(matter_node, 1, 258, 65529, [0, 1, 2, 5])
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -205,7 +183,7 @@ async def test_cover_lift_only(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("window_covering_pa_lift", "cover.longan_link_wncv_da01_cover"),
],
@ -213,17 +191,11 @@ async def test_cover_lift_only(
async def test_cover_position_aware_lift(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test window covering devices with position aware lift features."""
window_covering = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
state = hass.states.get(entity_id)
assert state
mask = (
@ -235,8 +207,8 @@ async def test_cover_position_aware_lift(
assert state.attributes["supported_features"] & mask == mask
for position in (0, 9999):
set_node_attribute(window_covering, 1, 258, 14, position)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, position)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -244,8 +216,8 @@ async def test_cover_position_aware_lift(
assert state.attributes["current_position"] == 100 - floor(position / 100)
assert state.state == STATE_OPEN
set_node_attribute(window_covering, 1, 258, 14, 10000)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, 10000)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -257,7 +229,7 @@ async def test_cover_position_aware_lift(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("window_covering_tilt", "cover.mock_tilt_window_covering_cover"),
("window_covering_pa_tilt", "cover.mock_pa_tilt_window_covering_cover"),
@ -267,17 +239,11 @@ async def test_cover_position_aware_lift(
async def test_cover_tilt(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test window covering devices with tilt and position aware tilt features."""
window_covering = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
await hass.services.async_call(
"cover",
"set_cover_tilt_position",
@ -290,7 +256,7 @@ async def test_cover_tilt(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=window_covering.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.WindowCovering.Commands.GoToTiltPercentage(5000),
)
@ -298,13 +264,13 @@ async def test_cover_tilt(
await trigger_subscription_callback(hass, matter_client)
set_node_attribute(window_covering, 1, 258, 10, 0b100010)
set_node_attribute(matter_node, 1, 258, 10, 0b100010)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_CLOSING
set_node_attribute(window_covering, 1, 258, 10, 0b010001)
set_node_attribute(matter_node, 1, 258, 10, 0b010001)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -315,7 +281,7 @@ async def test_cover_tilt(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("window_covering_tilt", "cover.mock_tilt_window_covering_cover"),
],
@ -323,18 +289,12 @@ async def test_cover_tilt(
async def test_cover_tilt_only(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test window covering devices with tilt feature and without position aware tilt feature."""
window_covering = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
set_node_attribute(window_covering, 1, 258, 65529, [0, 1, 2])
set_node_attribute(matter_node, 1, 258, 65529, [0, 1, 2])
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -344,7 +304,7 @@ async def test_cover_tilt_only(
== 0
)
set_node_attribute(window_covering, 1, 258, 65529, [0, 1, 2, 8])
set_node_attribute(matter_node, 1, 258, 65529, [0, 1, 2, 8])
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -358,7 +318,7 @@ async def test_cover_tilt_only(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("window_covering_pa_tilt", "cover.mock_pa_tilt_window_covering_cover"),
],
@ -366,17 +326,11 @@ async def test_cover_tilt_only(
async def test_cover_position_aware_tilt(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test window covering devices with position aware tilt feature."""
window_covering = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
state = hass.states.get(entity_id)
assert state
mask = (
@ -388,8 +342,8 @@ async def test_cover_position_aware_tilt(
assert state.attributes["supported_features"] & mask == mask
for tilt_position in (0, 9999, 10000):
set_node_attribute(window_covering, 1, 258, 15, tilt_position)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 15, tilt_position)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -399,17 +353,13 @@ async def test_cover_position_aware_tilt(
)
@pytest.mark.parametrize("node_fixture", ["window_covering_full"])
async def test_cover_full_features(
hass: HomeAssistant,
matter_client: MagicMock,
matter_node: MatterNode,
) -> None:
"""Test window covering devices with all the features."""
window_covering = await setup_integration_with_node_fixture(
hass,
"window_covering_full",
matter_client,
)
entity_id = "cover.mock_full_window_covering_cover"
state = hass.states.get(entity_id)
@ -423,77 +373,77 @@ async def test_cover_full_features(
)
assert state.attributes["supported_features"] & mask == mask
set_node_attribute(window_covering, 1, 258, 14, 10000)
set_node_attribute(window_covering, 1, 258, 15, 10000)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, 10000)
set_node_attribute(matter_node, 1, 258, 15, 10000)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_CLOSED
set_node_attribute(window_covering, 1, 258, 14, 5000)
set_node_attribute(window_covering, 1, 258, 15, 10000)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, 5000)
set_node_attribute(matter_node, 1, 258, 15, 10000)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OPEN
set_node_attribute(window_covering, 1, 258, 14, 10000)
set_node_attribute(window_covering, 1, 258, 15, 5000)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, 10000)
set_node_attribute(matter_node, 1, 258, 15, 5000)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_CLOSED
set_node_attribute(window_covering, 1, 258, 14, 5000)
set_node_attribute(window_covering, 1, 258, 15, 5000)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, 5000)
set_node_attribute(matter_node, 1, 258, 15, 5000)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OPEN
set_node_attribute(window_covering, 1, 258, 14, 5000)
set_node_attribute(window_covering, 1, 258, 15, None)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, 5000)
set_node_attribute(matter_node, 1, 258, 15, None)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OPEN
set_node_attribute(window_covering, 1, 258, 14, None)
set_node_attribute(window_covering, 1, 258, 15, 5000)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, None)
set_node_attribute(matter_node, 1, 258, 15, 5000)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == "unknown"
set_node_attribute(window_covering, 1, 258, 14, 10000)
set_node_attribute(window_covering, 1, 258, 15, None)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, 10000)
set_node_attribute(matter_node, 1, 258, 15, None)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_CLOSED
set_node_attribute(window_covering, 1, 258, 14, None)
set_node_attribute(window_covering, 1, 258, 15, 10000)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, None)
set_node_attribute(matter_node, 1, 258, 15, 10000)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == "unknown"
set_node_attribute(window_covering, 1, 258, 14, None)
set_node_attribute(window_covering, 1, 258, 15, None)
set_node_attribute(window_covering, 1, 258, 10, 0b000000)
set_node_attribute(matter_node, 1, 258, 14, None)
set_node_attribute(matter_node, 1, 258, 15, None)
set_node_attribute(matter_node, 1, 258, 10, 0b000000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state

6
tests/components/matter/test_diagnostics.py
View File

@ -6,6 +6,7 @@ import json
from typing import Any
from unittest.mock import MagicMock
from matter_server.client.models.node import MatterNode
from matter_server.common.helpers.util import dataclass_from_dict
from matter_server.common.models import ServerDiagnostics
import pytest
@ -15,8 +16,6 @@ from homeassistant.components.matter.diagnostics import redact_matter_attributes
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, load_fixture
from tests.components.diagnostics import (
get_diagnostics_for_config_entry,
@ -79,6 +78,7 @@ async def test_config_entry_diagnostics(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["device_diagnostics"])
async def test_device_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
@ -86,9 +86,9 @@ async def test_device_diagnostics(
matter_client: MagicMock,
config_entry_diagnostics: dict[str, Any],
device_diagnostics: dict[str, Any],
matter_node: MatterNode,
) -> None:
"""Test the device diagnostics."""
await setup_integration_with_node_fixture(hass, "device_diagnostics", matter_client)
system_info_dict = config_entry_diagnostics["info"]
device_diagnostics_redacted = {
"server_info": system_info_dict,

32
tests/components/matter/test_event.py
View File

@ -9,35 +9,16 @@ import pytest
from homeassistant.components.event import ATTR_EVENT_TYPE, ATTR_EVENT_TYPES
from homeassistant.core import HomeAssistant
from .common import setup_integration_with_node_fixture, trigger_subscription_callback
@pytest.fixture(name="generic_switch_node")
async def switch_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a GenericSwitch node."""
return await setup_integration_with_node_fixture(
hass, "generic_switch", matter_client
)
@pytest.fixture(name="generic_switch_multi_node")
async def multi_switch_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a GenericSwitch node with multiple buttons."""
return await setup_integration_with_node_fixture(
hass, "generic_switch_multi", matter_client
)
from .common import trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["generic_switch"])
async def test_generic_switch_node(
hass: HomeAssistant,
matter_client: MagicMock,
generic_switch_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test event entity for a GenericSwitch node."""
state = hass.states.get("event.mock_generic_switch_button")
@ -57,7 +38,7 @@ async def test_generic_switch_node(
matter_client,
EventType.NODE_EVENT,
MatterNodeEvent(
node_id=generic_switch_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
cluster_id=59,
event_id=1,
@ -74,10 +55,11 @@ async def test_generic_switch_node(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["generic_switch_multi"])
async def test_generic_switch_multi_node(
hass: HomeAssistant,
matter_client: MagicMock,
generic_switch_multi_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test event entity for a GenericSwitch node with multiple buttons."""
state_button_1 = hass.states.get("event.mock_generic_switch_button_1")
@ -105,7 +87,7 @@ async def test_generic_switch_multi_node(
matter_client,
EventType.NODE_EVENT,
MatterNodeEvent(
node_id=generic_switch_multi_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
cluster_id=59,
event_id=6,

101
tests/components/matter/test_fan.py
View File

@ -1,6 +1,5 @@
"""Test Matter Fan platform."""
from typing import Any
from unittest.mock import MagicMock, call
from matter_server.client.models.node import MatterNode
@ -21,37 +20,16 @@ from homeassistant.components.fan import (
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF, SERVICE_TURN_ON
from homeassistant.core import HomeAssistant
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
trigger_subscription_callback,
)
@pytest.fixture(name="fan_node")
async def simple_fan_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Fan node."""
return await setup_integration_with_node_fixture(hass, "fan", matter_client)
@pytest.fixture(name="air_purifier")
async def air_purifier_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Air Purifier node (containing Fan cluster)."""
return await setup_integration_with_node_fixture(
hass, "air_purifier", matter_client
)
from .common import set_node_attribute, trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["air_purifier"])
async def test_fan_base(
hass: HomeAssistant,
matter_client: MagicMock,
air_purifier: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test Fan platform."""
entity_id = "fan.air_purifier_fan"
@ -78,47 +56,48 @@ async def test_fan_base(
)
assert state.attributes["supported_features"] & mask == mask
# handle fan mode update
set_node_attribute(air_purifier, 1, 514, 0, 1)
set_node_attribute(matter_node, 1, 514, 0, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state.attributes["preset_mode"] == "low"
# handle direction update
set_node_attribute(air_purifier, 1, 514, 11, 1)
set_node_attribute(matter_node, 1, 514, 11, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state.attributes["direction"] == "reverse"
# handle rock/oscillation update
set_node_attribute(air_purifier, 1, 514, 8, 1)
set_node_attribute(matter_node, 1, 514, 8, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state.attributes["oscillating"] is True
# handle wind mode active translates to correct preset
set_node_attribute(air_purifier, 1, 514, 10, 2)
set_node_attribute(matter_node, 1, 514, 10, 2)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state.attributes["preset_mode"] == "natural_wind"
set_node_attribute(air_purifier, 1, 514, 10, 1)
set_node_attribute(matter_node, 1, 514, 10, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state.attributes["preset_mode"] == "sleep_wind"
# set mains power to OFF (OnOff cluster)
set_node_attribute(air_purifier, 1, 6, 0, False)
set_node_attribute(matter_node, 1, 6, 0, False)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state.attributes["preset_mode"] is None
assert state.attributes["percentage"] == 0
# test featuremap update
set_node_attribute(air_purifier, 1, 514, 65532, 1)
set_node_attribute(matter_node, 1, 514, 65532, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state.attributes["supported_features"] & FanEntityFeature.SET_SPEED
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["air_purifier"])
async def test_fan_turn_on_with_percentage(
hass: HomeAssistant,
matter_client: MagicMock,
air_purifier: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test turning on the fan with a specific percentage."""
entity_id = "fan.air_purifier_fan"
@ -130,7 +109,7 @@ async def test_fan_turn_on_with_percentage(
)
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=air_purifier.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/2",
value=50,
)
@ -145,17 +124,18 @@ async def test_fan_turn_on_with_percentage(
)
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=air_purifier.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/2",
value=255,
)
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["fan"])
async def test_fan_turn_on_with_preset_mode(
hass: HomeAssistant,
matter_client: MagicMock,
fan_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test turning on the fan with a specific preset mode."""
entity_id = "fan.mocked_fan_switch_fan"
@ -167,7 +147,7 @@ async def test_fan_turn_on_with_preset_mode(
)
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=fan_node.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/0",
value=2,
)
@ -182,13 +162,13 @@ async def test_fan_turn_on_with_preset_mode(
)
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=fan_node.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/10",
value=value,
)
# test again if wind mode is explicitly turned off when we set a new preset mode
matter_client.write_attribute.reset_mock()
set_node_attribute(fan_node, 1, 514, 10, 2)
set_node_attribute(matter_node, 1, 514, 10, 2)
await trigger_subscription_callback(hass, matter_client)
await hass.services.async_call(
FAN_DOMAIN,
@ -198,20 +178,20 @@ async def test_fan_turn_on_with_preset_mode(
)
assert matter_client.write_attribute.call_count == 2
assert matter_client.write_attribute.call_args_list[0] == call(
node_id=fan_node.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/10",
value=0,
)
assert matter_client.write_attribute.call_args == call(
node_id=fan_node.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/0",
value=2,
)
# test again where preset_mode is omitted in the service call
# which should select the last active preset
matter_client.write_attribute.reset_mock()
set_node_attribute(fan_node, 1, 514, 0, 1)
set_node_attribute(fan_node, 1, 514, 10, 0)
set_node_attribute(matter_node, 1, 514, 0, 1)
set_node_attribute(matter_node, 1, 514, 10, 0)
await trigger_subscription_callback(hass, matter_client)
await hass.services.async_call(
FAN_DOMAIN,
@ -221,16 +201,17 @@ async def test_fan_turn_on_with_preset_mode(
)
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=fan_node.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/0",
value=1,
)
@pytest.mark.parametrize("node_fixture", ["air_purifier"])
async def test_fan_turn_off(
hass: HomeAssistant,
matter_client: MagicMock,
air_purifier: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test turning off the fan."""
entity_id = "fan.air_purifier_fan"
@ -242,13 +223,13 @@ async def test_fan_turn_off(
)
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=air_purifier.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/0",
value=0,
)
matter_client.write_attribute.reset_mock()
# test again if wind mode is turned off
set_node_attribute(air_purifier, 1, 514, 10, 2)
set_node_attribute(matter_node, 1, 514, 10, 2)
await trigger_subscription_callback(hass, matter_client)
await hass.services.async_call(
FAN_DOMAIN,
@ -258,21 +239,22 @@ async def test_fan_turn_off(
)
assert matter_client.write_attribute.call_count == 2
assert matter_client.write_attribute.call_args_list[0] == call(
node_id=air_purifier.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/10",
value=0,
)
assert matter_client.write_attribute.call_args_list[1] == call(
node_id=air_purifier.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/0",
value=0,
)
@pytest.mark.parametrize("node_fixture", ["air_purifier"])
async def test_fan_oscillate(
hass: HomeAssistant,
matter_client: MagicMock,
air_purifier: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test oscillating the fan."""
entity_id = "fan.air_purifier_fan"
@ -285,17 +267,18 @@ async def test_fan_oscillate(
)
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=air_purifier.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/8",
value=value,
)
matter_client.write_attribute.reset_mock()
@pytest.mark.parametrize("node_fixture", ["air_purifier"])
async def test_fan_set_direction(
hass: HomeAssistant,
matter_client: MagicMock,
air_purifier: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test oscillating the fan."""
entity_id = "fan.air_purifier_fan"
@ -308,7 +291,7 @@ async def test_fan_set_direction(
)
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args == call(
node_id=air_purifier.node_id,
node_id=matter_node.node_id,
attribute_path="1/514/11",
value=value,
)
@ -317,7 +300,7 @@ async def test_fan_set_direction(
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id", "attributes", "features"),
("node_fixture", "entity_id", "attributes", "features"),
[
(
"fan",
@ -369,13 +352,11 @@ async def test_fan_set_direction(
async def test_fan_supported_features(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
attributes: dict[str, Any],
features: int,
) -> None:
"""Test if the correct features get discovered from featuremap."""
await setup_integration_with_node_fixture(hass, fixture, matter_client, attributes)
state = hass.states.get(entity_id)
assert state
assert state.attributes["supported_features"] & features == features
@ -383,7 +364,7 @@ async def test_fan_supported_features(
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id", "attributes", "preset_modes"),
("node_fixture", "entity_id", "attributes", "preset_modes"),
[
(
"fan",
@ -433,13 +414,11 @@ async def test_fan_supported_features(
async def test_fan_features(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
attributes: dict[str, Any],
preset_modes: list[str],
) -> None:
"""Test if the correct presets get discovered from fanmodesequence."""
await setup_integration_with_node_fixture(hass, fixture, matter_client, attributes)
state = hass.states.get(entity_id)
assert state
assert state.attributes["preset_modes"] == preset_modes

8
tests/components/matter/test_helpers.py
View File

@ -4,6 +4,7 @@ from __future__ import annotations
from unittest.mock import MagicMock
from matter_server.client.models.node import MatterNode
import pytest
from homeassistant.components.matter.const import DOMAIN
@ -21,15 +22,14 @@ from tests.common import MockConfigEntry
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["device_diagnostics"])
async def test_get_device_id(
hass: HomeAssistant,
matter_client: MagicMock,
matter_node: MatterNode,
) -> None:
"""Test get_device_id."""
node = await setup_integration_with_node_fixture(
hass, "device_diagnostics", matter_client
)
device_id = get_device_id(matter_client.server_info, node.endpoints[0])
device_id = get_device_id(matter_client.server_info, matter_node.endpoints[0])
assert device_id == "00000000000004D2-0000000000000005-MatterNodeDevice"

98
tests/components/matter/test_light.py
View File

@ -3,22 +3,19 @@
from unittest.mock import MagicMock, call
from chip.clusters import Objects as clusters
from matter_server.client.models.node import MatterNode
import pytest
from homeassistant.components.light import ColorMode
from homeassistant.core import HomeAssistant
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
trigger_subscription_callback,
)
from .common import set_node_attribute, trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id", "supported_color_modes"),
("node_fixture", "entity_id", "supported_color_modes"),
[
(
"extended_color_light",
@ -38,20 +35,14 @@ from .common import (
async def test_light_turn_on_off(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
supported_color_modes: list[str],
) -> None:
"""Test basic light discovery and turn on/off."""
light_node = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
# Test that the light is off
set_node_attribute(light_node, 1, 6, 0, False)
set_node_attribute(matter_node, 1, 6, 0, False)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -64,7 +55,7 @@ async def test_light_turn_on_off(
assert state.attributes["supported_color_modes"] == supported_color_modes
# Test that the light is on
set_node_attribute(light_node, 1, 6, 0, True)
set_node_attribute(matter_node, 1, 6, 0, True)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -83,7 +74,7 @@ async def test_light_turn_on_off(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.Off(),
)
@ -101,7 +92,7 @@ async def test_light_turn_on_off(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.On(),
)
@ -111,7 +102,7 @@ async def test_light_turn_on_off(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("extended_color_light", "light.mock_extended_color_light_light"),
("color_temperature_light", "light.mock_color_temperature_light_light"),
@ -122,19 +113,13 @@ async def test_light_turn_on_off(
async def test_dimmable_light(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test a dimmable light."""
light_node = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
# Test that the light brightness is 50 (out of 254)
set_node_attribute(light_node, 1, 8, 0, 50)
set_node_attribute(matter_node, 1, 8, 0, 50)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -155,7 +140,7 @@ async def test_dimmable_light(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.LevelControl.Commands.MoveToLevelWithOnOff(
level=128,
@ -174,7 +159,7 @@ async def test_dimmable_light(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.LevelControl.Commands.MoveToLevelWithOnOff(
level=128,
@ -187,7 +172,7 @@ async def test_dimmable_light(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("extended_color_light", "light.mock_extended_color_light_light"),
("color_temperature_light", "light.mock_color_temperature_light_light"),
@ -196,20 +181,13 @@ async def test_dimmable_light(
async def test_color_temperature_light(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test a color temperature light."""
light_node = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
# Test that the light color temperature is 3000 (out of 50000)
set_node_attribute(light_node, 1, 768, 8, 2)
set_node_attribute(light_node, 1, 768, 7, 3000)
set_node_attribute(matter_node, 1, 768, 8, 2)
set_node_attribute(matter_node, 1, 768, 7, 3000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -233,7 +211,7 @@ async def test_color_temperature_light(
matter_client.send_device_command.assert_has_calls(
[
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.ColorControl.Commands.MoveToColorTemperature(
colorTemperatureMireds=300,
@ -243,7 +221,7 @@ async def test_color_temperature_light(
),
),
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.On(),
),
@ -263,7 +241,7 @@ async def test_color_temperature_light(
matter_client.send_device_command.assert_has_calls(
[
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.ColorControl.Commands.MoveToColorTemperature(
colorTemperatureMireds=300,
@ -273,7 +251,7 @@ async def test_color_temperature_light(
),
),
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.On(),
),
@ -285,7 +263,7 @@ async def test_color_temperature_light(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize(
("fixture", "entity_id"),
("node_fixture", "entity_id"),
[
("extended_color_light", "light.mock_extended_color_light_light"),
],
@ -293,21 +271,15 @@ async def test_color_temperature_light(
async def test_extended_color_light(
hass: HomeAssistant,
matter_client: MagicMock,
fixture: str,
matter_node: MatterNode,
entity_id: str,
) -> None:
"""Test an extended color light."""
light_node = await setup_integration_with_node_fixture(
hass,
fixture,
matter_client,
)
# Test that the XY color changes
set_node_attribute(light_node, 1, 768, 8, 1)
set_node_attribute(light_node, 1, 768, 3, 50)
set_node_attribute(light_node, 1, 768, 4, 100)
set_node_attribute(matter_node, 1, 768, 8, 1)
set_node_attribute(matter_node, 1, 768, 3, 50)
set_node_attribute(matter_node, 1, 768, 4, 100)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -317,9 +289,9 @@ async def test_extended_color_light(
assert state.attributes["xy_color"] == (0.0007630, 0.001526)
# Test that the HS color changes
set_node_attribute(light_node, 1, 768, 8, 0)
set_node_attribute(light_node, 1, 768, 1, 50)
set_node_attribute(light_node, 1, 768, 0, 100)
set_node_attribute(matter_node, 1, 768, 8, 0)
set_node_attribute(matter_node, 1, 768, 1, 50)
set_node_attribute(matter_node, 1, 768, 0, 100)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -343,7 +315,7 @@ async def test_extended_color_light(
matter_client.send_device_command.assert_has_calls(
[
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.ColorControl.Commands.MoveToColor(
colorX=0.5 * 65536,
@ -354,7 +326,7 @@ async def test_extended_color_light(
),
),
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.On(),
),
@ -374,7 +346,7 @@ async def test_extended_color_light(
matter_client.send_device_command.assert_has_calls(
[
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.ColorControl.Commands.MoveToColor(
colorX=0.5 * 65536,
@ -385,7 +357,7 @@ async def test_extended_color_light(
),
),
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.On(),
),
@ -419,7 +391,7 @@ async def test_extended_color_light(
),
),
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.On(),
),
@ -454,7 +426,7 @@ async def test_extended_color_light(
),
),
call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.On(),
),

39
tests/components/matter/test_lock.py
View File

@ -17,10 +17,11 @@ from .common import set_node_attribute, trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["door_lock"])
async def test_lock(
hass: HomeAssistant,
matter_client: MagicMock,
door_lock: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test door lock."""
await hass.services.async_call(
@ -34,7 +35,7 @@ async def test_lock(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=door_lock.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.DoorLock.Commands.UnlockDoor(),
timed_request_timeout_ms=1000,
@ -52,7 +53,7 @@ async def test_lock(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=door_lock.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.DoorLock.Commands.LockDoor(),
timed_request_timeout_ms=1000,
@ -64,28 +65,28 @@ async def test_lock(
assert state
assert state.state == LockState.LOCKING
set_node_attribute(door_lock, 1, 257, 0, 0)
set_node_attribute(matter_node, 1, 257, 0, 0)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("lock.mock_door_lock_lock")
assert state
assert state.state == LockState.UNLOCKED
set_node_attribute(door_lock, 1, 257, 0, 2)
set_node_attribute(matter_node, 1, 257, 0, 2)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("lock.mock_door_lock_lock")
assert state
assert state.state == LockState.UNLOCKED
set_node_attribute(door_lock, 1, 257, 0, 1)
set_node_attribute(matter_node, 1, 257, 0, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("lock.mock_door_lock_lock")
assert state
assert state.state == LockState.LOCKED
set_node_attribute(door_lock, 1, 257, 0, None)
set_node_attribute(matter_node, 1, 257, 0, None)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("lock.mock_door_lock_lock")
@ -93,7 +94,7 @@ async def test_lock(
assert state.state == STATE_UNKNOWN
# test featuremap update
set_node_attribute(door_lock, 1, 257, 65532, 4096)
set_node_attribute(matter_node, 1, 257, 65532, 4096)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("lock.mock_door_lock_lock")
assert state.attributes["supported_features"] & LockEntityFeature.OPEN
@ -101,10 +102,11 @@ async def test_lock(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["door_lock"])
async def test_lock_requires_pin(
hass: HomeAssistant,
matter_client: MagicMock,
door_lock: MatterNode,
matter_node: MatterNode,
entity_registry: er.EntityRegistry,
) -> None:
"""Test door lock with PINCode."""
@ -112,9 +114,9 @@ async def test_lock_requires_pin(
code = "1234567"
# set RequirePINforRemoteOperation
set_node_attribute(door_lock, 1, 257, 51, True)
set_node_attribute(matter_node, 1, 257, 51, True)
# set door state to unlocked
set_node_attribute(door_lock, 1, 257, 0, 2)
set_node_attribute(matter_node, 1, 257, 0, 2)
await trigger_subscription_callback(hass, matter_client)
with pytest.raises(ServiceValidationError):
@ -136,7 +138,7 @@ async def test_lock_requires_pin(
)
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=door_lock.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.DoorLock.Commands.LockDoor(code.encode()),
timed_request_timeout_ms=1000,
@ -156,7 +158,7 @@ async def test_lock_requires_pin(
)
assert matter_client.send_device_command.call_count == 2
assert matter_client.send_device_command.call_args == call(
node_id=door_lock.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.DoorLock.Commands.LockDoor(default_code.encode()),
timed_request_timeout_ms=1000,
@ -165,10 +167,11 @@ async def test_lock_requires_pin(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["door_lock_with_unbolt"])
async def test_lock_with_unbolt(
hass: HomeAssistant,
matter_client: MagicMock,
door_lock_with_unbolt: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test door lock."""
state = hass.states.get("lock.mock_door_lock_lock")
@ -187,7 +190,7 @@ async def test_lock_with_unbolt(
assert matter_client.send_device_command.call_count == 1
# unlock should unbolt on a lock with unbolt feature
assert matter_client.send_device_command.call_args == call(
node_id=door_lock_with_unbolt.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.DoorLock.Commands.UnboltDoor(),
timed_request_timeout_ms=1000,
@ -204,7 +207,7 @@ async def test_lock_with_unbolt(
)
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=door_lock_with_unbolt.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.DoorLock.Commands.UnlockDoor(),
timed_request_timeout_ms=1000,
@ -215,14 +218,14 @@ async def test_lock_with_unbolt(
assert state
assert state.state == LockState.OPENING
set_node_attribute(door_lock_with_unbolt, 1, 257, 0, 0)
set_node_attribute(matter_node, 1, 257, 0, 0)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("lock.mock_door_lock_lock")
assert state
assert state.state == LockState.UNLOCKED
set_node_attribute(door_lock_with_unbolt, 1, 257, 0, 3)
set_node_attribute(matter_node, 1, 257, 0, 3)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("lock.mock_door_lock_lock")

38
tests/components/matter/test_number.py
View File

@ -9,39 +9,16 @@ import pytest
from homeassistant.core import HomeAssistant
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
trigger_subscription_callback,
)
@pytest.fixture(name="light_node")
async def dimmable_light_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a flow sensor node."""
return await setup_integration_with_node_fixture(
hass, "dimmable_light", matter_client
)
@pytest.fixture(name="eve_weather_sensor_node")
async def eve_weather_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Eve Weather sensor node."""
return await setup_integration_with_node_fixture(
hass, "eve_weather_sensor", matter_client
)
from .common import set_node_attribute, trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["dimmable_light"])
async def test_level_control_config_entities(
hass: HomeAssistant,
matter_client: MagicMock,
light_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test number entities are created for the LevelControl cluster (config) attributes."""
state = hass.states.get("number.mock_dimmable_light_on_level")
@ -60,7 +37,7 @@ async def test_level_control_config_entities(
assert state
assert state.state == "0.0"
set_node_attribute(light_node, 1, 0x00000008, 0x0011, 20)
set_node_attribute(matter_node, 1, 0x00000008, 0x0011, 20)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("number.mock_dimmable_light_on_level")
@ -68,10 +45,11 @@ async def test_level_control_config_entities(
assert state.state == "20"
@pytest.mark.parametrize("node_fixture", ["eve_weather_sensor"])
async def test_eve_weather_sensor_altitude(
hass: HomeAssistant,
matter_client: MagicMock,
eve_weather_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test weather sensor created from (Eve) custom cluster."""
# pressure sensor on Eve custom cluster
@ -79,7 +57,7 @@ async def test_eve_weather_sensor_altitude(
assert state
assert state.state == "40.0"
set_node_attribute(eve_weather_sensor_node, 1, 319486977, 319422483, 800)
set_node_attribute(matter_node, 1, 319486977, 319422483, 800)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("number.eve_weather_altitude_above_sea_level")
assert state
@ -97,7 +75,7 @@ async def test_eve_weather_sensor_altitude(
)
assert matter_client.write_attribute.call_count == 1
assert matter_client.write_attribute.call_args_list[0] == call(
node_id=eve_weather_sensor_node.node_id,
node_id=matter_node.node_id,
attribute_path=create_attribute_path_from_attribute(
endpoint_id=1,
attribute=custom_clusters.EveCluster.Attributes.Altitude,

30
tests/components/matter/test_select.py
View File

@ -8,29 +8,16 @@ import pytest
from homeassistant.core import HomeAssistant
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
trigger_subscription_callback,
)
@pytest.fixture(name="light_node")
async def dimmable_light_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a dimmable light node."""
return await setup_integration_with_node_fixture(
hass, "dimmable_light", matter_client
)
from .common import set_node_attribute, trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["dimmable_light"])
async def test_mode_select_entities(
hass: HomeAssistant,
matter_client: MagicMock,
light_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test select entities are created for the ModeSelect cluster attributes."""
state = hass.states.get("select.mock_dimmable_light_led_color")
@ -53,7 +40,7 @@ async def test_mode_select_entities(
]
# name should be derived from description attribute
assert state.attributes["friendly_name"] == "Mock Dimmable Light LED Color"
set_node_attribute(light_node, 6, 80, 3, 1)
set_node_attribute(matter_node, 6, 80, 3, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("select.mock_dimmable_light_led_color")
assert state.state == "Orange"
@ -70,7 +57,7 @@ async def test_mode_select_entities(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=light_node.node_id,
node_id=matter_node.node_id,
endpoint_id=6,
command=clusters.ModeSelect.Commands.ChangeToMode(newMode=3),
)
@ -78,10 +65,11 @@ async def test_mode_select_entities(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["dimmable_light"])
async def test_attribute_select_entities(
hass: HomeAssistant,
matter_client: MagicMock,
light_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test select entities are created for attribute based discovery schema(s)."""
entity_id = "select.mock_dimmable_light_power_on_behavior_on_startup"
@ -93,12 +81,12 @@ async def test_attribute_select_entities(
state.attributes["friendly_name"]
== "Mock Dimmable Light Power-on behavior on startup"
)
set_node_attribute(light_node, 1, 6, 16387, 1)
set_node_attribute(matter_node, 1, 6, 16387, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state.state == "on"
# test that an invalid value (e.g. 253) leads to an unknown state
set_node_attribute(light_node, 1, 6, 16387, 253)
set_node_attribute(matter_node, 1, 6, 16387, 253)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state.state == "unknown"

204
tests/components/matter/test_sensor.py
View File

@ -12,141 +12,37 @@ from homeassistant.helpers import entity_registry as er
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
snapshot_matter_entities,
trigger_subscription_callback,
)
@pytest.fixture(name="flow_sensor_node")
async def flow_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a flow sensor node."""
return await setup_integration_with_node_fixture(hass, "flow_sensor", matter_client)
@pytest.fixture(name="humidity_sensor_node")
async def humidity_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a humidity sensor node."""
return await setup_integration_with_node_fixture(
hass, "humidity_sensor", matter_client
)
@pytest.fixture(name="light_sensor_node")
async def light_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a light sensor node."""
return await setup_integration_with_node_fixture(
hass, "light_sensor", matter_client
)
@pytest.fixture(name="pressure_sensor_node")
async def pressure_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a pressure sensor node."""
return await setup_integration_with_node_fixture(
hass, "pressure_sensor", matter_client
)
@pytest.fixture(name="temperature_sensor_node")
async def temperature_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a temperature sensor node."""
return await setup_integration_with_node_fixture(
hass, "temperature_sensor", matter_client
)
@pytest.fixture(name="eve_energy_plug_node")
async def eve_energy_plug_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Eve Energy Plug node."""
return await setup_integration_with_node_fixture(
hass, "eve_energy_plug", matter_client
)
@pytest.fixture(name="eve_thermo_node")
async def eve_thermo_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Eve Thermo node."""
return await setup_integration_with_node_fixture(hass, "eve_thermo", matter_client)
@pytest.fixture(name="eve_energy_plug_patched_node")
async def eve_energy_plug_patched_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Eve Energy Plug node (patched to include Matter 1.3 energy clusters)."""
return await setup_integration_with_node_fixture(
hass, "eve_energy_plug_patched", matter_client
)
@pytest.fixture(name="eve_weather_sensor_node")
async def eve_weather_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Eve Weather sensor node."""
return await setup_integration_with_node_fixture(
hass, "eve_weather_sensor", matter_client
)
@pytest.fixture(name="air_quality_sensor_node")
async def air_quality_sensor_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for an air quality sensor (LightFi AQ1) node."""
return await setup_integration_with_node_fixture(
hass, "air_quality_sensor", matter_client
)
@pytest.fixture(name="air_purifier_node")
async def air_purifier_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for an air purifier node."""
return await setup_integration_with_node_fixture(
hass, "air_purifier", matter_client
)
@pytest.fixture(name="dishwasher_node")
async def dishwasher_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for an dishwasher node."""
return await setup_integration_with_node_fixture(
hass, "silabs_dishwasher", matter_client
)
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.usefixtures("matter_devices")
async def test_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test sensors."""
snapshot_matter_entities(hass, entity_registry, snapshot, Platform.SENSOR)
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["flow_sensor"])
async def test_sensor_null_value(
hass: HomeAssistant,
matter_client: MagicMock,
flow_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test flow sensor."""
state = hass.states.get("sensor.mock_flow_sensor_flow")
assert state
assert state.state == "0.0"
set_node_attribute(flow_sensor_node, 1, 1028, 0, None)
set_node_attribute(matter_node, 1, 1028, 0, None)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.mock_flow_sensor_flow")
@ -156,17 +52,18 @@ async def test_sensor_null_value(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["flow_sensor"])
async def test_flow_sensor(
hass: HomeAssistant,
matter_client: MagicMock,
flow_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test flow sensor."""
state = hass.states.get("sensor.mock_flow_sensor_flow")
assert state
assert state.state == "0.0"
set_node_attribute(flow_sensor_node, 1, 1028, 0, 20)
set_node_attribute(matter_node, 1, 1028, 0, 20)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.mock_flow_sensor_flow")
@ -176,17 +73,18 @@ async def test_flow_sensor(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["humidity_sensor"])
async def test_humidity_sensor(
hass: HomeAssistant,
matter_client: MagicMock,
humidity_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test humidity sensor."""
state = hass.states.get("sensor.mock_humidity_sensor_humidity")
assert state
assert state.state == "0.0"
set_node_attribute(humidity_sensor_node, 1, 1029, 0, 4000)
set_node_attribute(matter_node, 1, 1029, 0, 4000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.mock_humidity_sensor_humidity")
@ -196,17 +94,18 @@ async def test_humidity_sensor(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["light_sensor"])
async def test_light_sensor(
hass: HomeAssistant,
matter_client: MagicMock,
light_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test light sensor."""
state = hass.states.get("sensor.mock_light_sensor_illuminance")
assert state
assert state.state == "1.3"
set_node_attribute(light_sensor_node, 1, 1024, 0, 3000)
set_node_attribute(matter_node, 1, 1024, 0, 3000)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.mock_light_sensor_illuminance")
@ -216,17 +115,18 @@ async def test_light_sensor(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["temperature_sensor"])
async def test_temperature_sensor(
hass: HomeAssistant,
matter_client: MagicMock,
temperature_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test temperature sensor."""
state = hass.states.get("sensor.mock_temperature_sensor_temperature")
assert state
assert state.state == "21.0"
set_node_attribute(temperature_sensor_node, 1, 1026, 0, 2500)
set_node_attribute(matter_node, 1, 1026, 0, 2500)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.mock_temperature_sensor_temperature")
@ -236,11 +136,12 @@ async def test_temperature_sensor(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["eve_contact_sensor"])
async def test_battery_sensor(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
matter_client: MagicMock,
eve_contact_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test battery sensor."""
entity_id = "sensor.eve_door_battery"
@ -248,7 +149,7 @@ async def test_battery_sensor(
assert state
assert state.state == "100"
set_node_attribute(eve_contact_sensor_node, 1, 47, 12, 100)
set_node_attribute(matter_node, 1, 47, 12, 100)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -263,11 +164,12 @@ async def test_battery_sensor(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["eve_contact_sensor"])
async def test_battery_sensor_voltage(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
matter_client: MagicMock,
eve_contact_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test battery voltage sensor."""
entity_id = "sensor.eve_door_voltage"
@ -275,7 +177,7 @@ async def test_battery_sensor_voltage(
assert state
assert state.state == "3.558"
set_node_attribute(eve_contact_sensor_node, 1, 47, 11, 4234)
set_node_attribute(matter_node, 1, 47, 11, 4234)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
@ -290,10 +192,11 @@ async def test_battery_sensor_voltage(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["eve_thermo"])
async def test_eve_thermo_sensor(
hass: HomeAssistant,
matter_client: MagicMock,
eve_thermo_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test Eve Thermo."""
# Valve position
@ -301,7 +204,7 @@ async def test_eve_thermo_sensor(
assert state
assert state.state == "10"
set_node_attribute(eve_thermo_node, 1, 319486977, 319422488, 0)
set_node_attribute(matter_node, 1, 319486977, 319422488, 0)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.eve_thermo_valve_position")
@ -311,17 +214,18 @@ async def test_eve_thermo_sensor(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["pressure_sensor"])
async def test_pressure_sensor(
hass: HomeAssistant,
matter_client: MagicMock,
pressure_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test pressure sensor."""
state = hass.states.get("sensor.mock_pressure_sensor_pressure")
assert state
assert state.state == "0.0"
set_node_attribute(pressure_sensor_node, 1, 1027, 0, 1010)
set_node_attribute(matter_node, 1, 1027, 0, 1010)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.mock_pressure_sensor_pressure")
@ -329,10 +233,11 @@ async def test_pressure_sensor(
assert state.state == "101.0"
@pytest.mark.parametrize("node_fixture", ["eve_weather_sensor"])
async def test_eve_weather_sensor_custom_cluster(
hass: HomeAssistant,
matter_client: MagicMock,
eve_weather_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test weather sensor created from (Eve) custom cluster."""
# pressure sensor on Eve custom cluster
@ -340,7 +245,7 @@ async def test_eve_weather_sensor_custom_cluster(
assert state
assert state.state == "1008.5"
set_node_attribute(eve_weather_sensor_node, 1, 319486977, 319422484, 800)
set_node_attribute(matter_node, 1, 319486977, 319422484, 800)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.eve_weather_pressure")
assert state
@ -349,10 +254,11 @@ async def test_eve_weather_sensor_custom_cluster(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["air_quality_sensor"])
async def test_air_quality_sensor(
hass: HomeAssistant,
matter_client: MagicMock,
air_quality_sensor_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test air quality sensor."""
# Carbon Dioxide
@ -360,7 +266,7 @@ async def test_air_quality_sensor(
assert state
assert state.state == "678.0"
set_node_attribute(air_quality_sensor_node, 1, 1037, 0, 789)
set_node_attribute(matter_node, 1, 1037, 0, 789)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.lightfi_aq1_air_quality_sensor_carbon_dioxide")
@ -372,7 +278,7 @@ async def test_air_quality_sensor(
assert state
assert state.state == "3.0"
set_node_attribute(air_quality_sensor_node, 1, 1068, 0, 50)
set_node_attribute(matter_node, 1, 1068, 0, 50)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.lightfi_aq1_air_quality_sensor_pm1")
@ -384,7 +290,7 @@ async def test_air_quality_sensor(
assert state
assert state.state == "3.0"
set_node_attribute(air_quality_sensor_node, 1, 1066, 0, 50)
set_node_attribute(matter_node, 1, 1066, 0, 50)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.lightfi_aq1_air_quality_sensor_pm2_5")
@ -396,7 +302,7 @@ async def test_air_quality_sensor(
assert state
assert state.state == "3.0"
set_node_attribute(air_quality_sensor_node, 1, 1069, 0, 50)
set_node_attribute(matter_node, 1, 1069, 0, 50)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.lightfi_aq1_air_quality_sensor_pm10")
@ -404,10 +310,11 @@ async def test_air_quality_sensor(
assert state.state == "50.0"
@pytest.mark.parametrize("node_fixture", ["silabs_dishwasher"])
async def test_operational_state_sensor(
hass: HomeAssistant,
matter_client: MagicMock,
dishwasher_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test dishwasher sensor."""
# OperationalState Cluster / OperationalState attribute (1/96/4)
@ -422,22 +329,9 @@ async def test_operational_state_sensor(
"extra_state",
]
set_node_attribute(dishwasher_node, 1, 96, 4, 8)
set_node_attribute(matter_node, 1, 96, 4, 8)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("sensor.dishwasher_operational_state")
assert state
assert state.state == "extra_state"
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
async def test_sensors(
hass: HomeAssistant,
matter_client: MagicMock,
matter_devices: MatterNode,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test sensors."""
snapshot_matter_entities(hass, entity_registry, snapshot, Platform.SENSOR)

52
tests/components/matter/test_switch.py
View File

@ -8,37 +8,16 @@ import pytest
from homeassistant.core import HomeAssistant
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
trigger_subscription_callback,
)
@pytest.fixture(name="powerplug_node")
async def powerplug_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Powerplug node."""
return await setup_integration_with_node_fixture(
hass, "on_off_plugin_unit", matter_client
)
@pytest.fixture(name="switch_unit")
async def switch_unit_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a Switch Unit node."""
return await setup_integration_with_node_fixture(hass, "switch_unit", matter_client)
from .common import set_node_attribute, trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["on_off_plugin_unit"])
async def test_turn_on(
hass: HomeAssistant,
matter_client: MagicMock,
powerplug_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test turning on a switch."""
state = hass.states.get("switch.mock_onoffpluginunit_switch")
@ -56,12 +35,12 @@ async def test_turn_on(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=powerplug_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.On(),
)
set_node_attribute(powerplug_node, 1, 6, 0, True)
set_node_attribute(matter_node, 1, 6, 0, True)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get("switch.mock_onoffpluginunit_switch")
@ -71,10 +50,11 @@ async def test_turn_on(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["on_off_plugin_unit"])
async def test_turn_off(
hass: HomeAssistant,
matter_client: MagicMock,
powerplug_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test turning off a switch."""
state = hass.states.get("switch.mock_onoffpluginunit_switch")
@ -92,7 +72,7 @@ async def test_turn_off(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=powerplug_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.OnOff.Commands.Off(),
)
@ -100,11 +80,8 @@ async def test_turn_off(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
async def test_switch_unit(
hass: HomeAssistant,
matter_client: MagicMock,
switch_unit: MatterNode,
) -> None:
@pytest.mark.parametrize("node_fixture", ["switch_unit"])
async def test_switch_unit(hass: HomeAssistant, matter_node: MatterNode) -> None:
"""Test if a switch entity is discovered from any (non-light) OnOf cluster device."""
# A switch entity should be discovered as fallback for ANY Matter device (endpoint)
# that has the OnOff cluster and does not fall into an explicit discovery schema
@ -117,14 +94,9 @@ async def test_switch_unit(
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
async def test_power_switch(
hass: HomeAssistant,
matter_client: MagicMock,
) -> None:
@pytest.mark.parametrize("node_fixture", ["room_airconditioner"])
async def test_power_switch(hass: HomeAssistant, matter_node: MatterNode) -> None:
"""Test if a Power switch entity is created for a device that supports that."""
await setup_integration_with_node_fixture(
hass, "room_airconditioner", matter_client
)
state = hass.states.get("switch.room_airconditioner_power")
assert state
assert state.state == "off"

35
tests/components/matter/test_update.py
View File

@ -78,21 +78,12 @@ async def update_node_fixture(matter_client: MagicMock) -> AsyncMock:
return matter_client.update_node
@pytest.fixture(name="updateable_node")
async def updateable_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a flow sensor node."""
return await setup_integration_with_node_fixture(
hass, "dimmable_light", matter_client
)
@pytest.mark.parametrize("node_fixture", ["dimmable_light"])
async def test_update_entity(
hass: HomeAssistant,
matter_client: MagicMock,
check_node_update: AsyncMock,
updateable_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test update entity exists and update check got made."""
state = hass.states.get("update.mock_dimmable_light")
@ -102,11 +93,12 @@ async def test_update_entity(
assert matter_client.check_node_update.call_count == 1
@pytest.mark.parametrize("node_fixture", ["dimmable_light"])
async def test_update_check_service(
hass: HomeAssistant,
matter_client: MagicMock,
check_node_update: AsyncMock,
updateable_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test check device update through service call."""
state = hass.states.get("update.mock_dimmable_light")
@ -149,11 +141,12 @@ async def test_update_check_service(
)
@pytest.mark.parametrize("node_fixture", ["dimmable_light"])
async def test_update_install(
hass: HomeAssistant,
matter_client: MagicMock,
check_node_update: AsyncMock,
updateable_node: MatterNode,
matter_node: MatterNode,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test device update with Matter attribute changes influence progress."""
@ -199,7 +192,7 @@ async def test_update_install(
)
set_node_attribute_typed(
updateable_node,
matter_node,
0,
clusters.OtaSoftwareUpdateRequestor.Attributes.UpdateState,
clusters.OtaSoftwareUpdateRequestor.Enums.UpdateStateEnum.kDownloading,
@ -212,7 +205,7 @@ async def test_update_install(
assert state.attributes.get("in_progress")
set_node_attribute_typed(
updateable_node,
matter_node,
0,
clusters.OtaSoftwareUpdateRequestor.Attributes.UpdateStateProgress,
50,
@ -225,19 +218,19 @@ async def test_update_install(
assert state.attributes.get("in_progress") == 50
set_node_attribute_typed(
updateable_node,
matter_node,
0,
clusters.OtaSoftwareUpdateRequestor.Attributes.UpdateState,
clusters.OtaSoftwareUpdateRequestor.Enums.UpdateStateEnum.kIdle,
)
set_node_attribute_typed(
updateable_node,
matter_node,
0,
clusters.BasicInformation.Attributes.SoftwareVersion,
2,
)
set_node_attribute_typed(
updateable_node,
matter_node,
0,
clusters.BasicInformation.Attributes.SoftwareVersionString,
"v2.0",
@ -249,12 +242,13 @@ async def test_update_install(
assert state.attributes.get("installed_version") == "v2.0"
@pytest.mark.parametrize("node_fixture", ["dimmable_light"])
async def test_update_install_failure(
hass: HomeAssistant,
matter_client: MagicMock,
check_node_update: AsyncMock,
update_node: AsyncMock,
updateable_node: MatterNode,
matter_node: MatterNode,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test update entity service call errors."""
@ -317,12 +311,13 @@ async def test_update_install_failure(
)
@pytest.mark.parametrize("node_fixture", ["dimmable_light"])
async def test_update_state_save_and_restore(
hass: HomeAssistant,
hass_storage: dict[str, Any],
matter_client: MagicMock,
check_node_update: AsyncMock,
updateable_node: MatterNode,
matter_node: MatterNode,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test latest update information is retained across reload/restart."""

39
tests/components/matter/test_valve.py
View File

@ -8,27 +8,16 @@ import pytest
from homeassistant.core import HomeAssistant
from .common import (
set_node_attribute,
setup_integration_with_node_fixture,
trigger_subscription_callback,
)
@pytest.fixture(name="valve_node")
async def valve_node_fixture(
hass: HomeAssistant, matter_client: MagicMock
) -> MatterNode:
"""Fixture for a valve node."""
return await setup_integration_with_node_fixture(hass, "valve", matter_client)
from .common import set_node_attribute, trigger_subscription_callback
# This tests needs to be adjusted to remove lingering tasks
@pytest.mark.parametrize("expected_lingering_tasks", [True])
@pytest.mark.parametrize("node_fixture", ["valve"])
async def test_valve(
hass: HomeAssistant,
matter_client: MagicMock,
valve_node: MatterNode,
matter_node: MatterNode,
) -> None:
"""Test valve entity is created for a Matter ValveConfigurationAndControl Cluster."""
entity_id = "valve.valve_valve"
@ -49,7 +38,7 @@ async def test_valve(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=valve_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.ValveConfigurationAndControl.Commands.Close(),
)
@ -67,45 +56,45 @@ async def test_valve(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=valve_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.ValveConfigurationAndControl.Commands.Open(),
)
matter_client.send_device_command.reset_mock()
# set changing state to 'opening'
set_node_attribute(valve_node, 1, 129, 4, 2)
set_node_attribute(valve_node, 1, 129, 5, 1)
set_node_attribute(matter_node, 1, 129, 4, 2)
set_node_attribute(matter_node, 1, 129, 5, 1)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == "opening"
# set changing state to 'closing'
set_node_attribute(valve_node, 1, 129, 4, 2)
set_node_attribute(valve_node, 1, 129, 5, 0)
set_node_attribute(matter_node, 1, 129, 4, 2)
set_node_attribute(matter_node, 1, 129, 5, 0)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == "closing"
# set changing state to 'open'
set_node_attribute(valve_node, 1, 129, 4, 1)
set_node_attribute(valve_node, 1, 129, 5, 0)
set_node_attribute(matter_node, 1, 129, 4, 1)
set_node_attribute(matter_node, 1, 129, 5, 0)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.state == "open"
# add support for setting position by updating the featuremap
set_node_attribute(valve_node, 1, 129, 65532, 2)
set_node_attribute(matter_node, 1, 129, 65532, 2)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
assert state.attributes["current_position"] == 0
# update current position
set_node_attribute(valve_node, 1, 129, 6, 50)
set_node_attribute(matter_node, 1, 129, 6, 50)
await trigger_subscription_callback(hass, matter_client)
state = hass.states.get(entity_id)
assert state
@ -124,7 +113,7 @@ async def test_valve(
assert matter_client.send_device_command.call_count == 1
assert matter_client.send_device_command.call_args == call(
node_id=valve_node.node_id,
node_id=matter_node.node_id,
endpoint_id=1,
command=clusters.ValveConfigurationAndControl.Commands.Open(targetLevel=100),
)