jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add config flow support to songpal integration (#34714)

Browse Source 
* Add config flow to songpal

* Add config flow to songpal

* Add songpal to migrated service in discovery

* Improve songpal/set_sound_setting service

* Remove songpal config flow from .coveragerc omit

* Bump python-songpal to 0.12 and fix exception handling

* Revert "Improve songpal/set_sound_setting service"

This reverts commit 9be076ab52e21f268322572c36709a17d41db771.

* Code style fix

* Add connections to device_info

* Fix pylint

* Ignore braava tv

* Fix test warning

* Add @shenxn as codeowner

* Remove model from configuration data

* Get name from device in user step

* Add unload entry support

* Delete translations as it will get generated as part of CI

* Code cleanup

* Fix typo

* Remove _show_setup_form

* Change configuration from media_player to songpal

Co-authored-by: Raman Gupta <7243222+raman325@users.noreply.github.com>
This commit is contained in:
Xiaonan Shen 2020-05-06 15:52:33 -07:00 committed by GitHub
parent 541b666a86
commit 33077f0cdd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 547 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.coveragerc
View File

@ -679,7 +679,8 @@ omit =
homeassistant/components/somfy/*
homeassistant/components/somfy_mylink/*
homeassistant/components/sonarr/sensor.py
homeassistant/components/songpal/*
homeassistant/components/songpal/__init__.py
homeassistant/components/songpal/media_player.py
homeassistant/components/sonos/*
homeassistant/components/sony_projector/switch.py
homeassistant/components/spc/*

2
CODEOWNERS
View File

@ -362,7 +362,7 @@ homeassistant/components/solax/* @squishykid
homeassistant/components/soma/* @ratsept
homeassistant/components/somfy/* @tetienne
homeassistant/components/sonarr/* @ctalkington
homeassistant/components/songpal/* @rytilahti
homeassistant/components/songpal/* @rytilahti @shenxn
homeassistant/components/sonos/* @amelchio
homeassistant/components/spaceapi/* @fabaff
homeassistant/components/speedtestdotnet/* @rohankapoorcom

2
homeassistant/components/discovery/__init__.py
View File

@ -71,7 +71,6 @@ SERVICE_HANDLERS = {
"openhome": ("media_player", "openhome"),
"bose_soundtouch": ("media_player", "soundtouch"),
"bluesound": ("media_player", "bluesound"),
"songpal": ("media_player", "songpal"),
"kodi": ("media_player", "kodi"),
"volumio": ("media_player", "volumio"),
"lg_smart_device": ("media_player", "lg_soundbar"),
@ -91,6 +90,7 @@ MIGRATED_SERVICE_HANDLERS = [
"ikea_tradfri",
"philips_hue",
"sonos",
"songpal",
SERVICE_WEMO,
]

49
homeassistant/components/songpal/__init__.py
View File

@ -1 +1,50 @@
"""The songpal component."""
from collections import OrderedDict
import logging
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_NAME
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import HomeAssistantType
from .const import CONF_ENDPOINT, DOMAIN
_LOGGER = logging.getLogger(__name__)
SONGPAL_CONFIG_SCHEMA = vol.Schema(
{vol.Optional(CONF_NAME): cv.string, vol.Required(CONF_ENDPOINT): cv.string}
)
CONFIG_SCHEMA = vol.Schema(
{vol.Optional(DOMAIN): vol.All(cv.ensure_list, [SONGPAL_CONFIG_SCHEMA])},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass: HomeAssistantType, config: OrderedDict) -> bool:
"""Set up songpal environment."""
conf = config.get(DOMAIN)
if conf is None:
return True
for config_entry in conf:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config_entry,
),
)
return True
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Set up songpal media player."""
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "media_player")
)
return True
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Unload songpal media player."""
return await hass.config_entries.async_forward_entry_unload(entry, "media_player")

153
homeassistant/components/songpal/config_flow.py Normal file
View File

@ -0,0 +1,153 @@
"""Config flow to configure songpal component."""
import logging
from typing import Optional
from urllib.parse import urlparse
from songpal import Device, SongpalException
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.const import CONF_HOST, CONF_NAME
from .const import CONF_ENDPOINT, DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
class SongpalConfig:
"""Device Configuration."""
def __init__(self, name, host, endpoint):
"""Initialize Configuration."""
self.name = name
self.host = host
self.endpoint = endpoint
class SongpalConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Songpal configuration flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
def __init__(self):
"""Initialize the flow."""
self.conf: Optional[SongpalConfig] = None
async def async_step_user(self, user_input=None):
"""Handle a flow initiated by the user."""
if user_input is None:
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_ENDPOINT): str}),
)
# Validate input
endpoint = user_input[CONF_ENDPOINT]
parsed_url = urlparse(endpoint)
# Try to connect and get device name
try:
device = Device(endpoint)
await device.get_supported_methods()
interface_info = await device.get_interface_information()
name = interface_info.modelName
except SongpalException as ex:
_LOGGER.debug("Connection failed: %s", ex)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(
CONF_ENDPOINT, default=user_input.get(CONF_ENDPOINT, "")
): str,
}
),
errors={"base": "connection"},
)
self.conf = SongpalConfig(name, parsed_url.hostname, endpoint)
return await self.async_step_init(user_input)
async def async_step_init(self, user_input=None):
"""Handle a flow start."""
# Check if already configured
if self._endpoint_already_configured():
return self.async_abort(reason="already_configured")
if user_input is None:
return self.async_show_form(
step_id="init",
description_placeholders={
CONF_NAME: self.conf.name,
CONF_HOST: self.conf.host,
},
)
await self.async_set_unique_id(self.conf.endpoint)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=self.conf.name,
data={CONF_NAME: self.conf.name, CONF_ENDPOINT: self.conf.endpoint},
)
async def async_step_ssdp(self, discovery_info):
"""Handle a discovered Songpal device."""
await self.async_set_unique_id(discovery_info[ssdp.ATTR_UPNP_UDN])
self._abort_if_unique_id_configured()
_LOGGER.debug("Discovered: %s", discovery_info)
friendly_name = discovery_info[ssdp.ATTR_UPNP_FRIENDLY_NAME]
parsed_url = urlparse(discovery_info[ssdp.ATTR_SSDP_LOCATION])
scalarweb_info = discovery_info["X_ScalarWebAPI_DeviceInfo"]
endpoint = scalarweb_info["X_ScalarWebAPI_BaseURL"]
service_types = scalarweb_info["X_ScalarWebAPI_ServiceList"][
"X_ScalarWebAPI_ServiceType"
]
# Ignore Bravia TVs
if "videoScreen" in service_types:
return self.async_abort(reason="not_songpal_device")
# pylint: disable=no-member
self.context["title_placeholders"] = {
CONF_NAME: friendly_name,
CONF_HOST: parsed_url.hostname,
}
self.conf = SongpalConfig(friendly_name, parsed_url.hostname, endpoint)
return await self.async_step_init()
async def async_step_import(self, user_input=None):
"""Import a config entry."""
name = user_input.get(CONF_NAME)
endpoint = user_input.get(CONF_ENDPOINT)
parsed_url = urlparse(endpoint)
# Try to connect to test the endpoint
try:
device = Device(endpoint)
await device.get_supported_methods()
# Get name
if name is None:
interface_info = await device.get_interface_information()
name = interface_info.modelName
except SongpalException as ex:
_LOGGER.error("Import from yaml configuration failed: %s", ex)
return self.async_abort(reason="connection")
self.conf = SongpalConfig(name, parsed_url.hostname, endpoint)
return await self.async_step_init(user_input)
def _endpoint_already_configured(self):
"""See if we already have an endpoint matching user input configured."""
existing_endpoints = [
entry.data[CONF_ENDPOINT] for entry in self._async_current_entries()
]
return self.conf.endpoint in existing_endpoints

2
homeassistant/components/songpal/const.py
View File

@ -1,3 +1,5 @@
"""Constants for the Songpal component."""
DOMAIN = "songpal"
SET_SOUND_SETTING = "set_sound_setting"
CONF_ENDPOINT = "endpoint"

11
homeassistant/components/songpal/manifest.json
View File

@ -1,7 +1,14 @@
{
"domain": "songpal",
"name": "Sony Songpal",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/songpal",
"requirements": ["python-songpal==0.11.2"],
"codeowners": ["@rytilahti"]
"requirements": ["python-songpal==0.12"],
"codeowners": ["@rytilahti", "@shenxn"],
"ssdp": [
{
"st": "urn:schemas-sony-com:service:ScalarWebAPI:1",
"manufacturer": "Sony Corporation"
}
]
}

72
homeassistant/components/songpal/media_player.py
View File

@ -13,7 +13,7 @@ from songpal import (
)
import voluptuous as vol
from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerEntity
from homeassistant.components.media_player import MediaPlayerEntity
from homeassistant.components.media_player.const import (
SUPPORT_SELECT_SOURCE,
SUPPORT_TURN_OFF,
@ -22,6 +22,7 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_SET,
SUPPORT_VOLUME_STEP,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_NAME,
@ -30,19 +31,16 @@ from homeassistant.const import (
STATE_ON,
)
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN, SET_SOUND_SETTING
from .const import CONF_ENDPOINT, DOMAIN, SET_SOUND_SETTING
_LOGGER = logging.getLogger(__name__)
CONF_ENDPOINT = "endpoint"
PARAM_NAME = "name"
PARAM_VALUE = "value"
PLATFORM = "songpal"
SUPPORT_SONGPAL = (
SUPPORT_VOLUME_SET
| SUPPORT_VOLUME_STEP
@ -52,10 +50,6 @@ SUPPORT_SONGPAL = (
| SUPPORT_TURN_OFF
)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{vol.Optional(CONF_NAME): cv.string, vol.Required(CONF_ENDPOINT): cv.string}
)
SET_SOUND_SCHEMA = vol.Schema(
{
vol.Optional(ATTR_ENTITY_ID): cv.entity_id,
@ -65,33 +59,37 @@ SET_SOUND_SCHEMA = vol.Schema(
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the Songpal platform."""
if PLATFORM not in hass.data:
hass.data[PLATFORM] = {}
async def async_setup_platform(
hass: HomeAssistantType, config: dict, async_add_entities, discovery_info=None
) -> None:
"""Set up from legacy configuration file. Obsolete."""
_LOGGER.error(
"Configuring Songpal through media_player platform is no longer supported. Convert to songpal platform or UI configuration."
)
if discovery_info is not None:
name = discovery_info["name"]
endpoint = discovery_info["properties"]["endpoint"]
_LOGGER.debug("Got autodiscovered %s - endpoint: %s", name, endpoint)
device = SongpalDevice(name, endpoint)
else:
name = config.get(CONF_NAME)
endpoint = config.get(CONF_ENDPOINT)
device = SongpalDevice(name, endpoint, poll=False)
async def async_setup_entry(
hass: HomeAssistantType, config_entry: ConfigEntry, async_add_entities
) -> None:
"""Set up songpal media player."""
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}
if endpoint in hass.data[PLATFORM]:
name = config_entry.data[CONF_NAME]
endpoint = config_entry.data[CONF_ENDPOINT]
if endpoint in hass.data[DOMAIN]:
_LOGGER.debug("The endpoint exists already, skipping setup.")
return
device = SongpalDevice(name, endpoint)
try:
await device.initialize()
except SongpalException as ex:
_LOGGER.error("Unable to get methods from songpal: %s", ex)
raise PlatformNotReady
hass.data[PLATFORM][endpoint] = device
hass.data[DOMAIN][endpoint] = device
async_add_entities([device], True)
@ -102,7 +100,7 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
key: value for key, value in service.data.items() if key != ATTR_ENTITY_ID
}
for device in hass.data[PLATFORM].values():
for device in hass.data[DOMAIN].values():
if device.entity_id == entity_id or entity_id is None:
_LOGGER.debug(
"Calling %s (entity: %s) with params %s", service, entity_id, params
@ -127,6 +125,7 @@ class SongpalDevice(MediaPlayerEntity):
self._poll = poll
self.dev = Device(self._endpoint)
self._sysinfo = None
self._model = None
self._state = False
self._available = False
@ -150,6 +149,13 @@ class SongpalDevice(MediaPlayerEntity):
"""Initialize the device."""
await self.dev.get_supported_methods()
self._sysinfo = await self.dev.get_system_info()
interface_info = await self.dev.get_interface_information()
self._model = interface_info.modelName
async def async_will_remove_from_hass(self):
"""Run when entity will be removed from hass."""
self.hass.data[DOMAIN].pop(self._endpoint)
await self.dev.stop_listen_notifications()
async def async_activate_websocket(self):
"""Activate websocket for listening if wanted."""
@ -221,6 +227,18 @@ class SongpalDevice(MediaPlayerEntity):
"""Return a unique ID."""
return self._sysinfo.macAddr
@property
def device_info(self):
"""Return the device info."""
return {
"connections": {(dr.CONNECTION_NETWORK_MAC, self._sysinfo.macAddr)},
"identifiers": {(DOMAIN, self.unique_id)},
"manufacturer": "Sony Corporation",
"name": self.name,
"sw_version": self._sysinfo.version,
"model": self._model,
}
@property
def available(self):
"""Return availability of the device."""

24
homeassistant/components/songpal/strings.json Normal file
View File

@ -0,0 +1,24 @@
{
"config": {
"flow_title": "Sony Songpal {name} ({host})",
"step": {
"user": {
"data": {
"endpoint": "Endpoint"
},
"title": "Sony Songpal"
},
"init": {
"description": "Do you want to set up {name} ({host})?",
"title": "Sony Songpal"
}
},
"error": {
"connection": "Connection error: please check your endpoint"
},
"abort": {
"already_configured": "Device already configured",
"not_songpal_device": "Not a Songpal device"
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -120,6 +120,7 @@ FLOWS = [
"solarlog",
"soma",
"somfy",
"songpal",
"sonos",
"spotify",
"starline",

6
homeassistant/generated/ssdp.py
View File

@ -70,6 +70,12 @@ SSDP = {
"st": "urn:samsung.com:device:RemoteControlReceiver:1"
}
],
"songpal": [
{
"manufacturer": "Sony Corporation",
"st": "urn:schemas-sony-com:service:ScalarWebAPI:1"
}
],
"sonos": [
{
"st": "urn:schemas-upnp-org:device:ZonePlayer:1"

2
requirements_all.txt
View File

@ -1699,7 +1699,7 @@ python-ripple-api==0.0.3
python-sochain-api==0.0.2
# homeassistant.components.songpal
python-songpal==0.11.2
python-songpal==0.12
# homeassistant.components.synology_dsm
python-synology==0.8.0

3
requirements_test_all.txt
View File

@ -677,6 +677,9 @@ python-nest==4.1.0
# homeassistant.components.zwave_mqtt
python-openzwave-mqtt==1.0.1
# homeassistant.components.songpal
python-songpal==0.12
# homeassistant.components.synology_dsm
python-synology==0.8.0

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

@ -0,0 +1 @@
"""Test the songpal integration."""

249
tests/components/songpal/test_config_flow.py Normal file
View File

@ -0,0 +1,249 @@
"""Test the songpal config flow."""
import copy
from asynctest import MagicMock, patch
from songpal import SongpalException
from songpal.containers import InterfaceInfo
from homeassistant.components import ssdp
from homeassistant.components.songpal.const import CONF_ENDPOINT, DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from tests.common import MockConfigEntry
UDN = "uuid:1234"
FRIENDLY_NAME = "friendly name"
HOST = "0.0.0.0"
ENDPOINT = f"http://{HOST}:10000/sony"
MODEL = "model"
SSDP_DATA = {
ssdp.ATTR_UPNP_UDN: UDN,
ssdp.ATTR_UPNP_FRIENDLY_NAME: FRIENDLY_NAME,
ssdp.ATTR_SSDP_LOCATION: f"http://{HOST}:52323/dmr.xml",
"X_ScalarWebAPI_DeviceInfo": {
"X_ScalarWebAPI_BaseURL": ENDPOINT,
"X_ScalarWebAPI_ServiceList": {
"X_ScalarWebAPI_ServiceType": ["guide", "system", "audio", "avContent"],
},
},
}
CONF_DATA = {
CONF_NAME: FRIENDLY_NAME,
CONF_ENDPOINT: ENDPOINT,
}
async def _async_return_value():
pass
def _get_supported_methods(throw_exception):
def get_supported_methods():
if throw_exception:
raise SongpalException("Unable to do POST request: ")
return _async_return_value()
return get_supported_methods
async def _get_interface_information():
return InterfaceInfo(
productName="product name",
modelName=MODEL,
productCategory="product category",
interfaceVersion="interface version",
serverName="server name",
)
def _create_mocked_device(throw_exception=False):
mocked_device = MagicMock()
type(mocked_device).get_supported_methods = MagicMock(
side_effect=_get_supported_methods(throw_exception)
)
type(mocked_device).get_interface_information = MagicMock(
side_effect=_get_interface_information
)
return mocked_device
def _patch_config_flow_device(mocked_device):
return patch(
"homeassistant.components.songpal.config_flow.Device",
return_value=mocked_device,
)
def _flow_next(hass, flow_id):
return next(
flow
for flow in hass.config_entries.flow.async_progress()
if flow["flow_id"] == flow_id
)
async def test_flow_ssdp(hass):
"""Test working ssdp flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_SSDP}, data=SSDP_DATA,
)
assert result["type"] == "form"
assert result["step_id"] == "init"
assert result["description_placeholders"] == {
CONF_NAME: FRIENDLY_NAME,
CONF_HOST: HOST,
}
flow = _flow_next(hass, result["flow_id"])
assert flow["context"]["unique_id"] == UDN
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == FRIENDLY_NAME
assert result["data"] == CONF_DATA
async def test_flow_user(hass):
"""Test working user initialized flow."""
mocked_device = _create_mocked_device()
with _patch_config_flow_device(mocked_device):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] is None
_flow_next(hass, result["flow_id"])
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_ENDPOINT: ENDPOINT},
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == MODEL
assert result["data"] == {
CONF_NAME: MODEL,
CONF_ENDPOINT: ENDPOINT,
}
mocked_device.get_supported_methods.assert_called_once()
mocked_device.get_interface_information.assert_called_once()
async def test_flow_import(hass):
"""Test working import flow."""
mocked_device = _create_mocked_device()
with _patch_config_flow_device(mocked_device):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=CONF_DATA
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == FRIENDLY_NAME
assert result["data"] == CONF_DATA
mocked_device.get_supported_methods.assert_called_once()
mocked_device.get_interface_information.assert_not_called()
def _create_mock_config_entry(hass):
MockConfigEntry(domain=DOMAIN, unique_id="uuid:0000", data=CONF_DATA,).add_to_hass(
hass
)
async def test_ssdp_bravia(hass):
"""Test discovering a bravia TV."""
ssdp_data = copy.deepcopy(SSDP_DATA)
ssdp_data["X_ScalarWebAPI_DeviceInfo"]["X_ScalarWebAPI_ServiceList"][
"X_ScalarWebAPI_ServiceType"
].append("videoScreen")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_SSDP}, data=ssdp_data,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "not_songpal_device"
async def test_sddp_exist(hass):
"""Test discovering existed device."""
_create_mock_config_entry(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_SSDP}, data=SSDP_DATA,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_user_exist(hass):
"""Test user adding existed device."""
mocked_device = _create_mocked_device()
_create_mock_config_entry(hass)
with _patch_config_flow_device(mocked_device):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
mocked_device.get_supported_methods.assert_called_once()
mocked_device.get_interface_information.assert_called_once()
async def test_import_exist(hass):
"""Test importing existed device."""
mocked_device = _create_mocked_device()
_create_mock_config_entry(hass)
with _patch_config_flow_device(mocked_device):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=CONF_DATA
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
mocked_device.get_supported_methods.assert_called_once()
mocked_device.get_interface_information.assert_not_called()
async def test_user_invalid(hass):
"""Test using adding invalid config."""
mocked_device = _create_mocked_device(True)
_create_mock_config_entry(hass)
with _patch_config_flow_device(mocked_device):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONF_DATA
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "connection"}
mocked_device.get_supported_methods.assert_called_once()
mocked_device.get_interface_information.assert_not_called()
async def test_import_invalid(hass):
"""Test importing invalid config."""
mocked_device = _create_mocked_device(True)
_create_mock_config_entry(hass)
with _patch_config_flow_device(mocked_device):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=CONF_DATA
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "connection"
mocked_device.get_supported_methods.assert_called_once()
mocked_device.get_interface_information.assert_not_called()