Add typings to Certificate Expiry integration (#75945)
This commit is contained in:
parent
abb7495ced
commit
1204b4f700
3 changed files with 33 additions and 7 deletions
|
@ -1,12 +1,15 @@
|
||||||
"""Config flow for the Cert Expiry platform."""
|
"""Config flow for the Cert Expiry platform."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Mapping
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||||
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
|
|
||||||
from .const import DEFAULT_PORT, DOMAIN
|
from .const import DEFAULT_PORT, DOMAIN
|
||||||
from .errors import (
|
from .errors import (
|
||||||
|
@ -29,7 +32,10 @@ class CertexpiryConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
"""Initialize the config flow."""
|
"""Initialize the config flow."""
|
||||||
self._errors: dict[str, str] = {}
|
self._errors: dict[str, str] = {}
|
||||||
|
|
||||||
async def _test_connection(self, user_input=None):
|
async def _test_connection(
|
||||||
|
self,
|
||||||
|
user_input: Mapping[str, Any],
|
||||||
|
):
|
||||||
"""Test connection to the server and try to get the certificate."""
|
"""Test connection to the server and try to get the certificate."""
|
||||||
try:
|
try:
|
||||||
await get_cert_expiry_timestamp(
|
await get_cert_expiry_timestamp(
|
||||||
|
@ -48,7 +54,10 @@ class CertexpiryConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
|
|
||||||
async def async_step_user(self, user_input=None):
|
async def async_step_user(
|
||||||
|
self,
|
||||||
|
user_input: Mapping[str, Any] | None = None,
|
||||||
|
) -> FlowResult:
|
||||||
"""Step when user initializes a integration."""
|
"""Step when user initializes a integration."""
|
||||||
self._errors = {}
|
self._errors = {}
|
||||||
if user_input is not None:
|
if user_input is not None:
|
||||||
|
@ -85,7 +94,10 @@ class CertexpiryConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
errors=self._errors,
|
errors=self._errors,
|
||||||
)
|
)
|
||||||
|
|
||||||
async def async_step_import(self, user_input=None):
|
async def async_step_import(
|
||||||
|
self,
|
||||||
|
user_input: Mapping[str, Any] | None = None,
|
||||||
|
) -> FlowResult:
|
||||||
"""Import a config entry.
|
"""Import a config entry.
|
||||||
|
|
||||||
Only host was required in the yaml file all other fields are optional
|
Only host was required in the yaml file all other fields are optional
|
||||||
|
|
|
@ -2,6 +2,7 @@
|
||||||
import socket
|
import socket
|
||||||
import ssl
|
import ssl
|
||||||
|
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.util import dt
|
from homeassistant.util import dt
|
||||||
|
|
||||||
from .const import TIMEOUT
|
from .const import TIMEOUT
|
||||||
|
@ -13,7 +14,10 @@ from .errors import (
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def get_cert(host, port):
|
def get_cert(
|
||||||
|
host: str,
|
||||||
|
port: int,
|
||||||
|
):
|
||||||
"""Get the certificate for the host and port combination."""
|
"""Get the certificate for the host and port combination."""
|
||||||
ctx = ssl.create_default_context()
|
ctx = ssl.create_default_context()
|
||||||
address = (host, port)
|
address = (host, port)
|
||||||
|
@ -23,7 +27,11 @@ def get_cert(host, port):
|
||||||
return cert
|
return cert
|
||||||
|
|
||||||
|
|
||||||
async def get_cert_expiry_timestamp(hass, hostname, port):
|
async def get_cert_expiry_timestamp(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
hostname: str,
|
||||||
|
port: int,
|
||||||
|
):
|
||||||
"""Return the certificate's expiration timestamp."""
|
"""Return the certificate's expiration timestamp."""
|
||||||
try:
|
try:
|
||||||
cert = await hass.async_add_executor_job(get_cert, hostname, port)
|
cert = await hass.async_add_executor_job(get_cert, hostname, port)
|
||||||
|
|
|
@ -19,6 +19,7 @@ from homeassistant.helpers.event import async_call_later
|
||||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||||
|
|
||||||
|
from . import CertExpiryDataUpdateCoordinator
|
||||||
from .const import DEFAULT_PORT, DOMAIN
|
from .const import DEFAULT_PORT, DOMAIN
|
||||||
|
|
||||||
SCAN_INTERVAL = timedelta(hours=12)
|
SCAN_INTERVAL = timedelta(hours=12)
|
||||||
|
@ -57,7 +58,9 @@ async def async_setup_platform(
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(
|
async def async_setup_entry(
|
||||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
hass: HomeAssistant,
|
||||||
|
entry: ConfigEntry,
|
||||||
|
async_add_entities: AddEntitiesCallback,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Add cert-expiry entry."""
|
"""Add cert-expiry entry."""
|
||||||
coordinator = hass.data[DOMAIN][entry.entry_id]
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
||||||
|
@ -88,7 +91,10 @@ class SSLCertificateTimestamp(CertExpiryEntity, SensorEntity):
|
||||||
|
|
||||||
_attr_device_class = SensorDeviceClass.TIMESTAMP
|
_attr_device_class = SensorDeviceClass.TIMESTAMP
|
||||||
|
|
||||||
def __init__(self, coordinator) -> None:
|
def __init__(
|
||||||
|
self,
|
||||||
|
coordinator: CertExpiryDataUpdateCoordinator,
|
||||||
|
) -> None:
|
||||||
"""Initialize a Cert Expiry timestamp sensor."""
|
"""Initialize a Cert Expiry timestamp sensor."""
|
||||||
super().__init__(coordinator)
|
super().__init__(coordinator)
|
||||||
self._attr_name = f"Cert Expiry Timestamp ({coordinator.name})"
|
self._attr_name = f"Cert Expiry Timestamp ({coordinator.name})"
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue