TP-Link Omada integration (#81223)

* TP-Link Omada integration
Support for PoE config of network switch ports

* Bump omada client version

* Fixing tests

* Refactored site config flow

* Code review comments

* Fixed tests and device display name issue

* Bump isort to fix pre-commit hooks

* Hassfest for the win

* Omada code review

* Black

* More config flow test coverage

* Full coverage for omada config_flow

---------

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>

.coveragerc

@@ -1268,6 +1268,10 @@ omit =
     homeassistant/components/totalconnect/binary_sensor.py
     homeassistant/components/touchline/climate.py
     homeassistant/components/tplink_lte/*
+    homeassistant/components/tplink_omada/__init__.py
+    homeassistant/components/tplink_omada/coordinator.py
+    homeassistant/components/tplink_omada/entity.py
+    homeassistant/components/tplink_omada/switch.py
     homeassistant/components/traccar/device_tracker.py
     homeassistant/components/tractive/__init__.py
     homeassistant/components/tractive/binary_sensor.py

.strict-typing

@@ -299,6 +299,7 @@ homeassistant.components.tile.*
 homeassistant.components.tilt_ble.*
 homeassistant.components.tolo.*
 homeassistant.components.tplink.*
+homeassistant.components.tplink_omada.*
 homeassistant.components.tractive.*
 homeassistant.components.tradfri.*
 homeassistant.components.trafikverket_ferry.*

CODEOWNERS

@@ -1231,6 +1231,8 @@ build.json @home-assistant/supervisor
 /tests/components/totalconnect/ @austinmroczek
 /homeassistant/components/tplink/ @rytilahti @thegardenmonkey
 /tests/components/tplink/ @rytilahti @thegardenmonkey
+/homeassistant/components/tplink_omada/ @MarkGodwin
+/tests/components/tplink_omada/ @MarkGodwin
 /homeassistant/components/traccar/ @ludeeus
 /tests/components/traccar/ @ludeeus
 /homeassistant/components/trace/ @home-assistant/core

homeassistant/brands/tplink.json

@@ -0,0 +1,5 @@
+{
+  "domain": "tplink",
+  "name": "TP-Link",
+  "integrations": ["tplink", "tplink_omada", "tplink_lte"]
+}

homeassistant/components/tplink_omada/__init__.py

@@ -0,0 +1,60 @@
+"""The TP-Link Omada integration."""
+from __future__ import annotations
+
+from tplink_omada_client.exceptions import (
+    ConnectionFailed,
+    LoginFailed,
+    OmadaClientException,
+    UnsupportedControllerVersion,
+)
+from tplink_omada_client.omadaclient import OmadaSite
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+
+from .config_flow import CONF_SITE, create_omada_client
+from .const import DOMAIN
+
+PLATFORMS: list[Platform] = [Platform.SWITCH]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up TP-Link Omada from a config entry."""
+
+    hass.data.setdefault(DOMAIN, {})
+
+    try:
+        client = await create_omada_client(hass, entry.data)
+        await client.login()
+
+    except (LoginFailed, UnsupportedControllerVersion) as ex:
+        raise ConfigEntryAuthFailed(
+            f"Omada controller refused login attempt: {ex}"
+        ) from ex
+    except ConnectionFailed as ex:
+        raise ConfigEntryNotReady(
+            f"Omada controller could not be reached: {ex}"
+        ) from ex
+
+    except OmadaClientException as ex:
+        raise ConfigEntryNotReady(
+            f"Unexpected error connecting to Omada controller: {ex}"
+        ) from ex
+
+    site_client = await client.get_site_client(OmadaSite(None, entry.data[CONF_SITE]))
+
+    hass.data[DOMAIN][entry.entry_id] = site_client
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/tplink_omada/config_flow.py

@@ -0,0 +1,199 @@
+"""Config flow for TP-Link Omada integration."""
+from __future__ import annotations
+
+from collections.abc import Mapping
+import logging
+from types import MappingProxyType
+from typing import Any, NamedTuple
+
+from tplink_omada_client.exceptions import (
+    ConnectionFailed,
+    LoginFailed,
+    OmadaClientException,
+    UnsupportedControllerVersion,
+)
+from tplink_omada_client.omadaclient import OmadaClient, OmadaSite
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_VERIFY_SSL
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import selector
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+CONF_SITE = "site"
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST): str,
+        vol.Required(CONF_VERIFY_SSL, default=True): bool,
+        vol.Required(CONF_USERNAME): str,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+
+async def create_omada_client(
+    hass: HomeAssistant, data: MappingProxyType[str, Any]
+) -> OmadaClient:
+    """Create a TP-Link Omada client API for the given config entry."""
+    host = data[CONF_HOST]
+    verify_ssl = bool(data[CONF_VERIFY_SSL])
+    username = data[CONF_USERNAME]
+    password = data[CONF_PASSWORD]
+    websession = async_get_clientsession(hass, verify_ssl=verify_ssl)
+    return OmadaClient(host, username, password, websession=websession)
+
+
+class HubInfo(NamedTuple):
+    """Discovered controller information."""
+
+    controller_id: str
+    name: str
+    sites: list[OmadaSite]
+
+
+async def _validate_input(hass: HomeAssistant, data: dict[str, Any]) -> HubInfo:
+    """Validate the user input allows us to connect."""
+
+    client = await create_omada_client(hass, MappingProxyType(data))
+    controller_id = await client.login()
+    name = await client.get_controller_name()
+    sites = await client.get_sites()
+
+    return HubInfo(controller_id, name, sites)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for TP-Link Omada."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Create the config flow for a new integration."""
+        self._omada_opts: dict[str, Any] = {}
+        self._sites: list[OmadaSite] = []
+        self._controller_name = ""
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+
+        errors: dict[str, str] = {}
+        info = None
+        if user_input is not None:
+            info = await self._test_login(user_input, errors)
+
+        if info is None or user_input is None:
+            return self.async_show_form(
+                step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+            )
+
+        await self.async_set_unique_id(info.controller_id)
+        self._abort_if_unique_id_configured()
+
+        self._omada_opts.update(user_input)
+        self._sites = info.sites
+        self._controller_name = info.name
+        if len(self._sites) > 1:
+            return await self.async_step_site()
+        return await self.async_step_site({CONF_SITE: self._sites[0].id})
+
+    async def async_step_site(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle step to select site to manage."""
+
+        if user_input is None:
+            schema = vol.Schema(
+                {
+                    vol.Required(CONF_SITE, "site"): selector.SelectSelector(
+                        selector.SelectSelectorConfig(
+                            options=[
+                                selector.SelectOptionDict(value=s.id, label=s.name)
+                                for s in self._sites
+                            ],
+                            multiple=False,
+                            mode=selector.SelectSelectorMode.DROPDOWN,
+                        )
+                    )
+                }
+            )
+
+            return self.async_show_form(step_id="site", data_schema=schema)
+
+        self._omada_opts.update(user_input)
+        site_name = next(
+            site for site in self._sites if site.id == user_input["site"]
+        ).name
+        display_name = f"{self._controller_name} ({site_name})"
+
+        return self.async_create_entry(title=display_name, data=self._omada_opts)
+
+    async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
+        """Perform reauth upon an API authentication error."""
+        self._omada_opts = dict(entry_data)
+        return await self.async_step_reauth_confirm()
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Dialog that informs the user that reauth is required."""
+
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            self._omada_opts.update(user_input)
+            info = await self._test_login(self._omada_opts, errors)
+
+            if info is not None:
+                # Auth successful - update the config entry with the new credentials
+                entry = self.hass.config_entries.async_get_entry(
+                    self.context["entry_id"]
+                )
+                assert entry is not None
+                self.hass.config_entries.async_update_entry(
+                    entry, data=self._omada_opts
+                )
+                await self.hass.config_entries.async_reload(entry.entry_id)
+                return self.async_abort(reason="reauth_successful")
+
+        return self.async_show_form(
+            step_id="reauth_confirm",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_USERNAME): str,
+                    vol.Required(CONF_PASSWORD): str,
+                }
+            ),
+            errors=errors,
+        )
+
+    async def _test_login(
+        self, data: dict[str, Any], errors: dict[str, str]
+    ) -> HubInfo | None:
+        try:
+            info = await _validate_input(self.hass, data)
+            if len(info.sites) > 0:
+                return info
+            errors["base"] = "no_sites_found"
+
+        except ConnectionFailed:
+            errors["base"] = "cannot_connect"
+        except LoginFailed:
+            errors["base"] = "invalid_auth"
+        except UnsupportedControllerVersion:
+            errors["base"] = "unsupported_controller"
+        except OmadaClientException as ex:
+            _LOGGER.error("Unexpected API error: %s", ex)
+            errors["base"] = "unknown"
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Unexpected exception")
+            errors["base"] = "unknown"
+        return None

homeassistant/components/tplink_omada/const.py

@@ -0,0 +1,3 @@
+"""Constants for the TP-Link Omada integration."""
+
+DOMAIN = "tplink_omada"

homeassistant/components/tplink_omada/coordinator.py

@@ -0,0 +1,44 @@
+"""Generic Omada API coordinator."""
+from collections.abc import Awaitable, Callable
+from datetime import timedelta
+import logging
+from typing import Generic, TypeVar
+
+import async_timeout
+from tplink_omada_client.exceptions import OmadaClientException
+from tplink_omada_client.omadaclient import OmadaClient
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+_LOGGER = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class OmadaCoordinator(DataUpdateCoordinator[dict[str, T]], Generic[T]):
+    """Coordinator for synchronizing bulk Omada data."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        omada_client: OmadaClient,
+        update_func: Callable[[OmadaClient], Awaitable[dict[str, T]]],
+    ) -> None:
+        """Initialize my coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Omada API Data",
+            update_interval=timedelta(seconds=300),
+        )
+        self.omada_client = omada_client
+        self._update_func = update_func
+
+    async def _async_update_data(self) -> dict[str, T]:
+        """Fetch data from API endpoint."""
+        try:
+            async with async_timeout.timeout(10):
+                return await self._update_func(self.omada_client)
+        except OmadaClientException as err:
+            raise UpdateFailed(f"Error communicating with API: {err}") from err

homeassistant/components/tplink_omada/entity.py

@@ -0,0 +1,33 @@
+"""Base entity definitions."""
+from tplink_omada_client.devices import OmadaSwitch, OmadaSwitchPortDetails
+
+from homeassistant.helpers import device_registry
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import OmadaCoordinator
+
+
+class OmadaSwitchDeviceEntity(
+    CoordinatorEntity[OmadaCoordinator[OmadaSwitchPortDetails]]
+):
+    """Common base class for all entities attached to Omada network switches."""
+
+    def __init__(
+        self, coordinator: OmadaCoordinator[OmadaSwitchPortDetails], device: OmadaSwitch
+    ) -> None:
+        """Initialize the switch."""
+        super().__init__(coordinator)
+        self.device = device
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return information about the device."""
+        return DeviceInfo(
+            connections={(device_registry.CONNECTION_NETWORK_MAC, self.device.mac)},
+            identifiers={(DOMAIN, (self.device.mac))},
+            manufacturer="TP-Link",
+            model=self.device.model_display_name,
+            name=self.device.name,
+        )

homeassistant/components/tplink_omada/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "tplink_omada",
+  "name": "TP-Link Omada",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/tplink_omada",
+  "integration_type": "hub",
+  "requirements": ["tplink-omada-client==1.1.0"],
+  "codeowners": ["@MarkGodwin"],
+  "iot_class": "local_polling"
+}

homeassistant/components/tplink_omada/strings.json

@@ -0,0 +1,41 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "host": "[%key:common::config_flow::data::host%]",
+          "verify_ssl": "[%key:common::config_flow::data::verify_ssl%]",
+          "username": "[%key:common::config_flow::data::username%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        },
+        "title": "TP-Link Omada Controller",
+        "description": "Enter the connection details for the Omada controller. Cloud controllers aren't supported."
+      },
+      "site": {
+        "data": {
+          "site": "Site"
+        },
+        "title": "Choose which site(s) to manage"
+      },
+      "reauth_confirm": {
+        "data": {
+          "username": "[%key:common::config_flow::data::username%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        },
+        "title": "Update TP-Link Omada Credentials",
+        "description": "The provided credentials have stopped working. Please update them."
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "unsupported_controller": "Omada Controller version not supported.",
+      "unknown": "[%key:common::config_flow::error::unknown%]",
+      "no_sites_found": "No sites found which the user can manage."
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
+    }
+  }
+}

homeassistant/components/tplink_omada/switch.py

@@ -0,0 +1,118 @@
+"""Support for TPLink Omada device toggle options."""
+from __future__ import annotations
+
+from functools import partial
+from typing import Any
+
+from tplink_omada_client.definitions import PoEMode
+from tplink_omada_client.devices import OmadaSwitch, OmadaSwitchPortDetails
+from tplink_omada_client.omadasiteclient import OmadaSiteClient, SwitchPortOverrides
+
+from homeassistant.components.switch import SwitchEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity import EntityCategory
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN
+from .coordinator import OmadaCoordinator
+from .entity import OmadaSwitchDeviceEntity
+
+POE_SWITCH_ICON = "mdi:ethernet"
+
+
+async def poll_switch_state(
+    client: OmadaSiteClient, network_switch: OmadaSwitch
+) -> dict[str, OmadaSwitchPortDetails]:
+    """Poll a switch's current state."""
+    ports = await client.get_switch_ports(network_switch)
+    return {p.port_id: p for p in ports}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up switches."""
+    omada_client: OmadaSiteClient = hass.data[DOMAIN][config_entry.entry_id]
+
+    # Naming fun. Omada switches, as in the network hardware
+    network_switches = await omada_client.get_switches()
+
+    entities: list = []
+    for switch in [
+        ns for ns in network_switches if ns.device_capabilities.supports_poe
+    ]:
+        coordinator = OmadaCoordinator[OmadaSwitchPortDetails](
+            hass, omada_client, partial(poll_switch_state, network_switch=switch)
+        )
+
+        await coordinator.async_request_refresh()
+
+        for idx, port_id in enumerate(coordinator.data):
+            if idx < switch.device_capabilities.poe_ports:
+                entities.append(
+                    OmadaNetworkSwitchPortPoEControl(coordinator, switch, port_id)
+                )
+
+    async_add_entities(entities)
+
+
+def get_port_base_name(port: OmadaSwitchPortDetails) -> str:
+    """Get display name for a switch port."""
+
+    if port.name == f"Port{port.port}":
+        return f"Port {port.port}"
+    return f"Port {port.port} ({port.name})"
+
+
+class OmadaNetworkSwitchPortPoEControl(OmadaSwitchDeviceEntity, SwitchEntity):
+    """Representation of a PoE control toggle on a single network port on a switch."""
+
+    _attr_has_entity_name = True
+    _attr_entity_category = EntityCategory.CONFIG
+    _attr_icon = POE_SWITCH_ICON
+
+    def __init__(
+        self,
+        coordinator: OmadaCoordinator[OmadaSwitchPortDetails],
+        device: OmadaSwitch,
+        port_id: str,
+    ) -> None:
+        """Initialize the PoE switch."""
+        super().__init__(coordinator, device)
+        self.port_id = port_id
+        self.port_details = self.coordinator.data[port_id]
+        self.omada_client = self.coordinator.omada_client
+        self._attr_unique_id = f"{device.mac}_{port_id}_poe"
+
+        self._attr_name = f"{get_port_base_name(self.port_details)} PoE"
+        self._refresh_state()
+
+    async def _async_turn_on_off_poe(self, enable: bool) -> None:
+        self.port_details = await self.omada_client.update_switch_port(
+            self.device,
+            self.port_details,
+            overrides=SwitchPortOverrides(enable_poe=enable),
+        )
+        self._refresh_state()
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn the entity on."""
+        await self._async_turn_on_off_poe(True)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn the entity off."""
+        await self._async_turn_on_off_poe(False)
+
+    def _refresh_state(self) -> None:
+        self._attr_is_on = self.port_details.poe_mode != PoEMode.DISABLED
+        if self.hass:
+            self.async_write_ha_state()
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self.port_details = self.coordinator.data[self.port_id]
+        self._refresh_state()

homeassistant/components/tplink_omada/translations/en.json

@@ -0,0 +1,40 @@
+{
+    "config": {
+        "abort": {
+            "already_configured": "Device is already configured"
+        },
+        "error": {
+            "cannot_connect": "Failed to connect",
+            "unsupported_controller": "Omada Controller version not supported.",
+            "invalid_auth": "Invalid authentication",
+            "unknown": "Unexpected error",
+            "no_sites_found": "No sites found which the user can manage."
+        },
+        "step": {
+            "user": {
+                "data": {
+                    "host": "Host",
+                    "verify_ssl": "Verify SSL certificate",
+                    "password": "Password",
+                    "username": "Username"
+                },
+                "title": "TP-Link Omada Controller",
+                "description": "Enter the connection details for the Omada controller. Cloud controllers aren't supported."
+            },
+            "site": {
+                "data": {
+                    "site": "Site"
+                },
+                "title": "Choose which site(s) to manage"
+            },
+            "reauth_confirm": {
+                "data": {
+                    "password": "Password",
+                    "username": "Username"
+                },
+                "title": "Update TP-Link Omada Credentials",
+                "description": "The provided username and password have stopped working. Please update them."
+            }
+        }
+    }
+}

homeassistant/generated/config_flows.py

@@ -443,6 +443,7 @@ FLOWS = {
         "toon",
         "totalconnect",
         "tplink",
+        "tplink_omada",
         "traccar",
         "tractive",
         "tradfri",

homeassistant/generated/integrations.json

@@ -5680,16 +5680,27 @@
       "iot_class": "local_polling"
     },
     "tplink": {
-      "name": "TP-Link Kasa Smart",
-      "integration_type": "hub",
-      "config_flow": true,
-      "iot_class": "local_polling"
-    },
-    "tplink_lte": {
-      "name": "TP-Link LTE",
-      "integration_type": "hub",
-      "config_flow": false,
-      "iot_class": "local_polling"
+      "name": "TP-Link",
+      "integrations": {
+        "tplink": {
+          "integration_type": "hub",
+          "config_flow": true,
+          "iot_class": "local_polling",
+          "name": "TP-Link Kasa Smart"
+        },
+        "tplink_omada": {
+          "integration_type": "hub",
+          "config_flow": true,
+          "iot_class": "local_polling",
+          "name": "TP-Link Omada"
+        },
+        "tplink_lte": {
+          "integration_type": "hub",
+          "config_flow": false,
+          "iot_class": "local_polling",
+          "name": "TP-Link LTE"
+        }
+      }
     },
     "traccar": {
       "name": "Traccar",

mypy.ini

@@ -2745,6 +2745,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.tplink_omada.*]
+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.tractive.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_all.txt

@@ -2517,6 +2517,9 @@ total_connect_client==2023.1
 # homeassistant.components.tplink_lte
 tp-connected==0.0.4
 
+# homeassistant.components.tplink_omada
+tplink-omada-client==1.1.0
+
 # homeassistant.components.transmission
 transmission-rpc==3.4.0
 

requirements_test_all.txt

@@ -1766,6 +1766,9 @@ toonapi==0.2.1
 # homeassistant.components.totalconnect
 total_connect_client==2023.1
 
+# homeassistant.components.tplink_omada
+tplink-omada-client==1.1.0
+
 # homeassistant.components.transmission
 transmission-rpc==3.4.0
 

tests/components/tplink_omada/__init__.py

@@ -0,0 +1 @@
+"""Tests for the TP-Link Omada integration."""

tests/components/tplink_omada/test_config_flow.py

@@ -0,0 +1,358 @@
+"""Test the TP-Link Omada config flows."""
+from unittest.mock import patch
+
+from tplink_omada_client.exceptions import (
+    ConnectionFailed,
+    LoginFailed,
+    OmadaClientException,
+    UnsupportedControllerVersion,
+)
+from tplink_omada_client.omadaclient import OmadaSite
+
+from homeassistant import config_entries
+from homeassistant.components.tplink_omada.config_flow import (
+    HubInfo,
+    _validate_input,
+    create_omada_client,
+)
+from homeassistant.components.tplink_omada.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from tests.common import MockConfigEntry
+
+MOCK_USER_DATA = {
+    "host": "1.1.1.1",
+    "verify_ssl": True,
+    "username": "test-username",
+    "password": "test-password",
+}
+
+MOCK_ENTRY_DATA = {
+    "host": "1.1.1.1",
+    "verify_ssl": True,
+    "site": "SiteId",
+    "username": "test-username",
+    "password": "test-password",
+}
+
+
+async def test_form_single_site(hass: HomeAssistant) -> None:
+    """Test we get the form."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["errors"] == {}
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        return_value=HubInfo(
+            "omada_id", "OC200", [OmadaSite("Display Name", "SiteId")]
+        ),
+    ) as mocked_validate, patch(
+        "homeassistant.components.tplink_omada.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            MOCK_USER_DATA,
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.CREATE_ENTRY
+    assert result2["title"] == "OC200 (Display Name)"
+    assert result2["data"] == MOCK_ENTRY_DATA
+    assert len(mock_setup_entry.mock_calls) == 1
+    mocked_validate.assert_called_once_with(hass, MOCK_USER_DATA)
+
+
+async def test_form_multiple_sites(hass: HomeAssistant) -> None:
+    """Test we get the form."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        return_value=HubInfo(
+            "omada_id",
+            "OC200",
+            [OmadaSite("Site 1", "first"), OmadaSite("Site 2", "second")],
+        ),
+    ), patch(
+        "homeassistant.components.tplink_omada.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            MOCK_USER_DATA,
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["step_id"] == "site"
+
+    with patch(
+        "homeassistant.components.tplink_omada.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result3 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "site": "second",
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == FlowResultType.CREATE_ENTRY
+    assert result3["title"] == "OC200 (Site 2)"
+    assert result3["data"] == {
+        "host": "1.1.1.1",
+        "verify_ssl": True,
+        "site": "second",
+        "username": "test-username",
+        "password": "test-password",
+    }
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_form_invalid_auth(hass: HomeAssistant) -> None:
+    """Test we handle invalid auth."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        side_effect=LoginFailed(-1000, "Invalid username/password"),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            MOCK_USER_DATA,
+        )
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["errors"] == {"base": "invalid_auth"}
+
+
+async def test_form_api_error(hass: HomeAssistant) -> None:
+    """Test we handle unknown API error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        side_effect=OmadaClientException,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            MOCK_USER_DATA,
+        )
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["errors"] == {"base": "unknown"}
+
+
+async def test_form_generic_exception(hass: HomeAssistant) -> None:
+    """Test we handle unknown API error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        side_effect=Exception,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            MOCK_USER_DATA,
+        )
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["errors"] == {"base": "unknown"}
+
+
+async def test_form_unsupported_controller(hass: HomeAssistant) -> None:
+    """Test we handle unknown API error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        side_effect=UnsupportedControllerVersion("4.0.0"),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            MOCK_USER_DATA,
+        )
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["errors"] == {"base": "unsupported_controller"}
+
+
+async def test_form_cannot_connect(hass: HomeAssistant) -> None:
+    """Test we handle cannot connect error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        side_effect=ConnectionFailed,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            MOCK_USER_DATA,
+        )
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["errors"] == {"base": "cannot_connect"}
+
+
+async def test_form_no_sites(hass: HomeAssistant) -> None:
+    """Test we handle invalid auth."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        return_value=HubInfo("omada_id", "OC200", []),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            MOCK_USER_DATA,
+        )
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["errors"] == {"base": "no_sites_found"}
+
+
+async def test_async_step_reauth_success(hass: HomeAssistant) -> None:
+    """Test reauth starts an interactive flow."""
+
+    mock_entry = MockConfigEntry(
+        domain="tplink_omada",
+        data=dict(MOCK_ENTRY_DATA),
+        unique_id="USERID",
+    )
+    mock_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={
+            "source": config_entries.SOURCE_REAUTH,
+            "entry_id": mock_entry.entry_id,
+        },
+        data=mock_entry.data,
+    )
+
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        return_value=HubInfo(
+            "omada_id", "OC200", [OmadaSite("Display Name", "SiteId")]
+        ),
+    ) as mocked_validate:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"username": "new_uname", "password": "new_passwd"}
+        )
+    await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.ABORT
+    assert result2["reason"] == "reauth_successful"
+    mocked_validate.assert_called_once_with(
+        hass,
+        {
+            "host": "1.1.1.1",
+            "verify_ssl": True,
+            "site": "SiteId",
+            "username": "new_uname",
+            "password": "new_passwd",
+        },
+    )
+
+
+async def test_async_step_reauth_invalid_auth(hass: HomeAssistant) -> None:
+    """Test reauth starts an interactive flow."""
+
+    mock_entry = MockConfigEntry(
+        domain="tplink_omada",
+        data=dict(MOCK_ENTRY_DATA),
+        unique_id="USERID",
+    )
+    mock_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={
+            "source": config_entries.SOURCE_REAUTH,
+            "entry_id": mock_entry.entry_id,
+        },
+        data=mock_entry.data,
+    )
+
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow._validate_input",
+        side_effect=LoginFailed(-1000, "Invalid username/password"),
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"username": "new_uname", "password": "new_passwd"}
+        )
+    await hass.async_block_till_done()
+
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "reauth_confirm"
+    assert result2["errors"] == {"base": "invalid_auth"}
+
+
+async def test_validate_input(hass: HomeAssistant) -> None:
+    """Test validate returns HubInfo."""
+
+    with patch(
+        "tplink_omada_client.omadaclient.OmadaClient", autospec=True
+    ) as mock_client, patch(
+        "homeassistant.components.tplink_omada.config_flow.create_omada_client",
+        return_value=mock_client,
+    ) as create_mock:
+        mock_client.login.return_value = "Id"
+        mock_client.get_controller_name.return_value = "Name"
+        mock_client.get_sites.return_value = [OmadaSite("x", "y")]
+        result = await _validate_input(hass, MOCK_USER_DATA)
+
+    create_mock.assert_awaited_once()
+    mock_client.login.assert_awaited_once()
+    mock_client.get_controller_name.assert_awaited_once()
+    mock_client.get_sites.assert_awaited_once()
+    assert result.controller_id == "Id"
+    assert result.name == "Name"
+    assert result.sites == [OmadaSite("x", "y")]
+
+
+async def test_create_omada_client_parses_args(hass: HomeAssistant) -> None:
+    """Test config arguments are passed to Omada client."""
+
+    with patch(
+        "homeassistant.components.tplink_omada.config_flow.OmadaClient", autospec=True
+    ) as mock_client, patch(
+        "homeassistant.components.tplink_omada.config_flow.async_get_clientsession",
+        return_value="ws",
+    ) as mock_clientsession:
+        result = await create_omada_client(hass, MOCK_USER_DATA)
+
+    assert result is not None
+    mock_client.assert_called_once_with(
+        "1.1.1.1", "test-username", "test-password", "ws"
+    )
+    mock_clientsession.assert_called_once_with(hass, verify_ssl=True)