Enable mypy for motionEye (aye aye!) (#49738)

homeassistant/components/motioneye/__init__.py

@@ -65,10 +65,10 @@ def get_motioneye_entity_unique_id(
 
 
 def get_camera_from_cameras(
-    camera_id: int, data: dict[str, Any]
+    camera_id: int, data: dict[str, Any] | None
 ) -> dict[str, Any] | None:
     """Get an individual camera dict from a multiple cameras data response."""
-    for camera in data.get(KEY_CAMERAS) or []:
+    for camera in data.get(KEY_CAMERAS, []) if data else []:
         if camera.get(KEY_ID) == camera_id:
             val: dict[str, Any] = camera
             return val
@@ -105,7 +105,7 @@ def _add_camera(
     entry: ConfigEntry,
     camera_id: int,
     camera: dict[str, Any],
-    device_identifier: tuple[str, str, int],
+    device_identifier: tuple[str, str],
 ) -> None:
     """Add a motionEye camera to hass."""
 
@@ -164,14 +164,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         CONF_COORDINATOR: coordinator,
     }
 
-    current_cameras: set[tuple[str, str, int]] = set()
+    current_cameras: set[tuple[str, str]] = set()
     device_registry = await dr.async_get_registry(hass)
 
     @callback
     def _async_process_motioneye_cameras() -> None:
         """Process motionEye camera additions and removals."""
-        inbound_camera: set[tuple[str, str, int]] = set()
-        if KEY_CAMERAS not in coordinator.data:
+        inbound_camera: set[tuple[str, str]] = set()
+        if coordinator.data is None or KEY_CAMERAS not in coordinator.data:
             return
 
         for camera in coordinator.data[KEY_CAMERAS]:

homeassistant/components/motioneye/camera.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
+from typing import Any, Dict, Optional
 
 import aiohttp
 from motioneye_client.client import MotionEyeClient
@@ -86,7 +86,7 @@ async def async_setup_entry(
     listen_for_new_cameras(hass, entry, camera_add)
 
 
-class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity):
+class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity[Optional[Dict[str, Any]]]):
     """motionEye mjpeg camera."""
 
     def __init__(
@@ -96,7 +96,7 @@ class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity):
         password: str,
         camera: dict[str, Any],
         client: MotionEyeClient,
-        coordinator: DataUpdateCoordinator,
+        coordinator: DataUpdateCoordinator[dict[str, Any] | None],
     ) -> None:
         """Initialize a MJPEG camera."""
         self._surveillance_username = username
@@ -191,7 +191,7 @@ class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity):
                 self._motion_detection_enabled = camera.get(KEY_MOTION_DETECTION, False)
                 available = True
         self._available = available
-        CoordinatorEntity._handle_coordinator_update(self)
+        super()._handle_coordinator_update()
 
     @property
     def brand(self) -> str:

homeassistant/components/motioneye/config_flow.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
+from typing import Any, Dict, cast
 
 from motioneye_client.client import (
     MotionEyeClientConnectionError,
@@ -13,8 +13,8 @@ import voluptuous as vol
 
 from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow
 from homeassistant.const import CONF_SOURCE, CONF_URL
+from homeassistant.data_entry_flow import FlowResult
 from homeassistant.helpers import config_validation as cv
-from homeassistant.helpers.typing import ConfigType
 
 from . import create_motioneye_client
 from .const import (
@@ -34,13 +34,13 @@ class MotionEyeConfigFlow(ConfigFlow, domain=DOMAIN):
     VERSION = 1
 
     async def async_step_user(
-        self, user_input: ConfigType | None = None
-    ) -> dict[str, Any]:
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Handle the initial step."""
 
         def _get_form(
-            user_input: ConfigType, errors: dict[str, str] | None = None
-        ) -> dict[str, Any]:
+            user_input: dict[str, Any], errors: dict[str, str] | None = None
+        ) -> FlowResult:
             """Show the form to the user."""
             return self.async_show_form(
                 step_id="user",
@@ -77,7 +77,9 @@ class MotionEyeConfigFlow(ConfigFlow, domain=DOMAIN):
             )
 
         if user_input is None:
-            return _get_form(reauth_entry.data if reauth_entry else {})
+            return _get_form(
+                cast(Dict[str, Any], reauth_entry.data) if reauth_entry else {}
+            )
 
         try:
             # Cannot use cv.url validation in the schema itself, so
@@ -130,7 +132,7 @@ class MotionEyeConfigFlow(ConfigFlow, domain=DOMAIN):
 
     async def async_step_reauth(
         self,
-        config_data: ConfigType | None = None,
-    ) -> dict[str, Any]:
+        config_data: dict[str, Any] | None = None,
+    ) -> FlowResult:
         """Handle a reauthentication flow."""
         return await self.async_step_user(config_data)

mypy.ini

@@ -996,9 +996,6 @@ ignore_errors = true
 [mypy-homeassistant.components.motion_blinds.*]
 ignore_errors = true
 
-[mypy-homeassistant.components.motioneye.*]
-ignore_errors = true
-
 [mypy-homeassistant.components.mqtt.*]
 ignore_errors = true
 

script/hassfest/mypy_config.py

@@ -135,7 +135,6 @@ IGNORED_MODULES: Final[list[str]] = [
     "homeassistant.components.minecraft_server.*",
     "homeassistant.components.mobile_app.*",
     "homeassistant.components.motion_blinds.*",
-    "homeassistant.components.motioneye.*",
     "homeassistant.components.mqtt.*",
     "homeassistant.components.mullvad.*",
     "homeassistant.components.mysensors.*",

tests/common.py

@@ -5,7 +5,7 @@ import asyncio
 import collections
 from collections import OrderedDict
 from contextlib import contextmanager
-from datetime import timedelta
+from datetime import datetime, timedelta
 import functools as ft
 from io import StringIO
 import json
@@ -44,7 +44,7 @@ from homeassistant.const import (
     STATE_OFF,
     STATE_ON,
 )
-from homeassistant.core import BLOCK_LOG_TIMEOUT, State
+from homeassistant.core import BLOCK_LOG_TIMEOUT, HomeAssistant, State
 from homeassistant.helpers import (
     area_registry,
     device_registry,
@@ -361,7 +361,9 @@ fire_mqtt_message = threadsafe_callback_factory(async_fire_mqtt_message)
 
 
 @ha.callback
-def async_fire_time_changed(hass, datetime_, fire_all=False):
+def async_fire_time_changed(
+    hass: HomeAssistant, datetime_: datetime, fire_all: bool = False
+) -> None:
     """Fire a time changes event."""
     hass.bus.async_fire(EVENT_TIME_CHANGED, {"now": date_util.as_utc(datetime_)})
 

tests/components/motioneye/test_camera.py

@@ -4,7 +4,7 @@ import logging
 from typing import Any
 from unittest.mock import AsyncMock, Mock
 
-from aiohttp import web  # type: ignore
+from aiohttp import web
 from aiohttp.web_exceptions import HTTPBadGateway
 from motioneye_client.client import (
     MotionEyeClientError,
@@ -165,6 +165,7 @@ async def test_setup_camera_new_data_error(hass: HomeAssistant) -> None:
     async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
     await hass.async_block_till_done()
     entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
+    assert entity_state
     assert entity_state.state == "unavailable"
 
 
@@ -173,6 +174,7 @@ async def test_setup_camera_new_data_without_streaming(hass: HomeAssistant) -> N
     client = create_mock_motioneye_client()
     await setup_mock_motioneye_config_entry(hass, client=client)
     entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
+    assert entity_state
     assert entity_state.state == "idle"
 
     cameras = copy.deepcopy(TEST_CAMERAS)
@@ -181,6 +183,7 @@ async def test_setup_camera_new_data_without_streaming(hass: HomeAssistant) -> N
     async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
     await hass.async_block_till_done()
     entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID)
+    assert entity_state
     assert entity_state.state == "unavailable"
 
 

tests/components/motioneye/test_config_flow.py

@@ -235,7 +235,7 @@ async def test_reauth(hass: HomeAssistant) -> None:
 
     assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
     assert result["reason"] == "reauth_successful"
-    assert config_entry.data == new_data
+    assert dict(config_entry.data) == new_data
 
     assert len(mock_setup_entry.mock_calls) == 1
     assert mock_client.async_client_close.called