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

Implement config and option flow for rfxtrx integration (#39117)

Browse source 
* Create option flow for Rfxtrx integration (#37982)

* Implement config flow for rfxtrx integration (#39299)

* Add config flow

* Add strings

* Add first series of tests

* Add tests

* Adjust tests according review comments

* Adjust strings

* Add executor for testing connection

* Change ports to dict

* Fix pylint issue

* Adjust tests

* Migrate config entry for rfxtrx integration (#39528)

* Add rfxtrx device connection validation when importing (#39582)

* Implement import connection validation

* Fix binary sensor tests

* Move rfxtrx data

* Fix cover tests

* Fix test init

* Fix light tests

* Fix sensor tests

* Fix switch tests

* Refactor rfxtrx test data

* Fix strings

* Fix check

* Rework device string in test code

* Add option to delete multiple rfxtrx devices (#39625)

* Opt to remove multiple devices

* Fix devices key

* Add tests (phase 1)

* Add tests (phase 2)

* Tweak remove devices test

* Implement device migration function in rfxtrx option flow (#39694)

* Prompt option to replace device

* Revert unwanted changes

* Add replace device function

* WIP replace entities

* Remove device/entities and update config entry

* Fix styling

* Add info

* Add test

* Fix strings

* Refactor building migration map

* Allow migration for all device types

* Add test to migrate control device

* Fixup some names

* Fixup entry names in test code

* Bump pyRFXtrx to 0.26 and deprecate debug config key (#40679)

* Create option flow for Rfxtrx integration (#37982)

* Implement config flow for rfxtrx integration (#39299)

* Add config flow

* Add strings

* Add first series of tests

* Add tests

* Adjust tests according review comments

* Adjust strings

* Add executor for testing connection

* Change ports to dict

* Fix pylint issue

* Adjust tests

* Migrate config entry for rfxtrx integration (#39528)

* Add rfxtrx device connection validation when importing (#39582)

* Implement import connection validation

* Fix binary sensor tests

* Move rfxtrx data

* Fix cover tests

* Fix test init

* Fix light tests

* Fix sensor tests

* Fix switch tests

* Refactor rfxtrx test data

* Fix strings

* Fix check

* Rework device string in test code

* Add option to delete multiple rfxtrx devices (#39625)

* Opt to remove multiple devices

* Fix devices key

* Add tests (phase 1)

* Add tests (phase 2)

* Tweak remove devices test

* Implement device migration function in rfxtrx option flow (#39694)

* Prompt option to replace device

* Revert unwanted changes

* Add replace device function

* WIP replace entities

* Remove device/entities and update config entry

* Fix styling

* Add info

* Add test

* Fix strings

* Refactor building migration map

* Allow migration for all device types

* Add test to migrate control device

* Fixup some names

* Fixup entry names in test code

* Bump version number

* Remove debug key from connect

* Remove debug option from config flow

* Remove debug from tests

* Fix event test

* Add cv.deprecated

* Fix test

* Fix config schema

* Add timeout on connection

* Rework config schema

* Fix schema...again

* Prevent creation of duplicate device in rfxtrx option flow (#40656)
This commit is contained in:
Rob Bierbooms 2020-10-01 08:55:57 +02:00 • committed by GitHub
parent 44e5ec58b0
commit c5041b41c8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
22 changed files with 2202 additions and 363 deletions
Download patch file Download diff file Expand all files Collapse all files

2
CODEOWNERS
View file

@ -355,7 +355,7 @@ homeassistant/components/rainmachine/* @bachya
homeassistant/components/random/* @fabaff
homeassistant/components/rejseplanen/* @DarkFox
homeassistant/components/repetier/* @MTrab
homeassistant/components/rfxtrx/* @danielhiversen @elupus
homeassistant/components/rfxtrx/* @danielhiversen @elupus @RobBie1221
homeassistant/components/ring/* @balloob
homeassistant/components/risco/* @OnFreund
homeassistant/components/rmvtransport/* @cgtobi

36
homeassistant/components/rfxtrx/__init__.py
View file

@ -5,6 +5,7 @@ from collections import OrderedDict
import logging
import RFXtrx as rfxtrxmod
import async_timeout
import voluptuous as vol
from homeassistant import config_entries
@ -33,11 +34,19 @@ from homeassistant.const import (
VOLT,
)
from homeassistant.core import callback
from homeassistant.exceptions import ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.restore_state import RestoreEntity
from .const import (
ATTR_EVENT,
CONF_AUTOMATIC_ADD,
CONF_DATA_BITS,
CONF_DEBUG,
CONF_FIRE_EVENT,
CONF_OFF_DELAY,
CONF_REMOVE_DEVICE,
CONF_SIGNAL_REPETITIONS,
DEVICE_PACKET_TYPE_LIGHTING4,
EVENT_RFXTRX_EVENT,
SERVICE_SEND,
@ -47,12 +56,6 @@ DOMAIN = "rfxtrx"
DEFAULT_SIGNAL_REPETITIONS = 1
CONF_FIRE_EVENT = "fire_event"
CONF_DATA_BITS = "data_bits"
CONF_AUTOMATIC_ADD = "automatic_add"
CONF_SIGNAL_REPETITIONS = "signal_repetitions"
CONF_DEBUG = "debug"
CONF_OFF_DELAY = "off_delay"
SIGNAL_EVENT = f"{DOMAIN}_event"
DATA_TYPES = OrderedDict(
@ -126,10 +129,10 @@ DEVICE_DATA_SCHEMA = vol.Schema(
BASE_SCHEMA = vol.Schema(
{
vol.Optional(CONF_DEBUG, default=False): cv.boolean,
vol.Optional(CONF_DEBUG): cv.boolean,
vol.Optional(CONF_AUTOMATIC_ADD, default=False): cv.boolean,
vol.Optional(CONF_DEVICES, default={}): {cv.string: _ensure_device},
}
},
)
DEVICE_SCHEMA = BASE_SCHEMA.extend({vol.Required(CONF_DEVICE): cv.string})
@ -139,7 +142,8 @@ PORT_SCHEMA = BASE_SCHEMA.extend(
)
CONFIG_SCHEMA = vol.Schema(
{DOMAIN: vol.Any(DEVICE_SCHEMA, PORT_SCHEMA)}, extra=vol.ALLOW_EXTRA
{DOMAIN: vol.All(cv.deprecated(CONF_DEBUG), vol.Any(DEVICE_SCHEMA, PORT_SCHEMA))},
extra=vol.ALLOW_EXTRA,
)
DOMAINS = ["switch", "sensor", "light", "binary_sensor", "cover"]
@ -154,7 +158,6 @@ async def async_setup(hass, config):
CONF_HOST: config[DOMAIN].get(CONF_HOST),
CONF_PORT: config[DOMAIN].get(CONF_PORT),
CONF_DEVICE: config[DOMAIN].get(CONF_DEVICE),
CONF_DEBUG: config[DOMAIN].get(CONF_DEBUG),
CONF_AUTOMATIC_ADD: config[DOMAIN].get(CONF_AUTOMATIC_ADD),
CONF_DEVICES: config[DOMAIN][CONF_DEVICES],
}
@ -223,11 +226,10 @@ def _create_rfx(config):
rfx = rfxtrxmod.Connect(
(config[CONF_HOST], config[CONF_PORT]),
None,
debug=config[CONF_DEBUG],
transport_protocol=rfxtrxmod.PyNetworkTransport,
)
else:
rfx = rfxtrxmod.Connect(config[CONF_DEVICE], None, debug=config[CONF_DEBUG])
rfx = rfxtrxmod.Connect(config[CONF_DEVICE], None)
return rfx
@ -251,7 +253,11 @@ async def async_setup_internal(hass, entry: config_entries.ConfigEntry):
config = entry.data
# Initialize library
try:
async with async_timeout.timeout(5):
rfx_object = await hass.async_add_executor_job(_create_rfx, config)
except asyncio.TimeoutError as err:
raise ConfigEntryNotReady from err
# Setup some per device config
devices = _get_device_lookup(config[CONF_DEVICES])
@ -444,6 +450,12 @@ class RfxtrxEntity(RestoreEntity):
)
)
self.async_on_remove(
self.hass.helpers.dispatcher.async_dispatcher_connect(
f"{DOMAIN}_{CONF_REMOVE_DEVICE}_{self._device_id}", self.async_remove
)
)
@property
def should_poll(self):
"""No polling needed for a RFXtrx switch."""

22
homeassistant/components/rfxtrx/binary_sensor.py
View file

@ -61,6 +61,18 @@ DEVICE_TYPE_DEVICE_CLASS = {
}
def supported(event):
"""Return whether an event supports binary_sensor."""
if isinstance(event, rfxtrxmod.ControlEvent):
return True
if isinstance(event, rfxtrxmod.SensorEvent):
return event.values.get("Sensor Status") in [
*SENSOR_STATUS_ON,
*SENSOR_STATUS_OFF,
]
return False
async def async_setup_entry(
hass,
config_entry,
@ -74,16 +86,6 @@ async def async_setup_entry(
discovery_info = config_entry.data
def supported(event):
if isinstance(event, rfxtrxmod.ControlEvent):
return True
if isinstance(event, rfxtrxmod.SensorEvent):
return event.values.get("Sensor Status") in [
*SENSOR_STATUS_ON,
*SENSOR_STATUS_OFF,
]
return False
for packet_id, entity_info in discovery_info[CONF_DEVICES].items():
event = get_rfx_object(packet_id)
if event is None:

581
homeassistant/components/rfxtrx/config_flow.py
View file

@ -1,12 +1,404 @@
"""Config flow for RFXCOM RFXtrx integration."""
import copy
import logging
import os
from homeassistant import config_entries
import RFXtrx as rfxtrxmod
import serial
import serial.tools.list_ports
import voluptuous as vol
from . import DOMAIN
from homeassistant import config_entries, exceptions
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_COMMAND_OFF,
CONF_COMMAND_ON,
CONF_DEVICE,
CONF_DEVICE_ID,
CONF_DEVICES,
CONF_HOST,
CONF_PORT,
CONF_TYPE,
)
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.device_registry import (
async_entries_for_config_entry,
async_get_registry as async_get_device_registry,
)
from homeassistant.helpers.entity_registry import (
async_entries_for_device,
async_get_registry as async_get_entity_registry,
)
from . import DOMAIN, get_device_id, get_rfx_object
from .binary_sensor import supported as binary_supported
from .const import (
CONF_AUTOMATIC_ADD,
CONF_DATA_BITS,
CONF_FIRE_EVENT,
CONF_OFF_DELAY,
CONF_REMOVE_DEVICE,
CONF_REPLACE_DEVICE,
CONF_SIGNAL_REPETITIONS,
DEVICE_PACKET_TYPE_LIGHTING4,
)
from .cover import supported as cover_supported
from .light import supported as light_supported
from .switch import supported as switch_supported
_LOGGER = logging.getLogger(__name__)
CONF_EVENT_CODE = "event_code"
CONF_MANUAL_PATH = "Enter Manually"
def none_or_int(value, base):
"""Check if strin is one otherwise convert to int."""
if value is None:
return None
return int(value, base)
class OptionsFlow(config_entries.OptionsFlow):
"""Handle Rfxtrx options."""
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize rfxtrx options flow."""
self._config_entry = config_entry
self._global_options = None
self._selected_device = None
self._selected_device_entry_id = None
self._selected_device_event_code = None
self._selected_device_object = None
self._device_entries = None
self._device_registry = None
async def async_step_init(self, user_input=None):
"""Manage the options."""
return await self.async_step_prompt_options()
async def async_step_prompt_options(self, user_input=None):
"""Prompt for options."""
errors = {}
if user_input is not None:
self._global_options = {
CONF_AUTOMATIC_ADD: user_input[CONF_AUTOMATIC_ADD],
}
if CONF_DEVICE in user_input:
entry_id = user_input[CONF_DEVICE]
device_data = self._get_device_data(entry_id)
self._selected_device_entry_id = entry_id
event_code = device_data[CONF_EVENT_CODE]
self._selected_device_event_code = event_code
self._selected_device = self._config_entry.data[CONF_DEVICES][
event_code
]
self._selected_device_object = get_rfx_object(event_code)
return await self.async_step_set_device_options()
if CONF_REMOVE_DEVICE in user_input:
remove_devices = user_input[CONF_REMOVE_DEVICE]
devices = {}
for entry_id in remove_devices:
device_data = self._get_device_data(entry_id)
event_code = device_data[CONF_EVENT_CODE]
device_id = device_data[CONF_DEVICE_ID]
self.hass.helpers.dispatcher.async_dispatcher_send(
f"{DOMAIN}_{CONF_REMOVE_DEVICE}_{device_id}"
)
self._device_registry.async_remove_device(entry_id)
devices[event_code] = None
self.update_config_data(
global_options=self._global_options, devices=devices
)
return self.async_create_entry(title="", data={})
if CONF_EVENT_CODE in user_input:
self._selected_device_event_code = user_input[CONF_EVENT_CODE]
self._selected_device = {}
selected_device_object = get_rfx_object(
self._selected_device_event_code
)
if selected_device_object is None:
errors[CONF_EVENT_CODE] = "invalid_event_code"
elif not self._can_add_device(selected_device_object):
errors[CONF_EVENT_CODE] = "already_configured_device"
else:
self._selected_device_object = selected_device_object
return await self.async_step_set_device_options()
if not errors:
self.update_config_data(global_options=self._global_options)
return self.async_create_entry(title="", data={})
device_registry = await async_get_device_registry(self.hass)
device_entries = async_entries_for_config_entry(
device_registry, self._config_entry.entry_id
)
self._device_registry = device_registry
self._device_entries = device_entries
devices = {
entry.id: entry.name_by_user if entry.name_by_user else entry.name
for entry in device_entries
}
options = {
vol.Optional(
CONF_AUTOMATIC_ADD,
default=self._config_entry.data[CONF_AUTOMATIC_ADD],
): bool,
vol.Optional(CONF_EVENT_CODE): str,
vol.Optional(CONF_DEVICE): vol.In(devices),
vol.Optional(CONF_REMOVE_DEVICE): cv.multi_select(devices),
}
return self.async_show_form(
step_id="prompt_options", data_schema=vol.Schema(options), errors=errors
)
async def async_step_set_device_options(self, user_input=None):
"""Manage device options."""
errors = {}
if user_input is not None:
device_id = get_device_id(
self._selected_device_object.device,
data_bits=user_input.get(CONF_DATA_BITS),
)
if CONF_REPLACE_DEVICE in user_input:
await self._async_replace_device(user_input[CONF_REPLACE_DEVICE])
devices = {self._selected_device_event_code: None}
self.update_config_data(
global_options=self._global_options, devices=devices
)
return self.async_create_entry(title="", data={})
try:
command_on = none_or_int(user_input.get(CONF_COMMAND_ON), 16)
except ValueError:
errors[CONF_COMMAND_ON] = "invalid_input_2262_on"
try:
command_off = none_or_int(user_input.get(CONF_COMMAND_OFF), 16)
except ValueError:
errors[CONF_COMMAND_OFF] = "invalid_input_2262_off"
try:
off_delay = none_or_int(user_input.get(CONF_OFF_DELAY), 10)
except ValueError:
errors[CONF_OFF_DELAY] = "invalid_input_off_delay"
if not errors:
devices = {}
device = {
CONF_DEVICE_ID: device_id,
CONF_FIRE_EVENT: user_input.get(CONF_FIRE_EVENT, False),
CONF_SIGNAL_REPETITIONS: user_input.get(CONF_SIGNAL_REPETITIONS, 1),
}
devices[self._selected_device_event_code] = device
if off_delay:
device[CONF_OFF_DELAY] = off_delay
if user_input.get(CONF_DATA_BITS):
device[CONF_DATA_BITS] = user_input[CONF_DATA_BITS]
if command_on:
device[CONF_COMMAND_ON] = command_on
if command_off:
device[CONF_COMMAND_OFF] = command_off
self.update_config_data(
global_options=self._global_options, devices=devices
)
return self.async_create_entry(title="", data={})
device_data = self._selected_device
data_schema = {
vol.Optional(
CONF_FIRE_EVENT, default=device_data.get(CONF_FIRE_EVENT, False)
): bool,
}
if binary_supported(self._selected_device_object):
if device_data.get(CONF_OFF_DELAY):
off_delay_schema = {
vol.Optional(
CONF_OFF_DELAY,
description={"suggested_value": device_data[CONF_OFF_DELAY]},
): str,
}
else:
off_delay_schema = {
vol.Optional(CONF_OFF_DELAY): str,
}
data_schema.update(off_delay_schema)
if (
binary_supported(self._selected_device_object)
or cover_supported(self._selected_device_object)
or light_supported(self._selected_device_object)
or switch_supported(self._selected_device_object)
):
data_schema.update(
{
vol.Optional(
CONF_SIGNAL_REPETITIONS,
default=device_data.get(CONF_SIGNAL_REPETITIONS, 1),
): int,
}
)
if (
self._selected_device_object.device.packettype
== DEVICE_PACKET_TYPE_LIGHTING4
):
data_schema.update(
{
vol.Optional(
CONF_DATA_BITS, default=device_data.get(CONF_DATA_BITS, 0)
): int,
vol.Optional(
CONF_COMMAND_ON,
default=hex(device_data.get(CONF_COMMAND_ON, 0)),
): str,
vol.Optional(
CONF_COMMAND_OFF,
default=hex(device_data.get(CONF_COMMAND_OFF, 0)),
): str,
}
)
devices = {
entry.id: entry.name_by_user if entry.name_by_user else entry.name
for entry in self._device_entries
if self._can_replace_device(entry.id)
}
if devices:
data_schema.update(
{
vol.Optional(CONF_REPLACE_DEVICE): vol.In(devices),
}
)
return self.async_show_form(
step_id="set_device_options",
data_schema=vol.Schema(data_schema),
errors=errors,
)
async def _async_replace_device(self, replace_device):
"""Migrate properties of a device into another."""
device_registry = self._device_registry
old_device = self._selected_device_entry_id
old_entry = device_registry.async_get(old_device)
device_registry.async_update_device(
replace_device,
area_id=old_entry.area_id,
name_by_user=old_entry.name_by_user,
)
old_device_data = self._get_device_data(old_device)
new_device_data = self._get_device_data(replace_device)
old_device_id = "_".join(x for x in old_device_data[CONF_DEVICE_ID])
new_device_id = "_".join(x for x in new_device_data[CONF_DEVICE_ID])
entity_registry = await async_get_entity_registry(self.hass)
entity_entries = async_entries_for_device(entity_registry, old_device)
entity_migration_map = {}
for entry in entity_entries:
unique_id = entry.unique_id
new_unique_id = unique_id.replace(old_device_id, new_device_id)
new_entity_id = entity_registry.async_get_entity_id(
entry.domain, entry.platform, new_unique_id
)
if new_entity_id is not None:
entity_migration_map[new_entity_id] = entry
for entry in entity_migration_map.values():
entity_registry.async_remove(entry.entity_id)
for entity_id, entry in entity_migration_map.items():
entity_registry.async_update_entity(
entity_id,
new_entity_id=entry.entity_id,
name=entry.name,
icon=entry.icon,
)
device_registry.async_remove_device(old_device)
def _can_add_device(self, new_rfx_obj):
"""Check if device does not already exist."""
new_device_id = get_device_id(new_rfx_obj.device)
for packet_id, entity_info in self._config_entry.data[CONF_DEVICES].items():
rfx_obj = get_rfx_object(packet_id)
device_id = get_device_id(rfx_obj.device, entity_info.get(CONF_DATA_BITS))
if new_device_id == device_id:
return False
return True
def _can_replace_device(self, entry_id):
"""Check if device can be replaced with selected device."""
device_data = self._get_device_data(entry_id)
event_code = device_data[CONF_EVENT_CODE]
rfx_obj = get_rfx_object(event_code)
if (
rfx_obj.device.packettype == self._selected_device_object.device.packettype
and rfx_obj.device.subtype == self._selected_device_object.device.subtype
and self._selected_device_event_code != event_code
):
return True
return False
def _get_device_data(self, entry_id):
"""Get event code based on device identifier."""
event_code = None
device_id = None
entry = self._device_registry.async_get(entry_id)
device_id = next(iter(entry.identifiers))[1:]
for packet_id, entity_info in self._config_entry.data[CONF_DEVICES].items():
if tuple(entity_info.get(CONF_DEVICE_ID)) == device_id:
event_code = packet_id
break
data = {CONF_EVENT_CODE: event_code, CONF_DEVICE_ID: device_id}
return data
@callback
def update_config_data(self, global_options=None, devices=None):
"""Update data in ConfigEntry."""
entry_data = self._config_entry.data.copy()
entry_data[CONF_DEVICES] = copy.deepcopy(self._config_entry.data[CONF_DEVICES])
if global_options:
entry_data.update(global_options)
if devices:
for event_code, options in devices.items():
if options is None:
entry_data[CONF_DEVICES].pop(event_code)
else:
entry_data[CONF_DEVICES][event_code] = options
self.hass.config_entries.async_update_entry(self._config_entry, data=entry_data)
self.hass.async_create_task(
self.hass.config_entries.async_reload(self._config_entry.entry_id)
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for RFXCOM RFXtrx."""
@ -14,11 +406,190 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
async def async_step_user(self, user_input=None):
"""Step when user initializes a integration."""
await self.async_set_unique_id(DOMAIN)
self._abort_if_unique_id_configured()
errors = {}
if user_input is not None:
user_selection = user_input[CONF_TYPE]
if user_selection == "Serial":
return await self.async_step_setup_serial()
return await self.async_step_setup_network()
list_of_types = ["Serial", "Network"]
schema = vol.Schema({vol.Required(CONF_TYPE): vol.In(list_of_types)})
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
async def async_step_setup_network(self, user_input=None):
"""Step when setting up network configuration."""
errors = {}
if user_input is not None:
host = user_input[CONF_HOST]
port = user_input[CONF_PORT]
try:
data = await self.async_validate_rfx(host=host, port=port)
except CannotConnect:
errors["base"] = "cannot_connect"
if not errors:
return self.async_create_entry(title="RFXTRX", data=data)
schema = vol.Schema(
{vol.Required(CONF_HOST): str, vol.Required(CONF_PORT): int}
)
return self.async_show_form(
step_id="setup_network",
data_schema=schema,
errors=errors,
)
async def async_step_setup_serial(self, user_input=None):
"""Step when setting up serial configuration."""
errors = {}
if user_input is not None:
user_selection = user_input[CONF_DEVICE]
if user_selection == CONF_MANUAL_PATH:
return await self.async_step_setup_serial_manual_path()
dev_path = await self.hass.async_add_executor_job(
get_serial_by_id, user_selection
)
try:
data = await self.async_validate_rfx(device=dev_path)
except CannotConnect:
errors["base"] = "cannot_connect"
if not errors:
return self.async_create_entry(title="RFXTRX", data=data)
ports = await self.hass.async_add_executor_job(serial.tools.list_ports.comports)
list_of_ports = {}
for port in ports:
list_of_ports[
port.device
] = f"{port}, s/n: {port.serial_number or 'n/a'}" + (
f" - {port.manufacturer}" if port.manufacturer else ""
)
list_of_ports[CONF_MANUAL_PATH] = CONF_MANUAL_PATH
schema = vol.Schema({vol.Required(CONF_DEVICE): vol.In(list_of_ports)})
return self.async_show_form(
step_id="setup_serial",
data_schema=schema,
errors=errors,
)
async def async_step_setup_serial_manual_path(self, user_input=None):
"""Select path manually."""
errors = {}
if user_input is not None:
device = user_input[CONF_DEVICE]
try:
data = await self.async_validate_rfx(device=device)
except CannotConnect:
errors["base"] = "cannot_connect"
if not errors:
return self.async_create_entry(title="RFXTRX", data=data)
schema = vol.Schema({vol.Required(CONF_DEVICE): str})
return self.async_show_form(
step_id="setup_serial_manual_path",
data_schema=schema,
errors=errors,
)
async def async_step_import(self, import_config=None):
"""Handle the initial step."""
entry = await self.async_set_unique_id(DOMAIN)
if entry and import_config.items() != entry.data.items():
self.hass.config_entries.async_update_entry(entry, data=import_config)
return self.async_abort(reason="already_configured")
if entry:
if CONF_DEVICES not in entry.data:
# In version 0.113, devices key was not written to config entry. Update the entry with import data
self._abort_if_unique_id_configured(import_config)
else:
self._abort_if_unique_id_configured()
host = import_config[CONF_HOST]
port = import_config[CONF_PORT]
device = import_config[CONF_DEVICE]
try:
if host is not None:
await self.async_validate_rfx(host=host, port=port)
else:
await self.async_validate_rfx(device=device)
except CannotConnect:
return self.async_abort(reason="cannot_connect")
return self.async_create_entry(title="RFXTRX", data=import_config)
async def async_validate_rfx(self, host=None, port=None, device=None):
"""Create data for rfxtrx entry."""
success = await self.hass.async_add_executor_job(
_test_transport, host, port, device
)
if not success:
raise CannotConnect
data = {
CONF_HOST: host,
CONF_PORT: port,
CONF_DEVICE: device,
CONF_AUTOMATIC_ADD: False,
CONF_DEVICES: {},
}
return data
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
"""Get the options flow for this handler."""
return OptionsFlow(config_entry)
def _test_transport(host, port, device):
"""Construct a rfx object based on config."""
if port is not None:
try:
conn = rfxtrxmod.PyNetworkTransport((host, port))
except OSError:
return False
conn.close()
else:
try:
conn = rfxtrxmod.PySerialTransport(device)
except serial.serialutil.SerialException:
return False
if conn.serial is None:
return False
conn.close()
return True
def get_serial_by_id(dev_path: str) -> str:
"""Return a /dev/serial/by-id match for given device if available."""
by_id = "/dev/serial/by-id"
if not os.path.isdir(by_id):
return dev_path
for path in (entry.path for entry in os.scandir(by_id) if entry.is_symlink()):
if os.path.realpath(path) == dev_path:
return path
return dev_path
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""

9
homeassistant/components/rfxtrx/const.py
View file

@ -1,5 +1,14 @@
"""Constants for RFXtrx integration."""
CONF_FIRE_EVENT = "fire_event"
CONF_DATA_BITS = "data_bits"
CONF_AUTOMATIC_ADD = "automatic_add"
CONF_SIGNAL_REPETITIONS = "signal_repetitions"
CONF_DEBUG = "debug"
CONF_OFF_DELAY = "off_delay"
CONF_REMOVE_DEVICE = "remove_device"
CONF_REPLACE_DEVICE = "replace_device"
COMMAND_ON_LIST = [
"On",

8
homeassistant/components/rfxtrx/cover.py
View file

@ -20,6 +20,11 @@ from .const import COMMAND_OFF_LIST, COMMAND_ON_LIST
_LOGGER = logging.getLogger(__name__)
def supported(event):
"""Return whether an event supports cover."""
return event.device.known_to_be_rollershutter
async def async_setup_entry(
hass,
config_entry,
@ -29,9 +34,6 @@ async def async_setup_entry(
discovery_info = config_entry.data
device_ids = set()
def supported(event):
return event.device.known_to_be_rollershutter
entities = []
for packet_id, entity_info in discovery_info[CONF_DEVICES].items():
event = get_rfx_object(packet_id)

14
homeassistant/components/rfxtrx/light.py
View file

@ -28,6 +28,14 @@ _LOGGER = logging.getLogger(__name__)
SUPPORT_RFXTRX = SUPPORT_BRIGHTNESS
def supported(event):
"""Return whether an event supports light."""
return (
isinstance(event.device, rfxtrxmod.LightingDevice)
and event.device.known_to_be_dimmable
)
async def async_setup_entry(
hass,
config_entry,
@ -37,12 +45,6 @@ async def async_setup_entry(
discovery_info = config_entry.data
device_ids = set()
def supported(event):
return (
isinstance(event.device, rfxtrxmod.LightingDevice)
and event.device.known_to_be_dimmable
)
# Add switch from config file
entities = []
for packet_id, entity_info in discovery_info[CONF_DEVICES].items():

6
homeassistant/components/rfxtrx/manifest.json
View file

@ -2,7 +2,7 @@
"domain": "rfxtrx",
"name": "RFXCOM RFXtrx",
"documentation": "https://www.home-assistant.io/integrations/rfxtrx",
"requirements": ["pyRFXtrx==0.25"],
"codeowners": ["@danielhiversen", "@elupus"],
"config_flow": false
"requirements": ["pyRFXtrx==0.26"],
"codeowners": ["@danielhiversen", "@elupus", "@RobBie1221"],
"config_flow": true
}

71
homeassistant/components/rfxtrx/strings.json
View file

@ -1,9 +1,74 @@
{
"title": "Rfxtrx",
"config": {
"step": {},
"error": {},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"already_configured": "[%key:common::config_flow::abort::single_instance_allowed%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"step": {
"user": {
"data": {
"type": "Connection type"
},
"title": "Select connection type"
},
"setup_network": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]"
},
"title": "Select connection address"
},
"setup_serial": {
"data": {
"device": "Select device"
},
"title": "Device"
},
"setup_serial_manual_path": {
"data": {
"device": "[%key:common::config_flow::data::usb_path%]"
},
"title": "Path"
}
}
},
"options": {
"step": {
"prompt_options": {
"data": {
"debug": "Enable debugging",
"automatic_add": "Enable automatic add",
"event_code": "Enter event code to add",
"device": "Select device to configure",
"remove_device": "Select device to delete"
},
"title": "Rfxtrx Options"
},
"set_device_options": {
"data": {
"fire_event": "Enable device event",
"off_delay": "Off delay",
"off_delay_enabled": "Enable off delay",
"data_bit": "Number of data bits",
"command_on": "Data bits value for command on",
"command_off": "Data bits value for command off",
"signal_repetitions": "Number of signal repetitions",
"replace_device": "Select device to replace"
},
"title": "Configure device options"
}
},
"error": {
"already_configured_device": "[%key:common::config_flow::abort::already_configured_device%]",
"invalid_event_code": "Invalid event code",
"invalid_input_2262_on": "Invalid input for command on",
"invalid_input_2262_off": "Invalid input for command off",
"invalid_input_off_delay": "Invalid input for off delay",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
}
}

18
homeassistant/components/rfxtrx/switch.py
View file

@ -25,6 +25,16 @@ DATA_SWITCH = f"{DOMAIN}_switch"
_LOGGER = logging.getLogger(__name__)
def supported(event):
"""Return whether an event supports switch."""
return (
isinstance(event.device, rfxtrxmod.LightingDevice)
and not event.device.known_to_be_dimmable
and not event.device.known_to_be_rollershutter
or isinstance(event.device, rfxtrxmod.RfyDevice)
)
async def async_setup_entry(
hass,
config_entry,
@ -34,14 +44,6 @@ async def async_setup_entry(
discovery_info = config_entry.data
device_ids = set()
def supported(event):
return (
isinstance(event.device, rfxtrxmod.LightingDevice)
and not event.device.known_to_be_dimmable
and not event.device.known_to_be_rollershutter
or isinstance(event.device, rfxtrxmod.RfyDevice)
)
# Add switch from config file
entities = []
for packet_id, entity_info in discovery_info[CONF_DEVICES].items():

69
homeassistant/components/rfxtrx/translations/en.json
View file

@ -1,7 +1,74 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
"already_configured": "Already configured. Only a single configuration possible.",
"cannot_connect": "Failed to connect"
},
"error": {
"cannot_connect": "Failed to connect"
},
"step": {
"user": {
"data": {
"type": "Connection type"
},
"title": "Select connection type"
},
"setup_network": {
"data": {
"host": "Host",
"port": "Port"
},
"title": "Select connection address"
},
"setup_serial": {
"data": {
"device": "Select device"
},
"title": "Device"
},
"setup_serial_manual_path": {
"data": {
"device": "Select path"
},
"title": "Path"
}
}
},
"options": {
"step": {
"prompt_options": {
"data": {
"debug": "Enable debugging",
"automatic_add": "Enable automatic add",
"event_code": "Enter event code to add",
"device": "Select device to configure",
"remove_device": "Select device to delete"
},
"title": "Rfxtrx Options"
},
"set_device_options": {
"data": {
"fire_event": "Enable device event",
"off_delay": "Off delay",
"off_delay_enabled": "Enable off delay",
"data_bit": "Number of data bits",
"command_on": "Data bits value for command on",
"command_off": "Data bits value for command off",
"signal_repetitions": "Number of signal repetitions",
"replace_device": "Select device to replace"
},
"title": "Configure device options"
}
},
"error": {
"already_configured_device": "Device is already configured",
"invalid_event_code": "Invalid event code",
"invalid_input_2262_on": "Invalid input for command on",
"invalid_input_2262_off": "Invalid input for command off",
"invalid_input_off_delay": "Invalid input for off delay",
"unknown": "Unexpected error"
}
},
"title": "Rfxtrx"
}

1
homeassistant/generated/config_flows.py
View file

@ -150,6 +150,7 @@ FLOWS = [
"pvpc_hourly_pricing",
"rachio",
"rainmachine",
"rfxtrx",
"ring",
"risco",
"roku",

2
requirements_all.txt
View file

@ -1211,7 +1211,7 @@ pyHS100==0.3.5.1
pyMetno==0.8.1
# homeassistant.components.rfxtrx
pyRFXtrx==0.25
pyRFXtrx==0.26
# homeassistant.components.switchmate
# pySwitchmate==0.4.6

2
requirements_test_all.txt
View file

@ -592,7 +592,7 @@ pyHS100==0.3.5.1
pyMetno==0.8.1
# homeassistant.components.rfxtrx
pyRFXtrx==0.25
pyRFXtrx==0.26
# homeassistant.components.tibber
pyTibber==0.15.3

26
tests/components/rfxtrx/conftest.py
View file

@ -4,11 +4,23 @@ from datetime import timedelta
import pytest
from homeassistant.components import rfxtrx
from homeassistant.setup import async_setup_component
from homeassistant.components.rfxtrx import DOMAIN
from homeassistant.util.dt import utcnow
from tests.async_mock import patch
from tests.common import async_fire_time_changed
from tests.common import MockConfigEntry, async_fire_time_changed
def create_rfx_test_cfg(device="abcd", automatic_add=False, devices=None):
"""Create rfxtrx config entry data."""
return {
"device": device,
"host": None,
"port": None,
"automatic_add": automatic_add,
"debug": False,
"devices": devices,
}
@pytest.fixture(autouse=True, name="rfxtrx")
@ -37,12 +49,12 @@ async def rfxtrx_fixture(hass):
@pytest.fixture(name="rfxtrx_automatic")
async def rfxtrx_automatic_fixture(hass, rfxtrx):
"""Fixture that starts up with automatic additions."""
entry_data = create_rfx_test_cfg(automatic_add=True, devices={})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "automatic_add": True}},
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
yield rfxtrx

126
tests/components/rfxtrx/test_binary_sensor.py
View file

@ -1,11 +1,12 @@
"""The tests for the Rfxtrx sensor platform."""
import pytest
from homeassistant.components.rfxtrx import DOMAIN
from homeassistant.components.rfxtrx.const import ATTR_EVENT
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from tests.common import mock_restore_cache
from tests.common import MockConfigEntry, mock_restore_cache
from tests.components.rfxtrx.conftest import create_rfx_test_cfg
EVENT_SMOKE_DETECTOR_PANIC = "08200300a109000670"
EVENT_SMOKE_DETECTOR_NO_PANIC = "08200300a109000770"
@ -21,11 +22,12 @@ EVENT_AC_118CDEA_2_ON = "0b1100100118cdea02010f70"
async def test_one(hass, rfxtrx):
"""Test with 1 sensor."""
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0b1100cd0213c7f230010f71": {}}}},
)
entry_data = create_rfx_test_cfg(devices={"0b1100cd0213c7f230010f71": {}})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.ac_213c7f2_48")
@ -36,22 +38,20 @@ async def test_one(hass, rfxtrx):
async def test_one_pt2262(hass, rfxtrx):
"""Test with 1 sensor."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
entry_data = create_rfx_test_cfg(
devices={
"0913000022670e013970": {
"data_bits": 4,
"command_on": 0xE,
"command_off": 0x7,
}
},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
@ -71,19 +71,14 @@ async def test_one_pt2262(hass, rfxtrx):
async def test_pt2262_unconfigured(hass, rfxtrx):
"""Test with discovery for PT2262."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
"0913000022670e013970": {},
"09130000226707013970": {},
},
}
},
entry_data = create_rfx_test_cfg(
devices={"0913000022670e013970": {}, "09130000226707013970": {}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
@ -109,11 +104,12 @@ async def test_state_restore(hass, rfxtrx, state, event):
mock_restore_cache(hass, [State(entity_id, state, attributes={ATTR_EVENT: event})])
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0b1100cd0213c7f230010f71": {}}}},
)
entry_data = create_rfx_test_cfg(devices={"0b1100cd0213c7f230010f71": {}})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == state
@ -121,20 +117,18 @@ async def test_state_restore(hass, rfxtrx, state, event):
async def test_several(hass, rfxtrx):
"""Test with 3."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
entry_data = create_rfx_test_cfg(
devices={
"0b1100cd0213c7f230010f71": {},
"0b1100100118cdea02010f70": {},
"0b1100101118cdea02010f70": {},
},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.ac_213c7f2_48")
@ -181,16 +175,12 @@ async def test_off_delay_restore(hass, rfxtrx):
],
)
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {EVENT_AC_118CDEA_2_ON: {"off_delay": 5}},
}
},
)
entry_data = create_rfx_test_cfg(devices={EVENT_AC_118CDEA_2_ON: {"off_delay": 5}})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
@ -201,16 +191,14 @@ async def test_off_delay_restore(hass, rfxtrx):
async def test_off_delay(hass, rfxtrx, timestep):
"""Test with discovery."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {"0b1100100118cdea02010f70": {"off_delay": 5}},
}
},
entry_data = create_rfx_test_cfg(
devices={"0b1100100118cdea02010f70": {"off_delay": 5}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
@ -295,13 +283,8 @@ async def test_light(hass, rfxtrx_automatic):
async def test_pt2262_duplicate_id(hass, rfxtrx):
"""Test with 1 sensor."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
entry_data = create_rfx_test_cfg(
devices={
"0913000022670e013970": {
"data_bits": 4,
"command_on": 0xE,
@ -312,10 +295,13 @@ async def test_pt2262_duplicate_id(hass, rfxtrx):
"command_on": 0xE,
"command_off": 0x7,
},
},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()

1113
tests/components/rfxtrx/test_config_flow.py
View file

File diff suppressed because it is too large Load diff

69
tests/components/rfxtrx/test_cover.py
View file

@ -3,19 +3,23 @@ from unittest.mock import call
import pytest
from homeassistant.components.rfxtrx import DOMAIN
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from tests.common import mock_restore_cache
from tests.common import MockConfigEntry, mock_restore_cache
from tests.components.rfxtrx.conftest import create_rfx_test_cfg
async def test_one_cover(hass, rfxtrx):
"""Test with 1 cover."""
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0b1400cd0213c7f20d010f51": {}}}},
entry_data = create_rfx_test_cfg(
devices={"0b1400cd0213c7f20d010f51": {"signal_repetitions": 1}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("cover.lightwaverf_siemens_0213c7_242")
@ -57,11 +61,14 @@ async def test_state_restore(hass, rfxtrx, state):
mock_restore_cache(hass, [State(entity_id, state)])
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0b1400cd0213c7f20d010f51": {}}}},
entry_data = create_rfx_test_cfg(
devices={"0b1400cd0213c7f20d010f51": {"signal_repetitions": 1}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == state
@ -69,20 +76,18 @@ async def test_state_restore(hass, rfxtrx, state):
async def test_several_covers(hass, rfxtrx):
"""Test with 3 covers."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
"0b1400cd0213c7f20d010f51": {},
"0A1400ADF394AB010D0060": {},
"09190000009ba8010100": {},
},
entry_data = create_rfx_test_cfg(
devices={
"0b1400cd0213c7f20d010f51": {"signal_repetitions": 1},
"0A1400ADF394AB010D0060": {"signal_repetitions": 1},
"09190000009ba8010100": {"signal_repetitions": 1},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("cover.lightwaverf_siemens_0213c7_242")
@ -118,19 +123,17 @@ async def test_discover_covers(hass, rfxtrx_automatic):
async def test_duplicate_cover(hass, rfxtrx):
"""Test with 2 duplicate covers."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
"0b1400cd0213c7f20d010f51": {},
"0b1400cd0213c7f20d010f50": {},
},
entry_data = create_rfx_test_cfg(
devices={
"0b1400cd0213c7f20d010f51": {"signal_repetitions": 1},
"0b1400cd0213c7f20d010f50": {"signal_repetitions": 1},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("cover.lightwaverf_siemens_0213c7_242")

34
tests/components/rfxtrx/test_init.py
View file

@ -1,10 +1,13 @@
"""The tests for the Rfxtrx component."""
from homeassistant.components.rfxtrx import DOMAIN
from homeassistant.components.rfxtrx.const import EVENT_RFXTRX_EVENT
from homeassistant.core import callback
from homeassistant.setup import async_setup_component
from tests.async_mock import call
from tests.common import MockConfigEntry
from tests.components.rfxtrx.conftest import create_rfx_test_cfg
async def test_valid_config(hass):
@ -55,21 +58,19 @@ async def test_invalid_config(hass):
async def test_fire_event(hass, rfxtrx):
"""Test fire event."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "/dev/serial/by-id/usb"
+ "-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
"automatic_add": True,
"devices": {
entry_data = create_rfx_test_cfg(
device="/dev/serial/by-id/usb-RFXCOM_RFXtrx433_A1Y0NJGR-if00-port0",
automatic_add=True,
devices={
"0b1100cd0213c7f210010f51": {"fire_event": True},
"0716000100900970": {"fire_event": True},
},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
@ -101,16 +102,19 @@ async def test_fire_event(hass, rfxtrx):
"type_string": "Byron SX",
"id_string": "00:90",
"data": "0716000100900970",
"values": {"Sound": 9, "Battery numeric": 0, "Rssi numeric": 7},
"values": {"Command": "Chime", "Rssi numeric": 7, "Sound": 9},
},
]
async def test_send(hass, rfxtrx):
"""Test configuration."""
assert await async_setup_component(
hass, "rfxtrx", {"rfxtrx": {"device": "/dev/null"}}
)
entry_data = create_rfx_test_cfg(device="/dev/null", devices={})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(

67
tests/components/rfxtrx/test_light.py
View file

@ -4,19 +4,23 @@ from unittest.mock import call
import pytest
from homeassistant.components.light import ATTR_BRIGHTNESS
from homeassistant.components.rfxtrx import DOMAIN
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from tests.common import mock_restore_cache
from tests.common import MockConfigEntry, mock_restore_cache
from tests.components.rfxtrx.conftest import create_rfx_test_cfg
async def test_one_light(hass, rfxtrx):
"""Test with 1 light."""
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0b1100cd0213c7f210020f51": {}}}},
entry_data = create_rfx_test_cfg(
devices={"0b1100cd0213c7f210020f51": {"signal_repetitions": 1}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("light.ac_213c7f2_16")
@ -95,11 +99,14 @@ async def test_state_restore(hass, rfxtrx, state, brightness):
hass, [State(entity_id, state, attributes={ATTR_BRIGHTNESS: brightness})]
)
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0b1100cd0213c7f210020f51": {}}}},
entry_data = create_rfx_test_cfg(
devices={"0b1100cd0213c7f210020f51": {"signal_repetitions": 1}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == state
@ -108,20 +115,18 @@ async def test_state_restore(hass, rfxtrx, state, brightness):
async def test_several_lights(hass, rfxtrx):
"""Test with 3 lights."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
"0b1100cd0213c7f230020f71": {},
"0b1100100118cdea02020f70": {},
"0b1100101118cdea02050f70": {},
},
entry_data = create_rfx_test_cfg(
devices={
"0b1100cd0213c7f230020f71": {"signal_repetitions": 1},
"0b1100100118cdea02020f70": {"signal_repetitions": 1},
"0b1100101118cdea02050f70": {"signal_repetitions": 1},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
@ -160,18 +165,14 @@ async def test_several_lights(hass, rfxtrx):
@pytest.mark.parametrize("repetitions", [1, 3])
async def test_repetitions(hass, rfxtrx, repetitions):
"""Test signal repetitions."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
"0b1100cd0213c7f230020f71": {"signal_repetitions": repetitions}
},
}
},
entry_data = create_rfx_test_cfg(
devices={"0b1100cd0213c7f230020f71": {"signal_repetitions": repetitions}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(

95
tests/components/rfxtrx/test_sensor.py
View file

@ -1,19 +1,23 @@
"""The tests for the Rfxtrx sensor platform."""
import pytest
from homeassistant.components.rfxtrx import DOMAIN
from homeassistant.components.rfxtrx.const import ATTR_EVENT
from homeassistant.const import ATTR_UNIT_OF_MEASUREMENT, PERCENTAGE, TEMP_CELSIUS
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from tests.common import mock_restore_cache
from tests.common import MockConfigEntry, mock_restore_cache
from tests.components.rfxtrx.conftest import create_rfx_test_cfg
async def test_default_config(hass, rfxtrx):
"""Test with 0 sensor."""
await async_setup_component(
hass, "sensor", {"sensor": {"platform": "rfxtrx", "devices": {}}}
)
entry_data = create_rfx_test_cfg(devices={})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 0
@ -21,11 +25,12 @@ async def test_default_config(hass, rfxtrx):
async def test_one_sensor(hass, rfxtrx):
"""Test with 1 sensor."""
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0a52080705020095220269": {}}}},
)
entry_data = create_rfx_test_cfg(devices={"0a52080705020095220269": {}})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("sensor.wt260_wt260h_wt440h_wt450_wt450h_05_02_temperature")
@ -49,11 +54,12 @@ async def test_state_restore(hass, rfxtrx, state, event):
mock_restore_cache(hass, [State(entity_id, state, attributes={ATTR_EVENT: event})])
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0a520801070100b81b0279": {}}}},
)
entry_data = create_rfx_test_cfg(devices={"0a520801070100b81b0279": {}})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == state
@ -61,11 +67,12 @@ async def test_state_restore(hass, rfxtrx, state, event):
async def test_one_sensor_no_datatype(hass, rfxtrx):
"""Test with 1 sensor."""
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0a52080705020095220269": {}}}},
)
entry_data = create_rfx_test_cfg(devices={"0a52080705020095220269": {}})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
base_id = "sensor.wt260_wt260h_wt440h_wt450_wt450h_05_02"
@ -104,19 +111,17 @@ async def test_one_sensor_no_datatype(hass, rfxtrx):
async def test_several_sensors(hass, rfxtrx):
"""Test with 3 sensors."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
entry_data = create_rfx_test_cfg(
devices={
"0a52080705020095220269": {},
"0a520802060100ff0e0269": {},
},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
@ -244,19 +249,17 @@ async def test_discover_sensor(hass, rfxtrx_automatic):
async def test_update_of_sensors(hass, rfxtrx):
"""Test with 3 sensors."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
entry_data = create_rfx_test_cfg(
devices={
"0a52080705020095220269": {},
"0a520802060100ff0e0269": {},
},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
@ -290,23 +293,21 @@ async def test_update_of_sensors(hass, rfxtrx):
async def test_rssi_sensor(hass, rfxtrx):
"""Test with 1 sensor."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
entry_data = create_rfx_test_cfg(
devices={
"0913000022670e013b70": {
"data_bits": 4,
"command_on": 0xE,
"command_off": 0x7,
},
"0b1100cd0213c7f230010f71": {},
},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()

82
tests/components/rfxtrx/test_switch.py
View file

@ -5,9 +5,9 @@ import pytest
from homeassistant.components.rfxtrx import DOMAIN
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from tests.common import mock_restore_cache
from tests.common import MockConfigEntry, mock_restore_cache
from tests.components.rfxtrx.conftest import create_rfx_test_cfg
EVENT_RFY_ENABLE_SUN_AUTO = "081a00000301010113"
EVENT_RFY_DISABLE_SUN_AUTO = "081a00000301010114"
@ -15,11 +15,14 @@ EVENT_RFY_DISABLE_SUN_AUTO = "081a00000301010114"
async def test_one_switch(hass, rfxtrx):
"""Test with 1 switch."""
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0b1100cd0213c7f210010f51": {}}}},
entry_data = create_rfx_test_cfg(
devices={"0b1100cd0213c7f210010f51": {"signal_repetitions": 1}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("switch.ac_213c7f2_16")
@ -55,11 +58,14 @@ async def test_state_restore(hass, rfxtrx, state):
mock_restore_cache(hass, [State(entity_id, state)])
assert await async_setup_component(
hass,
"rfxtrx",
{"rfxtrx": {"device": "abcd", "devices": {"0b1100cd0213c7f210010f51": {}}}},
entry_data = create_rfx_test_cfg(
devices={"0b1100cd0213c7f210010f51": {"signal_repetitions": 1}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == state
@ -67,20 +73,18 @@ async def test_state_restore(hass, rfxtrx, state):
async def test_several_switches(hass, rfxtrx):
"""Test with 3 switches."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
"0b1100cd0213c7f230010f71": {},
"0b1100100118cdea02010f70": {},
"0b1100101118cdea02010f70": {},
},
entry_data = create_rfx_test_cfg(
devices={
"0b1100cd0213c7f230010f71": {"signal_repetitions": 1},
"0b1100100118cdea02010f70": {"signal_repetitions": 1},
"0b1100101118cdea02010f70": {"signal_repetitions": 1},
}
},
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("switch.ac_213c7f2_48")
@ -102,18 +106,14 @@ async def test_several_switches(hass, rfxtrx):
@pytest.mark.parametrize("repetitions", [1, 3])
async def test_repetitions(hass, rfxtrx, repetitions):
"""Test signal repetitions."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {
"0b1100cd0213c7f230010f71": {"signal_repetitions": repetitions}
},
}
},
entry_data = create_rfx_test_cfg(
devices={"0b1100cd0213c7f230010f71": {"signal_repetitions": repetitions}}
)
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
@ -156,16 +156,12 @@ async def test_discover_rfy_sun_switch(hass, rfxtrx_automatic):
async def test_unknown_event_code(hass, rfxtrx):
"""Test with 3 switches."""
assert await async_setup_component(
hass,
"rfxtrx",
{
"rfxtrx": {
"device": "abcd",
"devices": {"1234567890": {}},
}
},
)
entry_data = create_rfx_test_cfg(devices={"1234567890": {"signal_repetitions": 1}})
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
conf_entries = hass.config_entries.async_entries(DOMAIN)