Remove legacy knx notify service (#128185)

This commit is contained in:
G Johansson 2024-10-11 20:19:15 +02:00 committed by GitHub
parent 8540343d7f
commit 2b2820018c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 3 additions and 217 deletions

View File

@ -29,7 +29,6 @@ from homeassistant.const import (
)
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import DeviceEntry
from homeassistant.helpers.reload import async_integration_yaml_config
@ -193,14 +192,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
},
)
# set up notify service for backwards compatibility - remove 2024.11
if NotifySchema.PLATFORM in config:
hass.async_create_task(
discovery.async_load_platform(
hass, Platform.NOTIFY, DOMAIN, {}, hass.data[DATA_HASS_CONFIG]
)
)
await register_panel(hass)
return True

View File

@ -2,86 +2,21 @@
from __future__ import annotations
from typing import Any
from xknx import XKNX
from xknx.devices import Notification as XknxNotification
from homeassistant import config_entries
from homeassistant.components.notify import (
BaseNotificationService,
NotifyEntity,
migrate_notify_issue,
)
from homeassistant.components.notify import NotifyEntity
from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME, CONF_TYPE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import ConfigType
from . import KNXModule
from .const import DOMAIN, KNX_ADDRESS, KNX_MODULE_KEY
from .const import KNX_ADDRESS, KNX_MODULE_KEY
from .entity import KnxYamlEntity
async def async_get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> KNXNotificationService | None:
"""Get the KNX notification service."""
if discovery_info is None:
return None
knx_module = hass.data[KNX_MODULE_KEY]
if platform_config := knx_module.config_yaml.get(Platform.NOTIFY):
xknx: XKNX = hass.data[KNX_MODULE_KEY].xknx
notification_devices = [
_create_notification_instance(xknx, device_config)
for device_config in platform_config
]
return KNXNotificationService(notification_devices)
return None
class KNXNotificationService(BaseNotificationService):
"""Implement notification service."""
def __init__(self, devices: list[XknxNotification]) -> None:
"""Initialize the service."""
self.devices = devices
@property
def targets(self) -> dict[str, str]:
"""Return a dictionary of registered targets."""
ret = {}
for device in self.devices:
ret[device.name] = device.name
return ret
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send a notification to knx bus."""
migrate_notify_issue(
self.hass, DOMAIN, "KNX", "2024.11.0", service_name=self._service_name
)
if "target" in kwargs:
await self._async_send_to_device(message, kwargs["target"])
else:
await self._async_send_to_all_devices(message)
async def _async_send_to_all_devices(self, message: str) -> None:
"""Send a notification to knx bus to all connected devices."""
for device in self.devices:
await device.set(message)
async def _async_send_to_device(self, message: str, names: str) -> None:
"""Send a notification to knx bus to device with given names."""
for device in self.devices:
if device.name in names:
await device.set(message)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: config_entries.ConfigEntry,

View File

@ -9,74 +9,6 @@ from homeassistant.core import HomeAssistant
from .conftest import KNXTestKit
async def test_legacy_notify_service_simple(
hass: HomeAssistant, knx: KNXTestKit
) -> None:
"""Test KNX notify can send to one device."""
await knx.setup_integration(
{
NotifySchema.PLATFORM: {
CONF_NAME: "test",
KNX_ADDRESS: "1/0/0",
}
}
)
await hass.services.async_call(
"notify", "notify", {"target": "test", "message": "I love KNX"}, blocking=True
)
await knx.assert_write(
"1/0/0",
(73, 32, 108, 111, 118, 101, 32, 75, 78, 88, 0, 0, 0, 0),
)
await hass.services.async_call(
"notify",
"notify",
{
"target": "test",
"message": "I love KNX, but this text is too long for KNX, poor KNX",
},
blocking=True,
)
await knx.assert_write(
"1/0/0",
(73, 32, 108, 111, 118, 101, 32, 75, 78, 88, 44, 32, 98, 117),
)
async def test_legacy_notify_service_multiple_sends_to_all_with_different_encodings(
hass: HomeAssistant, knx: KNXTestKit
) -> None:
"""Test KNX notify `type` configuration."""
await knx.setup_integration(
{
NotifySchema.PLATFORM: [
{
CONF_NAME: "ASCII",
KNX_ADDRESS: "1/0/0",
CONF_TYPE: "string",
},
{
CONF_NAME: "Latin-1",
KNX_ADDRESS: "1/0/1",
CONF_TYPE: "latin_1",
},
]
}
)
await hass.services.async_call(
"notify", "notify", {"message": "Gänsefüßchen"}, blocking=True
)
await knx.assert_write(
"1/0/0",
# "G?nsef??chen"
(71, 63, 110, 115, 101, 102, 63, 63, 99, 104, 101, 110, 0, 0),
)
await knx.assert_write(
"1/0/1",
(71, 228, 110, 115, 101, 102, 252, 223, 99, 104, 101, 110, 0, 0),
)
async def test_notify_simple(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test KNX notify can send to one device."""
await knx.setup_integration(

View File

@ -1,72 +0,0 @@
"""Test repairs for KNX integration."""
from homeassistant.components.knx.const import DOMAIN, KNX_ADDRESS
from homeassistant.components.knx.schema import NotifySchema
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
import homeassistant.helpers.issue_registry as ir
from .conftest import KNXTestKit
from tests.components.repairs import process_repair_fix_flow, start_repair_fix_flow
from tests.typing import ClientSessionGenerator
async def test_knx_notify_service_issue(
hass: HomeAssistant,
knx: KNXTestKit,
hass_client: ClientSessionGenerator,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test the legacy notify service still works before migration and repair flow is triggered."""
await knx.setup_integration(
{
NotifySchema.PLATFORM: {
CONF_NAME: "test",
KNX_ADDRESS: "1/0/0",
}
}
)
http_client = await hass_client()
# Assert no issue is present
assert len(issue_registry.issues) == 0
# Simulate legacy service being used
assert hass.services.has_service(NOTIFY_DOMAIN, NOTIFY_DOMAIN)
await hass.services.async_call(
NOTIFY_DOMAIN,
NOTIFY_DOMAIN,
service_data={"message": "It is too cold!", "target": "test"},
blocking=True,
)
await knx.assert_write(
"1/0/0",
(73, 116, 32, 105, 115, 32, 116, 111, 111, 32, 99, 111, 108, 100),
)
# Assert the issue is present
assert len(issue_registry.issues) == 1
assert issue_registry.async_get_issue(
domain="notify",
issue_id=f"migrate_notify_{DOMAIN}_notify",
)
# Test confirm step in repair flow
data = await start_repair_fix_flow(
http_client, "notify", f"migrate_notify_{DOMAIN}_notify"
)
flow_id = data["flow_id"]
assert data["step_id"] == "confirm"
data = await process_repair_fix_flow(http_client, flow_id)
assert data["type"] == "create_entry"
# Assert the issue is no longer present
assert not issue_registry.async_get_issue(
domain="notify",
issue_id=f"migrate_notify_{DOMAIN}_notify",
)
assert len(issue_registry.issues) == 0