jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 00:37:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add stream preview to options flow in generic camera (#133927)

Browse Source 
* Add stream preview to options flow

* Increase test coverage

* Code review: use correct flow handler type in cast

* Restore test coverage to 100%

* Remove error and test that can't be triggered yet
This commit is contained in:
Dave T 2024-12-30 23:46:42 +00:00 committed by GitHub
parent 57b7635b70
commit bf59241dab
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 148 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -343,7 +343,6 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
) -> ConfigFlowResult:
"""Handle the start of the config flow."""
errors = {}
description_placeholders = {}
hass = self.hass
if user_input:
# Secondary validation because serialised vol can't seem to handle this complexity:
@ -359,8 +358,6 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
)
except InvalidStreamException as err:
errors[CONF_STREAM_SOURCE] = str(err)
if err.details:
errors["error_details"] = err.details
self.preview_stream = None
if not errors:
user_input[CONF_CONTENT_TYPE] = still_format
@ -379,8 +376,6 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
# temporary preview for user to check the image
self.preview_cam = user_input
return await self.async_step_user_confirm()
if "error_details" in errors:
description_placeholders["error"] = errors.pop("error_details")
elif self.user_input:
user_input = self.user_input
else:
@ -388,7 +383,6 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_show_form(
step_id="user",
data_schema=build_schema(user_input),
description_placeholders=description_placeholders,
errors=errors,
)
@ -406,7 +400,6 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
title=self.title, data={}, options=self.user_input
)
register_preview(self.hass)
preview_url = f"/api/generic/preview_flow_image/{self.flow_id}?t={datetime.now().isoformat()}"
return self.async_show_form(
step_id="user_confirm",
data_schema=vol.Schema(
@ -414,7 +407,6 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
vol.Required(CONF_CONFIRMED_OK, default=False): bool,
}
),
description_placeholders={"preview_url": preview_url},
errors=None,
preview="generic_camera",
)
@ -431,6 +423,7 @@ class GenericOptionsFlowHandler(OptionsFlow):
def __init__(self) -> None:
"""Initialize Generic IP Camera options flow."""
self.preview_cam: dict[str, Any] = {}
self.preview_stream: Stream | None = None
self.user_input: dict[str, Any] = {}
async def async_step_init(
@ -438,42 +431,45 @@ class GenericOptionsFlowHandler(OptionsFlow):
) -> ConfigFlowResult:
"""Manage Generic IP Camera options."""
errors: dict[str, str] = {}
description_placeholders = {}
hass = self.hass
if user_input is not None:
errors, still_format = await async_test_still(
hass, self.config_entry.options | user_input
)
try:
await async_test_and_preview_stream(hass, user_input)
except InvalidStreamException as err:
errors[CONF_STREAM_SOURCE] = str(err)
if err.details:
errors["error_details"] = err.details
# Stream preview during options flow not yet implemented
still_url = user_input.get(CONF_STILL_IMAGE_URL)
if not errors:
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_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),
}
self.user_input = data
# temporary preview for user to check the image
self.preview_cam = data
return await self.async_step_confirm_still()
if "error_details" in errors:
description_placeholders["error"] = errors.pop("error_details")
if user_input:
# Secondary validation because serialised vol can't seem to handle this complexity:
if not user_input.get(CONF_STILL_IMAGE_URL) and not user_input.get(
CONF_STREAM_SOURCE
):
errors["base"] = "no_still_image_or_stream_url"
else:
errors, still_format = await async_test_still(hass, user_input)
try:
self.preview_stream = await async_test_and_preview_stream(
hass, user_input
)
except InvalidStreamException as err:
errors[CONF_STREAM_SOURCE] = str(err)
self.preview_stream = None
if not errors:
user_input[CONF_CONTENT_TYPE] = still_format
still_url = user_input.get(CONF_STILL_IMAGE_URL)
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_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),
}
self.user_input = data
# temporary preview for user to check the image
self.preview_cam = data
return await self.async_step_user_confirm()
elif self.user_input:
user_input = self.user_input
return self.async_show_form(
step_id="init",
data_schema=build_schema(
@ -481,15 +477,17 @@ class GenericOptionsFlowHandler(OptionsFlow):
True,
self.show_advanced_options,
),
description_placeholders=description_placeholders,
errors=errors,
)
async def async_step_confirm_still(
async def async_step_user_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle user clicking confirm after still preview."""
if user_input:
if ha_stream := self.preview_stream:
# Kill off the temp stream we created.
await ha_stream.stop()
if not user_input.get(CONF_CONFIRMED_OK):
return await self.async_step_init()
return self.async_create_entry(
@ -497,18 +495,22 @@ class GenericOptionsFlowHandler(OptionsFlow):
data=self.user_input,
)
register_preview(self.hass)
preview_url = f"/api/generic/preview_flow_image/{self.flow_id}?t={datetime.now().isoformat()}"
return self.async_show_form(
step_id="confirm_still",
step_id="user_confirm",
data_schema=vol.Schema(
{
vol.Required(CONF_CONFIRMED_OK, default=False): bool,
}
),
description_placeholders={"preview_url": preview_url},
errors=None,
preview="generic_camera",
)
@staticmethod
async def async_setup_preview(hass: HomeAssistant) -> None:
"""Set up preview WS API."""
websocket_api.async_register_command(hass, ws_start_preview)
class CameraImagePreview(HomeAssistantView):
"""Camera view to temporarily serve an image."""
@ -550,7 +552,7 @@ class CameraImagePreview(HomeAssistantView):
{
vol.Required("type"): "generic_camera/start_preview",
vol.Required("flow_id"): str,
vol.Optional("flow_type"): vol.Any("config_flow"),
vol.Optional("flow_type"): vol.Any("config_flow", "options_flow"),
vol.Optional("user_input"): dict,
}
)
@ -564,10 +566,17 @@ async def ws_start_preview(
_LOGGER.debug("Generating websocket handler for generic camera preview")
flow_id = msg["flow_id"]
flow = cast(
GenericIPCamConfigFlow,
hass.config_entries.flow._progress.get(flow_id), # noqa: SLF001
)
flow: GenericIPCamConfigFlow | GenericOptionsFlowHandler
if msg.get("flow_type", "config_flow") == "config_flow":
flow = cast(
GenericIPCamConfigFlow,
hass.config_entries.flow._progress.get(flow_id), # noqa: SLF001
)
else: # (flow type == "options flow")
flow = cast(
GenericOptionsFlowHandler,
hass.config_entries.options._progress.get(flow_id), # noqa: SLF001
)
user_input = flow.preview_cam
# Create an EntityPlatform, needed for name translations

8
homeassistant/components/generic/strings.json
View File

@ -67,11 +67,11 @@
"use_wallclock_as_timestamps": "This option may correct segmenting or crashing issues arising from buggy timestamp implementations on some cameras"
}
},
"confirm_still": {
"title": "Preview",
"description": "![Camera Still Image Preview]({preview_url})",
"user_confirm": {
"title": "Confirmation",
"description": "Please wait for previews to load...",
"data": {
"confirmed_ok": "This image looks good."
"confirmed_ok": "Everything looks good."
}
}
},

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

@ -92,12 +92,6 @@ async def test_form(
)
assert result1["type"] is FlowResultType.FORM
assert result1["step_id"] == "user_confirm"
client = await hass_client()
preview_url = result1["description_placeholders"]["preview_url"]
# Check the preview image works.
resp = await client.get(preview_url)
assert resp.status == HTTPStatus.OK
assert await resp.read() == fakeimgbytes_png
# HA should now be serving a WS connection for a preview stream.
ws_client = await hass_ws_client()
@ -108,7 +102,14 @@ async def test_form(
"flow_id": flow_id,
},
)
_ = await ws_client.receive_json()
json = await ws_client.receive_json()
client = await hass_client()
still_preview_url = json["event"]["attributes"]["still_url"]
# Check the preview image works.
resp = await client.get(still_preview_url)
assert resp.status == HTTPStatus.OK
assert await resp.read() == fakeimgbytes_png
result2 = await hass.config_entries.flow.async_configure(
result1["flow_id"],
@ -128,7 +129,7 @@ async def test_form(
}
# Check that the preview image is disabled after.
resp = await client.get(preview_url)
resp = await client.get(still_preview_url)
assert resp.status == HTTPStatus.NOT_FOUND
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@ -206,6 +207,7 @@ async def test_form_still_preview_cam_off(
mock_create_stream: _patch[MagicMock],
user_flow: ConfigFlowResult,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test camera errors are triggered during preview."""
with (
@ -221,10 +223,23 @@ async def test_form_still_preview_cam_off(
)
assert result1["type"] is FlowResultType.FORM
assert result1["step_id"] == "user_confirm"
preview_url = result1["description_placeholders"]["preview_url"]
# HA should now be serving a WS connection for a preview stream.
ws_client = await hass_ws_client()
flow_id = user_flow["flow_id"]
await ws_client.send_json_auto_id(
{
"type": "generic_camera/start_preview",
"flow_id": flow_id,
},
)
json = await ws_client.receive_json()
client = await hass_client()
still_preview_url = json["event"]["attributes"]["still_url"]
# Try to view the image, should be unavailable.
client = await hass_client()
resp = await client.get(preview_url)
resp = await client.get(still_preview_url)
assert resp.status == HTTPStatus.SERVICE_UNAVAILABLE
@ -686,7 +701,7 @@ async def test_form_no_route_to_host(
async def test_form_stream_io_error(
hass: HomeAssistant, user_flow: ConfigFlowResult
) -> None:
"""Test we handle no io error when setting up stream."""
"""Test we handle an io error when setting up stream."""
with patch(
"homeassistant.components.generic.config_flow.create_stream",
side_effect=OSError(errno.EIO, "Input/output error"),
@ -779,7 +794,7 @@ async def test_options_template_error(
user_input=data,
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "confirm_still"
assert result2["step_id"] == "user_confirm"
result2a = await hass.config_entries.options.async_configure(
result2["flow_id"], user_input={CONF_CONFIRMED_OK: True}
@ -874,7 +889,7 @@ async def test_options_only_stream(
user_input=data,
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "confirm_still"
assert result2["step_id"] == "user_confirm"
result3 = await hass.config_entries.options.async_configure(
result2["flow_id"], user_input={CONF_CONFIRMED_OK: True}
@ -883,6 +898,35 @@ async def test_options_only_stream(
assert result3["data"][CONF_CONTENT_TYPE] == "image/jpeg"
async def test_options_still_and_stream_not_provided(
hass: HomeAssistant,
) -> None:
"""Test we show a suitable error if neither still or stream URL are provided."""
data = TESTDATA.copy()
mock_entry = MockConfigEntry(
title="Test Camera",
domain=DOMAIN,
data={},
options=data,
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
result = await hass.config_entries.options.async_init(mock_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
data.pop(CONF_STILL_IMAGE_URL)
data.pop(CONF_STREAM_SOURCE)
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input=data,
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "no_still_image_or_stream_url"}
@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_options_permission_error(
@ -976,10 +1020,15 @@ async def test_migrate_existing_ids(
@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_use_wallclock_as_timestamps_option(
hass: HomeAssistant, mock_create_stream: _patch[MagicMock]
hass: HomeAssistant,
mock_create_stream: _patch[MagicMock],
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
fakeimgbytes_png: bytes,
) -> None:
"""Test the use_wallclock_as_timestamps option flow."""
respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
mock_entry = MockConfigEntry(
title="Test Camera",
domain=DOMAIN,
@ -1005,6 +1054,25 @@ async def test_use_wallclock_as_timestamps_option(
user_input={CONF_USE_WALLCLOCK_AS_TIMESTAMPS: True, **TESTDATA},
)
assert result2["type"] is FlowResultType.FORM
ws_client = await hass_ws_client()
flow_id = result2["flow_id"]
await ws_client.send_json_auto_id(
{
"type": "generic_camera/start_preview",
"flow_id": flow_id,
"flow_type": "options_flow",
},
)
json = await ws_client.receive_json()
client = await hass_client()
still_preview_url = json["event"]["attributes"]["still_url"]
# Check the preview image works.
resp = await client.get(still_preview_url)
assert resp.status == HTTPStatus.OK
assert await resp.read() == fakeimgbytes_png
# Test what happens if user rejects the preview
result3 = await hass.config_entries.options.async_configure(
result2["flow_id"], user_input={CONF_CONFIRMED_OK: False}
@ -1020,7 +1088,7 @@ async def test_use_wallclock_as_timestamps_option(
user_input={CONF_USE_WALLCLOCK_AS_TIMESTAMPS: True, **TESTDATA},
)
assert result4["type"] is FlowResultType.FORM
assert result4["step_id"] == "confirm_still"
assert result4["step_id"] == "user_confirm"
result5 = await hass.config_entries.options.async_configure(
result4["flow_id"],
user_input={CONF_CONFIRMED_OK: True},