Allow configuring username and password in generic camera config flow (#73804)

* Add ability to use user & pw not in stream url

* Increase test coverage to 100%

* Increase test coverage

* Verify that stream source includes user:pass

* Code review: refactor test to use MockConfigEntry

* Code review: Improve test docstring

* Edit comment; retrigger CI.

Co-authored-by: Dave T <davet2001@users.noreply.github.com>

homeassistant/components/generic/camera.py

@@ -7,6 +7,7 @@ from typing import Any
 
 import httpx
 import voluptuous as vol
+import yarl
 
 from homeassistant.components.camera import (
     DEFAULT_CONTENT_TYPE,
@@ -146,6 +147,8 @@ class GenericCamera(Camera):
         self.hass = hass
         self._attr_unique_id = identifier
         self._authentication = device_info.get(CONF_AUTHENTICATION)
+        self._username = device_info.get(CONF_USERNAME)
+        self._password = device_info.get(CONF_PASSWORD)
         self._name = device_info.get(CONF_NAME, title)
         self._still_image_url = device_info.get(CONF_STILL_IMAGE_URL)
         if (
@@ -223,7 +226,11 @@ class GenericCamera(Camera):
             return None
 
         try:
-            return self._stream_source.async_render(parse_result=False)
+            stream_url = self._stream_source.async_render(parse_result=False)
+            url = yarl.URL(stream_url)
+            if not url.user and not url.password and self._username and self._password:
+                url = url.with_user(self._username).with_password(self._password)
+            return str(url)
         except TemplateError as err:
             _LOGGER.error("Error parsing template %s: %s", self._stream_source, err)
             return None

homeassistant/components/generic/config_flow.py

@@ -221,6 +221,14 @@ async def async_test_stream(
         stream_options[CONF_RTSP_TRANSPORT] = rtsp_transport
     if info.get(CONF_USE_WALLCLOCK_AS_TIMESTAMPS):
         stream_options[CONF_USE_WALLCLOCK_AS_TIMESTAMPS] = True
+
+    url = yarl.URL(stream_source)
+    if not url.user and not url.password:
+        username = info.get(CONF_USERNAME)
+        password = info.get(CONF_PASSWORD)
+        if username and password:
+            url = url.with_user(username).with_password(password)
+            stream_source = str(url)
     try:
         stream = create_stream(hass, stream_source, stream_options, "test_stream")
         hls_provider = stream.add_provider(HLS_PROVIDER)

tests/components/generic/test_camera.py

@@ -8,12 +8,24 @@ import httpx
 import pytest
 import respx
 
-from homeassistant.components.camera import async_get_mjpeg_stream
+from homeassistant.components.camera import (
+    async_get_mjpeg_stream,
+    async_get_stream_source,
+)
+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,
+)
 from homeassistant.components.websocket_api.const import TYPE_RESULT
 from homeassistant.config_entries import SOURCE_IMPORT
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, CONF_VERIFY_SSL
 from homeassistant.setup import async_setup_component
 
-from tests.common import AsyncMock, Mock
+from tests.common import AsyncMock, Mock, MockConfigEntry
 
 
 @respx.mock
@@ -184,23 +196,29 @@ async def test_stream_source(hass, hass_client, hass_ws_client, fakeimgbytes_png
     respx.get("http://example.com/0a").respond(stream=fakeimgbytes_png)
 
     hass.states.async_set("sensor.temp", "0")
-    assert await async_setup_component(
-        hass,
-        "camera",
-        {
-            "camera": {
-                "name": "config_test",
-                "platform": "generic",
-                "still_image_url": "http://example.com",
-                "stream_source": 'http://example.com/{{ states.sensor.temp.state + "a" }}',
-                "limit_refetch_to_url_change": True,
-            },
+    mock_entry = MockConfigEntry(
+        title="config_test",
+        domain=DOMAIN,
+        data={},
+        options={
+            CONF_STILL_IMAGE_URL: "http://example.com",
+            CONF_STREAM_SOURCE: 'http://example.com/{{ states.sensor.temp.state + "a" }}',
+            CONF_LIMIT_REFETCH_TO_URL_CHANGE: True,
+            CONF_FRAMERATE: 2,
+            CONF_CONTENT_TYPE: "image/png",
+            CONF_VERIFY_SSL: False,
+            CONF_USERNAME: "barney",
+            CONF_PASSWORD: "betty",
         },
     )
+    mock_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_entry.entry_id)
     assert await async_setup_component(hass, "stream", {})
     await hass.async_block_till_done()
 
     hass.states.async_set("sensor.temp", "5")
+    stream_source = await async_get_stream_source(hass, "camera.config_test")
+    assert stream_source == "http://barney:betty@example.com/5a"
 
     with patch(
         "homeassistant.components.camera.Stream.endpoint_url",

tests/components/generic/test_config_flow.py

@@ -10,6 +10,7 @@ import respx
 
 from homeassistant import config_entries, data_entry_flow
 from homeassistant.components.camera import async_get_image
+from homeassistant.components.generic.config_flow import slug
 from homeassistant.components.generic.const import (
     CONF_CONTENT_TYPE,
     CONF_FRAMERATE,
@@ -517,6 +518,17 @@ async def test_options_template_error(hass, fakeimgbytes_png, mock_create_stream
     assert result5["errors"] == {"stream_source": "template_error"}
 
 
+async def test_slug(hass, caplog):
+    """
+    Test that the slug function generates an error in case of invalid template.
+
+    Other paths in the slug function are already tested by other tests.
+    """
+    result = slug(hass, "http://127.0.0.2/testurl/{{1/0}}")
+    assert result is None
+    assert "Syntax error in" in caplog.text
+
+
 @respx.mock
 async def test_options_only_stream(hass, fakeimgbytes_png, mock_create_stream):
     """Test the options flow without a still_image_url."""