starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-28 02:24:46 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add certificate expiration sensor to Gatus integration (#178660)

Browse Source
This commit is contained in:
Hamish
2026-08-26 19:29:26 +02:00
committed by GitHub
parent a77d8d05b3
commit bca346df5f
6 changed files with 111 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/gatus/coordinator.py
+3
View File
@@ -11,6 +11,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from .const import DOMAIN
@@ -27,6 +28,7 @@ class GatusDataUpdateCoordinator(DataUpdateCoordinator[dict[str, EndpointStatus]
self.url = url.rstrip("/")
self.client = GatusClient(url=self.url, session=async_get_clientsession(hass))
self._entry_id = entry.entry_id
self.last_update_time = dt_util.utcnow().replace(second=0, microsecond=0)
device_registry = dr.async_get(hass)
self._known_endpoint_keys = {
identifier[1].removeprefix(f"{entry.entry_id}_")
@@ -49,6 +51,7 @@ class GatusDataUpdateCoordinator(DataUpdateCoordinator[dict[str, EndpointStatus]
@override
async def _async_update_data(self) -> dict[str, EndpointStatus]:
"""Fetch endpoint statuses from the Gatus API."""
self.last_update_time = dt_util.utcnow().replace(second=0, microsecond=0)
try:
raw_endpoints = await self.client.get_endpoints_statuses()
except GatusClientError as err:
homeassistant/components/gatus/sensor.py
+31 -7
View File
@@ -2,6 +2,7 @@
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import override
from gatus_api import EndpointStatus
@@ -26,7 +27,10 @@ PARALLEL_UPDATES = 0
class GatusSensorEntityDescription(SensorEntityDescription):
"""Class describing Gatus sensor entities."""
value_fn: Callable[[EndpointStatus], float | int | str | None]
value_fn: Callable[
[GatusDataUpdateCoordinator, EndpointStatus],
datetime | float | int | str | None,
]
SENSOR_TYPES: tuple[GatusSensorEntityDescription, ...] = (
@@ -36,7 +40,7 @@ SENSOR_TYPES: tuple[GatusSensorEntityDescription, ...] = (
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.MILLISECONDS,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda endpoint: (
value_fn=lambda coordinator, endpoint: (
round(endpoint.results[-1].duration / 1_000_000, 2)
if endpoint.results and endpoint.results[-1].duration is not None
else None
@@ -46,7 +50,7 @@ SENSOR_TYPES: tuple[GatusSensorEntityDescription, ...] = (
key="status_code",
translation_key="status_code",
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda endpoint: (
value_fn=lambda coordinator, endpoint: (
endpoint.results[-1].status if endpoint.results else None
),
),
@@ -56,10 +60,25 @@ SENSOR_TYPES: tuple[GatusSensorEntityDescription, ...] = (
device_class=SensorDeviceClass.ENUM,
options=["start", "healthy", "unhealthy", "resolved"],
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda endpoint: (
value_fn=lambda coordinator, endpoint: (
endpoint.events[-1].type.lower() if endpoint.events else None
),
),
GatusSensorEntityDescription(
key="certificate_expiration",
translation_key="certificate_expiration",
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda coordinator, endpoint: (
coordinator.last_update_time
+ timedelta(
seconds=endpoint.results[-1].certificate_expiration // 1_000_000_000
)
if endpoint.results
and endpoint.results[-1].certificate_expiration is not None
else None
),
),
)
@@ -73,8 +92,12 @@ async def async_setup_entry(
async_add_entities(
GatusEndpointSensor(coordinator, entry, endpoint_key, description)
for endpoint_key in coordinator.data
for endpoint_key, endpoint in coordinator.data.items()
for description in SENSOR_TYPES
if description.key != "certificate_expiration"
or (
endpoint.results and endpoint.results[-1].certificate_expiration is not None
)
)
@@ -93,10 +116,11 @@ class GatusEndpointSensor(GatusEndpointEntity, SensorEntity):
"""Initialize the sensor."""
super().__init__(coordinator, entry, endpoint_key)
self.entity_description = description
self._attr_translation_key = description.translation_key
self._attr_unique_id = f"{entry.entry_id}_{endpoint_key}_{description.key}"
@property
@override
def native_value(self) -> float | int | str | None:
def native_value(self) -> datetime | float | int | str | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.endpoint_data)
return self.entity_description.value_fn(self.coordinator, self.endpoint_data)
homeassistant/components/gatus/strings.json
+3
View File
@@ -31,6 +31,9 @@
},
"entity": {
"sensor": {
"certificate_expiration": {
"name": "Certificate expiration"
},
"last_event": {
"name": "Last event",
"state": {
tests/components/gatus/conftest.py
+8 -1
View File
@@ -41,7 +41,14 @@ def mock_gatus_client() -> Generator[AsyncMock]:
key="backend_service",
name="Backend Service",
group="Core",
results=[Result(success=True, status=200, duration=23123100)],
results=[
Result(
success=True,
status=200,
duration=23123100,
certificate_expiration=7776000000000000,
)
],
events=[Event(type="HEALTHY", timestamp="2026-01-01T00:00:00Z")],
)
]
tests/components/gatus/snapshots/test_sensor.ambr
+51
View File
@@ -1,4 +1,55 @@
# serializer version: 1
# name: test_sensor_setup_and_states[sensor.core_backend_service_certificate_expiration-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.core_backend_service_certificate_expiration',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Certificate expiration',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Certificate expiration',
'platform': 'gatus',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'certificate_expiration',
'unique_id': '1234567890abcdef1234567890abcdef_backend_service_certificate_expiration',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_setup_and_states[sensor.core_backend_service_certificate_expiration-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'timestamp',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Core Backend Service Certificate expiration',
}),
'context': <ANY>,
'entity_id': 'sensor.core_backend_service_certificate_expiration',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2026-04-01T00:00:00+00:00',
})
# ---
# name: test_sensor_setup_and_states[sensor.core_backend_service_last_event-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
tests/components/gatus/test_sensor.py
+15 -7
View File
@@ -27,8 +27,10 @@ async def test_sensor_setup_and_states(
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test standard successful setup and entity snapshots using snapshot_platform."""
freezer.move_to("2026-01-01 00:00:00+00:00")
with patch("homeassistant.components.gatus._PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(
@@ -47,6 +49,7 @@ def _to_endpoint_statuses(raw_data: list[dict[str, Any]]) -> list[EndpointStatus
success=r["success"],
status=r.get("status"),
duration=r.get("duration"),
certificate_expiration=r.get("certificateExpiration"),
)
for r in ep.get("results", [])
],
@@ -167,24 +170,29 @@ async def test_sensor_missing_status_code(
assert state.state == STATE_UNKNOWN
async def test_sensor_missing_last_event(
async def test_sensor_missing_certificate_expiration(
hass: HomeAssistant,
mock_gatus_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that an endpoint missing events evaluates to STATE_UNKNOWN for last_event sensor."""
"""Test that a result missing certificate expiration creates no entity."""
mock_gatus_client.get_endpoints_statuses.return_value = [
EndpointStatus(
key="backend_service",
name="Backend Service",
group=None,
results=[Result(success=True, status=200, duration=12500000)],
events=[],
results=[
Result(
success=True,
status=200,
duration=12500000,
certificate_expiration=None,
)
],
)
]
await setup_integration(hass, mock_config_entry)
state = hass.states.get("sensor.backend_service_last_event")
assert state is not None
assert state.state == STATE_UNKNOWN
state = hass.states.get("sensor.backend_service_certificate_expiration")
assert state is None