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

Arcam config flow (#34384)

Browse source 
Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Joakim Plate 2020-06-06 22:43:28 +02:00 • committed by GitHub
parent 524b48be7d
commit 31973de2d5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 379 additions and 204 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

89
homeassistant/components/arcam_fmj/__init__.py
View file

@ -5,27 +5,15 @@ import logging
from arcam.fmj import ConnectionFailed
from arcam.fmj.client import Client
import async_timeout
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PORT,
CONF_SCAN_INTERVAL,
CONF_ZONE,
EVENT_HOMEASSISTANT_STOP,
SERVICE_TURN_ON,
)
from homeassistant.const import CONF_HOST, CONF_PORT, EVENT_HOMEASSISTANT_STOP
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from .const import (
DEFAULT_NAME,
DEFAULT_PORT,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
DOMAIN_DATA_CONFIG,
DOMAIN_DATA_ENTRIES,
DOMAIN_DATA_TASKS,
SIGNAL_CLIENT_DATA,
@ -35,44 +23,7 @@ from .const import (
_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
] = f"{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 = cv.deprecated(DOMAIN, invalidation_version="0.115")
async def _await_cancel(task):
@ -83,27 +34,10 @@ async def _await_cancel(task):
pass
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_TASKS] = {}
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]},
)
)
async def _stop(_):
asyncio.gather(
@ -116,21 +50,12 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
async def async_setup_entry(hass: HomeAssistantType, entry: config_entries.ConfigEntry):
"""Set up an access point from a config entry."""
"""Set up config entry."""
entries = hass.data[DOMAIN_DATA_ENTRIES]
tasks = hass.data[DOMAIN_DATA_TASKS]
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]}
),
)
tasks = hass.data.setdefault(DOMAIN_DATA_TASKS, {})
hass.data[DOMAIN_DATA_ENTRIES][entry.entry_id] = {
"client": client,
"config": config,
}
entries[entry.entry_id] = client
task = asyncio.create_task(_run_client(hass, client, DEFAULT_SCAN_INTERVAL))
tasks[entry.entry_id] = task

99
homeassistant/components/arcam_fmj/config_flow.py
View file

@ -1,27 +1,102 @@
"""Config flow to configure the Arcam FMJ component."""
from operator import itemgetter
import logging
from urllib.parse import urlparse
from arcam.fmj.client import Client, ConnectionFailed
from arcam.fmj.utils import get_uniqueid_from_host, get_uniqueid_from_udn
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_UDN
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
from .const import DEFAULT_NAME, DEFAULT_PORT, DOMAIN, DOMAIN_DATA_ENTRIES
_GETKEY = itemgetter(CONF_HOST, CONF_PORT)
_LOGGER = logging.getLogger(__name__)
def get_entry_client(hass, entry):
"""Retrieve client associated with a config entry."""
return hass.data[DOMAIN_DATA_ENTRIES][entry.entry_id]
@config_entries.HANDLERS.register(DOMAIN)
class ArcamFmjFlowHandler(config_entries.ConfigFlow):
"""Handle a SimpliSafe config flow."""
"""Handle 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")
async def _async_set_unique_id_and_update(self, host, port, uuid):
await self.async_set_unique_id(uuid)
self._abort_if_unique_id_configured({CONF_HOST: host, CONF_PORT: port})
return self.async_create_entry(title="Arcam FMJ", data=import_config)
async def _async_check_and_create(self, host, port):
client = Client(host, port)
try:
await client.start()
except ConnectionFailed:
return self.async_abort(reason="unable_to_connect")
finally:
await client.stop()
return self.async_create_entry(
title=f"{DEFAULT_NAME} ({host})", data={CONF_HOST: host, CONF_PORT: port},
)
async def async_step_user(self, user_info=None):
"""Handle a discovered device."""
errors = {}
if user_info is not None:
uuid = await get_uniqueid_from_host(
async_get_clientsession(self.hass), user_info[CONF_HOST]
)
if uuid:
await self._async_set_unique_id_and_update(
user_info[CONF_HOST], user_info[CONF_PORT], uuid
)
return await self._async_check_and_create(
user_info[CONF_HOST], user_info[CONF_PORT]
)
fields = {
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT, default=DEFAULT_PORT): int,
}
return self.async_show_form(
step_id="user", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_confirm(self, user_input=None):
"""Handle user-confirmation of discovered node."""
context = self.context # pylint: disable=no-member
placeholders = {
"host": context[CONF_HOST],
}
context["title_placeholders"] = placeholders
if user_input is not None:
return await self._async_check_and_create(
context[CONF_HOST], context[CONF_PORT]
)
return self.async_show_form(
step_id="confirm", description_placeholders=placeholders
)
async def async_step_ssdp(self, discovery_info):
"""Handle a discovered device."""
host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
port = DEFAULT_PORT
uuid = get_uniqueid_from_udn(discovery_info[ATTR_UPNP_UDN])
await self._async_set_unique_id_and_update(host, port, uuid)
context = self.context # pylint: disable=no-member
context[CONF_HOST] = host
context[CONF_PORT] = DEFAULT_PORT
return await self.async_step_confirm()

1
homeassistant/components/arcam_fmj/const.py
View file

@ -13,4 +13,3 @@ DEFAULT_SCAN_INTERVAL = 5
DOMAIN_DATA_ENTRIES = f"{DOMAIN}.entries"
DOMAIN_DATA_TASKS = f"{DOMAIN}.tasks"
DOMAIN_DATA_CONFIG = f"{DOMAIN}.config"

10
homeassistant/components/arcam_fmj/manifest.json
View file

@ -1,8 +1,14 @@
{
"domain": "arcam_fmj",
"name": "Arcam FMJ Receivers",
"config_flow": false,
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/arcam_fmj",
"requirements": ["arcam-fmj==0.4.6"],
"requirements": ["arcam-fmj==0.5.1"],
"ssdp": [
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:1",
"manufacturer": "ARCAM"
}
],
"codeowners": ["@elupus"]
}

53
homeassistant/components/arcam_fmj/media_player.py
View file

@ -1,6 +1,5 @@
"""Arcam media player."""
import logging
from typing import Optional
from arcam.fmj import DecodeMode2CH, DecodeModeMCH, IncomingAudioFormat, SourceCodes
from arcam.fmj.state import State
@ -17,21 +16,13 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_SET,
SUPPORT_VOLUME_STEP,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_NAME,
CONF_ZONE,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON
from homeassistant.core import callback
from homeassistant.helpers.service import async_call_from_config
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from homeassistant.helpers.typing import HomeAssistantType
from .config_flow import get_entry_client
from .const import (
DOMAIN,
DOMAIN_DATA_ENTRIES,
EVENT_TURN_ON,
SIGNAL_CLIENT_DATA,
SIGNAL_CLIENT_STARTED,
@ -47,19 +38,17 @@ async def async_setup_entry(
async_add_entities,
):
"""Set up the configuration entry."""
data = hass.data[DOMAIN_DATA_ENTRIES][config_entry.entry_id]
client = data["client"]
config = data["config"]
client = get_entry_client(hass, config_entry)
async_add_entities(
[
ArcamFmj(
config_entry.title,
State(client, zone),
config_entry.unique_id or config_entry.entry_id,
zone_config[CONF_NAME],
zone_config.get(SERVICE_TURN_ON),
)
for zone, zone_config in config[CONF_ZONE].items()
for zone in [1, 2]
],
True,
)
@ -71,13 +60,13 @@ class ArcamFmj(MediaPlayerEntity):
"""Representation of a media device."""
def __init__(
self, state: State, uuid: str, name: str, turn_on: Optional[ConfigType]
self, device_name, state: State, uuid: str,
):
"""Initialize device."""
self._state = state
self._device_name = device_name
self._name = f"{device_name} - Zone: {state.zn}"
self._uuid = uuid
self._name = name
self._turn_on = turn_on
self._support = (
SUPPORT_SELECT_SOURCE
| SUPPORT_VOLUME_SET
@ -102,6 +91,11 @@ class ArcamFmj(MediaPlayerEntity):
)
)
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return self._state.zn == 1
@property
def unique_id(self):
"""Return unique identifier if known."""
@ -111,8 +105,12 @@ class ArcamFmj(MediaPlayerEntity):
def device_info(self):
"""Return a device description for device registry."""
return {
"identifiers": {(DOMAIN, self._state.client.host, self._state.client.port)},
"model": "FMJ",
"name": self._device_name,
"identifiers": {
(DOMAIN, self._uuid),
(DOMAIN, self._state.client.host, self._state.client.port),
},
"model": "Arcam FMJ AVR",
"manufacturer": "Arcam",
}
@ -229,15 +227,6 @@ class ArcamFmj(MediaPlayerEntity):
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.debug("Firing event to turn on device")
self.hass.bus.async_fire(EVENT_TURN_ON, {ATTR_ENTITY_ID: self.entity_id})

23
homeassistant/components/arcam_fmj/strings.json
View file

@ -1,7 +1,28 @@
{
"config": {
"abort": {
"already_configured": "Device was already setup.",
"already_in_progress": "Config flow for device is already in progress.",
"unable_to_connect": "Unable to connect to device."
},
"error": {},
"flow_title": "Arcam FMJ on {host}",
"step": {
"confirm": {
"description": "Do you want to add Arcam FMJ on `{host}` to Home Assistant?"
},
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]"
},
"description": "Please enter the host name or IP address of device."
}
}
},
"device_automation": {
"trigger_type": {
"turn_on": "{entity_name} was requested to turn on"
}
}
}
}

22
homeassistant/components/arcam_fmj/translations/en.json
View file

@ -1,4 +1,26 @@
{
"title": "Arcam FMJ",
"config": {
"abort": {
"already_configured": "Device was already setup.",
"already_in_progress": "Config flow for device is already in progress.",
"unable_to_connect": "Unable to connect to device."
},
"error": {},
"flow_title": "Arcam FMJ on {host}",
"step": {
"confirm": {
"description": "Do you want to add Arcam FMJ on `{host}` to Home Assistant?"
},
"user": {
"data": {
"host": "Host",
"port": "Port"
},
"description": "Please enter the host name or IP address of device."
}
}
},
"device_automation": {
"trigger_type": {
"turn_on": "{entity_name} was requested to turn on"

1
homeassistant/generated/config_flows.py
View file

@ -15,6 +15,7 @@ FLOWS = [
"almond",
"ambiclimate",
"ambient_station",
"arcam_fmj",
"atag",
"august",
"avri",

6
homeassistant/generated/ssdp.py
View file

@ -6,6 +6,12 @@ To update, run python3 -m script.hassfest
# fmt: off
SSDP = {
"arcam_fmj": [
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:1",
"manufacturer": "ARCAM"
}
],
"deconz": [
{
"manufacturer": "Royal Philips Electronics"

2
requirements_all.txt
View file

@ -272,7 +272,7 @@ aprslib==0.6.46
aqualogic==1.0
# homeassistant.components.arcam_fmj
arcam-fmj==0.4.6
arcam-fmj==0.5.1
# homeassistant.components.arris_tg2492lg
arris-tg2492lg==1.0.0

2
requirements_test_all.txt
View file

@ -137,7 +137,7 @@ apprise==0.8.5
aprslib==0.6.46
# homeassistant.components.arcam_fmj
arcam-fmj==0.4.6
arcam-fmj==0.5.1
# homeassistant.components.dlna_dmr
# homeassistant.components.upnp

32
tests/components/arcam_fmj/conftest.py
View file

@ -3,30 +3,24 @@ from arcam.fmj.client import Client
from arcam.fmj.state import State
import pytest
from homeassistant.components.arcam_fmj import DEVICE_SCHEMA
from homeassistant.components.arcam_fmj.const import DOMAIN
from homeassistant.components.arcam_fmj.const import DEFAULT_NAME
from homeassistant.components.arcam_fmj.media_player import ArcamFmj
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.setup import async_setup_component
from tests.async_mock import Mock, patch
from tests.common import MockConfigEntry
MOCK_HOST = "127.0.0.1"
MOCK_PORT = 1234
MOCK_PORT = 50000
MOCK_TURN_ON = {
"service": "switch.turn_on",
"data": {"entity_id": "switch.test"},
}
MOCK_NAME = "dummy"
MOCK_UUID = "1234"
MOCK_ENTITY_ID = "media_player.arcam_fmj_127_0_0_1_1234_1"
MOCK_CONFIG = DEVICE_SCHEMA({CONF_HOST: MOCK_HOST, CONF_PORT: MOCK_PORT})
@pytest.fixture(name="config")
def config_fixture():
"""Create hass config fixture."""
return {DOMAIN: [{CONF_HOST: MOCK_HOST, CONF_PORT: MOCK_PORT}]}
MOCK_ENTITY_ID = "media_player.arcam_fmj_127_0_0_1_zone_1"
MOCK_UUID = "456789abcdef"
MOCK_UDN = f"uuid:01234567-89ab-cdef-0123-{MOCK_UUID}"
MOCK_NAME = f"{DEFAULT_NAME} ({MOCK_HOST})"
MOCK_CONFIG_ENTRY = {CONF_HOST: MOCK_HOST, CONF_PORT: MOCK_PORT}
@pytest.fixture(name="client")
@ -75,7 +69,7 @@ def state_fixture(state_1):
@pytest.fixture(name="player")
def player_fixture(hass, state):
"""Get standard player."""
player = ArcamFmj(state, MOCK_UUID, MOCK_NAME, None)
player = ArcamFmj(MOCK_NAME, state, MOCK_UUID)
player.entity_id = MOCK_ENTITY_ID
player.hass = hass
player.async_write_ha_state = Mock()
@ -83,8 +77,12 @@ def player_fixture(hass, state):
@pytest.fixture(name="player_setup")
async def player_setup_fixture(hass, config, state_1, state_2, client):
async def player_setup_fixture(hass, state_1, state_2, client):
"""Get standard player."""
config_entry = MockConfigEntry(
domain="arcam_fmj", data=MOCK_CONFIG_ENTRY, title=MOCK_NAME
)
config_entry.add_to_hass(hass)
def state_mock(cli, zone):
if zone == 1:
@ -95,6 +93,6 @@ async def player_setup_fixture(hass, config, state_1, state_2, client):
with patch("homeassistant.components.arcam_fmj.Client", return_value=client), patch(
"homeassistant.components.arcam_fmj.media_player.State", side_effect=state_mock
), patch("homeassistant.components.arcam_fmj._run_client", return_value=None):
assert await async_setup_component(hass, "arcam_fmj", config)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
yield MOCK_ENTITY_ID

189
tests/components/arcam_fmj/test_config_flow.py
View file

@ -1,37 +1,182 @@
"""Tests for the Arcam FMJ config flow module."""
from arcam.fmj.client import ConnectionFailed
import pytest
from homeassistant import data_entry_flow
from homeassistant.components.arcam_fmj.config_flow import ArcamFmjFlowHandler
from homeassistant.components.arcam_fmj.const import DOMAIN
from homeassistant.components import ssdp
from homeassistant.components.arcam_fmj.config_flow import get_entry_client
from homeassistant.components.arcam_fmj.const import DOMAIN, DOMAIN_DATA_ENTRIES
from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SOURCE
from .conftest import MOCK_CONFIG, MOCK_NAME
from .conftest import (
MOCK_CONFIG_ENTRY,
MOCK_HOST,
MOCK_NAME,
MOCK_PORT,
MOCK_UDN,
MOCK_UUID,
)
from tests.async_mock import AsyncMock, patch
from tests.common import MockConfigEntry
MOCK_UPNP_DEVICE = f"""
<root xmlns="urn:schemas-upnp-org:device-1-0">
<device>
<UDN>{MOCK_UDN}</UDN>
</device>
</root>
"""
@pytest.fixture(name="config_entry")
def config_entry_fixture():
"""Create a mock Arcam config entry."""
return MockConfigEntry(domain=DOMAIN, data=MOCK_CONFIG, title=MOCK_NAME)
MOCK_UPNP_LOCATION = f"http://{MOCK_HOST}:8080/dd.xml"
MOCK_DISCOVER = {
ssdp.ATTR_UPNP_MANUFACTURER: "ARCAM",
ssdp.ATTR_UPNP_MODEL_NAME: " ",
ssdp.ATTR_UPNP_MODEL_NUMBER: "AVR450, AVR750",
ssdp.ATTR_UPNP_FRIENDLY_NAME: f"Arcam media client {MOCK_UUID}",
ssdp.ATTR_UPNP_SERIAL: "12343",
ssdp.ATTR_SSDP_LOCATION: f"http://{MOCK_HOST}:8080/dd.xml",
ssdp.ATTR_UPNP_UDN: MOCK_UDN,
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:MediaRenderer:1",
}
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"
@pytest.fixture(name="dummy_client", autouse=True)
def dummy_client_fixture(hass):
"""Mock out the real client."""
with patch("homeassistant.components.arcam_fmj.config_flow.Client") as client:
client.return_value.start.side_effect = AsyncMock(return_value=None)
client.return_value.stop.side_effect = AsyncMock(return_value=None)
yield client.return_value
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)
async def test_ssdp(hass, dummy_client):
"""Test a ssdp import flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=MOCK_DISCOVER,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Arcam FMJ"
assert result["data"] == MOCK_CONFIG
assert result["title"] == f"Arcam FMJ ({MOCK_HOST})"
assert result["data"] == MOCK_CONFIG_ENTRY
async def test_ssdp_abort(hass):
"""Test a ssdp import flow."""
entry = MockConfigEntry(
domain=DOMAIN, data=MOCK_CONFIG_ENTRY, title=MOCK_NAME, unique_id=MOCK_UUID
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=MOCK_DISCOVER,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_ssdp_unable_to_connect(hass, dummy_client):
"""Test a ssdp import flow."""
dummy_client.start.side_effect = AsyncMock(side_effect=ConnectionFailed)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=MOCK_DISCOVER,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "unable_to_connect"
async def test_ssdp_update(hass):
"""Test a ssdp import flow."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "old_host", CONF_PORT: MOCK_PORT},
title=MOCK_NAME,
unique_id=MOCK_UUID,
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=MOCK_DISCOVER,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert entry.data[CONF_HOST] == MOCK_HOST
async def test_user(hass, aioclient_mock):
"""Test a manual user configuration flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER}, data=None,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
user_input = {
CONF_HOST: MOCK_HOST,
CONF_PORT: MOCK_PORT,
}
aioclient_mock.get(MOCK_UPNP_LOCATION, text=MOCK_UPNP_DEVICE)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == f"Arcam FMJ ({MOCK_HOST})"
assert result["data"] == MOCK_CONFIG_ENTRY
assert result["result"].unique_id == MOCK_UUID
async def test_invalid_ssdp(hass, aioclient_mock):
"""Test a a config flow where ssdp fails."""
user_input = {
CONF_HOST: MOCK_HOST,
CONF_PORT: MOCK_PORT,
}
aioclient_mock.get(MOCK_UPNP_LOCATION, text="")
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER}, data=user_input,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == f"Arcam FMJ ({MOCK_HOST})"
assert result["data"] == MOCK_CONFIG_ENTRY
assert result["result"].unique_id is None
async def test_user_wrong(hass, aioclient_mock):
"""Test a manual user configuration flow with no ssdp response."""
user_input = {
CONF_HOST: MOCK_HOST,
CONF_PORT: MOCK_PORT,
}
aioclient_mock.get(MOCK_UPNP_LOCATION, status=404)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER}, data=user_input,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == f"Arcam FMJ ({MOCK_HOST})"
assert result["result"].unique_id is None
async def test_get_entry_client(hass):
"""Test helper for configuration."""
entry = MockConfigEntry(
domain=DOMAIN, data=MOCK_CONFIG_ENTRY, title=MOCK_NAME, unique_id=MOCK_UUID
)
hass.data[DOMAIN_DATA_ENTRIES] = {entry.entry_id: "dummy"}
assert get_entry_client(hass, entry) == "dummy"

54
tests/components/arcam_fmj/test_media_player.py
View file

@ -4,10 +4,14 @@ from math import isclose
from arcam.fmj import DecodeMode2CH, DecodeModeMCH, IncomingAudioFormat, SourceCodes
import pytest
from homeassistant.components.media_player.const import MEDIA_TYPE_MUSIC
from homeassistant.core import HomeAssistant
from homeassistant.components.media_player.const import (
ATTR_INPUT_SOURCE,
MEDIA_TYPE_MUSIC,
SERVICE_SELECT_SOURCE,
)
from homeassistant.const import ATTR_ENTITY_ID
from .conftest import MOCK_ENTITY_ID, MOCK_HOST, MOCK_NAME, MOCK_PORT, MOCK_UUID
from .conftest import MOCK_HOST, MOCK_NAME, MOCK_PORT, MOCK_UUID
from tests.async_mock import ANY, MagicMock, Mock, PropertyMock, patch
@ -27,8 +31,9 @@ async def test_properties(player, state):
"""Test standard properties."""
assert player.unique_id == f"{MOCK_UUID}-1"
assert player.device_info == {
"identifiers": {("arcam_fmj", MOCK_HOST, MOCK_PORT)},
"model": "FMJ",
"name": f"Arcam FMJ ({MOCK_HOST})",
"identifiers": {("arcam_fmj", MOCK_UUID), ("arcam_fmj", MOCK_HOST, MOCK_PORT)},
"model": "Arcam FMJ AVR",
"manufacturer": "Arcam",
}
assert not player.should_poll
@ -55,12 +60,12 @@ async def test_powered_on(player, state):
async def test_supported_features(player, state):
"""Test support when turn on service exist."""
"""Test supported features."""
data = await update(player)
assert data.attributes["supported_features"] == 69004
async def test_turn_on_without_service(player, state):
async def test_turn_on(player, state):
"""Test turn on service."""
state.get_power.return_value = None
await player.async_turn_on()
@ -71,29 +76,6 @@ async def test_turn_on_without_service(player, state):
state.set_power.assert_called_with(True)
async def test_turn_on_with_service(hass, state):
"""Test support when turn on service exist."""
from homeassistant.components.arcam_fmj.media_player import ArcamFmj
player = ArcamFmj(state, MOCK_UUID, "dummy", MOCK_TURN_ON)
player.hass = Mock(HomeAssistant)
player.entity_id = MOCK_ENTITY_ID
with patch(
"homeassistant.components.arcam_fmj.media_player.async_call_from_config"
) as async_call_from_config:
state.get_power.return_value = None
await player.async_turn_on()
state.set_power.assert_not_called()
async_call_from_config.assert_called_with(
player.hass,
MOCK_TURN_ON,
variables=None,
blocking=True,
validate_config=False,
)
async def test_turn_off(player, state):
"""Test command to turn off."""
await player.async_turn_off()
@ -110,7 +92,7 @@ async def test_mute_volume(player, state, mute):
async def test_name(player):
"""Test name."""
assert player.name == MOCK_NAME
assert player.name == f"{MOCK_NAME} - Zone: 1"
async def test_update(player, state):
@ -138,9 +120,15 @@ async def test_2ch(player, state, fmt, result):
"source, value",
[("PVR", SourceCodes.PVR), ("BD", SourceCodes.BD), ("INVALID", None)],
)
async def test_select_source(player, state, source, value):
async def test_select_source(hass, player_setup, state, source, value):
"""Test selection of source."""
await player.async_select_source(source)
await hass.services.async_call(
"media_player",
SERVICE_SELECT_SOURCE,
service_data={ATTR_ENTITY_ID: player_setup, ATTR_INPUT_SOURCE: source},
blocking=True,
)
if value:
state.set_source.assert_called_with(value)
else: