mirror of
https://github.com/home-assistant/core.git
synced 2025-11-15 22:10:09 +00:00
357 lines
12 KiB
Python
357 lines
12 KiB
Python
"""Tests for the OpenRGB integration init."""
|
|
|
|
import socket
|
|
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.config_entries import ConfigEntryState
|
|
from homeassistant.const import STATE_ON, STATE_UNAVAILABLE
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers import device_registry as dr
|
|
|
|
from tests.common import MockConfigEntry, async_fire_time_changed
|
|
|
|
|
|
async def test_entry_setup_unload(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_openrgb_client: MagicMock,
|
|
) -> None:
|
|
"""Test entry setup and unload."""
|
|
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
|
|
assert mock_config_entry.runtime_data is not None
|
|
|
|
await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
|
|
|
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
|
|
assert mock_openrgb_client.disconnect.called
|
|
|
|
|
|
@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)
|
|
|
|
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
device_registry = dr.async_get(hass)
|
|
server_device = device_registry.async_get_device(
|
|
identifiers={(DOMAIN, mock_config_entry.entry_id)}
|
|
)
|
|
|
|
assert server_device is not None
|
|
|
|
# Try to remove server device - should be blocked
|
|
result = await async_remove_config_entry_device(
|
|
hass, mock_config_entry, server_device
|
|
)
|
|
|
|
assert result is False
|
|
|
|
|
|
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)
|
|
|
|
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
device_registry = dr.async_get(hass)
|
|
|
|
# Get a device that's in coordinator.data (still connected)
|
|
devices = dr.async_entries_for_config_entry(
|
|
device_registry, mock_config_entry.entry_id
|
|
)
|
|
rgb_device = next(
|
|
(d for d in devices if d.identifiers != {(DOMAIN, mock_config_entry.entry_id)}),
|
|
None,
|
|
)
|
|
|
|
if rgb_device:
|
|
# Try to remove device that's still connected - should be blocked
|
|
result = await async_remove_config_entry_device(
|
|
hass, mock_config_entry, rgb_device
|
|
)
|
|
assert result is False
|
|
|
|
|
|
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."""
|
|
mock_config_entry.add_to_hass(hass)
|
|
|
|
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
# Create a device that's not in coordinator.data (disconnected)
|
|
entry_id = mock_config_entry.entry_id
|
|
disconnected_device = device_registry.async_get_or_create(
|
|
config_entry_id=mock_config_entry.entry_id,
|
|
identifiers={
|
|
(
|
|
DOMAIN,
|
|
f"{entry_id}||KEYBOARD||Old Vendor||Old Device||OLD123||Old Location",
|
|
)
|
|
},
|
|
name="Old Disconnected Device",
|
|
via_device=(DOMAIN, entry_id),
|
|
)
|
|
|
|
# Try to remove disconnected device - should succeed
|
|
result = await async_remove_config_entry_device(
|
|
hass, mock_config_entry, disconnected_device
|
|
)
|
|
|
|
assert result is True
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_openrgb_client")
|
|
async def test_remove_config_entry_device_with_multiple_identifiers(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
device_registry: dr.DeviceRegistry,
|
|
) -> None:
|
|
"""Test device removal with multiple domain identifiers."""
|
|
mock_config_entry.add_to_hass(hass)
|
|
|
|
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
entry_id = mock_config_entry.entry_id
|
|
|
|
# Create a device with identifiers from multiple domains
|
|
device_with_multiple_identifiers = device_registry.async_get_or_create(
|
|
config_entry_id=mock_config_entry.entry_id,
|
|
identifiers={
|
|
("other_domain", "some_other_id"), # This should be skipped
|
|
(
|
|
DOMAIN,
|
|
f"{entry_id}||DEVICE||Vendor||Name||SERIAL123||Location",
|
|
), # This is a disconnected OpenRGB device
|
|
},
|
|
name="Multi-Domain Device",
|
|
via_device=(DOMAIN, entry_id),
|
|
)
|
|
|
|
# Try to remove device - should succeed because the OpenRGB identifier is disconnected
|
|
result = await async_remove_config_entry_device(
|
|
hass, mock_config_entry, device_with_multiple_identifiers
|
|
)
|
|
|
|
assert result is True
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("exception", "expected_state"),
|
|
[
|
|
(ConnectionRefusedError, ConfigEntryState.SETUP_RETRY),
|
|
(OpenRGBDisconnected, ConfigEntryState.SETUP_RETRY),
|
|
(ControllerParsingError, ConfigEntryState.SETUP_RETRY),
|
|
(TimeoutError, ConfigEntryState.SETUP_RETRY),
|
|
(socket.gaierror, ConfigEntryState.SETUP_RETRY),
|
|
(SDKVersionError, ConfigEntryState.SETUP_RETRY),
|
|
(RuntimeError("Test error"), ConfigEntryState.SETUP_RETRY),
|
|
],
|
|
)
|
|
async def test_setup_entry_exceptions(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_openrgb_client: MagicMock,
|
|
exception: Exception,
|
|
expected_state: ConfigEntryState,
|
|
) -> None:
|
|
"""Test setup entry with various exceptions."""
|
|
mock_config_entry.add_to_hass(hass)
|
|
|
|
mock_openrgb_client.client_class_mock.side_effect = exception
|
|
|
|
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert mock_config_entry.state is expected_state
|
|
|
|
|
|
async def test_reconnection_on_update_failure(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_openrgb_client: MagicMock,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""Test that coordinator reconnects when update fails."""
|
|
mock_config_entry.add_to_hass(hass)
|
|
|
|
# Set up the integration
|
|
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
# Verify initial state
|
|
state = hass.states.get("light.ene_dram")
|
|
assert state
|
|
assert state.state == STATE_ON
|
|
|
|
# Reset mock call counts after initial setup
|
|
mock_openrgb_client.update.reset_mock()
|
|
mock_openrgb_client.connect.reset_mock()
|
|
|
|
# Simulate the first update call failing, then second succeeding
|
|
mock_openrgb_client.update.side_effect = [
|
|
OpenRGBDisconnected(),
|
|
None, # Second call succeeds after reconnect
|
|
]
|
|
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
await hass.async_block_till_done()
|
|
|
|
# Verify that disconnect and connect were called (reconnection happened)
|
|
mock_openrgb_client.disconnect.assert_called_once()
|
|
mock_openrgb_client.connect.assert_called_once()
|
|
|
|
# Verify that update was called twice (once failed, once after reconnect)
|
|
assert mock_openrgb_client.update.call_count == 2
|
|
|
|
# Verify that the light is still available after successful reconnect
|
|
state = hass.states.get("light.ene_dram")
|
|
assert state
|
|
assert state.state == STATE_ON
|
|
|
|
|
|
async def test_reconnection_fails_second_attempt(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_openrgb_client: MagicMock,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""Test that coordinator fails when reconnection also fails."""
|
|
mock_config_entry.add_to_hass(hass)
|
|
|
|
# Set up the integration
|
|
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
# Verify initial state
|
|
state = hass.states.get("light.ene_dram")
|
|
assert state
|
|
assert state.state == STATE_ON
|
|
|
|
# Reset mock call counts after initial setup
|
|
mock_openrgb_client.update.reset_mock()
|
|
mock_openrgb_client.connect.reset_mock()
|
|
|
|
# Simulate the first update call failing, and reconnection also failing
|
|
mock_openrgb_client.update.side_effect = [
|
|
OpenRGBDisconnected(),
|
|
None, # Second call would succeed if reconnect worked
|
|
]
|
|
|
|
# Simulate connect raising an exception to mimic failed reconnection
|
|
mock_openrgb_client.connect.side_effect = ConnectionRefusedError()
|
|
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
await hass.async_block_till_done()
|
|
await hass.async_block_till_done()
|
|
|
|
# Verify that the light became unavailable after failed reconnection
|
|
state = hass.states.get("light.ene_dram")
|
|
assert state
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
# Verify that disconnect and connect were called (reconnection was attempted)
|
|
mock_openrgb_client.disconnect.assert_called_once()
|
|
mock_openrgb_client.connect.assert_called_once()
|
|
|
|
# Verify that update was only called in the first attempt
|
|
mock_openrgb_client.update.assert_called_once()
|
|
|
|
|
|
async def test_normal_update_without_errors(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
mock_openrgb_client: MagicMock,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""Test that normal updates work without triggering reconnection."""
|
|
mock_config_entry.add_to_hass(hass)
|
|
|
|
# Set up the integration
|
|
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
# Verify initial state
|
|
state = hass.states.get("light.ene_dram")
|
|
assert state
|
|
assert state.state == STATE_ON
|
|
|
|
# Reset mock call counts after initial setup
|
|
mock_openrgb_client.update.reset_mock()
|
|
mock_openrgb_client.connect.reset_mock()
|
|
|
|
# Simulate successful update
|
|
mock_openrgb_client.update.side_effect = None
|
|
mock_openrgb_client.update.return_value = None
|
|
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
await hass.async_block_till_done()
|
|
|
|
# Verify that disconnect and connect were NOT called (no reconnection needed)
|
|
mock_openrgb_client.disconnect.assert_not_called()
|
|
mock_openrgb_client.connect.assert_not_called()
|
|
|
|
# Verify that update was called only once
|
|
mock_openrgb_client.update.assert_called_once()
|
|
|
|
# Verify that the light is still available
|
|
state = hass.states.get("light.ene_dram")
|
|
assert state
|
|
assert state.state == STATE_ON
|