Swap pyvizio for vizaio in vizio integration (#176555)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Raman Gupta
2026-07-29 11:20:22 +02:00
committed by GitHub
co-authored by Claude Fable 5
parent 3c022e7be2
commit e5ca671ec9
15 changed files with 603 additions and 413 deletions
+4 -7
View File
@@ -1,13 +1,12 @@
"""The vizio component."""
from pyvizio import VizioAsync
from vizaio import Vizio
from homeassistant.components.media_player import MediaPlayerDeviceClass
from homeassistant.const import (
CONF_ACCESS_TOKEN,
CONF_DEVICE_CLASS,
CONF_HOST,
CONF_NAME,
Platform,
)
from homeassistant.core import HomeAssistant
@@ -17,7 +16,7 @@ from homeassistant.helpers.storage import Store
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.hass_dict import HassKey
from .const import DEFAULT_TIMEOUT, DEVICE_ID, DOMAIN, VIZIO_DEVICE_CLASSES
from .const import DEFAULT_TIMEOUT, DOMAIN, VIZIO_DEVICE_CLASSES
from .coordinator import (
VizioAppsDataUpdateCoordinator,
VizioConfigEntry,
@@ -45,12 +44,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: VizioConfigEntry) -> boo
device_class = entry.data[CONF_DEVICE_CLASS]
# Create device
device = VizioAsync(
DEVICE_ID,
device = Vizio(
host,
entry.data[CONF_NAME],
auth_token=token,
device_type=VIZIO_DEVICE_CLASSES[device_class],
auth_token=token,
session=async_get_clientsession(hass, False),
timeout=DEFAULT_TIMEOUT,
)
+86 -58
View File
@@ -4,8 +4,8 @@ import copy
import logging
from typing import Any, override
from pyvizio import VizioAsync, async_guess_device_type
from pyvizio.const import APP_HOME, APPS
from vizaio import AppRecord, PairChallenge, Vizio, VizioError, async_is_tv
from vizaio.apps import APP_HOME, BUNDLED_APPS
import voluptuous as vol
from homeassistant.components.media_player import MediaPlayerDeviceClass
@@ -24,7 +24,7 @@ from homeassistant.const import (
CONF_NAME,
CONF_PIN,
)
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
@@ -40,6 +40,7 @@ from .const import (
DEFAULT_VOLUME_STEP,
DEVICE_ID,
DOMAIN,
VIZIO_DEVICE_CLASSES,
)
from .coordinator import VizioConfigEntry
@@ -93,16 +94,56 @@ def _get_pairing_schema(input_dict: dict[str, Any] | None = None) -> vol.Schema:
)
def _get_device(
hass: HomeAssistant,
host: str,
device_class: str,
auth_token: str | None = None,
) -> Vizio:
"""Build a client for config flow validation calls."""
return Vizio(
host,
device_type=VIZIO_DEVICE_CLASSES[MediaPlayerDeviceClass(device_class)],
auth_token=auth_token,
session=async_get_clientsession(hass, False),
)
async def _async_get_unique_id(
hass: HomeAssistant, host: str, device_class: str
) -> str | None:
"""Return the device serial number, or None if unavailable."""
try:
return await _get_device(hass, host, device_class).get_serial_number()
except VizioError:
return None
async def _async_validate_config(
hass: HomeAssistant, host: str, auth_token: str | None, device_class: str
) -> bool:
"""Return whether the device is reachable (and the token valid, if any)."""
device = _get_device(hass, host, device_class, auth_token)
try:
if auth_token:
await device.ping_auth()
else:
await device.ping()
except VizioError:
return False
return True
class VizioOptionsConfigFlow(OptionsFlow):
"""Handle Vizio options."""
def _get_app_list(self) -> list[dict[str, Any]]:
def _get_app_list(self) -> tuple[AppRecord, ...]:
"""Return the current apps list, falling back to defaults."""
if (
apps_coordinator := self.hass.data.get(DATA_APPS)
) and apps_coordinator.data:
return apps_coordinator.data
return APPS
return BUNDLED_APPS
async def async_step_init(
self, user_input: dict[str, Any] | None = None
@@ -154,8 +195,8 @@ class VizioOptionsConfigFlow(OptionsFlow):
),
): cv.multi_select(
[
APP_HOME["name"],
*(app["name"] for app in self._get_app_list()),
APP_HOME.name,
*(app.name for app in self._get_app_list()),
]
),
}
@@ -182,8 +223,7 @@ class VizioConfigFlow(ConfigFlow, domain=DOMAIN):
"""Initialize config flow."""
self._user_schema: vol.Schema | None = None
self._must_show_form: bool | None = None
self._ch_type: str | None = None
self._pairing_token: str | None = None
self._pair_challenge: PairChallenge | None = None
self._data: dict[str, Any] | None = None
self._apps: dict[str, list] = {}
@@ -209,10 +249,8 @@ class VizioConfigFlow(ConfigFlow, domain=DOMAIN):
# Store current values in case setup fails and user needs to edit
self._user_schema = _get_config_schema(user_input)
if self.unique_id is None:
unique_id = await VizioAsync.get_unique_id(
user_input[CONF_HOST],
user_input[CONF_DEVICE_CLASS],
session=async_get_clientsession(self.hass, False),
unique_id = await _async_get_unique_id(
self.hass, user_input[CONF_HOST], user_input[CONF_DEVICE_CLASS]
)
# Check if unique ID was found, set unique ID, and abort if a flow with
@@ -238,11 +276,11 @@ class VizioConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_ACCESS_TOKEN
):
# Ensure config is valid for a device
if not await VizioAsync.validate_ha_config(
if not await _async_validate_config(
self.hass,
user_input[CONF_HOST],
user_input.get(CONF_ACCESS_TOKEN),
user_input[CONF_DEVICE_CLASS],
session=async_get_clientsession(self.hass, False),
):
errors["base"] = "cannot_connect"
@@ -270,14 +308,14 @@ class VizioConfigFlow(ConfigFlow, domain=DOMAIN):
num_chars_to_strip = len(discovery_info.type) + 1
name = discovery_info.name[:-num_chars_to_strip]
device_class = await async_guess_device_type(host)
device_class = (
MediaPlayerDeviceClass.TV
if await async_is_tv(host)
else MediaPlayerDeviceClass.SPEAKER
)
# Set unique ID early for discovery flow so we can abort if needed
unique_id = await VizioAsync.get_unique_id(
host,
device_class,
session=async_get_clientsession(self.hass, False),
)
unique_id = await _async_get_unique_id(self.hass, host, device_class)
if not unique_id:
return self.async_abort(reason="cannot_connect")
@@ -307,51 +345,41 @@ class VizioConfigFlow(ConfigFlow, domain=DOMAIN):
assert self._data
# Start pairing process if it hasn't already started
if not self._ch_type and not self._pairing_token:
dev = VizioAsync(
DEVICE_ID,
self._data[CONF_HOST],
self._data[CONF_NAME],
None,
self._data[CONF_DEVICE_CLASS],
session=async_get_clientsession(self.hass, False),
)
pair_data = await dev.start_pair()
if pair_data:
self._ch_type = pair_data.ch_type
self._pairing_token = pair_data.token
return await self.async_step_pair_tv()
return self.async_show_form(
step_id="user",
data_schema=_get_config_schema(self._data),
errors={"base": "cannot_connect"},
if not self._pair_challenge:
dev = _get_device(
self.hass, self._data[CONF_HOST], self._data[CONF_DEVICE_CLASS]
)
try:
self._pair_challenge = await dev.begin_pair(
device_id=DEVICE_ID, device_name=self._data[CONF_NAME]
)
except VizioError:
return self.async_show_form(
step_id="user",
data_schema=_get_config_schema(self._data),
errors={"base": "cannot_connect"},
)
return await self.async_step_pair_tv()
# Complete pairing process if PIN has been provided
if user_input and user_input.get(CONF_PIN):
dev = VizioAsync(
DEVICE_ID,
self._data[CONF_HOST],
self._data[CONF_NAME],
None,
self._data[CONF_DEVICE_CLASS],
session=async_get_clientsession(self.hass, False),
dev = _get_device(
self.hass, self._data[CONF_HOST], self._data[CONF_DEVICE_CLASS]
)
pair_data = await dev.pair(
self._ch_type, self._pairing_token, user_input[CONF_PIN]
)
if pair_data:
self._data[CONF_ACCESS_TOKEN] = pair_data.auth_token
try:
auth_token = await dev.finish_pair(
device_id=DEVICE_ID,
challenge=self._pair_challenge,
pin=user_input[CONF_PIN],
)
except VizioError:
# If pairing failed, it's assumed the PIN was invalid
errors[CONF_PIN] = "complete_pairing_failed"
else:
self._data[CONF_ACCESS_TOKEN] = auth_token
self._must_show_form = True
return await self.async_step_pairing_complete()
# If no data was retrieved, it's assumed that the pairing attempt was not
# successful
errors[CONF_PIN] = "complete_pairing_failed"
return self.async_show_form(
step_id="pair_tv",
data_schema=_get_pairing_schema(user_input),
+5 -9
View File
@@ -1,9 +1,6 @@
"""Constants used by vizio component."""
from pyvizio.const import (
DEVICE_CLASS_SPEAKER as VIZIO_DEVICE_CLASS_SPEAKER,
DEVICE_CLASS_TV as VIZIO_DEVICE_CLASS_TV,
)
from vizaio import DeviceType
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
@@ -55,10 +52,9 @@ VIZIO_MUTE_ON = "on"
VIZIO_VOLUME = "volume"
VIZIO_MUTE = "mute"
# Since Vizio component relies on device class, this dict will ensure that changes to
# the values of DEVICE_CLASS_SPEAKER or DEVICE_CLASS_TV
# don't require changes to pyvizio.
# Maps HA device class to the vizaio device type so changes to vizaio's
# DeviceType values never require a config entry migration.
VIZIO_DEVICE_CLASSES = {
MediaPlayerDeviceClass.SPEAKER: VIZIO_DEVICE_CLASS_SPEAKER,
MediaPlayerDeviceClass.TV: VIZIO_DEVICE_CLASS_TV,
MediaPlayerDeviceClass.SPEAKER: DeviceType.SOUNDBAR,
MediaPlayerDeviceClass.TV: DeviceType.TV,
}
+109 -58
View File
@@ -1,15 +1,24 @@
"""Coordinator for the vizio component."""
from dataclasses import dataclass
from collections.abc import Coroutine
from dataclasses import asdict, dataclass
from datetime import timedelta
import logging
from typing import TYPE_CHECKING, Any, override
from pyvizio import VizioAsync
from pyvizio.api.apps import AppConfig
from pyvizio.api.input import InputItem
from pyvizio.const import APPS, INPUT_APPS
from pyvizio.util import gen_apps_list_from_url
from vizaio import (
AppAvailability,
AppConfig,
AppRecord,
InputInfo,
SettingInfo,
Vizio,
VizioError,
fetch_app_availability,
fetch_remote_app_catalog,
is_app_input,
)
from vizaio.apps import BUNDLED_APPS, BUNDLED_AVAILABILITY
from homeassistant.components.media_player import MediaPlayerDeviceClass
from homeassistant.config_entries import ConfigEntry
@@ -29,6 +38,44 @@ _LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=30)
async def _optional[T](coro: Coroutine[Any, Any, T]) -> T | None:
"""Return the call result, or None when the device API call fails."""
try:
return await coro
except VizioError:
return None
def _records_to_storage(records: tuple[AppRecord, ...]) -> list[dict[str, Any]]:
"""Serialize AppRecords for the store."""
return [asdict(record) for record in records]
def _records_from_storage(
data: list[dict[str, Any]],
) -> tuple[AppRecord, ...] | None:
"""Deserialize stored AppRecords, or None if the data is unreadable.
Data stored by the previous pyvizio-based version has a different
shape (uppercase config keys) and is discarded; the next daily
refresh replaces it.
"""
try:
return tuple(
AppRecord(
name=item["name"],
country=tuple(item["country"]),
config=tuple(AppConfig(**config) for config in item["config"]),
id=item["id"],
description=item["description"],
icon_url=item["icon_url"],
)
for item in data
)
except KeyError, TypeError:
return None
@dataclass(frozen=True)
class VizioRuntimeData:
"""Runtime data for Vizio integration."""
@@ -43,17 +90,17 @@ class VizioDeviceData:
# Power state
is_on: bool
# Audio settings from get_all_settings("audio")
audio_settings: dict[str, Any] | None = None
# Audio settings from get_settings("audio")
audio_settings: dict[str, SettingInfo] | None = None
# Sound mode options from get_setting_options("audio", "eq")
# Sound mode options from get_setting("audio", "eq")
sound_mode_list: list[str] | None = None
# Current input from get_current_input()
current_input: str | None = None
# Available inputs from get_inputs_list()
input_list: list[InputItem] | None = None
# Available inputs from get_inputs()
input_list: list[InputInfo] | None = None
# Current app config from get_current_app_config() (TVs only)
current_app_config: AppConfig | None = None
@@ -68,7 +115,7 @@ class VizioDeviceCoordinator(DataUpdateCoordinator[VizioDeviceData]):
self,
hass: HomeAssistant,
config_entry: VizioConfigEntry,
device: VizioAsync,
device: Vizio,
) -> None:
"""Initialize the coordinator."""
super().__init__(
@@ -83,8 +130,8 @@ class VizioDeviceCoordinator(DataUpdateCoordinator[VizioDeviceData]):
@override
async def _async_setup(self) -> None:
"""Fetch device info and update device registry."""
model = await self.device.get_model_name(log_api_exception=False)
version = await self.device.get_version(log_api_exception=False)
model = await _optional(self.device.get_model_name())
version = await _optional(self.device.get_version())
if TYPE_CHECKING:
assert self.config_entry.unique_id
@@ -102,40 +149,38 @@ class VizioDeviceCoordinator(DataUpdateCoordinator[VizioDeviceData]):
@override
async def _async_update_data(self) -> VizioDeviceData:
"""Fetch all device data."""
is_on = await self.device.get_power_state(log_api_exception=False)
if is_on is None:
try:
is_on = await self.device.get_power_state()
except VizioError as err:
raise UpdateFailed(
f"Unable to connect to {self.config_entry.data[CONF_HOST]}"
)
) from err
if not is_on:
return VizioDeviceData(is_on=False)
# Device is on - fetch all data
audio_settings = await self.device.get_all_settings(
VIZIO_AUDIO_SETTINGS, log_api_exception=False
)
audio_settings = await _optional(self.device.get_settings(VIZIO_AUDIO_SETTINGS))
sound_mode_list = None
if audio_settings and VIZIO_SOUND_MODE in audio_settings:
sound_mode_list = await self.device.get_setting_options(
VIZIO_AUDIO_SETTINGS, VIZIO_SOUND_MODE, log_api_exception=False
sound_mode = await _optional(
self.device.get_setting(VIZIO_AUDIO_SETTINGS, VIZIO_SOUND_MODE)
)
if sound_mode:
sound_mode_list = list(sound_mode.options)
current_input = await self.device.get_current_input(log_api_exception=False)
input_list = await self.device.get_inputs_list(log_api_exception=False)
current_input = await _optional(self.device.get_current_input())
input_list = await _optional(self.device.get_inputs())
current_app_config = None
# Only attempt to fetch app config if the device is a TV and supports apps
if (
self.config_entry.data[CONF_DEVICE_CLASS] == MediaPlayerDeviceClass.TV
and input_list
and any(input_item.name in INPUT_APPS for input_item in input_list)
and any(is_app_input(input_item.name) for input_item in input_list)
):
current_app_config = await self.device.get_current_app_config(
log_api_exception=False
)
current_app_config = await _optional(self.device.get_current_app_config())
return VizioDeviceData(
is_on=True,
@@ -147,7 +192,7 @@ class VizioDeviceCoordinator(DataUpdateCoordinator[VizioDeviceData]):
)
class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]):
class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator[tuple[AppRecord, ...]]):
"""Define an object to hold Vizio app config data."""
def __init__(
@@ -166,38 +211,44 @@ class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]
self.fail_count = 0
self.fail_threshold = 10
self.store = store
self.availability: tuple[AppAvailability, ...] = BUNDLED_AVAILABILITY
async def async_setup(self) -> None:
"""Load initial data from storage and register shutdown."""
await self.async_register_shutdown()
self.data = await self.store.async_load() or APPS
stored = await self.store.async_load()
self.data = (_records_from_storage(stored) if stored else None) or BUNDLED_APPS
@override
async def _async_update_data(self) -> list[dict[str, Any]]:
async def _async_update_data(self) -> tuple[AppRecord, ...]:
"""Update data via library."""
if data := await gen_apps_list_from_url(
session=async_get_clientsession(self.hass)
):
# Reset the fail count and threshold when the data is successfully retrieved
self.fail_count = 0
self.fail_threshold = 10
# Store the new data if it has changed so we have it for the next restart
if data != self.data:
await self.store.async_save(data)
return data
# For every failure, increase the fail count until we reach the threshold.
# We then log a warning, increase the threshold, and reset the fail count.
# This is here to prevent silent failures but to reduce repeat logs.
if self.fail_count == self.fail_threshold:
_LOGGER.warning(
(
"Unable to retrieve the apps list from the external server for the "
"last %s days"
),
self.fail_threshold,
)
self.fail_count = 0
self.fail_threshold += 10
else:
self.fail_count += 1
return self.data
session = async_get_clientsession(self.hass)
# Availability complements the catalog for app-name resolution; it has
# its own bundled fallback and is not persisted.
self.availability = await fetch_app_availability(session)
try:
data = await fetch_remote_app_catalog(session)
except VizioError:
# For every failure, increase the fail count until we reach the threshold.
# We then log a warning, increase the threshold, and reset the fail count.
# This is here to prevent silent failures but to reduce repeat logs.
if self.fail_count == self.fail_threshold:
_LOGGER.warning(
(
"Unable to retrieve the apps list from the external server "
"for the last %s days"
),
self.fail_threshold,
)
self.fail_count = 0
self.fail_threshold += 10
else:
self.fail_count += 1
return self.data
# Reset the fail count and threshold when the data is successfully retrieved
self.fail_count = 0
self.fail_threshold = 10
# Store the new data if it has changed so we have it for the next restart
if data != self.data:
await self.store.async_save(_records_to_storage(data))
return data
+22
View File
@@ -0,0 +1,22 @@
"""Helpers for the vizio integration."""
from collections.abc import Coroutine
from typing import Any
from vizaio import VizioError
from homeassistant.exceptions import HomeAssistantError
from .const import DOMAIN
async def async_device_command[T](coro: Coroutine[Any, Any, T]) -> T:
"""Run a device command, raising HomeAssistantError on API failure."""
try:
return await coro
except VizioError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_error",
translation_placeholders={"error": str(err)},
) from err
+2 -2
View File
@@ -6,7 +6,7 @@
"documentation": "https://www.home-assistant.io/integrations/vizio",
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["pyvizio"],
"requirements": ["pyvizio==0.1.64"],
"loggers": ["vizaio"],
"requirements": ["vizaio==0.3.2"],
"zeroconf": ["_viziocast._tcp.local."]
}
+98 -50
View File
@@ -2,8 +2,14 @@
from typing import Any, override
from pyvizio.api.apps import AppConfig, find_app_name
from pyvizio.const import APP_HOME, INPUT_APPS, NO_APP_RUNNING, UNKNOWN_APP
from vizaio import AppConfig, AppRecord, RemoteKey
from vizaio.apps import (
APP_HOME,
NO_APP_RUNNING,
UNKNOWN_APP,
find_app_name,
is_app_input,
)
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
@@ -20,7 +26,11 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import DATA_APPS
from .const import (
CONF_ADDITIONAL_CONFIGS,
CONF_APP_ID,
CONF_APPS,
CONF_CONFIG,
CONF_MESSAGE,
CONF_NAME_SPACE,
CONF_VOLUME_STEP,
DEFAULT_VOLUME_STEP,
DOMAIN,
@@ -36,6 +46,7 @@ from .coordinator import (
VizioConfigEntry,
VizioDeviceCoordinator,
)
from .helpers import async_device_command
PARALLEL_UPDATES = 0
@@ -94,6 +105,15 @@ async def async_setup_entry(
async_add_entities([entity])
def _app_config_from_conf(config: dict[str, Any]) -> AppConfig:
"""Convert a stored uppercase-key app config to a vizaio AppConfig."""
return AppConfig(
app_id=str(config[CONF_APP_ID]),
name_space=int(config[CONF_NAME_SPACE]),
message=config.get(CONF_MESSAGE),
)
class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
"""Media Player implementation which performs REST requests to device."""
@@ -123,7 +143,10 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
CONF_ADDITIONAL_CONFIGS, []
)
self._device = coordinator.device
self._max_volume = float(coordinator.device.get_max_volume())
if apps_coordinator:
self._device.set_app_catalog(apps_coordinator.data)
self._device.set_app_availability(apps_coordinator.availability)
self._max_volume = float(self._device.profile.max_volume)
# Entity class attributes that will change with each update (we only include
# the ones that are initialized differently from the defaults)
@@ -180,11 +203,11 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
# Audio settings
if data.audio_settings:
self._attr_volume_level = (
float(data.audio_settings[VIZIO_VOLUME]) / self._max_volume
float(data.audio_settings[VIZIO_VOLUME].value) / self._max_volume
)
if VIZIO_MUTE in data.audio_settings:
self._attr_is_volume_muted = (
data.audio_settings[VIZIO_MUTE].lower() == VIZIO_MUTE_ON
str(data.audio_settings[VIZIO_MUTE].value).lower() == VIZIO_MUTE_ON
)
else:
self._attr_is_volume_muted = None
@@ -192,7 +215,7 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
self._attr_supported_features |= (
MediaPlayerEntityFeature.SELECT_SOUND_MODE
)
self._attr_sound_mode = data.audio_settings[VIZIO_SOUND_MODE]
self._attr_sound_mode = str(data.audio_settings[VIZIO_SOUND_MODE].value)
if not self._attr_sound_mode_list:
self._attr_sound_mode_list = data.sound_mode_list or []
else:
@@ -210,17 +233,28 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
if (
self._attr_device_class == MediaPlayerDeviceClass.TV
and self._available_inputs
and any(app in self._available_inputs for app in INPUT_APPS)
and any(is_app_input(name) for name in self._available_inputs)
):
all_apps = self._all_apps or ()
self._available_apps = self._apps_list([app["name"] for app in all_apps])
self._available_apps = self._apps_list([app.name for app in all_apps])
self._current_app_config = data.current_app_config
self._attr_app_name = find_app_name(
app_name = find_app_name(
self._current_app_config,
[APP_HOME, *all_apps, *self._additional_app_configs],
[APP_HOME, *all_apps, *self._additional_app_records()],
availability=(
self._apps_coordinator.availability
if self._apps_coordinator
else ()
),
)
if self._attr_app_name == NO_APP_RUNNING:
# find_app_name returns None on a catalog miss; the app_name state
# attribute contract expects the UNKNOWN_APP sentinel instead
if app_name == NO_APP_RUNNING:
self._attr_app_name = None
elif app_name is None:
self._attr_app_name = UNKNOWN_APP
else:
self._attr_app_name = app_name
super()._handle_coordinator_update()
@@ -230,15 +264,23 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
additional_app["name"] for additional_app in self._additional_app_configs
]
def _additional_app_records(self) -> list[AppRecord]:
"""Return AppRecords for additional apps from configuration.yaml."""
return [
AppRecord(
name=app["name"],
country=("*",),
config=(_app_config_from_conf(app[CONF_CONFIG]),),
)
for app in self._additional_app_configs
]
async def async_update_setting(
self, setting_type: str, setting_name: str, new_value: int | str
) -> None:
"""Update a setting when update_setting service is called."""
await self._device.set_setting(
setting_type,
setting_name,
new_value,
log_api_exception=False,
await async_device_command(
self._device.set_setting(setting_type, setting_name, new_value)
)
@override
@@ -262,6 +304,8 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
def apps_list_update() -> None:
"""Update list of all apps."""
self._all_apps = apps_coordinator.data
self._device.set_app_catalog(apps_coordinator.data)
self._device.set_app_availability(apps_coordinator.availability)
self.async_write_ha_state()
self.async_on_remove(apps_coordinator.async_add_listener(apps_list_update))
@@ -270,7 +314,11 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
@override
def source(self) -> str | None:
"""Return current input of the device."""
if self._attr_app_name is not None and self._current_input in INPUT_APPS:
if (
self._attr_app_name is not None
and self._current_input is not None
and is_app_input(self._current_input)
):
return self._attr_app_name
return self._current_input
@@ -286,7 +334,7 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
*(
_input
for _input in self._available_inputs
if _input not in INPUT_APPS
if not is_app_input(_input)
),
*self._available_apps,
*(
@@ -301,12 +349,12 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
@property
@override
def app_id(self):
"""Return the ID of the current app if it is unknown by pyvizio."""
"""Return the ID of the current app if it is unknown by vizaio."""
if self._current_app_config and self.source == UNKNOWN_APP:
return {
"APP_ID": self._current_app_config.APP_ID,
"NAME_SPACE": self._current_app_config.NAME_SPACE,
"MESSAGE": self._current_app_config.MESSAGE,
CONF_APP_ID: self._current_app_config.app_id,
CONF_NAME_SPACE: self._current_app_config.name_space,
CONF_MESSAGE: self._current_app_config.message,
}
return None
@@ -315,66 +363,66 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
async def async_select_sound_mode(self, sound_mode: str) -> None:
"""Select sound mode."""
if sound_mode in (self._attr_sound_mode_list or ()):
await self._device.set_setting(
VIZIO_AUDIO_SETTINGS,
VIZIO_SOUND_MODE,
sound_mode,
log_api_exception=False,
await async_device_command(
self._device.set_setting(
VIZIO_AUDIO_SETTINGS, VIZIO_SOUND_MODE, sound_mode
)
)
@override
async def async_turn_on(self) -> None:
"""Turn the device on."""
await self._device.pow_on(log_api_exception=False)
await async_device_command(self._device.power_on())
@override
async def async_turn_off(self) -> None:
"""Turn the device off."""
await self._device.pow_off(log_api_exception=False)
await async_device_command(self._device.power_off())
@override
async def async_mute_volume(self, mute: bool) -> None:
"""Mute the volume."""
if mute:
await self._device.mute_on(log_api_exception=False)
await async_device_command(self._device.mute())
self._attr_is_volume_muted = True
else:
await self._device.mute_off(log_api_exception=False)
await async_device_command(self._device.unmute())
self._attr_is_volume_muted = False
@override
async def async_media_previous_track(self) -> None:
"""Send previous channel command."""
await self._device.ch_down(log_api_exception=False)
await async_device_command(self._device.send_key(RemoteKey.CH_DOWN))
@override
async def async_media_next_track(self) -> None:
"""Send next channel command."""
await self._device.ch_up(log_api_exception=False)
await async_device_command(self._device.send_key(RemoteKey.CH_UP))
@override
async def async_select_source(self, source: str) -> None:
"""Select input source."""
if source in self._available_inputs:
await self._device.set_input(source, log_api_exception=False)
await async_device_command(self._device.set_input(source))
elif source in self._get_additional_app_names():
await self._device.launch_app_config(
**next(
app["config"]
for app in self._additional_app_configs
if app["name"] == source
),
log_api_exception=False,
await async_device_command(
self._device.launch_app_config(
_app_config_from_conf(
next(
app[CONF_CONFIG]
for app in self._additional_app_configs
if app["name"] == source
)
)
)
)
elif source in self._available_apps:
await self._device.launch_app(
source, self._all_apps, log_api_exception=False
)
await async_device_command(self._device.launch_app(source))
@override
async def async_volume_up(self) -> None:
"""Increase volume of the device."""
await self._device.vol_up(num=self._volume_step, log_api_exception=False)
await async_device_command(self._device.volume_up(steps=self._volume_step))
if self._attr_volume_level is not None:
self._attr_volume_level = min(
@@ -384,7 +432,7 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
@override
async def async_volume_down(self) -> None:
"""Decrease volume of the device."""
await self._device.vol_down(num=self._volume_step, log_api_exception=False)
await async_device_command(self._device.volume_down(steps=self._volume_step))
if self._attr_volume_level is not None:
self._attr_volume_level = max(
@@ -397,20 +445,20 @@ class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
if self._attr_volume_level is not None:
if volume > self._attr_volume_level:
num = int(self._max_volume * (volume - self._attr_volume_level))
await self._device.vol_up(num=num, log_api_exception=False)
await async_device_command(self._device.volume_up(steps=num))
self._attr_volume_level = volume
elif volume < self._attr_volume_level:
num = int(self._max_volume * (self._attr_volume_level - volume))
await self._device.vol_down(num=num, log_api_exception=False)
await async_device_command(self._device.volume_down(steps=num))
self._attr_volume_level = volume
@override
async def async_media_play(self) -> None:
"""Play whatever media is currently active."""
await self._device.play(log_api_exception=False)
await async_device_command(self._device.send_key(RemoteKey.PLAY))
@override
async def async_media_pause(self) -> None:
"""Pause whatever media is currently active."""
await self._device.pause(log_api_exception=False)
await async_device_command(self._device.send_key(RemoteKey.PAUSE))
+8 -7
View File
@@ -20,10 +20,11 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import VizioConfigEntry, VizioDeviceCoordinator
from .helpers import async_device_command
PARALLEL_UPDATES = 0
# Maps native pyvizio key names to human-friendly aliases.
# Maps native vizaio key names to human-friendly aliases.
# Keys are uppercase native names (e.g. "CC_TOGGLE"), values are lists of lowercase aliases.
REMOTE_KEY_ALIASES: dict[str, list[str]] = {
"CC_TOGGLE": ["closed_captions", "cc"],
@@ -74,8 +75,8 @@ class VizioRemote(CoordinatorEntity[VizioDeviceCoordinator], RemoteEntity):
assert unique_id is not None
self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, unique_id)})
self._device = coordinator.device
valid_keys = set(self._device.get_remote_keys_list())
# Map lowercased native keys to their original uppercase pyvizio names
valid_keys = set(self._device.available_keys)
# Map lowercased native keys to their original uppercase vizaio names
self._command_map: dict[str, str] = {key.lower(): key for key in valid_keys}
# Add aliases only for native keys this device actually supports
for alias, target in _ALIAS_LOOKUP.items():
@@ -89,7 +90,7 @@ class VizioRemote(CoordinatorEntity[VizioDeviceCoordinator], RemoteEntity):
return self.coordinator.data.is_on
def _resolve_command(self, command: str) -> str:
"""Resolve an lowercased command string to a pyvizio key name."""
"""Resolve an lowercased command string to a vizaio key name."""
if resolved := self._command_map.get(command):
return resolved
raise ServiceValidationError(
@@ -101,12 +102,12 @@ class VizioRemote(CoordinatorEntity[VizioDeviceCoordinator], RemoteEntity):
@override
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the device."""
await self._device.pow_on(log_api_exception=False)
await async_device_command(self._device.power_on())
@override
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the device."""
await self._device.pow_off(log_api_exception=False)
await async_device_command(self._device.power_off())
@override
async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None:
@@ -117,6 +118,6 @@ class VizioRemote(CoordinatorEntity[VizioDeviceCoordinator], RemoteEntity):
for i in range(num_repeats):
for cmd in resolved:
await self._device.remote(cmd, log_api_exception=False)
await async_device_command(self._device.send_key(cmd))
if i < num_repeats - 1:
await asyncio.sleep(delay)
@@ -41,6 +41,9 @@
}
},
"exceptions": {
"command_error": {
"message": "Failed to send command to the device: {error}"
},
"unknown_command": {
"message": "Unknown remote command `{command}`. Valid commands for this device are listed in the integration documentation."
}
+3 -3
View File
@@ -2831,9 +2831,6 @@ pyversasense==0.0.6
# homeassistant.components.vesync
pyvesync==3.4.2
# homeassistant.components.vizio
pyvizio==0.1.64
# homeassistant.components.velux
pyvlx==0.2.36
@@ -3329,6 +3326,9 @@ vilfo-api-client==0.5.0
# homeassistant.components.watts
visionpluspython==1.1.0
# homeassistant.components.vizio
vizaio==0.3.2
# homeassistant.components.caldav
vobject==0.9.9
+100 -80
View File
@@ -4,17 +4,16 @@ from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from pyvizio.api.apps import AppConfig
from pyvizio.const import DEVICE_CLASS_SPEAKER, MAX_VOLUME
from vizaio import AppConfig, InputInfo, SettingInfo, SettingType, VizioConnectionError
from vizaio.profiles import SOUNDBAR_PROFILE
from homeassistant.components.vizio.const import DOMAIN
from homeassistant.core import HomeAssistant
from .const import (
ACCESS_TOKEN,
APP_LIST,
CH_TYPE,
CURRENT_APP_CONFIG,
APP_RECORDS,
CURRENT_APP_CONFIG_OBJ,
CURRENT_EQ,
CURRENT_INPUT,
EQ_LIST,
@@ -23,28 +22,20 @@ from .const import (
MOCK_SPEAKER_CONFIG,
MOCK_USER_VALID_TV_CONFIG,
MODEL,
RESPONSE_TOKEN,
PAIR_CHALLENGE,
UNIQUE_ID,
VERSION,
MockCompletePairingResponse,
MockStartPairingResponse,
audio_setting,
)
from tests.common import MockConfigEntry
class MockInput:
"""Mock Vizio device input."""
def __init__(self, name) -> None:
"""Initialize mock Vizio device input."""
self.meta_name = name
self.name = name
def get_mock_inputs(input_list) -> list[MockInput]:
"""Return list of MockInput."""
return [MockInput(device_input) for device_input in input_list]
def get_mock_inputs(input_list: list[str]) -> list[InputInfo]:
"""Return list of InputInfo for the given input names."""
return [
InputInfo(name=name, meta_name=name, is_current=False) for name in input_list
]
@pytest.fixture
@@ -78,7 +69,7 @@ async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry)
def vizio_get_unique_id_fixture() -> Generator[None]:
"""Mock get vizio unique ID."""
with patch(
"homeassistant.components.vizio.config_flow.VizioAsync.get_unique_id",
"homeassistant.components.vizio.config_flow.Vizio.get_serial_number",
AsyncMock(return_value=UNIQUE_ID),
):
yield
@@ -87,9 +78,15 @@ def vizio_get_unique_id_fixture() -> Generator[None]:
@pytest.fixture(name="vizio_data_coordinator_update", autouse=True)
def vizio_data_coordinator_update_fixture() -> Generator[None]:
"""Mock get data coordinator update."""
with patch(
"homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
return_value=APP_LIST,
with (
patch(
"homeassistant.components.vizio.coordinator.fetch_remote_app_catalog",
return_value=APP_RECORDS,
),
patch(
"homeassistant.components.vizio.coordinator.fetch_app_availability",
return_value=(),
),
):
yield
@@ -107,9 +104,15 @@ def no_delay_secs() -> Generator[None]:
@pytest.fixture(name="vizio_data_coordinator_update_failure")
def vizio_data_coordinator_update_failure_fixture() -> Generator[None]:
"""Mock get data coordinator update failure."""
with patch(
"homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
return_value=None,
with (
patch(
"homeassistant.components.vizio.coordinator.fetch_remote_app_catalog",
side_effect=VizioConnectionError("fetch failed"),
),
patch(
"homeassistant.components.vizio.coordinator.fetch_app_availability",
return_value=(),
),
):
yield
@@ -118,8 +121,8 @@ def vizio_data_coordinator_update_failure_fixture() -> Generator[None]:
def vizio_no_unique_id_fixture() -> Generator[None]:
"""Mock no vizio unique ID returrned."""
with patch(
"homeassistant.components.vizio.config_flow.VizioAsync.get_unique_id",
return_value=None,
"homeassistant.components.vizio.config_flow.Vizio.get_serial_number",
side_effect=VizioConnectionError("cannot connect"),
):
yield
@@ -127,9 +130,15 @@ def vizio_no_unique_id_fixture() -> Generator[None]:
@pytest.fixture(name="vizio_connect")
def vizio_connect_fixture() -> Generator[None]:
"""Mock valid vizio device and entry setup."""
with patch(
"homeassistant.components.vizio.config_flow.VizioAsync.validate_ha_config",
AsyncMock(return_value=True),
with (
patch(
"homeassistant.components.vizio.config_flow.Vizio.ping",
AsyncMock(return_value=None),
),
patch(
"homeassistant.components.vizio.config_flow.Vizio.ping_auth",
AsyncMock(return_value=None),
),
):
yield
@@ -139,12 +148,12 @@ def vizio_complete_pairing_fixture() -> Generator[None]:
"""Mock complete vizio pairing workflow."""
with (
patch(
"homeassistant.components.vizio.config_flow.VizioAsync.start_pair",
return_value=MockStartPairingResponse(CH_TYPE, RESPONSE_TOKEN),
"homeassistant.components.vizio.config_flow.Vizio.begin_pair",
return_value=PAIR_CHALLENGE,
),
patch(
"homeassistant.components.vizio.config_flow.VizioAsync.pair",
return_value=MockCompletePairingResponse(ACCESS_TOKEN),
"homeassistant.components.vizio.config_flow.Vizio.finish_pair",
return_value=ACCESS_TOKEN,
),
):
yield
@@ -154,8 +163,8 @@ def vizio_complete_pairing_fixture() -> Generator[None]:
def vizio_start_pairing_failure_fixture() -> Generator[None]:
"""Mock vizio start pairing failure."""
with patch(
"homeassistant.components.vizio.config_flow.VizioAsync.start_pair",
return_value=None,
"homeassistant.components.vizio.config_flow.Vizio.begin_pair",
side_effect=VizioConnectionError("cannot connect"),
):
yield
@@ -165,12 +174,12 @@ def vizio_invalid_pin_failure_fixture() -> Generator[None]:
"""Mock vizio failure due to invalid pin."""
with (
patch(
"homeassistant.components.vizio.config_flow.VizioAsync.start_pair",
return_value=MockStartPairingResponse(CH_TYPE, RESPONSE_TOKEN),
"homeassistant.components.vizio.config_flow.Vizio.begin_pair",
return_value=PAIR_CHALLENGE,
),
patch(
"homeassistant.components.vizio.config_flow.VizioAsync.pair",
return_value=None,
"homeassistant.components.vizio.config_flow.Vizio.finish_pair",
side_effect=VizioConnectionError("invalid pin"),
),
):
yield
@@ -188,31 +197,31 @@ def vizio_bypass_update_fixture() -> Generator[None]:
"""Mock component update with minimal data."""
with (
patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
"homeassistant.components.vizio.Vizio.get_power_state",
return_value=True,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_all_settings",
"homeassistant.components.vizio.Vizio.get_settings",
return_value=None,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_current_input",
"homeassistant.components.vizio.Vizio.get_current_input",
return_value=None,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_inputs_list",
"homeassistant.components.vizio.Vizio.get_inputs",
return_value=None,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
"homeassistant.components.vizio.Vizio.get_current_app_config",
return_value=None,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_model_name",
"homeassistant.components.vizio.Vizio.get_model_name",
return_value=None,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_version",
"homeassistant.components.vizio.Vizio.get_version",
return_value=None,
),
):
@@ -221,10 +230,10 @@ def vizio_bypass_update_fixture() -> Generator[None]:
@pytest.fixture(name="vizio_guess_device_type")
def vizio_guess_device_type_fixture() -> Generator[None]:
"""Mock vizio async_guess_device_type function."""
"""Mock vizio device type probe to report a speaker."""
with patch(
"homeassistant.components.vizio.config_flow.async_guess_device_type",
return_value="speaker",
"homeassistant.components.vizio.config_flow.async_is_tv",
return_value=False,
):
yield
@@ -234,20 +243,24 @@ def vizio_cant_connect_fixture() -> Generator[None]:
"""Mock vizio device can't connect with valid auth."""
with (
patch(
"homeassistant.components.vizio.config_flow.VizioAsync.validate_ha_config",
AsyncMock(return_value=False),
"homeassistant.components.vizio.config_flow.Vizio.ping",
side_effect=VizioConnectionError("cannot connect"),
),
patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
return_value=None,
"homeassistant.components.vizio.config_flow.Vizio.ping_auth",
side_effect=VizioConnectionError("cannot connect"),
),
patch(
"homeassistant.components.vizio.VizioAsync.get_model_name",
return_value=None,
"homeassistant.components.vizio.Vizio.get_power_state",
side_effect=VizioConnectionError("cannot connect"),
),
patch(
"homeassistant.components.vizio.VizioAsync.get_version",
return_value=None,
"homeassistant.components.vizio.Vizio.get_model_name",
side_effect=VizioConnectionError("cannot connect"),
),
patch(
"homeassistant.components.vizio.Vizio.get_version",
side_effect=VizioConnectionError("cannot connect"),
),
):
yield
@@ -258,39 +271,46 @@ def vizio_update_fixture() -> Generator[None]:
"""Mock valid updates to vizio device."""
with (
patch(
"homeassistant.components.vizio.VizioAsync.get_all_settings",
"homeassistant.components.vizio.Vizio.get_settings",
return_value={
"volume": int(MAX_VOLUME[DEVICE_CLASS_SPEAKER] / 2),
"eq": CURRENT_EQ,
"mute": "Off",
"volume": audio_setting("volume", int(SOUNDBAR_PROFILE.max_volume / 2)),
"eq": audio_setting("eq", CURRENT_EQ),
"mute": audio_setting("mute", "Off"),
},
),
patch(
"homeassistant.components.vizio.VizioAsync.get_setting_options",
return_value=EQ_LIST,
"homeassistant.components.vizio.Vizio.get_setting",
return_value=SettingInfo(
setting_type="audio",
name="eq",
value=CURRENT_EQ,
hashval=0,
type=SettingType.LIST,
options=tuple(EQ_LIST),
),
),
patch(
"homeassistant.components.vizio.VizioAsync.get_current_input",
"homeassistant.components.vizio.Vizio.get_current_input",
return_value=CURRENT_INPUT,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_inputs_list",
"homeassistant.components.vizio.Vizio.get_inputs",
return_value=get_mock_inputs(INPUT_LIST),
),
patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
"homeassistant.components.vizio.Vizio.get_power_state",
return_value=True,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_model_name",
"homeassistant.components.vizio.Vizio.get_model_name",
return_value=MODEL,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_version",
"homeassistant.components.vizio.Vizio.get_version",
return_value=VERSION,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
"homeassistant.components.vizio.Vizio.get_current_app_config",
return_value=None,
),
):
@@ -302,16 +322,16 @@ def vizio_update_with_apps_fixture(vizio_update: None) -> Generator[None]:
"""Mock valid updates to vizio device that supports apps."""
with (
patch(
"homeassistant.components.vizio.VizioAsync.get_inputs_list",
"homeassistant.components.vizio.Vizio.get_inputs",
return_value=get_mock_inputs(INPUT_LIST_WITH_APPS),
),
patch(
"homeassistant.components.vizio.VizioAsync.get_current_input",
"homeassistant.components.vizio.Vizio.get_current_input",
return_value="CAST",
),
patch(
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
return_value=AppConfig(**CURRENT_APP_CONFIG),
"homeassistant.components.vizio.Vizio.get_current_app_config",
return_value=CURRENT_APP_CONFIG_OBJ,
),
):
yield
@@ -322,16 +342,16 @@ def vizio_update_with_apps_on_input_fixture(vizio_update: None) -> Generator[Non
"""Mock valid updates to vizio device that supports apps but is on a TV input."""
with (
patch(
"homeassistant.components.vizio.VizioAsync.get_inputs_list",
"homeassistant.components.vizio.Vizio.get_inputs",
return_value=get_mock_inputs(INPUT_LIST_WITH_APPS),
),
patch(
"homeassistant.components.vizio.VizioAsync.get_current_input",
"homeassistant.components.vizio.Vizio.get_current_input",
return_value=CURRENT_INPUT,
),
patch(
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
return_value=AppConfig("unknown", 1, "app"),
"homeassistant.components.vizio.Vizio.get_current_app_config",
return_value=AppConfig(app_id="unknown", name_space=1, message="app"),
),
):
yield
+38 -30
View File
@@ -2,6 +2,9 @@
from ipaddress import ip_address
from vizaio import AppConfig, AppRecord, PairChallenge, SettingInfo, SettingType
from vizaio.profiles import SOUNDBAR_PROFILE, TV_PROFILE
from homeassistant.components.media_player import (
DOMAIN as MP_DOMAIN,
MediaPlayerDeviceClass,
@@ -39,26 +42,28 @@ UNIQUE_ID = "testid"
MODEL = "model"
VERSION = "version"
CH_TYPE = 1
RESPONSE_TOKEN = 1234
PIN = "abcd"
PAIR_CHALLENGE = PairChallenge(challenge_type=1, token=1234)
class MockStartPairingResponse:
"""Mock Vizio start pairing response."""
def __init__(self, ch_type: int, token: int) -> None:
"""Initialize mock start pairing response."""
self.ch_type = ch_type
self.token = token
MAX_VOLUME = {
MediaPlayerDeviceClass.TV: TV_PROFILE.max_volume,
MediaPlayerDeviceClass.SPEAKER: SOUNDBAR_PROFILE.max_volume,
}
class MockCompletePairingResponse:
"""Mock Vizio complete pairing response."""
def __init__(self, auth_token: str) -> None:
"""Initialize mock complete pairing response."""
self.auth_token = auth_token
def audio_setting(
name: str, value: int | str, options: tuple[str, ...] = ()
) -> SettingInfo:
"""Build an audio SettingInfo for mock device responses."""
return SettingInfo(
setting_type="audio",
name=name,
value=value,
hashval=0,
type=SettingType.SLIDER if isinstance(value, int) else SettingType.LIST,
options=options,
)
CURRENT_EQ = "Music"
@@ -69,23 +74,25 @@ INPUT_LIST = ["HDMI", "USB", "Bluetooth", "AUX"]
CURRENT_APP = "Hulu"
CURRENT_APP_CONFIG = {CONF_APP_ID: "3", CONF_NAME_SPACE: 4, CONF_MESSAGE: None}
APP_LIST = [
{
"name": "Hulu",
"country": ["*"],
"id": ["1"],
"config": [{"NAME_SPACE": 4, "APP_ID": "3", "MESSAGE": None}],
},
{
"name": "Netflix",
"country": ["*"],
"id": ["2"],
"config": [{"NAME_SPACE": 1, "APP_ID": "2", "MESSAGE": None}],
},
]
APP_NAME_LIST = [app["name"] for app in APP_LIST]
CURRENT_APP_CONFIG_OBJ = AppConfig(app_id="3", name_space=4, message=None)
APP_RECORDS = (
AppRecord(
name="Hulu",
country=("*",),
config=(AppConfig(app_id="3", name_space=4, message=None),),
id="1",
),
AppRecord(
name="Netflix",
country=("*",),
config=(AppConfig(app_id="2", name_space=1, message=None),),
id="2",
),
)
APP_NAME_LIST = [app.name for app in APP_RECORDS]
INPUT_LIST_WITH_APPS = [*INPUT_LIST, "CAST"]
CUSTOM_CONFIG = {CONF_APP_ID: "test", CONF_MESSAGE: None, CONF_NAME_SPACE: 10}
CUSTOM_CONFIG_OBJ = AppConfig(app_id="test", name_space=10, message=None)
ADDITIONAL_APP_CONFIG = {
"name": CURRENT_APP,
CONF_CONFIG: CUSTOM_CONFIG,
@@ -95,6 +102,7 @@ UNKNOWN_APP_CONFIG = {
"NAME_SPACE": 10,
"MESSAGE": None,
}
UNKNOWN_APP_CONFIG_OBJ = AppConfig(app_id="UNKNOWN", name_space=10, message=None)
ENTITY_ID = f"{MP_DOMAIN}.{slugify(NAME)}"
+5 -5
View File
@@ -23,7 +23,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .conftest import setup_integration
from .const import APP_LIST, HOST2, MODEL, NAME2, UNIQUE_ID, VERSION
from .const import APP_RECORDS, HOST2, MODEL, NAME2, UNIQUE_ID, VERSION
from tests.common import MockConfigEntry, async_fire_time_changed
@@ -116,8 +116,8 @@ async def test_apps_coordinator_persists_until_last_tv_unloads(
await hass.async_block_till_done()
with patch(
"homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
return_value=APP_LIST,
"homeassistant.components.vizio.coordinator.fetch_remote_app_catalog",
return_value=APP_RECORDS,
) as mock_fetch:
freezer.tick(timedelta(days=1))
async_fire_time_changed(hass)
@@ -129,8 +129,8 @@ async def test_apps_coordinator_persists_until_last_tv_unloads(
await hass.async_block_till_done()
with patch(
"homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
return_value=APP_LIST,
"homeassistant.components.vizio.coordinator.fetch_remote_app_catalog",
return_value=APP_RECORDS,
) as mock_fetch:
freezer.tick(timedelta(days=2))
async_fire_time_changed(hass)
+105 -89
View File
@@ -8,16 +8,9 @@ from unittest.mock import call, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from pyvizio.api.apps import AppConfig
from pyvizio.const import (
APPS,
DEVICE_CLASS_SPEAKER as VIZIO_DEVICE_CLASS_SPEAKER,
DEVICE_CLASS_TV as VIZIO_DEVICE_CLASS_TV,
INPUT_APPS,
MAX_VOLUME,
UNKNOWN_APP,
)
from syrupy.assertion import SnapshotAssertion
from vizaio import AppConfig, RemoteKey, VizioConnectionError
from vizaio.apps import BUNDLED_APPS, UNKNOWN_APP, is_app_input
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
@@ -64,24 +57,27 @@ from homeassistant.util import dt as dt_util
from .conftest import setup_integration
from .const import (
ADDITIONAL_APP_CONFIG,
APP_LIST,
APP_NAME_LIST,
APP_RECORDS,
CURRENT_APP,
CURRENT_APP_CONFIG,
CURRENT_APP_CONFIG_OBJ,
CURRENT_EQ,
CURRENT_INPUT,
CUSTOM_CONFIG,
CUSTOM_CONFIG_OBJ,
ENTITY_ID,
EQ_LIST,
INPUT_LIST,
INPUT_LIST_WITH_APPS,
MAX_VOLUME,
MOCK_TV_WITH_ADDITIONAL_APPS_CONFIG,
MOCK_TV_WITH_EXCLUDE_CONFIG,
MOCK_TV_WITH_INCLUDE_CONFIG,
NAME,
UNIQUE_ID,
UNKNOWN_APP_CONFIG,
UNKNOWN_APP_CONFIG_OBJ,
VOLUME_STEP,
audio_setting,
)
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
@@ -124,7 +120,9 @@ def _get_ha_power_state(vizio_power_state: bool) -> str:
return STATE_OFF
def _assert_sources_and_volume(attr: dict[str, Any], vizio_device_class: str) -> None:
def _assert_sources_and_volume(
attr: dict[str, Any], vizio_device_class: MediaPlayerDeviceClass
) -> None:
"""Assert source list, source, and volume level based on device class."""
assert attr[ATTR_INPUT_SOURCE_LIST] == INPUT_LIST
assert attr[ATTR_INPUT_SOURCE] == CURRENT_INPUT
@@ -154,15 +152,17 @@ async def _cm_for_test_setup_without_apps(
"""Context manager to setup test for Vizio devices without app patches."""
with (
patch(
"homeassistant.components.vizio.VizioAsync.get_all_settings",
return_value=all_settings,
"homeassistant.components.vizio.Vizio.get_settings",
return_value={
name: audio_setting(name, value) for name, value in all_settings.items()
},
),
patch(
"homeassistant.components.vizio.VizioAsync.get_setting_options",
return_value=EQ_LIST,
"homeassistant.components.vizio.Vizio.get_setting",
return_value=audio_setting("eq", CURRENT_EQ, tuple(EQ_LIST)),
),
patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
"homeassistant.components.vizio.Vizio.get_power_state",
return_value=vizio_power_state,
),
):
@@ -177,7 +177,7 @@ async def _test_setup_tv(
async with _cm_for_test_setup_without_apps(
{
"volume": int(MAX_VOLUME[VIZIO_DEVICE_CLASS_TV] / 2),
"volume": int(MAX_VOLUME[MediaPlayerDeviceClass.TV] / 2),
"mute": "Off",
"eq": CURRENT_EQ,
},
@@ -189,7 +189,7 @@ async def _test_setup_tv(
hass, MediaPlayerDeviceClass.TV, ha_power_state
)
if ha_power_state == STATE_ON:
_assert_sources_and_volume(attr, VIZIO_DEVICE_CLASS_TV)
_assert_sources_and_volume(attr, MediaPlayerDeviceClass.TV)
assert attr[ATTR_SOUND_MODE] == CURRENT_EQ
@@ -200,7 +200,7 @@ async def _test_setup_speaker(
ha_power_state = _get_ha_power_state(vizio_power_state)
audio_settings = {
"volume": int(MAX_VOLUME[VIZIO_DEVICE_CLASS_SPEAKER] / 2),
"volume": int(MAX_VOLUME[MediaPlayerDeviceClass.SPEAKER] / 2),
"mute": "Off",
"eq": CURRENT_EQ,
}
@@ -215,22 +215,22 @@ async def _test_setup_speaker(
hass, MediaPlayerDeviceClass.SPEAKER, ha_power_state
)
if ha_power_state == STATE_ON:
_assert_sources_and_volume(attr, VIZIO_DEVICE_CLASS_SPEAKER)
_assert_sources_and_volume(attr, MediaPlayerDeviceClass.SPEAKER)
assert "sound_mode" in attr
@asynccontextmanager
async def _cm_for_test_setup_tv_with_apps(
hass: HomeAssistant, config_entry: MockConfigEntry, app_config: dict[str, Any]
hass: HomeAssistant, config_entry: MockConfigEntry, app_config: AppConfig | None
) -> AsyncIterator[None]:
"""Context manager to setup test for Vizio TV with support for apps."""
async with _cm_for_test_setup_without_apps(
{"volume": int(MAX_VOLUME[VIZIO_DEVICE_CLASS_TV] / 2), "mute": "Off"},
{"volume": int(MAX_VOLUME[MediaPlayerDeviceClass.TV] / 2), "mute": "Off"},
True,
):
with patch(
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
return_value=AppConfig(**app_config),
"homeassistant.components.vizio.Vizio.get_current_app_config",
return_value=app_config,
):
await setup_integration(hass, config_entry)
@@ -239,8 +239,8 @@ async def _cm_for_test_setup_tv_with_apps(
)
assert (
attr["volume_level"]
== float(int(MAX_VOLUME[VIZIO_DEVICE_CLASS_TV] / 2))
/ MAX_VOLUME[VIZIO_DEVICE_CLASS_TV]
== float(int(MAX_VOLUME[MediaPlayerDeviceClass.TV] / 2))
/ MAX_VOLUME[MediaPlayerDeviceClass.TV]
)
yield
@@ -249,10 +249,8 @@ async def _cm_for_test_setup_tv_with_apps(
def _assert_source_list_with_apps(
list_to_test: list[str], attr: dict[str, Any]
) -> None:
"""Assert source list matches list_to_test after removing INPUT_APPS from list."""
for app_to_remove in INPUT_APPS:
if app_to_remove in list_to_test:
list_to_test.remove(app_to_remove)
"""Assert source list matches list_to_test after removing app inputs."""
list_to_test = [item for item in list_to_test if not is_app_input(item)]
assert attr[ATTR_INPUT_SOURCE_LIST] == list_to_test
@@ -267,13 +265,12 @@ async def _test_service(
**kwargs,
) -> None:
"""Test generic Vizio media player entity service."""
kwargs["log_api_exception"] = False
service_data = {ATTR_ENTITY_ID: ENTITY_ID}
if additional_service_data:
service_data.update(additional_service_data)
with patch(
f"homeassistant.components.vizio.VizioAsync.{vizio_func_name}"
f"homeassistant.components.vizio.Vizio.{vizio_func_name}"
) as service_call:
await hass.services.async_call(
domain,
@@ -348,19 +345,19 @@ async def test_services(
"""Test all Vizio media player entity services."""
await _test_setup_tv(hass, mock_tv_config_entry, True)
await _test_service(hass, MP_DOMAIN, "pow_on", SERVICE_TURN_ON, None)
await _test_service(hass, MP_DOMAIN, "pow_off", SERVICE_TURN_OFF, None)
await _test_service(hass, MP_DOMAIN, "power_on", SERVICE_TURN_ON, None)
await _test_service(hass, MP_DOMAIN, "power_off", SERVICE_TURN_OFF, None)
await _test_service(
hass,
MP_DOMAIN,
"mute_on",
"mute",
SERVICE_VOLUME_MUTE,
{ATTR_MEDIA_VOLUME_MUTED: True},
)
await _test_service(
hass,
MP_DOMAIN,
"mute_off",
"unmute",
SERVICE_VOLUME_MUTE,
{ATTR_MEDIA_VOLUME_MUTED: False},
)
@@ -373,29 +370,43 @@ async def test_services(
"USB",
)
await _test_service(
hass, MP_DOMAIN, "vol_up", SERVICE_VOLUME_UP, None, num=DEFAULT_VOLUME_STEP
)
await _test_service(
hass, MP_DOMAIN, "vol_down", SERVICE_VOLUME_DOWN, None, num=DEFAULT_VOLUME_STEP
hass, MP_DOMAIN, "volume_up", SERVICE_VOLUME_UP, None, steps=DEFAULT_VOLUME_STEP
)
await _test_service(
hass,
MP_DOMAIN,
"vol_up",
"volume_down",
SERVICE_VOLUME_DOWN,
None,
steps=DEFAULT_VOLUME_STEP,
)
await _test_service(
hass,
MP_DOMAIN,
"volume_up",
SERVICE_VOLUME_SET,
{ATTR_MEDIA_VOLUME_LEVEL: 1},
num=50, # From 50% to 100% = 50 steps (TV max volume 100, starting at 50)
steps=50, # From 50% to 100% = 50 steps (TV max volume 100, starting at 50)
)
await _test_service(
hass,
MP_DOMAIN,
"vol_down",
"volume_down",
SERVICE_VOLUME_SET,
{ATTR_MEDIA_VOLUME_LEVEL: 0},
num=100, # From 100% (after previous vol_up) to 0% = 100 steps
steps=100, # From 100% (after previous vol_up) to 0% = 100 steps
)
await _test_service(
hass, MP_DOMAIN, "send_key", SERVICE_MEDIA_NEXT_TRACK, None, RemoteKey.CH_UP
)
await _test_service(
hass,
MP_DOMAIN,
"send_key",
SERVICE_MEDIA_PREVIOUS_TRACK,
None,
RemoteKey.CH_DOWN,
)
await _test_service(hass, MP_DOMAIN, "ch_up", SERVICE_MEDIA_NEXT_TRACK, None)
await _test_service(hass, MP_DOMAIN, "ch_down", SERVICE_MEDIA_PREVIOUS_TRACK, None)
await _test_service(
hass,
MP_DOMAIN,
@@ -428,8 +439,12 @@ async def test_services(
"eq",
"Music",
)
await _test_service(hass, MP_DOMAIN, "play", SERVICE_MEDIA_PLAY, None)
await _test_service(hass, MP_DOMAIN, "pause", SERVICE_MEDIA_PAUSE, None)
await _test_service(
hass, MP_DOMAIN, "send_key", SERVICE_MEDIA_PLAY, None, RemoteKey.PLAY
)
await _test_service(
hass, MP_DOMAIN, "send_key", SERVICE_MEDIA_PAUSE, None, RemoteKey.PAUSE
)
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
@@ -450,7 +465,7 @@ async def test_options_update(
assert config_entry.options == updated_options
await hass.async_block_till_done()
await _test_service(
hass, MP_DOMAIN, "vol_up", SERVICE_VOLUME_UP, None, num=VOLUME_STEP
hass, MP_DOMAIN, "volume_up", SERVICE_VOLUME_UP, None, steps=VOLUME_STEP
)
@@ -465,8 +480,8 @@ async def test_update_available_to_unavailable(
# Simulate device becoming unreachable
with patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
return_value=None,
"homeassistant.components.vizio.Vizio.get_power_state",
side_effect=VizioConnectionError("cannot connect"),
):
freezer.tick(timedelta(minutes=1))
async_fire_time_changed(hass)
@@ -485,8 +500,8 @@ async def test_update_unavailable_to_available(
# First, make device unavailable
with patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
return_value=None,
"homeassistant.components.vizio.Vizio.get_power_state",
side_effect=VizioConnectionError("cannot connect"),
):
freezer.tick(timedelta(minutes=1))
async_fire_time_changed(hass)
@@ -495,7 +510,7 @@ async def test_update_unavailable_to_available(
# Then, make device available again
with patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
"homeassistant.components.vizio.Vizio.get_power_state",
return_value=True,
):
freezer.tick(timedelta(minutes=1))
@@ -512,7 +527,7 @@ async def test_setup_with_apps(
) -> None:
"""Test device setup with apps."""
async with _cm_for_test_setup_tv_with_apps(
hass, mock_tv_config_entry, CURRENT_APP_CONFIG
hass, mock_tv_config_entry, CURRENT_APP_CONFIG_OBJ
):
attr = hass.states.get(ENTITY_ID).attributes
_assert_source_list_with_apps(list(INPUT_LIST_WITH_APPS + APP_NAME_LIST), attr)
@@ -528,7 +543,6 @@ async def test_setup_with_apps(
SERVICE_SELECT_SOURCE,
{ATTR_INPUT_SOURCE: CURRENT_APP},
CURRENT_APP,
APP_LIST,
)
@@ -541,7 +555,9 @@ async def test_setup_with_apps_include(
config_entry = MockConfigEntry(
domain=DOMAIN, data=MOCK_TV_WITH_INCLUDE_CONFIG, unique_id=UNIQUE_ID
)
async with _cm_for_test_setup_tv_with_apps(hass, config_entry, CURRENT_APP_CONFIG):
async with _cm_for_test_setup_tv_with_apps(
hass, config_entry, CURRENT_APP_CONFIG_OBJ
):
attr = hass.states.get(ENTITY_ID).attributes
_assert_source_list_with_apps([*INPUT_LIST_WITH_APPS, CURRENT_APP], attr)
assert CURRENT_APP in attr[ATTR_INPUT_SOURCE_LIST]
@@ -559,7 +575,9 @@ async def test_setup_with_apps_exclude(
config_entry = MockConfigEntry(
domain=DOMAIN, data=MOCK_TV_WITH_EXCLUDE_CONFIG, unique_id=UNIQUE_ID
)
async with _cm_for_test_setup_tv_with_apps(hass, config_entry, CURRENT_APP_CONFIG):
async with _cm_for_test_setup_tv_with_apps(
hass, config_entry, CURRENT_APP_CONFIG_OBJ
):
attr = hass.states.get(ENTITY_ID).attributes
_assert_source_list_with_apps([*INPUT_LIST_WITH_APPS, CURRENT_APP], attr)
assert CURRENT_APP in attr[ATTR_INPUT_SOURCE_LIST]
@@ -580,7 +598,7 @@ async def test_setup_with_apps_additional_apps_config(
async with _cm_for_test_setup_tv_with_apps(
hass,
config_entry,
ADDITIONAL_APP_CONFIG["config"],
CUSTOM_CONFIG_OBJ,
):
attr = hass.states.get(ENTITY_ID).attributes
assert attr[ATTR_INPUT_SOURCE_LIST].count(CURRENT_APP) == 1
@@ -610,7 +628,6 @@ async def test_setup_with_apps_additional_apps_config(
SERVICE_SELECT_SOURCE,
{ATTR_INPUT_SOURCE: "Netflix"},
"Netflix",
APP_LIST,
)
await _test_service(
hass,
@@ -618,14 +635,14 @@ async def test_setup_with_apps_additional_apps_config(
"launch_app_config",
SERVICE_SELECT_SOURCE,
{ATTR_INPUT_SOURCE: CURRENT_APP},
**CUSTOM_CONFIG,
CUSTOM_CONFIG_OBJ,
)
# Test that invalid app does nothing
with (
patch("homeassistant.components.vizio.VizioAsync.launch_app") as service_call1,
patch("homeassistant.components.vizio.Vizio.launch_app") as service_call1,
patch(
"homeassistant.components.vizio.VizioAsync.launch_app_config"
"homeassistant.components.vizio.Vizio.launch_app_config"
) as service_call2,
):
await hass.services.async_call(
@@ -646,7 +663,7 @@ async def test_setup_with_unknown_app_config(
) -> None:
"""Test device setup with apps where app config returned is unknown."""
async with _cm_for_test_setup_tv_with_apps(
hass, mock_tv_config_entry, UNKNOWN_APP_CONFIG
hass, mock_tv_config_entry, UNKNOWN_APP_CONFIG_OBJ
):
attr = hass.states.get(ENTITY_ID).attributes
_assert_source_list_with_apps(list(INPUT_LIST_WITH_APPS + APP_NAME_LIST), attr)
@@ -662,9 +679,7 @@ async def test_setup_with_no_running_app(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test device setup with apps where no app is running."""
async with _cm_for_test_setup_tv_with_apps(
hass, mock_tv_config_entry, vars(AppConfig())
):
async with _cm_for_test_setup_tv_with_apps(hass, mock_tv_config_entry, None):
attr = hass.states.get(ENTITY_ID).attributes
_assert_source_list_with_apps(list(INPUT_LIST_WITH_APPS + APP_NAME_LIST), attr)
assert attr[ATTR_INPUT_SOURCE] == "CAST"
@@ -678,13 +693,13 @@ async def test_setup_tv_without_mute(
) -> None:
"""Test Vizio TV entity setup when mute property isn't returned by Vizio API."""
async with _cm_for_test_setup_without_apps(
{"volume": int(MAX_VOLUME[VIZIO_DEVICE_CLASS_TV] / 2)},
{"volume": int(MAX_VOLUME[MediaPlayerDeviceClass.TV] / 2)},
True,
):
await setup_integration(hass, mock_tv_config_entry)
attr = _get_attr_and_assert_base_attr(hass, MediaPlayerDeviceClass.TV, STATE_ON)
_assert_sources_and_volume(attr, VIZIO_DEVICE_CLASS_TV)
_assert_sources_and_volume(attr, MediaPlayerDeviceClass.TV)
assert "sound_mode" not in attr
assert "is_volume_muted" not in attr
@@ -697,21 +712,19 @@ async def test_apps_update(
) -> None:
"""Test device setup with apps where no app is running."""
with patch(
"homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
return_value=None,
"homeassistant.components.vizio.coordinator.fetch_remote_app_catalog",
side_effect=VizioConnectionError("fetch failed"),
):
async with _cm_for_test_setup_tv_with_apps(
hass, mock_tv_config_entry, vars(AppConfig())
):
async with _cm_for_test_setup_tv_with_apps(hass, mock_tv_config_entry, None):
# Check source list, remove TV inputs, and verify that the integration is
# using the default APPS list
# using the default bundled apps list
sources = hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE_LIST]
apps = list(set(sources) - set(INPUT_LIST))
assert len(apps) == len(APPS)
assert len(apps) == len(BUNDLED_APPS)
with patch(
"homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
return_value=APP_LIST,
"homeassistant.components.vizio.coordinator.fetch_remote_app_catalog",
return_value=APP_RECORDS,
):
async_fire_time_changed(hass, dt_util.now() + timedelta(days=2))
await hass.async_block_till_done()
@@ -719,10 +732,10 @@ async def test_apps_update(
await hass.async_block_till_done()
# Check source list, remove TV inputs, and verify that
# the integration is
# now using the APP_LIST list
# now using the APP_RECORDS list
sources = hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE_LIST]
apps = list(set(sources) - set(INPUT_LIST))
assert len(apps) == len(APP_LIST)
assert len(apps) == len(APP_RECORDS)
@pytest.mark.usefixtures("vizio_connect", "vizio_update_with_apps_on_input")
@@ -752,7 +765,7 @@ async def test_coordinator_update_on_to_off(
# Device turns off
with patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
"homeassistant.components.vizio.Vizio.get_power_state",
return_value=False,
):
freezer.tick(timedelta(minutes=1))
@@ -808,11 +821,14 @@ async def test_sound_mode_feature_toggling(
# Update with audio settings that have no sound mode
with (
patch(
"homeassistant.components.vizio.VizioAsync.get_all_settings",
return_value={"volume": 50, "mute": "Off"},
"homeassistant.components.vizio.Vizio.get_settings",
return_value={
"volume": audio_setting("volume", 50),
"mute": audio_setting("mute", "Off"),
},
),
patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
"homeassistant.components.vizio.Vizio.get_power_state",
return_value=True,
),
):
@@ -842,11 +858,11 @@ async def test_sound_mode_list_cached(
# Update with different sound mode options — cached list should persist
with (
patch(
"homeassistant.components.vizio.VizioAsync.get_setting_options",
return_value=["Different1", "Different2"],
"homeassistant.components.vizio.Vizio.get_setting",
return_value=audio_setting("eq", CURRENT_EQ, ("Different1", "Different2")),
),
patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
"homeassistant.components.vizio.Vizio.get_power_state",
return_value=True,
),
):
+15 -15
View File
@@ -68,7 +68,7 @@ async def test_remote_is_off_when_device_off(
) -> None:
"""Test remote state is off when device is off."""
with patch(
"homeassistant.components.vizio.VizioAsync.get_power_state",
"homeassistant.components.vizio.Vizio.get_power_state",
return_value=False,
):
await setup_integration(hass, mock_speaker_config_entry)
@@ -79,8 +79,8 @@ async def test_remote_is_off_when_device_off(
@pytest.mark.parametrize(
("service", "mock_method"),
[
(SERVICE_TURN_ON, "pow_on"),
(SERVICE_TURN_OFF, "pow_off"),
(SERVICE_TURN_ON, "power_on"),
(SERVICE_TURN_OFF, "power_off"),
],
)
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
@@ -93,7 +93,7 @@ async def test_turn_on_off(
"""Test turning on/off the remote sends the correct power command."""
await setup_integration(hass, mock_speaker_config_entry)
with patch(
f"homeassistant.components.vizio.VizioAsync.{mock_method}",
f"homeassistant.components.vizio.Vizio.{mock_method}",
) as mock_power:
await hass.services.async_call(
REMOTE_DOMAIN,
@@ -101,7 +101,7 @@ async def test_turn_on_off(
{ATTR_ENTITY_ID: REMOTE_ENTITY_ID},
blocking=True,
)
mock_power.assert_called_once_with(log_api_exception=False)
mock_power.assert_called_once_with()
@pytest.mark.parametrize(
@@ -111,7 +111,7 @@ async def test_turn_on_off(
("ch_up", "CH_UP"),
("SMARTCAST", "SMARTCAST"),
# Aliases
("closed_captions", "CC_TOGGLE"),
("next_input", "INPUT_NEXT"),
("channel_up", "CH_UP"),
("enter", "OK"),
("volume_down", "VOL_DOWN"),
@@ -128,7 +128,7 @@ async def test_send_command_tv_valid(
"""Test send_command resolves valid TV commands."""
await setup_integration(hass, mock_tv_config_entry)
with patch(
"homeassistant.components.vizio.VizioAsync.remote",
"homeassistant.components.vizio.Vizio.send_key",
) as mock_remote:
await hass.services.async_call(
REMOTE_DOMAIN,
@@ -139,7 +139,7 @@ async def test_send_command_tv_valid(
},
blocking=True,
)
mock_remote.assert_called_once_with(expected_key, log_api_exception=False)
mock_remote.assert_called_once_with(expected_key)
@pytest.mark.parametrize("command", ["INVALID_KEY", "not_a_key"])
@@ -185,7 +185,7 @@ async def test_send_command_speaker_valid(
"""Test send_command resolves valid speaker commands."""
await setup_integration(hass, mock_speaker_config_entry)
with patch(
"homeassistant.components.vizio.VizioAsync.remote",
"homeassistant.components.vizio.Vizio.send_key",
) as mock_remote:
await hass.services.async_call(
REMOTE_DOMAIN,
@@ -196,7 +196,7 @@ async def test_send_command_speaker_valid(
},
blocking=True,
)
mock_remote.assert_called_once_with(expected_key, log_api_exception=False)
mock_remote.assert_called_once_with(expected_key)
@pytest.mark.parametrize(
@@ -237,7 +237,7 @@ async def test_send_command_multiple(
"""Test send_command with multiple commands in one call."""
await setup_integration(hass, mock_tv_config_entry)
with patch(
"homeassistant.components.vizio.VizioAsync.remote",
"homeassistant.components.vizio.Vizio.send_key",
) as mock_remote:
await hass.services.async_call(
REMOTE_DOMAIN,
@@ -249,8 +249,8 @@ async def test_send_command_multiple(
blocking=True,
)
assert mock_remote.call_count == 2
mock_remote.assert_any_call("UP", log_api_exception=False)
mock_remote.assert_any_call("OK", log_api_exception=False)
mock_remote.assert_any_call("UP")
mock_remote.assert_any_call("OK")
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
@@ -261,7 +261,7 @@ async def test_send_command_invalid_skips_valid(
await setup_integration(hass, mock_tv_config_entry)
with (
patch(
"homeassistant.components.vizio.VizioAsync.remote",
"homeassistant.components.vizio.Vizio.send_key",
) as mock_remote,
pytest.raises(ServiceValidationError),
):
@@ -285,7 +285,7 @@ async def test_send_command_delay_between_repeats(
await setup_integration(hass, mock_tv_config_entry)
with (
patch(
"homeassistant.components.vizio.VizioAsync.remote",
"homeassistant.components.vizio.Vizio.send_key",
) as mock_remote,
patch(
"homeassistant.components.vizio.remote.asyncio.sleep",