mirror of
https://github.com/home-assistant/core.git
synced 2026-08-28 02:24:46 -05:00
Add OpenRGB profile select entity (#154732)
This commit is contained in:
@@ -10,7 +10,7 @@ from homeassistant.helpers.device_registry import DeviceEntry
|
||||
from .const import DOMAIN
|
||||
from .coordinator import OpenRGBConfigEntry, OpenRGBCoordinator
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.LIGHT]
|
||||
PLATFORMS: list[Platform] = [Platform.LIGHT, Platform.SELECT]
|
||||
|
||||
|
||||
def _setup_server_device_registry(
|
||||
@@ -37,6 +37,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenRGBConfigEntry) -> b
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
# The server device must be created first as other devices are children of it
|
||||
_setup_server_device_registry(hass, entry, coordinator)
|
||||
|
||||
entry.runtime_data = coordinator
|
||||
|
||||
@@ -13,6 +13,8 @@ from openrgb.utils import (
|
||||
|
||||
DOMAIN = "openrgb"
|
||||
|
||||
UID_SEPARATOR = "||"
|
||||
|
||||
# Defaults
|
||||
DEFAULT_PORT = 6742
|
||||
DEFAULT_CLIENT_NAME = "Home Assistant"
|
||||
|
||||
@@ -14,7 +14,13 @@ from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.debounce import Debouncer
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import CONNECTION_ERRORS, DEFAULT_CLIENT_NAME, DOMAIN, SCAN_INTERVAL
|
||||
from .const import (
|
||||
CONNECTION_ERRORS,
|
||||
DEFAULT_CLIENT_NAME,
|
||||
DOMAIN,
|
||||
SCAN_INTERVAL,
|
||||
UID_SEPARATOR,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@@ -111,7 +117,7 @@ class OpenRGBCoordinator(DataUpdateCoordinator[dict[str, Device]]):
|
||||
device.metadata.location or "none",
|
||||
)
|
||||
# Double pipe is readable and is unlikely to appear in metadata
|
||||
return "||".join(parts)
|
||||
return UID_SEPARATOR.join(parts)
|
||||
|
||||
async def async_client_disconnect(self, *args) -> None:
|
||||
"""Disconnect the OpenRGB client."""
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Select platform for OpenRGB integration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.select import SelectEntity
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import CONNECTION_ERRORS, DOMAIN, UID_SEPARATOR
|
||||
from .coordinator import OpenRGBConfigEntry, OpenRGBCoordinator
|
||||
|
||||
PARALLEL_UPDATES = 0
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: OpenRGBConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the OpenRGB select platform."""
|
||||
coordinator = config_entry.runtime_data
|
||||
async_add_entities([OpenRGBProfileSelect(coordinator, config_entry)])
|
||||
|
||||
|
||||
class OpenRGBProfileSelect(CoordinatorEntity[OpenRGBCoordinator], SelectEntity):
|
||||
"""Representation of an OpenRGB profile select entity."""
|
||||
|
||||
_attr_translation_key = "profile"
|
||||
_attr_has_entity_name = True
|
||||
|
||||
_state_hash: int | None = None
|
||||
_pending_profile: str | None = None
|
||||
|
||||
def __init__(
|
||||
self, coordinator: OpenRGBCoordinator, entry: OpenRGBConfigEntry
|
||||
) -> None:
|
||||
"""Initialize the select entity."""
|
||||
super().__init__(coordinator)
|
||||
self._attr_unique_id = UID_SEPARATOR.join([entry.entry_id, "profile"])
|
||||
self._attr_device_info = {
|
||||
"identifiers": {(DOMAIN, entry.entry_id)},
|
||||
}
|
||||
self._update_attrs()
|
||||
|
||||
def _compute_state_hash(self) -> int:
|
||||
"""Compute a hash of device states."""
|
||||
return hash(
|
||||
"\n".join(str(device.data) for device in self.coordinator.client.devices)
|
||||
)
|
||||
|
||||
@callback
|
||||
def _update_attrs(self) -> None:
|
||||
"""Update the attributes based on the current profile list."""
|
||||
profiles = self.coordinator.client.profiles
|
||||
self._attr_options = [profile.name for profile in profiles]
|
||||
|
||||
# If a profile was just applied, set it as current
|
||||
if self._pending_profile is not None:
|
||||
self._attr_current_option = self._pending_profile
|
||||
self._pending_profile = None
|
||||
self._state_hash = self._compute_state_hash()
|
||||
# Only check for state changes if we have a current option to potentially clear
|
||||
elif self._attr_current_option is not None:
|
||||
current_hash = self._compute_state_hash()
|
||||
# If state changed, we can no longer assume current profile
|
||||
if current_hash != self._state_hash:
|
||||
self._attr_current_option = None
|
||||
self._state_hash = None
|
||||
|
||||
@callback
|
||||
def _handle_coordinator_update(self) -> None:
|
||||
"""Handle updated data from the coordinator."""
|
||||
self._update_attrs()
|
||||
super()._handle_coordinator_update()
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""Return if the select is available."""
|
||||
return super().available and bool(self._attr_options)
|
||||
|
||||
async def async_select_option(self, option: str) -> None:
|
||||
"""Load the selected profile."""
|
||||
async with self.coordinator.client_lock:
|
||||
try:
|
||||
await self.hass.async_add_executor_job(
|
||||
self.coordinator.client.load_profile, option
|
||||
)
|
||||
except CONNECTION_ERRORS as err:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="communication_error",
|
||||
translation_placeholders={
|
||||
"server_address": self.coordinator.server_address,
|
||||
"error": str(err),
|
||||
},
|
||||
) from err
|
||||
except ValueError as err:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="openrgb_error",
|
||||
translation_placeholders={
|
||||
"error": str(err),
|
||||
},
|
||||
) from err
|
||||
|
||||
self._pending_profile = option
|
||||
await self.coordinator.async_refresh()
|
||||
@@ -65,6 +65,11 @@
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"select": {
|
||||
"profile": {
|
||||
"name": "Profile"
|
||||
}
|
||||
}
|
||||
},
|
||||
"exceptions": {
|
||||
@@ -77,6 +82,9 @@
|
||||
"openrgb_error": {
|
||||
"message": "An OpenRGB error occurred: {error}"
|
||||
},
|
||||
"profile_not_found": {
|
||||
"message": "Profile `{profile}` not found"
|
||||
},
|
||||
"unsupported_effect": {
|
||||
"message": "Effect `{effect}` is not supported by {device_name}"
|
||||
}
|
||||
|
||||
@@ -66,6 +66,11 @@ def mock_openrgb_device() -> MagicMock:
|
||||
device = MagicMock(spec=device_obj)
|
||||
device.configure_mock(**vars(device_obj))
|
||||
|
||||
# Used by select entity to track state changes
|
||||
type(device).data = property(
|
||||
lambda self: {attr: getattr(self, attr) for attr in vars(device_obj)}
|
||||
)
|
||||
|
||||
# Methods
|
||||
device.set_color = MagicMock()
|
||||
device.set_mode = MagicMock()
|
||||
@@ -96,11 +101,13 @@ def mock_openrgb_client(mock_openrgb_device: MagicMock) -> Generator[MagicMock]:
|
||||
# Attributes
|
||||
client.protocol_version = 4
|
||||
client.devices = [mock_openrgb_device]
|
||||
client.profiles = []
|
||||
|
||||
# Methods
|
||||
client.update = MagicMock()
|
||||
client.connect = MagicMock()
|
||||
client.disconnect = MagicMock()
|
||||
client.load_profile = MagicMock()
|
||||
|
||||
# Store the class mock so tests can set side_effect
|
||||
client.client_class_mock = client_mock
|
||||
|
||||
@@ -1,32 +0,0 @@
|
||||
# serializer version: 1
|
||||
# name: test_server_device_registry
|
||||
DeviceRegistryEntrySnapshot({
|
||||
'area_id': None,
|
||||
'config_entries': <ANY>,
|
||||
'config_entries_subentries': <ANY>,
|
||||
'configuration_url': None,
|
||||
'connections': set({
|
||||
}),
|
||||
'disabled_by': None,
|
||||
'entry_type': <DeviceEntryType.SERVICE: 'service'>,
|
||||
'hw_version': None,
|
||||
'id': <ANY>,
|
||||
'identifiers': set({
|
||||
tuple(
|
||||
'openrgb',
|
||||
'01J0EXAMPLE0CONFIGENTRY00',
|
||||
),
|
||||
}),
|
||||
'labels': set({
|
||||
}),
|
||||
'manufacturer': 'OpenRGB',
|
||||
'model': 'OpenRGB SDK Server',
|
||||
'model_id': None,
|
||||
'name': 'Test Computer',
|
||||
'name_by_user': None,
|
||||
'primary_config_entry': <ANY>,
|
||||
'serial_number': None,
|
||||
'sw_version': '4 (Protocol)',
|
||||
'via_device_id': None,
|
||||
})
|
||||
# ---
|
||||
@@ -0,0 +1,54 @@
|
||||
# serializer version: 1
|
||||
# name: test_entities[select.test_computer_profile-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': set({
|
||||
}),
|
||||
'area_id': None,
|
||||
'capabilities': dict({
|
||||
'options': list([
|
||||
]),
|
||||
}),
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'select',
|
||||
'entity_category': None,
|
||||
'entity_id': 'select.test_computer_profile',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': 'Profile',
|
||||
'platform': 'openrgb',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': 'profile',
|
||||
'unique_id': '01J0EXAMPLE0CONFIGENTRY00||profile',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_entities[select.test_computer_profile-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'friendly_name': 'Test Computer Profile',
|
||||
'options': list([
|
||||
]),
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'select.test_computer_profile',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'unavailable',
|
||||
})
|
||||
# ---
|
||||
@@ -6,10 +6,9 @@ from unittest.mock import MagicMock
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
from openrgb.utils import ControllerParsingError, OpenRGBDisconnected, SDKVersionError
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.openrgb import async_remove_config_entry_device
|
||||
from homeassistant.components.openrgb.const import DOMAIN, SCAN_INTERVAL
|
||||
from homeassistant.components.openrgb.const import DOMAIN, SCAN_INTERVAL, UID_SEPARATOR
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import STATE_ON, STATE_UNAVAILABLE
|
||||
from homeassistant.core import HomeAssistant
|
||||
@@ -39,31 +38,9 @@ async def test_entry_setup_unload(
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_server_device_registry(
|
||||
hass: HomeAssistant,
|
||||
snapshot: SnapshotAssertion,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
) -> None:
|
||||
"""Test server device is created in device registry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
server_device = device_registry.async_get_device(
|
||||
identifiers={(DOMAIN, mock_config_entry.entry_id)}
|
||||
)
|
||||
|
||||
assert server_device == snapshot
|
||||
|
||||
|
||||
async def test_remove_config_entry_device_server(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test that server device cannot be removed."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
@@ -86,10 +63,10 @@ async def test_remove_config_entry_device_server(
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_remove_config_entry_device_still_connected(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test that connected devices cannot be removed."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
@@ -116,10 +93,10 @@ async def test_remove_config_entry_device_still_connected(
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_remove_config_entry_device_disconnected(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
) -> None:
|
||||
"""Test that disconnected devices can be removed."""
|
||||
@@ -135,7 +112,16 @@ async def test_remove_config_entry_device_disconnected(
|
||||
identifiers={
|
||||
(
|
||||
DOMAIN,
|
||||
f"{entry_id}||KEYBOARD||Old Vendor||Old Device||OLD123||Old Location",
|
||||
UID_SEPARATOR.join(
|
||||
[
|
||||
entry_id,
|
||||
"KEYBOARD",
|
||||
"Old Vendor",
|
||||
"Old Device",
|
||||
"OLD123",
|
||||
"Old Location",
|
||||
]
|
||||
),
|
||||
)
|
||||
},
|
||||
name="Old Disconnected Device",
|
||||
@@ -171,7 +157,16 @@ async def test_remove_config_entry_device_with_multiple_identifiers(
|
||||
("other_domain", "some_other_id"), # This should be skipped
|
||||
(
|
||||
DOMAIN,
|
||||
f"{entry_id}||DEVICE||Vendor||Name||SERIAL123||Location",
|
||||
UID_SEPARATOR.join(
|
||||
[
|
||||
entry_id,
|
||||
"DEVICE",
|
||||
"Vendor",
|
||||
"Name",
|
||||
"SERIAL123",
|
||||
"Location",
|
||||
]
|
||||
),
|
||||
), # This is a disconnected OpenRGB device
|
||||
},
|
||||
name="Multi-Domain Device",
|
||||
|
||||
@@ -25,6 +25,7 @@ from homeassistant.components.openrgb.const import (
|
||||
DOMAIN,
|
||||
OFF_COLOR,
|
||||
SCAN_INTERVAL,
|
||||
UID_SEPARATOR,
|
||||
OpenRGBMode,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
@@ -72,7 +73,16 @@ async def test_entities(
|
||||
identifiers={
|
||||
(
|
||||
DOMAIN,
|
||||
f"{mock_config_entry.entry_id}||DRAM||ENE||ENE SMBus Device||none||I2C: PIIX4, address 0x70",
|
||||
UID_SEPARATOR.join(
|
||||
[
|
||||
mock_config_entry.entry_id,
|
||||
"DRAM",
|
||||
"ENE",
|
||||
"ENE SMBus Device",
|
||||
"none",
|
||||
"I2C: PIIX4, address 0x70",
|
||||
]
|
||||
),
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
@@ -0,0 +1,418 @@
|
||||
"""Tests for the OpenRGB select platform."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
from openrgb.utils import OpenRGBDisconnected
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.openrgb.const import DOMAIN, SCAN_INTERVAL
|
||||
from homeassistant.components.select import DOMAIN as SELECT_DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_OPTION,
|
||||
SERVICE_SELECT_OPTION,
|
||||
STATE_UNAVAILABLE,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
|
||||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def select_only() -> Generator[None]:
|
||||
"""Enable only the select platform."""
|
||||
with patch(
|
||||
"homeassistant.components.openrgb.PLATFORMS",
|
||||
[Platform.SELECT],
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_profiles() -> list[SimpleNamespace]:
|
||||
"""Return a list of mock profiles."""
|
||||
return [
|
||||
SimpleNamespace(name="Gaming"),
|
||||
SimpleNamespace(name="Work"),
|
||||
SimpleNamespace(name="Rainbow"),
|
||||
]
|
||||
|
||||
|
||||
# Test basic entity setup and configuration
|
||||
@pytest.mark.usefixtures("init_integration")
|
||||
async def test_entities(
|
||||
hass: HomeAssistant,
|
||||
snapshot: SnapshotAssertion,
|
||||
entity_registry: er.EntityRegistry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test the select entities."""
|
||||
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
|
||||
|
||||
# Ensure entity is correctly assigned to the OpenRGB server device
|
||||
device_entry = device_registry.async_get_device(
|
||||
identifiers={(DOMAIN, mock_config_entry.entry_id)}
|
||||
)
|
||||
assert device_entry
|
||||
entity_entries = er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry.entry_id
|
||||
)
|
||||
assert len(entity_entries) == 1
|
||||
assert entity_entries[0].device_id == device_entry.id
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_select_with_profiles(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
mock_profiles: list[SimpleNamespace],
|
||||
) -> None:
|
||||
"""Test select entity with available profiles."""
|
||||
mock_openrgb_client.profiles = mock_profiles
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Verify select entity has profile options
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.attributes.get("options") == ["Gaming", "Work", "Rainbow"]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_select_with_no_profiles(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
) -> None:
|
||||
"""Test select entity when no profiles are available."""
|
||||
mock_openrgb_client.profiles = []
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Verify select entity is unavailable when no profiles exist
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
assert state.attributes.get("options") == []
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_select_option_success(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
mock_profiles: list[SimpleNamespace],
|
||||
) -> None:
|
||||
"""Test selecting a profile successfully."""
|
||||
mock_openrgb_client.profiles = mock_profiles
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Select a profile
|
||||
await hass.services.async_call(
|
||||
SELECT_DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
{
|
||||
ATTR_ENTITY_ID: "select.test_computer_profile",
|
||||
ATTR_OPTION: "Gaming",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
# Verify load_profile was called with the correct profile name
|
||||
mock_openrgb_client.load_profile.assert_called_once_with("Gaming")
|
||||
|
||||
# Verify the current option is set to the selected profile
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.state == "Gaming"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_select_option_not_found(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
mock_profiles: list[SimpleNamespace],
|
||||
) -> None:
|
||||
"""Test selecting a profile that doesn't exist."""
|
||||
mock_openrgb_client.profiles = mock_profiles
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Try to select a non-existent profile
|
||||
# The select platform will validate that the option is in the options list
|
||||
with pytest.raises(
|
||||
ServiceValidationError,
|
||||
match="Option NonExistent is not valid for entity select.test_computer_profile",
|
||||
):
|
||||
await hass.services.async_call(
|
||||
SELECT_DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
{
|
||||
ATTR_ENTITY_ID: "select.test_computer_profile",
|
||||
ATTR_OPTION: "NonExistent",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_select_option_connection_error(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
mock_profiles: list[SimpleNamespace],
|
||||
) -> None:
|
||||
"""Test selecting a profile with connection error."""
|
||||
mock_openrgb_client.profiles = mock_profiles
|
||||
mock_openrgb_client.load_profile.side_effect = OpenRGBDisconnected(
|
||||
"Connection lost"
|
||||
)
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Try to select a profile - should raise HomeAssistantError
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
SELECT_DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
{
|
||||
ATTR_ENTITY_ID: "select.test_computer_profile",
|
||||
ATTR_OPTION: "Gaming",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_select_option_value_error(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
mock_profiles: list[SimpleNamespace],
|
||||
) -> None:
|
||||
"""Test selecting a profile with ValueError."""
|
||||
mock_openrgb_client.profiles = mock_profiles
|
||||
mock_openrgb_client.load_profile.side_effect = ValueError("Invalid profile data")
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Try to select a profile - should raise HomeAssistantError
|
||||
with pytest.raises(HomeAssistantError):
|
||||
await hass.services.async_call(
|
||||
SELECT_DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
{
|
||||
ATTR_ENTITY_ID: "select.test_computer_profile",
|
||||
ATTR_OPTION: "Gaming",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_profiles_update_on_refresh(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
mock_profiles: list[SimpleNamespace],
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test that profile list updates when profiles change."""
|
||||
# Start with initial profiles
|
||||
mock_openrgb_client.profiles = mock_profiles[:2] # Only Gaming and Work
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Verify initial profile options
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.attributes.get("options") == ["Gaming", "Work"]
|
||||
|
||||
# Add a new profile
|
||||
mock_openrgb_client.profiles = mock_profiles # All three profiles
|
||||
|
||||
freezer.tick(SCAN_INTERVAL)
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Verify updated profile options
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.attributes.get("options") == ["Gaming", "Work", "Rainbow"]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_select_becomes_unavailable_when_profiles_removed(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
mock_profiles: list[SimpleNamespace],
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test select becomes unavailable when all profiles are removed."""
|
||||
# Start with profiles
|
||||
mock_openrgb_client.profiles = mock_profiles
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Verify select entity is available with profiles
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.state != STATE_UNAVAILABLE
|
||||
assert state.attributes.get("options") == ["Gaming", "Work", "Rainbow"]
|
||||
|
||||
# Remove all profiles
|
||||
mock_openrgb_client.profiles = []
|
||||
|
||||
freezer.tick(SCAN_INTERVAL)
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Verify select entity becomes unavailable
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
assert state.attributes.get("options") == []
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_current_option_cleared_when_device_state_changes(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
mock_openrgb_device: MagicMock,
|
||||
mock_profiles: list[SimpleNamespace],
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test current option is cleared when device state changes externally."""
|
||||
mock_openrgb_client.profiles = mock_profiles
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Select a profile
|
||||
await hass.services.async_call(
|
||||
SELECT_DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
{
|
||||
ATTR_ENTITY_ID: "select.test_computer_profile",
|
||||
ATTR_OPTION: "Gaming",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
# Verify the current option is set
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.state == "Gaming"
|
||||
|
||||
# Simulate external device state change by modifying active_mode
|
||||
mock_openrgb_device.active_mode = 0
|
||||
|
||||
freezer.tick(SCAN_INTERVAL)
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Verify current option is cleared because device state changed
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.state == "unknown"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_openrgb_client")
|
||||
async def test_current_option_cleared_when_colors_change(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_openrgb_client: MagicMock,
|
||||
mock_openrgb_device: MagicMock,
|
||||
mock_profiles: list[SimpleNamespace],
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test current option is cleared when device colors change externally."""
|
||||
mock_openrgb_client.profiles = mock_profiles
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
# Select a profile
|
||||
await hass.services.async_call(
|
||||
SELECT_DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
{
|
||||
ATTR_ENTITY_ID: "select.test_computer_profile",
|
||||
ATTR_OPTION: "Work",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
# Verify the current option is set
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.state == "Work"
|
||||
|
||||
# Simulate external color change
|
||||
mock_openrgb_device.colors[0].red = 0
|
||||
mock_openrgb_device.colors[0].green = 255
|
||||
|
||||
freezer.tick(SCAN_INTERVAL)
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Verify current option is cleared because device colors changed
|
||||
state = hass.states.get("select.test_computer_profile")
|
||||
assert state
|
||||
assert state.state == "unknown"
|
||||
Reference in New Issue
Block a user