Make generic camera stream_source a template (#36123)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

homeassistant/components/generic/camera.py

@@ -40,7 +40,7 @@ DEFAULT_NAME = "Generic Camera"
 PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
     {
         vol.Required(CONF_STILL_IMAGE_URL): cv.template,
-        vol.Optional(CONF_STREAM_SOURCE, default=None): vol.Any(None, cv.string),
+        vol.Optional(CONF_STREAM_SOURCE): cv.template,
         vol.Optional(CONF_AUTHENTICATION, default=HTTP_BASIC_AUTHENTICATION): vol.In(
             [HTTP_BASIC_AUTHENTICATION, HTTP_DIGEST_AUTHENTICATION]
         ),
@@ -72,8 +72,10 @@ class GenericCamera(Camera):
         self._authentication = device_info.get(CONF_AUTHENTICATION)
         self._name = device_info.get(CONF_NAME)
         self._still_image_url = device_info[CONF_STILL_IMAGE_URL]
-        self._stream_source = device_info[CONF_STREAM_SOURCE]
+        self._stream_source = device_info.get(CONF_STREAM_SOURCE)
         self._still_image_url.hass = hass
+        if self._stream_source is not None:
+            self._stream_source.hass = hass
         self._limit_refetch = device_info[CONF_LIMIT_REFETCH_TO_URL_CHANGE]
         self._frame_interval = 1 / device_info[CONF_FRAMERATE]
         self._supported_features = SUPPORT_STREAM if self._stream_source else 0
@@ -166,4 +168,11 @@ class GenericCamera(Camera):
 
     async def stream_source(self):
         """Return the source of the stream."""
-        return self._stream_source
+        if self._stream_source is None:
+            return None
+
+        try:
+            return self._stream_source.async_render()
+        except TemplateError as err:
+            _LOGGER.error("Error parsing template %s: %s", self._stream_source, err)
+            return None

tests/components/generic/test_camera.py

@@ -1,10 +1,12 @@
 """The tests for generic camera component."""
 import asyncio
-from unittest import mock
 
+from homeassistant.components.websocket_api.const import TYPE_RESULT
 from homeassistant.const import HTTP_INTERNAL_SERVER_ERROR, HTTP_NOT_FOUND
 from homeassistant.setup import async_setup_component
 
+from tests.async_mock import patch
+
 
 async def test_fetching_url(aioclient_mock, hass, hass_client):
     """Test that it fetches the given url."""
@@ -119,7 +121,7 @@ async def test_limit_refetch(aioclient_mock, hass, hass_client):
 
     hass.states.async_set("sensor.temp", "5")
 
-    with mock.patch("async_timeout.timeout", side_effect=asyncio.TimeoutError()):
+    with patch("async_timeout.timeout", side_effect=asyncio.TimeoutError()):
         resp = await client.get("/api/camera_proxy/camera.config_test")
         assert aioclient_mock.call_count == 0
         assert resp.status == HTTP_INTERNAL_SERVER_ERROR
@@ -156,6 +158,104 @@ async def test_limit_refetch(aioclient_mock, hass, hass_client):
     assert body == "hello planet"
 
 
+async def test_stream_source(aioclient_mock, hass, hass_client, hass_ws_client):
+    """Test that the stream source is rendered."""
+    assert await async_setup_component(
+        hass,
+        "camera",
+        {
+            "camera": {
+                "name": "config_test",
+                "platform": "generic",
+                "still_image_url": "https://example.com",
+                "stream_source": 'http://example.com/{{ states.sensor.temp.state + "a" }}',
+                "limit_refetch_to_url_change": True,
+            }
+        },
+    )
+    await hass.async_block_till_done()
+
+    hass.states.async_set("sensor.temp", "5")
+
+    with patch(
+        "homeassistant.components.camera.request_stream",
+        return_value="http://home.assistant/playlist.m3u8",
+    ) as mock_request_stream:
+        # Request playlist through WebSocket
+        client = await hass_ws_client(hass)
+
+        await client.send_json(
+            {"id": 1, "type": "camera/stream", "entity_id": "camera.config_test"}
+        )
+        msg = await client.receive_json()
+
+        # Assert WebSocket response
+        assert mock_request_stream.call_count == 1
+        assert mock_request_stream.call_args[0][1] == "http://example.com/5a"
+        assert msg["id"] == 1
+        assert msg["type"] == TYPE_RESULT
+        assert msg["success"]
+        assert msg["result"]["url"][-13:] == "playlist.m3u8"
+
+        # Cause a template render error
+        hass.states.async_remove("sensor.temp")
+
+        await client.send_json(
+            {"id": 2, "type": "camera/stream", "entity_id": "camera.config_test"}
+        )
+        msg = await client.receive_json()
+
+        # Assert that no new call to the stream request should have been made
+        assert mock_request_stream.call_count == 1
+        # Assert the websocket error message
+        assert msg["id"] == 2
+        assert msg["type"] == TYPE_RESULT
+        assert msg["success"] is False
+        assert msg["error"] == {
+            "code": "start_stream_failed",
+            "message": "camera.config_test does not support play stream service",
+        }
+
+
+async def test_no_stream_source(aioclient_mock, hass, hass_client, hass_ws_client):
+    """Test a stream request without stream source option set."""
+    assert await async_setup_component(
+        hass,
+        "camera",
+        {
+            "camera": {
+                "name": "config_test",
+                "platform": "generic",
+                "still_image_url": "https://example.com",
+                "limit_refetch_to_url_change": True,
+            }
+        },
+    )
+    await hass.async_block_till_done()
+
+    with patch(
+        "homeassistant.components.camera.request_stream",
+        return_value="http://home.assistant/playlist.m3u8",
+    ) as mock_request_stream:
+        # Request playlist through WebSocket
+        client = await hass_ws_client(hass)
+
+        await client.send_json(
+            {"id": 3, "type": "camera/stream", "entity_id": "camera.config_test"}
+        )
+        msg = await client.receive_json()
+
+        # Assert the websocket error message
+        assert mock_request_stream.call_count == 0
+        assert msg["id"] == 3
+        assert msg["type"] == TYPE_RESULT
+        assert msg["success"] is False
+        assert msg["error"] == {
+            "code": "start_stream_failed",
+            "message": "camera.config_test does not support play stream service",
+        }
+
+
 async def test_camera_content_type(aioclient_mock, hass, hass_client):
     """Test generic camera with custom content_type."""
     svg_image = "<some image>"