mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 08:47:57 +00:00
LiteJet is now configured using config_flow (#44409)
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Jon Caruana
GitHub
parent
ffe42e150a
commit
6a8b5ee51b
@ -253,6 +253,7 @@ homeassistant/components/launch_library/* @ludeeus
|
||||
homeassistant/components/lcn/* @alengwenus
|
||||
homeassistant/components/life360/* @pnbruckner
|
||||
homeassistant/components/linux_battery/* @fabaff
|
||||
homeassistant/components/litejet/* @joncar
|
||||
homeassistant/components/litterrobot/* @natekspencer
|
||||
homeassistant/components/local_ip/* @issacg
|
||||
homeassistant/components/logger/* @home-assistant/core
|
||||
|
||||
@ -1,49 +1,86 @@
|
||||
"""Support for the LiteJet lighting system."""
|
||||
from pylitejet import LiteJet
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
import pylitejet
|
||||
from serial import SerialException
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import CONF_PORT
|
||||
from homeassistant.helpers import discovery
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
CONF_EXCLUDE_NAMES = "exclude_names"
|
||||
CONF_INCLUDE_SWITCHES = "include_switches"
|
||||
from .const import CONF_EXCLUDE_NAMES, CONF_INCLUDE_SWITCHES, DOMAIN, PLATFORMS
|
||||
|
||||
DOMAIN = "litejet"
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_PORT): cv.string,
|
||||
vol.Optional(CONF_EXCLUDE_NAMES): vol.All(cv.ensure_list, [cv.string]),
|
||||
vol.Optional(CONF_INCLUDE_SWITCHES, default=False): cv.boolean,
|
||||
}
|
||||
)
|
||||
},
|
||||
vol.All(
|
||||
cv.deprecated(DOMAIN),
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_PORT): cv.string,
|
||||
vol.Optional(CONF_EXCLUDE_NAMES): vol.All(
|
||||
cv.ensure_list, [cv.string]
|
||||
),
|
||||
vol.Optional(CONF_INCLUDE_SWITCHES, default=False): cv.boolean,
|
||||
}
|
||||
)
|
||||
},
|
||||
),
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
|
||||
def setup(hass, config):
|
||||
"""Set up the LiteJet component."""
|
||||
if DOMAIN in config and not hass.config_entries.async_entries(DOMAIN):
|
||||
# No config entry exists and configuration.yaml config exists, trigger the import flow.
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=config[DOMAIN]
|
||||
)
|
||||
)
|
||||
return True
|
||||
|
||||
url = config[DOMAIN].get(CONF_PORT)
|
||||
|
||||
hass.data["litejet_system"] = LiteJet(url)
|
||||
hass.data["litejet_config"] = config[DOMAIN]
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up LiteJet via a config entry."""
|
||||
port = entry.data[CONF_PORT]
|
||||
|
||||
discovery.load_platform(hass, "light", DOMAIN, {}, config)
|
||||
if config[DOMAIN].get(CONF_INCLUDE_SWITCHES):
|
||||
discovery.load_platform(hass, "switch", DOMAIN, {}, config)
|
||||
discovery.load_platform(hass, "scene", DOMAIN, {}, config)
|
||||
try:
|
||||
system = pylitejet.LiteJet(port)
|
||||
except SerialException as ex:
|
||||
_LOGGER.error("Error connecting to the LiteJet MCP at %s", port, exc_info=ex)
|
||||
raise ConfigEntryNotReady from ex
|
||||
|
||||
hass.data[DOMAIN] = system
|
||||
|
||||
for component in PLATFORMS:
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(entry, component)
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def is_ignored(hass, name):
|
||||
"""Determine if a load, switch, or scene should be ignored."""
|
||||
for prefix in hass.data["litejet_config"].get(CONF_EXCLUDE_NAMES, []):
|
||||
if name.startswith(prefix):
|
||||
return True
|
||||
return False
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
|
||||
"""Unload a LiteJet config entry."""
|
||||
|
||||
unload_ok = all(
|
||||
await asyncio.gather(
|
||||
*[
|
||||
hass.config_entries.async_forward_entry_unload(entry, component)
|
||||
for component in PLATFORMS
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].close()
|
||||
hass.data.pop(DOMAIN)
|
||||
|
||||
return unload_ok
|
||||
|
||||
53
homeassistant/components/litejet/config_flow.py
Normal file
53
homeassistant/components/litejet/config_flow.py
Normal file
@ -0,0 +1,53 @@
|
||||
"""Config flow for the LiteJet lighting system."""
|
||||
import logging
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
import pylitejet
|
||||
from serial import SerialException
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_PORT
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class LiteJetConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""LiteJet config flow."""
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: Optional[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Create a LiteJet config entry based upon user input."""
|
||||
if self.hass.config_entries.async_entries(DOMAIN):
|
||||
return self.async_abort(reason="single_instance_allowed")
|
||||
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
port = user_input[CONF_PORT]
|
||||
|
||||
await self.async_set_unique_id(port)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
try:
|
||||
system = pylitejet.LiteJet(port)
|
||||
system.close()
|
||||
except SerialException:
|
||||
errors[CONF_PORT] = "open_failed"
|
||||
else:
|
||||
return self.async_create_entry(
|
||||
title=port,
|
||||
data={CONF_PORT: port},
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema({vol.Required(CONF_PORT): str}),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_import(self, import_data):
|
||||
"""Import litejet config from configuration.yaml."""
|
||||
return self.async_create_entry(title=import_data[CONF_PORT], data=import_data)
|
||||
8
homeassistant/components/litejet/const.py
Normal file
8
homeassistant/components/litejet/const.py
Normal file
@ -0,0 +1,8 @@
|
||||
"""LiteJet constants."""
|
||||
|
||||
DOMAIN = "litejet"
|
||||
|
||||
CONF_EXCLUDE_NAMES = "exclude_names"
|
||||
CONF_INCLUDE_SWITCHES = "include_switches"
|
||||
|
||||
PLATFORMS = ["light", "switch", "scene"]
|
||||
@ -1,43 +1,53 @@
|
||||
"""Support for LiteJet lights."""
|
||||
import logging
|
||||
|
||||
from homeassistant.components import litejet
|
||||
from homeassistant.components.light import (
|
||||
ATTR_BRIGHTNESS,
|
||||
SUPPORT_BRIGHTNESS,
|
||||
LightEntity,
|
||||
)
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTR_NUMBER = "number"
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up lights for the LiteJet platform."""
|
||||
litejet_ = hass.data["litejet_system"]
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up entry."""
|
||||
|
||||
devices = []
|
||||
for i in litejet_.loads():
|
||||
name = litejet_.get_load_name(i)
|
||||
if not litejet.is_ignored(hass, name):
|
||||
devices.append(LiteJetLight(hass, litejet_, i, name))
|
||||
add_entities(devices, True)
|
||||
system = hass.data[DOMAIN]
|
||||
|
||||
def get_entities(system):
|
||||
entities = []
|
||||
for i in system.loads():
|
||||
name = system.get_load_name(i)
|
||||
entities.append(LiteJetLight(config_entry.entry_id, system, i, name))
|
||||
return entities
|
||||
|
||||
async_add_entities(await hass.async_add_executor_job(get_entities, system), True)
|
||||
|
||||
|
||||
class LiteJetLight(LightEntity):
|
||||
"""Representation of a single LiteJet light."""
|
||||
|
||||
def __init__(self, hass, lj, i, name):
|
||||
def __init__(self, entry_id, lj, i, name):
|
||||
"""Initialize a LiteJet light."""
|
||||
self._hass = hass
|
||||
self._entry_id = entry_id
|
||||
self._lj = lj
|
||||
self._index = i
|
||||
self._brightness = 0
|
||||
self._name = name
|
||||
|
||||
lj.on_load_activated(i, self._on_load_changed)
|
||||
lj.on_load_deactivated(i, self._on_load_changed)
|
||||
async def async_added_to_hass(self):
|
||||
"""Run when this Entity has been added to HA."""
|
||||
self._lj.on_load_activated(self._index, self._on_load_changed)
|
||||
self._lj.on_load_deactivated(self._index, self._on_load_changed)
|
||||
|
||||
async def async_will_remove_from_hass(self):
|
||||
"""Entity being removed from hass."""
|
||||
self._lj.unsubscribe(self._on_load_changed)
|
||||
|
||||
def _on_load_changed(self):
|
||||
"""Handle state changes."""
|
||||
@ -54,6 +64,11 @@ class LiteJetLight(LightEntity):
|
||||
"""Return the light's name."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return a unique identifier for this light."""
|
||||
return f"{self._entry_id}_{self._index}"
|
||||
|
||||
@property
|
||||
def brightness(self):
|
||||
"""Return the light's brightness."""
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
"domain": "litejet",
|
||||
"name": "LiteJet",
|
||||
"documentation": "https://www.home-assistant.io/integrations/litejet",
|
||||
"requirements": ["pylitejet==0.1"],
|
||||
"codeowners": []
|
||||
"requirements": ["pylitejet==0.3.0"],
|
||||
"codeowners": ["@joncar"],
|
||||
"config_flow": true
|
||||
}
|
||||
|
||||
@ -1,29 +1,37 @@
|
||||
"""Support for LiteJet scenes."""
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components import litejet
|
||||
from homeassistant.components.scene import Scene
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTR_NUMBER = "number"
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up scenes for the LiteJet platform."""
|
||||
litejet_ = hass.data["litejet_system"]
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up entry."""
|
||||
|
||||
devices = []
|
||||
for i in litejet_.scenes():
|
||||
name = litejet_.get_scene_name(i)
|
||||
if not litejet.is_ignored(hass, name):
|
||||
devices.append(LiteJetScene(litejet_, i, name))
|
||||
add_entities(devices)
|
||||
system = hass.data[DOMAIN]
|
||||
|
||||
def get_entities(system):
|
||||
entities = []
|
||||
for i in system.scenes():
|
||||
name = system.get_scene_name(i)
|
||||
entities.append(LiteJetScene(config_entry.entry_id, system, i, name))
|
||||
return entities
|
||||
|
||||
async_add_entities(await hass.async_add_executor_job(get_entities, system), True)
|
||||
|
||||
|
||||
class LiteJetScene(Scene):
|
||||
"""Representation of a single LiteJet scene."""
|
||||
|
||||
def __init__(self, lj, i, name):
|
||||
def __init__(self, entry_id, lj, i, name):
|
||||
"""Initialize the scene."""
|
||||
self._entry_id = entry_id
|
||||
self._lj = lj
|
||||
self._index = i
|
||||
self._name = name
|
||||
@ -33,6 +41,11 @@ class LiteJetScene(Scene):
|
||||
"""Return the name of the scene."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return a unique identifier for this scene."""
|
||||
return f"{self._entry_id}_{self._index}"
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the device-specific state attributes."""
|
||||
@ -41,3 +54,8 @@ class LiteJetScene(Scene):
|
||||
def activate(self, **kwargs: Any) -> None:
|
||||
"""Activate the scene."""
|
||||
self._lj.activate_scene(self._index)
|
||||
|
||||
@property
|
||||
def entity_registry_enabled_default(self) -> bool:
|
||||
"""Scenes are only enabled by explicit user choice."""
|
||||
return False
|
||||
|
||||
19
homeassistant/components/litejet/strings.json
Normal file
19
homeassistant/components/litejet/strings.json
Normal file
@ -0,0 +1,19 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Connect To LiteJet",
|
||||
"description": "Connect the LiteJet's RS232-2 port to your computer and enter the path to the serial port device.\n\nThe LiteJet MCP must be configured for 19.2 K baud, 8 data bits, 1 stop bit, no parity, and to transmit a 'CR' after each response.",
|
||||
"data": {
|
||||
"port": "[%key:common::config_flow::data::port%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"abort": {
|
||||
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]"
|
||||
},
|
||||
"error": {
|
||||
"open_failed": "Cannot open the specified serial port."
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,39 +1,50 @@
|
||||
"""Support for LiteJet switch."""
|
||||
import logging
|
||||
|
||||
from homeassistant.components import litejet
|
||||
from homeassistant.components.switch import SwitchEntity
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
ATTR_NUMBER = "number"
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the LiteJet switch platform."""
|
||||
litejet_ = hass.data["litejet_system"]
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up entry."""
|
||||
|
||||
devices = []
|
||||
for i in litejet_.button_switches():
|
||||
name = litejet_.get_switch_name(i)
|
||||
if not litejet.is_ignored(hass, name):
|
||||
devices.append(LiteJetSwitch(hass, litejet_, i, name))
|
||||
add_entities(devices, True)
|
||||
system = hass.data[DOMAIN]
|
||||
|
||||
def get_entities(system):
|
||||
entities = []
|
||||
for i in system.button_switches():
|
||||
name = system.get_switch_name(i)
|
||||
entities.append(LiteJetSwitch(config_entry.entry_id, system, i, name))
|
||||
return entities
|
||||
|
||||
async_add_entities(await hass.async_add_executor_job(get_entities, system), True)
|
||||
|
||||
|
||||
class LiteJetSwitch(SwitchEntity):
|
||||
"""Representation of a single LiteJet switch."""
|
||||
|
||||
def __init__(self, hass, lj, i, name):
|
||||
def __init__(self, entry_id, lj, i, name):
|
||||
"""Initialize a LiteJet switch."""
|
||||
self._hass = hass
|
||||
self._entry_id = entry_id
|
||||
self._lj = lj
|
||||
self._index = i
|
||||
self._state = False
|
||||
self._name = name
|
||||
|
||||
lj.on_switch_pressed(i, self._on_switch_pressed)
|
||||
lj.on_switch_released(i, self._on_switch_released)
|
||||
async def async_added_to_hass(self):
|
||||
"""Run when this Entity has been added to HA."""
|
||||
self._lj.on_switch_pressed(self._index, self._on_switch_pressed)
|
||||
self._lj.on_switch_released(self._index, self._on_switch_released)
|
||||
|
||||
async def async_will_remove_from_hass(self):
|
||||
"""Entity being removed from hass."""
|
||||
self._lj.unsubscribe(self._on_switch_pressed)
|
||||
self._lj.unsubscribe(self._on_switch_released)
|
||||
|
||||
def _on_switch_pressed(self):
|
||||
_LOGGER.debug("Updating pressed for %s", self._name)
|
||||
@ -50,6 +61,11 @@ class LiteJetSwitch(SwitchEntity):
|
||||
"""Return the name of the switch."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return a unique identifier for this switch."""
|
||||
return f"{self._entry_id}_{self._index}"
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return if the switch is pressed."""
|
||||
@ -72,3 +88,8 @@ class LiteJetSwitch(SwitchEntity):
|
||||
def turn_off(self, **kwargs):
|
||||
"""Release the switch."""
|
||||
self._lj.release_switch(self._index)
|
||||
|
||||
@property
|
||||
def entity_registry_enabled_default(self) -> bool:
|
||||
"""Switches are only enabled by explicit user choice."""
|
||||
return False
|
||||
|
||||
19
homeassistant/components/litejet/translations/en.json
Normal file
19
homeassistant/components/litejet/translations/en.json
Normal file
@ -0,0 +1,19 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"single_instance_allowed": "Already configured. Only a single configuration possible."
|
||||
},
|
||||
"error": {
|
||||
"open_failed": "Cannot open the specified serial port."
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"port": "Port"
|
||||
},
|
||||
"description": "Connect the LiteJet's RS232-2 port to your computer and enter the path to the serial port device.\n\nThe LiteJet MCP must be configured for 19.2 K baud, 8 data bits, 1 stop bit, no parity, and to transmit a 'CR' after each response.",
|
||||
"title": "Connect To LiteJet"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,4 +1,6 @@
|
||||
"""Trigger an automation when a LiteJet switch is released."""
|
||||
from typing import Callable
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.const import CONF_PLATFORM
|
||||
@ -7,7 +9,7 @@ import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.event import track_point_in_utc_time
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
# mypy: allow-untyped-defs, no-check-untyped-defs
|
||||
from .const import DOMAIN
|
||||
|
||||
CONF_NUMBER = "number"
|
||||
CONF_HELD_MORE_THAN = "held_more_than"
|
||||
@ -33,7 +35,7 @@ async def async_attach_trigger(hass, config, action, automation_info):
|
||||
held_more_than = config.get(CONF_HELD_MORE_THAN)
|
||||
held_less_than = config.get(CONF_HELD_LESS_THAN)
|
||||
pressed_time = None
|
||||
cancel_pressed_more_than = None
|
||||
cancel_pressed_more_than: Callable = None
|
||||
job = HassJob(action)
|
||||
|
||||
@callback
|
||||
@ -91,12 +93,15 @@ async def async_attach_trigger(hass, config, action, automation_info):
|
||||
):
|
||||
hass.add_job(call_action)
|
||||
|
||||
hass.data["litejet_system"].on_switch_pressed(number, pressed)
|
||||
hass.data["litejet_system"].on_switch_released(number, released)
|
||||
system = hass.data[DOMAIN]
|
||||
|
||||
system.on_switch_pressed(number, pressed)
|
||||
system.on_switch_released(number, released)
|
||||
|
||||
@callback
|
||||
def async_remove():
|
||||
"""Remove all subscriptions used for this trigger."""
|
||||
return
|
||||
system.unsubscribe(pressed)
|
||||
system.unsubscribe(released)
|
||||
|
||||
return async_remove
|
||||
|
||||
@ -121,6 +121,7 @@ FLOWS = [
|
||||
"kulersky",
|
||||
"life360",
|
||||
"lifx",
|
||||
"litejet",
|
||||
"litterrobot",
|
||||
"local_ip",
|
||||
"locative",
|
||||
|
||||
@ -1501,7 +1501,7 @@ pylgnetcast-homeassistant==0.2.0.dev0
|
||||
pylibrespot-java==0.1.0
|
||||
|
||||
# homeassistant.components.litejet
|
||||
pylitejet==0.1
|
||||
pylitejet==0.3.0
|
||||
|
||||
# homeassistant.components.litterrobot
|
||||
pylitterbot==2021.2.5
|
||||
|
||||
@ -791,7 +791,7 @@ pylast==4.1.0
|
||||
pylibrespot-java==0.1.0
|
||||
|
||||
# homeassistant.components.litejet
|
||||
pylitejet==0.1
|
||||
pylitejet==0.3.0
|
||||
|
||||
# homeassistant.components.litterrobot
|
||||
pylitterbot==2021.2.5
|
||||
|
||||
@ -1 +1,51 @@
|
||||
"""Tests for the litejet component."""
|
||||
from homeassistant.components import scene, switch
|
||||
from homeassistant.components.litejet import DOMAIN
|
||||
from homeassistant.const import CONF_PORT
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def async_init_integration(
|
||||
hass, use_switch=False, use_scene=False
|
||||
) -> MockConfigEntry:
|
||||
"""Set up the LiteJet integration in Home Assistant."""
|
||||
|
||||
registry = await hass.helpers.entity_registry.async_get_registry()
|
||||
|
||||
entry_data = {CONF_PORT: "/dev/mock"}
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN, unique_id=entry_data[CONF_PORT], data=entry_data
|
||||
)
|
||||
|
||||
if use_switch:
|
||||
registry.async_get_or_create(
|
||||
switch.DOMAIN,
|
||||
DOMAIN,
|
||||
f"{entry.entry_id}_1",
|
||||
suggested_object_id="mock_switch_1",
|
||||
disabled_by=None,
|
||||
)
|
||||
registry.async_get_or_create(
|
||||
switch.DOMAIN,
|
||||
DOMAIN,
|
||||
f"{entry.entry_id}_2",
|
||||
suggested_object_id="mock_switch_2",
|
||||
disabled_by=None,
|
||||
)
|
||||
|
||||
if use_scene:
|
||||
registry.async_get_or_create(
|
||||
scene.DOMAIN,
|
||||
DOMAIN,
|
||||
f"{entry.entry_id}_1",
|
||||
suggested_object_id="mock_scene_1",
|
||||
disabled_by=None,
|
||||
)
|
||||
|
||||
entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
return entry
|
||||
|
||||
@ -1,2 +1,62 @@
|
||||
"""litejet conftest."""
|
||||
from tests.components.light.conftest import mock_light_profiles # noqa
|
||||
"""Fixtures for LiteJet testing."""
|
||||
from datetime import timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_litejet():
|
||||
"""Mock LiteJet system."""
|
||||
with patch("pylitejet.LiteJet") as mock_pylitejet:
|
||||
|
||||
def get_load_name(number):
|
||||
return f"Mock Load #{number}"
|
||||
|
||||
def get_scene_name(number):
|
||||
return f"Mock Scene #{number}"
|
||||
|
||||
def get_switch_name(number):
|
||||
return f"Mock Switch #{number}"
|
||||
|
||||
mock_lj = mock_pylitejet.return_value
|
||||
|
||||
mock_lj.switch_pressed_callbacks = {}
|
||||
mock_lj.switch_released_callbacks = {}
|
||||
mock_lj.load_activated_callbacks = {}
|
||||
mock_lj.load_deactivated_callbacks = {}
|
||||
|
||||
def on_switch_pressed(number, callback):
|
||||
mock_lj.switch_pressed_callbacks[number] = callback
|
||||
|
||||
def on_switch_released(number, callback):
|
||||
mock_lj.switch_released_callbacks[number] = callback
|
||||
|
||||
def on_load_activated(number, callback):
|
||||
mock_lj.load_activated_callbacks[number] = callback
|
||||
|
||||
def on_load_deactivated(number, callback):
|
||||
mock_lj.load_deactivated_callbacks[number] = callback
|
||||
|
||||
mock_lj.on_switch_pressed.side_effect = on_switch_pressed
|
||||
mock_lj.on_switch_released.side_effect = on_switch_released
|
||||
mock_lj.on_load_activated.side_effect = on_load_activated
|
||||
mock_lj.on_load_deactivated.side_effect = on_load_deactivated
|
||||
|
||||
mock_lj.loads.return_value = range(1, 3)
|
||||
mock_lj.get_load_name.side_effect = get_load_name
|
||||
mock_lj.get_load_level.return_value = 0
|
||||
|
||||
mock_lj.button_switches.return_value = range(1, 3)
|
||||
mock_lj.all_switches.return_value = range(1, 6)
|
||||
mock_lj.get_switch_name.side_effect = get_switch_name
|
||||
|
||||
mock_lj.scenes.return_value = range(1, 3)
|
||||
mock_lj.get_scene_name.side_effect = get_scene_name
|
||||
|
||||
mock_lj.start_time = dt_util.utcnow()
|
||||
mock_lj.last_delta = timedelta(0)
|
||||
|
||||
yield mock_lj
|
||||
|
||||
77
tests/components/litejet/test_config_flow.py
Normal file
77
tests/components/litejet/test_config_flow.py
Normal file
@ -0,0 +1,77 @@
|
||||
"""The tests for the litejet component."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from serial import SerialException
|
||||
|
||||
from homeassistant.components.litejet.const import DOMAIN
|
||||
from homeassistant.const import CONF_PORT
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_show_config_form(hass):
|
||||
"""Test show configuration form."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "user"}
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
|
||||
async def test_create_entry(hass, mock_litejet):
|
||||
"""Test create entry from user input."""
|
||||
test_data = {CONF_PORT: "/dev/test"}
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "user"}, data=test_data
|
||||
)
|
||||
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == "/dev/test"
|
||||
assert result["data"] == test_data
|
||||
|
||||
|
||||
async def test_flow_entry_already_exists(hass):
|
||||
"""Test user input when a config entry already exists."""
|
||||
first_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={CONF_PORT: "/dev/first"},
|
||||
)
|
||||
first_entry.add_to_hass(hass)
|
||||
|
||||
test_data = {CONF_PORT: "/dev/test"}
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "user"}, data=test_data
|
||||
)
|
||||
|
||||
assert result["type"] == "abort"
|
||||
assert result["reason"] == "single_instance_allowed"
|
||||
|
||||
|
||||
async def test_flow_open_failed(hass):
|
||||
"""Test user input when serial port open fails."""
|
||||
test_data = {CONF_PORT: "/dev/test"}
|
||||
|
||||
with patch("pylitejet.LiteJet") as mock_pylitejet:
|
||||
mock_pylitejet.side_effect = SerialException
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "user"}, data=test_data
|
||||
)
|
||||
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"][CONF_PORT] == "open_failed"
|
||||
|
||||
|
||||
async def test_import_step(hass):
|
||||
"""Test initializing via import step."""
|
||||
test_data = {CONF_PORT: "/dev/imported"}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "import"}, data=test_data
|
||||
)
|
||||
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == test_data[CONF_PORT]
|
||||
assert result["data"] == test_data
|
||||
@ -1,41 +1,30 @@
|
||||
"""The tests for the litejet component."""
|
||||
import unittest
|
||||
|
||||
from homeassistant.components import litejet
|
||||
from homeassistant.components.litejet.const import DOMAIN
|
||||
from homeassistant.const import CONF_PORT
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import get_test_home_assistant
|
||||
from . import async_init_integration
|
||||
|
||||
|
||||
class TestLiteJet(unittest.TestCase):
|
||||
"""Test the litejet component."""
|
||||
async def test_setup_with_no_config(hass):
|
||||
"""Test that nothing happens."""
|
||||
assert await async_setup_component(hass, DOMAIN, {}) is True
|
||||
assert DOMAIN not in hass.data
|
||||
|
||||
def setup_method(self, method):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.hass.start()
|
||||
self.hass.block_till_done()
|
||||
|
||||
def teardown_method(self, method):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
async def test_setup_with_config_to_import(hass, mock_litejet):
|
||||
"""Test that import happens."""
|
||||
assert (
|
||||
await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PORT: "/dev/hello"}})
|
||||
is True
|
||||
)
|
||||
assert DOMAIN in hass.data
|
||||
|
||||
def test_is_ignored_unspecified(self):
|
||||
"""Ensure it is ignored when unspecified."""
|
||||
self.hass.data["litejet_config"] = {}
|
||||
assert not litejet.is_ignored(self.hass, "Test")
|
||||
|
||||
def test_is_ignored_empty(self):
|
||||
"""Ensure it is ignored when empty."""
|
||||
self.hass.data["litejet_config"] = {litejet.CONF_EXCLUDE_NAMES: []}
|
||||
assert not litejet.is_ignored(self.hass, "Test")
|
||||
async def test_unload_entry(hass, mock_litejet):
|
||||
"""Test being able to unload an entry."""
|
||||
entry = await async_init_integration(hass, use_switch=True, use_scene=True)
|
||||
|
||||
def test_is_ignored_normal(self):
|
||||
"""Test if usually ignored."""
|
||||
self.hass.data["litejet_config"] = {
|
||||
litejet.CONF_EXCLUDE_NAMES: ["Test", "Other One"]
|
||||
}
|
||||
assert litejet.is_ignored(self.hass, "Test")
|
||||
assert not litejet.is_ignored(self.hass, "Other one")
|
||||
assert not litejet.is_ignored(self.hass, "Other 0ne")
|
||||
assert litejet.is_ignored(self.hass, "Other One There")
|
||||
assert litejet.is_ignored(self.hass, "Other One")
|
||||
assert await litejet.async_unload_entry(hass, entry)
|
||||
assert DOMAIN not in hass.data
|
||||
|
||||
@ -1,14 +1,11 @@
|
||||
"""The tests for the litejet component."""
|
||||
import logging
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from homeassistant import setup
|
||||
from homeassistant.components import litejet
|
||||
import homeassistant.components.light as light
|
||||
from homeassistant.components import light
|
||||
from homeassistant.components.light import ATTR_BRIGHTNESS
|
||||
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF, SERVICE_TURN_ON
|
||||
|
||||
from tests.common import get_test_home_assistant
|
||||
from tests.components.light import common
|
||||
from . import async_init_integration
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@ -18,144 +15,113 @@ ENTITY_OTHER_LIGHT = "light.mock_load_2"
|
||||
ENTITY_OTHER_LIGHT_NUMBER = 2
|
||||
|
||||
|
||||
class TestLiteJetLight(unittest.TestCase):
|
||||
"""Test the litejet component."""
|
||||
async def test_on_brightness(hass, mock_litejet):
|
||||
"""Test turning the light on with brightness."""
|
||||
await async_init_integration(hass)
|
||||
|
||||
@mock.patch("homeassistant.components.litejet.LiteJet")
|
||||
def setup_method(self, method, mock_pylitejet):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.hass.start()
|
||||
assert hass.states.get(ENTITY_LIGHT).state == "off"
|
||||
assert hass.states.get(ENTITY_OTHER_LIGHT).state == "off"
|
||||
|
||||
self.load_activated_callbacks = {}
|
||||
self.load_deactivated_callbacks = {}
|
||||
assert not light.is_on(hass, ENTITY_LIGHT)
|
||||
|
||||
def get_load_name(number):
|
||||
return f"Mock Load #{number}"
|
||||
await hass.services.async_call(
|
||||
light.DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: ENTITY_LIGHT, ATTR_BRIGHTNESS: 102},
|
||||
blocking=True,
|
||||
)
|
||||
mock_litejet.activate_load_at.assert_called_with(ENTITY_LIGHT_NUMBER, 39, 0)
|
||||
|
||||
def on_load_activated(number, callback):
|
||||
self.load_activated_callbacks[number] = callback
|
||||
|
||||
def on_load_deactivated(number, callback):
|
||||
self.load_deactivated_callbacks[number] = callback
|
||||
async def test_on_off(hass, mock_litejet):
|
||||
"""Test turning the light on and off."""
|
||||
await async_init_integration(hass)
|
||||
|
||||
self.mock_lj = mock_pylitejet.return_value
|
||||
self.mock_lj.loads.return_value = range(1, 3)
|
||||
self.mock_lj.button_switches.return_value = range(0)
|
||||
self.mock_lj.all_switches.return_value = range(0)
|
||||
self.mock_lj.scenes.return_value = range(0)
|
||||
self.mock_lj.get_load_level.return_value = 0
|
||||
self.mock_lj.get_load_name.side_effect = get_load_name
|
||||
self.mock_lj.on_load_activated.side_effect = on_load_activated
|
||||
self.mock_lj.on_load_deactivated.side_effect = on_load_deactivated
|
||||
assert hass.states.get(ENTITY_LIGHT).state == "off"
|
||||
assert hass.states.get(ENTITY_OTHER_LIGHT).state == "off"
|
||||
|
||||
assert setup.setup_component(
|
||||
self.hass,
|
||||
litejet.DOMAIN,
|
||||
{"litejet": {"port": "/dev/serial/by-id/mock-litejet"}},
|
||||
)
|
||||
self.hass.block_till_done()
|
||||
assert not light.is_on(hass, ENTITY_LIGHT)
|
||||
|
||||
self.mock_lj.get_load_level.reset_mock()
|
||||
await hass.services.async_call(
|
||||
light.DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: ENTITY_LIGHT},
|
||||
blocking=True,
|
||||
)
|
||||
mock_litejet.activate_load.assert_called_with(ENTITY_LIGHT_NUMBER)
|
||||
|
||||
def light(self):
|
||||
"""Test for main light entity."""
|
||||
return self.hass.states.get(ENTITY_LIGHT)
|
||||
await hass.services.async_call(
|
||||
light.DOMAIN,
|
||||
SERVICE_TURN_OFF,
|
||||
{ATTR_ENTITY_ID: ENTITY_LIGHT},
|
||||
blocking=True,
|
||||
)
|
||||
mock_litejet.deactivate_load.assert_called_with(ENTITY_LIGHT_NUMBER)
|
||||
|
||||
def other_light(self):
|
||||
"""Test the other light."""
|
||||
return self.hass.states.get(ENTITY_OTHER_LIGHT)
|
||||
|
||||
def teardown_method(self, method):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
async def test_activated_event(hass, mock_litejet):
|
||||
"""Test handling an event from LiteJet."""
|
||||
|
||||
def test_on_brightness(self):
|
||||
"""Test turning the light on with brightness."""
|
||||
assert self.light().state == "off"
|
||||
assert self.other_light().state == "off"
|
||||
await async_init_integration(hass)
|
||||
|
||||
assert not light.is_on(self.hass, ENTITY_LIGHT)
|
||||
# Light 1
|
||||
mock_litejet.get_load_level.return_value = 99
|
||||
mock_litejet.get_load_level.reset_mock()
|
||||
mock_litejet.load_activated_callbacks[ENTITY_LIGHT_NUMBER]()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
common.turn_on(self.hass, ENTITY_LIGHT, brightness=102)
|
||||
self.hass.block_till_done()
|
||||
self.mock_lj.activate_load_at.assert_called_with(ENTITY_LIGHT_NUMBER, 39, 0)
|
||||
mock_litejet.get_load_level.assert_called_once_with(ENTITY_LIGHT_NUMBER)
|
||||
|
||||
def test_on_off(self):
|
||||
"""Test turning the light on and off."""
|
||||
assert self.light().state == "off"
|
||||
assert self.other_light().state == "off"
|
||||
assert light.is_on(hass, ENTITY_LIGHT)
|
||||
assert not light.is_on(hass, ENTITY_OTHER_LIGHT)
|
||||
assert hass.states.get(ENTITY_LIGHT).state == "on"
|
||||
assert hass.states.get(ENTITY_OTHER_LIGHT).state == "off"
|
||||
assert hass.states.get(ENTITY_LIGHT).attributes.get(ATTR_BRIGHTNESS) == 255
|
||||
|
||||
assert not light.is_on(self.hass, ENTITY_LIGHT)
|
||||
# Light 2
|
||||
|
||||
common.turn_on(self.hass, ENTITY_LIGHT)
|
||||
self.hass.block_till_done()
|
||||
self.mock_lj.activate_load.assert_called_with(ENTITY_LIGHT_NUMBER)
|
||||
mock_litejet.get_load_level.return_value = 40
|
||||
mock_litejet.get_load_level.reset_mock()
|
||||
mock_litejet.load_activated_callbacks[ENTITY_OTHER_LIGHT_NUMBER]()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
common.turn_off(self.hass, ENTITY_LIGHT)
|
||||
self.hass.block_till_done()
|
||||
self.mock_lj.deactivate_load.assert_called_with(ENTITY_LIGHT_NUMBER)
|
||||
mock_litejet.get_load_level.assert_called_once_with(ENTITY_OTHER_LIGHT_NUMBER)
|
||||
|
||||
def test_activated_event(self):
|
||||
"""Test handling an event from LiteJet."""
|
||||
self.mock_lj.get_load_level.return_value = 99
|
||||
assert light.is_on(hass, ENTITY_LIGHT)
|
||||
assert light.is_on(hass, ENTITY_OTHER_LIGHT)
|
||||
assert hass.states.get(ENTITY_LIGHT).state == "on"
|
||||
assert hass.states.get(ENTITY_OTHER_LIGHT).state == "on"
|
||||
assert (
|
||||
int(hass.states.get(ENTITY_OTHER_LIGHT).attributes.get(ATTR_BRIGHTNESS)) == 103
|
||||
)
|
||||
|
||||
# Light 1
|
||||
|
||||
_LOGGER.info(self.load_activated_callbacks[ENTITY_LIGHT_NUMBER])
|
||||
self.load_activated_callbacks[ENTITY_LIGHT_NUMBER]()
|
||||
self.hass.block_till_done()
|
||||
async def test_deactivated_event(hass, mock_litejet):
|
||||
"""Test handling an event from LiteJet."""
|
||||
await async_init_integration(hass)
|
||||
|
||||
self.mock_lj.get_load_level.assert_called_once_with(ENTITY_LIGHT_NUMBER)
|
||||
# Initial state is on.
|
||||
mock_litejet.get_load_level.return_value = 99
|
||||
|
||||
assert light.is_on(self.hass, ENTITY_LIGHT)
|
||||
assert not light.is_on(self.hass, ENTITY_OTHER_LIGHT)
|
||||
assert self.light().state == "on"
|
||||
assert self.other_light().state == "off"
|
||||
assert self.light().attributes.get(light.ATTR_BRIGHTNESS) == 255
|
||||
mock_litejet.load_activated_callbacks[ENTITY_OTHER_LIGHT_NUMBER]()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Light 2
|
||||
assert light.is_on(hass, ENTITY_OTHER_LIGHT)
|
||||
|
||||
self.mock_lj.get_load_level.return_value = 40
|
||||
# Event indicates it is off now.
|
||||
|
||||
self.mock_lj.get_load_level.reset_mock()
|
||||
mock_litejet.get_load_level.reset_mock()
|
||||
mock_litejet.get_load_level.return_value = 0
|
||||
|
||||
self.load_activated_callbacks[ENTITY_OTHER_LIGHT_NUMBER]()
|
||||
self.hass.block_till_done()
|
||||
mock_litejet.load_deactivated_callbacks[ENTITY_OTHER_LIGHT_NUMBER]()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
self.mock_lj.get_load_level.assert_called_once_with(ENTITY_OTHER_LIGHT_NUMBER)
|
||||
# (Requesting the level is not strictly needed with a deactivated
|
||||
# event but the implementation happens to do it. This could be
|
||||
# changed to an assert_not_called in the future.)
|
||||
mock_litejet.get_load_level.assert_called_with(ENTITY_OTHER_LIGHT_NUMBER)
|
||||
|
||||
assert light.is_on(self.hass, ENTITY_OTHER_LIGHT)
|
||||
assert light.is_on(self.hass, ENTITY_LIGHT)
|
||||
assert self.light().state == "on"
|
||||
assert self.other_light().state == "on"
|
||||
assert int(self.other_light().attributes[light.ATTR_BRIGHTNESS]) == 103
|
||||
|
||||
def test_deactivated_event(self):
|
||||
"""Test handling an event from LiteJet."""
|
||||
# Initial state is on.
|
||||
|
||||
self.mock_lj.get_load_level.return_value = 99
|
||||
|
||||
self.load_activated_callbacks[ENTITY_OTHER_LIGHT_NUMBER]()
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert light.is_on(self.hass, ENTITY_OTHER_LIGHT)
|
||||
|
||||
# Event indicates it is off now.
|
||||
|
||||
self.mock_lj.get_load_level.reset_mock()
|
||||
self.mock_lj.get_load_level.return_value = 0
|
||||
|
||||
self.load_deactivated_callbacks[ENTITY_OTHER_LIGHT_NUMBER]()
|
||||
self.hass.block_till_done()
|
||||
|
||||
# (Requesting the level is not strictly needed with a deactivated
|
||||
# event but the implementation happens to do it. This could be
|
||||
# changed to an assert_not_called in the future.)
|
||||
self.mock_lj.get_load_level.assert_called_with(ENTITY_OTHER_LIGHT_NUMBER)
|
||||
|
||||
assert not light.is_on(self.hass, ENTITY_OTHER_LIGHT)
|
||||
assert not light.is_on(self.hass, ENTITY_LIGHT)
|
||||
assert self.light().state == "off"
|
||||
assert self.other_light().state == "off"
|
||||
assert not light.is_on(hass, ENTITY_OTHER_LIGHT)
|
||||
assert not light.is_on(hass, ENTITY_LIGHT)
|
||||
assert hass.states.get(ENTITY_LIGHT).state == "off"
|
||||
assert hass.states.get(ENTITY_OTHER_LIGHT).state == "off"
|
||||
|
||||
@ -1,12 +1,8 @@
|
||||
"""The tests for the litejet component."""
|
||||
import unittest
|
||||
from unittest import mock
|
||||
from homeassistant.components import scene
|
||||
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_ON
|
||||
|
||||
from homeassistant import setup
|
||||
from homeassistant.components import litejet
|
||||
|
||||
from tests.common import get_test_home_assistant
|
||||
from tests.components.scene import common
|
||||
from . import async_init_integration
|
||||
|
||||
ENTITY_SCENE = "scene.mock_scene_1"
|
||||
ENTITY_SCENE_NUMBER = 1
|
||||
@ -14,46 +10,31 @@ ENTITY_OTHER_SCENE = "scene.mock_scene_2"
|
||||
ENTITY_OTHER_SCENE_NUMBER = 2
|
||||
|
||||
|
||||
class TestLiteJetScene(unittest.TestCase):
|
||||
"""Test the litejet component."""
|
||||
async def test_disabled_by_default(hass, mock_litejet):
|
||||
"""Test the scene is disabled by default."""
|
||||
await async_init_integration(hass)
|
||||
|
||||
@mock.patch("homeassistant.components.litejet.LiteJet")
|
||||
def setup_method(self, method, mock_pylitejet):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.hass.start()
|
||||
registry = await hass.helpers.entity_registry.async_get_registry()
|
||||
|
||||
def get_scene_name(number):
|
||||
return f"Mock Scene #{number}"
|
||||
state = hass.states.get(ENTITY_SCENE)
|
||||
assert state is None
|
||||
|
||||
self.mock_lj = mock_pylitejet.return_value
|
||||
self.mock_lj.loads.return_value = range(0)
|
||||
self.mock_lj.button_switches.return_value = range(0)
|
||||
self.mock_lj.all_switches.return_value = range(0)
|
||||
self.mock_lj.scenes.return_value = range(1, 3)
|
||||
self.mock_lj.get_scene_name.side_effect = get_scene_name
|
||||
entry = registry.async_get(ENTITY_SCENE)
|
||||
assert entry
|
||||
assert entry.disabled
|
||||
assert entry.disabled_by == "integration"
|
||||
|
||||
assert setup.setup_component(
|
||||
self.hass,
|
||||
litejet.DOMAIN,
|
||||
{"litejet": {"port": "/dev/serial/by-id/mock-litejet"}},
|
||||
)
|
||||
self.hass.block_till_done()
|
||||
|
||||
def teardown_method(self, method):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
async def test_activate(hass, mock_litejet):
|
||||
"""Test activating the scene."""
|
||||
|
||||
def scene(self):
|
||||
"""Get the current scene."""
|
||||
return self.hass.states.get(ENTITY_SCENE)
|
||||
await async_init_integration(hass, use_scene=True)
|
||||
|
||||
def other_scene(self):
|
||||
"""Get the other scene."""
|
||||
return self.hass.states.get(ENTITY_OTHER_SCENE)
|
||||
state = hass.states.get(ENTITY_SCENE)
|
||||
assert state is not None
|
||||
|
||||
def test_activate(self):
|
||||
"""Test activating the scene."""
|
||||
common.activate(self.hass, ENTITY_SCENE)
|
||||
self.hass.block_till_done()
|
||||
self.mock_lj.activate_scene.assert_called_once_with(ENTITY_SCENE_NUMBER)
|
||||
await hass.services.async_call(
|
||||
scene.DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_SCENE}, blocking=True
|
||||
)
|
||||
|
||||
mock_litejet.activate_scene.assert_called_once_with(ENTITY_SCENE_NUMBER)
|
||||
|
||||
@ -1,14 +1,10 @@
|
||||
"""The tests for the litejet component."""
|
||||
import logging
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from homeassistant import setup
|
||||
from homeassistant.components import litejet
|
||||
import homeassistant.components.switch as switch
|
||||
from homeassistant.components import switch
|
||||
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF, SERVICE_TURN_ON
|
||||
|
||||
from tests.common import get_test_home_assistant
|
||||
from tests.components.switch import common
|
||||
from . import async_init_integration
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@ -18,117 +14,67 @@ ENTITY_OTHER_SWITCH = "switch.mock_switch_2"
|
||||
ENTITY_OTHER_SWITCH_NUMBER = 2
|
||||
|
||||
|
||||
class TestLiteJetSwitch(unittest.TestCase):
|
||||
"""Test the litejet component."""
|
||||
async def test_on_off(hass, mock_litejet):
|
||||
"""Test turning the switch on and off."""
|
||||
|
||||
@mock.patch("homeassistant.components.litejet.LiteJet")
|
||||
def setup_method(self, method, mock_pylitejet):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.hass.start()
|
||||
await async_init_integration(hass, use_switch=True)
|
||||
|
||||
self.switch_pressed_callbacks = {}
|
||||
self.switch_released_callbacks = {}
|
||||
assert hass.states.get(ENTITY_SWITCH).state == "off"
|
||||
assert hass.states.get(ENTITY_OTHER_SWITCH).state == "off"
|
||||
|
||||
def get_switch_name(number):
|
||||
return f"Mock Switch #{number}"
|
||||
assert not switch.is_on(hass, ENTITY_SWITCH)
|
||||
|
||||
def on_switch_pressed(number, callback):
|
||||
self.switch_pressed_callbacks[number] = callback
|
||||
await hass.services.async_call(
|
||||
switch.DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_SWITCH}, blocking=True
|
||||
)
|
||||
mock_litejet.press_switch.assert_called_with(ENTITY_SWITCH_NUMBER)
|
||||
|
||||
def on_switch_released(number, callback):
|
||||
self.switch_released_callbacks[number] = callback
|
||||
await hass.services.async_call(
|
||||
switch.DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_SWITCH}, blocking=True
|
||||
)
|
||||
mock_litejet.release_switch.assert_called_with(ENTITY_SWITCH_NUMBER)
|
||||
|
||||
self.mock_lj = mock_pylitejet.return_value
|
||||
self.mock_lj.loads.return_value = range(0)
|
||||
self.mock_lj.button_switches.return_value = range(1, 3)
|
||||
self.mock_lj.all_switches.return_value = range(1, 6)
|
||||
self.mock_lj.scenes.return_value = range(0)
|
||||
self.mock_lj.get_switch_name.side_effect = get_switch_name
|
||||
self.mock_lj.on_switch_pressed.side_effect = on_switch_pressed
|
||||
self.mock_lj.on_switch_released.side_effect = on_switch_released
|
||||
|
||||
config = {"litejet": {"port": "/dev/serial/by-id/mock-litejet"}}
|
||||
if method == self.test_include_switches_False:
|
||||
config["litejet"]["include_switches"] = False
|
||||
elif method != self.test_include_switches_unspecified:
|
||||
config["litejet"]["include_switches"] = True
|
||||
async def test_pressed_event(hass, mock_litejet):
|
||||
"""Test handling an event from LiteJet."""
|
||||
|
||||
assert setup.setup_component(self.hass, litejet.DOMAIN, config)
|
||||
self.hass.block_till_done()
|
||||
await async_init_integration(hass, use_switch=True)
|
||||
|
||||
def teardown_method(self, method):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
# Switch 1
|
||||
mock_litejet.switch_pressed_callbacks[ENTITY_SWITCH_NUMBER]()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
def switch(self):
|
||||
"""Return the switch state."""
|
||||
return self.hass.states.get(ENTITY_SWITCH)
|
||||
assert switch.is_on(hass, ENTITY_SWITCH)
|
||||
assert not switch.is_on(hass, ENTITY_OTHER_SWITCH)
|
||||
assert hass.states.get(ENTITY_SWITCH).state == "on"
|
||||
assert hass.states.get(ENTITY_OTHER_SWITCH).state == "off"
|
||||
|
||||
def other_switch(self):
|
||||
"""Return the other switch state."""
|
||||
return self.hass.states.get(ENTITY_OTHER_SWITCH)
|
||||
# Switch 2
|
||||
mock_litejet.switch_pressed_callbacks[ENTITY_OTHER_SWITCH_NUMBER]()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
def test_include_switches_unspecified(self):
|
||||
"""Test that switches are ignored by default."""
|
||||
self.mock_lj.button_switches.assert_not_called()
|
||||
self.mock_lj.all_switches.assert_not_called()
|
||||
assert switch.is_on(hass, ENTITY_OTHER_SWITCH)
|
||||
assert switch.is_on(hass, ENTITY_SWITCH)
|
||||
assert hass.states.get(ENTITY_SWITCH).state == "on"
|
||||
assert hass.states.get(ENTITY_OTHER_SWITCH).state == "on"
|
||||
|
||||
def test_include_switches_False(self):
|
||||
"""Test that switches can be explicitly ignored."""
|
||||
self.mock_lj.button_switches.assert_not_called()
|
||||
self.mock_lj.all_switches.assert_not_called()
|
||||
|
||||
def test_on_off(self):
|
||||
"""Test turning the switch on and off."""
|
||||
assert self.switch().state == "off"
|
||||
assert self.other_switch().state == "off"
|
||||
async def test_released_event(hass, mock_litejet):
|
||||
"""Test handling an event from LiteJet."""
|
||||
|
||||
assert not switch.is_on(self.hass, ENTITY_SWITCH)
|
||||
await async_init_integration(hass, use_switch=True)
|
||||
|
||||
common.turn_on(self.hass, ENTITY_SWITCH)
|
||||
self.hass.block_till_done()
|
||||
self.mock_lj.press_switch.assert_called_with(ENTITY_SWITCH_NUMBER)
|
||||
# Initial state is on.
|
||||
mock_litejet.switch_pressed_callbacks[ENTITY_OTHER_SWITCH_NUMBER]()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
common.turn_off(self.hass, ENTITY_SWITCH)
|
||||
self.hass.block_till_done()
|
||||
self.mock_lj.release_switch.assert_called_with(ENTITY_SWITCH_NUMBER)
|
||||
assert switch.is_on(hass, ENTITY_OTHER_SWITCH)
|
||||
|
||||
def test_pressed_event(self):
|
||||
"""Test handling an event from LiteJet."""
|
||||
# Switch 1
|
||||
_LOGGER.info(self.switch_pressed_callbacks[ENTITY_SWITCH_NUMBER])
|
||||
self.switch_pressed_callbacks[ENTITY_SWITCH_NUMBER]()
|
||||
self.hass.block_till_done()
|
||||
# Event indicates it is off now.
|
||||
mock_litejet.switch_released_callbacks[ENTITY_OTHER_SWITCH_NUMBER]()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert switch.is_on(self.hass, ENTITY_SWITCH)
|
||||
assert not switch.is_on(self.hass, ENTITY_OTHER_SWITCH)
|
||||
assert self.switch().state == "on"
|
||||
assert self.other_switch().state == "off"
|
||||
|
||||
# Switch 2
|
||||
self.switch_pressed_callbacks[ENTITY_OTHER_SWITCH_NUMBER]()
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert switch.is_on(self.hass, ENTITY_OTHER_SWITCH)
|
||||
assert switch.is_on(self.hass, ENTITY_SWITCH)
|
||||
assert self.other_switch().state == "on"
|
||||
assert self.switch().state == "on"
|
||||
|
||||
def test_released_event(self):
|
||||
"""Test handling an event from LiteJet."""
|
||||
# Initial state is on.
|
||||
self.switch_pressed_callbacks[ENTITY_OTHER_SWITCH_NUMBER]()
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert switch.is_on(self.hass, ENTITY_OTHER_SWITCH)
|
||||
|
||||
# Event indicates it is off now.
|
||||
|
||||
self.switch_released_callbacks[ENTITY_OTHER_SWITCH_NUMBER]()
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert not switch.is_on(self.hass, ENTITY_OTHER_SWITCH)
|
||||
assert not switch.is_on(self.hass, ENTITY_SWITCH)
|
||||
assert self.other_switch().state == "off"
|
||||
assert self.switch().state == "off"
|
||||
assert not switch.is_on(hass, ENTITY_OTHER_SWITCH)
|
||||
assert not switch.is_on(hass, ENTITY_SWITCH)
|
||||
assert hass.states.get(ENTITY_SWITCH).state == "off"
|
||||
assert hass.states.get(ENTITY_OTHER_SWITCH).state == "off"
|
||||
|
||||
@ -2,14 +2,16 @@
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from unittest import mock
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant import setup
|
||||
from homeassistant.components import litejet
|
||||
import homeassistant.components.automation as automation
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
from . import async_init_integration
|
||||
|
||||
from tests.common import async_fire_time_changed, async_mock_service
|
||||
from tests.components.blueprint.conftest import stub_blueprint_populate # noqa
|
||||
|
||||
@ -27,88 +29,51 @@ def calls(hass):
|
||||
return async_mock_service(hass, "test", "automation")
|
||||
|
||||
|
||||
def get_switch_name(number):
|
||||
"""Get a mock switch name."""
|
||||
return f"Mock Switch #{number}"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_lj(hass):
|
||||
"""Initialize components."""
|
||||
with mock.patch("homeassistant.components.litejet.LiteJet") as mock_pylitejet:
|
||||
mock_lj = mock_pylitejet.return_value
|
||||
|
||||
mock_lj.switch_pressed_callbacks = {}
|
||||
mock_lj.switch_released_callbacks = {}
|
||||
|
||||
def on_switch_pressed(number, callback):
|
||||
mock_lj.switch_pressed_callbacks[number] = callback
|
||||
|
||||
def on_switch_released(number, callback):
|
||||
mock_lj.switch_released_callbacks[number] = callback
|
||||
|
||||
mock_lj.loads.return_value = range(0)
|
||||
mock_lj.button_switches.return_value = range(1, 3)
|
||||
mock_lj.all_switches.return_value = range(1, 6)
|
||||
mock_lj.scenes.return_value = range(0)
|
||||
mock_lj.get_switch_name.side_effect = get_switch_name
|
||||
mock_lj.on_switch_pressed.side_effect = on_switch_pressed
|
||||
mock_lj.on_switch_released.side_effect = on_switch_released
|
||||
|
||||
config = {"litejet": {"port": "/dev/serial/by-id/mock-litejet"}}
|
||||
assert hass.loop.run_until_complete(
|
||||
setup.async_setup_component(hass, litejet.DOMAIN, config)
|
||||
)
|
||||
|
||||
mock_lj.start_time = dt_util.utcnow()
|
||||
mock_lj.last_delta = timedelta(0)
|
||||
return mock_lj
|
||||
|
||||
|
||||
async def simulate_press(hass, mock_lj, number):
|
||||
async def simulate_press(hass, mock_litejet, number):
|
||||
"""Test to simulate a press."""
|
||||
_LOGGER.info("*** simulate press of %d", number)
|
||||
callback = mock_lj.switch_pressed_callbacks.get(number)
|
||||
callback = mock_litejet.switch_pressed_callbacks.get(number)
|
||||
with mock.patch(
|
||||
"homeassistant.helpers.condition.dt_util.utcnow",
|
||||
return_value=mock_lj.start_time + mock_lj.last_delta,
|
||||
return_value=mock_litejet.start_time + mock_litejet.last_delta,
|
||||
):
|
||||
if callback is not None:
|
||||
await hass.async_add_executor_job(callback)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def simulate_release(hass, mock_lj, number):
|
||||
async def simulate_release(hass, mock_litejet, number):
|
||||
"""Test to simulate releasing."""
|
||||
_LOGGER.info("*** simulate release of %d", number)
|
||||
callback = mock_lj.switch_released_callbacks.get(number)
|
||||
callback = mock_litejet.switch_released_callbacks.get(number)
|
||||
with mock.patch(
|
||||
"homeassistant.helpers.condition.dt_util.utcnow",
|
||||
return_value=mock_lj.start_time + mock_lj.last_delta,
|
||||
return_value=mock_litejet.start_time + mock_litejet.last_delta,
|
||||
):
|
||||
if callback is not None:
|
||||
await hass.async_add_executor_job(callback)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def simulate_time(hass, mock_lj, delta):
|
||||
async def simulate_time(hass, mock_litejet, delta):
|
||||
"""Test to simulate time."""
|
||||
_LOGGER.info(
|
||||
"*** simulate time change by %s: %s", delta, mock_lj.start_time + delta
|
||||
"*** simulate time change by %s: %s", delta, mock_litejet.start_time + delta
|
||||
)
|
||||
mock_lj.last_delta = delta
|
||||
mock_litejet.last_delta = delta
|
||||
with mock.patch(
|
||||
"homeassistant.helpers.condition.dt_util.utcnow",
|
||||
return_value=mock_lj.start_time + delta,
|
||||
return_value=mock_litejet.start_time + delta,
|
||||
):
|
||||
_LOGGER.info("now=%s", dt_util.utcnow())
|
||||
async_fire_time_changed(hass, mock_lj.start_time + delta)
|
||||
async_fire_time_changed(hass, mock_litejet.start_time + delta)
|
||||
await hass.async_block_till_done()
|
||||
_LOGGER.info("done with now=%s", dt_util.utcnow())
|
||||
|
||||
|
||||
async def setup_automation(hass, trigger):
|
||||
"""Test setting up the automation."""
|
||||
await async_init_integration(hass, use_switch=True)
|
||||
assert await setup.async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
@ -125,19 +90,19 @@ async def setup_automation(hass, trigger):
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def test_simple(hass, calls, mock_lj):
|
||||
async def test_simple(hass, calls, mock_litejet):
|
||||
"""Test the simplest form of a LiteJet trigger."""
|
||||
await setup_automation(
|
||||
hass, {"platform": "litejet", "number": ENTITY_OTHER_SWITCH_NUMBER}
|
||||
)
|
||||
|
||||
await simulate_press(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_release(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_press(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_release(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
|
||||
assert len(calls) == 1
|
||||
|
||||
|
||||
async def test_held_more_than_short(hass, calls, mock_lj):
|
||||
async def test_held_more_than_short(hass, calls, mock_litejet):
|
||||
"""Test a too short hold."""
|
||||
await setup_automation(
|
||||
hass,
|
||||
@ -148,13 +113,13 @@ async def test_held_more_than_short(hass, calls, mock_lj):
|
||||
},
|
||||
)
|
||||
|
||||
await simulate_press(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_time(hass, mock_lj, timedelta(seconds=0.1))
|
||||
await simulate_release(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_press(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_time(hass, mock_litejet, timedelta(seconds=0.1))
|
||||
await simulate_release(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 0
|
||||
|
||||
|
||||
async def test_held_more_than_long(hass, calls, mock_lj):
|
||||
async def test_held_more_than_long(hass, calls, mock_litejet):
|
||||
"""Test a hold that is long enough."""
|
||||
await setup_automation(
|
||||
hass,
|
||||
@ -165,15 +130,15 @@ async def test_held_more_than_long(hass, calls, mock_lj):
|
||||
},
|
||||
)
|
||||
|
||||
await simulate_press(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_press(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 0
|
||||
await simulate_time(hass, mock_lj, timedelta(seconds=0.3))
|
||||
await simulate_time(hass, mock_litejet, timedelta(seconds=0.3))
|
||||
assert len(calls) == 1
|
||||
await simulate_release(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_release(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 1
|
||||
|
||||
|
||||
async def test_held_less_than_short(hass, calls, mock_lj):
|
||||
async def test_held_less_than_short(hass, calls, mock_litejet):
|
||||
"""Test a hold that is short enough."""
|
||||
await setup_automation(
|
||||
hass,
|
||||
@ -184,14 +149,14 @@ async def test_held_less_than_short(hass, calls, mock_lj):
|
||||
},
|
||||
)
|
||||
|
||||
await simulate_press(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_time(hass, mock_lj, timedelta(seconds=0.1))
|
||||
await simulate_press(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_time(hass, mock_litejet, timedelta(seconds=0.1))
|
||||
assert len(calls) == 0
|
||||
await simulate_release(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_release(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 1
|
||||
|
||||
|
||||
async def test_held_less_than_long(hass, calls, mock_lj):
|
||||
async def test_held_less_than_long(hass, calls, mock_litejet):
|
||||
"""Test a hold that is too long."""
|
||||
await setup_automation(
|
||||
hass,
|
||||
@ -202,15 +167,15 @@ async def test_held_less_than_long(hass, calls, mock_lj):
|
||||
},
|
||||
)
|
||||
|
||||
await simulate_press(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_press(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 0
|
||||
await simulate_time(hass, mock_lj, timedelta(seconds=0.3))
|
||||
await simulate_time(hass, mock_litejet, timedelta(seconds=0.3))
|
||||
assert len(calls) == 0
|
||||
await simulate_release(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_release(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 0
|
||||
|
||||
|
||||
async def test_held_in_range_short(hass, calls, mock_lj):
|
||||
async def test_held_in_range_short(hass, calls, mock_litejet):
|
||||
"""Test an in-range trigger with a too short hold."""
|
||||
await setup_automation(
|
||||
hass,
|
||||
@ -222,13 +187,13 @@ async def test_held_in_range_short(hass, calls, mock_lj):
|
||||
},
|
||||
)
|
||||
|
||||
await simulate_press(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_time(hass, mock_lj, timedelta(seconds=0.05))
|
||||
await simulate_release(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_press(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_time(hass, mock_litejet, timedelta(seconds=0.05))
|
||||
await simulate_release(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 0
|
||||
|
||||
|
||||
async def test_held_in_range_just_right(hass, calls, mock_lj):
|
||||
async def test_held_in_range_just_right(hass, calls, mock_litejet):
|
||||
"""Test an in-range trigger with a just right hold."""
|
||||
await setup_automation(
|
||||
hass,
|
||||
@ -240,15 +205,15 @@ async def test_held_in_range_just_right(hass, calls, mock_lj):
|
||||
},
|
||||
)
|
||||
|
||||
await simulate_press(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_press(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 0
|
||||
await simulate_time(hass, mock_lj, timedelta(seconds=0.2))
|
||||
await simulate_time(hass, mock_litejet, timedelta(seconds=0.2))
|
||||
assert len(calls) == 0
|
||||
await simulate_release(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_release(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 1
|
||||
|
||||
|
||||
async def test_held_in_range_long(hass, calls, mock_lj):
|
||||
async def test_held_in_range_long(hass, calls, mock_litejet):
|
||||
"""Test an in-range trigger with a too long hold."""
|
||||
await setup_automation(
|
||||
hass,
|
||||
@ -260,9 +225,50 @@ async def test_held_in_range_long(hass, calls, mock_lj):
|
||||
},
|
||||
)
|
||||
|
||||
await simulate_press(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_press(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 0
|
||||
await simulate_time(hass, mock_lj, timedelta(seconds=0.4))
|
||||
await simulate_time(hass, mock_litejet, timedelta(seconds=0.4))
|
||||
assert len(calls) == 0
|
||||
await simulate_release(hass, mock_lj, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
await simulate_release(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 0
|
||||
|
||||
|
||||
async def test_reload(hass, calls, mock_litejet):
|
||||
"""Test reloading automation."""
|
||||
await setup_automation(
|
||||
hass,
|
||||
{
|
||||
"platform": "litejet",
|
||||
"number": ENTITY_OTHER_SWITCH_NUMBER,
|
||||
"held_more_than": {"milliseconds": "100"},
|
||||
"held_less_than": {"milliseconds": "300"},
|
||||
},
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.config.load_yaml_config_file",
|
||||
autospec=True,
|
||||
return_value={
|
||||
"automation": {
|
||||
"trigger": {
|
||||
"platform": "litejet",
|
||||
"number": ENTITY_OTHER_SWITCH_NUMBER,
|
||||
"held_more_than": {"milliseconds": "1000"},
|
||||
},
|
||||
"action": {"service": "test.automation"},
|
||||
}
|
||||
},
|
||||
):
|
||||
await hass.services.async_call(
|
||||
"automation",
|
||||
"reload",
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await simulate_press(hass, mock_litejet, ENTITY_OTHER_SWITCH_NUMBER)
|
||||
assert len(calls) == 0
|
||||
await simulate_time(hass, mock_litejet, timedelta(seconds=0.5))
|
||||
assert len(calls) == 0
|
||||
await simulate_time(hass, mock_litejet, timedelta(seconds=1.25))
|
||||
assert len(calls) == 1
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user