Move vizio coordinator to separate module (#117498)

homeassistant/components/vizio/__init__.py

@@ -2,12 +2,8 @@
 
 from __future__ import annotations
 
-from datetime import timedelta
-import logging
 from typing import Any
 
-from pyvizio.const import APPS
-from pyvizio.util import gen_apps_list_from_url
 import voluptuous as vol
 
 from homeassistant.components.media_player import MediaPlayerDeviceClass
@@ -15,14 +11,11 @@ from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry, ConfigEntry
 from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import config_validation as cv
-from homeassistant.helpers.aiohttp_client import async_get_clientsession
 from homeassistant.helpers.storage import Store
 from homeassistant.helpers.typing import ConfigType
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
 from .const import CONF_APPS, CONF_DEVICE_CLASS, DOMAIN, VIZIO_SCHEMA
-
-_LOGGER = logging.getLogger(__name__)
+from .coordinator import VizioAppsDataUpdateCoordinator
 
 
 def validate_apps(config: ConfigType) -> ConfigType:
@@ -96,53 +89,3 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
         hass.data.pop(DOMAIN)
 
     return unload_ok
-
-
-class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]):  # pylint: disable=hass-enforce-coordinator-module
-    """Define an object to hold Vizio app config data."""
-
-    def __init__(self, hass: HomeAssistant, store: Store[list[dict[str, Any]]]) -> None:
-        """Initialize."""
-        super().__init__(
-            hass,
-            _LOGGER,
-            name=DOMAIN,
-            update_interval=timedelta(days=1),
-        )
-        self.fail_count = 0
-        self.fail_threshold = 10
-        self.store = store
-
-    async def async_config_entry_first_refresh(self) -> None:
-        """Refresh data for the first time when a config entry is setup."""
-        self.data = await self.store.async_load() or APPS
-        await super().async_config_entry_first_refresh()
-
-    async def _async_update_data(self) -> list[dict[str, Any]]:
-        """Update data via library."""
-        if data := await gen_apps_list_from_url(
-            session=async_get_clientsession(self.hass)
-        ):
-            # Reset the fail count and threshold when the data is successfully retrieved
-            self.fail_count = 0
-            self.fail_threshold = 10
-            # Store the new data if it has changed so we have it for the next restart
-            if data != self.data:
-                await self.store.async_save(data)
-            return data
-        # For every failure, increase the fail count until we reach the threshold.
-        # We then log a warning, increase the threshold, and reset the fail count.
-        # This is here to prevent silent failures but to reduce repeat logs.
-        if self.fail_count == self.fail_threshold:
-            _LOGGER.warning(
-                (
-                    "Unable to retrieve the apps list from the external server for the "
-                    "last %s days"
-                ),
-                self.fail_threshold,
-            )
-            self.fail_count = 0
-            self.fail_threshold += 10
-        else:
-            self.fail_count += 1
-        return self.data

homeassistant/components/vizio/coordinator.py

@@ -0,0 +1,69 @@
+"""Coordinator for the vizio component."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+from typing import Any
+
+from pyvizio.const import APPS
+from pyvizio.util import gen_apps_list_from_url
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.storage import Store
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]):
+    """Define an object to hold Vizio app config data."""
+
+    def __init__(self, hass: HomeAssistant, store: Store[list[dict[str, Any]]]) -> None:
+        """Initialize."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(days=1),
+        )
+        self.fail_count = 0
+        self.fail_threshold = 10
+        self.store = store
+
+    async def async_config_entry_first_refresh(self) -> None:
+        """Refresh data for the first time when a config entry is setup."""
+        self.data = await self.store.async_load() or APPS
+        await super().async_config_entry_first_refresh()
+
+    async def _async_update_data(self) -> list[dict[str, Any]]:
+        """Update data via library."""
+        if data := await gen_apps_list_from_url(
+            session=async_get_clientsession(self.hass)
+        ):
+            # Reset the fail count and threshold when the data is successfully retrieved
+            self.fail_count = 0
+            self.fail_threshold = 10
+            # Store the new data if it has changed so we have it for the next restart
+            if data != self.data:
+                await self.store.async_save(data)
+            return data
+        # For every failure, increase the fail count until we reach the threshold.
+        # We then log a warning, increase the threshold, and reset the fail count.
+        # This is here to prevent silent failures but to reduce repeat logs.
+        if self.fail_count == self.fail_threshold:
+            _LOGGER.warning(
+                (
+                    "Unable to retrieve the apps list from the external server for the "
+                    "last %s days"
+                ),
+                self.fail_threshold,
+            )
+            self.fail_count = 0
+            self.fail_threshold += 10
+        else:
+            self.fail_count += 1
+        return self.data

homeassistant/components/vizio/media_player.py

@@ -34,7 +34,6 @@ from homeassistant.helpers.dispatcher import (
 )
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import VizioAppsDataUpdateCoordinator
 from .const import (
     CONF_ADDITIONAL_CONFIGS,
     CONF_APPS,
@@ -53,6 +52,7 @@ from .const import (
     VIZIO_SOUND_MODE,
     VIZIO_VOLUME,
 )
+from .coordinator import VizioAppsDataUpdateCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 

tests/components/vizio/conftest.py

@@ -54,7 +54,7 @@ def vizio_get_unique_id_fixture():
 def vizio_data_coordinator_update_fixture():
     """Mock get data coordinator update."""
     with patch(
-        "homeassistant.components.vizio.gen_apps_list_from_url",
+        "homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
         return_value=APP_LIST,
     ):
         yield
@@ -64,7 +64,7 @@ def vizio_data_coordinator_update_fixture():
 def vizio_data_coordinator_update_failure_fixture():
     """Mock get data coordinator update failure."""
     with patch(
-        "homeassistant.components.vizio.gen_apps_list_from_url",
+        "homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
         return_value=None,
     ):
         yield

tests/components/vizio/test_media_player.py

@@ -745,7 +745,7 @@ async def test_apps_update(
 ) -> None:
     """Test device setup with apps where no app is running."""
     with patch(
-        "homeassistant.components.vizio.gen_apps_list_from_url",
+        "homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
         return_value=None,
     ):
         async with _cm_for_test_setup_tv_with_apps(
@@ -758,7 +758,7 @@ async def test_apps_update(
             assert len(apps) == len(APPS)
 
             with patch(
-                "homeassistant.components.vizio.gen_apps_list_from_url",
+                "homeassistant.components.vizio.coordinator.gen_apps_list_from_url",
                 return_value=APP_LIST,
             ):
                 async_fire_time_changed(hass, dt_util.now() + timedelta(days=2))