Add switch platform to pyload integration (#120352)
Mr. Bubbles
GitHub
parent
caa57c56f6
commit
d76a82e340
|
|
@ -22,7 +22,7 @@ from homeassistant.helpers.aiohttp_client import async_create_clientsession
|
|||
|
||||
from .coordinator import PyLoadCoordinator
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.SENSOR]
|
||||
PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.SENSOR, Platform.SWITCH]
|
||||
|
||||
type PyLoadConfigEntry = ConfigEntry[PyLoadCoordinator]
|
||||
|
||||
|
|
|
|||
|
|
@ -27,6 +27,22 @@
|
|||
"total": {
|
||||
"default": "mdi:cloud-alert"
|
||||
}
|
||||
},
|
||||
"switch": {
|
||||
"download": {
|
||||
"default": "mdi:play",
|
||||
"state": {
|
||||
"on": "mdi:play",
|
||||
"off": "mdi:pause"
|
||||
}
|
||||
},
|
||||
"reconnect": {
|
||||
"default": "mdi:restart",
|
||||
"state": {
|
||||
"on": "mdi:restart",
|
||||
"off": "mdi:restart-off"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -79,6 +79,14 @@
|
|||
"free_space": {
|
||||
"name": "Free space"
|
||||
}
|
||||
},
|
||||
"switch": {
|
||||
"download": {
|
||||
"name": "Pause/Resume queue"
|
||||
},
|
||||
"reconnect": {
|
||||
"name": "Auto-Reconnect"
|
||||
}
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,122 @@
|
|||
"""Support for monitoring pyLoad."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
from dataclasses import dataclass
|
||||
from enum import StrEnum
|
||||
from typing import Any
|
||||
|
||||
from pyloadapi.api import PyLoadAPI
|
||||
|
||||
from homeassistant.components.switch import (
|
||||
SwitchDeviceClass,
|
||||
SwitchEntity,
|
||||
SwitchEntityDescription,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import PyLoadConfigEntry
|
||||
from .const import DOMAIN, MANUFACTURER, SERVICE_NAME
|
||||
from .coordinator import PyLoadCoordinator
|
||||
|
||||
|
||||
class PyLoadSwitchEntity(StrEnum):
|
||||
"""PyLoad Switch Entities."""
|
||||
|
||||
PAUSE_RESUME_QUEUE = "download"
|
||||
RECONNECT = "reconnect"
|
||||
|
||||
|
||||
@dataclass(kw_only=True, frozen=True)
|
||||
class PyLoadSwitchEntityDescription(SwitchEntityDescription):
|
||||
"""Describes pyLoad switch entity."""
|
||||
|
||||
turn_on_fn: Callable[[PyLoadAPI], Awaitable[Any]]
|
||||
turn_off_fn: Callable[[PyLoadAPI], Awaitable[Any]]
|
||||
toggle_fn: Callable[[PyLoadAPI], Awaitable[Any]]
|
||||
|
||||
|
||||
SENSOR_DESCRIPTIONS: tuple[PyLoadSwitchEntityDescription, ...] = (
|
||||
PyLoadSwitchEntityDescription(
|
||||
key=PyLoadSwitchEntity.PAUSE_RESUME_QUEUE,
|
||||
translation_key=PyLoadSwitchEntity.PAUSE_RESUME_QUEUE,
|
||||
device_class=SwitchDeviceClass.SWITCH,
|
||||
turn_on_fn=lambda api: api.unpause(),
|
||||
turn_off_fn=lambda api: api.pause(),
|
||||
toggle_fn=lambda api: api.toggle_pause(),
|
||||
),
|
||||
PyLoadSwitchEntityDescription(
|
||||
key=PyLoadSwitchEntity.RECONNECT,
|
||||
translation_key=PyLoadSwitchEntity.RECONNECT,
|
||||
device_class=SwitchDeviceClass.SWITCH,
|
||||
turn_on_fn=lambda api: api.toggle_reconnect(),
|
||||
turn_off_fn=lambda api: api.toggle_reconnect(),
|
||||
toggle_fn=lambda api: api.toggle_reconnect(),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: PyLoadConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the pyLoad sensors."""
|
||||
|
||||
coordinator = entry.runtime_data
|
||||
|
||||
async_add_entities(
|
||||
PyLoadBinarySensor(coordinator, description)
|
||||
for description in SENSOR_DESCRIPTIONS
|
||||
)
|
||||
|
||||
|
||||
class PyLoadBinarySensor(CoordinatorEntity[PyLoadCoordinator], SwitchEntity):
|
||||
"""Representation of a pyLoad sensor."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
entity_description: PyLoadSwitchEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: PyLoadCoordinator,
|
||||
entity_description: PyLoadSwitchEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self._attr_unique_id = (
|
||||
f"{coordinator.config_entry.entry_id}_{entity_description.key}"
|
||||
)
|
||||
self.entity_description = entity_description
|
||||
self._attr_device_info = DeviceInfo(
|
||||
entry_type=DeviceEntryType.SERVICE,
|
||||
manufacturer=MANUFACTURER,
|
||||
model=SERVICE_NAME,
|
||||
configuration_url=coordinator.pyload.api_url,
|
||||
identifiers={(DOMAIN, coordinator.config_entry.entry_id)},
|
||||
sw_version=coordinator.version,
|
||||
)
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool | None:
|
||||
"""Return the state of the device."""
|
||||
return getattr(self.coordinator.data, self.entity_description.key)
|
||||
|
||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn the entity on."""
|
||||
await self.entity_description.turn_on_fn(self.coordinator.pyload)
|
||||
await self.coordinator.async_refresh()
|
||||
|
||||
async def async_turn_off(self, **kwargs: Any) -> None:
|
||||
"""Turn the entity on."""
|
||||
await self.entity_description.turn_off_fn(self.coordinator.pyload)
|
||||
await self.coordinator.async_refresh()
|
||||
|
||||
async def async_toggle(self, **kwargs: Any) -> None:
|
||||
"""Toggle the entity."""
|
||||
await self.entity_description.toggle_fn(self.coordinator.pyload)
|
||||
await self.coordinator.async_refresh()
|
||||
|
|
@ -0,0 +1,142 @@
|
|||
# serializer version: 1
|
||||
# name: test_state[switch.pyload_auto_reconnect-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': set({
|
||||
}),
|
||||
'area_id': None,
|
||||
'capabilities': None,
|
||||
'config_entry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'switch',
|
||||
'entity_category': None,
|
||||
'entity_id': 'switch.pyload_auto_reconnect',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': <SwitchDeviceClass.SWITCH: 'switch'>,
|
||||
'original_icon': None,
|
||||
'original_name': 'Auto-Reconnect',
|
||||
'platform': 'pyload',
|
||||
'previous_unique_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': <PyLoadSwitchEntity.RECONNECT: 'reconnect'>,
|
||||
'unique_id': 'XXXXXXXXXXXXXX_reconnect',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_state[switch.pyload_auto_reconnect-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'device_class': 'switch',
|
||||
'friendly_name': 'pyLoad Auto-Reconnect',
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'switch.pyload_auto_reconnect',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'off',
|
||||
})
|
||||
# ---
|
||||
# name: test_state[switch.pyload_pause_resume_queue-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': set({
|
||||
}),
|
||||
'area_id': None,
|
||||
'capabilities': None,
|
||||
'config_entry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'switch',
|
||||
'entity_category': None,
|
||||
'entity_id': 'switch.pyload_pause_resume_queue',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': <SwitchDeviceClass.SWITCH: 'switch'>,
|
||||
'original_icon': None,
|
||||
'original_name': 'Pause/Resume queue',
|
||||
'platform': 'pyload',
|
||||
'previous_unique_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': <PyLoadSwitchEntity.PAUSE_RESUME_QUEUE: 'download'>,
|
||||
'unique_id': 'XXXXXXXXXXXXXX_download',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_state[switch.pyload_pause_resume_queue-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'device_class': 'switch',
|
||||
'friendly_name': 'pyLoad Pause/Resume queue',
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'switch.pyload_pause_resume_queue',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'on',
|
||||
})
|
||||
# ---
|
||||
# name: test_state[switch.pyload_reconnect-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': set({
|
||||
}),
|
||||
'area_id': None,
|
||||
'capabilities': None,
|
||||
'config_entry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'switch',
|
||||
'entity_category': None,
|
||||
'entity_id': 'switch.pyload_reconnect',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': <SwitchDeviceClass.SWITCH: 'switch'>,
|
||||
'original_icon': None,
|
||||
'original_name': 'Reconnect',
|
||||
'platform': 'pyload',
|
||||
'previous_unique_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': <PyLoadSwitchEntity.RECONNECT: 'reconnect'>,
|
||||
'unique_id': 'XXXXXXXXXXXXXX_reconnect',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_state[switch.pyload_reconnect-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'device_class': 'switch',
|
||||
'friendly_name': 'pyLoad Reconnect',
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'switch.pyload_reconnect',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'off',
|
||||
})
|
||||
# ---
|
||||
|
|
@ -0,0 +1,105 @@
|
|||
"""Tests for the pyLoad Switches."""
|
||||
|
||||
from collections.abc import AsyncGenerator
|
||||
from unittest.mock import AsyncMock, call, patch
|
||||
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.pyload.switch import PyLoadSwitchEntity
|
||||
from homeassistant.components.switch import (
|
||||
DOMAIN as SWITCH_DOMAIN,
|
||||
SERVICE_TOGGLE,
|
||||
SERVICE_TURN_OFF,
|
||||
SERVICE_TURN_ON,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import ATTR_ENTITY_ID, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from tests.common import MockConfigEntry, snapshot_platform
|
||||
|
||||
# Maps entity to the mock calls to assert
|
||||
API_CALL = {
|
||||
PyLoadSwitchEntity.PAUSE_RESUME_QUEUE: {
|
||||
SERVICE_TURN_ON: call.unpause,
|
||||
SERVICE_TURN_OFF: call.pause,
|
||||
SERVICE_TOGGLE: call.toggle_pause,
|
||||
},
|
||||
PyLoadSwitchEntity.RECONNECT: {
|
||||
SERVICE_TURN_ON: call.toggle_reconnect,
|
||||
SERVICE_TURN_OFF: call.toggle_reconnect,
|
||||
SERVICE_TOGGLE: call.toggle_reconnect,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def switch_only() -> AsyncGenerator[None, None]:
|
||||
"""Enable only the switch platform."""
|
||||
with patch(
|
||||
"homeassistant.components.pyload.PLATFORMS",
|
||||
[Platform.SWITCH],
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
async def test_state(
|
||||
hass: HomeAssistant,
|
||||
config_entry: MockConfigEntry,
|
||||
snapshot: SnapshotAssertion,
|
||||
entity_registry: er.EntityRegistry,
|
||||
mock_pyloadapi: AsyncMock,
|
||||
) -> None:
|
||||
"""Test switch state."""
|
||||
|
||||
config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("service_call"),
|
||||
[
|
||||
SERVICE_TURN_ON,
|
||||
SERVICE_TURN_OFF,
|
||||
SERVICE_TOGGLE,
|
||||
],
|
||||
)
|
||||
async def test_turn_on_off(
|
||||
hass: HomeAssistant,
|
||||
config_entry: MockConfigEntry,
|
||||
mock_pyloadapi: AsyncMock,
|
||||
service_call: str,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test switch turn on/off, toggle method."""
|
||||
|
||||
config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
entity_entries = er.async_entries_for_config_entry(
|
||||
entity_registry, config_entry.entry_id
|
||||
)
|
||||
|
||||
for entity_entry in entity_entries:
|
||||
await hass.services.async_call(
|
||||
SWITCH_DOMAIN,
|
||||
service_call,
|
||||
{ATTR_ENTITY_ID: entity_entry.entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
API_CALL[entity_entry.translation_key][service_call]
|
||||
in mock_pyloadapi.method_calls
|
||||
)
|
||||
mock_pyloadapi.reset_mock()
|
||||
Loading…
Reference in New Issue