Add scripts

add_override_decorators.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+"""Add @override decorator to methods listed in explicit_override_errors.txt."""
+
+from __future__ import annotations
+
+import re
+import subprocess
+from collections import defaultdict
+from pathlib import Path
+
+INPUT = Path("explicit_override_errors.txt")
+ERROR_RE = re.compile(r"^(.+?):(\d+): error:.*\[explicit-override\]")
+
+
+def decorator_stack_top(lines: list[str], def_idx: int) -> int:
+    """Return the index of the topmost decorator above the def at def_idx."""
+    i = def_idx - 1
+    while i >= 0 and lines[i].lstrip().startswith("@"):
+        i -= 1
+    return i + 1
+
+
+by_file: dict[Path, set[int]] = defaultdict(set)
+for line in INPUT.read_text().splitlines():
+    if m := ERROR_RE.match(line):
+        by_file[Path(m.group(1))].add(int(m.group(2)))
+
+for path, line_nums in by_file.items():
+    lines = path.read_text().splitlines(keepends=True)
+    for lineno in sorted(line_nums, reverse=True):
+        insert_idx = decorator_stack_top(lines, lineno - 1)
+        target = lines[insert_idx]
+        indent = target[: len(target) - len(target.lstrip())]
+        lines.insert(insert_idx, f"{indent}@override\n")
+    first_import = next(
+        i for i, ln in enumerate(lines) if ln.startswith(("import ", "from "))
+    )
+    lines.insert(first_import, "from typing import override\n")
+    path.write_text("".join(lines))
+    print(f"Updated {path} ({len(line_nums)} methods)")
+
+if by_file:
+    subprocess.run(["ruff", "check", "--fix", *map(str, by_file)], check=False)

extract_override_errors.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+"""Run mypy on a directory and write `[explicit-override]` errors to a file."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+OUTPUT = Path("explicit_override_errors.txt")
+
+target = sys.argv[1]
+result = subprocess.run(
+    ["mypy", "--enable-error-code=explicit-override", target],
+    capture_output=True,
+    text=True,
+    check=False,
+)
+matches = [line for line in result.stdout.splitlines() if "[explicit-override]" in line]
+OUTPUT.write_text("\n".join(matches) + ("\n" if matches else ""))
+print(f"Wrote {len(matches)} errors to {OUTPUT}")

homeassistant/components/homewizard/sensor.py

@@ -35,6 +35,7 @@ from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 from homeassistant.helpers.typing import StateType
 from homeassistant.util.dt import utcnow
+from homeassistant.util.variance import ignore_variance
 
 from .const import DOMAIN
 from .coordinator import HomeWizardConfigEntry, HWEnergyDeviceUpdateCoordinator
@@ -65,6 +66,13 @@ def to_percentage(value: float | None) -> float | None:
     return value * 100 if value is not None else None
 
 
+def uptime_to_datetime(value: int) -> datetime:
+    """Convert seconds to datetime timestamp."""
+    return utcnow().replace(microsecond=0) - timedelta(seconds=value)
+
+
+uptime_to_stable_datetime = ignore_variance(uptime_to_datetime, timedelta(minutes=5))
+
 SENSORS: Final[tuple[HomeWizardSensorEntityDescription, ...]] = (
     HomeWizardSensorEntityDescription(
         key="smr_version",
@@ -635,7 +643,7 @@ SENSORS: Final[tuple[HomeWizardSensorEntityDescription, ...]] = (
     HomeWizardSensorEntityDescription(
         key="uptime",
         translation_key="uptime",
-        device_class=SensorDeviceClass.UPTIME,
+        device_class=SensorDeviceClass.TIMESTAMP,
         entity_category=EntityCategory.DIAGNOSTIC,
         entity_registry_enabled_default=False,
         has_fn=(
@@ -643,7 +651,7 @@ SENSORS: Final[tuple[HomeWizardSensorEntityDescription, ...]] = (
         ),
         value_fn=(
             lambda data: (
-                utcnow() - timedelta(seconds=data.system.uptime_s)
+                uptime_to_stable_datetime(data.system.uptime_s)
                 if data.system is not None and data.system.uptime_s is not None
                 else None
             )

homeassistant/components/hue/v2/device_trigger.py

@@ -87,8 +87,6 @@ def async_get_triggers(
 
     # Get Hue device id from device identifier
     hue_dev_id = get_hue_device_id(device_entry)
-    if hue_dev_id is None or hue_dev_id not in api.devices:
-        return []
     # extract triggers from all button resources of this Hue device
     triggers: list[dict[str, Any]] = []
     model_id = api.devices[hue_dev_id].product_data.product_name

homeassistant/components/insteon/__init__.py

@@ -118,8 +118,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         )
         device = devices.add_x10_device(housecode, unitcode, x10_type, steps)
 
-    create_insteon_device(hass, devices.modem, entry.entry_id)
-
     await hass.config_entries.async_forward_entry_setups(entry, INSTEON_PLATFORMS)
 
     for address in devices:
@@ -133,6 +131,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     register_new_device_callback(hass)
     async_setup_services(hass)
 
+    create_insteon_device(hass, devices.modem, entry.entry_id)
+
     entry.async_create_background_task(
         hass, async_get_device_config(hass, entry), "insteon-get-device-config"
     )

homeassistant/components/nina/binary_sensor.py

@@ -6,7 +6,6 @@ from homeassistant.components.binary_sensor import (
     BinarySensorDeviceClass,
     BinarySensorEntity,
 )
-from homeassistant.const import ATTR_ID
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
@@ -15,6 +14,7 @@ from .const import (
     ATTR_DESCRIPTION,
     ATTR_EXPIRES,
     ATTR_HEADLINE,
+    ATTR_ID,
     ATTR_RECOMMENDED_ACTIONS,
     ATTR_SENDER,
     ATTR_SENT,

homeassistant/components/nina/const.py

@@ -29,6 +29,8 @@ ATTR_SEVERITY: str = "severity"
 ATTR_RECOMMENDED_ACTIONS: str = "recommended_actions"
 ATTR_AFFECTED_AREAS: str = "affected_areas"
 ATTR_WEB: str = "web"
+# pylint: disable-next=home-assistant-duplicate-const
+ATTR_ID: str = "id"
 ATTR_SENT: str = "sent"
 ATTR_START: str = "start"
 ATTR_EXPIRES: str = "expires"

homeassistant/components/openhome/config_flow.py

@@ -37,15 +37,11 @@ class OpenhomeConfigFlow(ConfigFlow, domain=DOMAIN):
             _LOGGER.debug("async_step_ssdp: Incomplete discovery, ignoring")
             return self.async_abort(reason="incomplete_discovery")
 
-        udn = discovery_info.upnp[ATTR_UPNP_UDN]
-        if isinstance(udn, list):
-            if not udn:
-                return self.async_abort(reason="incomplete_discovery")
-            udn = udn[0]
+        _LOGGER.debug(
+            "async_step_ssdp: setting unique id %s", discovery_info.upnp[ATTR_UPNP_UDN]
+        )
 
-        _LOGGER.debug("async_step_ssdp: setting unique id %s", udn)
-
-        await self.async_set_unique_id(udn)
+        await self.async_set_unique_id(discovery_info.upnp[ATTR_UPNP_UDN])
         self._abort_if_unique_id_configured({CONF_HOST: discovery_info.ssdp_location})
 
         _LOGGER.debug(

homeassistant/components/system_bridge/services.yaml

@@ -89,4 +89,3 @@ power_command:
             - "restart"
             - "shutdown"
             - "sleep"
-          translation_key: "power_command"

homeassistant/components/system_bridge/strings.json

@@ -178,18 +178,6 @@
       "title": "System Bridge upgrade required"
     }
   },
-  "selector": {
-    "power_command": {
-      "options": {
-        "hibernate": "Hibernate",
-        "lock": "Lock",
-        "logout": "Logout",
-        "restart": "[%key:common::action::restart%]",
-        "shutdown": "Shutdown",
-        "sleep": "Sleep"
-      }
-    }
-  },
   "services": {
     "get_process_by_id": {
       "description": "Gets a process by the ID.",

homeassistant/components/xthings_cloud/const.py

@@ -21,4 +21,4 @@ CONF_INSTANCE_ID = "instance_id"
 # Polling interval (seconds)
 DEFAULT_SCAN_INTERVAL = 1800
 
-PLATFORMS: list[Platform] = [Platform.LIGHT, Platform.LOCK, Platform.SWITCH]
+PLATFORMS: list[Platform] = [Platform.LIGHT, Platform.SWITCH]

homeassistant/components/xthings_cloud/lock.py

@@ -1,47 +0,0 @@
-"""Lock platform for Xthings Cloud."""
-
-from typing import Any
-
-from homeassistant.components.lock import LockEntity
-from homeassistant.core import HomeAssistant
-from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
-
-from .coordinator import XthingsCloudConfigEntry
-from .entity import XthingsCloudEntity
-
-
-async def async_setup_entry(
-    hass: HomeAssistant,
-    entry: XthingsCloudConfigEntry,
-    async_add_entities: AddConfigEntryEntitiesCallback,
-) -> None:
-    """Set up lock platform."""
-    coordinator = entry.runtime_data
-    entities = [
-        XthingsCloudLock(coordinator, device_id, device_data)
-        for device_id, device_data in coordinator.data.items()
-        if device_data["type"] == "lock"
-    ]
-    async_add_entities(entities)
-
-
-class XthingsCloudLock(XthingsCloudEntity, LockEntity):
-    """Xthings Cloud lock entity."""
-
-    @property
-    def is_locked(self) -> bool | None:
-        """Return true if lock is locked."""
-        return self.device_data["status"].get("locked")
-
-    @property
-    def is_jammed(self) -> bool | None:
-        """Return true if lock is jammed."""
-        return self.device_data["status"].get("jammed")
-
-    async def async_lock(self, **kwargs: Any) -> None:
-        """Lock the device."""
-        await self.coordinator.client.async_lock_lock(self._device_id)
-
-    async def async_unlock(self, **kwargs: Any) -> None:
-        """Unlock the device."""
-        await self.coordinator.client.async_lock_unlock(self._device_id)

mypy.ini

@@ -17,7 +17,7 @@ no_implicit_optional = true
 warn_incomplete_stub = true
 warn_redundant_casts = true
 warn_unused_ignores = true
-enable_error_code = deprecated, ignore-without-code, redundant-self, truthy-iterable
+enable_error_code = deprecated, explicit-override, ignore-without-code, redundant-self, truthy-iterable
 disable_error_code = annotation-unchecked, import-not-found, import-untyped
 extra_checks = false
 check_untyped_defs = true

script/hassfest/mypy_config.py

@@ -54,6 +54,7 @@ GENERAL_SETTINGS: Final[dict[str, str]] = {
     "enable_error_code": ", ".join(  # noqa: FLY002
         [
             "deprecated",
+            "explicit-override",
             "ignore-without-code",
             "redundant-self",
             "truthy-iterable",

tests/components/homewizard/snapshots/test_sensor.ambr

@@ -798,7 +798,7 @@
     'object_id_base': 'Uptime',
     'options': dict({
     }),
-    'original_device_class': <SensorDeviceClass.UPTIME: 'uptime'>,
+    'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
     'original_icon': None,
     'original_name': 'Uptime',
     'platform': 'homewizard',
@@ -813,7 +813,7 @@
 # name: test_sensors[HWE-BAT-entity_ids10][sensor.device_uptime:state]
   StateSnapshot({
     'attributes': ReadOnlyDict({
-      'device_class': 'uptime',
+      'device_class': 'timestamp',
       'friendly_name': 'Device Uptime',
     }),
     'context': <ANY>,

tests/components/hue/test_device_trigger_v2.py

@@ -116,30 +116,3 @@ async def test_get_triggers(
     ]
 
     assert triggers == unordered(expected_triggers)
-
-
-async def test_get_triggers_for_removed_device(
-    hass: HomeAssistant,
-    mock_bridge_v2: Mock,
-    v2_resources_test_data: JsonArrayType,
-    device_registry: dr.DeviceRegistry,
-) -> None:
-    """Test triggers for a device removed from the bridge.
-
-    Regression test for https://github.com/home-assistant/core/issues/152937
-    """
-    await mock_bridge_v2.api.load_test_data(v2_resources_test_data)
-    await setup_platform(
-        hass, mock_bridge_v2, [Platform.BINARY_SENSOR, Platform.SENSOR]
-    )
-
-    # Create a device entry with a Hue ID that doesn't exist on the bridge
-    orphaned_device = device_registry.async_get_or_create(
-        config_entry_id=mock_bridge_v2.config_entry.entry_id,
-        identifiers={(hue.DOMAIN, "non-existent-hue-device-id")},
-    )
-
-    triggers = await async_get_device_automations(
-        hass, DeviceAutomationType.TRIGGER, orphaned_device.id
-    )
-    assert triggers == []

tests/components/openhome/test_config_flow.py

@@ -116,54 +116,3 @@ async def test_host_updated(hass: HomeAssistant) -> None:
     assert result["reason"] == "already_configured"
 
     assert entry.data[CONF_HOST] == MOCK_SSDP_LOCATION
-
-
-async def test_ssdp_udn_as_list(hass: HomeAssistant) -> None:
-    """Test SSDP discovery when UDN is a list instead of a string.
-
-    Regression test for https://github.com/home-assistant/core/issues/171837
-    """
-    list_udn_discovery = SsdpServiceInfo(
-        ssdp_usn="usn",
-        ssdp_st="st",
-        ssdp_location=MOCK_SSDP_LOCATION,
-        upnp={
-            ATTR_UPNP_FRIENDLY_NAME: MOCK_FRIENDLY_NAME,
-            ATTR_UPNP_UDN: [MOCK_UDN, "uuid:other"],
-        },
-    )
-    result = await hass.config_entries.flow.async_init(
-        DOMAIN,
-        context={CONF_SOURCE: SOURCE_SSDP},
-        data=list_udn_discovery,
-    )
-
-    assert result["type"] is FlowResultType.FORM
-    assert result["step_id"] == "confirm"
-    assert result["description_placeholders"] == {CONF_NAME: MOCK_FRIENDLY_NAME}
-
-    result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
-    assert result2["type"] is FlowResultType.CREATE_ENTRY
-    assert result2["title"] == MOCK_FRIENDLY_NAME
-    assert result2["data"] == {CONF_HOST: MOCK_SSDP_LOCATION}
-
-
-async def test_ssdp_udn_as_empty_list(hass: HomeAssistant) -> None:
-    """Test SSDP discovery when UDN is an empty list."""
-    empty_udn_discovery = SsdpServiceInfo(
-        ssdp_usn="usn",
-        ssdp_st="st",
-        ssdp_location=MOCK_SSDP_LOCATION,
-        upnp={
-            ATTR_UPNP_FRIENDLY_NAME: MOCK_FRIENDLY_NAME,
-            ATTR_UPNP_UDN: [],
-        },
-    )
-    result = await hass.config_entries.flow.async_init(
-        DOMAIN,
-        context={CONF_SOURCE: SOURCE_SSDP},
-        data=empty_udn_discovery,
-    )
-
-    assert result["type"] is FlowResultType.ABORT
-    assert result["reason"] == "incomplete_discovery"

tests/components/xthings_cloud/conftest.py

@@ -50,7 +50,6 @@ def device_fixtures() -> list[str]:
         "XT-LT200",
         "XT-PL50",
         "XT-PL100",
-        "XT-LK50",
     ]
 
 

tests/components/xthings_cloud/fixtures/XT-LK50.json

@@ -1,13 +0,0 @@
-{
-  "id": "dev_lock_001",
-  "name": "Front Door Lock",
-  "type": "lock",
-  "model": "XT-LK50",
-  "version": "1.0.0",
-  "online": true,
-  "status": {
-    "locked": true,
-    "jammed": false,
-    "battery": 85
-  }
-}

tests/components/xthings_cloud/snapshots/test_init.ambr

@@ -154,34 +154,3 @@
     'via_device_id': None,
   })
 # ---
-# name: test_devices[XT-LK50]
-  DeviceRegistryEntrySnapshot({
-    'area_id': None,
-    'config_entries': <ANY>,
-    'config_entries_subentries': <ANY>,
-    'configuration_url': None,
-    'connections': set({
-    }),
-    'disabled_by': None,
-    'entry_type': None,
-    'hw_version': None,
-    'id': <ANY>,
-    'identifiers': set({
-      tuple(
-        'xthings_cloud',
-        'dev_lock_001',
-      ),
-    }),
-    'labels': set({
-    }),
-    'manufacturer': 'Xthings',
-    'model': 'XT-LK50',
-    'model_id': None,
-    'name': 'Front Door Lock',
-    'name_by_user': None,
-    'primary_config_entry': <ANY>,
-    'serial_number': None,
-    'sw_version': '1.0.0',
-    'via_device_id': None,
-  })
-# ---

tests/components/xthings_cloud/snapshots/test_lock.ambr

@@ -1,52 +0,0 @@
-# serializer version: 1
-# name: test_locks[lock.front_door_lock-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': 'lock',
-    'entity_category': None,
-    'entity_id': 'lock.front_door_lock',
-    'has_entity_name': True,
-    'hidden_by': None,
-    'icon': None,
-    'id': <ANY>,
-    'labels': set({
-    }),
-    'name': None,
-    'object_id_base': None,
-    'options': dict({
-    }),
-    'original_device_class': None,
-    'original_icon': None,
-    'original_name': None,
-    'platform': 'xthings_cloud',
-    'previous_unique_id': None,
-    'suggested_object_id': None,
-    'supported_features': 0,
-    'translation_key': None,
-    'unique_id': 'dev_lock_001',
-    'unit_of_measurement': None,
-  })
-# ---
-# name: test_locks[lock.front_door_lock-state]
-  StateSnapshot({
-    'attributes': ReadOnlyDict({
-      'friendly_name': 'Front Door Lock',
-      'supported_features': <LockEntityFeature: 0>,
-    }),
-    'context': <ANY>,
-    'entity_id': 'lock.front_door_lock',
-    'last_changed': <ANY>,
-    'last_reported': <ANY>,
-    'last_updated': <ANY>,
-    'state': 'locked',
-  })
-# ---

tests/components/xthings_cloud/test_config_flow.py

@@ -27,10 +27,9 @@ from .const import (
 from tests.common import MockConfigEntry
 
 
+@pytest.mark.usefixtures("mock_setup_entry")
 async def test_user_flow_success(
-    hass: HomeAssistant,
-    mock_setup_entry: AsyncMock,
-    mock_api_client: AsyncMock,
+    hass: HomeAssistant, mock_api_client: AsyncMock
 ) -> None:
     """Test successful user login flow."""
     result = await hass.config_entries.flow.async_init(
@@ -62,9 +61,9 @@ async def test_user_flow_success(
         (RuntimeError("unexpected"), "unknown"),
     ],
 )
+@pytest.mark.usefixtures("mock_setup_entry")
 async def test_user_flow_error_and_recover(
     hass: HomeAssistant,
-    mock_setup_entry: AsyncMock,
     mock_api_client: AsyncMock,
     side_effect: Exception,
     expected_error: str,
@@ -91,11 +90,9 @@ async def test_user_flow_error_and_recover(
     assert result["type"] is FlowResultType.CREATE_ENTRY
 
 
+@pytest.mark.usefixtures("mock_setup_entry")
 async def test_user_flow_already_configured(
-    hass: HomeAssistant,
-    mock_setup_entry: AsyncMock,
-    mock_api_client: AsyncMock,
-    mock_config_entry: MockConfigEntry,
+    hass: HomeAssistant, mock_api_client: AsyncMock, mock_config_entry: MockConfigEntry
 ) -> None:
     """Test user flow aborts if same account already configured."""
     mock_config_entry.add_to_hass(hass)

tests/components/xthings_cloud/test_lock.py

@@ -1,105 +0,0 @@
-"""Tests for Xthings Cloud lock platform."""
-
-from unittest.mock import AsyncMock, patch
-
-import pytest
-from syrupy.assertion import SnapshotAssertion
-
-from homeassistant.components.lock import (
-    DOMAIN as LOCK_DOMAIN,
-    SERVICE_LOCK,
-    SERVICE_UNLOCK,
-)
-from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE, Platform
-from homeassistant.core import HomeAssistant
-from homeassistant.helpers import entity_registry as er
-
-from . import get_device_by_id, setup_integration
-
-from tests.common import MockConfigEntry, snapshot_platform
-
-
-async def test_locks(
-    hass: HomeAssistant,
-    mock_config_entry: MockConfigEntry,
-    mock_api_client: AsyncMock,
-    entity_registry: er.EntityRegistry,
-    snapshot: SnapshotAssertion,
-) -> None:
-    """Test lock entities are created correctly."""
-    with patch("homeassistant.components.xthings_cloud.PLATFORMS", [Platform.LOCK]):
-        await setup_integration(hass, mock_config_entry)
-
-        await snapshot_platform(
-            hass, entity_registry, snapshot, mock_config_entry.entry_id
-        )
-
-
-@pytest.mark.parametrize(
-    ("service", "method"),
-    [
-        (SERVICE_LOCK, "async_lock_lock"),
-        (SERVICE_UNLOCK, "async_lock_unlock"),
-    ],
-)
-async def test_lock_lock_unlock(
-    hass: HomeAssistant,
-    mock_config_entry: MockConfigEntry,
-    mock_api_client: AsyncMock,
-    service: str,
-    method: str,
-) -> None:
-    """Test locking and unlocking a lock."""
-    with patch("homeassistant.components.xthings_cloud.PLATFORMS", [Platform.LOCK]):
-        await setup_integration(hass, mock_config_entry)
-
-    await hass.services.async_call(
-        LOCK_DOMAIN,
-        service,
-        {ATTR_ENTITY_ID: "lock.front_door_lock"},
-        blocking=True,
-    )
-    getattr(mock_api_client, method).assert_called_once_with("dev_lock_001")
-
-
-async def test_lock_unavailable_when_offline(
-    hass: HomeAssistant,
-    mock_config_entry: MockConfigEntry,
-    mock_api_client: AsyncMock,
-) -> None:
-    """Test lock shows unavailable when device is offline."""
-    get_device_by_id(mock_api_client, "dev_lock_001")["online"] = False
-    with patch("homeassistant.components.xthings_cloud.PLATFORMS", [Platform.LOCK]):
-        await setup_integration(hass, mock_config_entry)
-
-    state = hass.states.get("lock.front_door_lock")
-    assert state is not None
-    assert state.state == STATE_UNAVAILABLE
-
-
-async def test_updating_state(
-    hass: HomeAssistant,
-    mock_config_entry: MockConfigEntry,
-    mock_api_client: AsyncMock,
-    mock_websocket: AsyncMock,
-) -> None:
-    """Test updating state."""
-    with patch("homeassistant.components.xthings_cloud.PLATFORMS", [Platform.LOCK]):
-        await setup_integration(hass, mock_config_entry)
-
-    state = hass.states.get("lock.front_door_lock")
-    assert state is not None
-    assert state.state == "locked"
-
-    mock_websocket.call_args[1]["on_device_status"](
-        "dev_lock_001",
-        {
-            "locked": False,
-            "jammed": False,
-            "battery": 80,
-        },
-    )
-
-    state = hass.states.get("lock.front_door_lock")
-    assert state is not None
-    assert state.state == "unlocked"