Add advanced section for generic camera config flow

This commit is contained in:
Robert Resch 2025-07-08 17:18:35 +02:00
parent a35299d94c
commit c5a0eec441
No known key found for this signature in database
GPG Key ID: 9D9D9DCB43120143
9 changed files with 275 additions and 121 deletions

View File

@ -2,15 +2,23 @@
from __future__ import annotations
import logging
from typing import Any
from homeassistant.components.stream import (
CONF_RTSP_TRANSPORT,
CONF_USE_WALLCLOCK_AS_TIMESTAMPS,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.const import CONF_AUTHENTICATION, CONF_VERIFY_SSL, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from .const import CONF_FRAMERATE, CONF_LIMIT_REFETCH_TO_URL_CHANGE, SECTION_ADVANCED
DOMAIN = "generic"
PLATFORMS = [Platform.CAMERA]
_LOGGER = logging.getLogger(__name__)
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
@ -47,3 +55,38 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Migrate entry."""
_LOGGER.debug("Migrating from version %s:%s", entry.version, entry.minor_version)
if entry.version > 2:
# This means the user has downgraded from a future version
return False
if entry.version == 1:
# Migrate to advanced section
new_options = {**entry.options}
advanced = new_options[SECTION_ADVANCED] = {
CONF_FRAMERATE: new_options.pop(CONF_FRAMERATE),
CONF_VERIFY_SSL: new_options.pop(CONF_VERIFY_SSL),
}
# migrate optional fields
for key in (
CONF_RTSP_TRANSPORT,
CONF_USE_WALLCLOCK_AS_TIMESTAMPS,
CONF_AUTHENTICATION,
CONF_LIMIT_REFETCH_TO_URL_CHANGE,
):
if key in new_options:
advanced[key] = new_options.pop(key)
hass.config_entries.async_update_entry(entry, options=new_options, version=2)
_LOGGER.debug(
"Migration to version %s:%s successful", entry.version, entry.minor_version
)
return True

View File

@ -41,6 +41,7 @@ from .const import (
CONF_STILL_IMAGE_URL,
CONF_STREAM_SOURCE,
GET_IMAGE_TIMEOUT,
SECTION_ADVANCED,
)
_LOGGER = logging.getLogger(__name__)
@ -62,9 +63,11 @@ def generate_auth(device_info: Mapping[str, Any]) -> httpx.Auth | None:
"""Generate httpx.Auth object from credentials."""
username: str | None = device_info.get(CONF_USERNAME)
password: str | None = device_info.get(CONF_PASSWORD)
authentication = device_info.get(CONF_AUTHENTICATION)
if username and password:
if authentication == HTTP_DIGEST_AUTHENTICATION:
if (
device_info[SECTION_ADVANCED].get(CONF_AUTHENTICATION)
== HTTP_DIGEST_AUTHENTICATION
):
return httpx.DigestAuth(username=username, password=password)
return httpx.BasicAuth(username=username, password=password)
return None
@ -99,14 +102,16 @@ class GenericCamera(Camera):
if self._stream_source:
self._stream_source = Template(self._stream_source, hass)
self._attr_supported_features = CameraEntityFeature.STREAM
self._limit_refetch = device_info.get(CONF_LIMIT_REFETCH_TO_URL_CHANGE, False)
self._attr_frame_interval = 1 / device_info[CONF_FRAMERATE]
self._limit_refetch = device_info[SECTION_ADVANCED].get(
CONF_LIMIT_REFETCH_TO_URL_CHANGE, False
)
self._attr_frame_interval = 1 / device_info[SECTION_ADVANCED][CONF_FRAMERATE]
self.content_type = device_info[CONF_CONTENT_TYPE]
self.verify_ssl = device_info[CONF_VERIFY_SSL]
if device_info.get(CONF_RTSP_TRANSPORT):
self.stream_options[CONF_RTSP_TRANSPORT] = device_info[CONF_RTSP_TRANSPORT]
self.verify_ssl = device_info[SECTION_ADVANCED][CONF_VERIFY_SSL]
if rtsp_transport := device_info[SECTION_ADVANCED].get(CONF_RTSP_TRANSPORT):
self.stream_options[CONF_RTSP_TRANSPORT] = rtsp_transport
self._auth = generate_auth(device_info)
if device_info.get(CONF_USE_WALLCLOCK_AS_TIMESTAMPS):
if device_info[SECTION_ADVANCED].get(CONF_USE_WALLCLOCK_AS_TIMESTAMPS):
self.stream_options[CONF_USE_WALLCLOCK_AS_TIMESTAMPS] = True
self._last_url = None

View File

@ -50,10 +50,17 @@ from homeassistant.const import (
HTTP_DIGEST_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import section
from homeassistant.exceptions import HomeAssistantError, TemplateError
from homeassistant.helpers import config_validation as cv, template as template_helper
from homeassistant.helpers.entity_platform import EntityPlatform
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.selector import (
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from homeassistant.setup import async_prepare_setup_platform
from homeassistant.util import slugify
@ -68,16 +75,19 @@ from .const import (
DEFAULT_NAME,
DOMAIN,
GET_IMAGE_TIMEOUT,
SECTION_ADVANCED,
)
_LOGGER = logging.getLogger(__name__)
DEFAULT_DATA = {
CONF_NAME: DEFAULT_NAME,
SECTION_ADVANCED: {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
CONF_FRAMERATE: 2,
CONF_VERIFY_SSL: True,
},
}
SUPPORTED_IMAGE_TYPES = {"png", "jpeg", "gif", "svg+xml", "webp"}
@ -94,58 +104,47 @@ class InvalidStreamException(HomeAssistantError):
def build_schema(
user_input: Mapping[str, Any],
is_options_flow: bool = False,
show_advanced_options: bool = False,
) -> vol.Schema:
"""Create schema for camera config setup."""
rtsp_options = [
SelectOptionDict(
value=value,
label=name,
)
for value, name in RTSP_TRANSPORTS.items()
]
advanced_section = {
vol.Required(CONF_FRAMERATE): vol.All(
vol.Range(min=0, min_included=False), cv.positive_float
),
vol.Required(CONF_VERIFY_SSL): bool,
vol.Optional(CONF_RTSP_TRANSPORT): SelectSelector(
SelectSelectorConfig(
options=rtsp_options,
mode=SelectSelectorMode.DROPDOWN,
)
),
vol.Optional(CONF_AUTHENTICATION): vol.In(
[HTTP_BASIC_AUTHENTICATION, HTTP_DIGEST_AUTHENTICATION]
),
}
spec = {
vol.Optional(
CONF_STILL_IMAGE_URL,
description={"suggested_value": user_input.get(CONF_STILL_IMAGE_URL, "")},
): str,
vol.Optional(
CONF_STREAM_SOURCE,
description={"suggested_value": user_input.get(CONF_STREAM_SOURCE, "")},
): str,
vol.Optional(
CONF_RTSP_TRANSPORT,
description={"suggested_value": user_input.get(CONF_RTSP_TRANSPORT)},
): vol.In(RTSP_TRANSPORTS),
vol.Optional(
CONF_AUTHENTICATION,
description={"suggested_value": user_input.get(CONF_AUTHENTICATION)},
): vol.In([HTTP_BASIC_AUTHENTICATION, HTTP_DIGEST_AUTHENTICATION]),
vol.Optional(
CONF_USERNAME,
description={"suggested_value": user_input.get(CONF_USERNAME, "")},
): str,
vol.Optional(
CONF_PASSWORD,
description={"suggested_value": user_input.get(CONF_PASSWORD, "")},
): str,
vol.Required(
CONF_FRAMERATE,
description={"suggested_value": user_input.get(CONF_FRAMERATE, 2)},
): vol.All(vol.Range(min=0, min_included=False), cv.positive_float),
vol.Required(
CONF_VERIFY_SSL, default=user_input.get(CONF_VERIFY_SSL, True)
): bool,
vol.Optional(CONF_STREAM_SOURCE): str,
vol.Optional(CONF_STILL_IMAGE_URL): str,
vol.Optional(CONF_USERNAME): str,
vol.Optional(CONF_PASSWORD): str,
vol.Required(SECTION_ADVANCED): section(
vol.Schema(advanced_section), {"collapsed": True}
),
}
if is_options_flow:
spec[
vol.Required(
CONF_LIMIT_REFETCH_TO_URL_CHANGE,
default=user_input.get(CONF_LIMIT_REFETCH_TO_URL_CHANGE, False),
)
] = bool
advanced_section[vol.Optional(CONF_LIMIT_REFETCH_TO_URL_CHANGE)] = bool
if show_advanced_options:
spec[
vol.Required(
CONF_USE_WALLCLOCK_AS_TIMESTAMPS,
default=user_input.get(CONF_USE_WALLCLOCK_AS_TIMESTAMPS, False),
)
] = bool
advanced_section[vol.Optional(CONF_USE_WALLCLOCK_AS_TIMESTAMPS)] = bool
return vol.Schema(spec)
@ -187,7 +186,7 @@ async def async_test_still(
return {CONF_STILL_IMAGE_URL: "malformed_url"}, None
if not yarl_url.is_absolute():
return {CONF_STILL_IMAGE_URL: "relative_url"}, None
verify_ssl = info[CONF_VERIFY_SSL]
verify_ssl = info[SECTION_ADVANCED][CONF_VERIFY_SSL]
auth = generate_auth(info)
try:
async_client = get_async_client(hass, verify_ssl=verify_ssl)
@ -266,9 +265,9 @@ async def async_test_and_preview_stream(
_LOGGER.warning("Problem rendering template %s: %s", stream_source, err)
raise InvalidStreamException("template_error") from err
stream_options: dict[str, str | bool | float] = {}
if rtsp_transport := info.get(CONF_RTSP_TRANSPORT):
if rtsp_transport := info[SECTION_ADVANCED].get(CONF_RTSP_TRANSPORT):
stream_options[CONF_RTSP_TRANSPORT] = rtsp_transport
if info.get(CONF_USE_WALLCLOCK_AS_TIMESTAMPS):
if info[SECTION_ADVANCED].get(CONF_USE_WALLCLOCK_AS_TIMESTAMPS):
stream_options[CONF_USE_WALLCLOCK_AS_TIMESTAMPS] = True
try:
@ -324,7 +323,7 @@ def register_still_preview(hass: HomeAssistant) -> None:
class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
"""Config flow for generic IP camera."""
VERSION = 1
VERSION = 2
def __init__(self) -> None:
"""Initialize Generic ConfigFlow."""
@ -379,7 +378,7 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
user_input = DEFAULT_DATA.copy()
return self.async_show_form(
step_id="user",
data_schema=build_schema(user_input),
data_schema=self.add_suggested_values_to_schema(build_schema(), user_input),
errors=errors,
)
@ -447,13 +446,19 @@ class GenericOptionsFlowHandler(OptionsFlow):
self.preview_stream = None
if not errors:
data = {
CONF_USE_WALLCLOCK_AS_TIMESTAMPS: self.config_entry.options.get(
CONF_USE_WALLCLOCK_AS_TIMESTAMPS, False
),
**user_input,
CONF_CONTENT_TYPE: still_format
or self.config_entry.options.get(CONF_CONTENT_TYPE),
}
if (
CONF_USE_WALLCLOCK_AS_TIMESTAMPS
not in user_input[SECTION_ADVANCED]
):
user_input[SECTION_ADVANCED][
CONF_USE_WALLCLOCK_AS_TIMESTAMPS
] = self.config_entry.options[SECTION_ADVANCED].get(
CONF_USE_WALLCLOCK_AS_TIMESTAMPS, False
)
self.user_input = data
# temporary preview for user to check the image
self.preview_image_settings = data
@ -462,11 +467,13 @@ class GenericOptionsFlowHandler(OptionsFlow):
user_input = self.user_input
return self.async_show_form(
step_id="init",
data_schema=build_schema(
user_input or self.config_entry.options,
data_schema=self.add_suggested_values_to_schema(
build_schema(
True,
self.show_advanced_options,
),
user_input or self.config_entry.options,
),
errors=errors,
)

View File

@ -9,3 +9,4 @@ CONF_STILL_IMAGE_URL = "still_image_url"
CONF_STREAM_SOURCE = "stream_source"
CONF_FRAMERATE = "framerate"
GET_IMAGE_TIMEOUT = 10
SECTION_ADVANCED = "advanced"

View File

@ -30,13 +30,21 @@
"data": {
"still_image_url": "Still image URL (e.g. http://...)",
"stream_source": "Stream source URL (e.g. rtsp://...)",
"rtsp_transport": "RTSP transport protocol",
"authentication": "Authentication",
"limit_refetch_to_url_change": "Limit refetch to URL change",
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]",
"username": "[%key:common::config_flow::data::username%]"
},
"sections": {
"advanced": {
"name": "Advanced settings",
"description": "Advanced settings are only needed for special cases. Leave them unchanged unless you know what you are doing.",
"data": {
"rtsp_transport": "RTSP transport protocol",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]",
"framerate": "Frame rate (Hz)",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
"limit_refetch_to_url_change": "Limit refetch to URL change",
"authentication": "Authentication"
}
}
}
},
"user_confirm": {
@ -54,18 +62,26 @@
"data": {
"still_image_url": "[%key:component::generic::config::step::user::data::still_image_url%]",
"stream_source": "[%key:component::generic::config::step::user::data::stream_source%]",
"rtsp_transport": "[%key:component::generic::config::step::user::data::rtsp_transport%]",
"authentication": "[%key:component::generic::config::step::user::data::authentication%]",
"limit_refetch_to_url_change": "[%key:component::generic::config::step::user::data::limit_refetch_to_url_change%]",
"password": "[%key:common::config_flow::data::password%]",
"use_wallclock_as_timestamps": "Use wallclock as timestamps",
"username": "[%key:common::config_flow::data::username%]",
"framerate": "[%key:component::generic::config::step::user::data::framerate%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
"username": "[%key:common::config_flow::data::username%]"
},
"sections": {
"advanced": {
"name": "[%key:component::generic::config::step::user::sections::advanced::name%]",
"description": "[%key:component::generic::config::step::user::sections::advanced::description%]",
"data": {
"rtsp_transport": "[%key:component::generic::config::step::user::sections::advanced::data::rtsp_transport%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]",
"framerate": "[%key:component::generic::config::step::user::sections::advanced::data::framerate%]",
"limit_refetch_to_url_change": "[%key:component::generic::config::step::user::sections::advanced::data::limit_refetch_to_url_change%]",
"authentication": "[%key:component::generic::config::step::user::sections::advanced::data::authentication%]",
"use_wallclock_as_timestamps": "Use wallclock as timestamps"
},
"data_description": {
"use_wallclock_as_timestamps": "This option may correct segmenting or crashing issues arising from buggy timestamp implementations on some cameras"
}
}
}
},
"user_confirm": {
"title": "Confirmation",

View File

@ -129,13 +129,15 @@ def config_entry_fixture(hass: HomeAssistant) -> MockConfigEntry:
"stream_source": "http://janebloggs:letmein2@example.com/stream",
"username": "johnbloggs",
"password": "letmein123",
"limit_refetch_to_url_change": False,
"authentication": "basic",
"content_type": "image/jpeg",
"advanced": {
"framerate": 2.0,
"verify_ssl": True,
"content_type": "image/jpeg",
"limit_refetch_to_url_change": False,
"authentication": "basic",
},
version=1,
},
version=2,
)
entry.add_to_hass(hass)
return entry

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from collections.abc import Generator
import contextlib
from copy import deepcopy
import errno
from http import HTTPStatus
import os.path
@ -24,6 +25,7 @@ from homeassistant.components.generic.const import (
CONF_STILL_IMAGE_URL,
CONF_STREAM_SOURCE,
DOMAIN,
SECTION_ADVANCED,
)
from homeassistant.components.stream import (
CONF_RTSP_TRANSPORT,
@ -49,16 +51,13 @@ from tests.typing import ClientSessionGenerator, WebSocketGenerator
TESTDATA = {
CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_USERNAME: "fred_flintstone",
CONF_PASSWORD: "bambam",
SECTION_ADVANCED: {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_FRAMERATE: 5,
CONF_VERIFY_SSL: False,
}
TESTDATA_OPTIONS = {
CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
**TESTDATA,
},
}
TESTDATA_YAML = {
@ -114,12 +113,14 @@ async def test_form(
assert result2["options"] == {
CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_USERNAME: "fred_flintstone",
CONF_PASSWORD: "bambam",
CONF_CONTENT_TYPE: "image/png",
SECTION_ADVANCED: {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_FRAMERATE: 5.0,
CONF_VERIFY_SSL: False,
},
}
# Check that the preview image is disabled after.
@ -152,12 +153,14 @@ async def test_form_only_stillimage(
assert result2["title"] == "127_0_0_1"
assert result2["options"] == {
CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_USERNAME: "fred_flintstone",
CONF_PASSWORD: "bambam",
CONF_CONTENT_TYPE: "image/png",
SECTION_ADVANCED: {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_FRAMERATE: 5.0,
CONF_VERIFY_SSL: False,
},
}
assert respx.calls.call_count == 1
@ -385,8 +388,8 @@ async def test_form_rtsp_mode(
mock_setup_entry: _patch[MagicMock],
) -> None:
"""Test we complete ok if the user enters a stream url."""
data = TESTDATA.copy()
data[CONF_RTSP_TRANSPORT] = "tcp"
data = deepcopy(TESTDATA)
data[SECTION_ADVANCED][CONF_RTSP_TRANSPORT] = "tcp"
data[CONF_STREAM_SOURCE] = "rtsp://127.0.0.1/testurl/2"
result1 = await hass.config_entries.flow.async_configure(user_flow["flow_id"], data)
assert result1["type"] is FlowResultType.FORM
@ -399,14 +402,16 @@ async def test_form_rtsp_mode(
assert result2["title"] == "127_0_0_1"
assert result2["options"] == {
CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_STREAM_SOURCE: "rtsp://127.0.0.1/testurl/2",
CONF_RTSP_TRANSPORT: "tcp",
CONF_USERNAME: "fred_flintstone",
CONF_PASSWORD: "bambam",
CONF_CONTENT_TYPE: "image/png",
SECTION_ADVANCED: {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_FRAMERATE: 5.0,
CONF_VERIFY_SSL: False,
CONF_RTSP_TRANSPORT: "tcp",
},
}
@ -433,13 +438,15 @@ async def test_form_only_stream(
assert result2["title"] == "127_0_0_1"
assert result2["options"] == {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_STREAM_SOURCE: "rtsp://user:pass@127.0.0.1/testurl/2",
CONF_USERNAME: "fred_flintstone",
CONF_PASSWORD: "bambam",
CONF_CONTENT_TYPE: "image/jpeg",
SECTION_ADVANCED: {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_FRAMERATE: 5.0,
CONF_VERIFY_SSL: False,
},
}
with patch(
@ -457,10 +464,12 @@ async def test_form_still_and_stream_not_provided(
result2 = await hass.config_entries.flow.async_configure(
user_flow["flow_id"],
{
SECTION_ADVANCED: {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_FRAMERATE: 5,
CONF_VERIFY_SSL: False,
},
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "no_still_image_or_stream_url"}
@ -879,8 +888,17 @@ async def test_migrate_existing_ids(
hass: HomeAssistant, entity_registry: er.EntityRegistry
) -> None:
"""Test that existing ids are migrated for issue #70568."""
test_data = {
CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_USERNAME: "fred_flintstone",
CONF_PASSWORD: "bambam",
CONF_FRAMERATE: 5,
CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
CONF_VERIFY_SSL: False,
}
test_data = TESTDATA_OPTIONS.copy()
test_data[CONF_CONTENT_TYPE] = "image/png"
old_unique_id = "54321"
entity_id = "camera.sample_camera"
@ -926,9 +944,12 @@ async def test_options_use_wallclock_as_timestamps(
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
data = deepcopy(TESTDATA)
data[SECTION_ADVANCED][CONF_USE_WALLCLOCK_AS_TIMESTAMPS] = True
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_USE_WALLCLOCK_AS_TIMESTAMPS: True, **TESTDATA},
user_input=data,
)
assert result2["type"] is FlowResultType.FORM
@ -958,7 +979,7 @@ async def test_options_use_wallclock_as_timestamps(
assert result3["step_id"] == "init"
result4 = await hass.config_entries.options.async_configure(
result3["flow_id"],
user_input={CONF_USE_WALLCLOCK_AS_TIMESTAMPS: True, **TESTDATA},
user_input=data,
)
assert result4["type"] is FlowResultType.FORM
assert result4["step_id"] == "user_confirm"

View File

@ -26,11 +26,13 @@ async def test_entry_diagnostics(
"stream_source": "http://****:****@example.com/****",
"username": REDACTED,
"password": REDACTED,
"content_type": "image/jpeg",
"advanced": {
"limit_refetch_to_url_change": False,
"authentication": "basic",
"framerate": 2.0,
"verify_ssl": True,
"content_type": "image/jpeg",
},
},
}

View File

@ -2,7 +2,23 @@
import pytest
from homeassistant.components.generic.const import (
CONF_CONTENT_TYPE,
CONF_FRAMERATE,
CONF_LIMIT_REFETCH_TO_URL_CHANGE,
CONF_STILL_IMAGE_URL,
CONF_STREAM_SOURCE,
DOMAIN,
SECTION_ADVANCED,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
CONF_AUTHENTICATION,
CONF_PASSWORD,
CONF_USERNAME,
CONF_VERIFY_SSL,
HTTP_BASIC_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@ -35,3 +51,44 @@ async def test_reload_on_title_change(
assert (
hass.states.get("camera.test_camera").attributes["friendly_name"] == "New Title"
)
@pytest.mark.usefixtures("fakeimg_png")
async def test_migration_to_version_2(hass: HomeAssistant) -> None:
"""Test the File sensor with JSON entries."""
entry = MockConfigEntry(
domain=DOMAIN,
title="Test Camera",
unique_id="abc123",
data={},
options={
CONF_STILL_IMAGE_URL: "http://joebloggs:letmein1@example.com/secret1/file.jpg?pw=qwerty",
CONF_STREAM_SOURCE: "http://janebloggs:letmein2@example.com/stream",
CONF_USERNAME: "johnbloggs",
CONF_PASSWORD: "letmein123",
CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_FRAMERATE: 2.0,
CONF_VERIFY_SSL: True,
CONF_CONTENT_TYPE: "image/jpeg",
},
version=1,
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.LOADED
assert entry.version == 2
assert entry.options == {
CONF_STILL_IMAGE_URL: "http://joebloggs:letmein1@example.com/secret1/file.jpg?pw=qwerty",
CONF_STREAM_SOURCE: "http://janebloggs:letmein2@example.com/stream",
CONF_USERNAME: "johnbloggs",
CONF_PASSWORD: "letmein123",
CONF_CONTENT_TYPE: "image/jpeg",
SECTION_ADVANCED: {
CONF_FRAMERATE: 2.0,
CONF_VERIFY_SSL: True,
CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
},
}