Add certificate expiration sensor to Gatus integration

homeassistant/components/gatus/coordinator.py

@@ -11,6 +11,7 @@
 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 @@ def __init__(self, hass: HomeAssistant, entry: GatusConfigEntry, url: str) -> No
         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 @@ def __init__(self, hass: HomeAssistant, entry: GatusConfigEntry, url: str) -> No
     @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

@@ -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 @@
 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 @@ class GatusSensorEntityDescription(SensorEntityDescription):
         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 @@ class GatusSensorEntityDescription(SensorEntityDescription):
         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 @@ class GatusSensorEntityDescription(SensorEntityDescription):
         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 @@ def __init__(
         """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

@@ -31,6 +31,9 @@
   },
   "entity": {
     "sensor": {
+      "certificate_expiration": {
+        "name": "Certificate expiration"
+      },
       "last_event": {
         "name": "Last event",
         "state": {

tests/components/gatus/conftest.py

@@ -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

@@ -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

@@ -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