Add tests for Comelit cover platform (#141740)

* Add  tests for Comelit cover platform

* cleanup up
This commit is contained in:
Simone Chemelli 2025-03-30 12:04:00 +02:00 committed by GitHub
parent 5b5efb5aaa
commit 600aedc9a1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 216 additions and 4 deletions

View File

@ -49,10 +49,10 @@ def mock_serial_bridge() -> Generator[AsyncMock]:
),
):
bridge = mock_comelit_serial_bridge.return_value
bridge.get_all_devices.return_value = BRIDGE_DEVICE_QUERY
bridge.get_all_devices.return_value = deepcopy(BRIDGE_DEVICE_QUERY)
bridge.host = BRIDGE_HOST
bridge.port = BRIDGE_PORT
bridge.pin = BRIDGE_PIN
bridge.device_pin = BRIDGE_PIN
yield bridge
@ -67,6 +67,7 @@ def mock_serial_bridge_config_entry() -> Generator[MockConfigEntry]:
CONF_PIN: BRIDGE_PIN,
CONF_TYPE: BRIDGE,
},
entry_id="serial_bridge_config_entry_id",
)

View File

@ -35,7 +35,7 @@ BRIDGE_DEVICE_QUERY = {
index=0,
name="Cover0",
status=0,
human_status="closed",
human_status="stopped",
type="cover",
val=0,
protected=0,

View File

@ -0,0 +1,50 @@
# serializer version: 1
# name: test_all_entities[cover.cover0-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'cover',
'entity_category': None,
'entity_id': 'cover.cover0',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <CoverDeviceClass.SHUTTER: 'shutter'>,
'original_icon': None,
'original_name': None,
'platform': 'comelit',
'previous_unique_id': None,
'supported_features': <CoverEntityFeature: 11>,
'translation_key': None,
'unique_id': 'serial_bridge_config_entry_id-0',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[cover.cover0-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'shutter',
'friendly_name': 'Cover0',
'supported_features': <CoverEntityFeature: 11>,
}),
'context': <ANY>,
'entity_id': 'cover.cover0',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@ -11,7 +11,7 @@
'shutter': list([
dict({
'0': dict({
'human_status': 'closed',
'human_status': 'stopped',
'name': 'Cover0',
'power': 0.0,
'power_unit': 'W',

View File

@ -0,0 +1,161 @@
"""Tests for Comelit SimpleHome cover platform."""
from unittest.mock import AsyncMock, patch
from aiocomelit.api import ComelitSerialBridgeObject
from aiocomelit.const import COVER, WATT
from freezegun.api import FrozenDateTimeFactory
from syrupy import SnapshotAssertion
from homeassistant.components.comelit.const import SCAN_INTERVAL
from homeassistant.components.cover import (
DOMAIN as COVER_DOMAIN,
SERVICE_CLOSE_COVER,
SERVICE_OPEN_COVER,
SERVICE_STOP_COVER,
STATE_CLOSED,
STATE_CLOSING,
STATE_OPENING,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
ENTITY_ID = "cover.cover0"
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_serial_bridge: AsyncMock,
mock_serial_bridge_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test all entities."""
with patch("homeassistant.components.comelit.BRIDGE_PLATFORMS", [Platform.COVER]):
await setup_integration(hass, mock_serial_bridge_config_entry)
await snapshot_platform(
hass,
entity_registry,
snapshot(),
mock_serial_bridge_config_entry.entry_id,
)
async def test_cover_open(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_serial_bridge: AsyncMock,
mock_serial_bridge_config_entry: MockConfigEntry,
) -> None:
"""Test cover open service."""
mock_serial_bridge.reset_mock()
await setup_integration(hass, mock_serial_bridge_config_entry)
assert (state := hass.states.get(ENTITY_ID))
assert state.state == STATE_UNKNOWN
# Open cover
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
mock_serial_bridge.set_device_status.assert_called()
assert (state := hass.states.get(ENTITY_ID))
assert state.state == STATE_OPENING
# Finish opening, update status
mock_serial_bridge.get_all_devices.return_value[COVER] = {
0: ComelitSerialBridgeObject(
index=0,
name="Cover0",
status=0,
human_status="stopped",
type="cover",
val=0,
protected=0,
zone="Open space",
power=0.0,
power_unit=WATT,
),
}
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (state := hass.states.get(ENTITY_ID))
assert state.state == STATE_UNKNOWN
async def test_cover_close(
hass: HomeAssistant,
mock_serial_bridge: AsyncMock,
mock_serial_bridge_config_entry: MockConfigEntry,
) -> None:
"""Test cover close and stop service."""
mock_serial_bridge.reset_mock()
await setup_integration(hass, mock_serial_bridge_config_entry)
assert (state := hass.states.get(ENTITY_ID))
assert state.state == STATE_UNKNOWN
# Close cover
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
mock_serial_bridge.set_device_status.assert_called()
assert (state := hass.states.get(ENTITY_ID))
assert state.state == STATE_CLOSING
# Stop cover
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
mock_serial_bridge.set_device_status.assert_called()
assert (state := hass.states.get(ENTITY_ID))
assert state.state == STATE_CLOSED
async def test_cover_stop_if_stopped(
hass: HomeAssistant,
mock_serial_bridge: AsyncMock,
mock_serial_bridge_config_entry: MockConfigEntry,
) -> None:
"""Test cover stop service when already stopped."""
mock_serial_bridge.reset_mock()
await setup_integration(hass, mock_serial_bridge_config_entry)
assert (state := hass.states.get(ENTITY_ID))
assert state.state == STATE_UNKNOWN
# Stop cover while not opening/closing
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
mock_serial_bridge.set_device_status.assert_not_called()
assert (state := hass.states.get(ENTITY_ID))
assert state.state == STATE_UNKNOWN