Split homeassistant_alerts constants and coordinator (#117475)

homeassistant/components/homeassistant_alerts/__init__.py

@@ -2,15 +2,9 @@
 
 from __future__ import annotations
 
-import dataclasses
-from datetime import timedelta
 import logging
 
-import aiohttp
+from homeassistant.const import EVENT_COMPONENT_LOADED
-from awesomeversion import AwesomeVersion, AwesomeVersionStrategy
-
-from homeassistant.components.hassio import get_supervisor_info, is_hassio
-from homeassistant.const import EVENT_COMPONENT_LOADED, __version__
 from homeassistant.core import Event, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
@@ -22,15 +16,12 @@ from homeassistant.helpers.issue_registry import (
 )
 from homeassistant.helpers.start import async_at_started
 from homeassistant.helpers.typing import ConfigType
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 from homeassistant.setup import EventComponentLoaded
 
-COMPONENT_LOADED_COOLDOWN = 30
+from .const import COMPONENT_LOADED_COOLDOWN, DOMAIN, REQUEST_TIMEOUT
-DOMAIN = "homeassistant_alerts"
+from .coordinator import AlertUpdateCoordinator
-UPDATE_INTERVAL = timedelta(hours=3)
-_LOGGER = logging.getLogger(__name__)
 
-REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=30)
+_LOGGER = logging.getLogger(__name__)
 
 CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
 
@@ -114,98 +105,3 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     async_at_started(hass, initial_refresh)
 
     return True
-
-
-@dataclasses.dataclass(slots=True, frozen=True)
-class IntegrationAlert:
-    """Issue Registry Entry."""
-
-    alert_id: str
-    integration: str
-    filename: str
-    date_updated: str | None
-
-    @property
-    def issue_id(self) -> str:
-        """Return the issue id."""
-        return f"{self.filename}_{self.integration}"
-
-
-class AlertUpdateCoordinator(DataUpdateCoordinator[dict[str, IntegrationAlert]]):  # pylint: disable=hass-enforce-coordinator-module
-    """Data fetcher for HA Alerts."""
-
-    def __init__(self, hass: HomeAssistant) -> None:
-        """Initialize the data updater."""
-        super().__init__(
-            hass,
-            _LOGGER,
-            name=DOMAIN,
-            update_interval=UPDATE_INTERVAL,
-        )
-        self.ha_version = AwesomeVersion(
-            __version__,
-            ensure_strategy=AwesomeVersionStrategy.CALVER,
-        )
-        self.supervisor = is_hassio(self.hass)
-
-    async def _async_update_data(self) -> dict[str, IntegrationAlert]:
-        response = await async_get_clientsession(self.hass).get(
-            "https://alerts.home-assistant.io/alerts.json",
-            timeout=REQUEST_TIMEOUT,
-        )
-        alerts = await response.json()
-
-        result = {}
-
-        for alert in alerts:
-            if "integrations" not in alert:
-                continue
-
-            if "homeassistant" in alert:
-                if "affected_from_version" in alert["homeassistant"]:
-                    affected_from_version = AwesomeVersion(
-                        alert["homeassistant"]["affected_from_version"],
-                    )
-                    if self.ha_version < affected_from_version:
-                        continue
-                if "resolved_in_version" in alert["homeassistant"]:
-                    resolved_in_version = AwesomeVersion(
-                        alert["homeassistant"]["resolved_in_version"],
-                    )
-                    if self.ha_version >= resolved_in_version:
-                        continue
-
-            if self.supervisor and "supervisor" in alert:
-                if (supervisor_info := get_supervisor_info(self.hass)) is None:
-                    continue
-
-                if "affected_from_version" in alert["supervisor"]:
-                    affected_from_version = AwesomeVersion(
-                        alert["supervisor"]["affected_from_version"],
-                    )
-                    if supervisor_info["version"] < affected_from_version:
-                        continue
-                if "resolved_in_version" in alert["supervisor"]:
-                    resolved_in_version = AwesomeVersion(
-                        alert["supervisor"]["resolved_in_version"],
-                    )
-                    if supervisor_info["version"] >= resolved_in_version:
-                        continue
-
-            for integration in alert["integrations"]:
-                if "package" not in integration:
-                    continue
-
-                if integration["package"] not in self.hass.config.components:
-                    continue
-
-                integration_alert = IntegrationAlert(
-                    alert_id=alert["id"],
-                    integration=integration["package"],
-                    filename=alert["filename"],
-                    date_updated=alert.get("updated"),
-                )
-
-                result[integration_alert.issue_id] = integration_alert
-
-        return result

homeassistant/components/homeassistant_alerts/const.py

@@ -0,0 +1,11 @@
+"""Constants for the Home Assistant alerts integration."""
+
+from datetime import timedelta
+
+import aiohttp
+
+COMPONENT_LOADED_COOLDOWN = 30
+DOMAIN = "homeassistant_alerts"
+UPDATE_INTERVAL = timedelta(hours=3)
+
+REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=30)

homeassistant/components/homeassistant_alerts/coordinator.py

@@ -0,0 +1,111 @@
+"""Coordinator for the Home Assistant alerts integration."""
+
+import dataclasses
+import logging
+
+from awesomeversion import AwesomeVersion, AwesomeVersionStrategy
+
+from homeassistant.components.hassio import get_supervisor_info, is_hassio
+from homeassistant.const import __version__
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN, REQUEST_TIMEOUT, UPDATE_INTERVAL
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclasses.dataclass(slots=True, frozen=True)
+class IntegrationAlert:
+    """Issue Registry Entry."""
+
+    alert_id: str
+    integration: str
+    filename: str
+    date_updated: str | None
+
+    @property
+    def issue_id(self) -> str:
+        """Return the issue id."""
+        return f"{self.filename}_{self.integration}"
+
+
+class AlertUpdateCoordinator(DataUpdateCoordinator[dict[str, IntegrationAlert]]):
+    """Data fetcher for HA Alerts."""
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the data updater."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=UPDATE_INTERVAL,
+        )
+        self.ha_version = AwesomeVersion(
+            __version__,
+            ensure_strategy=AwesomeVersionStrategy.CALVER,
+        )
+        self.supervisor = is_hassio(self.hass)
+
+    async def _async_update_data(self) -> dict[str, IntegrationAlert]:
+        response = await async_get_clientsession(self.hass).get(
+            "https://alerts.home-assistant.io/alerts.json",
+            timeout=REQUEST_TIMEOUT,
+        )
+        alerts = await response.json()
+
+        result = {}
+
+        for alert in alerts:
+            if "integrations" not in alert:
+                continue
+
+            if "homeassistant" in alert:
+                if "affected_from_version" in alert["homeassistant"]:
+                    affected_from_version = AwesomeVersion(
+                        alert["homeassistant"]["affected_from_version"],
+                    )
+                    if self.ha_version < affected_from_version:
+                        continue
+                if "resolved_in_version" in alert["homeassistant"]:
+                    resolved_in_version = AwesomeVersion(
+                        alert["homeassistant"]["resolved_in_version"],
+                    )
+                    if self.ha_version >= resolved_in_version:
+                        continue
+
+            if self.supervisor and "supervisor" in alert:
+                if (supervisor_info := get_supervisor_info(self.hass)) is None:
+                    continue
+
+                if "affected_from_version" in alert["supervisor"]:
+                    affected_from_version = AwesomeVersion(
+                        alert["supervisor"]["affected_from_version"],
+                    )
+                    if supervisor_info["version"] < affected_from_version:
+                        continue
+                if "resolved_in_version" in alert["supervisor"]:
+                    resolved_in_version = AwesomeVersion(
+                        alert["supervisor"]["resolved_in_version"],
+                    )
+                    if supervisor_info["version"] >= resolved_in_version:
+                        continue
+
+            for integration in alert["integrations"]:
+                if "package" not in integration:
+                    continue
+
+                if integration["package"] not in self.hass.config.components:
+                    continue
+
+                integration_alert = IntegrationAlert(
+                    alert_id=alert["id"],
+                    integration=integration["package"],
+                    filename=alert["filename"],
+                    date_updated=alert.get("updated"),
+                )
+
+                result[integration_alert.issue_id] = integration_alert
+
+        return result

tests/components/homeassistant_alerts/test_init.py

@@ -10,7 +10,7 @@ from freezegun.api import FrozenDateTimeFactory
 import pytest
 from pytest_unordered import unordered
 
-from homeassistant.components.homeassistant_alerts import (
+from homeassistant.components.homeassistant_alerts.const import (
     COMPONENT_LOADED_COOLDOWN,
     DOMAIN,
     UPDATE_INTERVAL,
@@ -134,15 +134,15 @@ async def test_alerts(
 
     with (
         patch(
-            "homeassistant.components.homeassistant_alerts.__version__",
+            "homeassistant.components.homeassistant_alerts.coordinator.__version__",
             ha_version,
         ),
         patch(
-            "homeassistant.components.homeassistant_alerts.is_hassio",
+            "homeassistant.components.homeassistant_alerts.coordinator.is_hassio",
             return_value=supervisor_info is not None,
         ),
         patch(
-            "homeassistant.components.homeassistant_alerts.get_supervisor_info",
+            "homeassistant.components.homeassistant_alerts.coordinator.get_supervisor_info",
             return_value=supervisor_info,
         ),
     ):
@@ -317,15 +317,15 @@ async def test_alerts_refreshed_on_component_load(
 
     with (
         patch(
-            "homeassistant.components.homeassistant_alerts.__version__",
+            "homeassistant.components.homeassistant_alerts.coordinator.__version__",
             ha_version,
         ),
         patch(
-            "homeassistant.components.homeassistant_alerts.is_hassio",
+            "homeassistant.components.homeassistant_alerts.coordinator.is_hassio",
             return_value=supervisor_info is not None,
         ),
         patch(
-            "homeassistant.components.homeassistant_alerts.get_supervisor_info",
+            "homeassistant.components.homeassistant_alerts.coordinator.get_supervisor_info",
             return_value=supervisor_info,
         ),
     ):
@@ -361,15 +361,15 @@ async def test_alerts_refreshed_on_component_load(
 
     with (
         patch(
-            "homeassistant.components.homeassistant_alerts.__version__",
+            "homeassistant.components.homeassistant_alerts.coordinator.__version__",
             ha_version,
         ),
         patch(
-            "homeassistant.components.homeassistant_alerts.is_hassio",
+            "homeassistant.components.homeassistant_alerts.coordinator.is_hassio",
             return_value=supervisor_info is not None,
         ),
         patch(
-            "homeassistant.components.homeassistant_alerts.get_supervisor_info",
+            "homeassistant.components.homeassistant_alerts.coordinator.get_supervisor_info",
             return_value=supervisor_info,
         ),
     ):
@@ -456,7 +456,7 @@ async def test_bad_alerts(
         hass.config.components.add(domain)
 
     with patch(
-        "homeassistant.components.homeassistant_alerts.__version__",
+        "homeassistant.components.homeassistant_alerts.coordinator.__version__",
         ha_version,
     ):
         assert await async_setup_component(hass, DOMAIN, {})
@@ -615,7 +615,7 @@ async def test_alerts_change(
         hass.config.components.add(domain)
 
     with patch(
-        "homeassistant.components.homeassistant_alerts.__version__",
+        "homeassistant.components.homeassistant_alerts.coordinator.__version__",
         ha_version,
     ):
         assert await async_setup_component(hass, DOMAIN, {})