Add select platform to roku (#66133)

This commit is contained in:
Chris Talkington 2022-02-11 20:52:31 -06:00 committed by GitHub
parent 578456bbb5
commit f344ea7bbb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 484 additions and 37 deletions

View File

@ -24,6 +24,7 @@ PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.MEDIA_PLAYER,
Platform.REMOTE,
Platform.SELECT,
Platform.SENSOR,
]
_LOGGER = logging.getLogger(__name__)

View File

@ -21,6 +21,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.network import is_internal_request
from .coordinator import RokuDataUpdateCoordinator
from .helpers import format_channel_name
CONTENT_TYPE_MEDIA_CLASS = {
MEDIA_TYPE_APP: MEDIA_CLASS_APP,
@ -191,11 +192,11 @@ def build_item_response(
title = "TV Channels"
media = [
{
"channel_number": item.number,
"title": item.name,
"channel_number": channel.number,
"title": format_channel_name(channel.number, channel.name),
"type": MEDIA_TYPE_CHANNEL,
}
for item in coordinator.data.channels
for channel in coordinator.data.channels
]
children_media_class = MEDIA_CLASS_CHANNEL

View File

@ -0,0 +1,10 @@
"""Helpers for Roku."""
from __future__ import annotations
def format_channel_name(channel_number: str, channel_name: str | None = None) -> str:
"""Format a Roku Channel name."""
if channel_name is not None and channel_name != "":
return f"{channel_name} ({channel_number})"
return channel_number

View File

@ -2,7 +2,7 @@
"domain": "roku",
"name": "Roku",
"documentation": "https://www.home-assistant.io/integrations/roku",
"requirements": ["rokuecp==0.13.1"],
"requirements": ["rokuecp==0.13.2"],
"homekit": {
"models": ["3810X", "4660X", "7820X", "C105X", "C135X"]
},

View File

@ -58,6 +58,7 @@ from .const import (
)
from .coordinator import RokuDataUpdateCoordinator
from .entity import RokuEntity
from .helpers import format_channel_name
_LOGGER = logging.getLogger(__name__)
@ -212,10 +213,9 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
if self.app_id != "tvinput.dtv" or self.coordinator.data.channel is None:
return None
if self.coordinator.data.channel.name is not None:
return f"{self.coordinator.data.channel.name} ({self.coordinator.data.channel.number})"
channel = self.coordinator.data.channel
return self.coordinator.data.channel.number
return format_channel_name(channel.number, channel.name)
@property
def media_title(self) -> str | None:

View File

@ -0,0 +1,174 @@
"""Support for Roku selects."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from rokuecp import Roku
from rokuecp.models import Device as RokuDevice
from homeassistant.components.select import SelectEntity, SelectEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import roku_exception_handler
from .const import DOMAIN
from .coordinator import RokuDataUpdateCoordinator
from .entity import RokuEntity
from .helpers import format_channel_name
@dataclass
class RokuSelectEntityDescriptionMixin:
"""Mixin for required keys."""
options_fn: Callable[[RokuDevice], list[str]]
value_fn: Callable[[RokuDevice], str | None]
set_fn: Callable[[RokuDevice, Roku, str], Awaitable[None]]
def _get_application_name(device: RokuDevice) -> str | None:
if device.app is None or device.app.name is None:
return None
if device.app.name == "Roku":
return "Home"
return device.app.name
def _get_applications(device: RokuDevice) -> list[str]:
return ["Home"] + sorted(app.name for app in device.apps if app.name is not None)
def _get_channel_name(device: RokuDevice) -> str | None:
if device.channel is None:
return None
return format_channel_name(device.channel.number, device.channel.name)
def _get_channels(device: RokuDevice) -> list[str]:
return sorted(
format_channel_name(channel.number, channel.name) for channel in device.channels
)
async def _launch_application(device: RokuDevice, roku: Roku, value: str) -> None:
if value == "Home":
await roku.remote("home")
appl = next(
(app for app in device.apps if value == app.name),
None,
)
if appl is not None and appl.app_id is not None:
await roku.launch(appl.app_id)
async def _tune_channel(device: RokuDevice, roku: Roku, value: str) -> None:
_channel = next(
(
channel
for channel in device.channels
if (
channel.name is not None
and value == format_channel_name(channel.number, channel.name)
)
or value == channel.number
),
None,
)
if _channel is not None:
await roku.tune(_channel.number)
@dataclass
class RokuSelectEntityDescription(
SelectEntityDescription, RokuSelectEntityDescriptionMixin
):
"""Describes Roku select entity."""
ENTITIES: tuple[RokuSelectEntityDescription, ...] = (
RokuSelectEntityDescription(
key="application",
name="Application",
icon="mdi:application",
set_fn=_launch_application,
value_fn=_get_application_name,
options_fn=_get_applications,
entity_registry_enabled_default=False,
),
)
CHANNEL_ENTITY = RokuSelectEntityDescription(
key="channel",
name="Channel",
icon="mdi:television",
set_fn=_tune_channel,
value_fn=_get_channel_name,
options_fn=_get_channels,
)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Roku select based on a config entry."""
coordinator: RokuDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
device: RokuDevice = coordinator.data
unique_id = device.info.serial_number
entities: list[RokuSelectEntity] = []
for description in ENTITIES:
entities.append(
RokuSelectEntity(
device_id=unique_id,
coordinator=coordinator,
description=description,
)
)
if len(device.channels) > 0:
entities.append(
RokuSelectEntity(
device_id=unique_id,
coordinator=coordinator,
description=CHANNEL_ENTITY,
)
)
async_add_entities(entities)
class RokuSelectEntity(RokuEntity, SelectEntity):
"""Defines a Roku select entity."""
entity_description: RokuSelectEntityDescription
@property
def current_option(self) -> str | None:
"""Return the current value."""
return self.entity_description.value_fn(self.coordinator.data)
@property
def options(self) -> list[str]:
"""Return a set of selectable options."""
return self.entity_description.options_fn(self.coordinator.data)
@roku_exception_handler
async def async_select_option(self, option: str) -> None:
"""Set the option."""
await self.entity_description.set_fn(
self.coordinator.data,
self.coordinator.roku,
option,
)
await self.coordinator.async_request_refresh()

View File

@ -2117,7 +2117,7 @@ rjpl==0.3.6
rocketchat-API==0.6.1
# homeassistant.components.roku
rokuecp==0.13.1
rokuecp==0.13.2
# homeassistant.components.roomba
roombapy==1.6.5

View File

@ -1306,7 +1306,7 @@ rflink==0.0.62
ring_doorbell==0.7.2
# homeassistant.components.roku
rokuecp==0.13.1
rokuecp==0.13.2
# homeassistant.components.roomba
roombapy==1.6.5

View File

@ -38,38 +38,44 @@ def mock_setup_entry() -> Generator[None, None, None]:
@pytest.fixture
def mock_roku_config_flow(
async def mock_device(
request: pytest.FixtureRequest,
) -> Generator[None, MagicMock, None]:
"""Return a mocked Roku client."""
) -> RokuDevice:
"""Return the mocked roku device."""
fixture: str = "roku/roku3.json"
if hasattr(request, "param") and request.param:
fixture = request.param
device = RokuDevice(json.loads(load_fixture(fixture)))
return RokuDevice(json.loads(load_fixture(fixture)))
@pytest.fixture
def mock_roku_config_flow(
mock_device: RokuDevice,
) -> Generator[None, MagicMock, None]:
"""Return a mocked Roku client."""
with patch(
"homeassistant.components.roku.config_flow.Roku", autospec=True
) as roku_mock:
client = roku_mock.return_value
client.app_icon_url.side_effect = app_icon_url
client.update.return_value = device
client.update.return_value = mock_device
yield client
@pytest.fixture
def mock_roku(request: pytest.FixtureRequest) -> Generator[None, MagicMock, None]:
def mock_roku(
request: pytest.FixtureRequest, mock_device: RokuDevice
) -> Generator[None, MagicMock, None]:
"""Return a mocked Roku client."""
fixture: str = "roku/roku3.json"
if hasattr(request, "param") and request.param:
fixture = request.param
device = RokuDevice(json.loads(load_fixture(fixture)))
with patch(
"homeassistant.components.roku.coordinator.Roku", autospec=True
) as roku_mock:
client = roku_mock.return_value
client.app_icon_url.side_effect = app_icon_url
client.update.return_value = device
client.update.return_value = mock_device
yield client

View File

@ -167,6 +167,18 @@
"name": "QVC",
"type": "air-digital",
"user-hidden": "false"
},
{
"number": "14.3",
"name": "getTV",
"type": "air-digital",
"user-hidden": "false"
},
{
"number": "99.1",
"name": "",
"type": "air-digital",
"user-hidden": "false"
}
],
"media": {

View File

@ -2,6 +2,7 @@
from unittest.mock import MagicMock
import pytest
from rokuecp import Device as RokuDevice
from homeassistant.components.binary_sensor import STATE_OFF, STATE_ON
from homeassistant.components.roku.const import DOMAIN
@ -82,10 +83,11 @@ async def test_roku_binary_sensors(
assert device_entry.suggested_area is None
@pytest.mark.parametrize("mock_roku", ["roku/rokutv-7820x.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_rokutv_binary_sensors(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_device: RokuDevice,
mock_roku: MagicMock,
) -> None:
"""Test the Roku binary sensors."""

View File

@ -158,9 +158,7 @@ async def test_homekit_unknown_error(
assert result["reason"] == "unknown"
@pytest.mark.parametrize(
"mock_roku_config_flow", ["roku/rokutv-7820x.json"], indirect=True
)
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_homekit_discovery(
hass: HomeAssistant,
mock_roku_config_flow: MagicMock,

View File

@ -115,7 +115,7 @@ async def test_setup(hass: HomeAssistant, init_integration: MockConfigEntry) ->
assert device_entry.suggested_area is None
@pytest.mark.parametrize("mock_roku", ["roku/roku3-idle.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/roku3-idle.json"], indirect=True)
async def test_idle_setup(
hass: HomeAssistant,
init_integration: MockConfigEntry,
@ -127,7 +127,7 @@ async def test_idle_setup(
assert state.state == STATE_STANDBY
@pytest.mark.parametrize("mock_roku", ["roku/rokutv-7820x.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_tv_setup(
hass: HomeAssistant,
init_integration: MockConfigEntry,
@ -215,7 +215,7 @@ async def test_supported_features(
)
@pytest.mark.parametrize("mock_roku", ["roku/rokutv-7820x.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_tv_supported_features(
hass: HomeAssistant,
init_integration: MockConfigEntry,
@ -254,7 +254,7 @@ async def test_attributes(
assert state.attributes.get(ATTR_INPUT_SOURCE) == "Roku"
@pytest.mark.parametrize("mock_roku", ["roku/roku3-app.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/roku3-app.json"], indirect=True)
async def test_attributes_app(
hass: HomeAssistant,
init_integration: MockConfigEntry,
@ -271,7 +271,9 @@ async def test_attributes_app(
assert state.attributes.get(ATTR_INPUT_SOURCE) == "Netflix"
@pytest.mark.parametrize("mock_roku", ["roku/roku3-media-playing.json"], indirect=True)
@pytest.mark.parametrize(
"mock_device", ["roku/roku3-media-playing.json"], indirect=True
)
async def test_attributes_app_media_playing(
hass: HomeAssistant,
init_integration: MockConfigEntry,
@ -290,7 +292,7 @@ async def test_attributes_app_media_playing(
assert state.attributes.get(ATTR_INPUT_SOURCE) == "Pluto TV - It's Free TV"
@pytest.mark.parametrize("mock_roku", ["roku/roku3-media-paused.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/roku3-media-paused.json"], indirect=True)
async def test_attributes_app_media_paused(
hass: HomeAssistant,
init_integration: MockConfigEntry,
@ -309,7 +311,7 @@ async def test_attributes_app_media_paused(
assert state.attributes.get(ATTR_INPUT_SOURCE) == "Pluto TV - It's Free TV"
@pytest.mark.parametrize("mock_roku", ["roku/roku3-screensaver.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/roku3-screensaver.json"], indirect=True)
async def test_attributes_screensaver(
hass: HomeAssistant,
init_integration: MockConfigEntry,
@ -326,7 +328,7 @@ async def test_attributes_screensaver(
assert state.attributes.get(ATTR_INPUT_SOURCE) == "Roku"
@pytest.mark.parametrize("mock_roku", ["roku/rokutv-7820x.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_tv_attributes(
hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
@ -557,7 +559,7 @@ async def test_services_play_media_local_source(
assert "/media/local/Epic%20Sax%20Guy%2010%20Hours.mp4?authSig=" in call_args[0]
@pytest.mark.parametrize("mock_roku", ["roku/rokutv-7820x.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_tv_services(
hass: HomeAssistant,
init_integration: MockConfigEntry,
@ -836,7 +838,7 @@ async def test_media_browse_local_source(
)
@pytest.mark.parametrize("mock_roku", ["roku/rokutv-7820x.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_tv_media_browse(
hass,
init_integration,
@ -933,10 +935,10 @@ async def test_tv_media_browse(
assert msg["result"]["children_media_class"] == MEDIA_CLASS_CHANNEL
assert msg["result"]["can_expand"]
assert not msg["result"]["can_play"]
assert len(msg["result"]["children"]) == 2
assert len(msg["result"]["children"]) == 4
assert msg["result"]["children_media_class"] == MEDIA_CLASS_CHANNEL
assert msg["result"]["children"][0]["title"] == "WhatsOn"
assert msg["result"]["children"][0]["title"] == "WhatsOn (1.1)"
assert msg["result"]["children"][0]["media_content_type"] == MEDIA_TYPE_CHANNEL
assert msg["result"]["children"][0]["media_content_id"] == "1.1"
assert msg["result"]["children"][0]["can_play"]

View File

@ -0,0 +1,241 @@
"""Tests for the Roku select platform."""
from unittest.mock import MagicMock
import pytest
from rokuecp import Application, Device as RokuDevice, RokuError
from homeassistant.components.roku.const import DOMAIN
from homeassistant.components.roku.coordinator import SCAN_INTERVAL
from homeassistant.components.select import DOMAIN as SELECT_DOMAIN
from homeassistant.components.select.const import ATTR_OPTION, ATTR_OPTIONS
from homeassistant.const import ATTR_ENTITY_ID, ATTR_ICON, SERVICE_SELECT_OPTION
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_application_state(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_device: RokuDevice,
mock_roku: MagicMock,
) -> None:
"""Test the creation and values of the Roku selects."""
entity_registry = er.async_get(hass)
entity_registry.async_get_or_create(
SELECT_DOMAIN,
DOMAIN,
"1GU48T017973_application",
suggested_object_id="my_roku_3_application",
disabled_by=None,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("select.my_roku_3_application")
assert state
assert state.attributes.get(ATTR_ICON) == "mdi:application"
assert state.attributes.get(ATTR_OPTIONS) == [
"Home",
"Amazon Video on Demand",
"Free FrameChannel Service",
"MLB.TV" + "\u00AE",
"Mediafly",
"Netflix",
"Pandora",
"Pluto TV - It's Free TV",
"Roku Channel Store",
]
assert state.state == "Home"
entry = entity_registry.async_get("select.my_roku_3_application")
assert entry
assert entry.unique_id == "1GU48T017973_application"
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: "select.my_roku_3_application",
ATTR_OPTION: "Netflix",
},
blocking=True,
)
assert mock_roku.launch.call_count == 1
mock_roku.launch.assert_called_with("12")
mock_device.app = mock_device.apps[1]
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get("select.my_roku_3_application")
assert state
assert state.state == "Netflix"
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: "select.my_roku_3_application",
ATTR_OPTION: "Home",
},
blocking=True,
)
assert mock_roku.remote.call_count == 1
mock_roku.remote.assert_called_with("home")
mock_device.app = Application(
app_id=None, name="Roku", version=None, screensaver=None
)
async_fire_time_changed(hass, dt_util.utcnow() + (SCAN_INTERVAL * 2))
await hass.async_block_till_done()
state = hass.states.get("select.my_roku_3_application")
assert state
assert state.state == "Home"
async def test_application_select_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_roku: MagicMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test error handling of the Roku selects."""
entity_registry = er.async_get(hass)
entity_registry.async_get_or_create(
SELECT_DOMAIN,
DOMAIN,
"1GU48T017973_application",
suggested_object_id="my_roku_3_application",
disabled_by=None,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_roku.launch.side_effect = RokuError
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: "select.my_roku_3_application",
ATTR_OPTION: "Netflix",
},
blocking=True,
)
state = hass.states.get("select.my_roku_3_application")
assert state
assert state.state == "Home"
assert "Invalid response from API" in caplog.text
assert mock_roku.launch.call_count == 1
mock_roku.launch.assert_called_with("12")
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_channel_state(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_device: RokuDevice,
mock_roku: MagicMock,
) -> None:
"""Test the creation and values of the Roku selects."""
entity_registry = er.async_get(hass)
state = hass.states.get("select.58_onn_roku_tv_channel")
assert state
assert state.attributes.get(ATTR_ICON) == "mdi:television"
assert state.attributes.get(ATTR_OPTIONS) == [
"99.1",
"QVC (1.3)",
"WhatsOn (1.1)",
"getTV (14.3)",
]
assert state.state == "getTV (14.3)"
entry = entity_registry.async_get("select.58_onn_roku_tv_channel")
assert entry
assert entry.unique_id == "YN00H5555555_channel"
# channel name
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: "select.58_onn_roku_tv_channel",
ATTR_OPTION: "WhatsOn (1.1)",
},
blocking=True,
)
assert mock_roku.tune.call_count == 1
mock_roku.tune.assert_called_with("1.1")
mock_device.channel = mock_device.channels[0]
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get("select.58_onn_roku_tv_channel")
assert state
assert state.state == "WhatsOn (1.1)"
# channel number
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: "select.58_onn_roku_tv_channel",
ATTR_OPTION: "99.1",
},
blocking=True,
)
assert mock_roku.tune.call_count == 2
mock_roku.tune.assert_called_with("99.1")
mock_device.channel = mock_device.channels[3]
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get("select.58_onn_roku_tv_channel")
assert state
assert state.state == "99.1"
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_channel_select_error(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_roku: MagicMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test error handling of the Roku selects."""
mock_roku.tune.side_effect = RokuError
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: "select.58_onn_roku_tv_channel",
ATTR_OPTION: "99.1",
},
blocking=True,
)
state = hass.states.get("select.58_onn_roku_tv_channel")
assert state
assert state.state == "getTV (14.3)"
assert "Invalid response from API" in caplog.text
assert mock_roku.tune.call_count == 1
mock_roku.tune.assert_called_with("99.1")

View File

@ -65,7 +65,7 @@ async def test_roku_sensors(
assert device_entry.suggested_area is None
@pytest.mark.parametrize("mock_roku", ["roku/rokutv-7820x.json"], indirect=True)
@pytest.mark.parametrize("mock_device", ["roku/rokutv-7820x.json"], indirect=True)
async def test_rokutv_sensors(
hass: HomeAssistant,
init_integration: MockConfigEntry,