Install and start Mosquitto MQTT broker add on from MQTT config flow (#124106)

* Opt in to install Mosquitto broker add-on in MQTT config flow

* rephrase

* Tests with supervisor and running add-on

* Complete tests for success flows

* Also set up entry in success flow

* Use realistic names for addon and broker

* Finetuning and fail test cases

* Spelling

* Improve translation strings

* Update addon docstr

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Raise AddonError if add-on does not start

* Only show the option to use the add-on

* Simplify flow, rework and cleanup

* Revert unrelated cleanup, process suggestion

* Move ADDON_SLUG const to addon module

* Move fixture to component level

* Move back supervisor fixture

* Move addon_setup_time_fixture and superfixe to config flow model tests

* Refactor hassio fixture

* Rename helpers as they are no fixtures, remove fixture from their names

---------

Co-authored-by: Erik Montnemery <erik@montnemery.com>

homeassistant/components/mqtt/addon.py

@@ -0,0 +1,22 @@
+"""Provide MQTT add-on management.
+
+Currently only supports the official mosquitto add-on.
+"""
+
+from __future__ import annotations
+
+from homeassistant.components.hassio import AddonManager
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.singleton import singleton
+
+from .const import DOMAIN, LOGGER
+
+ADDON_SLUG = "core_mosquitto"
+DATA_ADDON_MANAGER = f"{DOMAIN}_addon_manager"
+
+
+@singleton(DATA_ADDON_MANAGER)
+@callback
+def get_addon_manager(hass: HomeAssistant) -> AddonManager:
+    """Get the add-on manager."""
+    return AddonManager(hass, LOGGER, "Mosquitto Mqtt Broker", ADDON_SLUG)

homeassistant/components/mqtt/config_flow.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import asyncio
 from collections import OrderedDict
 from collections.abc import Callable, Mapping
+import logging
 import queue
 from ssl import PROTOCOL_TLS_CLIENT, SSLContext, SSLError
 from types import MappingProxyType
@@ -14,7 +16,12 @@ from cryptography.x509 import load_pem_x509_certificate
 import voluptuous as vol
 
 from homeassistant.components.file_upload import process_uploaded_file
-from homeassistant.components.hassio import HassioServiceInfo
+from homeassistant.components.hassio import HassioServiceInfo, is_hassio
+from homeassistant.components.hassio.addon_manager import (
+    AddonError,
+    AddonManager,
+    AddonState,
+)
 from homeassistant.config_entries import (
     ConfigEntry,
     ConfigFlow,
@@ -32,6 +39,7 @@ from homeassistant.const import (
     CONF_USERNAME,
 )
 from homeassistant.core import callback
+from homeassistant.data_entry_flow import AbortFlow
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.json import json_dumps
 from homeassistant.helpers.selector import (
@@ -51,6 +59,7 @@ from homeassistant.helpers.selector import (
 )
 from homeassistant.util.json import JSON_DECODE_EXCEPTIONS, json_loads
 
+from .addon import get_addon_manager
 from .client import MqttClientSetup
 from .const import (
     ATTR_PAYLOAD,
@@ -91,6 +100,11 @@ from .util import (
     valid_publish_topic,
 )
 
+_LOGGER = logging.getLogger(__name__)
+
+ADDON_SETUP_TIMEOUT = 5
+ADDON_SETUP_TIMEOUT_ROUNDS = 5
+
 MQTT_TIMEOUT = 5
 
 ADVANCED_OPTIONS = "advanced_options"
@@ -197,6 +211,12 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
 
     entry: ConfigEntry | None
     _hassio_discovery: dict[str, Any] | None = None
+    _addon_manager: AddonManager
+
+    def __init__(self) -> None:
+        """Set up flow instance."""
+        self.install_task: asyncio.Task | None = None
+        self.start_task: asyncio.Task | None = None
 
     @staticmethod
     @callback
@@ -206,6 +226,118 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
         """Get the options flow for this handler."""
         return MQTTOptionsFlowHandler(config_entry)
 
+    async def _async_install_addon(self) -> None:
+        """Install the Mosquitto Mqtt broker add-on."""
+        addon_manager: AddonManager = get_addon_manager(self.hass)
+        await addon_manager.async_schedule_install_addon()
+
+    async def async_step_install_failed(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Add-on installation failed."""
+        return self.async_abort(
+            reason="addon_install_failed",
+            description_placeholders={"addon": self._addon_manager.addon_name},
+        )
+
+    async def async_step_install_addon(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Install Mosquitto Broker add-on."""
+        if self.install_task is None:
+            self.install_task = self.hass.async_create_task(self._async_install_addon())
+
+        if not self.install_task.done():
+            return self.async_show_progress(
+                step_id="install_addon",
+                progress_action="install_addon",
+                progress_task=self.install_task,
+            )
+
+        try:
+            await self.install_task
+        except AddonError as err:
+            _LOGGER.error(err)
+            return self.async_show_progress_done(next_step_id="install_failed")
+        finally:
+            self.install_task = None
+
+        return self.async_show_progress_done(next_step_id="start_addon")
+
+    async def async_step_start_failed(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Add-on start failed."""
+        return self.async_abort(
+            reason="addon_start_failed",
+            description_placeholders={"addon": self._addon_manager.addon_name},
+        )
+
+    async def async_step_start_addon(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Start Mosquitto Broker add-on."""
+        if not self.start_task:
+            self.start_task = self.hass.async_create_task(self._async_start_addon())
+        if not self.start_task.done():
+            return self.async_show_progress(
+                step_id="start_addon",
+                progress_action="start_addon",
+                progress_task=self.start_task,
+            )
+        try:
+            await self.start_task
+        except AddonError as err:
+            _LOGGER.error(err)
+            return self.async_show_progress_done(next_step_id="start_failed")
+        finally:
+            self.start_task = None
+
+        return self.async_show_progress_done(next_step_id="setup_entry_from_discovery")
+
+    async def _async_get_config_and_try(self) -> dict[str, Any] | None:
+        """Get the MQTT add-on discovery info and try the connection."""
+        if self._hassio_discovery is not None:
+            return self._hassio_discovery
+        addon_manager: AddonManager = get_addon_manager(self.hass)
+        try:
+            addon_discovery_config = (
+                await addon_manager.async_get_addon_discovery_info()
+            )
+            config: dict[str, Any] = {
+                CONF_BROKER: addon_discovery_config[CONF_HOST],
+                CONF_PORT: addon_discovery_config[CONF_PORT],
+                CONF_USERNAME: addon_discovery_config.get(CONF_USERNAME),
+                CONF_PASSWORD: addon_discovery_config.get(CONF_PASSWORD),
+                CONF_DISCOVERY: DEFAULT_DISCOVERY,
+            }
+        except AddonError:
+            # We do not have discovery information yet
+            return None
+        if await self.hass.async_add_executor_job(
+            try_connection,
+            config,
+        ):
+            self._hassio_discovery = config
+            return config
+        return None
+
+    async def _async_start_addon(self) -> None:
+        """Start the Mosquitto Broker add-on."""
+        addon_manager: AddonManager = get_addon_manager(self.hass)
+        await addon_manager.async_schedule_start_addon()
+
+        # Sleep some seconds to let the add-on start properly before connecting.
+        for _ in range(ADDON_SETUP_TIMEOUT_ROUNDS):
+            await asyncio.sleep(ADDON_SETUP_TIMEOUT)
+            # Finish setup using discovery info to test the connection
+            if await self._async_get_config_and_try():
+                break
+        else:
+            raise AddonError(
+                f"Failed to correctly start {addon_manager.addon_name} add-on"
+            )
+
     async def async_step_user(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
@@ -213,8 +345,57 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
         if self._async_current_entries():
             return self.async_abort(reason="single_instance_allowed")
 
+        if is_hassio(self.hass):
+            # Offer to set up broker add-on if supervisor is available
+            self._addon_manager = get_addon_manager(self.hass)
+            return self.async_show_menu(
+                step_id="user",
+                menu_options=["addon", "broker"],
+                description_placeholders={"addon": self._addon_manager.addon_name},
+            )
+
+        # Start up a flow for manual setup
         return await self.async_step_broker()
 
+    async def async_step_setup_entry_from_discovery(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Set up mqtt entry from discovery info."""
+        if (config := await self._async_get_config_and_try()) is not None:
+            return self.async_create_entry(
+                title=self._addon_manager.addon_name,
+                data=config,
+            )
+
+        raise AbortFlow(
+            "addon_connection_failed",
+            description_placeholders={"addon": self._addon_manager.addon_name},
+        )
+
+    async def async_step_addon(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Install and start MQTT broker add-on."""
+        addon_manager = self._addon_manager
+
+        try:
+            addon_info = await addon_manager.async_get_addon_info()
+        except AddonError as err:
+            raise AbortFlow(
+                "addon_info_failed",
+                description_placeholders={"addon": self._addon_manager.addon_name},
+            ) from err
+
+        if addon_info.state == AddonState.RUNNING:
+            # Finish setup using discovery info
+            return await self.async_step_setup_entry_from_discovery()
+
+        if addon_info.state == AddonState.NOT_RUNNING:
+            return await self.async_step_start_addon()
+
+        # Install the add-on and start it
+        return await self.async_step_install_addon()
+
     async def async_step_reauth(
         self, entry_data: Mapping[str, Any]
     ) -> ConfigFlowResult:
@@ -293,7 +474,7 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
     async def async_step_hassio(
         self, discovery_info: HassioServiceInfo
     ) -> ConfigFlowResult:
-        """Receive a Hass.io discovery."""
+        """Receive a Hass.io discovery or process setup after addon install."""
         await self._async_handle_discovery_without_unique_id()
 
         self._hassio_discovery = discovery_info.config

homeassistant/components/mqtt/const.py

@@ -1,5 +1,7 @@
 """Constants used by multiple MQTT modules."""
 
+import logging
+
 import jinja2
 
 from homeassistant.const import CONF_PAYLOAD, Platform
@@ -148,6 +150,7 @@ DEFAULT_WILL = {
 }
 
 DOMAIN = "mqtt"
+LOGGER = logging.getLogger(__package__)
 
 MQTT_CONNECTION_STATE = "mqtt_connection_state"
 

homeassistant/components/mqtt/strings.json

@@ -23,6 +23,13 @@
   },
   "config": {
     "step": {
+      "user": {
+        "description": "Please choose how you want to connect to the MQTT broker:",
+        "menu_options": {
+          "addon": "Use the official {addon} add-on.",
+          "broker": "Manually enter the MQTT broker connection details"
+        }
+      },
       "broker": {
         "description": "Please enter the connection information of your MQTT broker.",
         "data": {
@@ -63,6 +70,12 @@
           "ws_path": "The WebSocket path to be used for the connection to your MQTT broker."
         }
       },
+      "install_addon": {
+        "title": "Installing add-on"
+      },
+      "start_addon": {
+        "title": "Starting add-on"
+      },
       "hassio_confirm": {
         "title": "MQTT Broker via Home Assistant add-on",
         "description": "Do you want to configure Home Assistant to connect to the MQTT broker provided by the add-on {addon}?",
@@ -87,6 +100,10 @@
       }
     },
     "abort": {
+      "addon_info_failed": "Failed get info for the {addon} add-on.",
+      "addon_install_failed": "Failed to install the {addon} add-on.",
+      "addon_start_failed": "Failed to start the {addon} add-on.",
+      "addon_connection_failed": "Failed to connect to the {addon} add-on. Check the add-on status and try again later.",
       "already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
       "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
       "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]"

tests/components/conftest.py

@@ -6,7 +6,7 @@ from collections.abc import Callable, Generator
 from importlib.util import find_spec
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -180,3 +180,92 @@ def mock_legacy_device_tracker_setup() -> Callable[[HomeAssistant, MockScanner],
     from .device_tracker.common import mock_legacy_device_tracker_setup
 
     return mock_legacy_device_tracker_setup
+
+
+@pytest.fixture(name="discovery_info")
+def discovery_info_fixture() -> Any:
+    """Return the discovery info from the supervisor."""
+    # pylint: disable-next=import-outside-toplevel
+    from .hassio.common import mock_discovery_info
+
+    return mock_discovery_info()
+
+
+@pytest.fixture(name="get_addon_discovery_info")
+def get_addon_discovery_info_fixture(discovery_info: Any) -> Generator[AsyncMock]:
+    """Mock get add-on discovery info."""
+    # pylint: disable-next=import-outside-toplevel
+    from .hassio.common import mock_get_addon_discovery_info
+
+    yield from mock_get_addon_discovery_info(discovery_info)
+
+
+@pytest.fixture(name="addon_store_info")
+def addon_store_info_fixture() -> Generator[AsyncMock]:
+    """Mock Supervisor add-on store info."""
+    # pylint: disable-next=import-outside-toplevel
+    from .hassio.common import mock_addon_store_info
+
+    yield from mock_addon_store_info()
+
+
+@pytest.fixture(name="addon_info")
+def addon_info_fixture() -> Generator[AsyncMock]:
+    """Mock Supervisor add-on info."""
+    # pylint: disable-next=import-outside-toplevel
+    from .hassio.common import mock_addon_info
+
+    yield from mock_addon_info()
+
+
+@pytest.fixture(name="addon_not_installed")
+def addon_not_installed_fixture(
+    addon_store_info: AsyncMock, addon_info: AsyncMock
+) -> AsyncMock:
+    """Mock add-on not installed."""
+    # pylint: disable-next=import-outside-toplevel
+    from .hassio.common import mock_addon_not_installed
+
+    return mock_addon_not_installed(addon_store_info, addon_info)
+
+
+@pytest.fixture(name="addon_installed")
+def addon_installed_fixture(
+    addon_store_info: AsyncMock, addon_info: AsyncMock
+) -> AsyncMock:
+    """Mock add-on already installed but not running."""
+    # pylint: disable-next=import-outside-toplevel
+    from .hassio.common import mock_addon_installed
+
+    return mock_addon_installed(addon_store_info, addon_info)
+
+
+@pytest.fixture(name="addon_running")
+def addon_running_fixture(
+    addon_store_info: AsyncMock, addon_info: AsyncMock
+) -> AsyncMock:
+    """Mock add-on already running."""
+    # pylint: disable-next=import-outside-toplevel
+    from .hassio.common import mock_addon_running
+
+    return mock_addon_running(addon_store_info, addon_info)
+
+
+@pytest.fixture(name="install_addon")
+def install_addon_fixture(
+    addon_store_info: AsyncMock, addon_info: AsyncMock
+) -> Generator[AsyncMock]:
+    """Mock install add-on."""
+    # pylint: disable-next=import-outside-toplevel
+    from .hassio.common import mock_install_addon
+
+    yield from mock_install_addon(addon_store_info, addon_info)
+
+
+@pytest.fixture(name="start_addon")
+def start_addon_fixture() -> Generator[AsyncMock]:
+    """Mock start add-on."""
+    # pylint: disable-next=import-outside-toplevel
+    from .hassio.common import mock_start_addon
+
+    yield from mock_start_addon()

tests/components/hassio/common.py

@@ -0,0 +1,125 @@
+"""Provide common test tools for hassio."""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from typing import Any
+from unittest.mock import DEFAULT, AsyncMock, patch
+
+from homeassistant.core import HomeAssistant
+
+
+def mock_discovery_info() -> Any:
+    """Return the discovery info from the supervisor."""
+    return DEFAULT
+
+
+def mock_get_addon_discovery_info(discovery_info: Any) -> Generator[AsyncMock]:
+    """Mock get add-on discovery info."""
+    with patch(
+        "homeassistant.components.hassio.addon_manager.async_get_addon_discovery_info",
+        return_value=discovery_info,
+    ) as get_addon_discovery_info:
+        yield get_addon_discovery_info
+
+
+def mock_addon_store_info() -> Generator[AsyncMock]:
+    """Mock Supervisor add-on store info."""
+    with patch(
+        "homeassistant.components.hassio.addon_manager.async_get_addon_store_info"
+    ) as addon_store_info:
+        addon_store_info.return_value = {
+            "available": False,
+            "installed": None,
+            "state": None,
+            "version": "1.0.0",
+        }
+        yield addon_store_info
+
+
+def mock_addon_info() -> Generator[AsyncMock]:
+    """Mock Supervisor add-on info."""
+    with patch(
+        "homeassistant.components.hassio.addon_manager.async_get_addon_info",
+    ) as addon_info:
+        addon_info.return_value = {
+            "available": False,
+            "hostname": None,
+            "options": {},
+            "state": None,
+            "update_available": False,
+            "version": None,
+        }
+        yield addon_info
+
+
+def mock_addon_not_installed(
+    addon_store_info: AsyncMock, addon_info: AsyncMock
+) -> AsyncMock:
+    """Mock add-on not installed."""
+    addon_store_info.return_value["available"] = True
+    return addon_info
+
+
+def mock_addon_installed(
+    addon_store_info: AsyncMock, addon_info: AsyncMock
+) -> AsyncMock:
+    """Mock add-on already installed but not running."""
+    addon_store_info.return_value = {
+        "available": True,
+        "installed": "1.0.0",
+        "state": "stopped",
+        "version": "1.0.0",
+    }
+    addon_info.return_value["available"] = True
+    addon_info.return_value["hostname"] = "core-matter-server"
+    addon_info.return_value["state"] = "stopped"
+    addon_info.return_value["version"] = "1.0.0"
+    return addon_info
+
+
+def mock_addon_running(addon_store_info: AsyncMock, addon_info: AsyncMock) -> AsyncMock:
+    """Mock add-on already running."""
+    addon_store_info.return_value = {
+        "available": True,
+        "installed": "1.0.0",
+        "state": "started",
+        "version": "1.0.0",
+    }
+    addon_info.return_value["available"] = True
+    addon_info.return_value["hostname"] = "core-mosquitto"
+    addon_info.return_value["state"] = "started"
+    addon_info.return_value["version"] = "1.0.0"
+    return addon_info
+
+
+def mock_install_addon(
+    addon_store_info: AsyncMock, addon_info: AsyncMock
+) -> Generator[AsyncMock]:
+    """Mock install add-on."""
+
+    async def install_addon_side_effect(hass: HomeAssistant, slug: str) -> None:
+        """Mock install add-on."""
+        addon_store_info.return_value = {
+            "available": True,
+            "installed": "1.0.0",
+            "state": "stopped",
+            "version": "1.0.0",
+        }
+        addon_info.return_value["available"] = True
+        addon_info.return_value["state"] = "stopped"
+        addon_info.return_value["version"] = "1.0.0"
+
+    with patch(
+        "homeassistant.components.hassio.addon_manager.async_install_addon"
+    ) as install_addon:
+        install_addon.side_effect = install_addon_side_effect
+        yield install_addon
+
+
+def mock_start_addon() -> Generator[AsyncMock]:
+    """Mock start add-on."""
+    with patch(
+        "homeassistant.components.hassio.addon_manager.async_start_addon"
+    ) as start_addon:
+        yield start_addon

tests/components/mqtt/test_config_flow.py

@@ -14,6 +14,8 @@ import voluptuous as vol
 from homeassistant import config_entries
 from homeassistant.components import mqtt
 from homeassistant.components.hassio import HassioServiceInfo
+from homeassistant.components.hassio.addon_manager import AddonError
+from homeassistant.components.hassio.handler import HassioAPIError
 from homeassistant.components.mqtt.config_flow import PWD_NOT_CHANGED
 from homeassistant.const import (
     CONF_CLIENT_ID,
@@ -28,6 +30,15 @@ from homeassistant.data_entry_flow import FlowResultType
 from tests.common import MockConfigEntry
 from tests.typing import MqttMockHAClientGenerator, MqttMockPahoClient
 
+ADD_ON_DISCOVERY_INFO = {
+    "addon": "Mosquitto Mqtt Broker",
+    "host": "core-mosquitto",
+    "port": 1883,
+    "username": "mock-user",
+    "password": "mock-pass",
+    "protocol": "3.1.1",
+    "ssl": False,
+}
 MOCK_CLIENT_CERT = b"## mock client certificate file ##"
 MOCK_CLIENT_KEY = b"## mock key file ##"
 
@@ -186,6 +197,29 @@ def mock_process_uploaded_file(
         yield mock_upload
 
 
+@pytest.fixture(name="supervisor")
+def supervisor_fixture() -> Generator[MagicMock]:
+    """Mock Supervisor."""
+    with patch(
+        "homeassistant.components.mqtt.config_flow.is_hassio", return_value=True
+    ) as is_hassio:
+        yield is_hassio
+
+
+@pytest.fixture(name="addon_setup_time", autouse=True)
+def addon_setup_time_fixture() -> Generator[int]:
+    """Mock add-on setup sleep time."""
+    with patch(
+        "homeassistant.components.mqtt.config_flow.ADDON_SETUP_TIMEOUT", new=0
+    ) as addon_setup_time:
+        yield addon_setup_time
+
+
+@pytest.fixture(autouse=True)
+def mock_get_addon_discovery_info(get_addon_discovery_info: AsyncMock) -> None:
+    """Mock get add-on discovery info."""
+
+
 @pytest.mark.usefixtures("mqtt_client_mock")
 async def test_user_connection_works(
     hass: HomeAssistant,
@@ -216,6 +250,47 @@ async def test_user_connection_works(
     assert len(mock_finish_setup.mock_calls) == 1
 
 
+@pytest.mark.usefixtures("mqtt_client_mock", "supervisor")
+async def test_user_connection_works_with_supervisor(
+    hass: HomeAssistant,
+    mock_try_connection: MagicMock,
+    mock_finish_setup: MagicMock,
+) -> None:
+    """Test we can finish a config flow with a supervised install."""
+    mock_try_connection.return_value = True
+
+    result = await hass.config_entries.flow.async_init(
+        "mqtt", context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["menu_options"] == ["addon", "broker"]
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "broker"},
+    )
+
+    # Assert a manual setup flow
+    assert result["type"] is FlowResultType.FORM
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], {"broker": "127.0.0.1"}
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["result"].data == {
+        "broker": "127.0.0.1",
+        "port": 1883,
+        "discovery": True,
+    }
+    # Check we tried the connection
+    assert len(mock_try_connection.mock_calls) == 1
+    # Check config entry got setup
+    assert len(mock_finish_setup.mock_calls) == 1
+    await hass.async_block_till_done(wait_background_tasks=True)
+
+
 @pytest.mark.usefixtures("mqtt_client_mock")
 async def test_user_v5_connection_works(
     hass: HomeAssistant,
@@ -382,16 +457,8 @@ async def test_hassio_confirm(
     result = await hass.config_entries.flow.async_init(
         "mqtt",
         data=HassioServiceInfo(
-            config={
+            config=ADD_ON_DISCOVERY_INFO.copy(),
-                "addon": "Mock Addon",
+            name="Mosquitto Mqtt Broker",
-                "host": "mock-broker",
-                "port": 1883,
-                "username": "mock-user",
-                "password": "mock-pass",
-                "protocol": "3.1.1",  # Set by the addon's discovery, ignored by HA
-                "ssl": False,  # Set by the addon's discovery, ignored by HA
-            },
-            name="Mock Addon",
             slug="mosquitto",
             uuid="1234",
         ),
@@ -399,7 +466,7 @@ async def test_hassio_confirm(
     )
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "hassio_confirm"
-    assert result["description_placeholders"] == {"addon": "Mock Addon"}
+    assert result["description_placeholders"] == {"addon": "Mosquitto Mqtt Broker"}
 
     mock_try_connection_success.reset_mock()
     result = await hass.config_entries.flow.async_configure(
@@ -408,7 +475,7 @@ async def test_hassio_confirm(
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
     assert result["result"].data == {
-        "broker": "mock-broker",
+        "broker": "core-mosquitto",
         "port": 1883,
         "username": "mock-user",
         "password": "mock-pass",
@@ -426,14 +493,12 @@ async def test_hassio_cannot_connect(
     mock_finish_setup: MagicMock,
 ) -> None:
     """Test a config flow is aborted when a connection was not successful."""
-    mock_try_connection.return_value = True
-
     result = await hass.config_entries.flow.async_init(
         "mqtt",
         data=HassioServiceInfo(
             config={
                 "addon": "Mock Addon",
-                "host": "mock-broker",
+                "host": "core-mosquitto",
                 "port": 1883,
                 "username": "mock-user",
                 "password": "mock-pass",
@@ -463,6 +528,362 @@ async def test_hassio_cannot_connect(
     assert len(mock_finish_setup.mock_calls) == 0
 
 
+@pytest.mark.usefixtures(
+    "mqtt_client_mock", "supervisor", "addon_info", "addon_running"
+)
+@pytest.mark.parametrize("discovery_info", [{"config": ADD_ON_DISCOVERY_INFO.copy()}])
+async def test_addon_flow_with_supervisor_addon_running(
+    hass: HomeAssistant,
+    mock_try_connection_success: MagicMock,
+    mock_finish_setup: MagicMock,
+) -> None:
+    """Test we perform an auto config flow with a supervised install.
+
+    Case: The Mosquitto add-on is already installed, and running.
+    """
+    # show menu
+    result = await hass.config_entries.flow.async_init(
+        "mqtt", context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["menu_options"] == ["addon", "broker"]
+    assert result["step_id"] == "user"
+
+    # select install via add-on
+    mock_try_connection_success.reset_mock()
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "addon"},
+    )
+    await hass.async_block_till_done(wait_background_tasks=True)
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["result"].data == {
+        "broker": "core-mosquitto",
+        "port": 1883,
+        "username": "mock-user",
+        "password": "mock-pass",
+        "discovery": True,
+    }
+    # Check we tried the connection
+    assert len(mock_try_connection_success.mock_calls)
+    # Check config entry got setup
+    assert len(mock_finish_setup.mock_calls) == 1
+
+
+@pytest.mark.usefixtures(
+    "mqtt_client_mock", "supervisor", "addon_info", "addon_installed", "start_addon"
+)
+@pytest.mark.parametrize("discovery_info", [{"config": ADD_ON_DISCOVERY_INFO.copy()}])
+async def test_addon_flow_with_supervisor_addon_installed(
+    hass: HomeAssistant,
+    mock_try_connection_success: MagicMock,
+    mock_finish_setup: MagicMock,
+) -> None:
+    """Test we perform an auto config flow with a supervised install.
+
+    Case: The Mosquitto add-on is installed, but not running.
+    """
+    # show menu
+    result = await hass.config_entries.flow.async_init(
+        "mqtt", context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["menu_options"] == ["addon", "broker"]
+    assert result["step_id"] == "user"
+
+    # select install via add-on
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "addon"},
+    )
+
+    # add-on installed but not started, so we wait for start-up
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["progress_action"] == "start_addon"
+    assert result["step_id"] == "start_addon"
+    await hass.async_block_till_done()
+    await hass.async_block_till_done(wait_background_tasks=True)
+    mock_try_connection_success.reset_mock()
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "start_addon"},
+    )
+
+    # add-on is running, so entry can be installed
+    await hass.async_block_till_done(wait_background_tasks=True)
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["result"].data == {
+        "broker": "core-mosquitto",
+        "port": 1883,
+        "username": "mock-user",
+        "password": "mock-pass",
+        "discovery": True,
+    }
+    # Check we tried the connection
+    assert len(mock_try_connection_success.mock_calls)
+    # Check config entry got setup
+    assert len(mock_finish_setup.mock_calls) == 1
+
+
+@pytest.mark.usefixtures(
+    "mqtt_client_mock", "supervisor", "addon_info", "addon_running"
+)
+@pytest.mark.parametrize("discovery_info", [{"config": ADD_ON_DISCOVERY_INFO.copy()}])
+async def test_addon_flow_with_supervisor_addon_running_connection_fails(
+    hass: HomeAssistant,
+    mock_try_connection: MagicMock,
+) -> None:
+    """Test we perform an auto config flow with a supervised install.
+
+    Case: The Mosquitto add-on is already installed, and running.
+    """
+    # show menu
+    result = await hass.config_entries.flow.async_init(
+        "mqtt", context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["menu_options"] == ["addon", "broker"]
+    assert result["step_id"] == "user"
+
+    # select install via add-on but the connection fails and the flow will be aborted.
+    mock_try_connection.return_value = False
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "addon"},
+    )
+    await hass.async_block_till_done(wait_background_tasks=True)
+    assert result["type"] is FlowResultType.ABORT
+
+
+@pytest.mark.usefixtures(
+    "mqtt_client_mock",
+    "supervisor",
+    "addon_info",
+    "addon_installed",
+)
+async def test_addon_not_running_api_error(
+    hass: HomeAssistant,
+    start_addon: AsyncMock,
+) -> None:
+    """Test we perform an auto config flow with a supervised install.
+
+    Case: The Mosquitto add-on start fails on a API error.
+    """
+    start_addon.side_effect = HassioAPIError()
+
+    result = await hass.config_entries.flow.async_init(
+        "mqtt", context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["menu_options"] == ["addon", "broker"]
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "addon"},
+    )
+    # add-on not installed, so we wait for install
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["progress_action"] == "start_addon"
+    assert result["step_id"] == "start_addon"
+    await hass.async_block_till_done()
+    await hass.async_block_till_done(wait_background_tasks=True)
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "install_addon"},
+    )
+
+    # add-on start-up failed
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "addon_start_failed"
+
+
+@pytest.mark.usefixtures(
+    "mqtt_client_mock",
+    "supervisor",
+    "start_addon",
+    "addon_installed",
+)
+async def test_addon_discovery_info_error(
+    hass: HomeAssistant,
+    addon_info: AsyncMock,
+    get_addon_discovery_info: AsyncMock,
+) -> None:
+    """Test we perform an auto config flow with a supervised install.
+
+    Case: The Mosquitto add-on start on a discovery error.
+    """
+    get_addon_discovery_info.side_effect = AddonError
+
+    result = await hass.config_entries.flow.async_init(
+        "mqtt", context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["menu_options"] == ["addon", "broker"]
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "addon"},
+    )
+    # Addon will retry
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["progress_action"] == "start_addon"
+    assert result["step_id"] == "start_addon"
+    await hass.async_block_till_done()
+    await hass.async_block_till_done(wait_background_tasks=True)
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "start_addon"},
+    )
+
+    # add-on start-up failed
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "addon_start_failed"
+
+
+@pytest.mark.usefixtures(
+    "mqtt_client_mock",
+    "supervisor",
+    "start_addon",
+    "addon_installed",
+)
+async def test_addon_info_error(
+    hass: HomeAssistant,
+    addon_info: AsyncMock,
+) -> None:
+    """Test we perform an auto config flow with a supervised install.
+
+    Case: The Mosquitto add-on info could not be retrieved.
+    """
+    addon_info.side_effect = AddonError()
+
+    result = await hass.config_entries.flow.async_init(
+        "mqtt", context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["menu_options"] == ["addon", "broker"]
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "addon"},
+    )
+
+    # add-on info failed
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "addon_info_failed"
+
+
+@pytest.mark.usefixtures(
+    "mqtt_client_mock",
+    "supervisor",
+    "addon_info",
+    "addon_not_installed",
+    "install_addon",
+    "start_addon",
+)
+@pytest.mark.parametrize("discovery_info", [{"config": ADD_ON_DISCOVERY_INFO.copy()}])
+async def test_addon_flow_with_supervisor_addon_not_installed(
+    hass: HomeAssistant,
+    mock_try_connection_success: MagicMock,
+    mock_finish_setup: MagicMock,
+) -> None:
+    """Test we perform an auto config flow with a supervised install.
+
+    Case: The Mosquitto add-on is not yet installed nor running.
+    """
+    result = await hass.config_entries.flow.async_init(
+        "mqtt", context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["menu_options"] == ["addon", "broker"]
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "addon"},
+    )
+    # add-on not installed, so we wait for install
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["progress_action"] == "install_addon"
+    assert result["step_id"] == "install_addon"
+    await hass.async_block_till_done()
+    await hass.async_block_till_done(wait_background_tasks=True)
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "install_addon"},
+    )
+
+    # add-on installed but not started, so we wait for start-up
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["progress_action"] == "start_addon"
+    assert result["step_id"] == "start_addon"
+    await hass.async_block_till_done()
+    await hass.async_block_till_done(wait_background_tasks=True)
+    mock_try_connection_success.reset_mock()
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "start_addon"},
+    )
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["result"].data == {
+        "broker": "core-mosquitto",
+        "port": 1883,
+        "username": "mock-user",
+        "password": "mock-pass",
+        "discovery": True,
+    }
+    # Check we tried the connection
+    assert len(mock_try_connection_success.mock_calls)
+    # Check config entry got setup
+    assert len(mock_finish_setup.mock_calls) == 1
+
+
+@pytest.mark.usefixtures(
+    "mqtt_client_mock",
+    "supervisor",
+    "addon_info",
+    "addon_not_installed",
+    "start_addon",
+)
+async def test_addon_not_installed_failures(
+    hass: HomeAssistant,
+    install_addon: AsyncMock,
+) -> None:
+    """Test we perform an auto config flow with a supervised install.
+
+    Case: The Mosquitto add-on install fails.
+    """
+    install_addon.side_effect = HassioAPIError()
+
+    result = await hass.config_entries.flow.async_init(
+        "mqtt", context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["menu_options"] == ["addon", "broker"]
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "addon"},
+    )
+    # add-on not installed, so we wait for install
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["progress_action"] == "install_addon"
+    assert result["step_id"] == "install_addon"
+    await hass.async_block_till_done()
+    await hass.async_block_till_done(wait_background_tasks=True)
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {"next_step_id": "install_addon"},
+    )
+
+    # add-on install failed
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "addon_install_failed"
+
+
 async def test_option_flow(
     hass: HomeAssistant,
     mqtt_mock_entry: MqttMockHAClientGenerator,