remote/adaptive-lighting
1
0
Fork 0
mirror of https://github.com/basnijholt/adaptive-lighting.git synced 2026-09-12 06:44:04 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge remote-tracking branch 'origin/main' into test-multiple-ha-versions

Browse source
This commit is contained in:
Bas Nijholt 2023-04-06 13:35:28 -07:00
commit d839571e2b
4 changed files with 394 additions and 129 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/update-readme.yml vendored
View file

@ -24,7 +24,7 @@ jobs:
- name: Install markdown-code-runner and README code dependencies
run: |
pip install markdown-code-runner pandas tabulate
pip install markdown-code-runner==1.0.0 pandas tabulate
- name: Link custom_components/adaptive_lighting
run: |

4
custom_components/adaptive_lighting/manifest.json
View file

@ -7,6 +7,6 @@
"documentation": "https://github.com/basnijholt/adaptive-lighting#readme",
"iot_class": "calculated",
"issue_tracker": "https://github.com/basnijholt/adaptive-lighting/issues",
"requirements": [],
"version": "1.10.0"
"requirements": ["ulid-transform"],
"version": "1.10.1"
}

136
custom_components/adaptive_lighting/switch.py
View file

@ -88,6 +88,7 @@ from homeassistant.util.color import (
color_xy_to_RGB,
)
import homeassistant.util.dt as dt_util
import ulid_transform
import voluptuous as vol
from .const import (
@ -182,21 +183,58 @@ BRIGHTNESS_ATTRS = {
}
# Keep a short domain version for the context instances (which can only be 36 chars)
_DOMAIN_SHORT = "adapt_lgt"
_DOMAIN_SHORT = "al"
def _int_to_bytes(i: int, signed: bool = False) -> bytes:
bits = i.bit_length()
if signed:
# Make room for the sign bit.
bits += 1
return i.to_bytes((bits + 7) // 8, "little", signed=signed)
def _int_to_base36(num: int) -> str:
"""
Convert an integer to its base-36 representation using numbers and uppercase letters.
Base-36 encoding uses digits 0-9 and uppercase letters A-Z, providing a case-insensitive
alphanumeric representation. The function takes an integer `num` as input and returns
its base-36 representation as a string.
Parameters
----------
num
The integer to convert to base-36.
Returns
-------
str
The base-36 representation of the input integer.
Examples
--------
>>> num = 123456
>>> base36_num = int_to_base36(num)
>>> print(base36_num)
'2N9'
"""
ALPHANUMERIC_CHARS = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
if num == 0:
return ALPHANUMERIC_CHARS[0]
base36_str = ""
base = len(ALPHANUMERIC_CHARS)
while num:
num, remainder = divmod(num, base)
base36_str = ALPHANUMERIC_CHARS[remainder] + base36_str
return base36_str
def _short_hash(string: str, length: int = 4) -> str:
"""Create a hash of 'string' with length 'length'."""
str_hash_bytes = _int_to_bytes(hash(string), signed=True)
return base64.b85encode(str_hash_bytes)[:length]
return base64.b32encode(string.encode()).decode("utf-8").zfill(length)[:length]
def _remove_vowels(input_str: str, length: int = 4) -> str:
vowels = "aeiouAEIOU"
output_str = "".join([char for char in input_str if char not in vowels])
return output_str.zfill(length)[:length]
def create_context(
@ -204,12 +242,16 @@ def create_context(
) -> Context:
"""Create a context that can identify this integration."""
# Use a hash for the name because otherwise the context might become
# too long (max len == 36) to fit in the database.
name_hash = _short_hash(name)
# too long (max len == 26) to fit in the database.
# Pack index with base85 to maximize the number of contexts we can create
# before we exceed the 36-character limit and are forced to wrap.
index_packed = base64.b85encode(_int_to_bytes(index, signed=False))
context_id = f"{_DOMAIN_SHORT}:{name_hash}:{which}:{index_packed}"[:36]
# before we exceed the 26-character limit and are forced to wrap.
time_stamp = ulid_transform.ulid_now()[:10] # time part of a ULID
name_hash = _short_hash(name)
which_short = _remove_vowels(which)
context_id_start = f"{time_stamp}:{_DOMAIN_SHORT}:{name_hash}:{which_short}:"
chars_left = 26 - len(context_id_start)
index_packed = _int_to_base36(index).zfill(chars_left)[-chars_left:]
context_id = context_id_start + index_packed
parent_id = parent.id if parent else None
return Context(id=context_id, parent_id=parent_id)
@ -218,7 +260,7 @@ def is_our_context(context: Context | None) -> bool:
"""Check whether this integration created 'context'."""
if context is None:
return False
return context.id.startswith(_DOMAIN_SHORT)
return f":{_DOMAIN_SHORT}:" in context.id
def _split_service_data(service_data, adapt_brightness, adapt_color):
@ -1102,7 +1144,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
return
# See #80. Doesn't check if transitions differ but it does the job.
last_service_data = self.turn_on_off_listener.last_service_data
if last_service_data.get(light) == service_data:
if not force and last_service_data.get(light) == service_data:
_LOGGER.debug(
"%s: Cancelling adapt to light %s, there's no new values to set (context.id='%s')",
self._name,
@ -1167,14 +1209,23 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if lights is None:
lights = self._lights
if not force and self._only_once:
return
filtered_lights = []
for light in lights:
# Don't adapt lights that haven't finished prior transitions.
if force or not self.turn_on_off_listener.transition_timers.get(light):
filtered_lights.append(light)
if not force:
if self._only_once:
return
for light in lights:
# Don't adapt lights that haven't finished prior transitions.
timer = self.turn_on_off_listener.transition_timers.get(light)
if timer is not None and timer.is_running():
_LOGGER.debug(
"%s: Light '%s' is still transitioning",
self._name,
light,
)
else:
filtered_lights.append(light)
else:
filtered_lights = lights
if not filtered_lights:
return
@ -1620,33 +1671,28 @@ class TurnOnOffListener:
def start_transition_timer(self, light: str) -> None:
"""Mark a light as manually controlled."""
_LOGGER.debug("Start transition timer for %s", light)
last_service_data = self.last_service_data
if (
not last_service_data
or light not in last_service_data
or ATTR_TRANSITION not in last_service_data[light]
):
last_service_data = self.last_service_data.get(light)
if not last_service_data:
_LOGGER.debug("This should not ever happen. Please report to the devs.")
return
delay = last_service_data[light][ATTR_TRANSITION]
last_transition = last_service_data.get(ATTR_TRANSITION)
if not last_transition:
_LOGGER.debug(
"No transition in last adapt for light %s, continuing...", light
)
return
_LOGGER.debug(
"Start transition timer of %s seconds for light %s", last_transition, light
)
async def reset():
ValueError("TEST")
_LOGGER.debug(
"Transition finished for light %s",
light,
)
switches = _get_switches_with_lights(self.hass, [light])
for switch in switches:
if not switch.is_on:
continue
await switch._update_attrs_and_maybe_adapt_lights(
[light],
force=False,
context=switch.create_context("transit"),
)
self._handle_timer(light, self.transition_timers, delay, reset)
self._handle_timer(light, self.transition_timers, last_transition, reset)
def set_auto_reset_manual_control_times(self, lights: list[str], time: float):
"""Set the time after which the lights are automatically reset."""
@ -1769,7 +1815,7 @@ class TurnOnOffListener:
async def state_changed_event_listener(self, event: Event) -> None:
"""Track 'state_changed' events."""
entity_id = event.data.get(ATTR_ENTITY_ID, "")
if entity_id not in self.lights or entity_id.split(".")[0] != LIGHT_DOMAIN:
if entity_id not in self.lights:
return
new_state = event.data.get("new_state")
@ -1814,6 +1860,10 @@ class TurnOnOffListener:
entity_id,
)
self.last_state_change[entity_id] = [new_state]
_LOGGER.debug(
"Last transition: %s",
self.last_service_data[entity_id].get(ATTR_TRANSITION),
)
self.start_transition_timer(entity_id)
elif old_state is not None:
self.last_state_change[entity_id].append(new_state)

389
tests/test_switch.py
View file

@ -1,6 +1,7 @@
"""Tests for Adaptive Lighting switches."""
# pylint: disable=protected-access
import asyncio
from copy import deepcopy
import datetime
import logging
from random import randint
@ -47,6 +48,7 @@ from homeassistant.components.light import (
ATTR_BRIGHTNESS_PCT,
ATTR_COLOR_TEMP_KELVIN,
ATTR_RGB_COLOR,
ATTR_TRANSITION,
ATTR_XY_COLOR,
)
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
@ -61,6 +63,7 @@ from homeassistant.const import (
CONF_LIGHTS,
CONF_NAME,
CONF_PLATFORM,
EVENT_STATE_CHANGED,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
@ -71,6 +74,7 @@ from homeassistant.setup import async_setup_component
from homeassistant.util.color import color_temperature_mired_to_kelvin
import homeassistant.util.dt as dt_util
import pytest
import ulid_transform
import voluptuous.error
from tests.common import MockConfigEntry, mock_area_registry
@ -107,6 +111,15 @@ ENTITY_ADAPT_COLOR_SWITCH = f"{_SWITCH_FMT}_adapt_color_{DEFAULT_NAME}"
ORIG_TIMEZONE = dt_util.DEFAULT_TIME_ZONE
GLOBAL_TEST_DEPENDENCIES = [
"test_adaptive_lighting_switches",
"test_light_settings",
]
def create_random_context() -> str:
return Context(id=ulid_transform.ulid_now(), parent_id=None)
@pytest.fixture
def reset_time_zone():
@ -209,6 +222,59 @@ async def setup_lights_and_switch(hass, extra_conf=None):
return switch, lights_instances
# see https://github.com/home-assistant/core/blob/dev/homeassistant/scripts/benchmark/__init__.py
# basically just search the repo for EVENT_STATE_CHANGED look for how it's fired.
def create_transition_events(
light: str,
state: State,
last: dict | None = None,
current: dict | None = None,
total_events: int = 4,
) -> list[dict]:
assert light is not None
all_events = []
for i in range(1, total_events):
# Build basic event data.
attributes = {}
# The first state change always has the context from our integration.
# That one will not be in all_events.
# It's very possible it stores the parent_id though.
# If it stores the parent_id in all situations, there's a great improvement
# that could added in future updates.
# Simulate the events the bulb would send to HASS.
last_brightness = last.get(ATTR_BRIGHTNESS) or state[ATTR_BRIGHTNESS]
current_brightness = current.get(ATTR_BRIGHTNESS)
if (
last_brightness
and current_brightness
and last_brightness != current_brightness
):
diff = (current_brightness - last_brightness) * (i / total_events)
attributes[ATTR_BRIGHTNESS] = last_brightness + diff
elif current_brightness:
attributes[ATTR_BRIGHTNESS] = current_brightness
current_kelvin = current.get(ATTR_COLOR_TEMP_KELVIN)
last_kelvin = last.get(ATTR_COLOR_TEMP_KELVIN) or state[ATTR_COLOR_TEMP_KELVIN]
if last_kelvin and current_kelvin and last_kelvin != current_kelvin:
diff = (current_kelvin - last_kelvin) * (i / total_events)
attributes[ATTR_COLOR_TEMP_KELVIN] = last_kelvin + diff
elif current_kelvin:
attributes[ATTR_COLOR_TEMP_KELVIN] = current_kelvin
# Pack event
event_data = {
ATTR_ENTITY_ID: light,
"old_state": State(light, "on", attributes=last),
"new_state": State(
light, "on", attributes=attributes, context=create_random_context()
),
}
all_events.append(event_data)
return all_events
async def test_adaptive_lighting_switches(hass):
"""Test switches created for adaptive_lighting integration."""
entry, _ = await setup_switch(hass, {})
@ -236,6 +302,7 @@ async def test_adaptive_lighting_switches(hass):
@pytest.mark.parametrize("lat,long,timezone", LAT_LONG_TZS)
@pytest.mark.dependency("test_adaptive_lighting_switches")
async def test_adaptive_lighting_time_zones_with_default_settings(
hass, lat, long, timezone, reset_time_zone # pylint: disable=redefined-outer-name
):
@ -428,6 +495,7 @@ async def test_light_settings(hass):
assert_expected_color_temp(state)
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
async def test_turn_on_off_listener_not_tracking_untracked_lights(hass):
"""Test that lights that are not in a Adaptive Lighting switch aren't tracked."""
switch, _ = await setup_lights_and_switch(hass)
@ -447,6 +515,7 @@ async def test_turn_on_off_listener_not_tracking_untracked_lights(hass):
assert light not in switch.turn_on_off_listener.lights
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
async def test_manual_control(hass):
"""Test the 'manual control' tracking."""
switch, (light, *_) = await setup_lights_and_switch(hass)
@ -594,6 +663,7 @@ async def test_manual_control(hass):
assert all([not manual_control[eid] for eid in switch._lights])
@pytest.mark.dependency(depends=[*GLOBAL_TEST_DEPENDENCIES, "test_manual_control"])
async def test_auto_reset_manual_control(hass):
switch, (light, *_) = await setup_lights_and_switch(
hass, {CONF_AUTORESET_CONTROL: 0.1}
@ -638,6 +708,7 @@ async def test_auto_reset_manual_control(hass):
assert not manual_control[light.entity_id]
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
async def test_apply_service(hass):
"""Test adaptive_lighting.apply service."""
switch, (_, _, light) = await setup_lights_and_switch(hass)
@ -701,6 +772,9 @@ async def test_apply_service(hass):
assert old_state[ATTR_COLOR_TEMP_KELVIN] == new_state[ATTR_COLOR_TEMP_KELVIN]
@pytest.mark.dependency(
depends=[*GLOBAL_TEST_DEPENDENCIES, "test_apply_service", "test_manual_control"]
)
async def test_switch_off_on_off(hass):
"""Test switch rapid off_on_off."""
@ -751,85 +825,7 @@ async def test_switch_off_on_off(hass):
assert state == STATE_OFF
async def test_significant_change(hass):
"""Test significant change."""
async def turn_light(state, **kwargs):
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON if state else SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: ENTITY_LIGHT, **kwargs},
blocking=True,
)
await hass.async_block_till_done()
async def update(force):
await switch._update_attrs_and_maybe_adapt_lights(
transition=0,
context=switch.create_context("test"),
force=force,
)
await hass.async_block_till_done()
async def set_brightness(val: int):
hass.states.async_set(
ENTITY_LIGHT, "on", {ATTR_BRIGHTNESS: val, ATTR_SUPPORTED_FEATURES: 1}
)
await hass.async_block_till_done()
switch, _ = await setup_lights_and_switch(hass)
_LOGGER.debug("Test detect_non_ha_changes:")
switch._take_over_control = True
assert switch._take_over_control
switch._detect_non_ha_changes = True
assert switch._detect_non_ha_changes
# build last service data
await update(force=False)
# force=True should not reset manual control.
await turn_light(True, brightness=40)
await turn_light(True, brightness=20)
await update(force=False)
assert switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
await update(force=True)
assert switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
# turn light off then on should reset manual control.
await turn_light(False)
await turn_light(True)
assert not switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
# Assert last_service_data got filled from update()
await update(force=True)
assert switch.turn_on_off_listener.last_service_data.get(ENTITY_LIGHT) is not None
# Simulate a transition to 255 where the update() is already using brightness 255.
await set_brightness(240)
await set_brightness(244)
await set_brightness(247)
await set_brightness(250)
# last_state_change should have our state changes.
# Change brightness by async_set (not using 'light.turn_on')
new_brightness = 50
await set_brightness(new_brightness)
_LOGGER.debug("Test: Brightness set to %s", new_brightness)
# mock homeassistant.core.HomeAssistant.helpers.entity_component.async_update_entity
# Otherwise what happens is update_entity() refreshes the state to the last call of
# light.turn_on(). This is because we are not using hass.states.async_set() to
# set the brightness of the light. We mock `async_update_ha_state` because
# `async_update_entity` calls it.
with patch("homeassistant.helpers.entity.Entity.async_update_ha_state"):
# On next update ENTITY_LIGHT should be marked as manually controlled
await update(force=False)
assert (
switch.turn_on_off_listener.last_service_data.get(ENTITY_LIGHT) is not None
)
assert switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
def test_color_difference_redmean():
"""Test color_difference_redmean function."""
for _ in range(10):
@ -839,14 +835,6 @@ def test_color_difference_redmean():
color_difference_redmean((0, 0, 0), (255, 255, 255))
def test_is_our_context():
"""Test is_our_context function."""
context = create_context(DOMAIN, "test", 0)
assert is_our_context(context)
assert not is_our_context(None)
assert not is_our_context(Context())
def test_attributes_have_changed():
"""Test _attributes_have_changed function."""
attributes_1 = {
@ -900,6 +888,229 @@ def test_attributes_have_changed():
)
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
async def test_state_change_handlers(hass):
"""
Test TurnOnOffListener's EVENT_STATE_CHANGED listener.
======================
Sequence of events:
1. Transition from sleep mode to normal.
2. Create simulated transition events for that adapt.
3. Fire all simulated transition events.
4. Assert all possible problems that would result.
Also tests significant changes.
"""
switch, (light, *_) = await setup_lights_and_switch(hass)
context = switch.create_context("test") # needs to be passed to update method
# [Config options]:
transition_used = 2
total_events = 5
async def set_brightness(val: int):
# 'Unsafe' set but we know what we're doing.
hass.states.async_set(
ENTITY_LIGHT, "on", {ATTR_BRIGHTNESS: val, ATTR_SUPPORTED_FEATURES: 1}
)
await hass.async_block_till_done()
# Call code in TurnOnOffListener
hass.bus.async_fire(
EVENT_STATE_CHANGED,
{
"new_state": {
ATTR_ENTITY_ID: ENTITY_LIGHT,
"state": "on",
ATTR_BRIGHTNESS: val,
}
},
)
await hass.async_block_till_done()
async def turn_light(state, **kwargs):
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON if state else SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: ENTITY_LIGHT, **kwargs},
blocking=True,
)
await hass.async_block_till_done()
async def update(force: bool = False):
await switch._update_attrs_and_maybe_adapt_lights(
force=force, transition=0, context=context
)
await hass.async_block_till_done()
# 1. Adapt to sleep without a transition.
# Should only be one state change.
_LOGGER.debug('test_state_change_handling: Turn on "sleep mode"')
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: ENTITY_SLEEP_MODE_SWITCH},
blocking=True,
)
await hass.async_block_till_done()
assert switch.turn_on_off_listener.last_state_change.get(ENTITY_LIGHT)
assert len(switch.turn_on_off_listener.last_state_change[ENTITY_LIGHT]) == 1
assert not switch.turn_on_off_listener.transition_timers.get(ENTITY_LIGHT)
last_service_data = deepcopy(switch.turn_on_off_listener.last_service_data)
assert last_service_data.get(ENTITY_LIGHT)
# 2 Adapt from sleep with a 'transition'.
await switch.sleep_mode_switch.async_turn_off()
await switch._update_attrs_and_maybe_adapt_lights(
force=False, transition=0, context=context
)
await hass.async_block_till_done()
current_service_data = switch.turn_on_off_listener.last_service_data
assert current_service_data != last_service_data
for light in switch._lights:
# current_service_data should have changed after the last update.
assert current_service_data.get(light)
assert last_service_data.get(light)
assert current_service_data[light] != last_service_data[light]
# Test same context id events.
current_service_data[light][ATTR_TRANSITION] = transition_used
hass.bus.async_fire(
EVENT_STATE_CHANGED,
{
ATTR_ENTITY_ID: light,
"old_state": State(light, "on", attributes=last_service_data),
"new_state": State(
light, "on", attributes=current_service_data, context=context
),
},
)
assert not switch.turn_on_off_listener.transition_timers.get(light)
# 2.3 Refire and overwrite the original state_changed event with our 'transition'
hass.bus.async_fire(
EVENT_STATE_CHANGED,
{
ATTR_ENTITY_ID: light,
"old_state": State(light, "on", attributes=last_service_data),
"new_state": State(
light,
"on",
attributes=current_service_data,
# We need to overwrite the old context_id
context=switch.create_context("test"),
),
},
)
await hass.async_block_till_done()
# Assert our transition timer was created.
assert switch.turn_on_off_listener.transition_timers.get(light)
# 2.5 Simulate a transition. There's no other way to do this in the demo.
events = create_transition_events(
light=light,
state=hass.states.get(light),
last=last_service_data[light],
current=current_service_data[light],
total_events=total_events,
)
# 3. Fire simulated events for our TurnOnOffListener
for event in events:
_LOGGER.debug("Test EVENT_STATE_CHANGED listener")
hass.bus.async_fire(EVENT_STATE_CHANGED, event)
await hass.async_block_till_done()
# On real systems HA fires transition state changes every ~3 seconds.
# asyncio.sleep(3)
# 4. Assert the transition timer started and everything was filled.
listener = switch.turn_on_off_listener
assert listener.last_state_change.get(ENTITY_LIGHT)
assert len(listener.last_state_change[ENTITY_LIGHT]) == total_events
assert listener.transition_timers.get(ENTITY_LIGHT)
# 5. Execute some checks during a transition
_LOGGER.debug("Test detect_non_ha_changes:")
switch._take_over_control = True
assert switch._take_over_control
switch._detect_non_ha_changes = True
assert switch._detect_non_ha_changes
await asyncio.sleep(transition_used / 3)
# Ensure the timer still exists
timer = listener.transition_timers.get(ENTITY_LIGHT)
assert timer and timer.is_running()
last_service_data = deepcopy(current_service_data)
await update()
assert not switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
await update()
assert not switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
timer = listener.transition_timers.get(ENTITY_LIGHT)
assert timer and timer.is_running()
# Ensure the light did not adapt during the transition.
assert last_service_data == current_service_data
# 6. Assert everything after the transition finishes.
await asyncio.sleep(transition_used)
assert listener.last_state_change.get(ENTITY_LIGHT)
assert len(listener.last_state_change[ENTITY_LIGHT]) == total_events
# Timer should be done and reset now.
# This is the assert that I can't fix.
timer = listener.transition_timers.get(ENTITY_LIGHT)
assert not timer or not timer.is_running()
# build last service data
await update(force=False)
# force=True should not reset manual control.
await turn_light(True, brightness=40)
await turn_light(True, brightness=20)
await update(force=False)
assert switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
await update(force=True)
assert switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
# turn light off then on should reset manual control.
await turn_light(False)
await turn_light(True)
assert not switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
# last_state_change should have our state changes.
# Change brightness by async_set (not using 'light.turn_on')
new_brightness = 50
await set_brightness(new_brightness)
_LOGGER.debug("Test: Brightness set to %s", new_brightness)
# mock homeassistant.core.HomeAssistant.helpers.entity_component.async_update_entity
# Otherwise what happens is update_entity() refreshes the state to the last call of
# light.turn_on(). This is because we are not using hass.states.async_set() to
# set the brightness of the light. We mock `async_update_ha_state` because
# `async_update_entity` calls it.
with patch("homeassistant.helpers.entity.Entity.async_update_ha_state"):
# On next update ENTITY_LIGHT should be marked as manually controlled
await update(force=False)
assert (
switch.turn_on_off_listener.last_service_data.get(ENTITY_LIGHT) is not None
)
assert (
switch.turn_on_off_listener.last_state_change.get(ENTITY_LIGHT) is not None
)
assert switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
@pytest.mark.dependency(
depends=[
*GLOBAL_TEST_DEPENDENCIES,
"test_manual_control",
"test_apply_service",
"test_attributes_have_changed",
"test_state_change_handling",
]
)
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
def test_is_our_context():
"""Test is_our_context function."""
context = create_context(DOMAIN, "test", 0)
assert is_our_context(context)
assert not is_our_context(None)
assert not is_our_context(Context())
async def test_unload_switch(hass):
"""Test removing Adaptive Lighting."""
entry, _ = await setup_switch(hass, {})
@ -966,6 +1177,7 @@ async def test_turn_on_and_off_when_already_at_that_state(hass):
await hass.async_block_till_done()
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
async def test_async_update_at_interval(hass):
"""Test '_async_update_at_interval' method."""
_, switch = await setup_switch(hass, {})
@ -973,6 +1185,7 @@ async def test_async_update_at_interval(hass):
@pytest.mark.parametrize("separate_turn_on_commands", (True, False))
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
async def test_separate_turn_on_commands(hass, separate_turn_on_commands):
"""Test 'separate_turn_on_commands' argument."""
switch, (light, *_) = await setup_lights_and_switch(
@ -1009,6 +1222,7 @@ async def test_separate_turn_on_commands(hass, separate_turn_on_commands):
assert sleep_color_temp != color_temp
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
async def test_area(hass):
switch, (light, *_) = await setup_lights_and_switch(hass)
@ -1045,6 +1259,7 @@ async def test_area(hass):
assert light.entity_id not in switch.turn_on_off_listener.last_service_data
@pytest.mark.dependency(depends=GLOBAL_TEST_DEPENDENCIES)
async def test_change_switch_settings_service(hass):
"""Test adaptive_lighting.change_switch_settings service."""
switch, (_, _, light) = await setup_lights_and_switch(hass)