mirror of
https://github.com/home-assistant/core.git
synced 2025-07-17 02:07:09 +00:00
Code quality file (#65258)
This commit is contained in:
parent
7806494816
commit
a8304392b5
@ -1,5 +1,8 @@
|
|||||||
"""Support for file notification."""
|
"""Support for file notification."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import os
|
import os
|
||||||
|
from typing import TextIO
|
||||||
|
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
@ -10,7 +13,9 @@ from homeassistant.components.notify import (
|
|||||||
BaseNotificationService,
|
BaseNotificationService,
|
||||||
)
|
)
|
||||||
from homeassistant.const import CONF_FILENAME
|
from homeassistant.const import CONF_FILENAME
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
import homeassistant.helpers.config_validation as cv
|
import homeassistant.helpers.config_validation as cv
|
||||||
|
from homeassistant.helpers.typing import ConfigType
|
||||||
import homeassistant.util.dt as dt_util
|
import homeassistant.util.dt as dt_util
|
||||||
|
|
||||||
CONF_TIMESTAMP = "timestamp"
|
CONF_TIMESTAMP = "timestamp"
|
||||||
@ -23,26 +28,33 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def get_service(hass, config, discovery_info=None):
|
def get_service(
|
||||||
|
hass: HomeAssistant, config: ConfigType, discovery_info=None
|
||||||
|
) -> FileNotificationService:
|
||||||
"""Get the file notification service."""
|
"""Get the file notification service."""
|
||||||
filename = config[CONF_FILENAME]
|
filename: str = config[CONF_FILENAME]
|
||||||
timestamp = config[CONF_TIMESTAMP]
|
timestamp: bool = config[CONF_TIMESTAMP]
|
||||||
|
|
||||||
return FileNotificationService(hass, filename, timestamp)
|
return FileNotificationService(filename, timestamp)
|
||||||
|
|
||||||
|
|
||||||
class FileNotificationService(BaseNotificationService):
|
class FileNotificationService(BaseNotificationService):
|
||||||
"""Implement the notification service for the File service."""
|
"""Implement the notification service for the File service."""
|
||||||
|
|
||||||
def __init__(self, hass, filename, add_timestamp):
|
def __init__(self, filename: str, add_timestamp: bool) -> None:
|
||||||
"""Initialize the service."""
|
"""Initialize the service."""
|
||||||
self.filepath = os.path.join(hass.config.config_dir, filename)
|
self.filename = filename
|
||||||
self.add_timestamp = add_timestamp
|
self.add_timestamp = add_timestamp
|
||||||
|
|
||||||
def send_message(self, message="", **kwargs):
|
def send_message(self, message="", **kwargs) -> None:
|
||||||
"""Send a message to a file."""
|
"""Send a message to a file."""
|
||||||
with open(self.filepath, "a", encoding="utf8") as file:
|
file: TextIO
|
||||||
if os.stat(self.filepath).st_size == 0:
|
if not self.hass.config.config_dir:
|
||||||
|
return
|
||||||
|
|
||||||
|
filepath: str = os.path.join(self.hass.config.config_dir, self.filename)
|
||||||
|
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 started: {dt_util.utcnow().isoformat()})\n{'-' * 80}\n"
|
title = f"{kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)} notifications (Log started: {dt_util.utcnow().isoformat()})\n{'-' * 80}\n"
|
||||||
file.write(title)
|
file.write(title)
|
||||||
|
|
||||||
|
@ -16,6 +16,7 @@ from homeassistant.const import (
|
|||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
import homeassistant.helpers.config_validation as cv
|
import homeassistant.helpers.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.typing import ConfigType, DiscoveryInfoType
|
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
@ -41,11 +42,12 @@ async def async_setup_platform(
|
|||||||
discovery_info: DiscoveryInfoType | None = None,
|
discovery_info: DiscoveryInfoType | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Set up the file sensor."""
|
"""Set up the file sensor."""
|
||||||
file_path = config[CONF_FILE_PATH]
|
file_path: str = config[CONF_FILE_PATH]
|
||||||
name = config[CONF_NAME]
|
name: str = config[CONF_NAME]
|
||||||
unit = config.get(CONF_UNIT_OF_MEASUREMENT)
|
unit: str | None = config.get(CONF_UNIT_OF_MEASUREMENT)
|
||||||
|
value_template: Template | None = config.get(CONF_VALUE_TEMPLATE)
|
||||||
|
|
||||||
if (value_template := config.get(CONF_VALUE_TEMPLATE)) is not None:
|
if value_template is not None:
|
||||||
value_template.hass = hass
|
value_template.hass = hass
|
||||||
|
|
||||||
if hass.config.is_allowed_path(file_path):
|
if hass.config.is_allowed_path(file_path):
|
||||||
@ -57,33 +59,20 @@ async def async_setup_platform(
|
|||||||
class FileSensor(SensorEntity):
|
class FileSensor(SensorEntity):
|
||||||
"""Implementation of a file sensor."""
|
"""Implementation of a file sensor."""
|
||||||
|
|
||||||
def __init__(self, name, file_path, unit_of_measurement, value_template):
|
_attr_icon = ICON
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
name: str,
|
||||||
|
file_path: str,
|
||||||
|
unit_of_measurement: str | None,
|
||||||
|
value_template: Template | None,
|
||||||
|
) -> None:
|
||||||
"""Initialize the file sensor."""
|
"""Initialize the file sensor."""
|
||||||
self._name = name
|
self._attr_name = name
|
||||||
self._file_path = file_path
|
self._file_path = file_path
|
||||||
self._unit_of_measurement = unit_of_measurement
|
self._attr_native_unit_of_measurement = unit_of_measurement
|
||||||
self._val_tpl = value_template
|
self._val_tpl = value_template
|
||||||
self._state = None
|
|
||||||
|
|
||||||
@property
|
|
||||||
def name(self):
|
|
||||||
"""Return the name of the sensor."""
|
|
||||||
return self._name
|
|
||||||
|
|
||||||
@property
|
|
||||||
def native_unit_of_measurement(self):
|
|
||||||
"""Return the unit the value is expressed in."""
|
|
||||||
return self._unit_of_measurement
|
|
||||||
|
|
||||||
@property
|
|
||||||
def icon(self):
|
|
||||||
"""Return the icon to use in the frontend, if any."""
|
|
||||||
return ICON
|
|
||||||
|
|
||||||
@property
|
|
||||||
def native_value(self):
|
|
||||||
"""Return the state of the sensor."""
|
|
||||||
return self._state
|
|
||||||
|
|
||||||
def update(self):
|
def update(self):
|
||||||
"""Get the latest entry from a file and updates the state."""
|
"""Get the latest entry from a file and updates the state."""
|
||||||
@ -100,8 +89,8 @@ class FileSensor(SensorEntity):
|
|||||||
return
|
return
|
||||||
|
|
||||||
if self._val_tpl is not None:
|
if self._val_tpl is not None:
|
||||||
self._state = self._val_tpl.async_render_with_possible_json_value(
|
self._attr_native_value = (
|
||||||
data, None
|
self._val_tpl.async_render_with_possible_json_value(data, None)
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
self._state = data
|
self._attr_native_value = data
|
||||||
|
@ -4,15 +4,16 @@ from unittest.mock import call, mock_open, patch
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
import homeassistant.components.notify as notify
|
from homeassistant.components import notify
|
||||||
from homeassistant.components.notify import ATTR_TITLE_DEFAULT
|
from homeassistant.components.notify import ATTR_TITLE_DEFAULT
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.setup import async_setup_component
|
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 assert_setup_component
|
from tests.common import assert_setup_component
|
||||||
|
|
||||||
|
|
||||||
async def test_bad_config(hass):
|
async def test_bad_config(hass: HomeAssistant):
|
||||||
"""Test set up the platform with bad/missing config."""
|
"""Test set up the platform with bad/missing config."""
|
||||||
config = {notify.DOMAIN: {"name": "test", "platform": "file"}}
|
config = {notify.DOMAIN: {"name": "test", "platform": "file"}}
|
||||||
with assert_setup_component(0) as handle_config:
|
with assert_setup_component(0) as handle_config:
|
||||||
@ -27,7 +28,7 @@ async def test_bad_config(hass):
|
|||||||
True,
|
True,
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
async def test_notify_file(hass, timestamp):
|
async def test_notify_file(hass: HomeAssistant, timestamp: bool):
|
||||||
"""Test the notify file output."""
|
"""Test the notify file output."""
|
||||||
filename = "mock_file"
|
filename = "mock_file"
|
||||||
message = "one, two, testing, testing"
|
message = "one, two, testing, testing"
|
||||||
|
@ -4,6 +4,7 @@ from unittest.mock import Mock, mock_open, patch
|
|||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from homeassistant.const import STATE_UNKNOWN
|
from homeassistant.const import STATE_UNKNOWN
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
|
||||||
from tests.common import mock_registry
|
from tests.common import mock_registry
|
||||||
@ -17,7 +18,7 @@ def entity_reg(hass):
|
|||||||
|
|
||||||
@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(hass, entity_reg):
|
async def test_file_value(hass: HomeAssistant) -> None:
|
||||||
"""Test the File sensor."""
|
"""Test the File sensor."""
|
||||||
config = {
|
config = {
|
||||||
"sensor": {"platform": "file", "name": "file1", "file_path": "mock.file1"}
|
"sensor": {"platform": "file", "name": "file1", "file_path": "mock.file1"}
|
||||||
@ -36,7 +37,7 @@ async def test_file_value(hass, entity_reg):
|
|||||||
|
|
||||||
@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_template(hass, entity_reg):
|
async def test_file_value_template(hass: HomeAssistant) -> None:
|
||||||
"""Test the File sensor with JSON entries."""
|
"""Test the File sensor with JSON entries."""
|
||||||
config = {
|
config = {
|
||||||
"sensor": {
|
"sensor": {
|
||||||
@ -47,7 +48,9 @@ async def test_file_value_template(hass, entity_reg):
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
data = '{"temperature": 29, "humidity": 31}\n' '{"temperature": 26, "humidity": 36}'
|
data = (
|
||||||
|
'{"temperature": 29, "humidity": 31}\n' + '{"temperature": 26, "humidity": 36}'
|
||||||
|
)
|
||||||
|
|
||||||
m_open = mock_open(read_data=data)
|
m_open = mock_open(read_data=data)
|
||||||
with patch(
|
with patch(
|
||||||
@ -62,7 +65,7 @@ async def test_file_value_template(hass, entity_reg):
|
|||||||
|
|
||||||
@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_empty(hass, entity_reg):
|
async def test_file_empty(hass: HomeAssistant) -> None:
|
||||||
"""Test the File sensor with an empty file."""
|
"""Test the File sensor with an empty file."""
|
||||||
config = {"sensor": {"platform": "file", "name": "file3", "file_path": "mock.file"}}
|
config = {"sensor": {"platform": "file", "name": "file3", "file_path": "mock.file"}}
|
||||||
|
|
||||||
@ -75,3 +78,21 @@ async def test_file_empty(hass, entity_reg):
|
|||||||
|
|
||||||
state = hass.states.get("sensor.file3")
|
state = hass.states.get("sensor.file3")
|
||||||
assert state.state == STATE_UNKNOWN
|
assert state.state == STATE_UNKNOWN
|
||||||
|
|
||||||
|
|
||||||
|
@patch("os.path.isfile", Mock(return_value=True))
|
||||||
|
@patch("os.access", Mock(return_value=True))
|
||||||
|
async def test_file_path_invalid(hass: HomeAssistant) -> None:
|
||||||
|
"""Test the File sensor with invalid path."""
|
||||||
|
config = {
|
||||||
|
"sensor": {"platform": "file", "name": "file4", "file_path": "mock.file4"}
|
||||||
|
}
|
||||||
|
|
||||||
|
m_open = mock_open(read_data="43\n45\n21")
|
||||||
|
with patch(
|
||||||
|
"homeassistant.components.file.sensor.open", m_open, create=True
|
||||||
|
), patch.object(hass.config, "is_allowed_path", return_value=False):
|
||||||
|
assert await async_setup_component(hass, "sensor", config)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
assert len(hass.states.async_entity_ids("sensor")) == 0
|
||||||
|
Loading…
x
Reference in New Issue
Block a user