home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-08-17 19:41:45 +02:00
Code Issues Packages Projects Releases Wiki Activity

use async_setup for config

Browse Source
This commit is contained in:
escoand
2019-11-06 22:49:22 +01:00
parent d80505f1e0
commit d664008651
5 changed files with 161 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
homeassistant/components/samsungtv/__init__.py
View File

@@ -1,8 +1,42 @@
"""The Samsung TV integration."""
import voluptuous as vol
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, CONF_PORT, CONF_TIMEOUT
import homeassistant.helpers.config_validation as cv
from .const import DEFAULT_NAME, DEFAULT_TIMEOUT, DOMAIN
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
[
vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PORT): cv.port,
vol.Optional(CONF_MAC): cv.string,
vol.Optional(
CONF_TIMEOUT, default=DEFAULT_TIMEOUT
): cv.positive_int,
}
)
]
)
},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass, config):
"""Set up the Samsung TV integration."""
if DOMAIN in config:
for entry_config in config[DOMAIN]:
await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "import"}, data=entry_config
)
return True

47
homeassistant/components/samsungtv/config_flow.py
View File

@@ -6,7 +6,15 @@ from samsungctl.exceptions import AccessDenied, UnhandledResponse
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_ID, CONF_IP_ADDRESS, CONF_NAME
from homeassistant.const import (
CONF_HOST,
CONF_ID,
CONF_IP_ADDRESS,
CONF_MAC,
CONF_NAME,
CONF_PORT,
CONF_TIMEOUT,
)
from homeassistant.components.ssdp import (
ATTR_HOST,
ATTR_NAME,
@@ -81,22 +89,28 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Initialize flow."""
self._host = None
self._ip = None
self._mac = None
self._manufacturer = None
self._model = None
self._uuid = None
self._name = None
self._port = None
self._timeout = None
self._title = None
self._uuid = None
def _get_entry(self):
return self.async_create_entry(
title=self._title,
data={
CONF_HOST: self._host,
CONF_ID: self._uuid,
CONF_IP_ADDRESS: self._ip,
CONF_NAME: self._name,
CONF_MAC: self._mac,
CONF_MANUFACTURER: self._manufacturer,
CONF_MODEL: self._model,
CONF_ID: self._uuid,
CONF_NAME: self._name,
CONF_PORT: self._port,
CONF_TIMEOUT: self._timeout,
},
)
@@ -104,15 +118,15 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a flow initialized by the user."""
if user_input is not None:
ip_address = await self.hass.async_add_executor_job(
_get_ip, user_input[ATTR_HOST]
_get_ip, user_input[CONF_HOST]
)
if _is_already_configured(self.hass, ip_address):
return self.async_abort(reason="already_configured")
self._host = user_input[CONF_HOST]
self._title = user_input[CONF_NAME]
self._ip = self.context[CONF_IP_ADDRESS] = ip_address
self._title = user_input[CONF_NAME]
result = await self.hass.async_add_executor_job(
_try_connect, self._host, self._title
@@ -143,13 +157,13 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._ip = self.context[CONF_IP_ADDRESS] = ip_address
self._manufacturer = user_input[ATTR_MANUFACTURER]
self._model = user_input[ATTR_MODEL_NAME]
self._uuid = user_input[ATTR_UDN]
if self._uuid.startswith("uuid:"):
self._uuid = self._uuid[5:]
self._name = user_input[ATTR_NAME]
if self._name.startswith("[TV]"):
self._name = self._name[4:]
self._title = f"{self._name} ({self._model})"
self._uuid = user_input[ATTR_UDN]
if self._uuid.startswith("uuid:"):
self._uuid = self._uuid[5:]
return await self.async_step_confirm()
@@ -167,3 +181,18 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form(
step_id="confirm", description_placeholders={"model": self._model}
)
async def async_step_import(self, user_input):
"""Handle a flow initialized by config import."""
ip_address = await self.hass.async_add_executor_job(
_get_ip, user_input.get(CONF_HOST)
)
self._host = user_input.get(CONF_HOST)
self._ip = self.context[CONF_IP_ADDRESS] = ip_address
self._mac = user_input.get(CONF_MAC)
self._port = user_input.get(CONF_PORT)
self._timeout = user_input.get(CONF_TIMEOUT)
self._title = user_input.get(CONF_NAME)
return self._get_entry()

3
homeassistant/components/samsungtv/const.py
View File

@@ -4,6 +4,9 @@ import logging
LOGGER = logging.getLogger(__package__)
DOMAIN = "samsungtv"
DEFAULT_NAME = "Samsung TV Remote"
DEFAULT_TIMEOUT = 1
CONF_MANUFACTURER = "manufacturer"
CONF_MODEL = "model"

29
homeassistant/components/samsungtv/media_player.py
View File

@@ -39,12 +39,17 @@ from homeassistant.const import (
import homeassistant.helpers.config_validation as cv
from homeassistant.util import dt as dt_util
from .const import CONF_MANUFACTURER, CONF_MODEL, DOMAIN, LOGGER, METHODS
from .const import (
CONF_MANUFACTURER,
CONF_MODEL,
DEFAULT_NAME,
DEFAULT_TIMEOUT,
DOMAIN,
LOGGER,
METHODS,
)
DEFAULT_NAME = "Samsung TV Remote"
DEFAULT_TIMEOUT = 1
DEFAULT_BROADCAST_ADDRESS = "255.255.255.255"
KEY_PRESS_TIMEOUT = 1.2
KNOWN_DEVICES_KEY = "samsungtv_known_devices"
SOURCES = {"TV": "KEY_TV", "HDMI": "KEY_HDMI"}
@@ -119,15 +124,15 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Samsung TV from a config entry."""
host = config_entry.data[CONF_HOST]
port = None
name = config_entry.title
timeout = DEFAULT_TIMEOUT
mac = None
host = config_entry.data.get(CONF_HOST)
mac = config_entry.data.get(CONF_MAC)
broadcast = config_entry.data.get(CONF_BROADCAST_ADDRESS, DEFAULT_BROADCAST_ADDRESS)
uuid = config_entry.data[CONF_ID]
manufacturer = config_entry.data[CONF_MANUFACTURER]
model = config_entry.data[CONF_MODEL]
manufacturer = config_entry.data.get(CONF_MANUFACTURER)
model = config_entry.data.get(CONF_MODEL)
name = config_entry.title
port = config_entry.data.get(CONF_PORT)
timeout = config_entry.data.get(CONF_TIMEOUT)
uuid = config_entry.data.get(CONF_ID)
async_add_devices(
[

69
tests/components/samsungtv/test_init.py Normal file
View File

@@ -0,0 +1,69 @@
"""Tests for the Samsung TV Integration."""
import pytest
from unittest.mock import call, patch
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_SUPPORTED_FEATURES,
CONF_HOST,
CONF_MAC,
CONF_NAME,
CONF_PORT,
CONF_TIMEOUT,
SERVICE_VOLUME_UP,
)
from homeassistant.components.media_player.const import DOMAIN, SUPPORT_TURN_ON
from homeassistant.components import samsungtv
from homeassistant.components.samsungtv.const import DOMAIN as SAMSUNGTV_DOMAIN
from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV
ENTITY_ID = f"{DOMAIN}.fake_name"
MOCK_CONFIG = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake_host",
CONF_MAC: "fake_mac",
CONF_NAME: "fake_name",
CONF_PORT: 1234,
CONF_TIMEOUT: 999,
}
]
}
REMOTE_CALL = {
"name": "HomeAssistant",
"description": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_NAME],
"id": "ha.component.samsung",
"method": "websocket",
"port": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_PORT],
"host": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_HOST],
"timeout": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_TIMEOUT],
}
@pytest.fixture(name="remote")
def remote_fixture():
"""Patch the samsungctl Remote."""
with patch("samsungctl.Remote") as remote, patch(
"homeassistant.components.samsungtv.config_flow.socket"
):
yield remote
async def test_setup(hass, remote):
"""Test Samsung TV integration is setup."""
await samsungtv.async_setup(hass, MOCK_CONFIG)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
# test name and mac
assert state
assert state.name == "fake_name"
assert (
state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON
)
# test host, port and timeout
assert await hass.services.async_call(
DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
)
assert remote.mock_calls[0] == call(REMOTE_CALL)