Add vlc telnet config flow (#57513)

.coveragerc

@@ -1173,6 +1173,7 @@ omit =
     homeassistant/components/vilfo/const.py
     homeassistant/components/vivotek/camera.py
     homeassistant/components/vlc/media_player.py
+    homeassistant/components/vlc_telnet/__init__.py
     homeassistant/components/vlc_telnet/media_player.py
     homeassistant/components/volkszaehler/sensor.py
     homeassistant/components/volumio/__init__.py

CODEOWNERS

@@ -571,7 +571,7 @@ homeassistant/components/vicare/* @oischinger
 homeassistant/components/vilfo/* @ManneW
 homeassistant/components/vivotek/* @HarlemSquirrel
 homeassistant/components/vizio/* @raman325
-homeassistant/components/vlc_telnet/* @rodripf @dmcc
+homeassistant/components/vlc_telnet/* @rodripf @dmcc @MartinHjelmare
 homeassistant/components/volkszaehler/* @fabaff
 homeassistant/components/volumio/* @OnFreund
 homeassistant/components/wake_on_lan/* @ntilley905

homeassistant/components/vlc_telnet/__init__.py

@@ -1 +1,67 @@
-"""The vlc component."""
+"""The VLC media player Telnet integration."""
+from aiovlc.client import Client
+from aiovlc.exceptions import AuthError, ConnectError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+
+from .const import DATA_AVAILABLE, DATA_VLC, DOMAIN, LOGGER
+
+PLATFORMS = ["media_player"]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up VLC media player Telnet from a config entry."""
+    config = entry.data
+
+    host = config[CONF_HOST]
+    port = config[CONF_PORT]
+    password = config[CONF_PASSWORD]
+
+    vlc = Client(password=password, host=host, port=port)
+
+    available = True
+
+    try:
+        await vlc.connect()
+    except ConnectError as err:
+        LOGGER.warning("Failed to connect to VLC: %s. Trying again", err)
+        available = False
+
+    if available:
+        try:
+            await vlc.login()
+        except AuthError as err:
+            await disconnect_vlc(vlc)
+            raise ConfigEntryAuthFailed() from err
+
+    domain_data = hass.data.setdefault(DOMAIN, {})
+    domain_data[entry.entry_id] = {DATA_VLC: vlc, DATA_AVAILABLE: available}
+
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+    if unload_ok:
+        entry_data = hass.data[DOMAIN].pop(entry.entry_id)
+        vlc = entry_data[DATA_VLC]
+
+        await hass.async_add_executor_job(disconnect_vlc, vlc)
+
+    return unload_ok
+
+
+async def disconnect_vlc(vlc: Client) -> None:
+    """Disconnect from VLC."""
+    LOGGER.debug("Disconnecting from VLC")
+    try:
+        await vlc.disconnect()
+    except ConnectError as err:
+        LOGGER.warning("Connection error: %s", err)

homeassistant/components/vlc_telnet/config_flow.py

@@ -0,0 +1,159 @@
+"""Config flow for VLC media player Telnet integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from aiovlc.client import Client
+from aiovlc.exceptions import AuthError, ConnectError
+import voluptuous as vol
+
+from homeassistant import core, exceptions
+from homeassistant.config_entries import ConfigEntry, ConfigFlow
+from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_PORT
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DEFAULT_PORT, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def user_form_schema(user_input: dict[str, Any] | None) -> vol.Schema:
+    """Return user form schema."""
+    user_input = user_input or {}
+    return vol.Schema(
+        {
+            vol.Required(CONF_PASSWORD): str,
+            vol.Optional(
+                CONF_HOST, default=user_input.get(CONF_HOST, "localhost")
+            ): str,
+            vol.Optional(
+                CONF_PORT, default=user_input.get(CONF_PORT, DEFAULT_PORT)
+            ): int,
+        }
+    )
+
+
+STEP_REAUTH_DATA_SCHEMA = vol.Schema({vol.Required(CONF_PASSWORD): str})
+
+
+async def vlc_connect(vlc: Client) -> None:
+    """Connect to VLC."""
+    await vlc.connect()
+    await vlc.login()
+    await vlc.disconnect()
+
+
+async def validate_input(
+    hass: core.HomeAssistant, data: dict[str, Any]
+) -> dict[str, str]:
+    """Validate the user input allows us to connect."""
+    vlc = Client(
+        password=data[CONF_PASSWORD],
+        host=data[CONF_HOST],
+        port=data[CONF_PORT],
+    )
+
+    try:
+        await vlc_connect(vlc)
+    except ConnectError as err:
+        raise CannotConnect from err
+    except AuthError as err:
+        raise InvalidAuth from err
+
+    # CONF_NAME is only present in the imported YAML data.
+    return {"title": data.get(CONF_NAME) or data[CONF_HOST]}
+
+
+class VLCTelnetConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for VLC media player Telnet."""
+
+    VERSION = 1
+    entry: ConfigEntry | None = None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        if user_input is None:
+            return self.async_show_form(
+                step_id="user", data_schema=user_form_schema(user_input)
+            )
+
+        self._async_abort_entries_match(
+            {CONF_HOST: user_input[CONF_HOST], CONF_PORT: user_input[CONF_PORT]}
+        )
+
+        errors = {}
+
+        try:
+            info = await validate_input(self.hass, user_input)
+        except CannotConnect:
+            errors["base"] = "cannot_connect"
+        except InvalidAuth:
+            errors["base"] = "invalid_auth"
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Unexpected exception")
+            errors["base"] = "unknown"
+        else:
+            return self.async_create_entry(title=info["title"], data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=user_form_schema(user_input), errors=errors
+        )
+
+    async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
+        """Handle the import step."""
+        return await self.async_step_user(user_input)
+
+    async def async_step_reauth(self, data: dict[str, Any]) -> FlowResult:
+        """Handle reauth flow."""
+        self.entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
+        assert self.entry
+        self.context["title_placeholders"] = {"host": self.entry.data[CONF_HOST]}
+        return await self.async_step_reauth_confirm()
+
+    async def async_step_reauth_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle reauth confirm."""
+        assert self.entry
+        errors = {}
+
+        if user_input is not None:
+            try:
+                await validate_input(self.hass, {**self.entry.data, **user_input})
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                self.hass.config_entries.async_update_entry(
+                    self.entry,
+                    data={
+                        **self.entry.data,
+                        CONF_PASSWORD: user_input[CONF_PASSWORD],
+                    },
+                )
+                self.hass.async_create_task(
+                    self.hass.config_entries.async_reload(self.entry.entry_id)
+                )
+                return self.async_abort(reason="reauth_successful")
+
+        return self.async_show_form(
+            step_id="reauth_confirm",
+            description_placeholders={CONF_HOST: self.entry.data[CONF_HOST]},
+            data_schema=STEP_REAUTH_DATA_SCHEMA,
+            errors=errors,
+        )
+
+
+class CannotConnect(exceptions.HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(exceptions.HomeAssistantError):
+    """Error to indicate there is invalid auth."""

homeassistant/components/vlc_telnet/const.py

@@ -0,0 +1,9 @@
+"""Integration shared constants."""
+import logging
+
+DATA_VLC = "vlc"
+DATA_AVAILABLE = "available"
+DEFAULT_NAME = "VLC-TELNET"
+DEFAULT_PORT = 4212
+DOMAIN = "vlc_telnet"
+LOGGER = logging.getLogger(__package__)

homeassistant/components/vlc_telnet/manifest.json

@@ -1,8 +1,9 @@
 {
   "domain": "vlc_telnet",
   "name": "VLC media player Telnet",
+  "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/vlc_telnet",
-  "requirements": ["python-telnet-vlc==2.0.1"],
-  "codeowners": ["@rodripf", "@dmcc"],
+  "requirements": ["aiovlc==0.1.0"],
+  "codeowners": ["@rodripf", "@dmcc", "@MartinHjelmare"],
   "iot_class": "local_polling"
 }

homeassistant/components/vlc_telnet/media_player.py

@@ -1,13 +1,11 @@
 """Provide functionality to interact with the vlc telnet interface."""
-import logging
+from __future__ import annotations
 
-from python_telnet_vlc import (
-    CommandError,
-    ConnectionError as ConnErr,
-    LuaError,
-    ParseError,
-    VLCTelnet,
-)
+from datetime import datetime
+from typing import Any
+
+from aiovlc.client import Client
+from aiovlc.exceptions import AuthError, CommandError, ConnectError
 import voluptuous as vol
 
 from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerEntity
@@ -25,6 +23,7 @@ from homeassistant.components.media_player.const import (
     SUPPORT_VOLUME_MUTE,
     SUPPORT_VOLUME_SET,
 )
+from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
 from homeassistant.const import (
     CONF_HOST,
     CONF_NAME,
@@ -33,17 +32,15 @@ from homeassistant.const import (
     STATE_IDLE,
     STATE_PAUSED,
     STATE_PLAYING,
-    STATE_UNAVAILABLE,
 )
+from homeassistant.core import HomeAssistant
 import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 import homeassistant.util.dt as dt_util
 
-_LOGGER = logging.getLogger(__name__)
+from .const import DATA_AVAILABLE, DATA_VLC, DEFAULT_NAME, DEFAULT_PORT, DOMAIN, LOGGER
 
-DOMAIN = "vlc_telnet"
-
-DEFAULT_NAME = "VLC-TELNET"
-DEFAULT_PORT = 4212
 MAX_VOLUME = 500
 
 SUPPORT_VLC = (
@@ -69,106 +66,129 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
 )
 
 
-def setup_platform(hass, config, add_entities, discovery_info=None):
+async def async_setup_platform(
+    hass: HomeAssistant,
+    config: ConfigType,
+    async_add_entities: AddEntitiesCallback,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> None:
     """Set up the vlc platform."""
-    add_entities(
-        [
-            VlcDevice(
-                config.get(CONF_NAME),
-                config.get(CONF_HOST),
-                config.get(CONF_PORT),
-                config.get(CONF_PASSWORD),
-            )
-        ],
-        True,
+    LOGGER.warning(
+        "Loading VLC media player Telnet integration via platform setup is deprecated; "
+        "Please remove it from your configuration"
     )
+    hass.async_create_task(
+        hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_IMPORT}, data=config
+        )
+    )
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up the vlc platform."""
+    # CONF_NAME is only present in imported YAML.
+    name = entry.data.get(CONF_NAME) or DEFAULT_NAME
+    vlc = hass.data[DOMAIN][entry.entry_id][DATA_VLC]
+    available = hass.data[DOMAIN][entry.entry_id][DATA_AVAILABLE]
+
+    async_add_entities([VlcDevice(entry, vlc, name, available)], True)
 
 
 class VlcDevice(MediaPlayerEntity):
     """Representation of a vlc player."""
 
-    def __init__(self, name, host, port, passwd):
+    def __init__(
+        self, config_entry: ConfigEntry, vlc: Client, name: str, available: bool
+    ) -> None:
         """Initialize the vlc device."""
+        self._config_entry = config_entry
         self._name = name
-        self._volume = None
-        self._muted = None
-        self._state = STATE_UNAVAILABLE
-        self._media_position_updated_at = None
-        self._media_position = None
-        self._media_duration = None
-        self._host = host
-        self._port = port
-        self._password = passwd
-        self._vlc = None
-        self._available = True
-        self._volume_bkp = 0
-        self._media_artist = ""
-        self._media_title = ""
+        self._volume: float | None = None
+        self._muted: bool | None = None
+        self._state: str | None = None
+        self._media_position_updated_at: datetime | None = None
+        self._media_position: int | None = None
+        self._media_duration: int | None = None
+        self._vlc = vlc
+        self._available = available
+        self._volume_bkp = 0.0
+        self._media_artist: str | None = None
+        self._media_title: str | None = None
+        config_entry_id = config_entry.entry_id
+        self._attr_unique_id = config_entry_id
+        self._attr_device_info = {
+            "name": name,
+            "identifiers": {(DOMAIN, config_entry_id)},
+            "manufacturer": "VideoLAN",
+            "entry_type": "service",
+        }
 
-    def update(self):
+    async def async_update(self) -> None:
         """Get the latest details from the device."""
-        if self._vlc is None:
+        if not self._available:
             try:
-                self._vlc = VLCTelnet(self._host, self._password, self._port)
-            except (ConnErr, EOFError) as err:
-                if self._available:
-                    _LOGGER.error("Connection error: %s", err)
-                    self._available = False
-                self._vlc = None
+                await self._vlc.connect()
+            except ConnectError as err:
+                LOGGER.debug("Connection error: %s", err)
+                return
+
+            try:
+                await self._vlc.login()
+            except AuthError:
+                LOGGER.debug("Failed to login to VLC")
+                self.hass.async_create_task(
+                    self.hass.config_entries.async_reload(self._config_entry.entry_id)
+                )
                 return
 
             self._state = STATE_IDLE
             self._available = True
+            LOGGER.info("Connected to vlc host: %s", self._vlc.host)
 
         try:
-            status = self._vlc.status()
-            _LOGGER.debug("Status: %s", status)
+            status = await self._vlc.status()
+            LOGGER.debug("Status: %s", status)
 
-            if status:
-                if "volume" in status:
-                    self._volume = status["volume"] / MAX_VOLUME
-                else:
-                    self._volume = None
-                if "state" in status:
-                    state = status["state"]
-                    if state == "playing":
-                        self._state = STATE_PLAYING
-                    elif state == "paused":
-                        self._state = STATE_PAUSED
-                    else:
-                        self._state = STATE_IDLE
-                else:
-                    self._state = STATE_IDLE
+            self._volume = status.audio_volume / MAX_VOLUME
+            state = status.state
+            if state == "playing":
+                self._state = STATE_PLAYING
+            elif state == "paused":
+                self._state = STATE_PAUSED
+            else:
+                self._state = STATE_IDLE
 
             if self._state != STATE_IDLE:
-                self._media_duration = self._vlc.get_length()
-                vlc_position = self._vlc.get_time()
+                self._media_duration = (await self._vlc.get_length()).length
+                time_output = await self._vlc.get_time()
+                vlc_position = time_output.time
 
                 # Check if current position is stale.
                 if vlc_position != self._media_position:
                     self._media_position_updated_at = dt_util.utcnow()
                     self._media_position = vlc_position
 
-            info = self._vlc.info()
-            _LOGGER.debug("Info: %s", info)
+            info = await self._vlc.info()
+            data = info.data
+            LOGGER.debug("Info data: %s", data)
 
-            if info:
-                self._media_artist = info.get(0, {}).get("artist")
-                self._media_title = info.get(0, {}).get("title")
+            self._media_artist = data.get(0, {}).get("artist")
+            self._media_title = data.get(0, {}).get("title")
 
-                if not self._media_title:
-                    # Fall back to filename.
-                    data_info = info.get("data")
-                    if data_info:
-                        self._media_title = data_info["filename"]
+            if not self._media_title:
+                # Fall back to filename.
+                data_info = data.get("data")
+                if data_info:
+                    self._media_title = data_info["filename"]
 
-        except (CommandError, LuaError, ParseError) as err:
-            _LOGGER.error("Command error: %s", err)
-        except (ConnErr, EOFError) as err:
+        except CommandError as err:
+            LOGGER.error("Command error: %s", err)
+        except ConnectError as err:
             if self._available:
-                _LOGGER.error("Connection error: %s", err)
+                LOGGER.error("Connection error: %s", err)
                 self._available = False
-            self._vlc = None
 
     @property
     def name(self):
@@ -186,7 +206,7 @@ class VlcDevice(MediaPlayerEntity):
         return self._available
 
     @property
-    def volume_level(self):
+    def volume_level(self) -> float | None:
         """Volume level of the media player (0..1)."""
         return self._volume
 
@@ -230,72 +250,79 @@ class VlcDevice(MediaPlayerEntity):
         """Artist of current playing media, music track only."""
         return self._media_artist
 
-    def media_seek(self, position):
+    async def async_media_seek(self, position: float) -> None:
         """Seek the media to a specific location."""
-        self._vlc.seek(int(position))
+        await self._vlc.seek(round(position))
 
-    def mute_volume(self, mute):
+    async def async_mute_volume(self, mute: bool) -> None:
         """Mute the volume."""
+        assert self._volume is not None
         if mute:
             self._volume_bkp = self._volume
-            self.set_volume_level(0)
+            await self.async_set_volume_level(0)
         else:
-            self.set_volume_level(self._volume_bkp)
+            await self.async_set_volume_level(self._volume_bkp)
 
         self._muted = mute
 
-    def set_volume_level(self, volume):
+    async def async_set_volume_level(self, volume: float) -> None:
         """Set volume level, range 0..1."""
-        self._vlc.set_volume(volume * MAX_VOLUME)
+        await self._vlc.set_volume(round(volume * MAX_VOLUME))
         self._volume = volume
 
         if self._muted and self._volume > 0:
             # This can happen if we were muted and then see a volume_up.
             self._muted = False
 
-    def media_play(self):
+    async def async_media_play(self) -> None:
         """Send play command."""
-        self._vlc.play()
+        await self._vlc.play()
         self._state = STATE_PLAYING
 
-    def media_pause(self):
+    async def async_media_pause(self) -> None:
         """Send pause command."""
-        current_state = self._vlc.status().get("state")
+        status = await self._vlc.status()
+        current_state = status.state
         if current_state != "paused":
             # Make sure we're not already paused since VLCTelnet.pause() toggles
             # pause.
-            self._vlc.pause()
+            await self._vlc.pause()
+
         self._state = STATE_PAUSED
 
-    def media_stop(self):
+    async def async_media_stop(self) -> None:
         """Send stop command."""
-        self._vlc.stop()
+        await self._vlc.stop()
         self._state = STATE_IDLE
 
-    def play_media(self, media_type, media_id, **kwargs):
+    async def async_play_media(
+        self, media_type: str, media_id: str, **kwargs: Any
+    ) -> None:
         """Play media from a URL or file."""
         if media_type != MEDIA_TYPE_MUSIC:
-            _LOGGER.error(
+            LOGGER.error(
                 "Invalid media type %s. Only %s is supported",
                 media_type,
                 MEDIA_TYPE_MUSIC,
             )
             return
-        self._vlc.add(media_id)
+
+        await self._vlc.add(media_id)
         self._state = STATE_PLAYING
 
-    def media_previous_track(self):
+    async def async_media_previous_track(self) -> None:
         """Send previous track command."""
-        self._vlc.prev()
+        await self._vlc.prev()
 
-    def media_next_track(self):
+    async def async_media_next_track(self) -> None:
         """Send next track command."""
-        self._vlc.next()
+        await self._vlc.next()
 
-    def clear_playlist(self):
+    async def async_clear_playlist(self) -> None:
         """Clear players playlist."""
-        self._vlc.clear()
+        await self._vlc.clear()
 
-    def set_shuffle(self, shuffle):
+    async def async_set_shuffle(self, shuffle: bool) -> None:
         """Enable/disable shuffle mode."""
-        self._vlc.random(shuffle)
+        shuffle_command = "on" if shuffle else "off"
+        await self._vlc.random(shuffle_command)

homeassistant/components/vlc_telnet/strings.json

@@ -0,0 +1,30 @@
+{
+  "config": {
+    "flow_title": "{host}",
+    "step": {
+      "reauth_confirm": {
+        "description": "Please enter the correct password for host: {host}",
+        "data": {
+          "password": "[%key:common::config_flow::data::password%]"
+        }
+      },
+      "user": {
+        "data": {
+          "host": "[%key:common::config_flow::data::host%]",
+          "port": "[%key:common::config_flow::data::port%]",
+          "password": "[%key:common::config_flow::data::password%]",
+          "name": "[%key:common::config_flow::data::name%]"
+        }
+      }
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
+      "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    }
+  }
+}

homeassistant/components/vlc_telnet/translations/en.json

@@ -0,0 +1,30 @@
+{
+    "config": {
+        "abort": {
+            "already_configured": "Service is already configured",
+            "reauth_successful": "Re-authentication was successful"
+        },
+        "error": {
+            "cannot_connect": "Failed to connect",
+            "invalid_auth": "Invalid authentication",
+            "unknown": "Unexpected error"
+        },
+        "flow_title": "{host}",
+        "step": {
+            "reauth_confirm": {
+                "data": {
+                    "password": "Password"
+                },
+                "description": "Please enter the correct password for host: {host}"
+            },
+            "user": {
+                "data": {
+                    "host": "Host",
+                    "name": "Name",
+                    "password": "Password",
+                    "port": "Port"
+                }
+            }
+        }
+    }
+}

homeassistant/generated/config_flows.py

@@ -309,6 +309,7 @@ FLOWS = [
     "vesync",
     "vilfo",
     "vizio",
+    "vlc_telnet",
     "volumio",
     "wallbox",
     "watttime",

requirements_all.txt

@@ -254,6 +254,9 @@ aiotractive==0.5.2
 # homeassistant.components.unifi
 aiounifi==27
 
+# homeassistant.components.vlc_telnet
+aiovlc==0.1.0
+
 # homeassistant.components.watttime
 aiowatttime==0.1.1
 
@@ -1930,9 +1933,6 @@ python-tado==0.12.0
 # homeassistant.components.telegram_bot
 python-telegram-bot==13.1
 
-# homeassistant.components.vlc_telnet
-python-telnet-vlc==2.0.1
-
 # homeassistant.components.twitch
 python-twitch-client==0.6.0
 

requirements_test_all.txt

@@ -181,6 +181,9 @@ aiotractive==0.5.2
 # homeassistant.components.unifi
 aiounifi==27
 
+# homeassistant.components.vlc_telnet
+aiovlc==0.1.0
+
 # homeassistant.components.watttime
 aiowatttime==0.1.1
 

tests/components/vlc_telnet/__init__.py

@@ -0,0 +1 @@
+"""Test the VLC media player Telnet integration."""

tests/components/vlc_telnet/test_config_flow.py

@@ -0,0 +1,272 @@
+"""Test the VLC media player Telnet config flow."""
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import patch
+
+from aiovlc.exceptions import AuthError, ConnectError
+import pytest
+
+from homeassistant import config_entries
+from homeassistant.components.vlc_telnet.const import DOMAIN
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+# mypy: allow-untyped-calls
+
+
+@pytest.mark.parametrize(
+    "input_data, entry_data",
+    [
+        (
+            {
+                "password": "test-password",
+                "host": "1.1.1.1",
+                "port": 8888,
+            },
+            {
+                "password": "test-password",
+                "host": "1.1.1.1",
+                "port": 8888,
+            },
+        ),
+        (
+            {
+                "password": "test-password",
+            },
+            {
+                "password": "test-password",
+                "host": "localhost",
+                "port": 4212,
+            },
+        ),
+    ],
+)
+async def test_user_flow(
+    hass: HomeAssistant, input_data: dict[str, Any], entry_data: dict[str, Any]
+) -> None:
+    """Test successful user flow."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["errors"] is None
+
+    with patch("homeassistant.components.vlc_telnet.config_flow.Client.connect"), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.login"
+    ), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.disconnect"
+    ), patch(
+        "homeassistant.components.vlc_telnet.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            input_data,
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == entry_data["host"]
+    assert result["data"] == entry_data
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_import_flow(hass: HomeAssistant) -> None:
+    """Test successful import flow."""
+    with patch("homeassistant.components.vlc_telnet.config_flow.Client.connect"), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.login"
+    ), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.disconnect"
+    ), patch(
+        "homeassistant.components.vlc_telnet.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_IMPORT},
+            data={
+                "password": "test-password",
+                "host": "1.1.1.1",
+                "port": 8888,
+                "name": "custom name",
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == "custom name"
+    assert result["data"] == {
+        "password": "test-password",
+        "host": "1.1.1.1",
+        "port": 8888,
+        "name": "custom name",
+    }
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+@pytest.mark.parametrize(
+    "source", [config_entries.SOURCE_USER, config_entries.SOURCE_IMPORT]
+)
+async def test_abort_already_configured(hass: HomeAssistant, source: str) -> None:
+    """Test we handle already configured host."""
+    entry_data = {
+        "password": "test-password",
+        "host": "1.1.1.1",
+        "port": 8888,
+        "name": "custom name",
+    }
+
+    entry = MockConfigEntry(domain=DOMAIN, data=entry_data)
+    entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": source},
+        data=entry_data,
+    )
+
+    assert result["type"] == "abort"
+    assert result["reason"] == "already_configured"
+
+
+@pytest.mark.parametrize(
+    "source", [config_entries.SOURCE_USER, config_entries.SOURCE_IMPORT]
+)
+@pytest.mark.parametrize(
+    "error, connect_side_effect, login_side_effect",
+    [
+        ("invalid_auth", None, AuthError),
+        ("cannot_connect", ConnectError, None),
+        ("unknown", Exception, None),
+    ],
+)
+async def test_errors(
+    hass: HomeAssistant,
+    error: str,
+    connect_side_effect: Exception | None,
+    login_side_effect: Exception | None,
+    source: str,
+) -> None:
+    """Test we handle form errors."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": source}
+    )
+
+    with patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.connect",
+        side_effect=connect_side_effect,
+    ), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.login",
+        side_effect=login_side_effect,
+    ), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.disconnect"
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"password": "test-password"},
+        )
+
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"base": error}
+
+
+async def test_reauth_flow(hass: HomeAssistant) -> None:
+    """Test successful reauth flow."""
+    entry_data = {
+        "password": "old-password",
+        "host": "1.1.1.1",
+        "port": 8888,
+        "name": "custom name",
+    }
+
+    entry = MockConfigEntry(domain=DOMAIN, data=entry_data)
+    entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={
+            "source": config_entries.SOURCE_REAUTH,
+            "entry_id": entry.entry_id,
+            "unique_id": entry.unique_id,
+        },
+        data=entry_data,
+    )
+
+    with patch("homeassistant.components.vlc_telnet.config_flow.Client.connect"), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.login"
+    ), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.disconnect"
+    ), patch(
+        "homeassistant.components.vlc_telnet.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"password": "new-password"},
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == "abort"
+    assert result["reason"] == "reauth_successful"
+    assert len(mock_setup_entry.mock_calls) == 1
+    assert dict(entry.data) == {
+        "password": "new-password",
+        "host": "1.1.1.1",
+        "port": 8888,
+        "name": "custom name",
+    }
+
+
+@pytest.mark.parametrize(
+    "error, connect_side_effect, login_side_effect",
+    [
+        ("invalid_auth", None, AuthError),
+        ("cannot_connect", ConnectError, None),
+        ("unknown", Exception, None),
+    ],
+)
+async def test_reauth_errors(
+    hass: HomeAssistant,
+    error: str,
+    connect_side_effect: Exception | None,
+    login_side_effect: Exception | None,
+) -> None:
+    """Test we handle reauth errors."""
+    entry_data = {
+        "password": "old-password",
+        "host": "1.1.1.1",
+        "port": 8888,
+        "name": "custom name",
+    }
+
+    entry = MockConfigEntry(domain=DOMAIN, data=entry_data)
+    entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={
+            "source": config_entries.SOURCE_REAUTH,
+            "entry_id": entry.entry_id,
+            "unique_id": entry.unique_id,
+        },
+        data=entry_data,
+    )
+
+    with patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.connect",
+        side_effect=connect_side_effect,
+    ), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.login",
+        side_effect=login_side_effect,
+    ), patch(
+        "homeassistant.components.vlc_telnet.config_flow.Client.disconnect"
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"password": "test-password"},
+        )
+
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"base": error}