evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add component media player.ps4 (#21074)

Browse source 
* Added PS4/ __init__.py

* Create en.json

* Create config_flow.py

* Create const.py

* Create media_player.py

* Create services.yaml

* Create strings.json

* Create __init__.py

* Add test_config_flow.py/ Finished adding PS4 files

* Rewrote for loop into short-hand

* bumped pyps4 to 0.2.8

* Pass in helper()

* Rewrite func

* Fixed test

* Added import in init

* bump to 0.2.9

* bump to 0.3.0

* Removed line

* lint

* Add ps4 to flows list

* Added pyps4-homeassistant with script

* Added pyps4

* Added pypys4 to test

* removed list def

* reformatted service call dicts

* removed config from device class

* typo

* removed line

* reformatted .. format

* redefined property

* reformat load games func

* Add __init__ and media_player.py to coveragerc

* Fix for test

* remove init

* remove blank line

* remove mock_coro

* Revert "remove init"

This reverts commit b68996aa34699bf38781e153acdd597579e8131f.

* Correct permissions

* fixes

* fixes
This commit is contained in:
ktnrg45 2019-02-17 13:41:55 -07:00 • committed by Charles Garwood
parent 4e7cfc923d
commit 72ef9670e6
14 changed files with 766 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.coveragerc
View file

@ -381,6 +381,8 @@ omit =
homeassistant/components/plum_lightpad/*
homeassistant/components/point/*
homeassistant/components/prometheus/*
homeassistant/components/ps4/__init__.py
homeassistant/components/ps4/media_player.py
homeassistant/components/qwikswitch/*
homeassistant/components/rachio/*
homeassistant/components/rainbird/*

32
homeassistant/components/ps4/.translations/en.json Normal file
View file

@ -0,0 +1,32 @@
{
"config": {
"title": "PlayStation 4",
"step": {
"creds": {
"title": "PlayStation 4",
"description": "Credentials needed. Press 'Submit' and then in the PS4 2nd Screen App, refresh devices and select the 'Home-Assistant' device to continue."
},
"link": {
"title": "PlayStation 4",
"description": "Enter your PlayStation 4 information. For 'PIN', navigate to 'Settings' on your PlayStation 4 console. Then navigate to 'Mobile App Connection Settings' and select 'Add Device'. Enter the PIN that is displayed.",
"data": {
"region": "Region",
"name": "Name",
"code": "PIN",
"ip_address": "IP Address"
}
}
},
"error": {
"not_ready": "PlayStation 4 is not on or connected to network.",
"login_failed": "Failed to pair to PlayStation 4. Verify PIN is correct."
},
"abort": {
"credential_error": "Error fetching credentials.",
"no_devices_found": "No PlayStation 4 devices found on the network.",
"devices_configured": "All devices found are already configured.",
"port_987_bind_error": "Could not bind to UDP port 987. Port in use or additional configuration required. See component documentation at: https://home-assistant.io/components/media_player.ps4/",
"port_997_bind_error": "Could not bind to TCP port 997. Port in use or additional configuration required. See component documentation at: https://home-assistant.io/components/media_player.ps4/"
}
}
}

33
homeassistant/components/ps4/__init__.py Normal file
View file

@ -0,0 +1,33 @@
"""
Support for PlayStation 4 consoles.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/ps4/
"""
import logging
from homeassistant.components.ps4.config_flow import PlayStation4FlowHandler # noqa: pylint: disable=unused-import
from homeassistant.components.ps4.const import DOMAIN # noqa: pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
REQUIREMENTS = ['pyps4-homeassistant==0.3.0']
async def async_setup(hass, config):
"""Set up the PS4 Component."""
return True
async def async_setup_entry(hass, config_entry):
"""Set up PS4 from a config entry."""
hass.async_create_task(hass.config_entries.async_forward_entry_setup(
config_entry, 'media_player'))
return True
async def async_unload_entry(hass, entry):
"""Unload a PS4 config entry."""
await hass.config_entries.async_forward_entry_unload(
entry, 'media_player')
return True

123
homeassistant/components/ps4/config_flow.py Normal file
View file

@ -0,0 +1,123 @@
"""Config Flow for PlayStation 4."""
from collections import OrderedDict
import logging
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.ps4.const import (
DEFAULT_NAME, DEFAULT_REGION, DOMAIN, REGIONS)
from homeassistant.const import (
CONF_CODE, CONF_HOST, CONF_IP_ADDRESS, CONF_NAME, CONF_REGION, CONF_TOKEN)
_LOGGER = logging.getLogger(__name__)
UDP_PORT = 987
TCP_PORT = 997
PORT_MSG = {UDP_PORT: 'port_987_bind_error', TCP_PORT: 'port_997_bind_error'}
@config_entries.HANDLERS.register(DOMAIN)
class PlayStation4FlowHandler(config_entries.ConfigFlow):
"""Handle a PlayStation 4 config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
"""Initialize the config flow."""
from pyps4_homeassistant import Helper
self.helper = Helper()
self.creds = None
self.name = None
self.host = None
self.region = None
self.pin = None
async def async_step_user(self, user_input=None):
"""Handle a user config flow."""
# Abort if device is configured.
if self.hass.config_entries.async_entries(DOMAIN):
return self.async_abort(reason='devices_configured')
# Check if able to bind to ports: UDP 987, TCP 997.
ports = PORT_MSG.keys()
failed = await self.hass.async_add_executor_job(
self.helper.port_bind, ports)
if failed in ports:
reason = PORT_MSG[failed]
return self.async_abort(reason=reason)
return await self.async_step_creds()
async def async_step_creds(self, user_input=None):
"""Return PS4 credentials from 2nd Screen App."""
if user_input is not None:
self.creds = await self.hass.async_add_executor_job(
self.helper.get_creds)
if self.creds is not None:
return await self.async_step_link()
return self.async_abort(reason='credential_error')
return self.async_show_form(
step_id='creds')
async def async_step_link(self, user_input=None):
"""Prompt user input. Create or edit entry."""
errors = {}
# Search for device.
devices = await self.hass.async_add_executor_job(
self.helper.has_devices)
# Abort if can't find device.
if not devices:
return self.async_abort(reason='no_devices_found')
device_list = [
device['host-ip'] for device in devices]
# Login to PS4 with user data.
if user_input is not None:
self.region = user_input[CONF_REGION]
self.name = user_input[CONF_NAME]
self.pin = user_input[CONF_CODE]
self.host = user_input[CONF_IP_ADDRESS]
is_ready, is_login = await self.hass.async_add_executor_job(
self.helper.link, self.host, self.creds, self.pin)
if is_ready is False:
errors['base'] = 'not_ready'
elif is_login is False:
errors['base'] = 'login_failed'
else:
device = {
CONF_HOST: self.host,
CONF_NAME: self.name,
CONF_REGION: self.region
}
# Create entry.
return self.async_create_entry(
title='PlayStation 4',
data={
CONF_TOKEN: self.creds,
'devices': [device],
},
)
# Show User Input form.
link_schema = OrderedDict()
link_schema[vol.Required(CONF_IP_ADDRESS)] = vol.In(list(device_list))
link_schema[vol.Required(
CONF_REGION, default=DEFAULT_REGION)] = vol.In(list(REGIONS))
link_schema[vol.Required(CONF_CODE)] = str
link_schema[vol.Required(CONF_NAME, default=DEFAULT_NAME)] = str
return self.async_show_form(
step_id='link',
data_schema=vol.Schema(link_schema),
errors=errors,
)

5
homeassistant/components/ps4/const.py Normal file
View file

@ -0,0 +1,5 @@
"""Constants for PlayStation 4."""
DEFAULT_NAME = "PlayStation 4"
DEFAULT_REGION = "R1"
DOMAIN = 'ps4'
REGIONS = ('R1', 'R2', 'R3', 'R4', 'R5')

372
homeassistant/components/ps4/media_player.py Normal file
View file

@ -0,0 +1,372 @@
"""
Support for PlayStation 4 consoles.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/media_player.ps4/
"""
from datetime import timedelta
import logging
import socket
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
import homeassistant.util as util
from homeassistant.components.media_player import (
MediaPlayerDevice, ENTITY_IMAGE_URL)
from homeassistant.components.media_player.const import (
MEDIA_TYPE_MUSIC, SUPPORT_SELECT_SOURCE,
SUPPORT_STOP, SUPPORT_TURN_OFF, SUPPORT_TURN_ON,
)
from homeassistant.components.ps4.const import DOMAIN as PS4_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_COMMAND, CONF_HOST, CONF_NAME, CONF_REGION,
CONF_TOKEN, STATE_IDLE, STATE_OFF, STATE_PLAYING,
)
from homeassistant.util.json import load_json, save_json
DEPENDENCIES = ['ps4']
_LOGGER = logging.getLogger(__name__)
SUPPORT_PS4 = SUPPORT_TURN_OFF | SUPPORT_TURN_ON | \
SUPPORT_STOP | SUPPORT_SELECT_SOURCE
PS4_DATA = 'ps4_data'
ICON = 'mdi:playstation'
GAMES_FILE = '.ps4-games.json'
MEDIA_IMAGE_DEFAULT = None
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=5)
MIN_TIME_BETWEEN_FORCED_SCANS = timedelta(seconds=10)
COMMANDS = (
'up',
'down',
'right',
'left',
'enter',
'back',
'option',
'ps',
)
SERVICE_COMMAND = 'send_command'
PS4_COMMAND_SCHEMA = vol.Schema({
vol.Required(ATTR_ENTITY_ID): cv.entity_ids,
vol.Required(ATTR_COMMAND): vol.In(list(COMMANDS))
})
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up PS4 from a config entry."""
config = config_entry
def add_entities(entities, update_before_add=False):
"""Sync version of async add devices."""
hass.add_job(async_add_entities, entities, update_before_add)
await hass.async_add_executor_job(
setup_platform, hass, config,
add_entities, None)
async def async_service_handle(hass):
"""Handle for services."""
def service_command(call):
entity_ids = call.data[ATTR_ENTITY_ID]
command = call.data[ATTR_COMMAND]
for device in hass.data[PS4_DATA].devices:
if device.entity_id in entity_ids:
device.send_command(command)
hass.services.async_register(
PS4_DOMAIN, SERVICE_COMMAND, service_command,
schema=PS4_COMMAND_SCHEMA)
await async_service_handle(hass)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up PS4 Platform."""
import pyps4_homeassistant as pyps4
hass.data[PS4_DATA] = PS4Data()
games_file = hass.config.path(GAMES_FILE)
creds = config.data[CONF_TOKEN]
device_list = []
for device in config.data['devices']:
host = device[CONF_HOST]
region = device[CONF_REGION]
name = device[CONF_NAME]
ps4 = pyps4.Ps4(host, creds)
device_list.append(PS4Device(
name, host, region, ps4, games_file))
add_entities(device_list, True)
class PS4Data():
"""Init Data Class."""
def __init__(self):
"""Init Class."""
self.devices = []
class PS4Device(MediaPlayerDevice):
"""Representation of a PS4."""
def __init__(self, name, host, region, ps4, games_file):
"""Initialize the ps4 device."""
self._ps4 = ps4
self._host = host
self._name = name
self._region = region
self._state = None
self._games_filename = games_file
self._media_content_id = None
self._media_title = None
self._media_image = None
self._source = None
self._games = {}
self._source_list = []
self._retry = 0
self._info = None
self._unique_id = None
async def async_added_to_hass(self):
"""Subscribe PS4 events."""
self.hass.data[PS4_DATA].devices.append(self)
@util.Throttle(MIN_TIME_BETWEEN_SCANS, MIN_TIME_BETWEEN_FORCED_SCANS)
def update(self):
"""Retrieve the latest data."""
try:
status = self._ps4.get_status()
if self._info is None:
self.get_device_info(status)
self._games = self.load_games()
if self._games is not None:
self._source_list = list(sorted(self._games.values()))
except socket.timeout:
status = None
if status is not None:
self._retry = 0
if status.get('status') == 'Ok':
title_id = status.get('running-app-titleid')
name = status.get('running-app-name')
if title_id and name is not None:
self._state = STATE_PLAYING
if self._media_content_id != title_id:
self._media_content_id = title_id
self.get_title_data(title_id, name)
else:
self.idle()
else:
self.state_off()
elif self._retry > 5:
self.state_unknown()
else:
self._retry += 1
def idle(self):
"""Set states for state idle."""
self.reset_title()
self._state = STATE_IDLE
def state_off(self):
"""Set states for state off."""
self.reset_title()
self._state = STATE_OFF
def state_unknown(self):
"""Set states for state unknown."""
self.reset_title()
self._state = None
_LOGGER.warning("PS4 could not be reached")
self._retry = 0
def reset_title(self):
"""Update if there is no title."""
self._media_title = None
self._media_content_id = None
self._source = None
def get_title_data(self, title_id, name):
"""Get PS Store Data."""
app_name = None
art = None
try:
app_name, art = self._ps4.get_ps_store_data(
name, title_id, self._region)
except TypeError:
_LOGGER.error(
"Could not find data in region: %s for PS ID: %s",
self._region, title_id)
finally:
self._media_title = app_name or name
self._source = self._media_title
self._media_image = art
self.update_list()
def update_list(self):
"""Update Game List, Correct data if different."""
if self._media_content_id in self._games:
store = self._games[self._media_content_id]
if store != self._media_title:
self._games.pop(self._media_content_id)
if self._media_content_id not in self._games:
self.add_games(self._media_content_id, self._media_title)
self._games = self.load_games()
self._source_list = list(sorted(self._games.values()))
def load_games(self):
"""Load games for sources."""
g_file = self._games_filename
try:
games = load_json(g_file)
# If file does not exist, create empty file.
except FileNotFoundError:
games = {}
self.save_games(games)
return games
def save_games(self, games):
"""Save games to file."""
g_file = self._games_filename
try:
save_json(g_file, games)
except OSError as error:
_LOGGER.error("Could not save game list, %s", error)
# Retry loading file
if games is None:
self.load_games()
def add_games(self, title_id, app_name):
"""Add games to list."""
games = self._games
if title_id is not None and title_id not in games:
game = {title_id: app_name}
games.update(game)
self.save_games(games)
def get_device_info(self, status):
"""Return device info for registry."""
_sw_version = status['system-version']
_sw_version = _sw_version[1:4]
sw_version = "{}.{}".format(_sw_version[0], _sw_version[1:])
self._info = {
'name': status['host-name'],
'model': 'PlayStation 4',
'identifiers': {
(PS4_DOMAIN, status['host-id'])
},
'manufacturer': 'Sony Interactive Entertainment Inc.',
'sw_version': sw_version
}
self._unique_id = status['host-id']
@property
def device_info(self):
"""Return information about the device."""
return self._info
@property
def unique_id(self):
"""Return Unique ID for entity."""
return self._unique_id
@property
def entity_picture(self):
"""Return picture."""
if self._state == STATE_PLAYING and self._media_content_id is not None:
image_hash = self.media_image_hash
if image_hash is not None:
return ENTITY_IMAGE_URL.format(
self.entity_id, self.access_token, image_hash)
return MEDIA_IMAGE_DEFAULT
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def state(self):
"""Return the state of the device."""
return self._state
@property
def icon(self):
"""Icon."""
return ICON
@property
def media_content_id(self):
"""Content ID of current playing media."""
return self._media_content_id
@property
def media_content_type(self):
"""Content type of current playing media."""
return MEDIA_TYPE_MUSIC
@property
def media_image_url(self):
"""Image url of current playing media."""
if self._media_content_id is None:
return MEDIA_IMAGE_DEFAULT
return self._media_image
@property
def media_title(self):
"""Title of current playing media."""
return self._media_title
@property
def supported_features(self):
"""Media player features that are supported."""
return SUPPORT_PS4
@property
def source(self):
"""Return the current input source."""
return self._source
@property
def source_list(self):
"""List of available input sources."""
return self._source_list
def turn_off(self):
"""Turn off media player."""
self._ps4.standby()
def turn_on(self):
"""Turn on the media player."""
self._ps4.wakeup()
def media_pause(self):
"""Send keypress ps to return to menu."""
self._ps4.remote_control('ps')
def media_stop(self):
"""Send keypress ps to return to menu."""
self._ps4.remote_control('ps')
def select_source(self, source):
"""Select input source."""
for title_id, game in self._games.items():
if source == game:
_LOGGER.debug(
"Starting PS4 game %s (%s) using source %s",
game, title_id, source)
self._ps4.start_title(
title_id, running_id=self._media_content_id)
return
def send_command(self, command):
"""Send Button Command."""
self._ps4.remote_control(command)

9
homeassistant/components/ps4/services.yaml Normal file
View file

@ -0,0 +1,9 @@
send_command:
description: Emulate button press for PlayStation 4.
fields:
entity_id:
description: Name(s) of entities to send command.
example: 'media_player.playstation_4'
command:
description: Button to press.
example: 'ps'

32
homeassistant/components/ps4/strings.json Normal file
View file

@ -0,0 +1,32 @@
{
"config": {
"title": "PlayStation 4",
"step": {
"creds": {
"title": "PlayStation 4",
"description": "Credentials needed. Press 'Submit' and then in the PS4 2nd Screen App, refresh devices and select the 'Home-Assistant' device to continue."
},
"link": {
"title": "PlayStation 4",
"description": "Enter your PlayStation 4 information. For 'PIN', navigate to 'Settings' on your PlayStation 4 console. Then navigate to 'Mobile App Connection Settings' and select 'Add Device'. Enter the PIN that is displayed.",
"data": {
"region": "Region",
"name": "Name",
"code": "PIN",
"ip_address": "IP Address"
}
}
},
"error": {
"not_ready": "PlayStation 4 is not on or connected to network.",
"login_failed": "Failed to pair to PlayStation 4. Verify PIN is correct."
},
"abort": {
"credential_error": "Error fetching credentials.",
"no_devices_found": "No PlayStation 4 devices found on the network.",
"devices_configured": "All devices found are already configured.",
"port_987_bind_error": "Could not bind to port 987.",
"port_997_bind_error": "Could not bind to port 997."
}
}
}

1
homeassistant/config_entries.py
View file

@ -165,6 +165,7 @@ FLOWS = [
'openuv',
'owntracks',
'point',
'ps4',
'rainmachine',
'simplisafe',
'smartthings',

3
requirements_all.txt
View file

@ -1207,6 +1207,9 @@ pypoint==1.1.1
# homeassistant.components.sensor.pollen
pypollencom==2.2.2
# homeassistant.components.ps4
pyps4-homeassistant==0.3.0
# homeassistant.components.qwikswitch
pyqwikswitch==0.8

3
requirements_test_all.txt
View file

@ -210,6 +210,9 @@ pyopenuv==1.0.4
# homeassistant.components.sensor.otp
pyotp==2.2.6
# homeassistant.components.ps4
pyps4-homeassistant==0.3.0
# homeassistant.components.qwikswitch
pyqwikswitch==0.8

1
script/gen_requirements_all.py
View file

@ -90,6 +90,7 @@ TEST_REQUIREMENTS = (
'pynx584',
'pyopenuv',
'pyotp',
'pyps4-homeassistant',
'pysmartapp',
'pysmartthings',
'pysonos',

1
tests/components/ps4/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Tests for the PlayStation 4 component."""

149
tests/components/ps4/test_config_flow.py Normal file
View file

@ -0,0 +1,149 @@
"""Define tests for the PlayStation 4 config flow."""
from unittest.mock import patch
from homeassistant import data_entry_flow
from homeassistant.components import ps4
from homeassistant.components.ps4.const import (
DEFAULT_NAME, DEFAULT_REGION)
from homeassistant.const import (
CONF_CODE, CONF_HOST, CONF_IP_ADDRESS, CONF_NAME, CONF_REGION, CONF_TOKEN)
from tests.common import MockConfigEntry
MOCK_TITLE = 'PlayStation 4'
MOCK_CODE = '12345678'
MOCK_CREDS = '000aa000'
MOCK_HOST = '192.0.0.0'
MOCK_DEVICE = {
CONF_HOST: MOCK_HOST,
CONF_NAME: DEFAULT_NAME,
CONF_REGION: DEFAULT_REGION
}
MOCK_CONFIG = {
CONF_IP_ADDRESS: MOCK_HOST,
CONF_NAME: DEFAULT_NAME,
CONF_REGION: DEFAULT_REGION,
CONF_CODE: MOCK_CODE
}
MOCK_DATA = {
CONF_TOKEN: MOCK_CREDS,
'devices': MOCK_DEVICE
}
MOCK_UDP_PORT = int(987)
MOCK_TCP_PORT = int(997)
async def test_full_flow_implementation(hass):
"""Test registering an implementation and flow works."""
flow = ps4.PlayStation4FlowHandler()
flow.hass = hass
# User Step Started, results in Step Creds
with patch('pyps4_homeassistant.Helper.port_bind',
return_value=None):
result = await flow.async_step_user()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'creds'
# Step Creds results with form in Step Link.
with patch('pyps4_homeassistant.Helper.get_creds',
return_value=MOCK_CREDS), \
patch('pyps4_homeassistant.Helper.has_devices',
return_value=[{'host-ip': MOCK_HOST}]):
result = await flow.async_step_creds({})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
# User Input results in created entry.
with patch('pyps4_homeassistant.Helper.link',
return_value=(True, True)), \
patch('pyps4_homeassistant.Helper.has_devices',
return_value=[{'host-ip': MOCK_HOST}]):
result = await flow.async_step_link(MOCK_CONFIG)
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['data'][CONF_TOKEN] == MOCK_CREDS
assert result['data']['devices'] == [MOCK_DEVICE]
assert result['title'] == MOCK_TITLE
async def test_port_bind_abort(hass):
"""Test that flow aborted when cannot bind to ports 987, 997."""
flow = ps4.PlayStation4FlowHandler()
flow.hass = hass
with patch('pyps4_homeassistant.Helper.port_bind',
return_value=MOCK_UDP_PORT):
reason = 'port_987_bind_error'
result = await flow.async_step_user(user_input=None)
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == reason
with patch('pyps4_homeassistant.Helper.port_bind',
return_value=MOCK_TCP_PORT):
reason = 'port_997_bind_error'
result = await flow.async_step_user(user_input=None)
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == reason
async def test_duplicate_abort(hass):
"""Test that Flow aborts when already configured."""
MockConfigEntry(domain=ps4.DOMAIN, data=MOCK_DATA).add_to_hass(hass)
flow = ps4.PlayStation4FlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=None)
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'devices_configured'
async def test_no_devices_found_abort(hass):
"""Test that failure to find devices aborts flow."""
flow = ps4.PlayStation4FlowHandler()
flow.hass = hass
with patch('pyps4_homeassistant.Helper.has_devices', return_value=None):
result = await flow.async_step_link(MOCK_CONFIG)
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'no_devices_found'
async def test_credential_abort(hass):
"""Test that failure to get credentials aborts flow."""
flow = ps4.PlayStation4FlowHandler()
flow.hass = hass
with patch('pyps4_homeassistant.Helper.get_creds', return_value=None):
result = await flow.async_step_creds({})
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'credential_error'
async def test_invalid_pin_error(hass):
"""Test that invalid pin throws an error."""
flow = ps4.PlayStation4FlowHandler()
flow.hass = hass
with patch('pyps4_homeassistant.Helper.link',
return_value=(True, False)), \
patch('pyps4_homeassistant.Helper.has_devices',
return_value=[{'host-ip': MOCK_HOST}]):
result = await flow.async_step_link(MOCK_CONFIG)
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
assert result['errors'] == {'base': 'login_failed'}
async def test_device_connection_error(hass):
"""Test that device not connected or on throws an error."""
flow = ps4.PlayStation4FlowHandler()
flow.hass = hass
with patch('pyps4_homeassistant.Helper.link',
return_value=(False, True)), \
patch('pyps4_homeassistant.Helper.has_devices',
return_value=[{'host-ip': MOCK_HOST}]):
result = await flow.async_step_link(MOCK_CONFIG)
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
assert result['errors'] == {'base': 'not_ready'}