mirror of
https://github.com/home-assistant/core.git
synced 2025-07-20 11:47:06 +00:00
Code quality Filesize (#71768)
This commit is contained in:
parent
991f0b40f2
commit
c0ae31d86c
@ -11,13 +11,19 @@ from homeassistant.exceptions import ConfigEntryNotReady
|
|||||||
from .const import PLATFORMS
|
from .const import PLATFORMS
|
||||||
|
|
||||||
|
|
||||||
|
def check_path(path: pathlib.Path) -> bool:
|
||||||
|
"""Check path."""
|
||||||
|
return path.exists() and path.is_file()
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
"""Set up from a config entry."""
|
"""Set up from a config entry."""
|
||||||
|
|
||||||
path = entry.data[CONF_FILE_PATH]
|
path = entry.data[CONF_FILE_PATH]
|
||||||
get_path = await hass.async_add_executor_job(pathlib.Path, path)
|
get_path = pathlib.Path(path)
|
||||||
|
|
||||||
if not get_path.exists() and not get_path.is_file():
|
check_file = await hass.async_add_executor_job(check_path, get_path)
|
||||||
|
if not check_file:
|
||||||
raise ConfigEntryNotReady(f"Can not access file {path}")
|
raise ConfigEntryNotReady(f"Can not access file {path}")
|
||||||
|
|
||||||
if not hass.config.is_allowed_path(path):
|
if not hass.config.is_allowed_path(path):
|
||||||
|
@ -11,6 +11,7 @@ from homeassistant.config_entries import ConfigFlow
|
|||||||
from homeassistant.const import CONF_FILE_PATH
|
from homeassistant.const import CONF_FILE_PATH
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.data_entry_flow import FlowResult
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
|
from homeassistant.exceptions import HomeAssistantError
|
||||||
|
|
||||||
from .const import DOMAIN
|
from .const import DOMAIN
|
||||||
|
|
||||||
@ -19,19 +20,20 @@ DATA_SCHEMA = vol.Schema({vol.Required(CONF_FILE_PATH): str})
|
|||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def validate_path(hass: HomeAssistant, path: str) -> pathlib.Path:
|
def validate_path(hass: HomeAssistant, path: str) -> str:
|
||||||
"""Validate path."""
|
"""Validate path."""
|
||||||
try:
|
|
||||||
get_path = pathlib.Path(path)
|
get_path = pathlib.Path(path)
|
||||||
except OSError as error:
|
if not get_path.exists() or not get_path.is_file():
|
||||||
_LOGGER.error("Can not access file %s, error %s", path, error)
|
_LOGGER.error("Can not access file %s", path)
|
||||||
raise NotValidError from error
|
raise NotValidError
|
||||||
|
|
||||||
if not hass.config.is_allowed_path(path):
|
if not hass.config.is_allowed_path(path):
|
||||||
_LOGGER.error("Filepath %s is not valid or allowed", path)
|
_LOGGER.error("Filepath %s is not allowed", path)
|
||||||
raise NotAllowedError
|
raise NotAllowedError
|
||||||
|
|
||||||
return get_path
|
full_path = get_path.absolute()
|
||||||
|
|
||||||
|
return str(full_path)
|
||||||
|
|
||||||
|
|
||||||
class FilesizeConfigFlow(ConfigFlow, domain=DOMAIN):
|
class FilesizeConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||||
@ -47,14 +49,13 @@ class FilesizeConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||||||
|
|
||||||
if user_input is not None:
|
if user_input is not None:
|
||||||
try:
|
try:
|
||||||
get_path = validate_path(self.hass, user_input[CONF_FILE_PATH])
|
full_path = validate_path(self.hass, user_input[CONF_FILE_PATH])
|
||||||
except NotValidError:
|
except NotValidError:
|
||||||
errors["base"] = "not_valid"
|
errors["base"] = "not_valid"
|
||||||
except NotAllowedError:
|
except NotAllowedError:
|
||||||
errors["base"] = "not_allowed"
|
errors["base"] = "not_allowed"
|
||||||
else:
|
else:
|
||||||
fullpath = str(get_path.absolute())
|
await self.async_set_unique_id(full_path)
|
||||||
await self.async_set_unique_id(fullpath)
|
|
||||||
self._abort_if_unique_id_configured()
|
self._abort_if_unique_id_configured()
|
||||||
|
|
||||||
name = str(user_input[CONF_FILE_PATH]).rsplit("/", maxsplit=1)[-1]
|
name = str(user_input[CONF_FILE_PATH]).rsplit("/", maxsplit=1)[-1]
|
||||||
@ -68,9 +69,9 @@ class FilesizeConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class NotValidError(Exception):
|
class NotValidError(HomeAssistantError):
|
||||||
"""Path is not valid error."""
|
"""Path is not valid error."""
|
||||||
|
|
||||||
|
|
||||||
class NotAllowedError(Exception):
|
class NotAllowedError(HomeAssistantError):
|
||||||
"""Path is not allowed error."""
|
"""Path is not allowed error."""
|
||||||
|
@ -74,12 +74,9 @@ async def async_setup_entry(
|
|||||||
coordinator = FileSizeCoordinator(hass, fullpath)
|
coordinator = FileSizeCoordinator(hass, fullpath)
|
||||||
await coordinator.async_config_entry_first_refresh()
|
await coordinator.async_config_entry_first_refresh()
|
||||||
|
|
||||||
if get_path.exists() and get_path.is_file():
|
|
||||||
async_add_entities(
|
async_add_entities(
|
||||||
[
|
|
||||||
FilesizeEntity(description, fullpath, entry.entry_id, coordinator)
|
FilesizeEntity(description, fullpath, entry.entry_id, coordinator)
|
||||||
for description in SENSOR_TYPES
|
for description in SENSOR_TYPES
|
||||||
]
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@ -119,7 +116,7 @@ class FileSizeCoordinator(DataUpdateCoordinator):
|
|||||||
|
|
||||||
|
|
||||||
class FilesizeEntity(CoordinatorEntity[FileSizeCoordinator], SensorEntity):
|
class FilesizeEntity(CoordinatorEntity[FileSizeCoordinator], SensorEntity):
|
||||||
"""Encapsulates file size information."""
|
"""Filesize sensor."""
|
||||||
|
|
||||||
entity_description: SensorEntityDescription
|
entity_description: SensorEntityDescription
|
||||||
|
|
||||||
@ -130,7 +127,7 @@ class FilesizeEntity(CoordinatorEntity[FileSizeCoordinator], SensorEntity):
|
|||||||
entry_id: str,
|
entry_id: str,
|
||||||
coordinator: FileSizeCoordinator,
|
coordinator: FileSizeCoordinator,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Initialize the data object."""
|
"""Initialize the Filesize sensor."""
|
||||||
super().__init__(coordinator)
|
super().__init__(coordinator)
|
||||||
base_name = path.split("/")[-1]
|
base_name = path.split("/")[-1]
|
||||||
self._attr_name = f"{base_name} {description.name}"
|
self._attr_name = f"{base_name} {description.name}"
|
||||||
|
@ -1,6 +1,8 @@
|
|||||||
"""Tests for the filesize component."""
|
"""Tests for the filesize component."""
|
||||||
import os
|
import os
|
||||||
|
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
TEST_DIR = os.path.join(os.path.dirname(__file__))
|
TEST_DIR = os.path.join(os.path.dirname(__file__))
|
||||||
TEST_FILE_NAME = "mock_file_test_filesize.txt"
|
TEST_FILE_NAME = "mock_file_test_filesize.txt"
|
||||||
TEST_FILE_NAME2 = "mock_file_test_filesize2.txt"
|
TEST_FILE_NAME2 = "mock_file_test_filesize2.txt"
|
||||||
@ -8,7 +10,12 @@ TEST_FILE = os.path.join(TEST_DIR, TEST_FILE_NAME)
|
|||||||
TEST_FILE2 = os.path.join(TEST_DIR, TEST_FILE_NAME2)
|
TEST_FILE2 = os.path.join(TEST_DIR, TEST_FILE_NAME2)
|
||||||
|
|
||||||
|
|
||||||
def create_file(path) -> None:
|
async def async_create_file(hass: HomeAssistant, path: str) -> None:
|
||||||
"""Create a test file."""
|
"""Create a test file."""
|
||||||
|
await hass.async_add_executor_job(create_file, path)
|
||||||
|
|
||||||
|
|
||||||
|
def create_file(path: str) -> None:
|
||||||
|
"""Create the test file."""
|
||||||
with open(path, "w", encoding="utf-8") as test_file:
|
with open(path, "w", encoding="utf-8") as test_file:
|
||||||
test_file.write("test")
|
test_file.write("test")
|
||||||
|
@ -11,14 +11,14 @@ from homeassistant.data_entry_flow import (
|
|||||||
RESULT_TYPE_FORM,
|
RESULT_TYPE_FORM,
|
||||||
)
|
)
|
||||||
|
|
||||||
from . import TEST_DIR, TEST_FILE, TEST_FILE_NAME, create_file
|
from . import TEST_DIR, TEST_FILE, TEST_FILE_NAME, async_create_file
|
||||||
|
|
||||||
from tests.common import MockConfigEntry
|
from tests.common import MockConfigEntry
|
||||||
|
|
||||||
|
|
||||||
async def test_full_user_flow(hass: HomeAssistant) -> None:
|
async def test_full_user_flow(hass: HomeAssistant) -> None:
|
||||||
"""Test the full user configuration flow."""
|
"""Test the full user configuration flow."""
|
||||||
create_file(TEST_FILE)
|
await async_create_file(hass, TEST_FILE)
|
||||||
hass.config.allowlist_external_dirs = {TEST_DIR}
|
hass.config.allowlist_external_dirs = {TEST_DIR}
|
||||||
result = await hass.config_entries.flow.async_init(
|
result = await hass.config_entries.flow.async_init(
|
||||||
DOMAIN, context={"source": SOURCE_USER}
|
DOMAIN, context={"source": SOURCE_USER}
|
||||||
@ -43,6 +43,7 @@ async def test_unique_path(
|
|||||||
mock_config_entry: MockConfigEntry,
|
mock_config_entry: MockConfigEntry,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test we abort if already setup."""
|
"""Test we abort if already setup."""
|
||||||
|
await async_create_file(hass, TEST_FILE)
|
||||||
hass.config.allowlist_external_dirs = {TEST_DIR}
|
hass.config.allowlist_external_dirs = {TEST_DIR}
|
||||||
mock_config_entry.add_to_hass(hass)
|
mock_config_entry.add_to_hass(hass)
|
||||||
|
|
||||||
@ -56,7 +57,7 @@ async def test_unique_path(
|
|||||||
|
|
||||||
async def test_flow_fails_on_validation(hass: HomeAssistant) -> None:
|
async def test_flow_fails_on_validation(hass: HomeAssistant) -> None:
|
||||||
"""Test config flow errors."""
|
"""Test config flow errors."""
|
||||||
create_file(TEST_FILE)
|
|
||||||
hass.config.allowlist_external_dirs = {}
|
hass.config.allowlist_external_dirs = {}
|
||||||
|
|
||||||
result = await hass.config_entries.flow.async_init(
|
result = await hass.config_entries.flow.async_init(
|
||||||
@ -66,10 +67,6 @@ async def test_flow_fails_on_validation(hass: HomeAssistant) -> None:
|
|||||||
assert result["type"] == RESULT_TYPE_FORM
|
assert result["type"] == RESULT_TYPE_FORM
|
||||||
assert result["step_id"] == SOURCE_USER
|
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(
|
result2 = await hass.config_entries.flow.async_configure(
|
||||||
result["flow_id"],
|
result["flow_id"],
|
||||||
user_input={
|
user_input={
|
||||||
@ -79,6 +76,8 @@ async def test_flow_fails_on_validation(hass: HomeAssistant) -> None:
|
|||||||
|
|
||||||
assert result2["errors"] == {"base": "not_valid"}
|
assert result2["errors"] == {"base": "not_valid"}
|
||||||
|
|
||||||
|
await async_create_file(hass, TEST_FILE)
|
||||||
|
|
||||||
with patch("homeassistant.components.filesize.config_flow.pathlib.Path",), patch(
|
with patch("homeassistant.components.filesize.config_flow.pathlib.Path",), patch(
|
||||||
"homeassistant.components.filesize.async_setup_entry",
|
"homeassistant.components.filesize.async_setup_entry",
|
||||||
return_value=True,
|
return_value=True,
|
||||||
|
@ -4,7 +4,7 @@ from homeassistant.config_entries import ConfigEntryState
|
|||||||
from homeassistant.const import CONF_FILE_PATH
|
from homeassistant.const import CONF_FILE_PATH
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
from . import create_file
|
from . import async_create_file
|
||||||
|
|
||||||
from tests.common import MockConfigEntry
|
from tests.common import MockConfigEntry
|
||||||
|
|
||||||
@ -14,7 +14,7 @@ async def test_load_unload_config_entry(
|
|||||||
) -> None:
|
) -> None:
|
||||||
"""Test the Filesize configuration entry loading/unloading."""
|
"""Test the Filesize configuration entry loading/unloading."""
|
||||||
testfile = f"{tmpdir}/file.txt"
|
testfile = f"{tmpdir}/file.txt"
|
||||||
create_file(testfile)
|
await async_create_file(hass, testfile)
|
||||||
hass.config.allowlist_external_dirs = {tmpdir}
|
hass.config.allowlist_external_dirs = {tmpdir}
|
||||||
mock_config_entry.add_to_hass(hass)
|
mock_config_entry.add_to_hass(hass)
|
||||||
hass.config_entries.async_update_entry(
|
hass.config_entries.async_update_entry(
|
||||||
@ -54,7 +54,7 @@ async def test_not_valid_path_to_file(
|
|||||||
) -> None:
|
) -> None:
|
||||||
"""Test that an invalid path is caught."""
|
"""Test that an invalid path is caught."""
|
||||||
testfile = f"{tmpdir}/file.txt"
|
testfile = f"{tmpdir}/file.txt"
|
||||||
create_file(testfile)
|
await async_create_file(hass, testfile)
|
||||||
mock_config_entry.add_to_hass(hass)
|
mock_config_entry.add_to_hass(hass)
|
||||||
hass.config_entries.async_update_entry(
|
hass.config_entries.async_update_entry(
|
||||||
mock_config_entry, unique_id=testfile, data={CONF_FILE_PATH: testfile}
|
mock_config_entry, unique_id=testfile, data={CONF_FILE_PATH: testfile}
|
||||||
|
@ -5,7 +5,7 @@ from homeassistant.const import CONF_FILE_PATH, STATE_UNAVAILABLE
|
|||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.entity_component import async_update_entity
|
from homeassistant.helpers.entity_component import async_update_entity
|
||||||
|
|
||||||
from . import TEST_FILE, TEST_FILE_NAME, create_file
|
from . import TEST_FILE, TEST_FILE_NAME, async_create_file
|
||||||
|
|
||||||
from tests.common import MockConfigEntry
|
from tests.common import MockConfigEntry
|
||||||
|
|
||||||
@ -28,7 +28,7 @@ async def test_valid_path(
|
|||||||
) -> None:
|
) -> None:
|
||||||
"""Test for a valid path."""
|
"""Test for a valid path."""
|
||||||
testfile = f"{tmpdir}/file.txt"
|
testfile = f"{tmpdir}/file.txt"
|
||||||
create_file(testfile)
|
await async_create_file(hass, testfile)
|
||||||
hass.config.allowlist_external_dirs = {tmpdir}
|
hass.config.allowlist_external_dirs = {tmpdir}
|
||||||
mock_config_entry.add_to_hass(hass)
|
mock_config_entry.add_to_hass(hass)
|
||||||
hass.config_entries.async_update_entry(
|
hass.config_entries.async_update_entry(
|
||||||
@ -50,7 +50,7 @@ async def test_state_unavailable(
|
|||||||
) -> None:
|
) -> None:
|
||||||
"""Verify we handle state unavailable."""
|
"""Verify we handle state unavailable."""
|
||||||
testfile = f"{tmpdir}/file.txt"
|
testfile = f"{tmpdir}/file.txt"
|
||||||
create_file(testfile)
|
await async_create_file(hass, testfile)
|
||||||
hass.config.allowlist_external_dirs = {tmpdir}
|
hass.config.allowlist_external_dirs = {tmpdir}
|
||||||
mock_config_entry.add_to_hass(hass)
|
mock_config_entry.add_to_hass(hass)
|
||||||
hass.config_entries.async_update_entry(
|
hass.config_entries.async_update_entry(
|
||||||
|
Loading…
x
Reference in New Issue
Block a user