config_flow.py

Go to the documentation of this file.

"""Config flow for the Cert Expiry platform."""
 
from __future__ import annotations
 
from collections.abc import Mapping
import logging
from typing import Any
 
import voluptuous as vol
 
from homeassistant.config_entries import SOURCE_IMPORT, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PORT
 
from .const import DEFAULT_PORT, DOMAIN
from .errors import (
    ConnectionRefused,
    ConnectionTimeout,
    ResolveFailed,
    ValidationFailure,
)
from .helper import get_cert_expiry_timestamp
 
_LOGGER = logging.getLogger(__name__)
 
 
class CertexpiryConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow."""
 
    VERSION = 1
 
    def __init__(self) -> None:
        """Initialize the config flow."""
        self._errors_errors: dict[str, str] = {}
 
    async def _test_connection(
        self,
        user_input: Mapping[str, Any],
    ) -> bool:
        """Test connection to the server and try to get the certificate."""
        try:
            await get_cert_expiry_timestamp(
                self.hass,
                user_input[CONF_HOST],
                user_input.get(CONF_PORT, DEFAULT_PORT),
            )
        except ResolveFailed:
            self._errors_errors[CONF_HOST] = "resolve_failed"
        except ConnectionTimeout:
            self._errors_errors[CONF_HOST] = "connection_timeout"
        except ConnectionRefused:
            self._errors_errors[CONF_HOST] = "connection_refused"
        except ValidationFailure:
            return True
        else:
            return True
        return False
 
    async def async_step_user(
        self,
        user_input: Mapping[str, Any] | None = None,
    ) -> ConfigFlowResult:
        """Step when user initializes a integration."""
        self._errors_errors = {}
        if user_input is not None:
            host = user_input[CONF_HOST]
            port = user_input.get(CONF_PORT, DEFAULT_PORT)
            await self.async_set_unique_idasync_set_unique_id(f"{host}:{port}")
            self._abort_if_unique_id_configured_abort_if_unique_id_configured()
 
            if await self._test_connection_test_connection(user_input):
                title_port = f":{port}" if port != DEFAULT_PORT else ""
                title = f"{host}{title_port}"
                return self.async_create_entryasync_create_entryasync_create_entry(
                    title=title,
                    data={CONF_HOST: host, CONF_PORT: port},
                )
            if self.context["source"] == SOURCE_IMPORT:
                _LOGGER.error("Config import failed for %s", user_input[CONF_HOST])
                return self.async_abortasync_abortasync_abort(reason="import_failed")
        else:
            user_input = {}
            user_input[CONF_HOST] = ""
            user_input[CONF_PORT] = DEFAULT_PORT
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_HOST, default=user_input[CONF_HOST]): str,
                    vol.Required(
                        CONF_PORT, default=user_input.get(CONF_PORT, DEFAULT_PORT)
                    ): int,
                }
            ),
            errors=self._errors_errors,
        )
 
    async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
        """Import a config entry.
 
        Only host was required in the yaml file all other fields are optional
        """
        return await self.async_step_userasync_step_userasync_step_user(import_data)