Add romy vacuum integration (#93750)

Co-authored-by: Erik Montnemery <erik@montnemery.com>
Co-authored-by: Robert Resch <robert@resch.dev>
Co-authored-by: Allen Porter <allen.porter@gmail.com>

.coveragerc

@@ -1103,6 +1103,9 @@ omit =
     homeassistant/components/ripple/sensor.py
     homeassistant/components/roborock/coordinator.py
     homeassistant/components/rocketchat/notify.py
+    homeassistant/components/romy/__init__.py
+    homeassistant/components/romy/coordinator.py
+    homeassistant/components/romy/vacuum.py
     homeassistant/components/roomba/__init__.py
     homeassistant/components/roomba/binary_sensor.py
     homeassistant/components/roomba/braava.py

.strict-typing

@@ -361,6 +361,7 @@ homeassistant.components.rhasspy.*
 homeassistant.components.ridwell.*
 homeassistant.components.rituals_perfume_genie.*
 homeassistant.components.roku.*
+homeassistant.components.romy.*
 homeassistant.components.rpi_power.*
 homeassistant.components.rss_feed_template.*
 homeassistant.components.rtsp_to_webrtc.*

CODEOWNERS

@@ -1122,6 +1122,8 @@ build.json @home-assistant/supervisor
 /tests/components/roborock/ @humbertogontijo @Lash-L
 /homeassistant/components/roku/ @ctalkington
 /tests/components/roku/ @ctalkington
+/homeassistant/components/romy/ @xeniter
+/tests/components/romy/ @xeniter
 /homeassistant/components/roomba/ @pschmitt @cyr-ius @shenxn @Xitee1
 /tests/components/roomba/ @pschmitt @cyr-ius @shenxn @Xitee1
 /homeassistant/components/roon/ @pavoni

homeassistant/components/romy/__init__.py

@@ -0,0 +1,42 @@
+"""ROMY Integration."""
+
+import romy
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PASSWORD
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN, LOGGER, PLATFORMS
+from .coordinator import RomyVacuumCoordinator
+
+
+async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
+    """Initialize the ROMY platform via config entry."""
+
+    new_romy = await romy.create_romy(
+        config_entry.data[CONF_HOST], config_entry.data.get(CONF_PASSWORD, "")
+    )
+
+    coordinator = RomyVacuumCoordinator(hass, new_romy)
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
+
+    config_entry.async_on_unload(config_entry.add_update_listener(update_listener))
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Handle removal of an entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+    return unload_ok
+
+
+async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
+    """Handle options update."""
+    LOGGER.debug("update_listener")
+    await hass.config_entries.async_reload(config_entry.entry_id)

homeassistant/components/romy/config_flow.py

@@ -0,0 +1,148 @@
+"""Config flow for ROMY integration."""
+from __future__ import annotations
+
+import romy
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.components import zeroconf
+from homeassistant.const import CONF_HOST, CONF_PASSWORD
+from homeassistant.data_entry_flow import FlowResult
+import homeassistant.helpers.config_validation as cv
+
+from .const import DOMAIN, LOGGER
+
+
+class RomyConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle config flow for ROMY."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Handle a config flow for ROMY."""
+        self.host: str = ""
+        self.password: str = ""
+        self.robot_name_given_by_user: str = ""
+
+    async def async_step_user(
+        self, user_input: dict[str, str] | None = None
+    ) -> FlowResult:
+        """Handle the user step."""
+        errors: dict[str, str] = {}
+
+        if user_input:
+            self.host = user_input[CONF_HOST]
+
+            new_romy = await romy.create_romy(self.host, "")
+
+            if not new_romy.is_initialized:
+                errors[CONF_HOST] = "cannot_connect"
+            else:
+                await self.async_set_unique_id(new_romy.unique_id)
+                self._abort_if_unique_id_configured()
+
+                self.robot_name_given_by_user = new_romy.user_name
+
+                if not new_romy.is_unlocked:
+                    return await self.async_step_password()
+                return await self._async_step_finish_config()
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_HOST): cv.string,
+                },
+            ),
+            errors=errors,
+        )
+
+    async def async_step_password(
+        self, user_input: dict[str, str] | None = None
+    ) -> FlowResult:
+        """Unlock the robots local http interface with password."""
+        errors: dict[str, str] = {}
+
+        if user_input:
+            self.password = user_input[CONF_PASSWORD]
+            new_romy = await romy.create_romy(self.host, self.password)
+
+            if not new_romy.is_initialized:
+                errors[CONF_PASSWORD] = "cannot_connect"
+            elif not new_romy.is_unlocked:
+                errors[CONF_PASSWORD] = "invalid_auth"
+
+            if not errors:
+                return await self._async_step_finish_config()
+
+        return self.async_show_form(
+            step_id="password",
+            data_schema=vol.Schema(
+                {vol.Required(CONF_PASSWORD): vol.All(cv.string, vol.Length(8))},
+            ),
+            errors=errors,
+        )
+
+    async def async_step_zeroconf(
+        self, discovery_info: zeroconf.ZeroconfServiceInfo
+    ) -> FlowResult:
+        """Handle zeroconf discovery."""
+
+        LOGGER.debug("Zeroconf discovery_info: %s", discovery_info)
+
+        # connect and gather information from your ROMY
+        self.host = discovery_info.host
+        LOGGER.debug("ZeroConf Host: %s", self.host)
+
+        new_discovered_romy = await romy.create_romy(self.host, "")
+
+        self.robot_name_given_by_user = new_discovered_romy.user_name
+        LOGGER.debug("ZeroConf Name: %s", self.robot_name_given_by_user)
+
+        # get unique id and stop discovery if robot is already added
+        unique_id = new_discovered_romy.unique_id
+        LOGGER.debug("ZeroConf Unique_id: %s", unique_id)
+        await self.async_set_unique_id(unique_id)
+        self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})
+
+        self.context.update(
+            {
+                "title_placeholders": {
+                    "name": f"{self.robot_name_given_by_user} ({self.host} / {unique_id})"
+                },
+                "configuration_url": f"http://{self.host}:{new_discovered_romy.port}",
+            }
+        )
+
+        # if robot got already unlocked with password add it directly
+        if not new_discovered_romy.is_initialized:
+            return self.async_abort(reason="cannot_connect")
+
+        if new_discovered_romy.is_unlocked:
+            return await self.async_step_zeroconf_confirm()
+
+        return await self.async_step_password()
+
+    async def async_step_zeroconf_confirm(
+        self, user_input: dict[str, str] | None = None
+    ) -> FlowResult:
+        """Handle a confirmation flow initiated by zeroconf."""
+        if user_input is None:
+            return self.async_show_form(
+                step_id="zeroconf_confirm",
+                description_placeholders={
+                    "name": self.robot_name_given_by_user,
+                    "host": self.host,
+                },
+            )
+        return await self._async_step_finish_config()
+
+    async def _async_step_finish_config(self) -> FlowResult:
+        """Finish the configuration setup."""
+        return self.async_create_entry(
+            title=self.robot_name_given_by_user,
+            data={
+                CONF_HOST: self.host,
+                CONF_PASSWORD: self.password,
+            },
+        )

homeassistant/components/romy/const.py

@@ -0,0 +1,11 @@
+"""Constants for the ROMY integration."""
+
+from datetime import timedelta
+import logging
+
+from homeassistant.const import Platform
+
+DOMAIN = "romy"
+PLATFORMS = [Platform.VACUUM]
+UPDATE_INTERVAL = timedelta(seconds=5)
+LOGGER = logging.getLogger(__package__)

homeassistant/components/romy/coordinator.py

@@ -0,0 +1,22 @@
+"""ROMY coordinator."""
+
+from romy import RomyRobot
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN, LOGGER, UPDATE_INTERVAL
+
+
+class RomyVacuumCoordinator(DataUpdateCoordinator[None]):
+    """ROMY Vacuum Coordinator."""
+
+    def __init__(self, hass: HomeAssistant, romy: RomyRobot) -> None:
+        """Initialize."""
+        super().__init__(hass, LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
+        self.hass = hass
+        self.romy = romy
+
+    async def _async_update_data(self) -> None:
+        """Update ROMY Vacuum Cleaner data."""
+        await self.romy.async_update()

homeassistant/components/romy/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "romy",
+  "name": "ROMY Vacuum Cleaner",
+  "codeowners": ["@xeniter"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/romy",
+  "iot_class": "local_polling",
+  "requirements": ["romy==0.0.7"],
+  "zeroconf": ["_aicu-http._tcp.local."]
+}

homeassistant/components/romy/strings.json

@@ -0,0 +1,51 @@
+{
+  "config": {
+    "flow_title": "{name}",
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
+    },
+    "step": {
+      "user": {
+        "data": {
+          "host": "[%key:common::config_flow::data::host%]"
+        }
+      },
+      "password": {
+        "title": "Password required",
+        "data": {
+          "password": "[%key:common::config_flow::data::password%]"
+        },
+        "data_description": {
+          "password": "(8 characters, see QR Code under the dustbin)."
+        }
+      },
+      "zeroconf_confirm": {
+        "description": "Do you want to add ROMY Vacuum Cleaner {name} to Home Assistant?"
+      }
+    }
+  },
+  "entity": {
+    "vacuum": {
+      "romy": {
+        "state_attributes": {
+          "fan_speed": {
+            "state": {
+              "default": "Default",
+              "normal": "Normal",
+              "silent": "Silent",
+              "intensive": "Intensive",
+              "super_silent": "Super silent",
+              "high": "High",
+              "auto": "Auto"
+            }
+          }
+        }
+      }
+    }
+  }
+}

homeassistant/components/romy/vacuum.py

@@ -0,0 +1,116 @@
+"""Support for Wi-Fi enabled ROMY vacuum cleaner robots.
+
+For more details about this platform, please refer to the documentation
+https://home-assistant.io/components/vacuum.romy/.
+"""
+
+
+from typing import Any
+
+from romy import RomyRobot
+
+from homeassistant.components.vacuum import StateVacuumEntity, VacuumEntityFeature
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN, LOGGER
+from .coordinator import RomyVacuumCoordinator
+
+ICON = "mdi:robot-vacuum"
+
+FAN_SPEED_NONE = "default"
+FAN_SPEED_NORMAL = "normal"
+FAN_SPEED_SILENT = "silent"
+FAN_SPEED_INTENSIVE = "intensive"
+FAN_SPEED_SUPER_SILENT = "super_silent"
+FAN_SPEED_HIGH = "high"
+FAN_SPEED_AUTO = "auto"
+
+FAN_SPEEDS: list[str] = [
+    FAN_SPEED_NONE,
+    FAN_SPEED_NORMAL,
+    FAN_SPEED_SILENT,
+    FAN_SPEED_INTENSIVE,
+    FAN_SPEED_SUPER_SILENT,
+    FAN_SPEED_HIGH,
+    FAN_SPEED_AUTO,
+]
+
+# Commonly supported features
+SUPPORT_ROMY_ROBOT = (
+    VacuumEntityFeature.BATTERY
+    | VacuumEntityFeature.RETURN_HOME
+    | VacuumEntityFeature.STATE
+    | VacuumEntityFeature.START
+    | VacuumEntityFeature.STOP
+    | VacuumEntityFeature.FAN_SPEED
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up ROMY vacuum cleaner."""
+
+    coordinator: RomyVacuumCoordinator = hass.data[DOMAIN][config_entry.entry_id]
+    async_add_entities([RomyVacuumEntity(coordinator, coordinator.romy)], True)
+
+
+class RomyVacuumEntity(CoordinatorEntity[RomyVacuumCoordinator], StateVacuumEntity):
+    """Representation of a ROMY vacuum cleaner robot."""
+
+    _attr_has_entity_name = True
+    _attr_name = None
+    _attr_supported_features = SUPPORT_ROMY_ROBOT
+    _attr_fan_speed_list = FAN_SPEEDS
+    _attr_icon = ICON
+
+    def __init__(
+        self,
+        coordinator: RomyVacuumCoordinator,
+        romy: RomyRobot,
+    ) -> None:
+        """Initialize the ROMY Robot."""
+        super().__init__(coordinator)
+        self.romy = romy
+        self._attr_unique_id = self.romy.unique_id
+        self._device_info = DeviceInfo(
+            identifiers={(DOMAIN, romy.unique_id)},
+            manufacturer="ROMY",
+            name=romy.name,
+            model=romy.model,
+        )
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self._attr_fan_speed = FAN_SPEEDS[self.romy.fan_speed]
+        self._attr_battery_level = self.romy.battery_level
+        self._attr_state = self.romy.status
+
+        self.async_write_ha_state()
+
+    async def async_start(self, **kwargs: Any) -> None:
+        """Turn the vacuum on."""
+        LOGGER.debug("async_start")
+        await self.romy.async_clean_start_or_continue()
+
+    async def async_stop(self, **kwargs: Any) -> None:
+        """Stop the vacuum cleaner."""
+        LOGGER.debug("async_stop")
+        await self.romy.async_stop()
+
+    async def async_return_to_base(self, **kwargs: Any) -> None:
+        """Return vacuum back to base."""
+        LOGGER.debug("async_return_to_base")
+        await self.romy.async_return_to_base()
+
+    async def async_set_fan_speed(self, fan_speed: str, **kwargs: Any) -> None:
+        """Set fan speed."""
+        LOGGER.debug("async_set_fan_speed to %s", fan_speed)
+        await self.romy.async_set_fan_speed(FAN_SPEEDS.index(fan_speed))

homeassistant/generated/config_flows.py

@@ -429,6 +429,7 @@ FLOWS = {
         "rituals_perfume_genie",
         "roborock",
         "roku",
+        "romy",
         "roomba",
         "roon",
         "rpi_power",

homeassistant/generated/integrations.json

@@ -4974,6 +4974,12 @@
       "config_flow": true,
       "iot_class": "local_polling"
     },
+    "romy": {
+      "name": "ROMY Vacuum Cleaner",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "roomba": {
       "name": "iRobot Roomba and Braava",
       "integration_type": "hub",

homeassistant/generated/zeroconf.py

@@ -248,6 +248,11 @@ ZEROCONF = {
             "domain": "volumio",
         },
     ],
+    "_aicu-http._tcp.local.": [
+        {
+            "domain": "romy",
+        },
+    ],
     "_airplay._tcp.local.": [
         {
             "domain": "apple_tv",

mypy.ini

@@ -3371,6 +3371,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.romy.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.rpi_power.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_all.txt

@@ -2439,6 +2439,9 @@ rocketchat-API==0.6.1
 # homeassistant.components.roku
 rokuecp==0.18.1
 
+# homeassistant.components.romy
+romy==0.0.7
+
 # homeassistant.components.roomba
 roombapy==1.6.10
 

requirements_test_all.txt

@@ -1858,6 +1858,9 @@ ring-doorbell[listen]==0.8.5
 # homeassistant.components.roku
 rokuecp==0.18.1
 
+# homeassistant.components.romy
+romy==0.0.7
+
 # homeassistant.components.roomba
 roombapy==1.6.10
 

tests/components/romy/__init__.py

@@ -0,0 +1 @@
+"""Tests for the ROMY integration."""

tests/components/romy/test_config_flow.py

@@ -0,0 +1,248 @@
+"""Test the ROMY config flow."""
+from ipaddress import ip_address
+from unittest.mock import Mock, PropertyMock, patch
+
+from romy import RomyRobot
+
+from homeassistant import config_entries, data_entry_flow
+from homeassistant.components import zeroconf
+from homeassistant.components.romy.const import DOMAIN
+from homeassistant.const import CONF_HOST, CONF_PASSWORD
+from homeassistant.core import HomeAssistant
+
+
+def _create_mocked_romy(
+    is_initialized,
+    is_unlocked,
+    name="Agon",
+    user_name="MyROMY",
+    unique_id="aicu-aicgsbksisfapcjqmqjq",
+    model="005:000:000:000:005",
+    port=8080,
+):
+    mocked_romy = Mock(spec_set=RomyRobot)
+    type(mocked_romy).is_initialized = PropertyMock(return_value=is_initialized)
+    type(mocked_romy).is_unlocked = PropertyMock(return_value=is_unlocked)
+    type(mocked_romy).name = PropertyMock(return_value=name)
+    type(mocked_romy).user_name = PropertyMock(return_value=user_name)
+    type(mocked_romy).unique_id = PropertyMock(return_value=unique_id)
+    type(mocked_romy).port = PropertyMock(return_value=port)
+    type(mocked_romy).model = PropertyMock(return_value=model)
+
+    return mocked_romy
+
+
+CONFIG = {CONF_HOST: "1.2.3.4", CONF_PASSWORD: "12345678"}
+
+INPUT_CONFIG_HOST = {
+    CONF_HOST: CONFIG[CONF_HOST],
+}
+
+
+async def test_show_user_form_robot_is_offline_and_locked(hass: HomeAssistant) -> None:
+    """Test that the user set up form with config."""
+
+    # Robot not reachable
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(False, False),
+    ):
+        result1 = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+            data=INPUT_CONFIG_HOST,
+        )
+
+        assert result1["errors"].get("host") == "cannot_connect"
+        assert result1["step_id"] == "user"
+        assert result1["type"] == data_entry_flow.FlowResultType.FORM
+
+    # Robot is locked
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(True, False),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result1["flow_id"], {"host": "1.2.3.4"}
+        )
+
+        assert result2["step_id"] == "password"
+        assert result2["type"] == data_entry_flow.FlowResultType.FORM
+
+    # Robot is initialized and unlocked
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(True, True),
+    ):
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"], {"password": "12345678"}
+        )
+
+        assert "errors" not in result3
+        assert result3["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
+
+
+async def test_show_user_form_robot_unlock_with_password(hass: HomeAssistant) -> None:
+    """Test that the user set up form with config."""
+
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(True, False),
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+            data=INPUT_CONFIG_HOST,
+        )
+
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(True, False),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"password": "12345678"}
+        )
+
+        assert result2["errors"] == {"password": "invalid_auth"}
+        assert result2["step_id"] == "password"
+        assert result2["type"] == data_entry_flow.FlowResultType.FORM
+
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(False, False),
+    ):
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"], {"password": "12345678"}
+        )
+
+        assert result3["errors"] == {"password": "cannot_connect"}
+        assert result3["step_id"] == "password"
+        assert result3["type"] == data_entry_flow.FlowResultType.FORM
+
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(True, True),
+    ):
+        result4 = await hass.config_entries.flow.async_configure(
+            result3["flow_id"], {"password": "12345678"}
+        )
+
+        assert "errors" not in result4
+        assert result4["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
+
+
+async def test_show_user_form_robot_reachable_again(hass: HomeAssistant) -> None:
+    """Test that the user set up form with config."""
+
+    # Robot not reachable
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(False, False),
+    ):
+        result1 = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_USER},
+            data=INPUT_CONFIG_HOST,
+        )
+
+        assert result1["errors"].get("host") == "cannot_connect"
+        assert result1["step_id"] == "user"
+        assert result1["type"] == data_entry_flow.FlowResultType.FORM
+
+    # Robot is locked
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(True, True),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result1["flow_id"], {"host": "1.2.3.4"}
+        )
+
+        assert "errors" not in result2
+        assert result2["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
+
+
+DISCOVERY_INFO = zeroconf.ZeroconfServiceInfo(
+    ip_address=ip_address("1.2.3.4"),
+    ip_addresses=[ip_address("1.2.3.4")],
+    port=8080,
+    hostname="aicu-aicgsbksisfapcjqmqjq.local",
+    type="mock_type",
+    name="myROMY",
+    properties={zeroconf.ATTR_PROPERTIES_ID: "aicu-aicgsbksisfapcjqmqjqZERO"},
+)
+
+
+async def test_zero_conf_locked_interface_robot(hass: HomeAssistant) -> None:
+    """Test zerconf which discovered locked robot."""
+
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(True, False),
+    ):
+        result1 = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            data=DISCOVERY_INFO,
+            context={"source": config_entries.SOURCE_ZEROCONF},
+        )
+
+    assert result1["step_id"] == "password"
+    assert result1["type"] == data_entry_flow.FlowResultType.FORM
+
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(True, True),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result1["flow_id"], {"password": "12345678"}
+        )
+
+        assert "errors" not in result2
+        assert result2["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
+
+
+async def test_zero_conf_uninitialized_robot(hass: HomeAssistant) -> None:
+    """Test zerconf which discovered locked robot."""
+
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(False, False),
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            data=DISCOVERY_INFO,
+            context={"source": config_entries.SOURCE_ZEROCONF},
+        )
+
+    assert result["reason"] == "cannot_connect"
+    assert result["type"] == data_entry_flow.FlowResultType.ABORT
+
+
+async def test_zero_conf_unlocked_interface_robot(hass: HomeAssistant) -> None:
+    """Test zerconf which discovered already unlocked robot."""
+
+    with patch(
+        "homeassistant.components.romy.config_flow.romy.create_romy",
+        return_value=_create_mocked_romy(True, True),
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            data=DISCOVERY_INFO,
+            context={"source": config_entries.SOURCE_ZEROCONF},
+        )
+
+    assert result["step_id"] == "zeroconf_confirm"
+    assert result["type"] == data_entry_flow.FlowResultType.FORM
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={CONF_HOST: "1.2.3.4"},
+    )
+
+    assert result["data"]
+    assert result["data"][CONF_HOST] == "1.2.3.4"
+
+    assert result["result"]
+    assert result["result"].unique_id == "aicu-aicgsbksisfapcjqmqjq"
+
+    assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY