jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Auto set content type for stream-only in generic camera (#70200)

Browse Source
This commit is contained in:
Dave T 2022-04-18 06:28:25 +01:00 committed by GitHub
parent 24e817a173
commit 32f9aefb40
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 22 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
homeassistant/components/generic/config_flow.py
View File

@ -109,20 +109,6 @@ def build_schema(
return vol.Schema(spec)
def build_schema_content_type(user_input: dict[str, Any] | MappingProxyType[str, Any]):
"""Create schema for conditional 2nd page specifying stream content_type."""
return vol.Schema(
{
vol.Required(
CONF_CONTENT_TYPE,
description={
"suggested_value": user_input.get(CONF_CONTENT_TYPE, "image/jpeg")
},
): str,
}
)
def get_image_type(image):
"""Get the format of downloaded bytes that could be an image."""
fmt = None
@ -283,17 +269,16 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
if not errors:
user_input[CONF_CONTENT_TYPE] = still_format
user_input[CONF_LIMIT_REFETCH_TO_URL_CHANGE] = False
if user_input.get(CONF_STILL_IMAGE_URL):
await self.async_set_unique_id(self.flow_id)
return self.async_create_entry(
title=name, data={}, options=user_input
)
# If user didn't specify a still image URL,
# we can't (yet) autodetect it from the stream.
# Show a conditional 2nd page to ask them the content type.
self.cached_user_input = user_input
self.cached_title = name
return await self.async_step_content_type()
if still_url is None:
# If user didn't specify a still image URL,
# The automatically generated still image that stream generates
# is always jpeg
user_input[CONF_CONTENT_TYPE] = "image/jpeg"
await self.async_set_unique_id(self.flow_id)
return self.async_create_entry(
title=name, data={}, options=user_input
)
else:
user_input = DEFAULT_DATA.copy()
@ -303,22 +288,6 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_content_type(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the user's choice for stream content_type."""
if user_input is not None:
user_input = self.cached_user_input | user_input
await self.async_set_unique_id(self.flow_id)
return self.async_create_entry(
title=self.cached_title, data={}, options=user_input
)
return self.async_show_form(
step_id="content_type",
data_schema=build_schema_content_type({}),
errors={},
)
async def async_step_import(self, import_config) -> FlowResult:
"""Handle config import from yaml."""
# abort if we've already got this one.
@ -362,6 +331,11 @@ class GenericOptionsFlowHandler(OptionsFlow):
stream_url = user_input.get(CONF_STREAM_SOURCE)
if not errors:
title = slug_url(still_url) or slug_url(stream_url) or DEFAULT_NAME
if still_url is None:
# If user didn't specify a still image URL,
# The automatically generated still image that stream generates
# is always jpeg
still_format = "image/jpeg"
data = {
CONF_AUTHENTICATION: user_input.get(CONF_AUTHENTICATION),
CONF_STREAM_SOURCE: user_input.get(CONF_STREAM_SOURCE),
@ -376,30 +350,12 @@ class GenericOptionsFlowHandler(OptionsFlow):
CONF_FRAMERATE: user_input[CONF_FRAMERATE],
CONF_VERIFY_SSL: user_input[CONF_VERIFY_SSL],
}
if still_url:
return self.async_create_entry(
title=title,
data=data,
)
self.cached_title = title
self.cached_user_input = data
return await self.async_step_content_type()
return self.async_create_entry(
title=title,
data=data,
)
return self.async_show_form(
step_id="init",
data_schema=build_schema(user_input or self.config_entry.options, True),
errors=errors,
)
async def async_step_content_type(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the user's choice for stream content_type."""
if user_input is not None:
user_input = self.cached_user_input | user_input
return self.async_create_entry(title=self.cached_title, data=user_input)
return self.async_show_form(
step_id="content_type",
data_schema=build_schema_content_type(self.cached_user_input),
errors={},
)

19
tests/components/generic/test_config_flow.py
View File

@ -203,15 +203,10 @@ async def test_form_only_stream(hass, mock_av_open, fakeimgbytes_jpg):
data = TESTDATA.copy()
data.pop(CONF_STILL_IMAGE_URL)
with mock_av_open as mock_setup:
result2 = await hass.config_entries.flow.async_configure(
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
data,
)
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_CONTENT_TYPE: "image/jpeg"},
)
assert result3["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result3["title"] == "127_0_0_1_testurl_2"
@ -516,20 +511,12 @@ async def test_options_only_stream(hass, fakeimgbytes_png, mock_av_open):
assert result["step_id"] == "init"
# try updating the config options
result2 = await hass.config_entries.options.async_configure(
result3 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input=data,
)
# Should be shown a 2nd form
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["step_id"] == "content_type"
result3 = await hass.config_entries.options.async_configure(
result2["flow_id"],
user_input={CONF_CONTENT_TYPE: "image/png"},
)
assert result3["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result3["data"][CONF_CONTENT_TYPE] == "image/png"
assert result3["data"][CONF_CONTENT_TYPE] == "image/jpeg"
# These below can be deleted after deprecation period is finished.