Fix lingering timer in ZHA tests (#91688)

* Fix lingering timer in ZHA discovery tests

* Add type hints

* Also update gateway tests
This commit is contained in:
epenet 2023-04-19 19:00:25 +02:00 committed by GitHub
parent d70ae8afc5
commit 1797dca0b8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 16 additions and 2 deletions

View File

@ -1,5 +1,7 @@
"""Test ZHA device discovery.""" """Test ZHA device discovery."""
from collections.abc import Callable
import re import re
from typing import Any
from unittest import mock from unittest import mock
from unittest.mock import AsyncMock, Mock, patch from unittest.mock import AsyncMock, Mock, patch
@ -16,6 +18,7 @@ import zigpy.zcl.foundation as zcl_f
import homeassistant.components.zha.binary_sensor import homeassistant.components.zha.binary_sensor
import homeassistant.components.zha.core.cluster_handlers as cluster_handlers import homeassistant.components.zha.core.cluster_handlers as cluster_handlers
import homeassistant.components.zha.core.const as zha_const import homeassistant.components.zha.core.const as zha_const
from homeassistant.components.zha.core.device import ZHADevice
import homeassistant.components.zha.core.discovery as disc import homeassistant.components.zha.core.discovery as disc
from homeassistant.components.zha.core.endpoint import Endpoint from homeassistant.components.zha.core.endpoint import Endpoint
import homeassistant.components.zha.core.registries as zha_regs import homeassistant.components.zha.core.registries as zha_regs
@ -301,7 +304,9 @@ def test_discover_probe_single_cluster() -> None:
@pytest.mark.parametrize("device_info", DEVICES) @pytest.mark.parametrize("device_info", DEVICES)
async def test_discover_endpoint( async def test_discover_endpoint(
device_info, zha_device_mock, hass: HomeAssistant device_info: dict[str, Any],
zha_device_mock: Callable[..., ZHADevice],
hass: HomeAssistant,
) -> None: ) -> None:
"""Test device discovery.""" """Test device discovery."""
@ -353,6 +358,8 @@ async def test_discover_endpoint(
ent_info[DEV_SIG_CLUSTER_HANDLERS] ent_info[DEV_SIG_CLUSTER_HANDLERS]
) )
device.async_cleanup_handles()
def _ch_mock(cluster): def _ch_mock(cluster):
"""Return mock of a cluster_handler with a cluster.""" """Return mock of a cluster_handler with a cluster."""

View File

@ -1,5 +1,6 @@
"""Test ZHA Gateway.""" """Test ZHA Gateway."""
import asyncio import asyncio
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
import pytest import pytest
@ -8,6 +9,7 @@ import zigpy.profiles.zha as zha
import zigpy.zcl.clusters.general as general import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.lighting as lighting import zigpy.zcl.clusters.lighting as lighting
from homeassistant.components.zha.core.device import ZHADevice
from homeassistant.components.zha.core.group import GroupMember from homeassistant.components.zha.core.group import GroupMember
from homeassistant.const import Platform from homeassistant.const import Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -238,7 +240,10 @@ async def test_gateway_create_group_with_id(
], ],
) )
async def test_gateway_initialize_success( async def test_gateway_initialize_success(
startup, hass: HomeAssistant, device_light_1, coordinator startup: list[Any],
hass: HomeAssistant,
device_light_1: ZHADevice,
coordinator: ZHADevice,
) -> None: ) -> None:
"""Test ZHA initializing the gateway successfully.""" """Test ZHA initializing the gateway successfully."""
zha_gateway = get_zha_gateway(hass) zha_gateway = get_zha_gateway(hass)
@ -253,6 +258,8 @@ async def test_gateway_initialize_success(
assert mock_new.call_count == len(startup) assert mock_new.call_count == len(startup)
device_light_1.async_cleanup_handles()
@patch("homeassistant.components.zha.core.gateway.STARTUP_FAILURE_DELAY_S", 0.01) @patch("homeassistant.components.zha.core.gateway.STARTUP_FAILURE_DELAY_S", 0.01)
async def test_gateway_initialize_failure( async def test_gateway_initialize_failure(