jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 00:37:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rework velbus services to deprecated the interface parameter (#134816)

Browse Source 
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Maikel Punie 2025-01-13 11:36:20 +01:00 committed by GitHub
parent 98ef32c668
commit 2d67aca550
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 344 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
homeassistant/components/velbus/const.py
View File

@ -11,6 +11,7 @@ from homeassistant.components.climate import (
DOMAIN: Final = "velbus"
CONF_CONFIG_ENTRY: Final = "config_entry"
CONF_INTERFACE: Final = "interface"
CONF_MEMO_TEXT: Final = "memo_text"

164
homeassistant/components/velbus/services.py
View File

@ -9,15 +9,19 @@ from typing import TYPE_CHECKING
import voluptuous as vol
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import config_validation as cv
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import config_validation as cv, selector
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.storage import STORAGE_DIR
if TYPE_CHECKING:
from . import VelbusConfigEntry
from .const import (
CONF_CONFIG_ENTRY,
CONF_INTERFACE,
CONF_MEMO_TEXT,
DOMAIN,
@ -32,6 +36,7 @@ def setup_services(hass: HomeAssistant) -> None:
"""Register the velbus services."""
def check_entry_id(interface: str) -> str:
"""Check the config_entry for a specific interface."""
for config_entry in hass.config_entries.async_entries(DOMAIN):
if "port" in config_entry.data and config_entry.data["port"] == interface:
return config_entry.entry_id
@ -39,51 +44,71 @@ def setup_services(hass: HomeAssistant) -> None:
"The interface provided is not defined as a port in a Velbus integration"
)
def get_config_entry(interface: str) -> VelbusConfigEntry | None:
for config_entry in hass.config_entries.async_entries(DOMAIN):
if "port" in config_entry.data and config_entry.data["port"] == interface:
return config_entry
return None
async def get_config_entry(call: ServiceCall) -> VelbusConfigEntry:
"""Get the config entry for this service call."""
if CONF_CONFIG_ENTRY in call.data:
entry_id = call.data[CONF_CONFIG_ENTRY]
elif CONF_INTERFACE in call.data:
# Deprecated in 2025.2, to remove in 2025.8
async_create_issue(
hass,
DOMAIN,
"deprecated_interface_parameter",
breaks_in_ha_version="2025.8.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_interface_parameter",
)
entry_id = call.data[CONF_INTERFACE]
if not (entry := hass.config_entries.async_get_entry(entry_id)):
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="integration_not_found",
translation_placeholders={"target": DOMAIN},
)
if entry.state is not ConfigEntryState.LOADED:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="not_loaded",
translation_placeholders={"target": entry.title},
)
return entry
async def scan(call: ServiceCall) -> None:
"""Handle a scan service call."""
entry = get_config_entry(call.data[CONF_INTERFACE])
if entry:
await entry.runtime_data.controller.scan()
entry = await get_config_entry(call)
await entry.runtime_data.controller.scan()
async def syn_clock(call: ServiceCall) -> None:
"""Handle a sync clock service call."""
entry = get_config_entry(call.data[CONF_INTERFACE])
if entry:
await entry.runtime_data.controller.sync_clock()
entry = await get_config_entry(call)
await entry.runtime_data.controller.sync_clock()
async def set_memo_text(call: ServiceCall) -> None:
"""Handle Memo Text service call."""
entry = get_config_entry(call.data[CONF_INTERFACE])
if entry:
memo_text = call.data[CONF_MEMO_TEXT]
module = entry.runtime_data.controller.get_module(call.data[CONF_ADDRESS])
if module:
await module.set_memo_text(memo_text.async_render())
entry = await get_config_entry(call)
memo_text = call.data[CONF_MEMO_TEXT]
module = entry.runtime_data.controller.get_module(call.data[CONF_ADDRESS])
if not module:
raise ServiceValidationError("Module not found")
await module.set_memo_text(memo_text.async_render())
async def clear_cache(call: ServiceCall) -> None:
"""Handle a clear cache service call."""
# clear the cache
entry = await get_config_entry(call)
with suppress(FileNotFoundError):
if call.data.get(CONF_ADDRESS):
await hass.async_add_executor_job(
os.unlink,
hass.config.path(
STORAGE_DIR,
f"velbuscache-{call.data[CONF_INTERFACE]}/{call.data[CONF_ADDRESS]}.p",
f"velbuscache-{entry.entry_id}/{call.data[CONF_ADDRESS]}.p",
),
)
else:
await hass.async_add_executor_job(
shutil.rmtree,
hass.config.path(
STORAGE_DIR, f"velbuscache-{call.data[CONF_INTERFACE]}/"
),
hass.config.path(STORAGE_DIR, f"velbuscache-{entry.entry_id}/"),
)
# call a scan to repopulate
await scan(call)
@ -92,28 +117,73 @@ def setup_services(hass: HomeAssistant) -> None:
DOMAIN,
SERVICE_SCAN,
scan,
vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}),
vol.Any(
vol.Schema(
{
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
}
),
vol.Schema(
{
vol.Required(CONF_CONFIG_ENTRY): selector.ConfigEntrySelector(
{
"integration": DOMAIN,
}
)
}
),
),
)
hass.services.async_register(
DOMAIN,
SERVICE_SYNC,
syn_clock,
vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}),
vol.Any(
vol.Schema(
{
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
}
),
vol.Schema(
{
vol.Required(CONF_CONFIG_ENTRY): selector.ConfigEntrySelector(
{
"integration": DOMAIN,
}
)
}
),
),
)
hass.services.async_register(
DOMAIN,
SERVICE_SET_MEMO_TEXT,
set_memo_text,
vol.Schema(
{
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
vol.Required(CONF_ADDRESS): vol.All(
vol.Coerce(int), vol.Range(min=0, max=255)
),
vol.Optional(CONF_MEMO_TEXT, default=""): cv.template,
}
vol.Any(
vol.Schema(
{
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
vol.Required(CONF_ADDRESS): vol.All(
vol.Coerce(int), vol.Range(min=0, max=255)
),
vol.Optional(CONF_MEMO_TEXT, default=""): cv.template,
}
),
vol.Schema(
{
vol.Required(CONF_CONFIG_ENTRY): selector.ConfigEntrySelector(
{
"integration": DOMAIN,
}
),
vol.Required(CONF_ADDRESS): vol.All(
vol.Coerce(int), vol.Range(min=0, max=255)
),
vol.Optional(CONF_MEMO_TEXT, default=""): cv.template,
}
),
),
)
@ -121,12 +191,26 @@ def setup_services(hass: HomeAssistant) -> None:
DOMAIN,
SERVICE_CLEAR_CACHE,
clear_cache,
vol.Schema(
{
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
vol.Optional(CONF_ADDRESS): vol.All(
vol.Coerce(int), vol.Range(min=0, max=255)
),
}
vol.Any(
vol.Schema(
{
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
vol.Optional(CONF_ADDRESS): vol.All(
vol.Coerce(int), vol.Range(min=0, max=255)
),
}
),
vol.Schema(
{
vol.Required(CONF_CONFIG_ENTRY): selector.ConfigEntrySelector(
{
"integration": DOMAIN,
}
),
vol.Optional(CONF_ADDRESS): vol.All(
vol.Coerce(int), vol.Range(min=0, max=255)
),
}
),
),
)

20
homeassistant/components/velbus/services.yaml
View File

@ -1,29 +1,38 @@
sync_clock:
fields:
interface:
required: true
example: "192.168.1.5:27015"
default: ""
selector:
text:
config_entry:
selector:
config_entry:
integration: velbus
scan:
fields:
interface:
required: true
example: "192.168.1.5:27015"
default: ""
selector:
text:
config_entry:
selector:
config_entry:
integration: velbus
clear_cache:
fields:
interface:
required: true
example: "192.168.1.5:27015"
default: ""
selector:
text:
config_entry:
selector:
config_entry:
integration: velbus
address:
required: false
selector:
@ -34,11 +43,14 @@ clear_cache:
set_memo_text:
fields:
interface:
required: true
example: "192.168.1.5:27015"
default: ""
selector:
text:
config_entry:
selector:
config_entry:
integration: velbus
address:
required: true
selector:

28
homeassistant/components/velbus/strings.json
View File

@ -20,6 +20,12 @@
"exceptions": {
"invalid_hvac_mode": {
"message": "Climate mode {hvac_mode} is not supported."
},
"not_loaded": {
"message": "{target} is not loaded."
},
"integration_not_found": {
"message": "Integration \"{target}\" not found in registry."
}
},
"services": {
@ -30,6 +36,10 @@
"interface": {
"name": "Interface",
"description": "The velbus interface to send the command to, this will be the same value as used during configuration."
},
"config_entry": {
"name": "Config entry",
"description": "The config entry of the velbus integration"
}
}
},
@ -40,6 +50,10 @@
"interface": {
"name": "[%key:component::velbus::services::sync_clock::fields::interface::name%]",
"description": "[%key:component::velbus::services::sync_clock::fields::interface::description%]"
},
"config_entry": {
"name": "[%key:component::velbus::services::sync_clock::fields::config_entry::name%]",
"description": "[%key:component::velbus::services::sync_clock::fields::config_entry::description%]"
}
}
},
@ -51,6 +65,10 @@
"name": "[%key:component::velbus::services::sync_clock::fields::interface::name%]",
"description": "[%key:component::velbus::services::sync_clock::fields::interface::description%]"
},
"config_entry": {
"name": "[%key:component::velbus::services::sync_clock::fields::config_entry::name%]",
"description": "[%key:component::velbus::services::sync_clock::fields::config_entry::description%]"
},
"address": {
"name": "Address",
"description": "The module address in decimal format, if this is provided we only clear this module, if nothing is provided we clear the whole cache directory (all modules) The decimal addresses are displayed in front of the modules listed at the integration page."
@ -65,6 +83,10 @@
"name": "[%key:component::velbus::services::sync_clock::fields::interface::name%]",
"description": "[%key:component::velbus::services::sync_clock::fields::interface::description%]"
},
"config_entry": {
"name": "[%key:component::velbus::services::sync_clock::fields::config_entry::name%]",
"description": "[%key:component::velbus::services::sync_clock::fields::config_entry::description%]"
},
"address": {
"name": "Address",
"description": "The module address in decimal format. The decimal addresses are displayed in front of the modules listed at the integration page."
@ -75,5 +97,11 @@
}
}
}
},
"issues": {
"deprecated_interface_parameter": {
"title": "Deprecated 'interface' parameter",
"description": "The 'interface' parameter in the Velbus service calls is deprecated. The 'config_entry' parameter should be used going forward.\n\nPlease adjust your automations or scripts to fix this issue."
}
}
}

5
tests/components/velbus/conftest.py
View File

@ -1,7 +1,7 @@
"""Fixtures for the Velbus tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, patch
import pytest
from velbusaio.channels import (
@ -72,6 +72,7 @@ def mock_controller(
4: mock_module_no_subdevices,
99: mock_module_subdevices,
}
cont.get_module.return_value = mock_module_subdevices
yield controller
@ -300,7 +301,7 @@ def mock_cover_no_position() -> AsyncMock:
@pytest.fixture(name="config_entry")
async def mock_config_entry(
hass: HomeAssistant,
controller: MagicMock,
controller: AsyncMock,
) -> VelbusConfigEntry:
"""Create and register mock config entry."""
config_entry = MockConfigEntry(

172
tests/components/velbus/test_services.py Normal file
View File

@ -0,0 +1,172 @@
"""Velbus services tests."""
from unittest.mock import AsyncMock
import pytest
import voluptuous as vol
from homeassistant.components.velbus.const import (
CONF_CONFIG_ENTRY,
CONF_INTERFACE,
CONF_MEMO_TEXT,
DOMAIN,
SERVICE_CLEAR_CACHE,
SERVICE_SCAN,
SERVICE_SET_MEMO_TEXT,
SERVICE_SYNC,
)
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
import homeassistant.helpers.issue_registry as ir
from . import init_integration
from tests.common import MockConfigEntry
async def test_global_services_with_interface(
hass: HomeAssistant,
config_entry: MockConfigEntry,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test services directed at the bus with an interface parameter."""
await init_integration(hass, config_entry)
await hass.services.async_call(
DOMAIN,
SERVICE_SCAN,
{CONF_INTERFACE: config_entry.data["port"]},
blocking=True,
)
config_entry.runtime_data.controller.scan.assert_called_once_with()
assert issue_registry.async_get_issue(DOMAIN, "deprecated_interface_parameter")
await hass.services.async_call(
DOMAIN,
SERVICE_SYNC,
{CONF_INTERFACE: config_entry.data["port"]},
blocking=True,
)
config_entry.runtime_data.controller.sync_clock.assert_called_once_with()
# Test invalid interface
with pytest.raises(vol.error.MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_SCAN,
{CONF_INTERFACE: "nonexistent"},
blocking=True,
)
# Test missing interface
with pytest.raises(vol.error.MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_SCAN,
{},
blocking=True,
)
async def test_global_survices_with_config_entry(
hass: HomeAssistant,
config_entry: MockConfigEntry,
) -> None:
"""Test services directed at the bus with a config_entry."""
await init_integration(hass, config_entry)
await hass.services.async_call(
DOMAIN,
SERVICE_SCAN,
{CONF_CONFIG_ENTRY: config_entry.entry_id},
blocking=True,
)
config_entry.runtime_data.controller.scan.assert_called_once_with()
await hass.services.async_call(
DOMAIN,
SERVICE_SYNC,
{CONF_CONFIG_ENTRY: config_entry.entry_id},
blocking=True,
)
config_entry.runtime_data.controller.sync_clock.assert_called_once_with()
# Test invalid interface
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
DOMAIN,
SERVICE_SCAN,
{CONF_CONFIG_ENTRY: "nonexistent"},
blocking=True,
)
# Test missing interface
with pytest.raises(vol.error.MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_SCAN,
{},
blocking=True,
)
async def test_set_memo_text(
hass: HomeAssistant,
config_entry: MockConfigEntry,
controller: AsyncMock,
) -> None:
"""Test the set_memo_text service."""
await init_integration(hass, config_entry)
await hass.services.async_call(
DOMAIN,
SERVICE_SET_MEMO_TEXT,
{
CONF_CONFIG_ENTRY: config_entry.entry_id,
CONF_MEMO_TEXT: "Test",
CONF_ADDRESS: 1,
},
blocking=True,
)
config_entry.runtime_data.controller.get_module(
1
).set_memo_text.assert_called_once_with("Test")
# Test with unfound module
controller.return_value.get_module.return_value = None
with pytest.raises(ServiceValidationError, match="Module not found"):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_MEMO_TEXT,
{
CONF_CONFIG_ENTRY: config_entry.entry_id,
CONF_MEMO_TEXT: "Test",
CONF_ADDRESS: 2,
},
blocking=True,
)
async def test_clear_cache(
hass: HomeAssistant,
config_entry: MockConfigEntry,
) -> None:
"""Test the clear_cache service."""
await init_integration(hass, config_entry)
await hass.services.async_call(
DOMAIN,
SERVICE_CLEAR_CACHE,
{CONF_CONFIG_ENTRY: config_entry.entry_id},
blocking=True,
)
config_entry.runtime_data.controller.scan.assert_called_once_with()
await hass.services.async_call(
DOMAIN,
SERVICE_CLEAR_CACHE,
{CONF_CONFIG_ENTRY: config_entry.entry_id, CONF_ADDRESS: 1},
blocking=True,
)
assert config_entry.runtime_data.controller.scan.call_count == 2