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

Clean up and add tests

Browse source
This commit is contained in:
Adam DeMuri 2026-01-18 21:24:01 -07:00
commit 282d7631c8
4 changed files with 156 additions and 57 deletions
Download patch file Download diff file Expand all files Collapse all files

27
custom_components/adaptive_lighting/__init__.py
View file

@ -16,12 +16,18 @@ from .const import (
CONF_NAME,
DOMAIN,
SERVICE_APPLY,
SERVICE_CHANGE_SWITCH_SETTINGS,
SERVICE_SET_MANUAL_CONTROL,
SET_MANUAL_CONTROL_SCHEMA,
UNDO_UPDATE_LISTENER,
apply_service_schema,
change_switch_settings_schema,
)
from .switch import (
handle_apply_service,
handle_change_switch_settings,
handle_set_manual_control_service,
)
from .switch import handle_apply_service, handle_set_manual_control_service
_LOGGER = logging.getLogger(__name__)
@ -54,19 +60,26 @@ async def reload_configuration_yaml(event: Event) -> None:
async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
"""Import integration from config."""
hass.services.async_register(
DOMAIN,
SERVICE_APPLY,
partial(handle_apply_service, hass),
domain=DOMAIN,
service=SERVICE_APPLY,
service_func=partial(handle_apply_service, hass),
schema=apply_service_schema(),
)
hass.services.async_register(
DOMAIN,
SERVICE_SET_MANUAL_CONTROL,
partial(handle_set_manual_control_service, hass),
domain=DOMAIN,
service=SERVICE_SET_MANUAL_CONTROL,
service_func=partial(handle_set_manual_control_service, hass),
schema=SET_MANUAL_CONTROL_SCHEMA,
)
hass.services.async_register(
domain=DOMAIN,
service=SERVICE_CHANGE_SWITCH_SETTINGS,
service_func=partial(handle_change_switch_settings, hass),
schema=change_switch_settings_schema(),
)
if DOMAIN in config:
for entry in config[DOMAIN]:
hass.async_create_task(

14
custom_components/adaptive_lighting/const.py
View file

@ -474,6 +474,20 @@ def apply_service_schema() -> vol.Schema:
)
def change_switch_settings_schema() -> vol.Schema:
"""Return the schema for the change_switch_settings service."""
args: dict[vol.Marker, Any] = {
vol.Optional(CONF_ENTITY_ID): cv.entity_ids,
vol.Optional(CONF_USE_DEFAULTS, default="current"): cv.string,
}
# Modifying these after init isn't possible
skip = (CONF_INTERVAL, CONF_NAME, CONF_LIGHTS)
for k, _, valid in VALIDATION_TUPLES:
if k not in skip:
args[vol.Optional(k)] = valid
return vol.Schema(args)
SET_MANUAL_CONTROL_SCHEMA = vol.Schema(
{
vol.Optional(CONF_ENTITY_ID): cv.entity_ids, # type: ignore[arg-type]

94
custom_components/adaptive_lighting/switch.py
View file

@ -57,6 +57,7 @@ from homeassistant.core import (
State,
callback,
)
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import entity_platform, entity_registry
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_component import async_update_entity
@ -158,7 +159,7 @@ if TYPE_CHECKING:
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import NoEventData, VolDictType
from homeassistant.helpers.typing import NoEventData
_LOGGER = logging.getLogger(__name__)
@ -280,7 +281,7 @@ def _switches_from_service_call(
service_call: ServiceCall,
) -> AdaptiveSwitches:
data = service_call.data
lights = data[CONF_LIGHTS]
lights = data.get(CONF_LIGHTS)
switch_entity_ids: list[str] | None = data.get("entity_id")
if not lights and not switch_entity_ids:
@ -291,7 +292,7 @@ def _switches_from_service_call(
" use case. Currently, you must pass either an adaptive-lighting switch or"
" the lights to an `adaptive_lighting` service call."
)
raise ValueError(msg)
raise ServiceValidationError(msg)
if switch_entity_ids is not None:
if len(switch_entity_ids) > 1 and lights:
@ -299,13 +300,22 @@ def _switches_from_service_call(
"adaptive-lighting: Cannot pass multiple switches with lights argument."
f" Invalid service data received: {service_call.data}"
)
raise ValueError(msg)
raise ServiceValidationError(msg)
switches: AdaptiveSwitches = []
ent_reg = entity_registry.async_get(hass)
for entity_id in switch_entity_ids:
ent_entry = ent_reg.async_get(entity_id)
assert ent_entry is not None
if ent_entry is None:
msg = (
f"adaptive-lighting: Entity '{entity_id}' not found in registry."
)
raise ServiceValidationError(msg)
config_id = ent_entry.config_entry_id
if config_id not in hass.data[DOMAIN]:
msg = (
f"adaptive-lighting: Entity '{entity_id}' does not belong to this integration or is not loaded."
)
raise ServiceValidationError(msg)
switches.append(hass.data[DOMAIN][config_id][SWITCH_DOMAIN])
return switches
@ -317,47 +327,48 @@ def _switches_from_service_call(
"adaptive-lighting: Incorrect data provided in service call."
f" Entities not found in the integration. Service data: {service_call.data}"
)
raise ValueError(msg)
raise ServiceValidationError(msg)
async def handle_change_switch_settings(
switch: AdaptiveSwitch,
hass: HomeAssistant,
service_call: ServiceCall,
) -> None:
"""Allows HASS to change config values via a service call."""
data = service_call.data
which = data.get(CONF_USE_DEFAULTS, "current")
if which == "current": # use whatever we're already using.
defaults = switch._current_settings # pylint: disable=protected-access
elif which == "factory": # use actual defaults listed in the documentation
defaults = None
elif which == "configuration":
# use whatever's in the config flow or configuration.yaml
defaults = switch._config_backup
else:
defaults = None
switches = _switches_from_service_call(hass, service_call)
for switch in switches:
which = data.get(CONF_USE_DEFAULTS, "current")
if which == "current": # use whatever we're already using.
defaults = switch._current_settings # pylint: disable=protected-access
elif which == "factory": # use actual defaults listed in the documentation
defaults = None
elif which == "configuration":
# use whatever's in the config flow or configuration.yaml
defaults = switch._config_backup
else:
defaults = None
# deep copy the defaults so we don't modify the original dicts
switch._set_changeable_settings(data=data, defaults=deepcopy(defaults))
if switch.is_on:
switch._update_time_interval_listener()
# deep copy the defaults so we don't modify the original dicts
switch._set_changeable_settings(data=data, defaults=deepcopy(defaults))
if switch.is_on:
switch._update_time_interval_listener()
_LOGGER.debug(
"Called 'adaptive_lighting.change_switch_settings' service with '%s'",
data,
)
switch.manager.reset(*switch.lights, reset_manual_control=False)
if switch.is_on:
await switch._update_attrs_and_maybe_adapt_lights( # pylint: disable=protected-access
context=switch.create_context("service", parent=service_call.context),
lights=switch.lights,
transition=switch.initial_transition,
force=True,
_LOGGER.debug(
"Called 'adaptive_lighting.change_switch_settings' service with '%s'",
data,
)
switch.manager.reset(*switch.lights, reset_manual_control=False)
if switch.is_on:
await switch._update_attrs_and_maybe_adapt_lights( # pylint: disable=protected-access
context=switch.create_context("service", parent=service_call.context),
lights=switch.lights,
transition=switch.initial_transition,
force=True,
)
@callback
async def handle_apply_service(hass: HomeAssistant, service_call: ServiceCall) -> None:
"""Handle the entity service apply."""
data = service_call.data
@ -392,7 +403,6 @@ async def handle_apply_service(hass: HomeAssistant, service_call: ServiceCall) -
)
@callback
async def handle_set_manual_control_service(
hass: HomeAssistant,
service_call: ServiceCall,
@ -508,20 +518,6 @@ async def async_setup_entry(
update_before_add=True,
)
args: VolDictType = {vol.Optional(CONF_USE_DEFAULTS, default="current"): cv.string}
# Modifying these after init isn't possible
skip = (CONF_INTERVAL, CONF_NAME, CONF_LIGHTS)
for k, _, valid in VALIDATION_TUPLES:
if k not in skip:
args[vol.Optional(k)] = valid
platform = entity_platform.current_platform.get()
assert platform is not None
platform.async_register_entity_service(
SERVICE_CHANGE_SWITCH_SETTINGS,
args,
handle_change_switch_settings,
)
def validate(
config_entry: ConfigEntry | None,

78
tests/test_switch.py
View file

@ -112,7 +112,8 @@ from homeassistant.const import (
STATE_ON,
)
from homeassistant.const import __version__ as ha_version
from homeassistant.core import Context, Event, HomeAssistant, State
from homeassistant.core import Context, Event, HomeAssistant, State, callback
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import area_registry as ar
from homeassistant.helpers import entity_registry
from homeassistant.helpers.entity_platform import async_get_platforms
@ -2914,3 +2915,78 @@ async def test_adapt_only_on_bare_turn_on_respects_pause_changed_mode(hass, inte
f"With take_over_control_mode=PAUSE_CHANGED and only brightness marked "
f"as manually controlled, color_temp should still be adapted."
)
async def test_service_validation_error_invalid_entity(hass):
"""Test that ServiceValidationError is raised for invalid entities."""
await setup_lights_and_switch(hass)
# Test change_switch_settings with non-existent entity
with pytest.raises(ServiceValidationError, match="not found in registry"):
await hass.services.async_call(
DOMAIN,
SERVICE_CHANGE_SWITCH_SETTINGS,
{ATTR_ENTITY_ID: "switch.non_existent"},
blocking=True,
)
async def test_service_validation_error_missing_input(hass):
"""Test that ServiceValidationError is raised for missing input."""
await setup_lights_and_switch(hass)
# Test change_switch_settings with no entity and no lights
with pytest.raises(ServiceValidationError, match="Neither a switch nor a light was provided"):
await hass.services.async_call(
DOMAIN,
SERVICE_CHANGE_SWITCH_SETTINGS,
{},
blocking=True,
)
async def test_change_switch_settings_multiple_entities(hass):
"""Test change_switch_settings with multiple entities."""
# Setup two switches
await setup_lights(hass)
_, switch1 = await setup_switch(hass, {CONF_LIGHTS: [ENTITY_LIGHT_1], CONF_NAME: "switch1"})
_, switch2 = await setup_switch(hass, {CONF_LIGHTS: [ENTITY_LIGHT_2], CONF_NAME: "switch2"})
assert switch1._sun_light_settings.min_color_temp != 3000
assert switch2._sun_light_settings.min_color_temp != 3000
# Call service for both switches
await hass.services.async_call(
DOMAIN,
SERVICE_CHANGE_SWITCH_SETTINGS,
{
ATTR_ENTITY_ID: ["switch.adaptive_lighting_switch1", "switch.adaptive_lighting_switch2"],
"min_color_temp": 3000,
"use_defaults": "configuration" # Required to set new value
},
blocking=True,
)
assert switch1._sun_light_settings.min_color_temp == 3000
assert switch2._sun_light_settings.min_color_temp == 3000
async def test_apply_service_validation(hass):
"""Test validation for apply service."""
await setup_lights_and_switch(hass)
with pytest.raises(ServiceValidationError, match="not found in registry"):
await hass.services.async_call(
DOMAIN,
SERVICE_APPLY,
{ATTR_ENTITY_ID: "switch.non_existent"},
blocking=True,
)
async def test_set_manual_control_validation(hass):
"""Test validation for set_manual_control service."""
await setup_lights_and_switch(hass)
with pytest.raises(ServiceValidationError, match="not found in registry"):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_MANUAL_CONTROL,
{ATTR_ENTITY_ID: "switch.non_existent"},
blocking=True,
)