Implement config flow for filesize (#67668)
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
parent
f5923be4e4
commit
67cf053260
19 changed files with 551 additions and 148 deletions
|
@ -336,7 +336,6 @@ omit =
|
|||
homeassistant/components/fastdotcom/*
|
||||
homeassistant/components/ffmpeg/camera.py
|
||||
homeassistant/components/fibaro/*
|
||||
homeassistant/components/filesize/sensor.py
|
||||
homeassistant/components/fints/sensor.py
|
||||
homeassistant/components/fireservicerota/__init__.py
|
||||
homeassistant/components/fireservicerota/binary_sensor.py
|
||||
|
|
|
@ -82,6 +82,7 @@ homeassistant.components.esphome.*
|
|||
homeassistant.components.energy.*
|
||||
homeassistant.components.evil_genius_labs.*
|
||||
homeassistant.components.fastdotcom.*
|
||||
homeassistant.components.filesize.*
|
||||
homeassistant.components.fitbit.*
|
||||
homeassistant.components.flunearyou.*
|
||||
homeassistant.components.flux_led.*
|
||||
|
|
|
@ -308,6 +308,8 @@ tests/components/fan/* @home-assistant/core
|
|||
homeassistant/components/fastdotcom/* @rohankapoorcom
|
||||
homeassistant/components/file/* @fabaff
|
||||
tests/components/file/* @fabaff
|
||||
homeassistant/components/filesize/* @gjohansson-ST
|
||||
tests/components/filesize/* @gjohansson-ST
|
||||
homeassistant/components/filter/* @dgomes
|
||||
tests/components/filter/* @dgomes
|
||||
homeassistant/components/fireservicerota/* @cyberjunky
|
||||
|
|
|
@ -1,6 +1,35 @@
|
|||
"""The filesize component."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.const import Platform
|
||||
import logging
|
||||
import pathlib
|
||||
|
||||
DOMAIN = "filesize"
|
||||
PLATFORMS = [Platform.SENSOR]
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_FILE_PATH
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
|
||||
from .const import PLATFORMS
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up from a config entry."""
|
||||
|
||||
path = entry.data[CONF_FILE_PATH]
|
||||
get_path = await hass.async_add_executor_job(pathlib.Path, path)
|
||||
|
||||
if not get_path.exists() and not get_path.is_file():
|
||||
raise ConfigEntryNotReady(f"Can not access file {path}")
|
||||
|
||||
if not hass.config.is_allowed_path(path):
|
||||
raise ConfigEntryNotReady(f"Filepath {path} is not valid or allowed")
|
||||
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
|
80
homeassistant/components/filesize/config_flow.py
Normal file
80
homeassistant/components/filesize/config_flow.py
Normal file
|
@ -0,0 +1,80 @@
|
|||
"""The filesize config flow."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import pathlib
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigFlow
|
||||
from homeassistant.const import CONF_FILE_PATH
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
DATA_SCHEMA = vol.Schema({vol.Required(CONF_FILE_PATH): str})
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def validate_path(hass: HomeAssistant, path: str) -> pathlib.Path:
|
||||
"""Validate path."""
|
||||
try:
|
||||
get_path = pathlib.Path(path)
|
||||
except OSError as error:
|
||||
_LOGGER.error("Can not access file %s, error %s", path, error)
|
||||
raise NotValidError from error
|
||||
|
||||
if not hass.config.is_allowed_path(path):
|
||||
_LOGGER.error("Filepath %s is not valid or allowed", path)
|
||||
raise NotAllowedError
|
||||
|
||||
return get_path
|
||||
|
||||
|
||||
class FilesizeConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Config flow for Filesize."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow initialized by the user."""
|
||||
errors: dict[str, Any] = {}
|
||||
|
||||
if user_input is not None:
|
||||
try:
|
||||
get_path = validate_path(self.hass, user_input[CONF_FILE_PATH])
|
||||
except NotValidError:
|
||||
errors["base"] = "not_valid"
|
||||
except NotAllowedError:
|
||||
errors["base"] = "not_allowed"
|
||||
else:
|
||||
fullpath = str(get_path.absolute())
|
||||
await self.async_set_unique_id(fullpath)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
name = str(user_input[CONF_FILE_PATH]).rsplit("/", maxsplit=1)[-1]
|
||||
return self.async_create_entry(
|
||||
title=name,
|
||||
data={CONF_FILE_PATH: user_input[CONF_FILE_PATH]},
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=DATA_SCHEMA, errors=errors
|
||||
)
|
||||
|
||||
async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
|
||||
"""Handle import from configuration.yaml."""
|
||||
return await self.async_step_user(user_input)
|
||||
|
||||
|
||||
class NotValidError(Exception):
|
||||
"""Path is not valid error."""
|
||||
|
||||
|
||||
class NotAllowedError(Exception):
|
||||
"""Path is not allowed error."""
|
8
homeassistant/components/filesize/const.py
Normal file
8
homeassistant/components/filesize/const.py
Normal file
|
@ -0,0 +1,8 @@
|
|||
"""The filesize constants."""
|
||||
|
||||
from homeassistant.const import Platform
|
||||
|
||||
DOMAIN = "filesize"
|
||||
PLATFORMS = [Platform.SENSOR]
|
||||
|
||||
CONF_FILE_PATHS = "file_paths"
|
|
@ -2,6 +2,7 @@
|
|||
"domain": "filesize",
|
||||
"name": "File Size",
|
||||
"documentation": "https://www.home-assistant.io/integrations/filesize",
|
||||
"codeowners": [],
|
||||
"iot_class": "local_polling"
|
||||
"codeowners": ["@gjohansson-ST"],
|
||||
"iot_class": "local_polling",
|
||||
"config_flow": true
|
||||
}
|
||||
|
|
|
@ -12,19 +12,17 @@ from homeassistant.components.sensor import (
|
|||
PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
|
||||
SensorEntity,
|
||||
)
|
||||
from homeassistant.const import DATA_MEGABYTES
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import CONF_FILE_PATH, DATA_MEGABYTES
|
||||
from homeassistant.core import HomeAssistant
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.reload import setup_reload_service
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from . import DOMAIN, PLATFORMS
|
||||
from .const import CONF_FILE_PATHS, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
CONF_FILE_PATHS = "file_paths"
|
||||
ICON = "mdi:file"
|
||||
|
||||
PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend(
|
||||
|
@ -32,49 +30,53 @@ PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend(
|
|||
)
|
||||
|
||||
|
||||
def setup_platform(
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
add_entities: AddEntitiesCallback,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the file size sensor."""
|
||||
|
||||
setup_reload_service(hass, DOMAIN, PLATFORMS)
|
||||
|
||||
sensors = []
|
||||
paths = set()
|
||||
_LOGGER.warning(
|
||||
# Filesize config flow added in 2022.4 and should be removed in 2022.8
|
||||
"Loading filesize via platform setup is deprecated; Please remove it from your configuration"
|
||||
)
|
||||
for path in config[CONF_FILE_PATHS]:
|
||||
try:
|
||||
fullpath = str(pathlib.Path(path).absolute())
|
||||
except OSError as error:
|
||||
_LOGGER.error("Can not access file %s, error %s", path, error)
|
||||
continue
|
||||
|
||||
if fullpath in paths:
|
||||
continue
|
||||
paths.add(fullpath)
|
||||
|
||||
if not hass.config.is_allowed_path(path):
|
||||
_LOGGER.error("Filepath %s is not valid or allowed", path)
|
||||
continue
|
||||
|
||||
sensors.append(Filesize(fullpath))
|
||||
|
||||
if sensors:
|
||||
add_entities(sensors, True)
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={CONF_FILE_PATH: path},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class Filesize(SensorEntity):
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the platform from config entry."""
|
||||
|
||||
path = entry.data[CONF_FILE_PATH]
|
||||
get_path = await hass.async_add_executor_job(pathlib.Path, path)
|
||||
fullpath = str(get_path.absolute())
|
||||
|
||||
if get_path.exists() and get_path.is_file():
|
||||
async_add_entities([FilesizeEntity(fullpath, entry.entry_id)], True)
|
||||
|
||||
|
||||
class FilesizeEntity(SensorEntity):
|
||||
"""Encapsulates file size information."""
|
||||
|
||||
_attr_native_unit_of_measurement = DATA_MEGABYTES
|
||||
_attr_icon = ICON
|
||||
|
||||
def __init__(self, path: str) -> None:
|
||||
def __init__(self, path: str, entry_id: str) -> None:
|
||||
"""Initialize the data object."""
|
||||
self._path = path # Need to check its a valid path
|
||||
self._attr_name = path.split("/")[-1]
|
||||
self._attr_unique_id = entry_id
|
||||
|
||||
def update(self) -> None:
|
||||
"""Update the sensor."""
|
||||
|
|
|
@ -1,3 +0,0 @@
|
|||
reload:
|
||||
name: Reload
|
||||
description: Reload all filesize entities.
|
18
homeassistant/components/filesize/strings.json
Normal file
18
homeassistant/components/filesize/strings.json
Normal file
|
@ -0,0 +1,18 @@
|
|||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"file_path": "Path to file"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"not_valid": "Path is not valid",
|
||||
"not_allowed": "Path is not allowed"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
|
||||
}
|
||||
}
|
||||
}
|
18
homeassistant/components/filesize/translations/en.json
Normal file
18
homeassistant/components/filesize/translations/en.json
Normal file
|
@ -0,0 +1,18 @@
|
|||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"file_path": "Path to file"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"not_valid": "Path is not valid",
|
||||
"not_allowed": "Path is not allowed"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "Filepath is already configured"
|
||||
}
|
||||
}
|
||||
}
|
|
@ -97,6 +97,7 @@ FLOWS = {
|
|||
"evil_genius_labs",
|
||||
"ezviz",
|
||||
"faa_delays",
|
||||
"filesize",
|
||||
"fireservicerota",
|
||||
"fivem",
|
||||
"fjaraskupan",
|
||||
|
|
11
mypy.ini
11
mypy.ini
|
@ -704,6 +704,17 @@ no_implicit_optional = true
|
|||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.filesize.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.fitbit.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
|
|
@ -1 +1,14 @@
|
|||
"""Tests for the filesize component."""
|
||||
import os
|
||||
|
||||
TEST_DIR = os.path.join(os.path.dirname(__file__))
|
||||
TEST_FILE_NAME = "mock_file_test_filesize.txt"
|
||||
TEST_FILE_NAME2 = "mock_file_test_filesize2.txt"
|
||||
TEST_FILE = os.path.join(TEST_DIR, TEST_FILE_NAME)
|
||||
TEST_FILE2 = os.path.join(TEST_DIR, TEST_FILE_NAME2)
|
||||
|
||||
|
||||
def create_file(path) -> None:
|
||||
"""Create a test file."""
|
||||
with open(path, "w", encoding="utf-8") as test_file:
|
||||
test_file.write("test")
|
||||
|
|
45
tests/components/filesize/conftest.py
Normal file
45
tests/components/filesize/conftest.py
Normal file
|
@ -0,0 +1,45 @@
|
|||
"""Fixtures for Filesize integration tests."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Generator
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.filesize.const import DOMAIN
|
||||
from homeassistant.const import CONF_FILE_PATH
|
||||
|
||||
from . import TEST_FILE, TEST_FILE2, TEST_FILE_NAME
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry() -> MockConfigEntry:
|
||||
"""Return the default mocked config entry."""
|
||||
return MockConfigEntry(
|
||||
title=TEST_FILE_NAME,
|
||||
domain=DOMAIN,
|
||||
data={CONF_FILE_PATH: TEST_FILE},
|
||||
unique_id=TEST_FILE,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[None, None, None]:
|
||||
"""Mock setting up a config entry."""
|
||||
with patch(
|
||||
"homeassistant.components.filesize.async_setup_entry", return_value=True
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def remove_file() -> None:
|
||||
"""Remove test file."""
|
||||
yield
|
||||
if os.path.isfile(TEST_FILE):
|
||||
os.remove(TEST_FILE)
|
||||
if os.path.isfile(TEST_FILE2):
|
||||
os.remove(TEST_FILE2)
|
|
@ -1,4 +0,0 @@
|
|||
sensor:
|
||||
- platform: filesize
|
||||
file_paths:
|
||||
- "/dev/null"
|
130
tests/components/filesize/test_config_flow.py
Normal file
130
tests/components/filesize/test_config_flow.py
Normal file
|
@ -0,0 +1,130 @@
|
|||
"""Tests for the Filesize config flow."""
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.filesize.const import DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
|
||||
from homeassistant.const import CONF_FILE_PATH
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import (
|
||||
RESULT_TYPE_ABORT,
|
||||
RESULT_TYPE_CREATE_ENTRY,
|
||||
RESULT_TYPE_FORM,
|
||||
)
|
||||
|
||||
from . import TEST_DIR, TEST_FILE, TEST_FILE_NAME, create_file
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_full_user_flow(hass: HomeAssistant) -> None:
|
||||
"""Test the full user configuration flow."""
|
||||
create_file(TEST_FILE)
|
||||
hass.config.allowlist_external_dirs = {TEST_DIR}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
assert result.get("type") == RESULT_TYPE_FORM
|
||||
assert result.get("step_id") == SOURCE_USER
|
||||
assert "flow_id" in result
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_FILE_PATH: TEST_FILE},
|
||||
)
|
||||
|
||||
assert result2.get("type") == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result2.get("title") == TEST_FILE_NAME
|
||||
assert result2.get("data") == {CONF_FILE_PATH: TEST_FILE}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("source", [SOURCE_USER, SOURCE_IMPORT])
|
||||
async def test_unique_path(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
source: str,
|
||||
) -> None:
|
||||
"""Test we abort if already setup."""
|
||||
hass.config.allowlist_external_dirs = {TEST_DIR}
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": source}, data={CONF_FILE_PATH: TEST_FILE}
|
||||
)
|
||||
|
||||
assert result.get("type") == RESULT_TYPE_ABORT
|
||||
assert result.get("reason") == "already_configured"
|
||||
|
||||
|
||||
async def test_import_flow(hass: HomeAssistant) -> None:
|
||||
"""Test the import configuration flow."""
|
||||
create_file(TEST_FILE)
|
||||
hass.config.allowlist_external_dirs = {TEST_DIR}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={CONF_FILE_PATH: TEST_FILE},
|
||||
)
|
||||
|
||||
assert result.get("type") == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result.get("title") == TEST_FILE_NAME
|
||||
assert result.get("data") == {CONF_FILE_PATH: TEST_FILE}
|
||||
|
||||
|
||||
async def test_flow_fails_on_validation(hass: HomeAssistant) -> None:
|
||||
"""Test config flow errors."""
|
||||
create_file(TEST_FILE)
|
||||
hass.config.allowlist_external_dirs = {}
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == SOURCE_USER
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.filesize.config_flow.pathlib.Path",
|
||||
side_effect=OSError,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_FILE_PATH: TEST_FILE,
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["errors"] == {"base": "not_valid"}
|
||||
|
||||
with patch("homeassistant.components.filesize.config_flow.pathlib.Path",), patch(
|
||||
"homeassistant.components.filesize.async_setup_entry",
|
||||
return_value=True,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_FILE_PATH: TEST_FILE,
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["errors"] == {"base": "not_allowed"}
|
||||
|
||||
hass.config.allowlist_external_dirs = {TEST_DIR}
|
||||
with patch("homeassistant.components.filesize.config_flow.pathlib.Path",), patch(
|
||||
"homeassistant.components.filesize.async_setup_entry",
|
||||
return_value=True,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_FILE_PATH: TEST_FILE,
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result2["title"] == TEST_FILE_NAME
|
||||
assert result2["data"] == {
|
||||
CONF_FILE_PATH: TEST_FILE,
|
||||
}
|
110
tests/components/filesize/test_init.py
Normal file
110
tests/components/filesize/test_init.py
Normal file
|
@ -0,0 +1,110 @@
|
|||
"""Tests for the Filesize integration."""
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from homeassistant.components.filesize.const import CONF_FILE_PATHS, DOMAIN
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import CONF_FILE_PATH
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import (
|
||||
TEST_DIR,
|
||||
TEST_FILE,
|
||||
TEST_FILE2,
|
||||
TEST_FILE_NAME,
|
||||
TEST_FILE_NAME2,
|
||||
create_file,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_load_unload_config_entry(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry, tmpdir: str
|
||||
) -> None:
|
||||
"""Test the Filesize configuration entry loading/unloading."""
|
||||
testfile = f"{tmpdir}/file.txt"
|
||||
create_file(testfile)
|
||||
hass.config.allowlist_external_dirs = {tmpdir}
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
hass.config_entries.async_update_entry(
|
||||
mock_config_entry, unique_id=testfile, data={CONF_FILE_PATH: testfile}
|
||||
)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert not hass.data.get(DOMAIN)
|
||||
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_cannot_access_file(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry, tmpdir: str
|
||||
) -> None:
|
||||
"""Test that an file not exist is caught."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
testfile = f"{tmpdir}/file_not_exist.txt"
|
||||
hass.config.allowlist_external_dirs = {tmpdir}
|
||||
hass.config_entries.async_update_entry(
|
||||
mock_config_entry, unique_id=testfile, data={CONF_FILE_PATH: testfile}
|
||||
)
|
||||
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_not_valid_path_to_file(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry, tmpdir: str
|
||||
) -> None:
|
||||
"""Test that an invalid path is caught."""
|
||||
testfile = f"{tmpdir}/file.txt"
|
||||
create_file(testfile)
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
hass.config_entries.async_update_entry(
|
||||
mock_config_entry, unique_id=testfile, data={CONF_FILE_PATH: testfile}
|
||||
)
|
||||
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_import_config(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: AsyncMock,
|
||||
) -> None:
|
||||
"""Test Filesize being set up from config via import."""
|
||||
create_file(TEST_FILE)
|
||||
create_file(TEST_FILE2)
|
||||
hass.config.allowlist_external_dirs = {TEST_DIR}
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
SENSOR_DOMAIN,
|
||||
{
|
||||
SENSOR_DOMAIN: {
|
||||
"platform": DOMAIN,
|
||||
CONF_FILE_PATHS: [TEST_FILE, TEST_FILE2],
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
config_entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert len(config_entries) == 2
|
||||
|
||||
entry = config_entries[0]
|
||||
assert entry.title == TEST_FILE_NAME
|
||||
assert entry.unique_id == TEST_FILE
|
||||
assert entry.data == {CONF_FILE_PATH: TEST_FILE}
|
||||
entry2 = config_entries[1]
|
||||
assert entry2.title == TEST_FILE_NAME2
|
||||
assert entry2.unique_id == TEST_FILE2
|
||||
assert entry2.data == {CONF_FILE_PATH: TEST_FILE2}
|
|
@ -1,130 +1,72 @@
|
|||
"""The tests for the filesize sensor."""
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant import config as hass_config
|
||||
from homeassistant.components.filesize import DOMAIN
|
||||
from homeassistant.components.filesize.sensor import CONF_FILE_PATHS
|
||||
from homeassistant.const import SERVICE_RELOAD, STATE_UNKNOWN
|
||||
from homeassistant.const import CONF_FILE_PATH, STATE_UNKNOWN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_component import async_update_entity
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import get_fixture_path
|
||||
from . import TEST_FILE, TEST_FILE_NAME, create_file
|
||||
|
||||
TEST_DIR = os.path.join(os.path.dirname(__file__))
|
||||
TEST_FILE = os.path.join(TEST_DIR, "mock_file_test_filesize.txt")
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
def create_file(path) -> None:
|
||||
"""Create a test file."""
|
||||
with open(path, "w") as test_file:
|
||||
test_file.write("test")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def remove_file() -> None:
|
||||
"""Remove test file."""
|
||||
yield
|
||||
if os.path.isfile(TEST_FILE):
|
||||
os.remove(TEST_FILE)
|
||||
|
||||
|
||||
async def test_invalid_path(hass: HomeAssistant) -> None:
|
||||
async def test_invalid_path(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test that an invalid path is caught."""
|
||||
config = {"sensor": {"platform": "filesize", CONF_FILE_PATHS: ["invalid_path"]}}
|
||||
assert await async_setup_component(hass, "sensor", config)
|
||||
await hass.async_block_till_done()
|
||||
assert len(hass.states.async_entity_ids("sensor")) == 0
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
hass.config_entries.async_update_entry(
|
||||
mock_config_entry, unique_id=TEST_FILE, data={CONF_FILE_PATH: TEST_FILE}
|
||||
)
|
||||
|
||||
state = hass.states.get("sensor." + TEST_FILE_NAME)
|
||||
assert not state
|
||||
|
||||
|
||||
async def test_cannot_access_file(hass: HomeAssistant) -> None:
|
||||
"""Test that an invalid path is caught."""
|
||||
config = {"sensor": {"platform": "filesize", CONF_FILE_PATHS: [TEST_FILE]}}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.filesize.sensor.pathlib",
|
||||
side_effect=OSError("Can not access"),
|
||||
):
|
||||
assert await async_setup_component(hass, "sensor", config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_entity_ids("sensor")) == 0
|
||||
|
||||
|
||||
async def test_valid_path(hass: HomeAssistant) -> None:
|
||||
async def test_valid_path(
|
||||
hass: HomeAssistant, tmpdir: str, mock_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test for a valid path."""
|
||||
create_file(TEST_FILE)
|
||||
config = {"sensor": {"platform": "filesize", CONF_FILE_PATHS: [TEST_FILE]}}
|
||||
hass.config.allowlist_external_dirs = {TEST_DIR}
|
||||
assert await async_setup_component(hass, "sensor", config)
|
||||
testfile = f"{tmpdir}/file.txt"
|
||||
create_file(testfile)
|
||||
hass.config.allowlist_external_dirs = {tmpdir}
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
hass.config_entries.async_update_entry(
|
||||
mock_config_entry, unique_id=testfile, data={CONF_FILE_PATH: testfile}
|
||||
)
|
||||
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert len(hass.states.async_entity_ids("sensor")) == 1
|
||||
state = hass.states.get("sensor.mock_file_test_filesize_txt")
|
||||
|
||||
state = hass.states.get("sensor.file_txt")
|
||||
assert state
|
||||
assert state.state == "0.0"
|
||||
assert state.attributes.get("bytes") == 4
|
||||
|
||||
await hass.async_add_executor_job(os.remove, testfile)
|
||||
|
||||
async def test_state_unknown(hass: HomeAssistant, tmpdir: str) -> None:
|
||||
|
||||
async def test_state_unknown(
|
||||
hass: HomeAssistant, tmpdir: str, mock_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Verify we handle state unavailable."""
|
||||
create_file(TEST_FILE)
|
||||
testfile = f"{tmpdir}/file"
|
||||
await hass.async_add_executor_job(create_file, testfile)
|
||||
with patch.object(hass.config, "is_allowed_path", return_value=True):
|
||||
await async_setup_component(
|
||||
hass,
|
||||
"sensor",
|
||||
{
|
||||
"sensor": {
|
||||
"platform": "filesize",
|
||||
"file_paths": [testfile],
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
create_file(testfile)
|
||||
hass.config.allowlist_external_dirs = {tmpdir}
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
hass.config_entries.async_update_entry(
|
||||
mock_config_entry, unique_id=testfile, data={CONF_FILE_PATH: testfile}
|
||||
)
|
||||
|
||||
assert hass.states.get("sensor.file")
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.file")
|
||||
assert state
|
||||
assert state.state == "0.0"
|
||||
|
||||
await hass.async_add_executor_job(os.remove, testfile)
|
||||
await async_update_entity(hass, "sensor.file")
|
||||
|
||||
state = hass.states.get("sensor.file")
|
||||
assert state.state == STATE_UNKNOWN
|
||||
|
||||
|
||||
async def test_reload(hass: HomeAssistant, tmpdir: str) -> None:
|
||||
"""Verify we can reload filesize sensors."""
|
||||
testfile = f"{tmpdir}/file"
|
||||
await hass.async_add_executor_job(create_file, testfile)
|
||||
with patch.object(hass.config, "is_allowed_path", return_value=True):
|
||||
await async_setup_component(
|
||||
hass,
|
||||
"sensor",
|
||||
{
|
||||
"sensor": {
|
||||
"platform": "filesize",
|
||||
"file_paths": [testfile],
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_all()) == 1
|
||||
|
||||
assert hass.states.get("sensor.file")
|
||||
|
||||
yaml_path = get_fixture_path("configuration.yaml", "filesize")
|
||||
with patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path), patch.object(
|
||||
hass.config, "is_allowed_path", return_value=True
|
||||
):
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_RELOAD,
|
||||
{},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get("sensor.file") is None
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue