Enable strict typing of analytics (#83119)

.strict-typing

@@ -56,6 +56,7 @@ homeassistant.components.amazon_polly.*
 homeassistant.components.ambient_station.*
 homeassistant.components.amcrest.*
 homeassistant.components.ampio.*
+homeassistant.components.analytics.*
 homeassistant.components.anthemav.*
 homeassistant.components.aqualogic.*
 homeassistant.components.aseko_pool_live.*

homeassistant/components/analytics/__init__.py

@@ -5,7 +5,7 @@ import voluptuous as vol
 
 from homeassistant.components import websocket_api
 from homeassistant.const import EVENT_HOMEASSISTANT_STARTED
-from homeassistant.core import HomeAssistant
+from homeassistant.core import Event, HomeAssistant, callback
 from homeassistant.helpers.event import async_call_later, async_track_time_interval
 from homeassistant.helpers.typing import ConfigType
 
@@ -20,7 +20,8 @@ async def async_setup(hass: HomeAssistant, _: ConfigType) -> bool:
     # Load stored data
     await analytics.load()
 
-    async def start_schedule(_event):
+    @callback
+    def start_schedule(_event: Event) -> None:
         """Start the send schedule after the started event."""
         # Wait 15 min after started
         async_call_later(hass, 900, analytics.send_analytics)
@@ -37,10 +38,10 @@ async def async_setup(hass: HomeAssistant, _: ConfigType) -> bool:
     return True
 
 
+@callback
 @websocket_api.require_admin
 @websocket_api.websocket_command({vol.Required("type"): "analytics"})
-@websocket_api.async_response
-async def websocket_analytics(
+def websocket_analytics(
     hass: HomeAssistant,
     connection: websocket_api.connection.ActiveConnection,
     msg: dict[str, Any],

homeassistant/components/analytics/analytics.py

@@ -1,5 +1,9 @@
 """Analytics helper class for the analytics integration."""
+from __future__ import annotations
+
 import asyncio
+from dataclasses import asdict as dataclass_asdict, dataclass
+from datetime import datetime
 from typing import Any
 import uuid
 
@@ -39,9 +43,7 @@ from .const import (
     ATTR_HEALTHY,
     ATTR_INTEGRATION_COUNT,
     ATTR_INTEGRATIONS,
-    ATTR_ONBOARDED,
     ATTR_OPERATING_SYSTEM,
-    ATTR_PREFERENCES,
     ATTR_PROTECTED,
     ATTR_SLUG,
     ATTR_STATE_COUNT,
@@ -59,6 +61,24 @@ from .const import (
 )
 
 
+@dataclass
+class AnalyticsData:
+    """Analytics data."""
+
+    onboarded: bool
+    preferences: dict[str, bool]
+    uuid: str | None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> AnalyticsData:
+        """Initialize analytics data from a dict."""
+        return cls(
+            data["onboarded"],
+            data["preferences"],
+            data["uuid"],
+        )
+
+
 class Analytics:
     """Analytics helper class for the analytics integration."""
 
@@ -66,17 +86,13 @@ class Analytics:
         """Initialize the Analytics class."""
         self.hass: HomeAssistant = hass
         self.session = async_get_clientsession(hass)
-        self._data: dict[str, Any] = {
-            ATTR_PREFERENCES: {},
-            ATTR_ONBOARDED: False,
-            ATTR_UUID: None,
-        }
+        self._data = AnalyticsData(False, {}, None)
         self._store = Store[dict[str, Any]](hass, STORAGE_VERSION, STORAGE_KEY)
 
     @property
     def preferences(self) -> dict:
         """Return the current active preferences."""
-        preferences = self._data[ATTR_PREFERENCES]
+        preferences = self._data.preferences
         return {
             ATTR_BASE: preferences.get(ATTR_BASE, False),
             ATTR_DIAGNOSTICS: preferences.get(ATTR_DIAGNOSTICS, False),
@@ -87,12 +103,12 @@ class Analytics:
     @property
     def onboarded(self) -> bool:
         """Return bool if the user has made a choice."""
-        return self._data[ATTR_ONBOARDED]
+        return self._data.onboarded
 
     @property
-    def uuid(self) -> bool:
+    def uuid(self) -> str | None:
         """Return the uuid for the analytics integration."""
-        return self._data[ATTR_UUID]
+        return self._data.uuid
 
     @property
     def endpoint(self) -> str:
@@ -111,7 +127,7 @@ class Analytics:
         """Load preferences."""
         stored = await self._store.async_load()
         if stored:
-            self._data = stored
+            self._data = AnalyticsData.from_dict(stored)
 
         if (
             self.supervisor
@@ -122,26 +138,26 @@ class Analytics:
                 if supervisor_info[ATTR_DIAGNOSTICS] and not self.preferences.get(
                     ATTR_DIAGNOSTICS, False
                 ):
-                    self._data[ATTR_PREFERENCES][ATTR_DIAGNOSTICS] = True
+                    self._data.preferences[ATTR_DIAGNOSTICS] = True
                 elif not supervisor_info[ATTR_DIAGNOSTICS] and self.preferences.get(
                     ATTR_DIAGNOSTICS, False
                 ):
-                    self._data[ATTR_PREFERENCES][ATTR_DIAGNOSTICS] = False
+                    self._data.preferences[ATTR_DIAGNOSTICS] = False
 
     async def save_preferences(self, preferences: dict) -> None:
         """Save preferences."""
         preferences = PREFERENCE_SCHEMA(preferences)
-        self._data[ATTR_PREFERENCES].update(preferences)
-        self._data[ATTR_ONBOARDED] = True
+        self._data.preferences.update(preferences)
+        self._data.onboarded = True
 
-        await self._store.async_save(self._data)
+        await self._store.async_save(dataclass_asdict(self._data))
 
         if self.supervisor:
             await hassio.async_update_diagnostics(
                 self.hass, self.preferences.get(ATTR_DIAGNOSTICS, False)
             )
 
-    async def send_analytics(self, _=None) -> None:
+    async def send_analytics(self, _: datetime | None = None) -> None:
         """Send analytics."""
         supervisor_info = None
         operating_system_info: dict[str, Any] = {}
@@ -150,9 +166,9 @@ class Analytics:
             LOGGER.debug("Nothing to submit")
             return
 
-        if self._data.get(ATTR_UUID) is None:
-            self._data[ATTR_UUID] = uuid.uuid4().hex
-            await self._store.async_save(self._data)
+        if self._data.uuid is None:
+            self._data.uuid = uuid.uuid4().hex
+            await self._store.async_save(dataclass_asdict(self._data))
 
         if self.supervisor:
             supervisor_info = hassio.get_supervisor_info(self.hass)

mypy.ini

@@ -313,6 +313,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.analytics.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.anthemav.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

tests/components/analytics/test_analytics.py

@@ -10,11 +10,9 @@ from homeassistant.components.analytics.const import (
     ANALYTICS_ENDPOINT_URL_DEV,
     ATTR_BASE,
     ATTR_DIAGNOSTICS,
-    ATTR_PREFERENCES,
     ATTR_STATISTICS,
     ATTR_USAGE,
 )
-from homeassistant.components.api import ATTR_UUID
 from homeassistant.const import ATTR_DOMAIN
 from homeassistant.loader import IntegrationNotFound
 from homeassistant.setup import async_setup_component
@@ -58,7 +56,7 @@ async def test_load_with_supervisor_diagnostics(hass):
 async def test_load_with_supervisor_without_diagnostics(hass):
     """Test loading with a supervisor that has not diagnostics enabled."""
     analytics = Analytics(hass)
-    analytics._data[ATTR_PREFERENCES][ATTR_DIAGNOSTICS] = True
+    analytics._data.preferences[ATTR_DIAGNOSTICS] = True
 
     assert analytics.preferences[ATTR_DIAGNOSTICS]
 
@@ -349,7 +347,7 @@ async def test_reusing_uuid(hass, aioclient_mock):
     """Test reusing the stored UUID."""
     aioclient_mock.post(ANALYTICS_ENDPOINT_URL, status=200)
     analytics = Analytics(hass)
-    analytics._data[ATTR_UUID] = "NOT_MOCK_UUID"
+    analytics._data.uuid = "NOT_MOCK_UUID"
 
     await analytics.save_preferences({ATTR_BASE: True})