mirror of https://github.com/home-assistant/core
Add support for arcam fmj receivers (#24621)
* Add arcam_fmj support * Just use use state in player avoid direct client access * Avoid leaking exceptions on invalid data * Fix return value for volume in case of 0 * Mark component as having no coverage * Add new requirement * Add myself as maintainer * Correct linting errors * Use async_create_task instead of async_add_job * Use new style string format instead of concat * Don't call init of base class without init * Annotate callbacks with @callback Otherwise they won't be called in loop * Reduce log level to debug * Use async_timeout instead of wait_for * Bump to version of arcam_fmj supporting 3.5 * Fix extra spaces * Drop somewhat flaky unique_id * Un-blackify ident to satisy pylint * Un-blackify ident to satisy pylint * Move default name calculation to config validation * Add test folder * Drop unused code * Add tests for config flow import
This commit is contained in:
Joakim Plate
Martin Hjelmare
parent
f90fe7e628
commit
ab832cda71
|
|
@ -38,6 +38,8 @@ omit =
|
|||
homeassistant/components/apple_tv/*
|
||||
homeassistant/components/aqualogic/*
|
||||
homeassistant/components/aquostv/media_player.py
|
||||
homeassistant/components/arcam_fmj/media_player.py
|
||||
homeassistant/components/arcam_fmj/__init__.py
|
||||
homeassistant/components/arduino/*
|
||||
homeassistant/components/arest/binary_sensor.py
|
||||
homeassistant/components/arest/sensor.py
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ homeassistant/components/ambiclimate/* @danielhiversen
|
|||
homeassistant/components/ambient_station/* @bachya
|
||||
homeassistant/components/api/* @home-assistant/core
|
||||
homeassistant/components/aprs/* @PhilRW
|
||||
homeassistant/components/arcam_fmj/* @elupus
|
||||
homeassistant/components/arduino/* @fabaff
|
||||
homeassistant/components/arest/* @fabaff
|
||||
homeassistant/components/asuswrt/* @kennedyshead
|
||||
|
|
|
|||
|
|
@ -0,0 +1,8 @@
|
|||
{
|
||||
"config": {
|
||||
"title": "Arcam FMJ",
|
||||
"step": {},
|
||||
"error": {},
|
||||
"abort": {}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,176 @@
|
|||
"""Arcam component."""
|
||||
import logging
|
||||
import asyncio
|
||||
|
||||
import voluptuous as vol
|
||||
import async_timeout
|
||||
from arcam.fmj.client import Client
|
||||
from arcam.fmj import ConnectionFailed
|
||||
|
||||
from homeassistant import config_entries
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.typing import HomeAssistantType, ConfigType
|
||||
from homeassistant.const import (
|
||||
EVENT_HOMEASSISTANT_STOP,
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
CONF_PORT,
|
||||
CONF_SCAN_INTERVAL,
|
||||
CONF_ZONE,
|
||||
SERVICE_TURN_ON,
|
||||
)
|
||||
from .const import (
|
||||
DOMAIN,
|
||||
DOMAIN_DATA_ENTRIES,
|
||||
DOMAIN_DATA_CONFIG,
|
||||
DEFAULT_NAME,
|
||||
DEFAULT_PORT,
|
||||
DEFAULT_SCAN_INTERVAL,
|
||||
SIGNAL_CLIENT_DATA,
|
||||
SIGNAL_CLIENT_STARTED,
|
||||
SIGNAL_CLIENT_STOPPED,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _optional_zone(value):
|
||||
if value:
|
||||
return ZONE_SCHEMA(value)
|
||||
return ZONE_SCHEMA({})
|
||||
|
||||
|
||||
def _zone_name_validator(config):
|
||||
for zone, zone_config in config[CONF_ZONE].items():
|
||||
if CONF_NAME not in zone_config:
|
||||
zone_config[CONF_NAME] = "{} ({}:{}) - {}".format(
|
||||
DEFAULT_NAME,
|
||||
config[CONF_HOST],
|
||||
config[CONF_PORT],
|
||||
zone)
|
||||
return config
|
||||
|
||||
|
||||
ZONE_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Optional(CONF_NAME): cv.string,
|
||||
vol.Optional(SERVICE_TURN_ON): cv.SERVICE_SCHEMA,
|
||||
}
|
||||
)
|
||||
|
||||
DEVICE_SCHEMA = vol.Schema(
|
||||
vol.All({
|
||||
vol.Required(CONF_HOST): cv.string,
|
||||
vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.positive_int,
|
||||
vol.Optional(
|
||||
CONF_ZONE, default={1: _optional_zone(None)}
|
||||
): {vol.In([1, 2]): _optional_zone},
|
||||
vol.Optional(
|
||||
CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL
|
||||
): cv.positive_int,
|
||||
}, _zone_name_validator)
|
||||
)
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{DOMAIN: vol.All(cv.ensure_list, [DEVICE_SCHEMA])}, extra=vol.ALLOW_EXTRA
|
||||
)
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistantType, config: ConfigType):
|
||||
"""Set up the component."""
|
||||
hass.data[DOMAIN_DATA_ENTRIES] = {}
|
||||
hass.data[DOMAIN_DATA_CONFIG] = {}
|
||||
|
||||
for device in config[DOMAIN]:
|
||||
hass.data[DOMAIN_DATA_CONFIG][
|
||||
(device[CONF_HOST], device[CONF_PORT])
|
||||
] = device
|
||||
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data={
|
||||
CONF_HOST: device[CONF_HOST],
|
||||
CONF_PORT: device[CONF_PORT],
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType, entry: config_entries.ConfigEntry
|
||||
):
|
||||
"""Set up an access point from a config entry."""
|
||||
client = Client(entry.data[CONF_HOST], entry.data[CONF_PORT])
|
||||
|
||||
config = hass.data[DOMAIN_DATA_CONFIG].get(
|
||||
(entry.data[CONF_HOST], entry.data[CONF_PORT]),
|
||||
DEVICE_SCHEMA(
|
||||
{
|
||||
CONF_HOST: entry.data[CONF_HOST],
|
||||
CONF_PORT: entry.data[CONF_PORT],
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
hass.data[DOMAIN_DATA_ENTRIES][entry.entry_id] = {
|
||||
"client": client,
|
||||
"config": config,
|
||||
}
|
||||
|
||||
asyncio.ensure_future(
|
||||
_run_client(hass, client, config[CONF_SCAN_INTERVAL])
|
||||
)
|
||||
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(entry, "media_player")
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def _run_client(hass, client, interval):
|
||||
task = asyncio.Task.current_task()
|
||||
run = True
|
||||
|
||||
async def _stop(_):
|
||||
nonlocal run
|
||||
run = False
|
||||
task.cancel()
|
||||
await task
|
||||
|
||||
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _stop)
|
||||
|
||||
def _listen(_):
|
||||
hass.helpers.dispatcher.async_dispatcher_send(
|
||||
SIGNAL_CLIENT_DATA, client.host
|
||||
)
|
||||
|
||||
while run:
|
||||
try:
|
||||
with async_timeout.timeout(interval):
|
||||
await client.start()
|
||||
|
||||
_LOGGER.debug("Client connected %s", client.host)
|
||||
hass.helpers.dispatcher.async_dispatcher_send(
|
||||
SIGNAL_CLIENT_STARTED, client.host
|
||||
)
|
||||
|
||||
try:
|
||||
with client.listen(_listen):
|
||||
await client.process()
|
||||
finally:
|
||||
await client.stop()
|
||||
|
||||
_LOGGER.debug("Client disconnected %s", client.host)
|
||||
hass.helpers.dispatcher.async_dispatcher_send(
|
||||
SIGNAL_CLIENT_STOPPED, client.host
|
||||
)
|
||||
|
||||
except ConnectionFailed:
|
||||
await asyncio.sleep(interval)
|
||||
except asyncio.TimeoutError:
|
||||
continue
|
||||
|
|
@ -0,0 +1,27 @@
|
|||
"""Config flow to configure the Arcam FMJ component."""
|
||||
from operator import itemgetter
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_GETKEY = itemgetter(CONF_HOST, CONF_PORT)
|
||||
|
||||
|
||||
@config_entries.HANDLERS.register(DOMAIN)
|
||||
class ArcamFmjFlowHandler(config_entries.ConfigFlow):
|
||||
"""Handle a SimpliSafe config flow."""
|
||||
|
||||
VERSION = 1
|
||||
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
|
||||
|
||||
async def async_step_import(self, import_config):
|
||||
"""Import a config entry from configuration.yaml."""
|
||||
entries = self.hass.config_entries.async_entries(DOMAIN)
|
||||
import_key = _GETKEY(import_config)
|
||||
for entry in entries:
|
||||
if _GETKEY(entry.data) == import_key:
|
||||
return self.async_abort(reason="already_setup")
|
||||
|
||||
return self.async_create_entry(title="Arcam FMJ", data=import_config)
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
"""Constants used for arcam."""
|
||||
DOMAIN = "arcam_fmj"
|
||||
|
||||
SIGNAL_CLIENT_STARTED = "arcam.client_started"
|
||||
SIGNAL_CLIENT_STOPPED = "arcam.client_stopped"
|
||||
SIGNAL_CLIENT_DATA = "arcam.client_data"
|
||||
|
||||
DEFAULT_PORT = 50000
|
||||
DEFAULT_NAME = "Arcam FMJ"
|
||||
DEFAULT_SCAN_INTERVAL = 5
|
||||
|
||||
DOMAIN_DATA_ENTRIES = "{}.entries".format(DOMAIN)
|
||||
DOMAIN_DATA_CONFIG = "{}.config".format(DOMAIN)
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
{
|
||||
"domain": "arcam_fmj",
|
||||
"name": "Arcam FMJ Receiver control",
|
||||
"config_flow": false,
|
||||
"documentation": "https://www.home-assistant.io/components/arcam_fmj",
|
||||
"requirements": [
|
||||
"arcam-fmj==0.4.3"
|
||||
],
|
||||
"dependencies": [],
|
||||
"codeowners": [
|
||||
"@elupus"
|
||||
]
|
||||
}
|
||||
|
|
@ -0,0 +1,342 @@
|
|||
"""Arcam media player."""
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from arcam.fmj import (
|
||||
DecodeMode2CH,
|
||||
DecodeModeMCH,
|
||||
IncomingAudioFormat,
|
||||
SourceCodes,
|
||||
)
|
||||
from arcam.fmj.state import State
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.components.media_player import MediaPlayerDevice
|
||||
from homeassistant.components.media_player.const import (
|
||||
MEDIA_TYPE_MUSIC,
|
||||
SUPPORT_SELECT_SOUND_MODE,
|
||||
SUPPORT_SELECT_SOURCE,
|
||||
SUPPORT_TURN_ON,
|
||||
SUPPORT_TURN_OFF,
|
||||
SUPPORT_VOLUME_MUTE,
|
||||
SUPPORT_VOLUME_SET,
|
||||
SUPPORT_VOLUME_STEP,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
CONF_NAME,
|
||||
CONF_ZONE,
|
||||
SERVICE_TURN_ON,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
)
|
||||
from homeassistant.helpers.typing import HomeAssistantType, ConfigType
|
||||
from homeassistant.helpers.service import async_call_from_config
|
||||
|
||||
from .const import (
|
||||
SIGNAL_CLIENT_DATA,
|
||||
SIGNAL_CLIENT_STARTED,
|
||||
SIGNAL_CLIENT_STOPPED,
|
||||
DOMAIN_DATA_ENTRIES,
|
||||
DOMAIN,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType,
|
||||
config_entry: config_entries.ConfigEntry,
|
||||
async_add_entities,
|
||||
):
|
||||
"""Set up the configuration entry."""
|
||||
data = hass.data[DOMAIN_DATA_ENTRIES][config_entry.entry_id]
|
||||
client = data["client"]
|
||||
config = data["config"]
|
||||
|
||||
async_add_entities(
|
||||
[
|
||||
ArcamFmj(
|
||||
State(client, zone),
|
||||
zone_config[CONF_NAME],
|
||||
zone_config.get(SERVICE_TURN_ON),
|
||||
)
|
||||
for zone, zone_config in config[CONF_ZONE].items()
|
||||
]
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class ArcamFmj(MediaPlayerDevice):
|
||||
"""Representation of a media device."""
|
||||
|
||||
def __init__(self, state: State, name: str, turn_on: Optional[ConfigType]):
|
||||
"""Initialize device."""
|
||||
self._state = state
|
||||
self._name = name
|
||||
self._turn_on = turn_on
|
||||
self._support = (
|
||||
SUPPORT_SELECT_SOURCE
|
||||
| SUPPORT_VOLUME_SET
|
||||
| SUPPORT_VOLUME_MUTE
|
||||
| SUPPORT_VOLUME_STEP
|
||||
| SUPPORT_TURN_OFF
|
||||
)
|
||||
if state.zn == 1:
|
||||
self._support |= SUPPORT_SELECT_SOUND_MODE
|
||||
|
||||
def _get_2ch(self):
|
||||
"""Return if source is 2 channel or not."""
|
||||
audio_format, _ = self._state.get_incoming_audio_format()
|
||||
return bool(
|
||||
audio_format
|
||||
in (
|
||||
IncomingAudioFormat.PCM,
|
||||
IncomingAudioFormat.ANALOGUE_DIRECT,
|
||||
None,
|
||||
)
|
||||
)
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Return a device description for device registry."""
|
||||
return {
|
||||
"identifiers": {
|
||||
(DOMAIN, self._state.client.host, self._state.client.port)
|
||||
},
|
||||
"model": "FMJ",
|
||||
"manufacturer": "Arcam",
|
||||
}
|
||||
|
||||
@property
|
||||
def should_poll(self) -> bool:
|
||||
"""No need to poll."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the controlled device."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the device."""
|
||||
if self._state.get_power():
|
||||
return STATE_ON
|
||||
return STATE_OFF
|
||||
|
||||
@property
|
||||
def supported_features(self):
|
||||
"""Flag media player features that are supported."""
|
||||
support = self._support
|
||||
if self._state.get_power() is not None or self._turn_on:
|
||||
support |= SUPPORT_TURN_ON
|
||||
return support
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Once registed add listener for events."""
|
||||
await self._state.start()
|
||||
|
||||
@callback
|
||||
def _data(host):
|
||||
if host == self._state.client.host:
|
||||
self.async_schedule_update_ha_state()
|
||||
|
||||
@callback
|
||||
def _started(host):
|
||||
if host == self._state.client.host:
|
||||
self.async_schedule_update_ha_state(force_refresh=True)
|
||||
|
||||
@callback
|
||||
def _stopped(host):
|
||||
if host == self._state.client.host:
|
||||
self.async_schedule_update_ha_state(force_refresh=True)
|
||||
|
||||
self.hass.helpers.dispatcher.async_dispatcher_connect(
|
||||
SIGNAL_CLIENT_DATA, _data
|
||||
)
|
||||
|
||||
self.hass.helpers.dispatcher.async_dispatcher_connect(
|
||||
SIGNAL_CLIENT_STARTED, _started
|
||||
)
|
||||
|
||||
self.hass.helpers.dispatcher.async_dispatcher_connect(
|
||||
SIGNAL_CLIENT_STOPPED, _stopped
|
||||
)
|
||||
|
||||
async def async_update(self):
|
||||
"""Force update of state."""
|
||||
_LOGGER.debug("Update state %s", self.name)
|
||||
await self._state.update()
|
||||
|
||||
async def async_mute_volume(self, mute):
|
||||
"""Send mute command."""
|
||||
await self._state.set_mute(mute)
|
||||
self.async_schedule_update_ha_state()
|
||||
|
||||
async def async_select_source(self, source):
|
||||
"""Select a specific source."""
|
||||
try:
|
||||
value = SourceCodes[source]
|
||||
except KeyError:
|
||||
_LOGGER.error("Unsupported source %s", source)
|
||||
return
|
||||
|
||||
await self._state.set_source(value)
|
||||
self.async_schedule_update_ha_state()
|
||||
|
||||
async def async_select_sound_mode(self, sound_mode):
|
||||
"""Select a specific source."""
|
||||
try:
|
||||
if self._get_2ch():
|
||||
await self._state.set_decode_mode_2ch(
|
||||
DecodeMode2CH[sound_mode]
|
||||
)
|
||||
else:
|
||||
await self._state.set_decode_mode_mch(
|
||||
DecodeModeMCH[sound_mode]
|
||||
)
|
||||
except KeyError:
|
||||
_LOGGER.error("Unsupported sound_mode %s", sound_mode)
|
||||
return
|
||||
|
||||
self.async_schedule_update_ha_state()
|
||||
|
||||
async def async_set_volume_level(self, volume):
|
||||
"""Set volume level, range 0..1."""
|
||||
await self._state.set_volume(round(volume * 99.0))
|
||||
self.async_schedule_update_ha_state()
|
||||
|
||||
async def async_volume_up(self):
|
||||
"""Turn volume up for media player."""
|
||||
await self._state.inc_volume()
|
||||
self.async_schedule_update_ha_state()
|
||||
|
||||
async def async_volume_down(self):
|
||||
"""Turn volume up for media player."""
|
||||
await self._state.dec_volume()
|
||||
self.async_schedule_update_ha_state()
|
||||
|
||||
async def async_turn_on(self):
|
||||
"""Turn the media player on."""
|
||||
if self._state.get_power() is not None:
|
||||
_LOGGER.debug("Turning on device using connection")
|
||||
await self._state.set_power(True)
|
||||
elif self._turn_on:
|
||||
_LOGGER.debug("Turning on device using service call")
|
||||
await async_call_from_config(
|
||||
self.hass,
|
||||
self._turn_on,
|
||||
variables=None,
|
||||
blocking=True,
|
||||
validate_config=False,
|
||||
)
|
||||
else:
|
||||
_LOGGER.error("Unable to turn on")
|
||||
|
||||
async def async_turn_off(self):
|
||||
"""Turn the media player off."""
|
||||
await self._state.set_power(False)
|
||||
|
||||
@property
|
||||
def source(self):
|
||||
"""Return the current input source."""
|
||||
value = self._state.get_source()
|
||||
if value is None:
|
||||
return None
|
||||
return value.name
|
||||
|
||||
@property
|
||||
def source_list(self):
|
||||
"""List of available input sources."""
|
||||
return [x.name for x in self._state.get_source_list()]
|
||||
|
||||
@property
|
||||
def sound_mode(self):
|
||||
"""Name of the current sound mode."""
|
||||
if self._state.zn != 1:
|
||||
return None
|
||||
|
||||
if self._get_2ch():
|
||||
value = self._state.get_decode_mode_2ch()
|
||||
else:
|
||||
value = self._state.get_decode_mode_mch()
|
||||
if value:
|
||||
return value.name
|
||||
return None
|
||||
|
||||
@property
|
||||
def sound_mode_list(self):
|
||||
"""List of available sound modes."""
|
||||
if self._state.zn != 1:
|
||||
return None
|
||||
|
||||
if self._get_2ch():
|
||||
return [x.name for x in DecodeMode2CH]
|
||||
return [x.name for x in DecodeModeMCH]
|
||||
|
||||
@property
|
||||
def is_volume_muted(self):
|
||||
"""Boolean if volume is currently muted."""
|
||||
value = self._state.get_mute()
|
||||
if value is None:
|
||||
return None
|
||||
return value
|
||||
|
||||
@property
|
||||
def volume_level(self):
|
||||
"""Volume level of device."""
|
||||
value = self._state.get_volume()
|
||||
if value is None:
|
||||
return None
|
||||
return value / 99.0
|
||||
|
||||
@property
|
||||
def media_content_type(self):
|
||||
"""Content type of current playing media."""
|
||||
source = self._state.get_source()
|
||||
if source == SourceCodes.DAB:
|
||||
value = MEDIA_TYPE_MUSIC
|
||||
elif source == SourceCodes.FM:
|
||||
value = MEDIA_TYPE_MUSIC
|
||||
else:
|
||||
value = None
|
||||
return value
|
||||
|
||||
@property
|
||||
def media_channel(self):
|
||||
"""Channel currently playing."""
|
||||
source = self._state.get_source()
|
||||
if source == SourceCodes.DAB:
|
||||
value = self._state.get_dab_station()
|
||||
elif source == SourceCodes.FM:
|
||||
value = self._state.get_rds_information()
|
||||
else:
|
||||
value = None
|
||||
return value
|
||||
|
||||
@property
|
||||
def media_artist(self):
|
||||
"""Artist of current playing media, music track only."""
|
||||
source = self._state.get_source()
|
||||
if source == SourceCodes.DAB:
|
||||
value = self._state.get_dls_pdt()
|
||||
else:
|
||||
value = None
|
||||
return value
|
||||
|
||||
@property
|
||||
def media_title(self):
|
||||
"""Title of current playing media."""
|
||||
source = self._state.get_source()
|
||||
if source is None:
|
||||
return None
|
||||
|
||||
channel = self.media_channel
|
||||
|
||||
if channel:
|
||||
value = "{} - {}".format(source.name, channel)
|
||||
else:
|
||||
value = source.name
|
||||
return value
|
||||
|
|
@ -0,0 +1,8 @@
|
|||
{
|
||||
"config": {
|
||||
"title": "Arcam FMJ",
|
||||
"step": {},
|
||||
"error": {},
|
||||
"abort": {}
|
||||
}
|
||||
}
|
||||
|
|
@ -201,6 +201,9 @@ aprslib==0.6.46
|
|||
# homeassistant.components.aqualogic
|
||||
aqualogic==1.0
|
||||
|
||||
# homeassistant.components.arcam_fmj
|
||||
arcam-fmj==0.4.3
|
||||
|
||||
# homeassistant.components.ampio
|
||||
asmog==0.0.6
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the arcam_fmj component."""
|
||||
|
|
@ -0,0 +1,50 @@
|
|||
|
||||
"""Tests for the Arcam FMJ config flow module."""
|
||||
import pytest
|
||||
from homeassistant import data_entry_flow
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
|
||||
from tests.common import MockConfigEntry, MockDependency
|
||||
|
||||
with MockDependency('arcam'), \
|
||||
MockDependency('arcam.fmj'), \
|
||||
MockDependency('arcam.fmj.client'):
|
||||
from homeassistant.components.arcam_fmj import DEVICE_SCHEMA
|
||||
from homeassistant.components.arcam_fmj.config_flow import (
|
||||
ArcamFmjFlowHandler)
|
||||
from homeassistant.components.arcam_fmj.const import DOMAIN
|
||||
|
||||
MOCK_HOST = "127.0.0.1"
|
||||
MOCK_PORT = 1234
|
||||
MOCK_NAME = "Arcam FMJ"
|
||||
MOCK_CONFIG = DEVICE_SCHEMA({
|
||||
CONF_HOST: MOCK_HOST,
|
||||
CONF_PORT: MOCK_PORT,
|
||||
})
|
||||
|
||||
@pytest.fixture(name="config_entry")
|
||||
def config_entry_fixture():
|
||||
"""Create a mock HEOS config entry."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=MOCK_CONFIG,
|
||||
title=MOCK_NAME,
|
||||
)
|
||||
|
||||
async def test_single_import_only(hass, config_entry):
|
||||
"""Test form is shown when host not provided."""
|
||||
config_entry.add_to_hass(hass)
|
||||
flow = ArcamFmjFlowHandler()
|
||||
flow.hass = hass
|
||||
result = await flow.async_step_import(MOCK_CONFIG)
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result['reason'] == 'already_setup'
|
||||
|
||||
async def test_import(hass):
|
||||
"""Test form is shown when host not provided."""
|
||||
flow = ArcamFmjFlowHandler()
|
||||
flow.hass = hass
|
||||
result = await flow.async_step_import(MOCK_CONFIG)
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result['title'] == MOCK_NAME
|
||||
assert result['data'] == MOCK_CONFIG
|
||||
Loading…
Reference in New Issue