Generic camera handle template adjacent to portnumber (#71031)

homeassistant/components/generic/config_flow.py

@@ -130,10 +130,9 @@ async def async_test_still(hass, info) -> tuple[dict[str, str], str | None]:
     fmt = None
     if not (url := info.get(CONF_STILL_IMAGE_URL)):
         return {}, info.get(CONF_CONTENT_TYPE, "image/jpeg")
-    if not isinstance(url, template_helper.Template) and url:
-        url = cv.template(url)
-        url.hass = hass
     try:
+        if not isinstance(url, template_helper.Template):
+            url = template_helper.Template(url, hass)
         url = url.async_render(parse_result=False)
     except TemplateError as err:
         _LOGGER.warning("Problem rendering template %s: %s", url, err)
@@ -168,11 +167,20 @@ async def async_test_still(hass, info) -> tuple[dict[str, str], str | None]:
     return {}, f"image/{fmt}"
 
 
-def slug_url(url) -> str | None:
+def slug(hass, template) -> str | None:
     """Convert a camera url into a string suitable for a camera name."""
-    if not url:
+    if not template:
         return None
-    return slugify(yarl.URL(url).host)
+    if not isinstance(template, template_helper.Template):
+        template = template_helper.Template(template, hass)
+    try:
+        url = template.async_render(parse_result=False)
+        return slugify(yarl.URL(url).host)
+    except TemplateError as err:
+        _LOGGER.error("Syntax error in '%s': %s", template.template, err)
+    except (ValueError, TypeError) as err:
+        _LOGGER.error("Syntax error in '%s': %s", url, err)
+    return None
 
 
 async def async_test_stream(hass, info) -> dict[str, str]:
@@ -252,6 +260,7 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
     ) -> FlowResult:
         """Handle the start of the config flow."""
         errors = {}
+        hass = self.hass
         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(
@@ -263,8 +272,7 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
                 errors = errors | await async_test_stream(self.hass, user_input)
                 still_url = user_input.get(CONF_STILL_IMAGE_URL)
                 stream_url = user_input.get(CONF_STREAM_SOURCE)
-                name = slug_url(still_url) or slug_url(stream_url) or DEFAULT_NAME
-
+                name = slug(hass, still_url) or slug(hass, stream_url) or DEFAULT_NAME
                 if not errors:
                     user_input[CONF_CONTENT_TYPE] = still_format
                     user_input[CONF_LIMIT_REFETCH_TO_URL_CHANGE] = False
@@ -295,7 +303,8 @@ class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
         still_url = import_config.get(CONF_STILL_IMAGE_URL)
         stream_url = import_config.get(CONF_STREAM_SOURCE)
         name = import_config.get(
-            CONF_NAME, slug_url(still_url) or slug_url(stream_url) or DEFAULT_NAME
+            CONF_NAME,
+            slug(self.hass, still_url) or slug(self.hass, stream_url) or DEFAULT_NAME,
         )
         if CONF_LIMIT_REFETCH_TO_URL_CHANGE not in import_config:
             import_config[CONF_LIMIT_REFETCH_TO_URL_CHANGE] = False
@@ -318,6 +327,7 @@ class GenericOptionsFlowHandler(OptionsFlow):
     ) -> FlowResult:
         """Manage Generic IP Camera options."""
         errors: dict[str, str] = {}
+        hass = self.hass
 
         if user_input is not None:
             errors, still_format = await async_test_still(
@@ -327,7 +337,7 @@ class GenericOptionsFlowHandler(OptionsFlow):
             still_url = user_input.get(CONF_STILL_IMAGE_URL)
             stream_url = user_input.get(CONF_STREAM_SOURCE)
             if not errors:
-                title = slug_url(still_url) or slug_url(stream_url) or DEFAULT_NAME
+                title = slug(hass, still_url) or slug(hass, 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

tests/components/generic/test_config_flow.py

@@ -165,6 +165,48 @@ async def test_form_only_still_sample(hass, user_flow, image_file):
     assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
 
 
+@respx.mock
+@pytest.mark.parametrize(
+    ("template", "url", "expected_result"),
+    [
+        # Test we can handle templates in strange parts of the url, #70961.
+        (
+            "http://localhost:812{{3}}/static/icons/favicon-apple-180x180.png",
+            "http://localhost:8123/static/icons/favicon-apple-180x180.png",
+            data_entry_flow.RESULT_TYPE_CREATE_ENTRY,
+        ),
+        (
+            "{% if 1 %}https://bla{% else %}https://yo{% endif %}",
+            "https://bla/",
+            data_entry_flow.RESULT_TYPE_CREATE_ENTRY,
+        ),
+        (
+            "http://{{example.org",
+            "http://example.org",
+            data_entry_flow.RESULT_TYPE_FORM,
+        ),
+        (
+            "invalid1://invalid:4\\1",
+            "invalid1://invalid:4%5c1",
+            data_entry_flow.RESULT_TYPE_CREATE_ENTRY,
+        ),
+    ],
+)
+async def test_still_template(
+    hass, user_flow, fakeimgbytes_png, template, url, expected_result
+) -> None:
+    """Test we can handle various templates."""
+    respx.get(url).respond(stream=fakeimgbytes_png)
+    data = TESTDATA.copy()
+    data.pop(CONF_STREAM_SOURCE)
+    data[CONF_STILL_IMAGE_URL] = template
+    result2 = await hass.config_entries.flow.async_configure(
+        user_flow["flow_id"],
+        data,
+    )
+    assert result2["type"] == expected_result
+
+
 @respx.mock
 async def test_form_rtsp_mode(hass, fakeimg_png, mock_av_open, user_flow):
     """Test we complete ok if the user enters a stream url."""