core/homeassistant/components/cync/config_flow.py

"""Config flow for the Cync integration."""

from collections.abc import Mapping
import logging
from typing import Any

from pycync import Auth
from pycync.exceptions import AuthFailedError, CyncError, TwoFactorRequiredError
import voluptuous as vol

from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_EMAIL, CONF_PASSWORD
from homeassistant.helpers.aiohttp_client import async_get_clientsession

from .const import (
    CONF_AUTHORIZE_STRING,
    CONF_EXPIRES_AT,
    CONF_REFRESH_TOKEN,
    CONF_TWO_FACTOR_CODE,
    CONF_USER_ID,
    DOMAIN,
)

_LOGGER = logging.getLogger(__name__)

STEP_USER_DATA_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_EMAIL): str,
        vol.Required(CONF_PASSWORD): str,
    }
)

STEP_TWO_FACTOR_SCHEMA = vol.Schema({vol.Required(CONF_TWO_FACTOR_CODE): str})


class CyncConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Cync."""

    VERSION = 1

    cync_auth: Auth = None

    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Attempt login with user credentials."""
        errors: dict[str, str] = {}

        if user_input:
            try:
                errors = await self._validate_credentials(user_input)
            except TwoFactorRequiredError:
                return await self.async_step_two_factor()

            if not errors:
                return await self._create_config_entry(self.cync_auth.username)

        return self.async_show_form(
            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
        )

    async def async_step_two_factor(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Attempt login with the two factor auth code sent to the user."""
        errors: dict[str, str] = {}

        if user_input:
            errors = await self._validate_credentials(user_input)

            if not errors:
                return await self._create_config_entry(self.cync_auth.username)

            return self.async_show_form(
                step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
            )

        return self.async_show_form(
            step_id="two_factor", data_schema=STEP_TWO_FACTOR_SCHEMA, errors=errors
        )

    async def async_step_reauth(
        self, entry_data: Mapping[str, Any]
    ) -> ConfigFlowResult:
        """Perform reauth upon an API authentication error."""
        return await self.async_step_reauth_confirm()

    async def async_step_reauth_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Inform the user that reauth is required and prompt for Cync credentials."""
        errors: dict[str, str] = {}

        reauth_entry = self._get_reauth_entry()

        if user_input:
            try:
                errors = await self._validate_credentials(user_input)
            except TwoFactorRequiredError:
                return await self.async_step_two_factor()

            if not errors:
                return await self._create_config_entry(self.cync_auth.username)

        return self.async_show_form(
            step_id="reauth_confirm",
            data_schema=STEP_USER_DATA_SCHEMA,
            errors=errors,
            description_placeholders={CONF_EMAIL: reauth_entry.title},
        )

    async def _validate_credentials(self, user_input: dict[str, Any]) -> dict[str, str]:
        """Attempt to log in with user email and password, and return the error dict."""
        errors: dict[str, str] = {}

        if not self.cync_auth:
            self.cync_auth = Auth(
                async_get_clientsession(self.hass),
                username=user_input[CONF_EMAIL],
                password=user_input[CONF_PASSWORD],
            )

        try:
            await self.cync_auth.login(user_input.get(CONF_TWO_FACTOR_CODE))
        except TwoFactorRequiredError:
            raise
        except AuthFailedError:
            errors["base"] = "invalid_auth"
        except CyncError:
            errors["base"] = "cannot_connect"
        except Exception:
            _LOGGER.exception("Unexpected exception")
            errors["base"] = "unknown"

        return errors

    async def _create_config_entry(self, user_email: str) -> ConfigFlowResult:
        """Create the Cync config entry using input user data."""

        cync_user = self.cync_auth.user
        await self.async_set_unique_id(str(cync_user.user_id))

        config_data = {
            CONF_USER_ID: cync_user.user_id,
            CONF_AUTHORIZE_STRING: cync_user.authorize,
            CONF_EXPIRES_AT: cync_user.expires_at,
            CONF_ACCESS_TOKEN: cync_user.access_token,
            CONF_REFRESH_TOKEN: cync_user.refresh_token,
        }

        if self.source == SOURCE_REAUTH:
            self._abort_if_unique_id_mismatch()
            return self.async_update_reload_and_abort(
                entry=self._get_reauth_entry(), title=user_email, data=config_data
            )

        self._abort_if_unique_id_configured()

        return self.async_create_entry(title=user_email, data=config_data)