Files
core/tests/components/wmspro/test_cover.py
Marc Hörsken c3abf5a190 Add support for WMS roller shutters and blinds (#132645)
* Add support for WMS roller shutters and blinds

* Add test variants for WMS device types and their diagnostics

* Add test variants for cover movement of WMS device types

* Move device entry tests to test_init and avoid snapshot list

Suggested-by: joostlek
2025-04-30 20:51:10 +02:00

289 lines
9.1 KiB
Python

"""Test the wmspro cover support."""
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.wmspro.const import DOMAIN
from homeassistant.components.wmspro.cover import SCAN_INTERVAL
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_CLOSE_COVER,
SERVICE_OPEN_COVER,
SERVICE_SET_COVER_POSITION,
SERVICE_STOP_COVER,
STATE_CLOSED,
STATE_OPEN,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import setup_config_entry
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_cover_device(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hub_ping: AsyncMock,
mock_hub_configuration_prod_awning_dimmer: AsyncMock,
mock_hub_status_prod_awning: AsyncMock,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test that a cover device is created correctly."""
assert await setup_config_entry(hass, mock_config_entry)
assert len(mock_hub_ping.mock_calls) == 1
assert len(mock_hub_configuration_prod_awning_dimmer.mock_calls) == 1
assert len(mock_hub_status_prod_awning.mock_calls) == 2
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "58717")})
assert device_entry is not None
assert device_entry == snapshot
async def test_cover_update(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hub_ping: AsyncMock,
mock_hub_configuration_prod_awning_dimmer: AsyncMock,
mock_hub_status_prod_awning: AsyncMock,
freezer: FrozenDateTimeFactory,
snapshot: SnapshotAssertion,
) -> None:
"""Test that a cover entity is created and updated correctly."""
assert await setup_config_entry(hass, mock_config_entry)
assert len(mock_hub_ping.mock_calls) == 1
assert len(mock_hub_configuration_prod_awning_dimmer.mock_calls) == 1
assert len(mock_hub_status_prod_awning.mock_calls) == 2
entity = hass.states.get("cover.markise")
assert entity is not None
assert entity == snapshot
# Move time to next update
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert len(mock_hub_status_prod_awning.mock_calls) >= 3
@pytest.mark.parametrize(
("mock_hub_configuration", "mock_hub_status", "entity_name"),
[
(
"mock_hub_configuration_prod_awning_dimmer",
"mock_hub_status_prod_awning",
"cover.markise",
),
(
"mock_hub_configuration_prod_roller_shutter",
"mock_hub_status_prod_roller_shutter",
"cover.wohnebene_alle",
),
],
)
async def test_cover_open_and_close(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hub_ping: AsyncMock,
mock_hub_configuration: AsyncMock,
mock_hub_status: AsyncMock,
mock_action_call: AsyncMock,
request: pytest.FixtureRequest,
entity_name: str,
) -> None:
"""Test that a cover entity is opened and closed correctly."""
mock_hub_configuration = request.getfixturevalue(mock_hub_configuration)
mock_hub_status = request.getfixturevalue(mock_hub_status)
assert await setup_config_entry(hass, mock_config_entry)
assert len(mock_hub_ping.mock_calls) == 1
assert len(mock_hub_configuration.mock_calls) == 1
assert len(mock_hub_status.mock_calls) >= 1
entity = hass.states.get(entity_name)
assert entity is not None
assert entity.state == STATE_CLOSED
assert entity.attributes["current_position"] == 0
with patch(
"wmspro.destination.Destination.refresh",
return_value=True,
):
before = len(mock_hub_status.mock_calls)
await hass.services.async_call(
Platform.COVER,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: entity.entity_id},
blocking=True,
)
entity = hass.states.get(entity_name)
assert entity is not None
assert entity.state == STATE_OPEN
assert entity.attributes["current_position"] == 100
assert len(mock_hub_status.mock_calls) == before
with patch(
"wmspro.destination.Destination.refresh",
return_value=True,
):
before = len(mock_hub_status.mock_calls)
await hass.services.async_call(
Platform.COVER,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: entity.entity_id},
blocking=True,
)
entity = hass.states.get(entity_name)
assert entity is not None
assert entity.state == STATE_CLOSED
assert entity.attributes["current_position"] == 0
assert len(mock_hub_status.mock_calls) == before
@pytest.mark.parametrize(
("mock_hub_configuration", "mock_hub_status", "entity_name"),
[
(
"mock_hub_configuration_prod_awning_dimmer",
"mock_hub_status_prod_awning",
"cover.markise",
),
(
"mock_hub_configuration_prod_roller_shutter",
"mock_hub_status_prod_roller_shutter",
"cover.wohnebene_alle",
),
],
)
async def test_cover_open_to_pos(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hub_ping: AsyncMock,
mock_hub_configuration: AsyncMock,
mock_hub_status: AsyncMock,
mock_action_call: AsyncMock,
request: pytest.FixtureRequest,
entity_name: str,
) -> None:
"""Test that a cover entity is opened to correct position."""
mock_hub_configuration = request.getfixturevalue(mock_hub_configuration)
mock_hub_status = request.getfixturevalue(mock_hub_status)
assert await setup_config_entry(hass, mock_config_entry)
assert len(mock_hub_ping.mock_calls) == 1
assert len(mock_hub_configuration.mock_calls) == 1
assert len(mock_hub_status.mock_calls) >= 1
entity = hass.states.get(entity_name)
assert entity is not None
assert entity.state == STATE_CLOSED
assert entity.attributes["current_position"] == 0
with patch(
"wmspro.destination.Destination.refresh",
return_value=True,
):
before = len(mock_hub_status.mock_calls)
await hass.services.async_call(
Platform.COVER,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: entity.entity_id, "position": 50},
blocking=True,
)
entity = hass.states.get(entity_name)
assert entity is not None
assert entity.state == STATE_OPEN
assert entity.attributes["current_position"] == 50
assert len(mock_hub_status.mock_calls) == before
@pytest.mark.parametrize(
("mock_hub_configuration", "mock_hub_status", "entity_name"),
[
(
"mock_hub_configuration_prod_awning_dimmer",
"mock_hub_status_prod_awning",
"cover.markise",
),
(
"mock_hub_configuration_prod_roller_shutter",
"mock_hub_status_prod_roller_shutter",
"cover.wohnebene_alle",
),
],
)
async def test_cover_open_and_stop(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hub_ping: AsyncMock,
mock_hub_configuration: AsyncMock,
mock_hub_status: AsyncMock,
mock_action_call: AsyncMock,
request: pytest.FixtureRequest,
entity_name: str,
) -> None:
"""Test that a cover entity is opened and stopped correctly."""
mock_hub_configuration = request.getfixturevalue(mock_hub_configuration)
mock_hub_status = request.getfixturevalue(mock_hub_status)
assert await setup_config_entry(hass, mock_config_entry)
assert len(mock_hub_ping.mock_calls) == 1
assert len(mock_hub_configuration.mock_calls) == 1
assert len(mock_hub_status.mock_calls) >= 1
entity = hass.states.get(entity_name)
assert entity is not None
assert entity.state == STATE_CLOSED
assert entity.attributes["current_position"] == 0
with patch(
"wmspro.destination.Destination.refresh",
return_value=True,
):
before = len(mock_hub_status.mock_calls)
await hass.services.async_call(
Platform.COVER,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: entity.entity_id, "position": 80},
blocking=True,
)
entity = hass.states.get(entity_name)
assert entity is not None
assert entity.state == STATE_OPEN
assert entity.attributes["current_position"] == 80
assert len(mock_hub_status.mock_calls) == before
with patch(
"wmspro.destination.Destination.refresh",
return_value=True,
):
before = len(mock_hub_status.mock_calls)
await hass.services.async_call(
Platform.COVER,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: entity.entity_id},
blocking=True,
)
entity = hass.states.get(entity_name)
assert entity is not None
assert entity.state == STATE_OPEN
assert entity.attributes["current_position"] == 80
assert len(mock_hub_status.mock_calls) == before