diff --git a/CODEOWNERS b/CODEOWNERS
index 183d699d2ae..42eb1f896f8 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -359,6 +359,8 @@ homeassistant/components/garages_amsterdam/* @klaasnicolaas
 tests/components/garages_amsterdam/* @klaasnicolaas
 homeassistant/components/gdacs/* @exxamalte
 tests/components/gdacs/* @exxamalte
+homeassistant/components/generic/* @davet2001
+tests/components/generic/* @davet2001
 homeassistant/components/generic_hygrostat/* @Shulyaka
 tests/components/generic_hygrostat/* @Shulyaka
 homeassistant/components/geniushub/* @zxdavb
diff --git a/homeassistant/components/generic/__init__.py b/homeassistant/components/generic/__init__.py
index 3ca526f2029..f243f1639b3 100644
--- a/homeassistant/components/generic/__init__.py
+++ b/homeassistant/components/generic/__init__.py
@@ -1,6 +1,28 @@
 """The generic component."""
 
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
 
 DOMAIN = "generic"
 PLATFORMS = [Platform.CAMERA]
+
+
+async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Handle options update."""
+    await hass.config_entries.async_reload(entry.entry_id)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up generic IP camera from a config entry."""
+
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+
+    entry.async_on_unload(entry.add_update_listener(_async_update_listener))
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
diff --git a/homeassistant/components/generic/camera.py b/homeassistant/components/generic/camera.py
index b4aaad38618..72fec27b733 100644
--- a/homeassistant/components/generic/camera.py
+++ b/homeassistant/components/generic/camera.py
@@ -12,6 +12,7 @@ from homeassistant.components.camera import (
     SUPPORT_STREAM,
     Camera,
 )
+from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
 from homeassistant.const import (
     CONF_AUTHENTICATION,
     CONF_NAME,
@@ -23,15 +24,13 @@ from homeassistant.const import (
 )
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import TemplateError
-from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers import config_validation as cv, template as template_helper
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.httpx_client import get_async_client
-from homeassistant.helpers.reload import async_setup_reload_service
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
-from . import DOMAIN, PLATFORMS
+from . import DOMAIN
 from .const import (
-    ALLOWED_RTSP_TRANSPORT_PROTOCOLS,
     CONF_CONTENT_TYPE,
     CONF_FRAMERATE,
     CONF_LIMIT_REFETCH_TO_URL_CHANGE,
@@ -41,6 +40,7 @@ from .const import (
     DEFAULT_NAME,
     FFMPEG_OPTION_MAP,
     GET_IMAGE_TIMEOUT,
+    RTSP_TRANSPORTS,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -62,7 +62,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
             cv.small_float, cv.positive_int
         ),
         vol.Optional(CONF_VERIFY_SSL, default=True): cv.boolean,
-        vol.Optional(CONF_RTSP_TRANSPORT): vol.In(ALLOWED_RTSP_TRANSPORT_PROTOCOLS),
+        vol.Optional(CONF_RTSP_TRANSPORT): vol.In(RTSP_TRANSPORTS.keys()),
     }
 )
 
@@ -75,25 +75,78 @@ async def async_setup_platform(
 ) -> None:
     """Set up a generic IP Camera."""
 
-    await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
+    _LOGGER.warning(
+        "Loading generic IP camera via configuration.yaml is deprecated, "
+        "it will be automatically imported.  Once you have confirmed correct "
+        "operation, please remove 'generic' (IP camera) section(s) from "
+        "configuration.yaml"
+    )
+    image = config.get(CONF_STILL_IMAGE_URL)
+    stream = config.get(CONF_STREAM_SOURCE)
+    config_new = {
+        CONF_NAME: config[CONF_NAME],
+        CONF_STILL_IMAGE_URL: image.template if image is not None else None,
+        CONF_STREAM_SOURCE: stream.template if stream is not None else None,
+        CONF_AUTHENTICATION: config.get(CONF_AUTHENTICATION),
+        CONF_USERNAME: config.get(CONF_USERNAME),
+        CONF_PASSWORD: config.get(CONF_PASSWORD),
+        CONF_LIMIT_REFETCH_TO_URL_CHANGE: config.get(CONF_LIMIT_REFETCH_TO_URL_CHANGE),
+        CONF_CONTENT_TYPE: config.get(CONF_CONTENT_TYPE),
+        CONF_FRAMERATE: config.get(CONF_FRAMERATE),
+        CONF_VERIFY_SSL: config.get(CONF_VERIFY_SSL),
+    }
 
-    async_add_entities([GenericCamera(hass, config)])
+    hass.async_create_task(
+        hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_IMPORT}, data=config_new
+        )
+    )
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up a generic IP Camera."""
+
+    async_add_entities(
+        [GenericCamera(hass, entry.options, entry.unique_id, entry.title)]
+    )
+
+
+def generate_auth(device_info) -> httpx.Auth | None:
+    """Generate httpx.Auth object from credentials."""
+    username = device_info.get(CONF_USERNAME)
+    password = device_info.get(CONF_PASSWORD)
+    authentication = device_info.get(CONF_AUTHENTICATION)
+    if username:
+        if authentication == HTTP_DIGEST_AUTHENTICATION:
+            return httpx.DigestAuth(username=username, password=password)
+        return httpx.BasicAuth(username=username, password=password)
+    return None
 
 
 class GenericCamera(Camera):
     """A generic implementation of an IP camera."""
 
-    def __init__(self, hass, device_info):
+    def __init__(self, hass, device_info, identifier, title):
         """Initialize a generic camera."""
         super().__init__()
         self.hass = hass
+        self._attr_unique_id = identifier
         self._authentication = device_info.get(CONF_AUTHENTICATION)
-        self._name = device_info.get(CONF_NAME)
+        self._name = device_info.get(CONF_NAME, title)
         self._still_image_url = device_info.get(CONF_STILL_IMAGE_URL)
-        if self._still_image_url:
+        if (
+            not isinstance(self._still_image_url, template_helper.Template)
+            and self._still_image_url
+        ):
+            self._still_image_url = cv.template(self._still_image_url)
+        if self._still_image_url not in [None, ""]:
             self._still_image_url.hass = hass
         self._stream_source = device_info.get(CONF_STREAM_SOURCE)
-        if self._stream_source is not None:
+        if self._stream_source not in (None, ""):
+            if not isinstance(self._stream_source, template_helper.Template):
+                self._stream_source = cv.template(self._stream_source)
             self._stream_source.hass = hass
         self._limit_refetch = device_info[CONF_LIMIT_REFETCH_TO_URL_CHANGE]
         self._attr_frame_interval = 1 / device_info[CONF_FRAMERATE]
@@ -104,17 +157,7 @@ class GenericCamera(Camera):
             self.stream_options[FFMPEG_OPTION_MAP[CONF_RTSP_TRANSPORT]] = device_info[
                 CONF_RTSP_TRANSPORT
             ]
-
-        username = device_info.get(CONF_USERNAME)
-        password = device_info.get(CONF_PASSWORD)
-
-        if username and password:
-            if self._authentication == HTTP_DIGEST_AUTHENTICATION:
-                self._auth = httpx.DigestAuth(username=username, password=password)
-            else:
-                self._auth = httpx.BasicAuth(username=username, password=password)
-        else:
-            self._auth = None
+        self._auth = generate_auth(device_info)
 
         self._last_url = None
         self._last_image = None
diff --git a/homeassistant/components/generic/config_flow.py b/homeassistant/components/generic/config_flow.py
new file mode 100644
index 00000000000..1b484821788
--- /dev/null
+++ b/homeassistant/components/generic/config_flow.py
@@ -0,0 +1,338 @@
+"""Config flow for generic (IP Camera)."""
+from __future__ import annotations
+
+import contextlib
+from errno import EHOSTUNREACH, EIO
+from functools import partial
+import imghdr
+import logging
+from types import MappingProxyType
+from typing import Any
+from urllib.parse import urlparse, urlunparse
+
+from async_timeout import timeout
+import av
+from httpx import HTTPStatusError, RequestError, TimeoutException
+import voluptuous as vol
+
+from homeassistant.components.stream.const import SOURCE_TIMEOUT
+from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
+from homeassistant.const import (
+    CONF_AUTHENTICATION,
+    CONF_NAME,
+    CONF_PASSWORD,
+    CONF_USERNAME,
+    CONF_VERIFY_SSL,
+    HTTP_BASIC_AUTHENTICATION,
+    HTTP_DIGEST_AUTHENTICATION,
+)
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.exceptions import TemplateError
+from homeassistant.helpers import config_validation as cv, template as template_helper
+from homeassistant.helpers.httpx_client import get_async_client
+from homeassistant.util import slugify
+
+from .camera import generate_auth
+from .const import (
+    CONF_CONTENT_TYPE,
+    CONF_FRAMERATE,
+    CONF_LIMIT_REFETCH_TO_URL_CHANGE,
+    CONF_RTSP_TRANSPORT,
+    CONF_STILL_IMAGE_URL,
+    CONF_STREAM_SOURCE,
+    DEFAULT_NAME,
+    DOMAIN,
+    FFMPEG_OPTION_MAP,
+    GET_IMAGE_TIMEOUT,
+    RTSP_TRANSPORTS,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+DEFAULT_DATA = {
+    CONF_NAME: DEFAULT_NAME,
+    CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
+    CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
+    CONF_FRAMERATE: 2,
+    CONF_VERIFY_SSL: True,
+}
+
+SUPPORTED_IMAGE_TYPES = ["png", "jpeg", "svg+xml"]
+
+
+def build_schema(
+    user_input: dict[str, Any] | MappingProxyType[str, Any],
+    is_options_flow: bool = False,
+):
+    """Create schema for camera config setup."""
+    spec = {
+        vol.Optional(
+            CONF_STILL_IMAGE_URL,
+            description={"suggested_value": user_input.get(CONF_STILL_IMAGE_URL, "")},
+        ): str,
+        vol.Optional(
+            CONF_STREAM_SOURCE,
+            description={"suggested_value": user_input.get(CONF_STREAM_SOURCE, "")},
+        ): str,
+        vol.Optional(
+            CONF_RTSP_TRANSPORT,
+            description={"suggested_value": user_input.get(CONF_RTSP_TRANSPORT)},
+        ): vol.In(RTSP_TRANSPORTS),
+        vol.Optional(
+            CONF_AUTHENTICATION,
+            description={"suggested_value": user_input.get(CONF_AUTHENTICATION)},
+        ): vol.In([HTTP_BASIC_AUTHENTICATION, HTTP_DIGEST_AUTHENTICATION]),
+        vol.Optional(
+            CONF_USERNAME,
+            description={"suggested_value": user_input.get(CONF_USERNAME, "")},
+        ): str,
+        vol.Optional(
+            CONF_PASSWORD,
+            description={"suggested_value": user_input.get(CONF_PASSWORD, "")},
+        ): str,
+        vol.Required(
+            CONF_FRAMERATE,
+            description={"suggested_value": user_input.get(CONF_FRAMERATE, 2)},
+        ): int,
+        vol.Required(
+            CONF_VERIFY_SSL, default=user_input.get(CONF_VERIFY_SSL, True)
+        ): bool,
+    }
+    if is_options_flow:
+        spec[
+            vol.Required(
+                CONF_LIMIT_REFETCH_TO_URL_CHANGE,
+                default=user_input.get(CONF_LIMIT_REFETCH_TO_URL_CHANGE, False),
+            )
+        ] = bool
+    return vol.Schema(spec)
+
+
+def get_image_type(image):
+    """Get the format of downloaded bytes that could be an image."""
+    fmt = imghdr.what(None, h=image)
+    if fmt is None:
+        # if imghdr can't figure it out, could be svg.
+        with contextlib.suppress(UnicodeDecodeError):
+            if image.decode("utf-8").startswith("<svg"):
+                return "svg+xml"
+    return fmt
+
+
+async def async_test_still(hass, info) -> tuple[dict[str, str], str | None]:
+    """Verify that the still image is valid before we create an entity."""
+    fmt = None
+    if not (url := info.get(CONF_STILL_IMAGE_URL)):
+        return {}, None
+    if not isinstance(url, template_helper.Template) and url:
+        url = cv.template(url)
+        url.hass = hass
+    try:
+        url = url.async_render(parse_result=False)
+    except TemplateError as err:
+        _LOGGER.error("Error parsing template %s: %s", url, err)
+        return {CONF_STILL_IMAGE_URL: "template_error"}, None
+    verify_ssl = info.get(CONF_VERIFY_SSL)
+    auth = generate_auth(info)
+    try:
+        async_client = get_async_client(hass, verify_ssl=verify_ssl)
+        async with timeout(GET_IMAGE_TIMEOUT):
+            response = await async_client.get(url, auth=auth, timeout=GET_IMAGE_TIMEOUT)
+            response.raise_for_status()
+            image = response.content
+    except (
+        TimeoutError,
+        RequestError,
+        HTTPStatusError,
+        TimeoutException,
+    ) as err:
+        _LOGGER.error("Error getting camera image from %s: %s", url, type(err).__name__)
+        return {CONF_STILL_IMAGE_URL: "unable_still_load"}, None
+
+    if not image:
+        return {CONF_STILL_IMAGE_URL: "unable_still_load"}, None
+    fmt = get_image_type(image)
+    _LOGGER.debug(
+        "Still image at '%s' detected format: %s",
+        info[CONF_STILL_IMAGE_URL],
+        fmt,
+    )
+    if fmt not in SUPPORTED_IMAGE_TYPES:
+        return {CONF_STILL_IMAGE_URL: "invalid_still_image"}, None
+    return {}, f"image/{fmt}"
+
+
+def slug_url(url) -> str | None:
+    """Convert a camera url into a string suitable for a camera name."""
+    if not url:
+        return None
+    url_no_scheme = urlparse(url)._replace(scheme="")
+    return slugify(urlunparse(url_no_scheme).strip("/"))
+
+
+async def async_test_stream(hass, info) -> dict[str, str]:
+    """Verify that the stream is valid before we create an entity."""
+    if not (stream_source := info.get(CONF_STREAM_SOURCE)):
+        return {}
+    try:
+        # For RTSP streams, prefer TCP. This code is duplicated from
+        # homeassistant.components.stream.__init__.py:create_stream()
+        # It may be possible & better to call create_stream() directly.
+        stream_options: dict[str, str] = {}
+        if isinstance(stream_source, str) and stream_source[:7] == "rtsp://":
+            stream_options = {
+                "rtsp_flags": "prefer_tcp",
+                "stimeout": "5000000",
+            }
+        if rtsp_transport := info.get(CONF_RTSP_TRANSPORT):
+            stream_options[FFMPEG_OPTION_MAP[CONF_RTSP_TRANSPORT]] = rtsp_transport
+        _LOGGER.debug("Attempting to open stream %s", stream_source)
+        container = await hass.async_add_executor_job(
+            partial(
+                av.open,
+                stream_source,
+                options=stream_options,
+                timeout=SOURCE_TIMEOUT,
+            )
+        )
+        _ = container.streams.video[0]
+    except (av.error.FileNotFoundError):  # pylint: disable=c-extension-no-member
+        return {CONF_STREAM_SOURCE: "stream_file_not_found"}
+    except (av.error.HTTPNotFoundError):  # pylint: disable=c-extension-no-member
+        return {CONF_STREAM_SOURCE: "stream_http_not_found"}
+    except (av.error.TimeoutError):  # pylint: disable=c-extension-no-member
+        return {CONF_STREAM_SOURCE: "timeout"}
+    except av.error.HTTPUnauthorizedError:  # pylint: disable=c-extension-no-member
+        return {CONF_STREAM_SOURCE: "stream_unauthorised"}
+    except (KeyError, IndexError):
+        return {CONF_STREAM_SOURCE: "stream_no_video"}
+    except PermissionError:
+        return {CONF_STREAM_SOURCE: "stream_not_permitted"}
+    except OSError as err:
+        if err.errno == EHOSTUNREACH:
+            return {CONF_STREAM_SOURCE: "stream_no_route_to_host"}
+        if err.errno == EIO:  # input/output error
+            return {CONF_STREAM_SOURCE: "stream_io_error"}
+        raise err
+    return {}
+
+
+class GenericIPCamConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Config flow for generic IP camera."""
+
+    VERSION = 1
+
+    @staticmethod
+    def async_get_options_flow(
+        config_entry: ConfigEntry,
+    ) -> GenericOptionsFlowHandler:
+        """Get the options flow for this handler."""
+        return GenericOptionsFlowHandler(config_entry)
+
+    def check_for_existing(self, options):
+        """Check whether an existing entry is using the same URLs."""
+        return any(
+            entry.options[CONF_STILL_IMAGE_URL] == options[CONF_STILL_IMAGE_URL]
+            and entry.options[CONF_STREAM_SOURCE] == options[CONF_STREAM_SOURCE]
+            for entry in self._async_current_entries()
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the start of the config flow."""
+        errors = {}
+        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(self.hass, user_input)
+                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
+
+                if not errors:
+                    user_input[CONF_CONTENT_TYPE] = still_format
+                    user_input[CONF_LIMIT_REFETCH_TO_URL_CHANGE] = False
+                    await self.async_set_unique_id(self.flow_id)
+                    return self.async_create_entry(
+                        title=name, data={}, options=user_input
+                    )
+        else:
+            user_input = DEFAULT_DATA.copy()
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=build_schema(user_input),
+            errors=errors,
+        )
+
+    async def async_step_import(self, import_config) -> FlowResult:
+        """Handle config import from yaml."""
+        # abort if we've already got this one.
+        if self.check_for_existing(import_config):
+            return self.async_abort(reason="already_exists")
+        errors, still_format = await async_test_still(self.hass, import_config)
+        errors = errors | await async_test_stream(self.hass, import_config)
+        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
+        )
+        if CONF_LIMIT_REFETCH_TO_URL_CHANGE not in import_config:
+            import_config[CONF_LIMIT_REFETCH_TO_URL_CHANGE] = False
+        if not errors:
+            import_config[CONF_CONTENT_TYPE] = still_format
+            await self.async_set_unique_id(self.flow_id)
+            return self.async_create_entry(title=name, data={}, options=import_config)
+        _LOGGER.error(
+            "Error importing generic IP camera platform config: unexpected error '%s'",
+            list(errors.values()),
+        )
+        return self.async_abort(reason="unknown")
+
+
+class GenericOptionsFlowHandler(OptionsFlow):
+    """Handle Generic IP Camera options."""
+
+    def __init__(self, config_entry: ConfigEntry) -> None:
+        """Initialize Generic IP Camera options flow."""
+        self.config_entry = config_entry
+
+    async def async_step_init(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Manage Generic IP Camera options."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            errors, still_format = await async_test_still(self.hass, user_input)
+            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)
+            if not errors:
+                return self.async_create_entry(
+                    title=slug_url(still_url) or slug_url(stream_url) or DEFAULT_NAME,
+                    data={
+                        CONF_AUTHENTICATION: user_input.get(CONF_AUTHENTICATION),
+                        CONF_STREAM_SOURCE: user_input.get(CONF_STREAM_SOURCE),
+                        CONF_PASSWORD: user_input.get(CONF_PASSWORD),
+                        CONF_STILL_IMAGE_URL: user_input.get(CONF_STILL_IMAGE_URL),
+                        CONF_CONTENT_TYPE: still_format,
+                        CONF_USERNAME: user_input.get(CONF_USERNAME),
+                        CONF_LIMIT_REFETCH_TO_URL_CHANGE: user_input[
+                            CONF_LIMIT_REFETCH_TO_URL_CHANGE
+                        ],
+                        CONF_FRAMERATE: user_input[CONF_FRAMERATE],
+                        CONF_VERIFY_SSL: user_input[CONF_VERIFY_SSL],
+                    },
+                )
+        return self.async_show_form(
+            step_id="init",
+            data_schema=build_schema(user_input or self.config_entry.options, True),
+            errors=errors,
+        )
diff --git a/homeassistant/components/generic/const.py b/homeassistant/components/generic/const.py
index 1b3ba657ecc..60b4cec61a6 100644
--- a/homeassistant/components/generic/const.py
+++ b/homeassistant/components/generic/const.py
@@ -1,5 +1,6 @@
 """Constants for the generic (IP Camera) integration."""
 
+DOMAIN = "generic"
 DEFAULT_NAME = "Generic Camera"
 CONF_CONTENT_TYPE = "content_type"
 CONF_LIMIT_REFETCH_TO_URL_CHANGE = "limit_refetch_to_url_change"
@@ -8,6 +9,15 @@ CONF_STREAM_SOURCE = "stream_source"
 CONF_FRAMERATE = "framerate"
 CONF_RTSP_TRANSPORT = "rtsp_transport"
 FFMPEG_OPTION_MAP = {CONF_RTSP_TRANSPORT: "rtsp_transport"}
-ALLOWED_RTSP_TRANSPORT_PROTOCOLS = {"tcp", "udp", "udp_multicast", "http"}
-
+RTSP_TRANSPORTS = {
+    "tcp": "TCP",
+    "udp": "UDP",
+    "udp_multicast": "UDP Multicast",
+    "http": "HTTP",
+}
 GET_IMAGE_TIMEOUT = 10
+
+DEFAULT_USERNAME = None
+DEFAULT_PASSWORD = None
+DEFAULT_IMAGE_URL = None
+DEFAULT_STREAM_SOURCE = None
diff --git a/homeassistant/components/generic/manifest.json b/homeassistant/components/generic/manifest.json
index ab6aa18c4d2..7b967108e77 100644
--- a/homeassistant/components/generic/manifest.json
+++ b/homeassistant/components/generic/manifest.json
@@ -1,7 +1,11 @@
 {
   "domain": "generic",
   "name": "Generic Camera",
+  "config_flow": true,
+  "requirements": ["av==9.0.0"],
   "documentation": "https://www.home-assistant.io/integrations/generic",
-  "codeowners": [],
+  "codeowners": [
+    "@davet2001"
+  ],
   "iot_class": "local_push"
 }
diff --git a/homeassistant/components/generic/strings.json b/homeassistant/components/generic/strings.json
new file mode 100644
index 00000000000..eb1bfcc3c55
--- /dev/null
+++ b/homeassistant/components/generic/strings.json
@@ -0,0 +1,76 @@
+{
+  "config": {
+    "error": {
+      "unknown": "[%key:common::config_flow::error::unknown%]",
+      "already_exists": "A camera with these URL settings already exists.",
+      "unable_still_load": "Unable to load valid image from still image URL (e.g. invalid host, URL or authentication failure).  Review log for more info.",
+      "no_still_image_or_stream_url": "You must specify at least a still image or stream URL",
+      "invalid_still_image": "URL did not return a valid still image",
+      "stream_file_not_found": "File not found while trying to connect to stream (is ffmpeg installed?)",
+      "stream_http_not_found": "HTTP 404 Not found while trying to connect to stream",
+      "timeout": "Timeout while loading URL",
+      "stream_no_route_to_host": "Could not find host while trying to connect to stream",
+      "stream_io_error": "Input/Output error while trying to connect to stream. Wrong RTSP transport protocol?",
+      "stream_unauthorised": "Authorisation failed while trying to connect to stream",
+      "stream_not_permitted": "Operation not permitted while trying to connect to stream. Wrong RTSP transport protocol?",
+      "stream_no_video": "Stream has no video"
+    },
+    "abort": {
+      "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
+      "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
+    },
+    "step": {
+      "user": {
+        "description": "Enter the settings to connect to the camera.",
+        "data": {
+          "still_image_url": "Still Image URL (e.g. http://...)",
+          "stream_source": "Stream Source URL (e.g. rtsp://...)",
+          "rtsp_transport": "RTSP transport protocol",
+          "authentication": "Authentication",
+          "limit_refetch_to_url_change": "Limit refetch to url change",
+          "password": "[%key:common::config_flow::data::password%]",
+          "username": "[%key:common::config_flow::data::username%]",
+          "content_type": "Content Type",
+          "framerate": "Frame Rate (Hz)",
+          "verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
+        }
+      },
+      "confirm": {
+        "description": "[%key:common::config_flow::description::confirm_setup%]"
+      }
+    }
+  },
+  "options": {
+    "step": {
+      "init": {
+        "data": {
+          "still_image_url": "[%key:component::generic::config::step::user::data::still_image_url%]",
+          "stream_source": "[%key:component::generic::config::step::user::data::stream_source%]",
+          "rtsp_transport": "[%key:component::generic::config::step::user::data::rtsp_transport%]",
+          "authentication": "[%key:component::generic::config::step::user::data::authentication%]",
+          "limit_refetch_to_url_change": "[%key:component::generic::config::step::user::data::limit_refetch_to_url_change%]",
+          "password": "[%key:common::config_flow::data::password%]",
+          "username": "[%key:common::config_flow::data::username%]",
+          "content_type": "[%key:component::generic::config::step::user::data::content_type%]",
+          "framerate": "[%key:component::generic::config::step::user::data::framerate%]",
+          "verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
+        }
+      }
+    },
+    "error": {
+      "unknown": "[%key:common::config_flow::error::unknown%]",
+      "already_exists": "[%key:component::generic::config::error::already_exists%]",
+      "unable_still_load": "[%key:component::generic::config::error::unable_still_load%]",
+      "no_still_image_or_stream_url": "[%key:component::generic::config::error::no_still_image_or_stream_url%]",
+      "invalid_still_image": "[%key:component::generic::config::error::invalid_still_image%]",
+      "stream_file_not_found": "[%key:component::generic::config::error::stream_file_not_found%]",
+      "stream_http_not_found": "[%key:component::generic::config::error::stream_http_not_found%]",
+      "timeout": "[%key:component::generic::config::error::timeout%]",
+      "stream_no_route_to_host": "[%key:component::generic::config::error::stream_no_route_to_host%]",
+      "stream_io_error": "[%key:component::generic::config::error::stream_io_error%]",
+      "stream_unauthorised": "[%key:component::generic::config::error::stream_unauthorised%]",
+      "stream_not_permitted": "[%key:component::generic::config::error::stream_not_permitted%]",
+      "stream_no_video": "[%key:component::generic::config::error::stream_no_video%]"
+    }
+  }
+}
diff --git a/homeassistant/components/generic/translations/en.json b/homeassistant/components/generic/translations/en.json
new file mode 100644
index 00000000000..5346c2b5106
--- /dev/null
+++ b/homeassistant/components/generic/translations/en.json
@@ -0,0 +1,76 @@
+{
+    "config": {
+        "abort": {
+            "no_devices_found": "No devices found on the network",
+            "single_instance_allowed": "Already configured. Only a single configuration possible."
+        },
+        "error": {
+            "already_exists": "A camera with these URL settings already exists.",
+            "invalid_still_image": "URL did not return a valid still image",
+            "no_still_image_or_stream_url": "You must specify at least a still image or stream URL",
+            "stream_file_not_found": "File not found while trying to connect to stream (is ffmpeg installed?)",
+            "stream_http_not_found": "HTTP 404 Not found while trying to connect to stream",
+            "stream_io_error": "Input/Output error while trying to connect to stream. Wrong RTSP transport protocol?",
+            "stream_no_route_to_host": "Could not find host while trying to connect to stream",
+            "stream_no_video": "Stream has no video",
+            "stream_not_permitted": "Operation not permitted while trying to connect to stream. Wrong RTSP transport protocol?",
+            "stream_unauthorised": "Authorisation failed while trying to connect to stream",
+            "timeout": "Timeout while loading URL",
+            "unable_still_load": "Unable to load valid image from still image URL (e.g. invalid host, URL or authentication failure).  Review log for more info.",
+            "unknown": "Unexpected error"
+        },
+        "step": {
+            "confirm": {
+                "description": "Do you want to start set up?"
+            },
+            "user": {
+                "data": {
+                    "authentication": "Authentication",
+                    "content_type": "Content Type",
+                    "framerate": "Frame Rate (Hz)",
+                    "limit_refetch_to_url_change": "Limit refetch to url change",
+                    "password": "Password",
+                    "rtsp_transport": "RTSP transport protocol",
+                    "still_image_url": "Still Image URL (e.g. http://...)",
+                    "stream_source": "Stream Source URL (e.g. rtsp://...)",
+                    "username": "Username",
+                    "verify_ssl": "Verify SSL certificate"
+                },
+                "description": "Enter the settings to connect to the camera."
+            }
+        }
+    },
+    "options": {
+        "error": {
+            "already_exists": "A camera with these URL settings already exists.",
+            "invalid_still_image": "URL did not return a valid still image",
+            "no_still_image_or_stream_url": "You must specify at least a still image or stream URL",
+            "stream_file_not_found": "File not found while trying to connect to stream (is ffmpeg installed?)",
+            "stream_http_not_found": "HTTP 404 Not found while trying to connect to stream",
+            "stream_io_error": "Input/Output error while trying to connect to stream. Wrong RTSP transport protocol?",
+            "stream_no_route_to_host": "Could not find host while trying to connect to stream",
+            "stream_no_video": "Stream has no video",
+            "stream_not_permitted": "Operation not permitted while trying to connect to stream. Wrong RTSP transport protocol?",
+            "stream_unauthorised": "Authorisation failed while trying to connect to stream",
+            "timeout": "Timeout while loading URL",
+            "unable_still_load": "Unable to load valid image from still image URL (e.g. invalid host, URL or authentication failure).  Review log for more info.",
+            "unknown": "Unexpected error"
+        },
+        "step": {
+            "init": {
+                "data": {
+                    "authentication": "Authentication",
+                    "content_type": "Content Type",
+                    "framerate": "Frame Rate (Hz)",
+                    "limit_refetch_to_url_change": "Limit refetch to url change",
+                    "password": "Password",
+                    "rtsp_transport": "RTSP transport protocol",
+                    "still_image_url": "Still Image URL (e.g. http://...)",
+                    "stream_source": "Stream Source URL (e.g. rtsp://...)",
+                    "username": "Username",
+                    "verify_ssl": "Verify SSL certificate"
+                }
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py
index 1795f0c74b4..b8a5651a9e8 100644
--- a/homeassistant/generated/config_flows.py
+++ b/homeassistant/generated/config_flows.py
@@ -119,6 +119,7 @@ FLOWS = {
         "fronius",
         "garages_amsterdam",
         "gdacs",
+        "generic",
         "geofency",
         "geonetnz_quakes",
         "geonetnz_volcano",
diff --git a/requirements_all.txt b/requirements_all.txt
index 0ff5a4d328a..36432c9c5ea 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -347,6 +347,7 @@ auroranoaa==0.0.2
 # homeassistant.components.aurora_abb_powerone
 aurorapy==0.2.6
 
+# homeassistant.components.generic
 # homeassistant.components.stream
 av==9.0.0
 
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index 50a8b9e110e..0d984a4a06c 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -274,6 +274,7 @@ auroranoaa==0.0.2
 # homeassistant.components.aurora_abb_powerone
 aurorapy==0.2.6
 
+# homeassistant.components.generic
 # homeassistant.components.stream
 av==9.0.0
 
diff --git a/tests/components/generic/conftest.py b/tests/components/generic/conftest.py
index 04de1aedca9..63f7a87cba0 100644
--- a/tests/components/generic/conftest.py
+++ b/tests/components/generic/conftest.py
@@ -1,9 +1,14 @@
 """Test fixtures for the generic component."""
 
 from io import BytesIO
+from unittest.mock import Mock, patch
 
 from PIL import Image
 import pytest
+import respx
+
+from homeassistant import config_entries, setup
+from homeassistant.components.generic.const import DOMAIN
 
 
 @pytest.fixture(scope="package")
@@ -29,3 +34,34 @@ def fakeimgbytes_svg():
         '<svg xmlns="http://www.w3.org/2000/svg"><circle r="50"/></svg>',
         encoding="utf-8",
     )
+
+
+@pytest.fixture
+def fakeimg_png(fakeimgbytes_png):
+    """Set up respx to respond to test url with fake image bytes."""
+    respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
+
+
+@pytest.fixture(scope="package")
+def mock_av_open():
+    """Fake container object with .streams.video[0] != None."""
+    fake = Mock()
+    fake.streams.video = ["fakevid"]
+    return patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        return_value=fake,
+    )
+
+
+@pytest.fixture
+async def user_flow(hass):
+    """Initiate a user flow."""
+
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["errors"] == {}
+
+    return result
diff --git a/tests/components/generic/test_camera.py b/tests/components/generic/test_camera.py
index b08ed58841d..5a96391e10e 100644
--- a/tests/components/generic/test_camera.py
+++ b/tests/components/generic/test_camera.py
@@ -8,47 +8,47 @@ import httpx
 import pytest
 import respx
 
-from homeassistant import config as hass_config
 from homeassistant.components.camera import async_get_mjpeg_stream
-from homeassistant.components.generic import DOMAIN
 from homeassistant.components.websocket_api.const import TYPE_RESULT
-from homeassistant.const import SERVICE_RELOAD
+from homeassistant.config_entries import SOURCE_IMPORT
 from homeassistant.setup import async_setup_component
 
-from tests.common import AsyncMock, Mock, get_fixture_path
+from tests.common import AsyncMock, Mock
 
 
 @respx.mock
-async def test_fetching_url(hass, hass_client, fakeimgbytes_png):
+async def test_fetching_url(hass, hass_client, fakeimgbytes_png, mock_av_open):
     """Test that it fetches the given url."""
     respx.get("http://example.com").respond(stream=fakeimgbytes_png)
 
-    await async_setup_component(
-        hass,
-        "camera",
-        {
-            "camera": {
-                "name": "config_test",
-                "platform": "generic",
-                "still_image_url": "http://example.com",
-                "username": "user",
-                "password": "pass",
-            }
-        },
-    )
-    await hass.async_block_till_done()
+    with mock_av_open:
+        await async_setup_component(
+            hass,
+            "camera",
+            {
+                "camera": {
+                    "name": "config_test",
+                    "platform": "generic",
+                    "still_image_url": "http://example.com",
+                    "username": "user",
+                    "password": "pass",
+                    "authentication": "basic",
+                }
+            },
+        )
+        await hass.async_block_till_done()
 
     client = await hass_client()
 
     resp = await client.get("/api/camera_proxy/camera.config_test")
 
     assert resp.status == HTTPStatus.OK
-    assert respx.calls.call_count == 1
+    assert respx.calls.call_count == 2
     body = await resp.read()
     assert body == fakeimgbytes_png
 
     resp = await client.get("/api/camera_proxy/camera.config_test")
-    assert respx.calls.call_count == 2
+    assert respx.calls.call_count == 3
 
 
 @respx.mock
@@ -110,11 +110,14 @@ async def test_fetching_url_with_verify_ssl(hass, hass_client, fakeimgbytes_png)
 @respx.mock
 async def test_limit_refetch(hass, hass_client, fakeimgbytes_png, fakeimgbytes_jpg):
     """Test that it fetches the given url."""
+    respx.get("http://example.com/0a").respond(stream=fakeimgbytes_png)
     respx.get("http://example.com/5a").respond(stream=fakeimgbytes_png)
     respx.get("http://example.com/10a").respond(stream=fakeimgbytes_png)
     respx.get("http://example.com/15a").respond(stream=fakeimgbytes_jpg)
     respx.get("http://example.com/20a").respond(status_code=HTTPStatus.NOT_FOUND)
 
+    hass.states.async_set("sensor.temp", "0")
+
     await async_setup_component(
         hass,
         "camera",
@@ -140,19 +143,19 @@ async def test_limit_refetch(hass, hass_client, fakeimgbytes_png, fakeimgbytes_j
     ):
         resp = await client.get("/api/camera_proxy/camera.config_test")
 
-    assert respx.calls.call_count == 0
-    assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
+    assert respx.calls.call_count == 2
+    assert resp.status == HTTPStatus.OK
 
     hass.states.async_set("sensor.temp", "10")
 
     resp = await client.get("/api/camera_proxy/camera.config_test")
-    assert respx.calls.call_count == 1
+    assert respx.calls.call_count == 3
     assert resp.status == HTTPStatus.OK
     body = await resp.read()
     assert body == fakeimgbytes_png
 
     resp = await client.get("/api/camera_proxy/camera.config_test")
-    assert respx.calls.call_count == 1
+    assert respx.calls.call_count == 3
     assert resp.status == HTTPStatus.OK
     body = await resp.read()
     assert body == fakeimgbytes_png
@@ -161,7 +164,7 @@ async def test_limit_refetch(hass, hass_client, fakeimgbytes_png, fakeimgbytes_j
 
     # Url change = fetch new image
     resp = await client.get("/api/camera_proxy/camera.config_test")
-    assert respx.calls.call_count == 2
+    assert respx.calls.call_count == 4
     assert resp.status == HTTPStatus.OK
     body = await resp.read()
     assert body == fakeimgbytes_jpg
@@ -169,31 +172,37 @@ async def test_limit_refetch(hass, hass_client, fakeimgbytes_png, fakeimgbytes_j
     # Cause a template render error
     hass.states.async_remove("sensor.temp")
     resp = await client.get("/api/camera_proxy/camera.config_test")
-    assert respx.calls.call_count == 2
+    assert respx.calls.call_count == 4
     assert resp.status == HTTPStatus.OK
     body = await resp.read()
     assert body == fakeimgbytes_jpg
 
 
-async def test_stream_source(hass, hass_client, hass_ws_client, fakeimgbytes_png):
+@respx.mock
+async def test_stream_source(
+    hass, hass_client, hass_ws_client, fakeimgbytes_png, mock_av_open
+):
     """Test that the stream source is rendered."""
     respx.get("http://example.com").respond(stream=fakeimgbytes_png)
+    respx.get("http://example.com/0a").respond(stream=fakeimgbytes_png)
 
-    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,
+    hass.states.async_set("sensor.temp", "0")
+    with mock_av_open:
+        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,
+                },
             },
-        },
-    )
-    assert await async_setup_component(hass, "stream", {})
-    await hass.async_block_till_done()
+        )
+        assert await async_setup_component(hass, "stream", {})
+        await hass.async_block_till_done()
 
     hass.states.async_set("sensor.temp", "5")
 
@@ -217,26 +226,30 @@ async def test_stream_source(hass, hass_client, hass_ws_client, fakeimgbytes_png
         assert msg["result"]["url"][-13:] == "playlist.m3u8"
 
 
-async def test_stream_source_error(hass, hass_client, hass_ws_client, fakeimgbytes_png):
+@respx.mock
+async def test_stream_source_error(
+    hass, hass_client, hass_ws_client, fakeimgbytes_png, mock_av_open
+):
     """Test that the stream source has an error."""
     respx.get("http://example.com").respond(stream=fakeimgbytes_png)
 
-    assert await async_setup_component(
-        hass,
-        "camera",
-        {
-            "camera": {
-                "name": "config_test",
-                "platform": "generic",
-                "still_image_url": "https://example.com",
-                # Does not exist
-                "stream_source": 'http://example.com/{{ states.sensor.temp.state + "a" }}',
-                "limit_refetch_to_url_change": True,
+    with mock_av_open:
+        assert await async_setup_component(
+            hass,
+            "camera",
+            {
+                "camera": {
+                    "name": "config_test",
+                    "platform": "generic",
+                    "still_image_url": "http://example.com",
+                    # Does not exist
+                    "stream_source": 'http://example.com/{{ states.sensor.temp.state + "a" }}',
+                    "limit_refetch_to_url_change": True,
+                },
             },
-        },
-    )
-    assert await async_setup_component(hass, "stream", {})
-    await hass.async_block_till_done()
+        )
+        assert await async_setup_component(hass, "stream", {})
+        await hass.async_block_till_done()
 
     with patch(
         "homeassistant.components.camera.Stream.endpoint_url",
@@ -261,31 +274,38 @@ async def test_stream_source_error(hass, hass_client, hass_ws_client, fakeimgbyt
         }
 
 
-async def test_setup_alternative_options(hass, hass_ws_client):
+@respx.mock
+async def test_setup_alternative_options(
+    hass, hass_ws_client, fakeimgbytes_png, mock_av_open
+):
     """Test that the stream source is setup with different config options."""
-    assert await async_setup_component(
-        hass,
-        "camera",
-        {
-            "camera": {
-                "name": "config_test",
-                "platform": "generic",
-                "still_image_url": "https://example.com",
-                "authentication": "digest",
-                "username": "user",
-                "password": "pass",
-                "stream_source": "rtsp://example.com:554/rtsp/",
-                "rtsp_transport": "udp",
+    respx.get("https://example.com").respond(stream=fakeimgbytes_png)
+
+    with mock_av_open:
+        assert await async_setup_component(
+            hass,
+            "camera",
+            {
+                "camera": {
+                    "name": "config_test",
+                    "platform": "generic",
+                    "still_image_url": "https://example.com",
+                    "authentication": "digest",
+                    "username": "user",
+                    "password": "pass",
+                    "stream_source": "rtsp://example.com:554/rtsp/",
+                    "rtsp_transport": "udp",
+                },
             },
-        },
-    )
-    await hass.async_block_till_done()
+        )
+        await hass.async_block_till_done()
     assert hass.states.get("camera.config_test")
 
 
+@respx.mock
 async def test_no_stream_source(hass, hass_client, hass_ws_client, fakeimgbytes_png):
     """Test a stream request without stream source option set."""
-    respx.get("http://example.com").respond(stream=fakeimgbytes_png)
+    respx.get("https://example.com").respond(stream=fakeimgbytes_png)
 
     assert await async_setup_component(
         hass,
@@ -326,7 +346,7 @@ async def test_no_stream_source(hass, hass_client, hass_ws_client, fakeimgbytes_
 
 @respx.mock
 async def test_camera_content_type(
-    hass, hass_client, fakeimgbytes_svg, fakeimgbytes_jpg
+    hass, hass_client, fakeimgbytes_svg, fakeimgbytes_jpg, mock_av_open
 ):
     """Test generic camera with custom content_type."""
     urlsvg = "https://upload.wikimedia.org/wikipedia/commons/0/02/SVG_logo.svg"
@@ -338,90 +358,54 @@ async def test_camera_content_type(
         "platform": "generic",
         "still_image_url": urlsvg,
         "content_type": "image/svg+xml",
+        "limit_refetch_to_url_change": False,
+        "framerate": 2,
+        "verify_ssl": True,
     }
     cam_config_jpg = {
         "name": "config_test_jpg",
         "platform": "generic",
         "still_image_url": urljpg,
         "content_type": "image/jpeg",
+        "limit_refetch_to_url_change": False,
+        "framerate": 2,
+        "verify_ssl": True,
     }
 
-    await async_setup_component(
-        hass, "camera", {"camera": [cam_config_svg, cam_config_jpg]}
-    )
-    await hass.async_block_till_done()
+    with mock_av_open:
+        result1 = await hass.config_entries.flow.async_init(
+            "generic",
+            data=cam_config_jpg,
+            context={"source": SOURCE_IMPORT, "unique_id": 12345},
+        )
+        await hass.async_block_till_done()
+    with mock_av_open:
+        result2 = await hass.config_entries.flow.async_init(
+            "generic",
+            data=cam_config_svg,
+            context={"source": SOURCE_IMPORT, "unique_id": 54321},
+        )
+        await hass.async_block_till_done()
 
+    assert result1["type"] == "create_entry"
+    assert result2["type"] == "create_entry"
     client = await hass_client()
 
     resp_1 = await client.get("/api/camera_proxy/camera.config_test_svg")
-    assert respx.calls.call_count == 1
+    assert respx.calls.call_count == 3
     assert resp_1.status == HTTPStatus.OK
     assert resp_1.content_type == "image/svg+xml"
     body = await resp_1.read()
     assert body == fakeimgbytes_svg
 
     resp_2 = await client.get("/api/camera_proxy/camera.config_test_jpg")
-    assert respx.calls.call_count == 2
+    assert respx.calls.call_count == 4
     assert resp_2.status == HTTPStatus.OK
     assert resp_2.content_type == "image/jpeg"
     body = await resp_2.read()
     assert body == fakeimgbytes_jpg
 
 
-@respx.mock
-async def test_reloading(hass, hass_client):
-    """Test we can cleanly reload."""
-    respx.get("http://example.com").respond(text="hello world")
-
-    await async_setup_component(
-        hass,
-        "camera",
-        {
-            "camera": {
-                "name": "config_test",
-                "platform": "generic",
-                "still_image_url": "http://example.com",
-                "username": "user",
-                "password": "pass",
-            }
-        },
-    )
-    await hass.async_block_till_done()
-
-    client = await hass_client()
-
-    resp = await client.get("/api/camera_proxy/camera.config_test")
-
-    assert resp.status == HTTPStatus.OK
-    assert respx.calls.call_count == 1
-    body = await resp.text()
-    assert body == "hello world"
-
-    yaml_path = get_fixture_path("configuration.yaml", "generic")
-
-    with patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path):
-        await hass.services.async_call(
-            DOMAIN,
-            SERVICE_RELOAD,
-            {},
-            blocking=True,
-        )
-        await hass.async_block_till_done()
-
-    assert len(hass.states.async_all()) == 1
-
-    resp = await client.get("/api/camera_proxy/camera.config_test")
-
-    assert resp.status == HTTPStatus.NOT_FOUND
-
-    resp = await client.get("/api/camera_proxy/camera.reload")
-
-    assert resp.status == HTTPStatus.OK
-    assert respx.calls.call_count == 2
-    body = await resp.text()
-    assert body == "hello world"
-
-
 @respx.mock
 async def test_timeout_cancelled(hass, hass_client, fakeimgbytes_png, fakeimgbytes_jpg):
     """Test that timeouts and cancellations return last image."""
@@ -448,7 +432,7 @@ async def test_timeout_cancelled(hass, hass_client, fakeimgbytes_png, fakeimgbyt
     resp = await client.get("/api/camera_proxy/camera.config_test")
 
     assert resp.status == HTTPStatus.OK
-    assert respx.calls.call_count == 1
+    assert respx.calls.call_count == 2
     assert await resp.read() == fakeimgbytes_png
 
     respx.get("http://example.com").respond(stream=fakeimgbytes_jpg)
@@ -458,7 +442,7 @@ async def test_timeout_cancelled(hass, hass_client, fakeimgbytes_png, fakeimgbyt
         side_effect=asyncio.CancelledError(),
     ):
         resp = await client.get("/api/camera_proxy/camera.config_test")
-        assert respx.calls.call_count == 1
+        assert respx.calls.call_count == 2
         assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
 
     respx.get("http://example.com").side_effect = [
@@ -466,27 +450,28 @@ async def test_timeout_cancelled(hass, hass_client, fakeimgbytes_png, fakeimgbyt
         httpx.TimeoutException,
     ]
 
-    for total_calls in range(2, 3):
+    for total_calls in range(3, 5):
         resp = await client.get("/api/camera_proxy/camera.config_test")
         assert respx.calls.call_count == total_calls
         assert resp.status == HTTPStatus.OK
         assert await resp.read() == fakeimgbytes_png
 
 
-async def test_no_still_image_url(hass, hass_client):
+async def test_no_still_image_url(hass, hass_client, mock_av_open):
     """Test that the component can grab images from stream with no still_image_url."""
-    assert await async_setup_component(
-        hass,
-        "camera",
-        {
-            "camera": {
-                "name": "config_test",
-                "platform": "generic",
-                "stream_source": "rtsp://example.com:554/rtsp/",
+    with mock_av_open:
+        assert await async_setup_component(
+            hass,
+            "camera",
+            {
+                "camera": {
+                    "name": "config_test",
+                    "platform": "generic",
+                    "stream_source": "rtsp://example.com:554/rtsp/",
+                },
             },
-        },
-    )
-    await hass.async_block_till_done()
+        )
+        await hass.async_block_till_done()
 
     client = await hass_client()
 
@@ -518,22 +503,23 @@ async def test_no_still_image_url(hass, hass_client):
         assert await resp.read() == b"stream_keyframe_image"
 
 
-async def test_frame_interval_property(hass):
+async def test_frame_interval_property(hass, mock_av_open):
     """Test that the frame interval is calculated and returned correctly."""
 
-    await async_setup_component(
-        hass,
-        "camera",
-        {
-            "camera": {
-                "name": "config_test",
-                "platform": "generic",
-                "stream_source": "rtsp://example.com:554/rtsp/",
-                "framerate": 5,
+    with mock_av_open:
+        await async_setup_component(
+            hass,
+            "camera",
+            {
+                "camera": {
+                    "name": "config_test",
+                    "platform": "generic",
+                    "stream_source": "rtsp://example.com:554/rtsp/",
+                    "framerate": 5,
+                },
             },
-        },
-    )
-    await hass.async_block_till_done()
+        )
+        await hass.async_block_till_done()
 
     request = Mock()
     with patch(
diff --git a/tests/components/generic/test_config_flow.py b/tests/components/generic/test_config_flow.py
new file mode 100644
index 00000000000..65bd8f7e386
--- /dev/null
+++ b/tests/components/generic/test_config_flow.py
@@ -0,0 +1,510 @@
+"""Test The generic (IP Camera) config flow."""
+
+import errno
+from unittest.mock import patch
+
+import av
+import httpx
+import pytest
+import respx
+
+from homeassistant import config_entries, data_entry_flow, setup
+from homeassistant.components.generic.const import (
+    CONF_CONTENT_TYPE,
+    CONF_FRAMERATE,
+    CONF_LIMIT_REFETCH_TO_URL_CHANGE,
+    CONF_RTSP_TRANSPORT,
+    CONF_STILL_IMAGE_URL,
+    CONF_STREAM_SOURCE,
+    DOMAIN,
+)
+from homeassistant.const import (
+    CONF_AUTHENTICATION,
+    CONF_NAME,
+    CONF_PASSWORD,
+    CONF_USERNAME,
+    CONF_VERIFY_SSL,
+    HTTP_BASIC_AUTHENTICATION,
+)
+
+from tests.common import MockConfigEntry
+
+TESTDATA = {
+    CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
+    CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
+    CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
+    CONF_USERNAME: "fred_flintstone",
+    CONF_PASSWORD: "bambam",
+    CONF_FRAMERATE: 5,
+    CONF_VERIFY_SSL: False,
+}
+
+TESTDATA_OPTIONS = {
+    CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
+    **TESTDATA,
+}
+
+TESTDATA_YAML = {
+    CONF_NAME: "Yaml Defined Name",
+    **TESTDATA,
+}
+
+
+@respx.mock
+async def test_form(hass, fakeimg_png, mock_av_open, user_flow):
+    """Test the form with a normal set of settings."""
+
+    with mock_av_open as mock_setup:
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result2["title"] == "127_0_0_1_testurl_1"
+    assert result2["options"] == {
+        CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
+        CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
+        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_VERIFY_SSL: False,
+    }
+
+    await hass.async_block_till_done()
+    assert len(mock_setup.mock_calls) == 1
+
+
+@respx.mock
+async def test_form_only_stillimage(hass, fakeimg_png, user_flow):
+    """Test we complete ok if the user wants still images only."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["errors"] == {}
+
+    data = TESTDATA.copy()
+    data.pop(CONF_STREAM_SOURCE)
+    result2 = await hass.config_entries.flow.async_configure(
+        user_flow["flow_id"],
+        data,
+    )
+    assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result2["title"] == "127_0_0_1_testurl_1"
+    assert result2["options"] == {
+        CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
+        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_VERIFY_SSL: False,
+    }
+
+    await hass.async_block_till_done()
+    assert respx.calls.call_count == 1
+
+
+@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."""
+    with mock_av_open as mock_setup:
+        data = TESTDATA
+        data[CONF_RTSP_TRANSPORT] = "tcp"
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"], data
+        )
+    assert "errors" not in result2, f"errors={result2['errors']}"
+    assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result2["title"] == "127_0_0_1_testurl_1"
+    assert result2["options"] == {
+        CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
+        CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
+        CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
+        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_VERIFY_SSL: False,
+    }
+
+    await hass.async_block_till_done()
+    assert len(mock_setup.mock_calls) == 1
+
+
+async def test_form_only_stream(hass, mock_av_open):
+    """Test we complete ok if the user wants stream only."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    data = TESTDATA.copy()
+    data.pop(CONF_STILL_IMAGE_URL)
+    with mock_av_open as mock_setup:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            data,
+        )
+    assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result2["title"] == "127_0_0_1_testurl_2"
+    assert result2["options"] == {
+        CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
+        CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
+        CONF_RTSP_TRANSPORT: "tcp",
+        CONF_USERNAME: "fred_flintstone",
+        CONF_PASSWORD: "bambam",
+        CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
+        CONF_CONTENT_TYPE: None,
+        CONF_FRAMERATE: 5,
+        CONF_VERIFY_SSL: False,
+    }
+
+    await hass.async_block_till_done()
+    assert len(mock_setup.mock_calls) == 1
+
+
+async def test_form_still_and_stream_not_provided(hass, user_flow):
+    """Test we show a suitable error if neither still or stream URL are provided."""
+    result2 = await hass.config_entries.flow.async_configure(
+        user_flow["flow_id"],
+        {
+            CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
+            CONF_FRAMERATE: 5,
+            CONF_VERIFY_SSL: False,
+        },
+    )
+    assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result2["errors"] == {"base": "no_still_image_or_stream_url"}
+
+
+@respx.mock
+async def test_form_image_timeout(hass, mock_av_open, user_flow):
+    """Test we handle invalid image timeout."""
+    respx.get("http://127.0.0.1/testurl/1").side_effect = [
+        httpx.TimeoutException,
+    ]
+
+    with mock_av_open:
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    await hass.async_block_till_done()
+
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"still_image_url": "unable_still_load"}
+
+
+@respx.mock
+async def test_form_stream_invalidimage(hass, mock_av_open, user_flow):
+    """Test we handle invalid image when a stream is specified."""
+    respx.get("http://127.0.0.1/testurl/1").respond(stream=b"invalid")
+    with mock_av_open:
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    await hass.async_block_till_done()
+
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"still_image_url": "invalid_still_image"}
+
+
+@respx.mock
+async def test_form_stream_invalidimage2(hass, mock_av_open, user_flow):
+    """Test we handle invalid image when a stream is specified."""
+    respx.get("http://127.0.0.1/testurl/1").respond(content=None)
+    with mock_av_open:
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    await hass.async_block_till_done()
+
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"still_image_url": "unable_still_load"}
+
+
+@respx.mock
+async def test_form_stream_invalidimage3(hass, mock_av_open, user_flow):
+    """Test we handle invalid image when a stream is specified."""
+    respx.get("http://127.0.0.1/testurl/1").respond(content=bytes([0xFF]))
+    with mock_av_open:
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    await hass.async_block_till_done()
+
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"still_image_url": "invalid_still_image"}
+
+
+@respx.mock
+async def test_form_stream_file_not_found(hass, fakeimg_png, user_flow):
+    """Test we handle file not found."""
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        side_effect=av.error.FileNotFoundError(0, 0),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"stream_source": "stream_file_not_found"}
+
+
+@respx.mock
+async def test_form_stream_http_not_found(hass, fakeimg_png, user_flow):
+    """Test we handle invalid auth."""
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        side_effect=av.error.HTTPNotFoundError(0, 0),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"stream_source": "stream_http_not_found"}
+
+
+@respx.mock
+async def test_form_stream_timeout(hass, fakeimg_png, user_flow):
+    """Test we handle invalid auth."""
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        side_effect=av.error.TimeoutError(0, 0),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"stream_source": "timeout"}
+
+
+@respx.mock
+async def test_form_stream_unauthorised(hass, fakeimg_png, user_flow):
+    """Test we handle invalid auth."""
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        side_effect=av.error.HTTPUnauthorizedError(0, 0),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"stream_source": "stream_unauthorised"}
+
+
+@respx.mock
+async def test_form_stream_novideo(hass, fakeimg_png, user_flow):
+    """Test we handle invalid stream."""
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open", side_effect=KeyError()
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"stream_source": "stream_no_video"}
+
+
+@respx.mock
+async def test_form_stream_permission_error(hass, fakeimgbytes_png, user_flow):
+    """Test we handle permission error."""
+    respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        side_effect=PermissionError(),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"stream_source": "stream_not_permitted"}
+
+
+@respx.mock
+async def test_form_no_route_to_host(hass, fakeimg_png, user_flow):
+    """Test we handle no route to host."""
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        side_effect=OSError(errno.EHOSTUNREACH, "No route to host"),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"stream_source": "stream_no_route_to_host"}
+
+
+@respx.mock
+async def test_form_stream_io_error(hass, fakeimg_png, user_flow):
+    """Test we handle no io error when setting up stream."""
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        side_effect=OSError(errno.EIO, "Input/output error"),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"stream_source": "stream_io_error"}
+
+
+@respx.mock
+async def test_form_oserror(hass, fakeimg_png, user_flow):
+    """Test we handle OS error when setting up stream."""
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        side_effect=OSError("Some other OSError"),
+    ), pytest.raises(OSError):
+        await hass.config_entries.flow.async_configure(
+            user_flow["flow_id"],
+            TESTDATA,
+        )
+
+
+@respx.mock
+async def test_options_template_error(hass, fakeimgbytes_png, mock_av_open):
+    """Test the options flow with a template error."""
+    respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
+    respx.get("http://127.0.0.1/testurl/2").respond(stream=fakeimgbytes_png)
+    await setup.async_setup_component(hass, "persistent_notification", {})
+
+    mock_entry = MockConfigEntry(
+        title="Test Camera",
+        domain=DOMAIN,
+        data={},
+        options=TESTDATA,
+    )
+
+    with mock_av_open:
+        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"] == data_entry_flow.RESULT_TYPE_FORM
+        assert result["step_id"] == "init"
+
+        # try updating the still image url
+        data = TESTDATA.copy()
+        data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/2"
+        result2 = await hass.config_entries.options.async_configure(
+            result["flow_id"],
+            user_input=data,
+        )
+        assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+
+        result3 = await hass.config_entries.options.async_init(mock_entry.entry_id)
+        assert result3["type"] == data_entry_flow.RESULT_TYPE_FORM
+        assert result3["step_id"] == "init"
+
+        # verify that an invalid template reports the correct UI error.
+        data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/{{1/0}}"
+        result4 = await hass.config_entries.options.async_configure(
+            result3["flow_id"],
+            user_input=data,
+        )
+        assert result4.get("type") == data_entry_flow.RESULT_TYPE_FORM
+        assert result4["errors"] == {"still_image_url": "template_error"}
+
+
+# These below can be deleted after deprecation period is finished.
+@respx.mock
+async def test_import(hass, fakeimg_png, mock_av_open):
+    """Test configuration.yaml import used during migration."""
+    with mock_av_open:
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=TESTDATA_YAML
+        )
+        # duplicate import should be aborted
+        result2 = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=TESTDATA_YAML
+        )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result["title"] == "Yaml Defined Name"
+    await hass.async_block_till_done()
+    # Any name defined in yaml should end up as the entity id.
+    assert hass.states.get("camera.yaml_defined_name")
+    assert result2["type"] == data_entry_flow.RESULT_TYPE_ABORT
+
+
+@respx.mock
+async def test_import_invalid_still_image(hass, mock_av_open):
+    """Test configuration.yaml import used during migration."""
+    respx.get("http://127.0.0.1/testurl/1").respond(stream=b"invalid")
+    with mock_av_open:
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=TESTDATA_YAML
+        )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "unknown"
+
+
+@respx.mock
+async def test_import_other_error(hass, fakeimgbytes_png):
+    """Test that non-specific import errors are raised."""
+    respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_png)
+    with patch(
+        "homeassistant.components.generic.config_flow.av.open",
+        side_effect=OSError("other error"),
+    ), pytest.raises(OSError):
+        await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=TESTDATA_YAML
+        )
+
+
+# These above can be deleted after deprecation period is finished.
+
+
+async def test_unload_entry(hass, fakeimg_png, mock_av_open):
+    """Test unloading the generic IP Camera entry."""
+    mock_entry = MockConfigEntry(domain=DOMAIN, options=TESTDATA)
+    mock_entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    assert mock_entry.state is config_entries.ConfigEntryState.LOADED
+
+    await hass.config_entries.async_unload(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    assert mock_entry.state is config_entries.ConfigEntryState.NOT_LOADED
+
+
+async def test_reload_on_title_change(hass) -> None:
+    """Test the integration gets reloaded when the title is updated."""
+
+    test_data = TESTDATA_OPTIONS
+    test_data[CONF_CONTENT_TYPE] = "image/png"
+    mock_entry = MockConfigEntry(
+        domain=DOMAIN, unique_id="54321", options=test_data, title="My Title"
+    )
+    mock_entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    assert mock_entry.state is config_entries.ConfigEntryState.LOADED
+    assert hass.states.get("camera.my_title").attributes["friendly_name"] == "My Title"
+
+    hass.config_entries.async_update_entry(mock_entry, title="New Title")
+    assert mock_entry.title == "New Title"
+    await hass.async_block_till_done()
+
+    assert hass.states.get("camera.my_title").attributes["friendly_name"] == "New Title"