evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Move cert_expiry coordinator to its own file (#100472)

Browse source 
* Move cert_expiry coordinator to its own file

* Add missing patched config flow test
This commit is contained in:
Jan-Philipp Benecke 2023-09-16 11:19:05 +02:00 • committed by GitHub
parent c504ca906d
commit 024db6dadf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 66 additions and 58 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

47
homeassistant/components/cert_expiry/__init__.py
View file

@ -1,22 +1,13 @@
"""The cert_expiry component."""
from __future__ import annotations
from datetime import datetime, timedelta
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.start import async_at_started
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DEFAULT_PORT, DOMAIN
from .errors import TemporaryFailure, ValidationFailure
from .helper import get_cert_expiry_timestamp
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(hours=12)
from .const import DOMAIN
from .coordinator import CertExpiryDataUpdateCoordinator
PLATFORMS = [Platform.SENSOR]
@ -45,37 +36,3 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
class CertExpiryDataUpdateCoordinator(DataUpdateCoordinator[datetime | None]):
"""Class to manage fetching Cert Expiry data from single endpoint."""
def __init__(self, hass, host, port):
"""Initialize global Cert Expiry data updater."""
self.host = host
self.port = port
self.cert_error = None
self.is_cert_valid = False
display_port = f":{port}" if port != DEFAULT_PORT else ""
name = f"{self.host}{display_port}"
super().__init__(
hass, _LOGGER, name=name, update_interval=SCAN_INTERVAL, always_update=False
)
async def _async_update_data(self) -> datetime | None:
"""Fetch certificate."""
try:
timestamp = await get_cert_expiry_timestamp(self.hass, self.host, self.port)
except TemporaryFailure as err:
raise UpdateFailed(err.args[0]) from err
except ValidationFailure as err:
self.cert_error = err
self.is_cert_valid = False
_LOGGER.error("Certificate validation error: %s [%s]", self.host, err)
return None
self.cert_error = None
self.is_cert_valid = True
return timestamp

51
homeassistant/components/cert_expiry/coordinator.py Normal file
View file

@ -0,0 +1,51 @@
"""DataUpdateCoordinator for cert_expiry coordinator."""
from __future__ import annotations
from datetime import datetime, timedelta
import logging
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DEFAULT_PORT
from .errors import TemporaryFailure, ValidationFailure
from .helper import get_cert_expiry_timestamp
_LOGGER = logging.getLogger(__name__)
class CertExpiryDataUpdateCoordinator(DataUpdateCoordinator[datetime | None]):
"""Class to manage fetching Cert Expiry data from single endpoint."""
def __init__(self, hass, host, port):
"""Initialize global Cert Expiry data updater."""
self.host = host
self.port = port
self.cert_error = None
self.is_cert_valid = False
display_port = f":{port}" if port != DEFAULT_PORT else ""
name = f"{self.host}{display_port}"
super().__init__(
hass,
_LOGGER,
name=name,
update_interval=timedelta(hours=12),
always_update=False,
)
async def _async_update_data(self) -> datetime | None:
"""Fetch certificate."""
try:
timestamp = await get_cert_expiry_timestamp(self.hass, self.host, self.port)
except TemporaryFailure as err:
raise UpdateFailed(err.args[0]) from err
except ValidationFailure as err:
self.cert_error = err
self.is_cert_valid = False
_LOGGER.error("Certificate validation error: %s [%s]", self.host, err)
return None
self.cert_error = None
self.is_cert_valid = True
return timestamp

8
tests/components/cert_expiry/test_config_flow.py
View file

@ -67,7 +67,7 @@ async def test_import_host_only(hass: HomeAssistant) -> None:
with patch(
"homeassistant.components.cert_expiry.config_flow.get_cert_expiry_timestamp"
), patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=future_timestamp(1),
):
result = await hass.config_entries.flow.async_init(
@ -89,7 +89,7 @@ async def test_import_host_and_port(hass: HomeAssistant) -> None:
with patch(
"homeassistant.components.cert_expiry.config_flow.get_cert_expiry_timestamp"
), patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=future_timestamp(1),
):
result = await hass.config_entries.flow.async_init(
@ -111,7 +111,7 @@ async def test_import_non_default_port(hass: HomeAssistant) -> None:
with patch(
"homeassistant.components.cert_expiry.config_flow.get_cert_expiry_timestamp"
), patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=future_timestamp(1),
):
result = await hass.config_entries.flow.async_init(
@ -133,7 +133,7 @@ async def test_import_with_name(hass: HomeAssistant) -> None:
with patch(
"homeassistant.components.cert_expiry.config_flow.get_cert_expiry_timestamp"
), patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=future_timestamp(1),
):
result = await hass.config_entries.flow.async_init(

8
tests/components/cert_expiry/test_init.py
View file

@ -42,7 +42,7 @@ async def test_setup_with_config(hass: HomeAssistant) -> None:
with patch(
"homeassistant.components.cert_expiry.config_flow.get_cert_expiry_timestamp"
), patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=future_timestamp(1),
):
await hass.async_block_till_done()
@ -63,7 +63,7 @@ async def test_update_unique_id(hass: HomeAssistant) -> None:
assert not entry.unique_id
with patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=future_timestamp(1),
):
assert await async_setup_component(hass, DOMAIN, {}) is True
@ -91,7 +91,7 @@ async def test_unload_config_entry(mock_now, hass: HomeAssistant) -> None:
timestamp = future_timestamp(100)
with patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=timestamp,
):
assert await async_setup_component(hass, DOMAIN, {}) is True
@ -134,7 +134,7 @@ async def test_delay_load_during_startup(hass: HomeAssistant) -> None:
timestamp = future_timestamp(100)
with patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=timestamp,
):
await hass.async_start()

10
tests/components/cert_expiry/test_sensors.py
View file

@ -29,7 +29,7 @@ async def test_async_setup_entry(mock_now, hass: HomeAssistant) -> None:
timestamp = future_timestamp(100)
with patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=timestamp,
):
entry.add_to_hass(hass)
@ -83,7 +83,7 @@ async def test_update_sensor(hass: HomeAssistant) -> None:
timestamp = future_timestamp(100)
with patch("homeassistant.util.dt.utcnow", return_value=starting_time), patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=timestamp,
):
entry.add_to_hass(hass)
@ -99,7 +99,7 @@ async def test_update_sensor(hass: HomeAssistant) -> None:
next_update = starting_time + timedelta(hours=24)
with patch("homeassistant.util.dt.utcnow", return_value=next_update), patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=timestamp,
):
async_fire_time_changed(hass, utcnow() + timedelta(hours=24))
@ -127,7 +127,7 @@ async def test_update_sensor_network_errors(hass: HomeAssistant) -> None:
timestamp = future_timestamp(100)
with patch("homeassistant.util.dt.utcnow", return_value=starting_time), patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=timestamp,
):
entry.add_to_hass(hass)
@ -156,7 +156,7 @@ async def test_update_sensor_network_errors(hass: HomeAssistant) -> None:
assert state.state == STATE_UNAVAILABLE
with patch("homeassistant.util.dt.utcnow", return_value=next_update), patch(
"homeassistant.components.cert_expiry.get_cert_expiry_timestamp",
"homeassistant.components.cert_expiry.coordinator.get_cert_expiry_timestamp",
return_value=timestamp,
):
async_fire_time_changed(hass, utcnow() + timedelta(hours=48))