jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add config flow to panasonic_viera component (#33829)

Browse Source 
* Updating the panasonic_viera component

* Updating .coveragerc

* Removing testplatform

* Updating strings.json

* Commit before rebase

* Commit before rebase

* Commit before rebase

* Commit before rebase

* Commit before rebase

* Adding tests and stuff

* Fixing permission issues

* Ignoring Pylint warnings

* Fixing one more Pylint warning

* Refactoring

* Commiting changes - part 1

* Commiting changes: part 2

* Turning unknown error logs into exception logs

* Update strings.json

* Rebasing

* Updating the panasonic_viera component

* Removing testplatform

* Updating strings.json

* Commit before rebase

* Commit before rebase

* Commit before rebase

* Commit before rebase

* Commit before rebase

* Adding tests and stuff

* Fixing permission issues

* Ignoring Pylint warnings

* Fixing one more Pylint warning

* Refactoring

* Commiting changes - part 1

* Commiting changes: part 2

* Turning unknown error logs into exception logs

* Adding pt-BR translation

* Removing Brazilian Portugues translations

* Modifying error handling

* Adding SOAPError to except handling

* Updating translation

* Refactoring async_step_import

* Fixing indentation

* Fixing requirements after rebase

* Fixing translations

* Fixing issues after rebase

* Routing import step to user step

* Adding myself as a codeowner
This commit is contained in:
João Gabriel 2020-04-18 00:19:01 -03:00 committed by GitHub
parent dd4a350bd5
commit 42b6ec2fb5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 1106 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -523,6 +523,7 @@ omit =
homeassistant/components/osramlightify/light.py
homeassistant/components/otp/sensor.py
homeassistant/components/panasonic_bluray/media_player.py
homeassistant/components/panasonic_viera/__init__.py
homeassistant/components/panasonic_viera/media_player.py
homeassistant/components/pandora/media_player.py
homeassistant/components/pcal9535a/*

1
CODEOWNERS
View File

@ -279,6 +279,7 @@ homeassistant/components/openweathermap/* @fabaff
homeassistant/components/opnsense/* @mtreinish
homeassistant/components/orangepi_gpio/* @pascallj
homeassistant/components/oru/* @bvlaicu
homeassistant/components/panasonic_viera/* @joogps
homeassistant/components/panel_custom/* @home-assistant/frontend
homeassistant/components/panel_iframe/* @home-assistant/frontend
homeassistant/components/pcal9535a/* @Shulyaka

31
homeassistant/components/panasonic_viera/.translations/en.json Normal file
View File

@ -0,0 +1,31 @@
{
"config": {
"abort": {
"already_configured": "This Panasonic Viera TV is already configured.",
"not_connected": "The remote connection with your Panasonic Viera TV was lost. Check the logs for more information.",
"unknown": "An unknown error occured. Check the logs for more information."
},
"error": {
"invalid_pin_code": "The PIN code you entered was invalid",
"not_connected": "Could not establish a remote connection with your Panasonic Viera TV"
},
"step": {
"pairing": {
"data": {
"pin": "PIN"
},
"description": "Enter the PIN displayed on your TV",
"title": "Pairing"
},
"user": {
"data": {
"host": "IP address",
"name": "Name"
},
"description": "Enter your Panasonic Viera TV's IP address",
"title": "Setup your TV"
}
}
},
"title": "Panasonic Viera"
}

70
homeassistant/components/panasonic_viera/__init__.py
View File

@ -1 +1,69 @@
"""The panasonic_viera component."""
"""The Panasonic Viera integration."""
import asyncio
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
import homeassistant.helpers.config_validation as cv
from .const import CONF_ON_ACTION, DEFAULT_NAME, DEFAULT_PORT, DOMAIN
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
cv.ensure_list,
[
vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_ON_ACTION): cv.SCRIPT_SCHEMA,
}
)
],
)
},
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = ["media_player"]
async def async_setup(hass, config):
"""Set up Panasonic Viera from configuration.yaml."""
if DOMAIN not in config:
return True
for conf in config[DOMAIN]:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=conf
)
)
return True
async def async_setup_entry(hass, config_entry):
"""Set up Panasonic Viera from a config entry."""
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, component)
)
return True
async def async_unload_entry(hass, config_entry):
"""Unload a config entry."""
return all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in PLATFORMS
]
)
)

152
homeassistant/components/panasonic_viera/config_flow.py Normal file
View File

@ -0,0 +1,152 @@
"""Config flow for Panasonic Viera TV integration."""
from functools import partial
import logging
from urllib.request import URLError
from panasonic_viera import TV_TYPE_ENCRYPTED, RemoteControl, SOAPError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PIN, CONF_PORT
from .const import ( # pylint: disable=unused-import
CONF_APP_ID,
CONF_ENCRYPTION_KEY,
CONF_ON_ACTION,
DEFAULT_NAME,
DEFAULT_PORT,
DOMAIN,
ERROR_INVALID_PIN_CODE,
ERROR_NOT_CONNECTED,
REASON_NOT_CONNECTED,
REASON_UNKNOWN,
)
_LOGGER = logging.getLogger(__name__)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Config flow for Panasonic Viera."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
"""Initialize the Panasonic Viera config flow."""
self._data = {
CONF_HOST: None,
CONF_NAME: None,
CONF_PORT: None,
CONF_ON_ACTION: None,
}
self._remote = None
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
await self.async_load_data(user_input)
try:
self._remote = await self.hass.async_add_executor_job(
partial(RemoteControl, self._data[CONF_HOST], self._data[CONF_PORT])
)
except (TimeoutError, URLError, SOAPError, OSError) as err:
_LOGGER.error("Could not establish remote connection: %s", err)
errors["base"] = ERROR_NOT_CONNECTED
except Exception as err: # pylint: disable=broad-except
_LOGGER.exception("An unknown error occurred: %s", err)
return self.async_abort(reason=REASON_UNKNOWN)
if "base" not in errors:
if self._remote.type == TV_TYPE_ENCRYPTED:
return await self.async_step_pairing()
return self.async_create_entry(
title=self._data[CONF_NAME], data=self._data,
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(
CONF_HOST,
default=self._data[CONF_HOST]
if self._data[CONF_HOST] is not None
else "",
): str,
vol.Optional(
CONF_NAME,
default=self._data[CONF_NAME]
if self._data[CONF_NAME] is not None
else DEFAULT_NAME,
): str,
}
),
errors=errors,
)
async def async_step_pairing(self, user_input=None):
"""Handle the pairing step."""
errors = {}
if user_input is not None:
pin = user_input[CONF_PIN]
try:
self._remote.authorize_pin_code(pincode=pin)
except SOAPError as err:
_LOGGER.error("Invalid PIN code: %s", err)
errors["base"] = ERROR_INVALID_PIN_CODE
except (TimeoutError, URLError, OSError) as err:
_LOGGER.error("The remote connection was lost: %s", err)
return self.async_abort(reason=REASON_NOT_CONNECTED)
except Exception as err: # pylint: disable=broad-except
_LOGGER.exception("Unknown error: %s", err)
return self.async_abort(reason=REASON_UNKNOWN)
if "base" not in errors:
encryption_data = {
CONF_APP_ID: self._remote.app_id,
CONF_ENCRYPTION_KEY: self._remote.enc_key,
}
self._data = {**self._data, **encryption_data}
return self.async_create_entry(
title=self._data[CONF_NAME], data=self._data,
)
try:
self._remote.request_pin_code(name="Home Assistant")
except (TimeoutError, URLError, SOAPError, OSError) as err:
_LOGGER.error("The remote connection was lost: %s", err)
return self.async_abort(reason=REASON_NOT_CONNECTED)
except Exception as err: # pylint: disable=broad-except
_LOGGER.exception("Unknown error: %s", err)
return self.async_abort(reason=REASON_UNKNOWN)
return self.async_show_form(
step_id="pairing",
data_schema=vol.Schema({vol.Required(CONF_PIN): str}),
errors=errors,
)
async def async_step_import(self, import_config):
"""Import a config entry from configuration.yaml."""
return await self.async_step_user(user_input=import_config)
async def async_load_data(self, config):
"""Load the data."""
self._data = config
self._data[CONF_PORT] = (
self._data[CONF_PORT] if CONF_PORT in self._data else DEFAULT_PORT
)
self._data[CONF_ON_ACTION] = (
self._data[CONF_ON_ACTION] if CONF_ON_ACTION in self._data else None
)
await self.async_set_unique_id(self._data[CONF_HOST])
self._abort_if_unique_id_configured()

17
homeassistant/components/panasonic_viera/const.py Normal file
View File

@ -0,0 +1,17 @@
"""Constants for the Panasonic Viera integration."""
DOMAIN = "panasonic_viera"
DEVICE_MANUFACTURER = "Panasonic"
CONF_ON_ACTION = "turn_on_action"
CONF_APP_ID = "app_id"
CONF_ENCRYPTION_KEY = "encryption_key"
DEFAULT_NAME = "Panasonic Viera TV"
DEFAULT_PORT = 55000
ERROR_NOT_CONNECTED = "not_connected"
ERROR_INVALID_PIN_CODE = "invalid_pin_code"
REASON_NOT_CONNECTED = "not_connected"
REASON_UNKNOWN = "unknown"

7
homeassistant/components/panasonic_viera/manifest.json
View File

@ -1,7 +1,8 @@
{
"domain": "panasonic_viera",
"name": "Panasonic Viera TV",
"name": "Panasonic Viera",
"documentation": "https://www.home-assistant.io/integrations/panasonic_viera",
"requirements": ["panasonic_viera==0.3.2", "wakeonlan==1.1.6"],
"codeowners": []
"requirements": ["panasonic_viera==0.3.5"],
"codeowners": ["@joogps"],
"config_flow": true
}

360
homeassistant/components/panasonic_viera/media_player.py
View File

@ -1,11 +1,11 @@
"""Support for interface with a Panasonic Viera TV."""
from functools import partial
import logging
from urllib.request import URLError
from panasonic_viera import RemoteControl
import voluptuous as vol
import wakeonlan
from panasonic_viera import EncryptionRequired, Keys, RemoteControl, SOAPError
from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerDevice
from homeassistant.components.media_player import MediaPlayerDevice
from homeassistant.components.media_player.const import (
MEDIA_TYPE_URL,
SUPPORT_NEXT_TRACK,
@ -20,25 +20,10 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_SET,
SUPPORT_VOLUME_STEP,
)
from homeassistant.const import (
CONF_BROADCAST_ADDRESS,
CONF_HOST,
CONF_MAC,
CONF_NAME,
CONF_PORT,
STATE_OFF,
STATE_ON,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, STATE_OFF, STATE_ON
from homeassistant.helpers.script import Script
_LOGGER = logging.getLogger(__name__)
CONF_APP_POWER = "app_power"
DEFAULT_NAME = "Panasonic Viera TV"
DEFAULT_PORT = 55000
DEFAULT_BROADCAST_ADDRESS = "255.255.255.255"
DEFAULT_APP_POWER = False
from .const import CONF_APP_ID, CONF_ENCRYPTION_KEY, CONF_ON_ACTION
SUPPORT_VIERATV = (
SUPPORT_PAUSE
@ -48,99 +33,58 @@ SUPPORT_VIERATV = (
| SUPPORT_PREVIOUS_TRACK
| SUPPORT_NEXT_TRACK
| SUPPORT_TURN_OFF
| SUPPORT_TURN_ON
| SUPPORT_PLAY
| SUPPORT_PLAY_MEDIA
| SUPPORT_STOP
)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_MAC): cv.string,
vol.Optional(
CONF_BROADCAST_ADDRESS, default=DEFAULT_BROADCAST_ADDRESS
): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_APP_POWER, default=DEFAULT_APP_POWER): cv.boolean,
}
)
_LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Panasonic Viera TV platform."""
mac = config.get(CONF_MAC)
broadcast = config.get(CONF_BROADCAST_ADDRESS)
name = config.get(CONF_NAME)
port = config.get(CONF_PORT)
app_power = config.get(CONF_APP_POWER)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up Panasonic Viera TV from a config entry."""
if discovery_info:
_LOGGER.debug("%s", discovery_info)
name = discovery_info.get("name")
host = discovery_info.get("host")
port = discovery_info.get("port")
udn = discovery_info.get("udn")
if udn and udn.startswith("uuid:"):
uuid = udn[len("uuid:") :]
else:
uuid = None
remote = RemoteControl(host, port)
add_entities([PanasonicVieraTVDevice(mac, name, remote, host, app_power, uuid)])
return True
config = config_entry.data
host = config.get(CONF_HOST)
remote = RemoteControl(host, port)
host = config[CONF_HOST]
port = config[CONF_PORT]
name = config[CONF_NAME]
add_entities(
[PanasonicVieraTVDevice(mac, name, remote, host, broadcast, app_power)]
)
return True
on_action = config[CONF_ON_ACTION]
if on_action is not None:
on_action = Script(hass, on_action)
params = {}
if CONF_APP_ID in config and CONF_ENCRYPTION_KEY in config:
params["app_id"] = config[CONF_APP_ID]
params["encryption_key"] = config[CONF_ENCRYPTION_KEY]
remote = Remote(hass, host, port, on_action, **params)
await remote.async_create_remote_control(during_setup=True)
tv_device = PanasonicVieraTVDevice(remote, name)
async_add_entities([tv_device])
class PanasonicVieraTVDevice(MediaPlayerDevice):
"""Representation of a Panasonic Viera TV."""
def __init__(self, mac, name, remote, host, broadcast, app_power, uuid=None):
def __init__(
self, remote, name, uuid=None,
):
"""Initialize the Panasonic device."""
# Save a reference to the imported class
self._wol = wakeonlan
self._mac = mac
self._remote = remote
self._name = name
self._uuid = uuid
self._muted = False
self._playing = True
self._state = None
self._remote = remote
self._host = host
self._broadcast = broadcast
self._volume = 0
self._app_power = app_power
@property
def unique_id(self) -> str:
"""Return the unique ID of this Viera TV."""
return self._uuid
def update(self):
"""Retrieve the latest data."""
try:
self._muted = self._remote.get_mute()
self._volume = self._remote.get_volume() / 100
self._state = STATE_ON
except OSError:
self._state = STATE_OFF
def send_key(self, key):
"""Send a key to the tv and handles exceptions."""
try:
self._remote.send_key(key)
self._state = STATE_ON
except OSError:
self._state = STATE_OFF
return False
return True
@property
def name(self):
"""Return the name of the device."""
@ -149,98 +93,208 @@ class PanasonicVieraTVDevice(MediaPlayerDevice):
@property
def state(self):
"""Return the state of the device."""
return self._state
return self._remote.state
@property
def available(self):
"""Return if True the device is available."""
return self._remote.available
@property
def volume_level(self):
"""Volume level of the media player (0..1)."""
return self._volume
return self._remote.volume
@property
def is_volume_muted(self):
"""Boolean if volume is currently muted."""
return self._muted
return self._remote.muted
@property
def supported_features(self):
"""Flag media player features that are supported."""
if self._mac or self._app_power:
return SUPPORT_VIERATV | SUPPORT_TURN_ON
return SUPPORT_VIERATV
def turn_on(self):
async def async_update(self):
"""Retrieve the latest data."""
await self._remote.async_update()
async def async_turn_on(self):
"""Turn on the media player."""
if self._mac:
self._wol.send_magic_packet(self._mac, ip_address=self._broadcast)
self._state = STATE_ON
elif self._app_power:
self._remote.turn_on()
self._state = STATE_ON
await self._remote.async_turn_on()
def turn_off(self):
async def async_turn_off(self):
"""Turn off media player."""
if self._state != STATE_OFF:
self._remote.turn_off()
self._state = STATE_OFF
await self._remote.async_turn_off()
def volume_up(self):
async def async_volume_up(self):
"""Volume up the media player."""
self._remote.volume_up()
await self._remote.async_send_key(Keys.volume_up)
def volume_down(self):
async def async_volume_down(self):
"""Volume down media player."""
self._remote.volume_down()
await self._remote.async_send_key(Keys.volume_down)
def mute_volume(self, mute):
async def async_mute_volume(self, mute):
"""Send mute command."""
self._remote.set_mute(mute)
await self._remote.async_set_mute(mute)
def set_volume_level(self, volume):
async def async_set_volume_level(self, volume):
"""Set volume level, range 0..1."""
await self._remote.async_set_volume(volume)
async def async_media_play_pause(self):
"""Simulate play pause media player."""
if self._remote.playing:
await self._remote.async_send_key(Keys.pause)
self._remote.playing = False
else:
await self._remote.async_send_key(Keys.play)
self._remote.playing = True
async def async_media_play(self):
"""Send play command."""
await self._remote.async_send_key(Keys.play)
self._remote.playing = True
async def async_media_pause(self):
"""Send pause command."""
await self._remote.async_send_key(Keys.pause)
self._remote.playing = False
async def async_media_stop(self):
"""Stop playback."""
await self._remote.async_send_key(Keys.stop)
async def async_media_next_track(self):
"""Send the fast forward command."""
await self._remote.async_send_key(Keys.fast_forward)
async def async_media_previous_track(self):
"""Send the rewind command."""
await self._remote.async_send_key(Keys.rewind)
async def async_play_media(self, media_type, media_id, **kwargs):
"""Play media."""
await self._remote.async_play_media(media_type, media_id)
class Remote:
"""The Remote class. It stores the TV properties and the remote control connection itself."""
def __init__(
self, hass, host, port, on_action=None, app_id=None, encryption_key=None,
):
"""Initialize the Remote class."""
self._hass = hass
self._host = host
self._port = port
self._on_action = on_action
self._app_id = app_id
self._encryption_key = encryption_key
self.state = None
self.available = False
self.volume = 0
self.muted = False
self.playing = True
self._control = None
async def async_create_remote_control(self, during_setup=False):
"""Create remote control."""
control_existed = self._control is not None
try:
params = {}
if self._app_id and self._encryption_key:
params["app_id"] = self._app_id
params["encryption_key"] = self._encryption_key
self._control = await self._hass.async_add_executor_job(
partial(RemoteControl, self._host, self._port, **params)
)
self.state = STATE_ON
self.available = True
except (TimeoutError, URLError, SOAPError, OSError) as err:
if control_existed or during_setup:
_LOGGER.error("Could not establish remote connection: %s", err)
self._control = None
self.state = STATE_OFF
self.available = self._on_action is not None
except Exception as err: # pylint: disable=broad-except
if control_existed or during_setup:
_LOGGER.exception("An unknown error occurred: %s", err)
self._control = None
self.state = STATE_OFF
self.available = self._on_action is not None
async def async_update(self):
"""Update device data."""
if self._control is None:
await self.async_create_remote_control()
return
await self._handle_errors(self._update)
async def _update(self):
"""Retrieve the latest data."""
self.muted = self._control.get_mute()
self.volume = self._control.get_volume() / 100
self.state = STATE_ON
self.available = True
async def async_send_key(self, key):
"""Send a key to the TV and handle exceptions."""
await self._handle_errors(self._control.send_key, key)
async def async_turn_on(self):
"""Turn on the TV."""
if self._on_action is not None:
await self._on_action.async_run()
self.state = STATE_ON
elif self.state != STATE_ON:
await self.async_send_key(Keys.power)
self.state = STATE_ON
async def async_turn_off(self):
"""Turn off the TV."""
if self.state != STATE_OFF:
await self.async_send_key(Keys.power)
self.state = STATE_OFF
await self.async_update()
async def async_set_mute(self, enable):
"""Set mute based on 'enable'."""
await self._handle_errors(self._control.set_mute, enable)
async def async_set_volume(self, volume):
"""Set volume level, range 0..1."""
volume = int(volume * 100)
try:
self._remote.set_volume(volume)
self._state = STATE_ON
except OSError:
self._state = STATE_OFF
await self._handle_errors(self._control.set_volume, volume)
def media_play_pause(self):
"""Simulate play pause media player."""
if self._playing:
self.media_pause()
else:
self.media_play()
def media_play(self):
"""Send play command."""
self._playing = True
self._remote.media_play()
def media_pause(self):
"""Send media pause command to media player."""
self._playing = False
self._remote.media_pause()
def media_next_track(self):
"""Send next track command."""
self._remote.media_next_track()
def media_previous_track(self):
"""Send the previous track command."""
self._remote.media_previous_track()
def play_media(self, media_type, media_id, **kwargs):
async def async_play_media(self, media_type, media_id):
"""Play media."""
_LOGGER.debug("Play media: %s (%s)", media_id, media_type)
if media_type == MEDIA_TYPE_URL:
try:
self._remote.open_webpage(media_id)
except (TimeoutError, OSError):
self._state = STATE_OFF
else:
if media_type != MEDIA_TYPE_URL:
_LOGGER.warning("Unsupported media_type: %s", media_type)
return
def media_stop(self):
"""Stop playback."""
self.send_key("NRC_CANCEL-ONOFF")
await self._handle_errors(self._control.open_webpage, media_id)
async def _handle_errors(self, func, *args):
"""Handle errors from func, set available and reconnect if needed."""
try:
await self._hass.async_add_executor_job(func, *args)
except EncryptionRequired:
_LOGGER.error("The connection couldn't be encrypted")
except (TimeoutError, URLError, SOAPError, OSError):
self.state = STATE_OFF
self.available = self._on_action is not None
await self.async_create_remote_control()

31
homeassistant/components/panasonic_viera/strings.json Normal file
View File

@ -0,0 +1,31 @@
{
"title": "Panasonic Viera",
"config": {
"step": {
"user": {
"title": "Setup your TV",
"description": "Enter your Panasonic Viera TV's IP address",
"data": {
"host": "IP address",
"name": "Name"
}
},
"pairing": {
"title": "Pairing",
"description": "Enter the PIN displayed on your TV",
"data": {
"pin": "PIN"
}
}
},
"error": {
"not_connected": "Could not establish a remote connection with your Panasonic Viera TV",
"invalid_pin_code": "The PIN code you entered was invalid"
},
"abort": {
"already_configured": "This Panasonic Viera TV is already configured.",
"not_connected": "The remote connection with your Panasonic Viera TV was lost. Check the logs for more information.",
"unknown": "An unknown error occured. Check the logs for more information."
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -89,6 +89,7 @@ FLOWS = [
"opentherm_gw",
"openuv",
"owntracks",
"panasonic_viera",
"plaato",
"plex",
"point",

3
requirements_all.txt
View File

@ -1000,7 +1000,7 @@ paho-mqtt==1.5.0
panacotta==0.1
# homeassistant.components.panasonic_viera
panasonic_viera==0.3.2
panasonic_viera==0.3.5
# homeassistant.components.pcal9535a
pcal9535a==0.7
@ -2105,7 +2105,6 @@ vtjp==0.1.14
# homeassistant.components.vultr
vultr==0.1.2
# homeassistant.components.panasonic_viera
# homeassistant.components.wake_on_lan
wakeonlan==1.1.6

4
requirements_test_all.txt
View File

@ -383,6 +383,9 @@ openerz-api==0.1.0
# homeassistant.components.shiftr
paho-mqtt==1.5.0
# homeassistant.components.panasonic_viera
panasonic_viera==0.3.5
# homeassistant.components.aruba
# homeassistant.components.cisco_ios
# homeassistant.components.pandora
@ -790,7 +793,6 @@ vsure==1.5.4
# homeassistant.components.vultr
vultr==0.1.2
# homeassistant.components.panasonic_viera
# homeassistant.components.wake_on_lan
wakeonlan==1.1.6

1
tests/components/panasonic_viera/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Panasonic Viera component."""

587
tests/components/panasonic_viera/test_config_flow.py Normal file
View File

@ -0,0 +1,587 @@
"""Test the Panasonic Viera config flow."""
from unittest.mock import Mock
from asynctest import patch
from panasonic_viera import TV_TYPE_ENCRYPTED, TV_TYPE_NONENCRYPTED, SOAPError
import pytest
from homeassistant import config_entries
from homeassistant.components.panasonic_viera.const import (
CONF_APP_ID,
CONF_ENCRYPTION_KEY,
CONF_ON_ACTION,
DEFAULT_NAME,
DEFAULT_PORT,
DOMAIN,
ERROR_INVALID_PIN_CODE,
ERROR_NOT_CONNECTED,
REASON_NOT_CONNECTED,
REASON_UNKNOWN,
)
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PIN, CONF_PORT
from tests.common import MockConfigEntry
@pytest.fixture(name="panasonic_viera_setup", autouse=True)
def panasonic_viera_setup_fixture():
"""Mock panasonic_viera setup."""
with patch(
"homeassistant.components.panasonic_viera.async_setup", return_value=True
), patch(
"homeassistant.components.panasonic_viera.async_setup_entry", return_value=True,
):
yield
def get_mock_remote(
host="1.2.3.4",
authorize_error=None,
encrypted=False,
app_id=None,
encryption_key=None,
):
"""Return a mock remote."""
mock_remote = Mock()
mock_remote.type = TV_TYPE_ENCRYPTED if encrypted else TV_TYPE_NONENCRYPTED
mock_remote.app_id = app_id
mock_remote.enc_key = encryption_key
def request_pin_code(name=None):
return
mock_remote.request_pin_code = request_pin_code
def authorize_pin_code(pincode):
if pincode == "1234":
return
if authorize_error is not None:
raise authorize_error
mock_remote.authorize_pin_code = authorize_pin_code
return mock_remote
async def test_flow_non_encrypted(hass):
"""Test flow without encryption."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
mock_remote = get_mock_remote(encrypted=False)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "create_entry"
assert result["title"] == DEFAULT_NAME
assert result["data"] == {
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: None,
}
async def test_flow_not_connected_error(hass):
"""Test flow with connection error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
side_effect=TimeoutError,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {"base": ERROR_NOT_CONNECTED}
async def test_flow_unknown_abort(hass):
"""Test flow with unknown error abortion."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
side_effect=Exception,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "abort"
assert result["reason"] == REASON_UNKNOWN
async def test_flow_encrypted_valid_pin_code(hass):
"""Test flow with encryption and valid PIN code."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
mock_remote = get_mock_remote(
encrypted=True, app_id="test-app-id", encryption_key="test-encryption-key",
)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_PIN: "1234"},
)
assert result["type"] == "create_entry"
assert result["title"] == DEFAULT_NAME
assert result["data"] == {
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: None,
CONF_APP_ID: "test-app-id",
CONF_ENCRYPTION_KEY: "test-encryption-key",
}
async def test_flow_encrypted_invalid_pin_code_error(hass):
"""Test flow with encryption and invalid PIN code error during pairing step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
mock_remote = get_mock_remote(encrypted=True, authorize_error=SOAPError)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_PIN: "0000"},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
assert result["errors"] == {"base": ERROR_INVALID_PIN_CODE}
async def test_flow_encrypted_not_connected_abort(hass):
"""Test flow with encryption and PIN code connection error abortion during pairing step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
mock_remote = get_mock_remote(encrypted=True, authorize_error=TimeoutError)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_PIN: "0000"},
)
assert result["type"] == "abort"
assert result["reason"] == REASON_NOT_CONNECTED
async def test_flow_encrypted_unknown_abort(hass):
"""Test flow with encryption and PIN code unknown error abortion during pairing step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
mock_remote = get_mock_remote(encrypted=True, authorize_error=Exception)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_PIN: "0000"},
)
assert result["type"] == "abort"
assert result["reason"] == REASON_UNKNOWN
async def test_flow_non_encrypted_already_configured_abort(hass):
"""Test flow without encryption and existing config entry abortion."""
MockConfigEntry(
domain=DOMAIN,
unique_id="1.2.3.4",
data={CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME, CONF_PORT: DEFAULT_PORT},
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
async def test_flow_encrypted_already_configured_abort(hass):
"""Test flow with encryption and existing config entry abortion."""
MockConfigEntry(
domain=DOMAIN,
unique_id="1.2.3.4",
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_APP_ID: "test-app-id",
CONF_ENCRYPTION_KEY: "test-encryption-key",
},
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
async def test_imported_flow_non_encrypted(hass):
"""Test imported flow without encryption."""
mock_remote = get_mock_remote(encrypted=False)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
},
)
assert result["type"] == "create_entry"
assert result["title"] == DEFAULT_NAME
assert result["data"] == {
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
}
async def test_imported_flow_encrypted_valid_pin_code(hass):
"""Test imported flow with encryption and valid PIN code."""
mock_remote = get_mock_remote(
encrypted=True, app_id="test-app-id", encryption_key="test-encryption-key",
)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_PIN: "1234"},
)
assert result["type"] == "create_entry"
assert result["title"] == DEFAULT_NAME
assert result["data"] == {
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
CONF_APP_ID: "test-app-id",
CONF_ENCRYPTION_KEY: "test-encryption-key",
}
async def test_imported_flow_encrypted_invalid_pin_code_error(hass):
"""Test imported flow with encryption and invalid PIN code error during pairing step."""
mock_remote = get_mock_remote(encrypted=True, authorize_error=SOAPError)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_PIN: "0000"},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
assert result["errors"] == {"base": ERROR_INVALID_PIN_CODE}
async def test_imported_flow_encrypted_not_connected_abort(hass):
"""Test imported flow with encryption and PIN code connection error abortion during pairing step."""
mock_remote = get_mock_remote(encrypted=True, authorize_error=TimeoutError)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_PIN: "0000"},
)
assert result["type"] == "abort"
assert result["reason"] == REASON_NOT_CONNECTED
async def test_imported_flow_encrypted_unknown_abort(hass):
"""Test imported flow with encryption and PIN code unknown error abortion during pairing step."""
mock_remote = get_mock_remote(encrypted=True, authorize_error=Exception)
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
return_value=mock_remote,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
},
)
assert result["type"] == "form"
assert result["step_id"] == "pairing"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_PIN: "0000"},
)
assert result["type"] == "abort"
assert result["reason"] == REASON_UNKNOWN
async def test_imported_flow_not_connected_error(hass):
"""Test imported flow with connection error abortion."""
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
side_effect=TimeoutError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
},
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {"base": ERROR_NOT_CONNECTED}
async def test_imported_flow_unknown_abort(hass):
"""Test imported flow with unknown error abortion."""
with patch(
"homeassistant.components.panasonic_viera.config_flow.RemoteControl",
side_effect=Exception,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
},
)
assert result["type"] == "abort"
assert result["reason"] == REASON_UNKNOWN
async def test_imported_flow_non_encrypted_already_configured_abort(hass):
"""Test imported flow without encryption and existing config entry abortion."""
MockConfigEntry(
domain=DOMAIN,
unique_id="1.2.3.4",
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
},
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
async def test_imported_flow_encrypted_already_configured_abort(hass):
"""Test imported flow with encryption and existing config entry abortion."""
MockConfigEntry(
domain=DOMAIN,
unique_id="1.2.3.4",
data={
CONF_HOST: "1.2.3.4",
CONF_NAME: DEFAULT_NAME,
CONF_PORT: DEFAULT_PORT,
CONF_ON_ACTION: "test-on-action",
CONF_APP_ID: "test-app-id",
CONF_ENCRYPTION_KEY: "test-encryption-key",
},
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={CONF_HOST: "1.2.3.4", CONF_NAME: DEFAULT_NAME},
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"