Add config flow stream preview to generic camera (#122563)

Co-authored-by: Allen Porter <allen.porter@gmail.com>

homeassistant/components/generic/camera.py

@@ -96,10 +96,9 @@ class GenericCamera(Camera):
         self._stream_source = device_info.get(CONF_STREAM_SOURCE)
         if self._stream_source:
             self._stream_source = Template(self._stream_source, hass)
-        self._limit_refetch = device_info[CONF_LIMIT_REFETCH_TO_URL_CHANGE]
-        self._attr_frame_interval = 1 / device_info[CONF_FRAMERATE]
-        if self._stream_source:
             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.content_type = device_info[CONF_CONTENT_TYPE]
         self.verify_ssl = device_info[CONF_VERIFY_SSL]
         if device_info.get(CONF_RTSP_TRANSPORT):

homeassistant/components/generic/config_flow.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import asyncio
 from collections.abc import Mapping
 import contextlib
-from datetime import datetime
+from datetime import datetime, timedelta
 from errno import EHOSTUNREACH, EIO
 import io
 import logging
@@ -17,18 +17,21 @@ import PIL.Image
 import voluptuous as vol
 import yarl
 
+from homeassistant.components import websocket_api
 from homeassistant.components.camera import (
     CAMERA_IMAGE_TIMEOUT,
+    DOMAIN as CAMERA_DOMAIN,
     DynamicStreamSettings,
     _async_get_image,
 )
-from homeassistant.components.http import HomeAssistantView
+from homeassistant.components.http.view import HomeAssistantView
 from homeassistant.components.stream import (
     CONF_RTSP_TRANSPORT,
     CONF_USE_WALLCLOCK_AS_TIMESTAMPS,
     HLS_PROVIDER,
     RTSP_TRANSPORTS,
     SOURCE_TIMEOUT,
+    Stream,
     create_stream,
 )
 from homeassistant.config_entries import (
@@ -49,7 +52,9 @@ from homeassistant.const import (
 from homeassistant.core import HomeAssistant
 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.setup import async_prepare_setup_platform
 from homeassistant.util import slugify
 
 from .camera import GenericCamera, generate_auth
@@ -79,6 +84,15 @@ SUPPORTED_IMAGE_TYPES = {"png", "jpeg", "gif", "svg+xml", "webp"}
 IMAGE_PREVIEWS_ACTIVE = "previews"
 
 
+class InvalidStreamException(HomeAssistantError):
+    """Error to indicate an invalid stream."""
+
+    def __init__(self, error: str, details: str | None = None) -> None:
+        """Initialize the error."""
+        super().__init__(error)
+        self.details = details
+
+
 def build_schema(
     user_input: Mapping[str, Any],
     is_options_flow: bool = False,
@@ -231,12 +245,16 @@ def slug(
     return None
 
 
-async def async_test_stream(
+async def async_test_and_preview_stream(
     hass: HomeAssistant, info: Mapping[str, Any]
-) -> dict[str, str]:
+) -> Stream | None:
-    """Verify that the stream is valid before we create an entity."""
+    """Verify that the stream is valid before we create an entity.
+
+    Returns the stream object if valid. Raises InvalidStreamException if not.
+    The stream object is used to preview the video in the UI.
+    """
     if not (stream_source := info.get(CONF_STREAM_SOURCE)):
-        return {}
+        return None
     # Import from stream.worker as stream cannot reexport from worker
     # without forcing the av dependency on default_config
     # pylint: disable-next=import-outside-toplevel
@@ -248,7 +266,7 @@ async def async_test_stream(
         stream_source = stream_source.async_render(parse_result=False)
     except TemplateError as err:
         _LOGGER.warning("Problem rendering template %s: %s", stream_source, err)
-        return {CONF_STREAM_SOURCE: "template_error"}
+        raise InvalidStreamException("template_error") from err
     stream_options: dict[str, str | bool | float] = {}
     if rtsp_transport := info.get(CONF_RTSP_TRANSPORT):
         stream_options[CONF_RTSP_TRANSPORT] = rtsp_transport
@@ -257,10 +275,10 @@ async def async_test_stream(
 
     try:
         url = yarl.URL(stream_source)
-    except ValueError:
+    except ValueError as err:
-        return {CONF_STREAM_SOURCE: "malformed_url"}
+        raise InvalidStreamException("malformed_url") from err
     if not url.is_absolute():
-        return {CONF_STREAM_SOURCE: "relative_url"}
+        raise InvalidStreamException("relative_url")
     if not url.user and not url.password:
         username = info.get(CONF_USERNAME)
         password = info.get(CONF_PASSWORD)
@@ -273,29 +291,28 @@ async def async_test_stream(
             stream_source,
             stream_options,
             DynamicStreamSettings(),
-            "test_stream",
+            f"{DOMAIN}.test_stream",
         )
         hls_provider = stream.add_provider(HLS_PROVIDER)
-        await stream.start()
-        if not await hls_provider.part_recv(timeout=SOURCE_TIMEOUT):
-            hass.async_create_task(stream.stop())
-            return {CONF_STREAM_SOURCE: "timeout"}
-        await stream.stop()
     except StreamWorkerError as err:
-        return {CONF_STREAM_SOURCE: "unknown_with_details", "error_details": str(err)}
+        raise InvalidStreamException("unknown_with_details", str(err)) from err
-    except PermissionError:
+    except PermissionError as err:
-        return {CONF_STREAM_SOURCE: "stream_not_permitted"}
+        raise InvalidStreamException("stream_not_permitted") from err
     except OSError as err:
         if err.errno == EHOSTUNREACH:
-            return {CONF_STREAM_SOURCE: "stream_no_route_to_host"}
+            raise InvalidStreamException("stream_no_route_to_host") from err
         if err.errno == EIO:  # input/output error
-            return {CONF_STREAM_SOURCE: "stream_io_error"}
+            raise InvalidStreamException("stream_io_error") from err
         raise
     except HomeAssistantError as err:
         if "Stream integration is not set up" in str(err):
-            return {CONF_STREAM_SOURCE: "stream_not_set_up"}
+            raise InvalidStreamException("stream_not_set_up") from err
         raise
-    return {}
+    await stream.start()
+    if not await hls_provider.part_recv(timeout=SOURCE_TIMEOUT):
+        hass.async_create_task(stream.stop())
+        raise InvalidStreamException("timeout")
+    return stream
 
 
 def register_preview(hass: HomeAssistant) -> None:
@@ -316,6 +333,7 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
     def __init__(self) -> None:
         """Initialize Generic ConfigFlow."""
         self.preview_cam: dict[str, Any] = {}
+        self.preview_stream: Stream | None = None
         self.user_input: dict[str, Any] = {}
         self.title = ""
 
@@ -326,14 +344,6 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
         """Get the options flow for this handler."""
         return GenericOptionsFlowHandler()
 
-    def check_for_existing(self, options: dict[str, Any]) -> bool:
-        """Check whether an existing entry is using the same URLs."""
-        return any(
-            entry.options.get(CONF_STILL_IMAGE_URL) == options.get(CONF_STILL_IMAGE_URL)
-            and entry.options.get(CONF_STREAM_SOURCE) == options.get(CONF_STREAM_SOURCE)
-            for entry in self._async_current_entries()
-        )
-
     async def async_step_user(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
@@ -349,10 +359,17 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
                 errors["base"] = "no_still_image_or_stream_url"
             else:
                 errors, still_format = await async_test_still(hass, user_input)
-                errors = errors | await async_test_stream(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)
+                    if err.details:
+                        errors["error_details"] = err.details
+                    self.preview_stream = None
                 if not errors:
                     user_input[CONF_CONTENT_TYPE] = still_format
-                    user_input[CONF_LIMIT_REFETCH_TO_URL_CHANGE] = False
                     still_url = user_input.get(CONF_STILL_IMAGE_URL)
                     stream_url = user_input.get(CONF_STREAM_SOURCE)
                     name = (
@@ -365,14 +382,9 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
                         user_input[CONF_CONTENT_TYPE] = "image/jpeg"
                     self.user_input = user_input
                     self.title = name
-
-                    if still_url is None:
-                        return self.async_create_entry(
-                            title=self.title, data={}, options=self.user_input
-                        )
                     # temporary preview for user to check the image
                     self.preview_cam = user_input
-                    return await self.async_step_user_confirm_still()
+                    return await self.async_step_user_confirm()
                 if "error_details" in errors:
                     description_placeholders["error"] = errors.pop("error_details")
         elif self.user_input:
@@ -386,11 +398,14 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
             errors=errors,
         )
 
-    async def async_step_user_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_user()
             return self.async_create_entry(
@@ -399,7 +414,7 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
         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_still",
+            step_id="user_confirm",
             data_schema=vol.Schema(
                 {
                     vol.Required(CONF_CONFIRMED_OK, default=False): bool,
@@ -407,8 +422,14 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
             ),
             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 GenericOptionsFlowHandler(OptionsFlow):
     """Handle Generic IP Camera options."""
@@ -423,13 +444,21 @@ 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
             )
-            errors = errors | await async_test_stream(hass, 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:
@@ -449,6 +478,8 @@ class GenericOptionsFlowHandler(OptionsFlow):
                 # 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")
         return self.async_show_form(
             step_id="init",
             data_schema=build_schema(
@@ -456,6 +487,7 @@ class GenericOptionsFlowHandler(OptionsFlow):
                 True,
                 self.show_advanced_options,
             ),
+            description_placeholders=description_placeholders,
             errors=errors,
         )
 
@@ -518,3 +550,59 @@ class CameraImagePreview(HomeAssistantView):
             CAMERA_IMAGE_TIMEOUT,
         )
         return web.Response(body=image.content, content_type=image.content_type)
+
+
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "generic_camera/start_preview",
+        vol.Required("flow_id"): str,
+        vol.Optional("flow_type"): vol.Any("config_flow"),
+        vol.Optional("user_input"): dict,
+    }
+)
+@websocket_api.async_response
+async def ws_start_preview(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Generate websocket handler for the camera still/stream 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
+    )
+    user_input = flow.preview_cam
+
+    # Create an EntityPlatform, needed for name translations
+    platform = await async_prepare_setup_platform(hass, {}, CAMERA_DOMAIN, DOMAIN)
+    entity_platform = EntityPlatform(
+        hass=hass,
+        logger=_LOGGER,
+        domain=CAMERA_DOMAIN,
+        platform_name=DOMAIN,
+        platform=platform,
+        scan_interval=timedelta(seconds=3600),
+        entity_namespace=None,
+    )
+    await entity_platform.async_load_translations()
+
+    ha_still_url = None
+    ha_stream_url = None
+
+    if user_input.get(CONF_STILL_IMAGE_URL):
+        ha_still_url = f"/api/generic/preview_flow_image/{msg['flow_id']}?t={datetime.now().isoformat()}"
+        _LOGGER.debug("Got preview still URL: %s", ha_still_url)
+
+    if ha_stream := flow.preview_stream:
+        ha_stream_url = ha_stream.endpoint_url(HLS_PROVIDER)
+        _LOGGER.debug("Got preview stream URL: %s", ha_stream_url)
+
+    connection.send_message(
+        websocket_api.event_message(
+            msg["id"],
+            {"attributes": {"still_url": ha_still_url, "stream_url": ha_stream_url}},
+        )
+    )

homeassistant/components/generic/manifest.json

@@ -3,7 +3,7 @@
   "name": "Generic Camera",
   "codeowners": ["@davet2001"],
   "config_flow": true,
-  "dependencies": ["http"],
+  "dependencies": ["http", "stream"],
   "documentation": "https://www.home-assistant.io/integrations/generic",
   "integration_type": "device",
   "iot_class": "local_push",

homeassistant/components/generic/strings.json

@@ -39,11 +39,11 @@
           "verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
         }
       },
-      "user_confirm_still": {
+      "user_confirm": {
-        "title": "Preview",
+        "title": "Confirmation",
-        "description": "![Camera Still Image Preview]({preview_url})",
+        "description": "Please wait for previews to load...",
         "data": {
-          "confirmed_ok": "This image looks good."
+          "confirmed_ok": "Everything looks good."
         }
       }
     }
@@ -68,15 +68,16 @@
         }
       },
       "confirm_still": {
-        "title": "[%key:component::generic::config::step::user_confirm_still::title%]",
+        "title": "Preview",
-        "description": "[%key:component::generic::config::step::user_confirm_still::description%]",
+        "description": "![Camera Still Image Preview]({preview_url})",
         "data": {
-          "confirmed_ok": "[%key:component::generic::config::step::user_confirm_still::data::confirmed_ok%]"
+          "confirmed_ok": "This image looks good."
         }
       }
     },
     "error": {
       "unknown": "[%key:common::config_flow::error::unknown%]",
+      "unknown_with_details": "[%key:common::config_flow::error::unknown_with_details]",
       "already_exists": "[%key:component::generic::config::error::already_exists%]",
       "unable_still_load": "[%key:component::generic::config::error::unable_still_load%]",
       "unable_still_load_auth": "[%key:component::generic::config::error::unable_still_load_auth%]",

tests/components/generic/conftest.py

@@ -71,16 +71,18 @@ def fakeimg_gif(fakeimgbytes_gif: bytes) -> Generator[None]:
     respx.pop("fake_img")
 
 
-@pytest.fixture(scope="package")
+@pytest.fixture
-def mock_create_stream() -> _patch[MagicMock]:
+def mock_create_stream(hass: HomeAssistant) -> _patch[MagicMock]:
     """Mock create stream."""
-    mock_stream = Mock()
+    mock_stream = MagicMock()
+    mock_stream.hass = hass
     mock_provider = Mock()
     mock_provider.part_recv = AsyncMock()
     mock_provider.part_recv.return_value = True
     mock_stream.add_provider.return_value = mock_provider
     mock_stream.start = AsyncMock()
     mock_stream.stop = AsyncMock()
+    mock_stream.endpoint_url.return_value = "http://127.0.0.1/nothing"
     return patch(
         "homeassistant.components.generic.config_flow.create_stream",
         return_value=mock_stream,

tests/components/generic/test_config_flow.py

@@ -9,6 +9,7 @@ import os.path
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, PropertyMock, _patch, patch
 
+from freezegun.api import FrozenDateTimeFactory
 import httpx
 import pytest
 import respx
@@ -44,8 +45,8 @@ from homeassistant.data_entry_flow import FlowResultType
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import entity_registry as er
 
-from tests.common import MockConfigEntry
+from tests.common import MockConfigEntry, async_fire_time_changed
-from tests.typing import ClientSessionGenerator
+from tests.typing import ClientSessionGenerator, WebSocketGenerator
 
 TESTDATA = {
     CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
@@ -75,6 +76,7 @@ async def test_form(
     hass_client: ClientSessionGenerator,
     user_flow: ConfigFlowResult,
     mock_create_stream: _patch[MagicMock],
+    hass_ws_client: WebSocketGenerator,
 ) -> None:
     """Test the form with a normal set of settings."""
 
@@ -90,18 +92,29 @@ async def test_form(
             TESTDATA,
         )
         assert result1["type"] is FlowResultType.FORM
-        assert result1["step_id"] == "user_confirm_still"
+        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()
+        flow_id = user_flow["flow_id"]
+        await ws_client.send_json_auto_id(
+            {
+                "type": "generic_camera/start_preview",
+                "flow_id": flow_id,
+            },
+        )
+        _ = await ws_client.receive_json()
+
         result2 = await hass.config_entries.flow.async_configure(
             result1["flow_id"],
             user_input={CONF_CONFIRMED_OK: True},
         )
-        await hass.async_block_till_done()
     assert result2["type"] is FlowResultType.CREATE_ENTRY
     assert result2["title"] == "127_0_0_1"
     assert result2["options"] == {
@@ -110,13 +123,11 @@ async def test_form(
         CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
         CONF_USERNAME: "fred_flintstone",
         CONF_PASSWORD: "bambam",
-        CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
         CONF_CONTENT_TYPE: "image/png",
-        CONF_FRAMERATE: 5,
+        CONF_FRAMERATE: 5.0,
         CONF_VERIFY_SSL: False,
     }
 
-    await hass.async_block_till_done()
     # Check that the preview image is disabled after.
     resp = await client.get(preview_url)
     assert resp.status == HTTPStatus.NOT_FOUND
@@ -145,7 +156,7 @@ async def test_form_only_stillimage(
         )
         await hass.async_block_till_done()
         assert result1["type"] is FlowResultType.FORM
-        assert result1["step_id"] == "user_confirm_still"
+        assert result1["step_id"] == "user_confirm"
         result2 = await hass.config_entries.flow.async_configure(
             result1["flow_id"],
             user_input={CONF_CONFIRMED_OK: True},
@@ -157,9 +168,8 @@ async def test_form_only_stillimage(
         CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
         CONF_USERNAME: "fred_flintstone",
         CONF_PASSWORD: "bambam",
-        CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
         CONF_CONTENT_TYPE: "image/png",
-        CONF_FRAMERATE: 5,
+        CONF_FRAMERATE: 5.0,
         CONF_VERIFY_SSL: False,
     }
 
@@ -167,13 +177,13 @@ async def test_form_only_stillimage(
 
 
 @respx.mock
-async def test_form_reject_still_preview(
+async def test_form_reject_preview(
     hass: HomeAssistant,
     fakeimgbytes_png: bytes,
     mock_create_stream: _patch[MagicMock],
     user_flow: ConfigFlowResult,
 ) -> None:
-    """Test we go back to the config screen if the user rejects the still preview."""
+    """Test we go back to the config screen if the user rejects the preview."""
     respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
     with mock_create_stream:
         result1 = await hass.config_entries.flow.async_configure(
@@ -181,7 +191,7 @@ async def test_form_reject_still_preview(
             TESTDATA,
         )
     assert result1["type"] is FlowResultType.FORM
-    assert result1["step_id"] == "user_confirm_still"
+    assert result1["step_id"] == "user_confirm"
     result2 = await hass.config_entries.flow.async_configure(
         result1["flow_id"],
         user_input={CONF_CONFIRMED_OK: False},
@@ -211,7 +221,7 @@ async def test_form_still_preview_cam_off(
             TESTDATA,
         )
         assert result1["type"] is FlowResultType.FORM
-        assert result1["step_id"] == "user_confirm_still"
+        assert result1["step_id"] == "user_confirm"
         preview_url = result1["description_placeholders"]["preview_url"]
         # Try to view the image, should be unavailable.
         client = await hass_client()
@@ -233,7 +243,7 @@ async def test_form_only_stillimage_gif(
             data,
         )
         assert result1["type"] is FlowResultType.FORM
-        assert result1["step_id"] == "user_confirm_still"
+        assert result1["step_id"] == "user_confirm"
         result2 = await hass.config_entries.flow.async_configure(
             result1["flow_id"],
             user_input={CONF_CONFIRMED_OK: True},
@@ -258,7 +268,7 @@ async def test_form_only_svg_whitespace(
             data,
         )
         assert result1["type"] is FlowResultType.FORM
-        assert result1["step_id"] == "user_confirm_still"
+        assert result1["step_id"] == "user_confirm"
         result2 = await hass.config_entries.flow.async_configure(
             result1["flow_id"],
             user_input={CONF_CONFIRMED_OK: True},
@@ -293,7 +303,7 @@ async def test_form_only_still_sample(
             data,
         )
         assert result1["type"] is FlowResultType.FORM
-        assert result1["step_id"] == "user_confirm_still"
+        assert result1["step_id"] == "user_confirm"
         result2 = await hass.config_entries.flow.async_configure(
             result1["flow_id"],
             user_input={CONF_CONFIRMED_OK: True},
@@ -310,13 +320,13 @@ async def test_form_only_still_sample(
         (
             "http://localhost:812{{3}}/static/icons/favicon-apple-180x180.png",
             "http://localhost:8123/static/icons/favicon-apple-180x180.png",
-            "user_confirm_still",
+            "user_confirm",
             None,
         ),
         (
             "{% if 1 %}https://bla{% else %}https://yo{% endif %}",
             "https://bla/",
-            "user_confirm_still",
+            "user_confirm",
             None,
         ),
         (
@@ -385,7 +395,7 @@ async def test_form_rtsp_mode(
             user_flow["flow_id"], data
         )
         assert result1["type"] is FlowResultType.FORM
-        assert result1["step_id"] == "user_confirm_still"
+        assert result1["step_id"] == "user_confirm"
         result2 = await hass.config_entries.flow.async_configure(
             result1["flow_id"],
             user_input={CONF_CONFIRMED_OK: True},
@@ -399,13 +409,11 @@ async def test_form_rtsp_mode(
         CONF_RTSP_TRANSPORT: "tcp",
         CONF_USERNAME: "fred_flintstone",
         CONF_PASSWORD: "bambam",
-        CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
         CONF_CONTENT_TYPE: "image/png",
-        CONF_FRAMERATE: 5,
+        CONF_FRAMERATE: 5.0,
         CONF_VERIFY_SSL: False,
     }
 
-    await hass.async_block_till_done()
     assert len(mock_setup.mock_calls) == 1
 
 
@@ -419,33 +427,36 @@ async def test_form_only_stream(
     data = TESTDATA.copy()
     data.pop(CONF_STILL_IMAGE_URL)
     data[CONF_STREAM_SOURCE] = "rtsp://user:pass@127.0.0.1/testurl/2"
-    with mock_create_stream as mock_setup:
+    with mock_create_stream:
         result1 = await hass.config_entries.flow.async_configure(
             user_flow["flow_id"],
             data,
         )
-    assert result1["type"] is FlowResultType.CREATE_ENTRY
+
-    assert result1["title"] == "127_0_0_1"
+    assert result1["type"] is FlowResultType.FORM
-    assert result1["options"] == {
+    with mock_create_stream:
+        result2 = await hass.config_entries.flow.async_configure(
+            result1["flow_id"],
+            user_input={CONF_CONFIRMED_OK: True},
+        )
+
+    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_LIMIT_REFETCH_TO_URL_CHANGE: False,
         CONF_CONTENT_TYPE: "image/jpeg",
-        CONF_FRAMERATE: 5,
+        CONF_FRAMERATE: 5.0,
         CONF_VERIFY_SSL: False,
     }
 
-    await hass.async_block_till_done()
-
     with patch(
         "homeassistant.components.camera._async_get_stream_image",
         return_value=fakeimgbytes_jpg,
     ):
         image_obj = await async_get_image(hass, "camera.127_0_0_1")
         assert image_obj.content == fakeimgbytes_jpg
-    assert len(mock_setup.mock_calls) == 1
 
 
 async def test_form_still_and_stream_not_provided(
@@ -512,7 +523,6 @@ async def test_form_image_http_exceptions(
             user_flow["flow_id"],
             TESTDATA,
         )
-    await hass.async_block_till_done()
 
     assert result2["type"] is FlowResultType.FORM
     assert result2["errors"] == expected_message
@@ -531,7 +541,6 @@ async def test_form_stream_invalidimage(
             user_flow["flow_id"],
             TESTDATA,
         )
-    await hass.async_block_till_done()
 
     assert result2["type"] is FlowResultType.FORM
     assert result2["errors"] == {"still_image_url": "invalid_still_image"}
@@ -550,7 +559,6 @@ async def test_form_stream_invalidimage2(
             user_flow["flow_id"],
             TESTDATA,
         )
-    await hass.async_block_till_done()
 
     assert result2["type"] is FlowResultType.FORM
     assert result2["errors"] == {"still_image_url": "unable_still_load_no_image"}
@@ -569,7 +577,6 @@ async def test_form_stream_invalidimage3(
             user_flow["flow_id"],
             TESTDATA,
         )
-    await hass.async_block_till_done()
 
     assert result2["type"] is FlowResultType.FORM
     assert result2["errors"] == {"still_image_url": "invalid_still_image"}
@@ -585,6 +592,8 @@ async def test_form_stream_timeout(
         "homeassistant.components.generic.config_flow.create_stream"
     ) as create_stream:
         create_stream.return_value.start = AsyncMock()
+        create_stream.return_value.stop = AsyncMock()
+        create_stream.return_value.hass = hass
         create_stream.return_value.add_provider.return_value.part_recv = AsyncMock()
         create_stream.return_value.add_provider.return_value.part_recv.return_value = (
             False
@@ -727,6 +736,37 @@ async def test_form_oserror(hass: HomeAssistant, user_flow: ConfigFlowResult) ->
         )
 
 
+@respx.mock
+async def test_form_stream_preview_auto_timeout(
+    hass: HomeAssistant,
+    user_flow: ConfigFlowResult,
+    mock_create_stream: _patch[MagicMock],
+    freezer: FrozenDateTimeFactory,
+    fakeimgbytes_png: bytes,
+) -> None:
+    """Test that the stream preview times out after 10mins."""
+    respx.get("http://fred_flintstone:bambam@127.0.0.1/testurl/2").respond(
+        stream=fakeimgbytes_png
+    )
+    data = TESTDATA.copy()
+    data.pop(CONF_STILL_IMAGE_URL)
+
+    with mock_create_stream as mock_stream:
+        result1 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            data,
+        )
+        assert result1["type"] is FlowResultType.FORM
+        assert result1["step_id"] == "user_confirm"
+
+        freezer.tick(600 + 12)
+        async_fire_time_changed(hass)
+        await hass.async_block_till_done()
+
+    mock_str = mock_stream.return_value
+    mock_str.start.assert_awaited_once()
+
+
 @respx.mock
 async def test_options_template_error(
     hass: HomeAssistant, fakeimgbytes_png: bytes, mock_create_stream: _patch[MagicMock]
@@ -842,7 +882,6 @@ async def test_options_only_stream(
     )
     mock_entry.add_to_hass(hass)
     await hass.config_entries.async_setup(mock_entry.entry_id)
-    await hass.async_block_till_done()
 
     result = await hass.config_entries.options.async_init(mock_entry.entry_id)
     assert result["type"] is FlowResultType.FORM
@@ -864,6 +903,27 @@ async def test_options_only_stream(
     assert result3["data"][CONF_CONTENT_TYPE] == "image/jpeg"
 
 
+@respx.mock
+@pytest.mark.usefixtures("fakeimg_png")
+async def test_form_options_stream_worker_error(
+    hass: HomeAssistant, config_entry: MockConfigEntry
+) -> None:
+    """Test we handle a StreamWorkerError and pass the message through."""
+
+    result = await hass.config_entries.options.async_init(config_entry.entry_id)
+    with patch(
+        "homeassistant.components.generic.config_flow.create_stream",
+        side_effect=StreamWorkerError("Some message"),
+    ):
+        result2 = await hass.config_entries.options.async_configure(
+            result["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] is FlowResultType.FORM
+    assert result2["errors"] == {"stream_source": "unknown_with_details"}
+    assert result2["description_placeholders"] == {"error": "Some message"}
+
+
 @pytest.mark.usefixtures("fakeimg_png")
 async def test_unload_entry(hass: HomeAssistant) -> None:
     """Test unloading the generic IP Camera entry."""