Add config flow to local_file (#125835)

* Add config flow to local_file

* Small mods

* Add/fix tests

* Fix

* slug

* Fix strings

* Mod strings
pull/128925/head
G Johansson 2024-10-21 19:04:43 +02:00 committed by GitHub
parent 1cc776d332
commit 1eaaa5c6d3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 788 additions and 182 deletions

View File

@ -1 +1,37 @@
"""The local_file component."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_FILE_PATH, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError
from .const import DOMAIN
from .util import check_file_path_access
PLATFORMS = [Platform.CAMERA]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Local file from a config entry."""
file_path: str = entry.options[CONF_FILE_PATH]
if not await hass.async_add_executor_job(check_file_path_access, file_path):
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="not_readable_path",
translation_placeholders={"file_path": file_path},
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload Local file config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)

View File

@ -4,7 +4,6 @@ from __future__ import annotations
import logging
import mimetypes
import os
import voluptuous as vol
@ -12,14 +11,21 @@ from homeassistant.components.camera import (
PLATFORM_SCHEMA as CAMERA_PLATFORM_SCHEMA,
Camera,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_FILE_PATH, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady, ServiceValidationError
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import (
config_validation as cv,
entity_platform,
issue_registry as ir,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import slugify
from .const import DEFAULT_NAME, SERVICE_UPDATE_FILE_PATH
from .const import DEFAULT_NAME, DOMAIN, SERVICE_UPDATE_FILE_PATH
from .util import check_file_path_access
_LOGGER = logging.getLogger(__name__)
@ -31,21 +37,12 @@ PLATFORM_SCHEMA = CAMERA_PLATFORM_SCHEMA.extend(
)
def check_file_path_access(file_path: str) -> bool:
"""Check that filepath given is readable."""
if not os.access(file_path, os.R_OK):
return False
return True
async def async_setup_platform(
async def async_setup_entry(
hass: HomeAssistant,
config: ConfigType,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Camera that works with local files."""
file_path: str = config[CONF_FILE_PATH]
"""Set up the Camera for local file from a config entry."""
platform = entity_platform.async_get_current_platform()
platform.async_register_entity_service(
@ -56,19 +53,76 @@ async def async_setup_platform(
"update_file_path",
)
if not await hass.async_add_executor_job(check_file_path_access, file_path):
raise PlatformNotReady(f"File path {file_path} is not readable")
async_add_entities(
[
LocalFile(
entry.options[CONF_NAME],
entry.options[CONF_FILE_PATH],
entry.entry_id,
)
]
)
async_add_entities([LocalFile(config[CONF_NAME], file_path)])
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Camera that works with local files."""
file_path: str = config[CONF_FILE_PATH]
file_path_slug = slugify(file_path)
if not await hass.async_add_executor_job(check_file_path_access, file_path):
ir.async_create_issue(
hass,
DOMAIN,
f"no_access_path_{file_path_slug}",
breaks_in_ha_version="2025.5.0",
is_fixable=False,
learn_more_url="https://www.home-assistant.io/integrations/local_file/",
severity=ir.IssueSeverity.WARNING,
translation_key="no_access_path",
translation_placeholders={
"file_path": file_path_slug,
},
)
return
ir.async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2025.5.0",
is_fixable=False,
issue_domain=DOMAIN,
learn_more_url="https://www.home-assistant.io/integrations/local_file/",
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Local file",
},
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
class LocalFile(Camera):
"""Representation of a local file camera."""
def __init__(self, name: str, file_path: str) -> None:
def __init__(self, name: str, file_path: str, unique_id: str) -> None:
"""Initialize Local File Camera component."""
super().__init__()
self._attr_name = name
self._attr_unique_id = unique_id
self._file_path = file_path
# Set content type of local file
content, _ = mimetypes.guess_type(file_path)

View File

@ -0,0 +1,77 @@
"""Config flow for Local file."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any, cast
import voluptuous as vol
from homeassistant.const import CONF_FILE_PATH, CONF_NAME
from homeassistant.helpers.schema_config_entry_flow import (
SchemaCommonFlowHandler,
SchemaConfigFlowHandler,
SchemaFlowError,
SchemaFlowFormStep,
)
from homeassistant.helpers.selector import TextSelector
from .const import DEFAULT_NAME, DOMAIN
from .util import check_file_path_access
async def validate_options(
handler: SchemaCommonFlowHandler, user_input: dict[str, Any]
) -> dict[str, Any]:
"""Validate options selected."""
file_path: str = user_input[CONF_FILE_PATH]
if not await handler.parent_handler.hass.async_add_executor_job(
check_file_path_access, file_path
):
raise SchemaFlowError("not_readable_path")
handler.parent_handler._async_abort_entries_match( # noqa: SLF001
{CONF_FILE_PATH: user_input[CONF_FILE_PATH]}
)
return user_input
DATA_SCHEMA_OPTIONS = vol.Schema(
{
vol.Required(CONF_FILE_PATH): TextSelector(),
}
)
DATA_SCHEMA_SETUP = vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): TextSelector(),
}
).extend(DATA_SCHEMA_OPTIONS.schema)
CONFIG_FLOW = {
"user": SchemaFlowFormStep(
schema=DATA_SCHEMA_SETUP,
validate_user_input=validate_options,
),
"import": SchemaFlowFormStep(
schema=DATA_SCHEMA_SETUP,
validate_user_input=validate_options,
),
}
OPTIONS_FLOW = {
"init": SchemaFlowFormStep(
DATA_SCHEMA_OPTIONS,
validate_user_input=validate_options,
)
}
class LocalFileConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
"""Handle a config flow for Local file."""
config_flow = CONFIG_FLOW
options_flow = OPTIONS_FLOW
def async_config_entry_title(self, options: Mapping[str, Any]) -> str:
"""Return config entry title."""
return cast(str, options[CONF_NAME])

View File

@ -1,4 +1,42 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"error": {
"not_readable_path": "The provided path to the file can not be read"
},
"step": {
"user": {
"data": {
"name": "[%key:common::config_flow::data::name%]",
"file_path": "File path"
},
"data_description": {
"name": "Name for the created entity.",
"file_path": "The full path to the image file to be displayed. Be sure the path of the file is in the allowed paths, you can read more about this in the documentation."
}
}
}
},
"options": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"error": {
"not_readable_path": "[%key:component::local_file::config::error::not_readable_path%]"
},
"step": {
"init": {
"data": {
"file_path": "[%key:component::local_file::config::step::user::data::file_path%]"
},
"data_description": {
"file_path": "[%key:component::local_file::config::step::user::data_description::file_path%]"
}
}
}
},
"services": {
"update_file_path": {
"name": "Updates file path",
@ -6,7 +44,7 @@
"fields": {
"file_path": {
"name": "File path",
"description": "The full path to the new image file to be displayed."
"description": "[%key:component::local_file::config::step::user::data_description::file_path%]"
}
}
}
@ -15,5 +53,11 @@
"file_path_not_accessible": {
"message": "Path {file_path} is not accessible"
}
},
"issues": {
"no_access_path": {
"title": "Incorrect file path",
"description": "While trying to import your configuration the provided file path {file_path} could not be read.\nPlease update your configuration to a correct file path and restart to fix this issue."
}
}
}

View File

@ -0,0 +1,10 @@
"""Utils for local file."""
import os
def check_file_path_access(file_path: str) -> bool:
"""Check that filepath given is readable."""
if not os.access(file_path, os.R_OK):
return False
return True

View File

@ -0,0 +1,63 @@
"""Fixtures for the Local file integration."""
from __future__ import annotations
from collections.abc import Generator
from typing import Any
from unittest.mock import AsyncMock, Mock, patch
import pytest
from homeassistant.components.local_file.const import DEFAULT_NAME, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_FILE_PATH, CONF_NAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Automatically patch setup."""
with patch(
"homeassistant.components.local_file.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture(name="get_config")
async def get_config_to_integration_load() -> dict[str, Any]:
"""Return configuration.
To override the config, tests can be marked with:
@pytest.mark.parametrize("get_config", [{...}])
"""
return {CONF_NAME: DEFAULT_NAME, CONF_FILE_PATH: "mock.file"}
@pytest.fixture(name="loaded_entry")
async def load_integration(
hass: HomeAssistant, get_config: dict[str, Any]
) -> MockConfigEntry:
"""Set up the Local file integration in Home Assistant."""
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
options=get_config,
entry_id="1",
)
config_entry.add_to_hass(hass)
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry

View File

@ -1,222 +1,189 @@
"""The tests for local file camera component."""
from http import HTTPStatus
from unittest import mock
from typing import Any
from unittest.mock import Mock, mock_open, patch
import pytest
from homeassistant.components.local_file.const import DOMAIN, SERVICE_UPDATE_FILE_PATH
from homeassistant.components.local_file.const import (
DEFAULT_NAME,
DOMAIN,
SERVICE_UPDATE_FILE_PATH,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import ATTR_ENTITY_ID, CONF_FILE_PATH
from homeassistant.core import HomeAssistant
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
from homeassistant.util import slugify
from tests.common import MockConfigEntry
from tests.typing import ClientSessionGenerator
async def test_loading_file(
hass: HomeAssistant, hass_client: ClientSessionGenerator
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
loaded_entry: MockConfigEntry,
) -> None:
"""Test that it loads image from disk."""
with (
mock.patch("os.path.isfile", mock.Mock(return_value=True)),
mock.patch("os.access", mock.Mock(return_value=True)),
mock.patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
mock.Mock(return_value=(None, None)),
),
):
await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "local_file",
"file_path": "mock.file",
}
},
)
await hass.async_block_till_done()
client = await hass_client()
m_open = mock.mock_open(read_data=b"hello")
with mock.patch(
"homeassistant.components.local_file.camera.open", m_open, create=True
):
resp = await client.get("/api/camera_proxy/camera.config_test")
m_open = mock_open(read_data=b"hello")
with patch("homeassistant.components.local_file.camera.open", m_open, create=True):
resp = await client.get("/api/camera_proxy/camera.local_file")
assert resp.status == HTTPStatus.OK
body = await resp.text()
assert body == "hello"
async def test_file_not_readable(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test a warning is shown setup when file is not readable."""
with (
mock.patch("os.path.isfile", mock.Mock(return_value=True)),
mock.patch("os.access", mock.Mock(return_value=False)),
):
await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "local_file",
"file_path": "mock.file",
}
},
)
await hass.async_block_till_done()
assert "File path mock.file is not readable;" in caplog.text
async def test_file_not_readable_after_setup(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
loaded_entry: MockConfigEntry,
) -> None:
"""Test a warning is shown setup when file is not readable."""
with (
mock.patch("os.path.isfile", mock.Mock(return_value=True)),
mock.patch("os.access", mock.Mock(return_value=True)),
mock.patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
mock.Mock(return_value=(None, None)),
),
):
await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "local_file",
"file_path": "mock.file",
}
},
)
await hass.async_block_till_done()
client = await hass_client()
with mock.patch(
with patch(
"homeassistant.components.local_file.camera.open", side_effect=FileNotFoundError
):
resp = await client.get("/api/camera_proxy/camera.config_test")
resp = await client.get("/api/camera_proxy/camera.local_file")
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
assert "Could not read camera config_test image from file: mock.file" in caplog.text
assert "Could not read camera Local File image from file: mock.file" in caplog.text
@pytest.mark.parametrize(
("config", "url", "content_type"),
[
(
{
"name": "test_jpg",
"file_path": "/path/to/image.jpg",
},
"/api/camera_proxy/camera.test_jpg",
"image/jpeg",
),
(
{
"name": "test_png",
"file_path": "/path/to/image.png",
},
"/api/camera_proxy/camera.test_png",
"image/png",
),
(
{
"name": "test_svg",
"file_path": "/path/to/image.svg",
},
"/api/camera_proxy/camera.test_svg",
"image/svg+xml",
),
(
{
"name": "test_no_ext",
"file_path": "/path/to/image",
},
"/api/camera_proxy/camera.test_no_ext",
"image/jpeg",
),
],
)
async def test_camera_content_type(
hass: HomeAssistant, hass_client: ClientSessionGenerator
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config: dict[str, Any],
url: str,
content_type: str,
) -> None:
"""Test local_file camera content_type."""
cam_config_jpg = {
"name": "test_jpg",
"platform": "local_file",
"file_path": "/path/to/image.jpg",
}
cam_config_png = {
"name": "test_png",
"platform": "local_file",
"file_path": "/path/to/image.png",
}
cam_config_svg = {
"name": "test_svg",
"platform": "local_file",
"file_path": "/path/to/image.svg",
}
cam_config_noext = {
"name": "test_no_ext",
"platform": "local_file",
"file_path": "/path/to/image",
}
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
options=config,
entry_id="1",
)
config_entry.add_to_hass(hass)
with (
mock.patch("os.path.isfile", mock.Mock(return_value=True)),
mock.patch("os.access", mock.Mock(return_value=True)),
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
):
await async_setup_component(
hass,
"camera",
{
"camera": [
cam_config_jpg,
cam_config_png,
cam_config_svg,
cam_config_noext,
]
},
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
client = await hass_client()
image = "hello"
m_open = mock.mock_open(read_data=image.encode())
with mock.patch(
"homeassistant.components.local_file.camera.open", m_open, create=True
):
resp_1 = await client.get("/api/camera_proxy/camera.test_jpg")
resp_2 = await client.get("/api/camera_proxy/camera.test_png")
resp_3 = await client.get("/api/camera_proxy/camera.test_svg")
resp_4 = await client.get("/api/camera_proxy/camera.test_no_ext")
m_open = mock_open(read_data=image.encode())
with patch("homeassistant.components.local_file.camera.open", m_open, create=True):
resp_1 = await client.get(url)
assert resp_1.status == HTTPStatus.OK
assert resp_1.content_type == "image/jpeg"
assert resp_1.content_type == content_type
body = await resp_1.text()
assert body == image
assert resp_2.status == HTTPStatus.OK
assert resp_2.content_type == "image/png"
body = await resp_2.text()
assert body == image
assert resp_3.status == HTTPStatus.OK
assert resp_3.content_type == "image/svg+xml"
body = await resp_3.text()
assert body == image
# default mime type
assert resp_4.status == HTTPStatus.OK
assert resp_4.content_type == "image/jpeg"
body = await resp_4.text()
assert body == image
async def test_update_file_path(hass: HomeAssistant) -> None:
@pytest.mark.parametrize(
"get_config",
[
{
"name": DEFAULT_NAME,
"file_path": "mock/path.jpg",
}
],
)
async def test_update_file_path(
hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test update_file_path service."""
# Setup platform
with (
mock.patch("os.path.isfile", mock.Mock(return_value=True)),
mock.patch("os.access", mock.Mock(return_value=True)),
mock.patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
mock.Mock(return_value=(None, None)),
),
):
camera_1 = {"platform": "local_file", "file_path": "mock/path.jpg"}
camera_2 = {
"platform": "local_file",
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
options={
"name": "local_file_camera_2",
"file_path": "mock/path_2.jpg",
}
await async_setup_component(hass, "camera", {"camera": [camera_1, camera_2]})
},
entry_id="2",
)
config_entry.add_to_hass(hass)
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
# Fetch state and check motion detection attribute
state = hass.states.get("camera.local_file")
assert state.attributes.get("friendly_name") == "Local File"
assert state.attributes.get("file_path") == "mock/path.jpg"
# Fetch state and check motion detection attribute
state = hass.states.get("camera.local_file")
assert state.attributes.get("friendly_name") == "Local File"
assert state.attributes.get("file_path") == "mock/path.jpg"
service_data = {"entity_id": "camera.local_file", "file_path": "new/path.jpg"}
service_data = {"entity_id": "camera.local_file", "file_path": "new/path.jpg"}
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
await hass.services.async_call(
DOMAIN,
SERVICE_UPDATE_FILE_PATH,
@ -224,12 +191,12 @@ async def test_update_file_path(hass: HomeAssistant) -> None:
blocking=True,
)
state = hass.states.get("camera.local_file")
assert state.attributes.get("file_path") == "new/path.jpg"
state = hass.states.get("camera.local_file")
assert state.attributes.get("file_path") == "new/path.jpg"
# Check that local_file_camera_2 file_path is still as configured
state = hass.states.get("camera.local_file_camera_2")
assert state.attributes.get("file_path") == "mock/path_2.jpg"
# Check that local_file_camera_2 file_path is still as configured
state = hass.states.get("camera.local_file_camera_2")
assert state.attributes.get("file_path") == "mock/path_2.jpg"
# Assert it fails if file is not readable
service_data = {
@ -245,3 +212,76 @@ async def test_update_file_path(hass: HomeAssistant) -> None:
service_data,
blocking=True,
)
async def test_import_from_yaml_success(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Test import."""
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "local_file",
"file_path": "mock.file",
}
},
)
await hass.async_block_till_done()
assert hass.config_entries.async_has_entries(DOMAIN)
state = hass.states.get("camera.config_test")
assert state.attributes.get("file_path") == "mock.file"
issue = issue_registry.async_get_issue(
HOMEASSISTANT_DOMAIN, f"deprecated_yaml_{DOMAIN}"
)
assert issue
assert issue.translation_key == "deprecated_yaml"
async def test_import_from_yaml_fails(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Test import fails due to not accessible file."""
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=False)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "local_file",
"file_path": "mock.file",
}
},
)
await hass.async_block_till_done()
assert not hass.config_entries.async_has_entries(DOMAIN)
assert not hass.states.get("camera.config_test")
issue = issue_registry.async_get_issue(
DOMAIN, f"no_access_path_{slugify("mock.file")}"
)
assert issue
assert issue.translation_key == "no_access_path"

View File

@ -0,0 +1,235 @@
"""Test the Scrape config flow."""
from __future__ import annotations
from unittest.mock import AsyncMock, Mock, patch
import pytest
from homeassistant import config_entries
from homeassistant.components.local_file.const import DEFAULT_NAME, DOMAIN
from homeassistant.const import CONF_FILE_PATH, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
async def test_form_sensor(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
"""Test we get the form for sensor."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: DEFAULT_NAME,
CONF_FILE_PATH: "mock.file",
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["version"] == 1
assert result["options"] == {
CONF_NAME: DEFAULT_NAME,
CONF_FILE_PATH: "mock.file",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_options_flow(hass: HomeAssistant, loaded_entry: MockConfigEntry) -> None:
"""Test options flow."""
result = await hass.config_entries.options.async_init(loaded_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_FILE_PATH: "mock.new.file"},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {CONF_NAME: DEFAULT_NAME, CONF_FILE_PATH: "mock.new.file"}
await hass.async_block_till_done()
# Check the entity was updated, no new entity was created
assert len(hass.states.async_all()) == 1
state = hass.states.get("camera.local_file")
assert state is not None
async def test_validation_options(
hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test validation."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=False)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: DEFAULT_NAME,
CONF_FILE_PATH: "mock.file",
},
)
await hass.async_block_till_done()
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "not_readable_path"}
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: DEFAULT_NAME,
CONF_FILE_PATH: "mock.new.file",
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["version"] == 1
assert result["options"] == {
CONF_NAME: DEFAULT_NAME,
CONF_FILE_PATH: "mock.new.file",
}
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.usefixtures("mock_setup_entry")
async def test_entry_already_exist(
hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test abort when entry already exist."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: DEFAULT_NAME,
CONF_FILE_PATH: "mock.file",
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_import(hass: HomeAssistant) -> None:
"""Test import."""
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
"name": DEFAULT_NAME,
"file_path": "mock/path.jpg",
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["version"] == 1
assert result["options"] == {
CONF_NAME: DEFAULT_NAME,
CONF_FILE_PATH: "mock/path.jpg",
}
@pytest.mark.usefixtures("mock_setup_entry")
async def test_import_already_exist(
hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test import abort existing entry."""
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=True)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
CONF_NAME: DEFAULT_NAME,
CONF_FILE_PATH: "mock.file",
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

View File

@ -0,0 +1,47 @@
"""Test Statistics component setup process."""
from __future__ import annotations
from unittest.mock import Mock, patch
from homeassistant.components.local_file.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER, ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_unload_entry(hass: HomeAssistant, loaded_entry: MockConfigEntry) -> None:
"""Test unload an entry."""
assert loaded_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(loaded_entry.entry_id)
await hass.async_block_till_done()
assert loaded_entry.state is ConfigEntryState.NOT_LOADED
async def test_file_not_readable_during_startup(
hass: HomeAssistant,
get_config: dict[str, str],
) -> None:
"""Test a warning is shown setup when file is not readable."""
config_entry = MockConfigEntry(
domain=DOMAIN,
source=SOURCE_USER,
options=get_config,
entry_id="1",
)
config_entry.add_to_hass(hass)
with (
patch("os.path.isfile", Mock(return_value=True)),
patch("os.access", Mock(return_value=False)),
patch(
"homeassistant.components.local_file.camera.mimetypes.guess_type",
Mock(return_value=(None, None)),
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_ERROR