jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-17 18:27:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cleanup yaml import and legacy file notify service (#130219)

Browse Source
This commit is contained in:
Jan Bouwhuis 2024-11-09 15:54:18 +01:00 committed by GitHub
parent 5f0f29704b
commit 6837ea947c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 18 additions and 434 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
homeassistant/components/file/__init__.py
View File

@ -3,88 +3,19 @@
from copy import deepcopy from copy import deepcopy
from typing import Any from typing import Any
from homeassistant.components.notify import migrate_notify_issue from homeassistant.config_entries import ConfigEntry
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry from homeassistant.const import CONF_FILE_PATH, CONF_NAME, CONF_PLATFORM, Platform
from homeassistant.const import ( from homeassistant.core import HomeAssistant
CONF_FILE_PATH,
CONF_NAME,
CONF_PLATFORM,
CONF_SCAN_INTERVAL,
Platform,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import ( from homeassistant.helpers import config_validation as cv
config_validation as cv,
discovery,
issue_registry as ir,
)
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN from .const import DOMAIN
from .notify import PLATFORM_SCHEMA as NOTIFY_PLATFORM_SCHEMA
from .sensor import PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA
IMPORT_SCHEMA = {
Platform.SENSOR: SENSOR_PLATFORM_SCHEMA,
Platform.NOTIFY: NOTIFY_PLATFORM_SCHEMA,
}
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
PLATFORMS = [Platform.NOTIFY, Platform.SENSOR] PLATFORMS = [Platform.NOTIFY, Platform.SENSOR]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the file integration."""
hass.data[DOMAIN] = config
if hass.config_entries.async_entries(DOMAIN):
# We skip import in case we already have config entries
return True
# The use of the legacy notify service was deprecated with HA Core 2024.6.0
# and will be removed with HA Core 2024.12
migrate_notify_issue(hass, DOMAIN, "File", "2024.12.0")
# The YAML config was imported with HA Core 2024.6.0 and will be removed with
# HA Core 2024.12
ir.async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2024.12.0",
is_fixable=False,
issue_domain=DOMAIN,
learn_more_url="https://www.home-assistant.io/integrations/file/",
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "File",
},
)
# Import the YAML config into separate config entries
platforms_config: dict[Platform, list[ConfigType]] = {
domain: config[domain] for domain in PLATFORMS if domain in config
}
for domain, items in platforms_config.items():
for item in items:
if item[CONF_PLATFORM] == DOMAIN:
file_config_item = IMPORT_SCHEMA[domain](item)
file_config_item[CONF_PLATFORM] = domain
if CONF_SCAN_INTERVAL in file_config_item:
del file_config_item[CONF_SCAN_INTERVAL]
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=file_config_item,
)
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a file component entry.""" """Set up a file component entry."""
config = {**entry.data, **entry.options} config = {**entry.data, **entry.options}
@ -102,20 +33,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
entry, [Platform(entry.data[CONF_PLATFORM])] entry, [Platform(entry.data[CONF_PLATFORM])]
) )
entry.async_on_unload(entry.add_update_listener(update_listener)) entry.async_on_unload(entry.add_update_listener(update_listener))
if entry.data[CONF_PLATFORM] == Platform.NOTIFY and CONF_NAME in entry.data:
# New notify entities are being setup through the config entry,
# but during the deprecation period we want to keep the legacy notify platform,
# so we forward the setup config through discovery.
# Only the entities from yaml will still be available as legacy service.
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.NOTIFY,
DOMAIN,
config,
hass.data[DOMAIN],
)
)
return True return True

23
homeassistant/components/file/config_flow.py
View File

@ -3,7 +3,6 @@
from __future__ import annotations from __future__ import annotations
from copy import deepcopy from copy import deepcopy
import os
from typing import Any from typing import Any
import voluptuous as vol import voluptuous as vol
@ -16,7 +15,6 @@ from homeassistant.config_entries import (
) )
from homeassistant.const import ( from homeassistant.const import (
CONF_FILE_PATH, CONF_FILE_PATH,
CONF_FILENAME,
CONF_NAME, CONF_NAME,
CONF_PLATFORM, CONF_PLATFORM,
CONF_UNIT_OF_MEASUREMENT, CONF_UNIT_OF_MEASUREMENT,
@ -132,27 +130,6 @@ class FileConfigFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle file sensor config flow.""" """Handle file sensor config flow."""
return await self._async_handle_step(Platform.SENSOR.value, user_input) return await self._async_handle_step(Platform.SENSOR.value, user_input)
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
"""Import `file`` config from configuration.yaml."""
self._async_abort_entries_match(import_data)
platform = import_data[CONF_PLATFORM]
name: str = import_data.get(CONF_NAME, DEFAULT_NAME)
file_name: str
if platform == Platform.NOTIFY:
file_name = import_data.pop(CONF_FILENAME)
file_path: str = os.path.join(self.hass.config.config_dir, file_name)
import_data[CONF_FILE_PATH] = file_path
else:
file_path = import_data[CONF_FILE_PATH]
title = f"{name} [{file_path}]"
data = deepcopy(import_data)
options = {}
for key, value in import_data.items():
if key not in (CONF_FILE_PATH, CONF_PLATFORM, CONF_NAME):
data.pop(key)
options[key] = value
return self.async_create_entry(title=title, data=data, options=options)
class FileOptionsFlowHandler(OptionsFlow): class FileOptionsFlowHandler(OptionsFlow):
"""Handle File options.""" """Handle File options."""

83
homeassistant/components/file/notify.py
View File

@ -2,104 +2,23 @@
from __future__ import annotations from __future__ import annotations
from functools import partial
import logging
import os import os
from typing import Any, TextIO from typing import Any, TextIO
import voluptuous as vol
from homeassistant.components.notify import ( from homeassistant.components.notify import (
ATTR_TITLE,
ATTR_TITLE_DEFAULT, ATTR_TITLE_DEFAULT,
PLATFORM_SCHEMA as NOTIFY_PLATFORM_SCHEMA,
BaseNotificationService,
NotifyEntity, NotifyEntity,
NotifyEntityFeature, NotifyEntityFeature,
migrate_notify_issue,
) )
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_FILE_PATH, CONF_FILENAME, CONF_NAME from homeassistant.const import CONF_FILE_PATH, CONF_NAME
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError from homeassistant.exceptions import ServiceValidationError
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from .const import CONF_TIMESTAMP, DEFAULT_NAME, DOMAIN, FILE_ICON from .const import CONF_TIMESTAMP, DEFAULT_NAME, DOMAIN, FILE_ICON
_LOGGER = logging.getLogger(__name__)
# The legacy platform schema uses a filename, after import
# The full file path is stored in the config entry
PLATFORM_SCHEMA = NOTIFY_PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_FILENAME): cv.string,
vol.Optional(CONF_TIMESTAMP, default=False): cv.boolean,
}
)
async def async_get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> FileNotificationService | None:
"""Get the file notification service."""
if discovery_info is None:
# We only set up through discovery
return None
file_path: str = discovery_info[CONF_FILE_PATH]
timestamp: bool = discovery_info[CONF_TIMESTAMP]
return FileNotificationService(file_path, timestamp)
class FileNotificationService(BaseNotificationService):
"""Implement the notification service for the File service."""
def __init__(self, file_path: str, add_timestamp: bool) -> None:
"""Initialize the service."""
self._file_path = file_path
self.add_timestamp = add_timestamp
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send a message to a file."""
# The use of the legacy notify service was deprecated with HA Core 2024.6.0
# and will be removed with HA Core 2024.12
migrate_notify_issue(
self.hass, DOMAIN, "File", "2024.12.0", service_name=self._service_name
)
await self.hass.async_add_executor_job(
partial(self.send_message, message, **kwargs)
)
def send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send a message to a file."""
file: TextIO
filepath = self._file_path
try:
with open(filepath, "a", encoding="utf8") as file:
if os.stat(filepath).st_size == 0:
title = (
f"{kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)} notifications (Log"
f" started: {dt_util.utcnow().isoformat()})\n{'-' * 80}\n"
)
file.write(title)
if self.add_timestamp:
text = f"{dt_util.utcnow().isoformat()} {message}\n"
else:
text = f"{message}\n"
file.write(text)
except OSError as exc:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="write_access_failed",
translation_placeholders={"filename": filepath, "exc": f"{exc!r}"},
) from exc
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistant, hass: HomeAssistant,

31
homeassistant/components/file/sensor.py
View File

@ -6,12 +6,8 @@ import logging
import os import os
from file_read_backwards import FileReadBackwards from file_read_backwards import FileReadBackwards
import voluptuous as vol
from homeassistant.components.sensor import ( from homeassistant.components.sensor import SensorEntity
PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
SensorEntity,
)
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
CONF_FILE_PATH, CONF_FILE_PATH,
@ -20,38 +16,13 @@ from homeassistant.const import (
CONF_VALUE_TEMPLATE, CONF_VALUE_TEMPLATE,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.template import Template from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import DEFAULT_NAME, FILE_ICON from .const import DEFAULT_NAME, FILE_ICON
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_FILE_PATH): cv.isfile,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_VALUE_TEMPLATE): cv.string,
vol.Optional(CONF_UNIT_OF_MEASUREMENT): cv.string,
}
)
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the file sensor from YAML.
The YAML platform config is automatically
imported to a config entry, this method can be removed
when YAML support is removed.
"""
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistant, hass: HomeAssistant,

201
tests/components/file/test_notify.py
View File

@ -12,222 +12,46 @@ from homeassistant.components.file import DOMAIN
from homeassistant.components.notify import ATTR_TITLE_DEFAULT from homeassistant.components.notify import ATTR_TITLE_DEFAULT
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry, assert_setup_component from tests.common import MockConfigEntry
async def test_bad_config(hass: HomeAssistant) -> None:
"""Test set up the platform with bad/missing config."""
config = {notify.DOMAIN: {"name": "test", "platform": "file"}}
with assert_setup_component(0, domain="notify") as handle_config:
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert not handle_config[notify.DOMAIN]
@pytest.mark.parametrize( @pytest.mark.parametrize(
("domain", "service", "params"), ("domain", "service", "params"),
[ [
(notify.DOMAIN, "test", {"message": "one, two, testing, testing"}),
( (
notify.DOMAIN, notify.DOMAIN,
"send_message", "send_message",
{"entity_id": "notify.test", "message": "one, two, testing, testing"}, {"entity_id": "notify.test", "message": "one, two, testing, testing"},
), ),
], ],
ids=["legacy", "entity"],
)
@pytest.mark.parametrize(
("timestamp", "config"),
[
(
False,
{
"notify": [
{
"name": "test",
"platform": "file",
"filename": "mock_file",
"timestamp": False,
}
]
},
),
(
True,
{
"notify": [
{
"name": "test",
"platform": "file",
"filename": "mock_file",
"timestamp": True,
}
]
},
),
],
ids=["no_timestamp", "timestamp"],
) )
@pytest.mark.parametrize("timestamp", [False, True], ids=["no_timestamp", "timestamp"])
async def test_notify_file( async def test_notify_file(
hass: HomeAssistant, hass: HomeAssistant,
freezer: FrozenDateTimeFactory, freezer: FrozenDateTimeFactory,
timestamp: bool,
mock_is_allowed_path: MagicMock, mock_is_allowed_path: MagicMock,
config: ConfigType, timestamp: bool,
domain: str, domain: str,
service: str, service: str,
params: dict[str, str], params: dict[str, str],
) -> None: ) -> None:
"""Test the notify file output.""" """Test the notify file output."""
filename = "mock_file" filename = "mock_file"
message = params["message"] full_filename = os.path.join(hass.config.path(), filename)
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done(wait_background_tasks=True)
freezer.move_to(dt_util.utcnow())
m_open = mock_open()
with (
patch("homeassistant.components.file.notify.open", m_open, create=True),
patch("homeassistant.components.file.notify.os.stat") as mock_st,
):
mock_st.return_value.st_size = 0
title = (
f"{ATTR_TITLE_DEFAULT} notifications "
f"(Log started: {dt_util.utcnow().isoformat()})\n{'-' * 80}\n"
)
await hass.services.async_call(domain, service, params, blocking=True)
full_filename = os.path.join(hass.config.path(), filename)
assert m_open.call_count == 1
assert m_open.call_args == call(full_filename, "a", encoding="utf8")
assert m_open.return_value.write.call_count == 2
if not timestamp:
assert m_open.return_value.write.call_args_list == [
call(title),
call(f"{message}\n"),
]
else:
assert m_open.return_value.write.call_args_list == [
call(title),
call(f"{dt_util.utcnow().isoformat()} {message}\n"),
]
@pytest.mark.parametrize(
("domain", "service", "params"),
[(notify.DOMAIN, "test", {"message": "one, two, testing, testing"})],
ids=["legacy"],
)
@pytest.mark.parametrize(
("is_allowed", "config"),
[
(
True,
{
"notify": [
{
"name": "test",
"platform": "file",
"filename": "mock_file",
}
]
},
),
],
ids=["allowed_but_access_failed"],
)
async def test_legacy_notify_file_exception(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_is_allowed_path: MagicMock,
config: ConfigType,
domain: str,
service: str,
params: dict[str, str],
) -> None:
"""Test legacy notify file output has exception."""
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done(wait_background_tasks=True)
freezer.move_to(dt_util.utcnow())
m_open = mock_open()
with (
patch("homeassistant.components.file.notify.open", m_open, create=True),
patch("homeassistant.components.file.notify.os.stat") as mock_st,
):
mock_st.side_effect = OSError("Access Failed")
with pytest.raises(ServiceValidationError) as exc:
await hass.services.async_call(domain, service, params, blocking=True)
assert f"{exc.value!r}" == "ServiceValidationError('write_access_failed')"
@pytest.mark.parametrize(
("timestamp", "data", "options"),
[
(
False,
{
"name": "test",
"platform": "notify",
"file_path": "mock_file",
},
{
"timestamp": False,
},
),
(
True,
{
"name": "test",
"platform": "notify",
"file_path": "mock_file",
},
{
"timestamp": True,
},
),
],
ids=["no_timestamp", "timestamp"],
)
async def test_legacy_notify_file_entry_only_setup(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
timestamp: bool,
mock_is_allowed_path: MagicMock,
data: dict[str, Any],
options: dict[str, Any],
) -> None:
"""Test the legacy notify file output in entry only setup."""
filename = "mock_file"
domain = notify.DOMAIN
service = "test"
params = {"message": "one, two, testing, testing"}
message = params["message"] message = params["message"]
entry = MockConfigEntry( entry = MockConfigEntry(
domain=DOMAIN, domain=DOMAIN,
data=data, data={"name": "test", "platform": "notify", "file_path": full_filename},
options={"timestamp": timestamp},
version=2, version=2,
options=options, title=f"test [{filename}]",
title=f"test [{data['file_path']}]",
) )
entry.add_to_hass(hass) entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id) assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.async_block_till_done(wait_background_tasks=True)
freezer.move_to(dt_util.utcnow()) freezer.move_to(dt_util.utcnow())
@ -245,7 +69,7 @@ async def test_legacy_notify_file_entry_only_setup(
await hass.services.async_call(domain, service, params, blocking=True) await hass.services.async_call(domain, service, params, blocking=True)
assert m_open.call_count == 1 assert m_open.call_count == 1
assert m_open.call_args == call(filename, "a", encoding="utf8") assert m_open.call_args == call(full_filename, "a", encoding="utf8")
assert m_open.return_value.write.call_count == 2 assert m_open.return_value.write.call_count == 2
if not timestamp: if not timestamp:
@ -277,14 +101,14 @@ async def test_legacy_notify_file_entry_only_setup(
], ],
ids=["not_allowed"], ids=["not_allowed"],
) )
async def test_legacy_notify_file_not_allowed( async def test_notify_file_not_allowed(
hass: HomeAssistant, hass: HomeAssistant,
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
mock_is_allowed_path: MagicMock, mock_is_allowed_path: MagicMock,
config: dict[str, Any], config: dict[str, Any],
options: dict[str, Any], options: dict[str, Any],
) -> None: ) -> None:
"""Test legacy notify file output not allowed.""" """Test notify file output not allowed."""
entry = MockConfigEntry( entry = MockConfigEntry(
domain=DOMAIN, domain=DOMAIN,
data=config, data=config,
@ -301,11 +125,10 @@ async def test_legacy_notify_file_not_allowed(
@pytest.mark.parametrize( @pytest.mark.parametrize(
("service", "params"), ("service", "params"),
[ [
("test", {"message": "one, two, testing, testing"}),
( (
"send_message", "send_message",
{"entity_id": "notify.test", "message": "one, two, testing, testing"}, {"entity_id": "notify.test", "message": "one, two, testing, testing"},
), )
], ],
) )
@pytest.mark.parametrize( @pytest.mark.parametrize(

23
tests/components/file/test_sensor.py
View File

@ -7,33 +7,10 @@ import pytest
from homeassistant.components.file import DOMAIN from homeassistant.components.file import DOMAIN
from homeassistant.const import STATE_UNKNOWN from homeassistant.const import STATE_UNKNOWN
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, get_fixture_path from tests.common import MockConfigEntry, get_fixture_path
@patch("os.path.isfile", Mock(return_value=True))
@patch("os.access", Mock(return_value=True))
async def test_file_value_yaml_setup(
hass: HomeAssistant, mock_is_allowed_path: MagicMock
) -> None:
"""Test the File sensor from YAML setup."""
config = {
"sensor": {
"platform": "file",
"scan_interval": 30,
"name": "file1",
"file_path": get_fixture_path("file_value.txt", "file"),
}
}
assert await async_setup_component(hass, "sensor", config)
await hass.async_block_till_done()
state = hass.states.get("sensor.file1")
assert state.state == "21"
@patch("os.path.isfile", Mock(return_value=True)) @patch("os.path.isfile", Mock(return_value=True))
@patch("os.access", Mock(return_value=True)) @patch("os.access", Mock(return_value=True))
async def test_file_value_entry_setup( async def test_file_value_entry_setup(