mirror of
https://github.com/home-assistant/core.git
synced 2025-07-22 20:57:21 +00:00
Add missing hass type hint in component tests (w) (#124284)
This commit is contained in:
parent
e6eedc0717
commit
c2dc4ef215
@ -19,7 +19,9 @@ from homeassistant.const import ATTR_ENTITY_ID, ATTR_TEMPERATURE, ENTITY_MATCH_A
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
|
||||
async def async_set_away_mode(hass, away_mode, entity_id=ENTITY_MATCH_ALL):
|
||||
async def async_set_away_mode(
|
||||
hass: HomeAssistant, away_mode: bool, entity_id: str = ENTITY_MATCH_ALL
|
||||
) -> None:
|
||||
"""Turn all or specified water_heater devices away mode on."""
|
||||
data = {ATTR_AWAY_MODE: away_mode}
|
||||
|
||||
@ -30,8 +32,11 @@ async def async_set_away_mode(hass, away_mode, entity_id=ENTITY_MATCH_ALL):
|
||||
|
||||
|
||||
async def async_set_temperature(
|
||||
hass, temperature=None, entity_id=ENTITY_MATCH_ALL, operation_mode=None
|
||||
):
|
||||
hass: HomeAssistant,
|
||||
temperature: float,
|
||||
entity_id: str = ENTITY_MATCH_ALL,
|
||||
operation_mode: str | None = None,
|
||||
) -> None:
|
||||
"""Set new target temperature."""
|
||||
kwargs = {
|
||||
key: value
|
||||
@ -48,7 +53,9 @@ async def async_set_temperature(
|
||||
)
|
||||
|
||||
|
||||
async def async_set_operation_mode(hass, operation_mode, entity_id=ENTITY_MATCH_ALL):
|
||||
async def async_set_operation_mode(
|
||||
hass: HomeAssistant, operation_mode: str, entity_id: str = ENTITY_MATCH_ALL
|
||||
) -> None:
|
||||
"""Set new target operation mode."""
|
||||
data = {ATTR_OPERATION_MODE: operation_mode}
|
||||
|
||||
|
@ -319,7 +319,9 @@ async def test_ws_webhook(
|
||||
|
||||
received = []
|
||||
|
||||
async def handler(hass, webhook_id, request):
|
||||
async def handler(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> web.Response:
|
||||
"""Handle a webhook."""
|
||||
received.append(request)
|
||||
return web.json_response({"from": "handler"})
|
||||
|
@ -2,6 +2,7 @@
|
||||
|
||||
from homeassistant.components.webostv.const import DOMAIN
|
||||
from homeassistant.const import CONF_CLIENT_SECRET, CONF_HOST
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from .const import CLIENT_KEY, FAKE_UUID, HOST, TV_NAME
|
||||
@ -9,7 +10,9 @@ from .const import CLIENT_KEY, FAKE_UUID, HOST, TV_NAME
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def setup_webostv(hass, unique_id=FAKE_UUID):
|
||||
async def setup_webostv(
|
||||
hass: HomeAssistant, unique_id: str | None = FAKE_UUID
|
||||
) -> MockConfigEntry:
|
||||
"""Initialize webostv and media_player for tests."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
|
@ -1,5 +1,7 @@
|
||||
"""Test decorators."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components import http, websocket_api
|
||||
@ -19,24 +21,40 @@ async def test_async_response_request_context(
|
||||
|
||||
@websocket_api.websocket_command({"type": "test-get-request-executor"})
|
||||
@websocket_api.async_response
|
||||
async def executor_get_request(hass, connection, msg):
|
||||
async def executor_get_request(
|
||||
hass: HomeAssistant,
|
||||
connection: websocket_api.ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
) -> None:
|
||||
handle_request(
|
||||
await hass.async_add_executor_job(http.current_request.get), connection, msg
|
||||
)
|
||||
|
||||
@websocket_api.websocket_command({"type": "test-get-request-async"})
|
||||
@websocket_api.async_response
|
||||
async def async_get_request(hass, connection, msg):
|
||||
async def async_get_request(
|
||||
hass: HomeAssistant,
|
||||
connection: websocket_api.ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
) -> None:
|
||||
handle_request(http.current_request.get(), connection, msg)
|
||||
|
||||
@websocket_api.websocket_command({"type": "test-get-request"})
|
||||
def get_request(hass, connection, msg):
|
||||
def get_request(
|
||||
hass: HomeAssistant,
|
||||
connection: websocket_api.ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
) -> None:
|
||||
handle_request(http.current_request.get(), connection, msg)
|
||||
|
||||
@websocket_api.websocket_command(
|
||||
{"type": "test-get-request-with-arg", vol.Required("arg"): str}
|
||||
)
|
||||
def get_with_arg_request(hass, connection, msg):
|
||||
def get_with_arg_request(
|
||||
hass: HomeAssistant,
|
||||
connection: websocket_api.ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
) -> None:
|
||||
handle_request(http.current_request.get(), connection, msg)
|
||||
|
||||
websocket_api.async_register_command(hass, executor_get_request)
|
||||
@ -145,7 +163,11 @@ async def test_supervisor_only(hass: HomeAssistant, websocket_client) -> None:
|
||||
|
||||
@websocket_api.ws_require_user(only_supervisor=True)
|
||||
@websocket_api.websocket_command({"type": "test-require-supervisor-user"})
|
||||
def require_supervisor_request(hass, connection, msg):
|
||||
def require_supervisor_request(
|
||||
hass: HomeAssistant,
|
||||
connection: websocket_api.ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
) -> None:
|
||||
connection.send_result(msg["id"])
|
||||
|
||||
websocket_api.async_register_command(hass, require_supervisor_request)
|
||||
|
@ -4,7 +4,11 @@ This is not a test module. These test methods are used by the platform test modu
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Callable, Coroutine
|
||||
import threading
|
||||
from typing import Any
|
||||
|
||||
import pywemo
|
||||
|
||||
from homeassistant.components.homeassistant import DOMAIN as HA_DOMAIN
|
||||
from homeassistant.components.wemo.coordinator import async_get_coordinator
|
||||
@ -17,6 +21,7 @@ from homeassistant.const import (
|
||||
STATE_UNAVAILABLE,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
|
||||
@ -40,7 +45,12 @@ def _perform_async_update(coordinator):
|
||||
return async_callback
|
||||
|
||||
|
||||
async def _async_multiple_call_helper(hass, pywemo_device, call1, call2):
|
||||
async def _async_multiple_call_helper(
|
||||
hass: HomeAssistant,
|
||||
pywemo_device: pywemo.WeMoDevice,
|
||||
call1: Callable[[], Coroutine[Any, Any, None]],
|
||||
call2: Callable[[], Coroutine[Any, Any, None]],
|
||||
) -> None:
|
||||
"""Create two calls (call1 & call2) in parallel; verify only one polls the device.
|
||||
|
||||
There should only be one poll on the device at a time. Any parallel updates
|
||||
@ -87,7 +97,7 @@ async def _async_multiple_call_helper(hass, pywemo_device, call1, call2):
|
||||
|
||||
|
||||
async def test_async_update_locked_callback_and_update(
|
||||
hass: HomeAssistant, pywemo_device, wemo_entity
|
||||
hass: HomeAssistant, pywemo_device: pywemo.WeMoDevice, wemo_entity: er.RegistryEntry
|
||||
) -> None:
|
||||
"""Test that a callback and a state update request can't both happen at the same time.
|
||||
|
||||
@ -102,7 +112,7 @@ async def test_async_update_locked_callback_and_update(
|
||||
|
||||
|
||||
async def test_async_update_locked_multiple_updates(
|
||||
hass: HomeAssistant, pywemo_device, wemo_entity
|
||||
hass: HomeAssistant, pywemo_device: pywemo.WeMoDevice, wemo_entity: er.RegistryEntry
|
||||
) -> None:
|
||||
"""Test that two hass async_update state updates do not proceed at the same time."""
|
||||
coordinator = async_get_coordinator(hass, wemo_entity.device_id)
|
||||
@ -112,7 +122,7 @@ async def test_async_update_locked_multiple_updates(
|
||||
|
||||
|
||||
async def test_async_update_locked_multiple_callbacks(
|
||||
hass: HomeAssistant, pywemo_device, wemo_entity
|
||||
hass: HomeAssistant, pywemo_device: pywemo.WeMoDevice, wemo_entity: er.RegistryEntry
|
||||
) -> None:
|
||||
"""Test that two device callback state updates do not proceed at the same time."""
|
||||
coordinator = async_get_coordinator(hass, wemo_entity.device_id)
|
||||
@ -158,24 +168,33 @@ class EntityTestHelpers:
|
||||
"""Common state update helpers."""
|
||||
|
||||
async def test_async_update_locked_multiple_updates(
|
||||
self, hass, pywemo_device, wemo_entity
|
||||
):
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
pywemo_device: pywemo.WeMoDevice,
|
||||
wemo_entity: er.RegistryEntry,
|
||||
) -> None:
|
||||
"""Test that two hass async_update state updates do not proceed at the same time."""
|
||||
await test_async_update_locked_multiple_updates(
|
||||
hass, pywemo_device, wemo_entity
|
||||
)
|
||||
|
||||
async def test_async_update_locked_multiple_callbacks(
|
||||
self, hass, pywemo_device, wemo_entity
|
||||
):
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
pywemo_device: pywemo.WeMoDevice,
|
||||
wemo_entity: er.RegistryEntry,
|
||||
) -> None:
|
||||
"""Test that two device callback state updates do not proceed at the same time."""
|
||||
await test_async_update_locked_multiple_callbacks(
|
||||
hass, pywemo_device, wemo_entity
|
||||
)
|
||||
|
||||
async def test_async_update_locked_callback_and_update(
|
||||
self, hass, pywemo_device, wemo_entity
|
||||
):
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
pywemo_device: pywemo.WeMoDevice,
|
||||
wemo_entity: er.RegistryEntry,
|
||||
) -> None:
|
||||
"""Test that a callback and a state update request can't both happen at the same time.
|
||||
|
||||
When a state update is received via a callback from the device at the same time
|
||||
|
@ -1,6 +1,7 @@
|
||||
"""Tests for the Wemo binary_sensor entity."""
|
||||
|
||||
import pytest
|
||||
import pywemo
|
||||
from pywemo import StandbyState
|
||||
|
||||
from homeassistant.components.homeassistant import (
|
||||
@ -12,6 +13,8 @@ from homeassistant.components.wemo.binary_sensor import (
|
||||
MakerBinarySensor,
|
||||
)
|
||||
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from .entity_test_helpers import EntityTestHelpers
|
||||
@ -26,8 +29,12 @@ class TestMotion(EntityTestHelpers):
|
||||
return "Motion"
|
||||
|
||||
async def test_binary_sensor_registry_state_callback(
|
||||
self, hass, pywemo_registry, pywemo_device, wemo_entity
|
||||
):
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
pywemo_registry: pywemo.SubscriptionRegistry,
|
||||
pywemo_device: pywemo.WeMoDevice,
|
||||
wemo_entity: er.RegistryEntry,
|
||||
) -> None:
|
||||
"""Verify that the binary_sensor receives state updates from the registry."""
|
||||
# On state.
|
||||
pywemo_device.get_state.return_value = 1
|
||||
@ -42,8 +49,12 @@ class TestMotion(EntityTestHelpers):
|
||||
assert hass.states.get(wemo_entity.entity_id).state == STATE_OFF
|
||||
|
||||
async def test_binary_sensor_update_entity(
|
||||
self, hass, pywemo_registry, pywemo_device, wemo_entity
|
||||
):
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
pywemo_registry: pywemo.SubscriptionRegistry,
|
||||
pywemo_device: pywemo.WeMoDevice,
|
||||
wemo_entity: er.RegistryEntry,
|
||||
) -> None:
|
||||
"""Verify that the binary_sensor performs state updates."""
|
||||
await async_setup_component(hass, HA_DOMAIN, {})
|
||||
|
||||
@ -82,8 +93,12 @@ class TestMaker(EntityTestHelpers):
|
||||
return MakerBinarySensor._name_suffix.lower()
|
||||
|
||||
async def test_registry_state_callback(
|
||||
self, hass, pywemo_registry, pywemo_device, wemo_entity
|
||||
):
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
pywemo_registry: pywemo.SubscriptionRegistry,
|
||||
pywemo_device: pywemo.WeMoDevice,
|
||||
wemo_entity: er.RegistryEntry,
|
||||
) -> None:
|
||||
"""Verify that the binary_sensor receives state updates from the registry."""
|
||||
# On state.
|
||||
pywemo_device.sensor_state = 0
|
||||
@ -112,8 +127,12 @@ class TestInsight(EntityTestHelpers):
|
||||
return InsightBinarySensor._name_suffix.lower()
|
||||
|
||||
async def test_registry_state_callback(
|
||||
self, hass, pywemo_registry, pywemo_device, wemo_entity
|
||||
):
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
pywemo_registry: pywemo.SubscriptionRegistry,
|
||||
pywemo_device: pywemo.WeMoDevice,
|
||||
wemo_entity: er.RegistryEntry,
|
||||
) -> None:
|
||||
"""Verify that the binary_sensor receives state updates from the registry."""
|
||||
# On state.
|
||||
pywemo_device.get_state.return_value = 1
|
||||
|
@ -3,9 +3,10 @@
|
||||
import asyncio
|
||||
from dataclasses import asdict
|
||||
from datetime import timedelta
|
||||
from unittest.mock import call, patch
|
||||
from unittest.mock import _Call, call, patch
|
||||
|
||||
import pytest
|
||||
import pywemo
|
||||
from pywemo.exceptions import ActionException, PyWeMoException
|
||||
from pywemo.subscribe import EVENT_TYPE_LONG_PRESS
|
||||
|
||||
@ -14,7 +15,7 @@ from homeassistant.components.wemo import CONF_DISCOVERY, CONF_STATIC
|
||||
from homeassistant.components.wemo.const import DOMAIN, WEMO_SUBSCRIPTION_EVENT
|
||||
from homeassistant.components.wemo.coordinator import Options, async_get_coordinator
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
from homeassistant.helpers.update_coordinator import UpdateFailed
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.util.dt import utcnow
|
||||
@ -248,14 +249,14 @@ class TestInsight:
|
||||
)
|
||||
async def test_should_poll(
|
||||
self,
|
||||
hass,
|
||||
subscribed,
|
||||
state,
|
||||
expected_calls,
|
||||
wemo_entity,
|
||||
pywemo_device,
|
||||
pywemo_registry,
|
||||
):
|
||||
hass: HomeAssistant,
|
||||
subscribed: bool,
|
||||
state: int,
|
||||
expected_calls: list[_Call],
|
||||
wemo_entity: er.RegistryEntry,
|
||||
pywemo_device: pywemo.WeMoDevice,
|
||||
pywemo_registry: pywemo.SubscriptionRegistry,
|
||||
) -> None:
|
||||
"""Validate the should_poll returns the correct value."""
|
||||
pywemo_registry.is_subscribed.return_value = subscribed
|
||||
pywemo_device.get_state.reset_mock()
|
||||
|
@ -16,6 +16,7 @@ from homeassistant.const import (
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import async_get_device_automations, async_mock_service
|
||||
@ -29,7 +30,9 @@ def pywemo_model():
|
||||
return "LightSwitchLongPress"
|
||||
|
||||
|
||||
async def setup_automation(hass, device_id, trigger_type):
|
||||
async def setup_automation(
|
||||
hass: HomeAssistant, device_id: str, trigger_type: str
|
||||
) -> None:
|
||||
"""Set up an automation trigger for testing triggering."""
|
||||
return await async_setup_component(
|
||||
hass,
|
||||
@ -96,7 +99,9 @@ async def test_get_triggers(hass: HomeAssistant, wemo_entity) -> None:
|
||||
assert triggers == unordered(expected_triggers)
|
||||
|
||||
|
||||
async def test_fires_on_long_press(hass: HomeAssistant, wemo_entity) -> None:
|
||||
async def test_fires_on_long_press(
|
||||
hass: HomeAssistant, wemo_entity: er.RegistryEntry
|
||||
) -> None:
|
||||
"""Test wemo long press trigger firing."""
|
||||
assert await setup_automation(hass, wemo_entity.device_id, EVENT_TYPE_LONG_PRESS)
|
||||
calls = async_mock_service(hass, "test", "automation")
|
||||
|
@ -2,6 +2,9 @@
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from .conftest import MOCK_INSIGHT_CURRENT_WATTS, MOCK_INSIGHT_TODAY_KWH
|
||||
from .entity_test_helpers import EntityTestHelpers
|
||||
|
||||
@ -24,7 +27,7 @@ class InsightTestTemplate(EntityTestHelpers):
|
||||
"""Select the appropriate entity for the test."""
|
||||
return cls.ENTITY_ID_SUFFIX
|
||||
|
||||
def test_state(self, hass, wemo_entity):
|
||||
def test_state(self, hass: HomeAssistant, wemo_entity: er.RegistryEntry) -> None:
|
||||
"""Test the sensor state."""
|
||||
assert hass.states.get(wemo_entity.entity_id).state == self.EXPECTED_STATE_VALUE
|
||||
|
||||
|
@ -1,9 +1,10 @@
|
||||
"""Tests for the WiZ Platform integration."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from contextlib import contextmanager
|
||||
from collections.abc import Callable, Generator
|
||||
from contextlib import _GeneratorContextManager, contextmanager
|
||||
from copy import deepcopy
|
||||
import json
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from pywizlight import SCENES, BulbType, PilotParser, wizlight
|
||||
@ -194,7 +195,11 @@ async def setup_integration(hass: HomeAssistant) -> MockConfigEntry:
|
||||
return entry
|
||||
|
||||
|
||||
def _mocked_wizlight(device, extended_white_range, bulb_type) -> wizlight:
|
||||
def _mocked_wizlight(
|
||||
device: dict[str, Any] | None,
|
||||
extended_white_range: list[int] | None,
|
||||
bulb_type: BulbType | None,
|
||||
) -> wizlight:
|
||||
bulb = MagicMock(auto_spec=wizlight, name="Mocked wizlight")
|
||||
|
||||
async def _save_setup_callback(callback: Callable) -> None:
|
||||
@ -228,9 +233,13 @@ def _mocked_wizlight(device, extended_white_range, bulb_type) -> wizlight:
|
||||
return bulb
|
||||
|
||||
|
||||
def _patch_wizlight(device=None, extended_white_range=None, bulb_type=None):
|
||||
def _patch_wizlight(
|
||||
device: dict[str, Any] | None = None,
|
||||
extended_white_range: list[int] | None = None,
|
||||
bulb_type: BulbType | None = None,
|
||||
) -> _GeneratorContextManager:
|
||||
@contextmanager
|
||||
def _patcher():
|
||||
def _patcher() -> Generator[None]:
|
||||
bulb = device or _mocked_wizlight(device, extended_white_range, bulb_type)
|
||||
with (
|
||||
patch("homeassistant.components.wiz.wizlight", return_value=bulb),
|
||||
@ -244,9 +253,9 @@ def _patch_wizlight(device=None, extended_white_range=None, bulb_type=None):
|
||||
return _patcher()
|
||||
|
||||
|
||||
def _patch_discovery():
|
||||
def _patch_discovery() -> _GeneratorContextManager[None]:
|
||||
@contextmanager
|
||||
def _patcher():
|
||||
def _patcher() -> Generator[None]:
|
||||
with patch(
|
||||
"homeassistant.components.wiz.discovery.find_wizlights",
|
||||
return_value=[DiscoveredBulb(FAKE_IP, FAKE_MAC)],
|
||||
@ -257,8 +266,12 @@ def _patch_discovery():
|
||||
|
||||
|
||||
async def async_setup_integration(
|
||||
hass, wizlight=None, device=None, extended_white_range=None, bulb_type=None
|
||||
):
|
||||
hass: HomeAssistant,
|
||||
wizlight: wizlight | None = None,
|
||||
device: dict[str, Any] | None = None,
|
||||
extended_white_range: list[int] | None = None,
|
||||
bulb_type: BulbType | None = None,
|
||||
) -> tuple[wizlight, MockConfigEntry]:
|
||||
"""Set up the integration with a mock device."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
@ -273,7 +286,9 @@ async def async_setup_integration(
|
||||
return bulb, entry
|
||||
|
||||
|
||||
async def async_push_update(hass, device, params):
|
||||
async def async_push_update(
|
||||
hass: HomeAssistant, device: wizlight, params: dict[str, Any]
|
||||
) -> None:
|
||||
"""Push an update to the device."""
|
||||
device.state = PilotParser(params)
|
||||
device.status = params.get("state")
|
||||
|
@ -1,6 +1,7 @@
|
||||
"""The tests for WS66i Media player platform."""
|
||||
|
||||
from collections import defaultdict
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
@ -170,7 +171,9 @@ async def _setup_ws66i_with_options(hass: HomeAssistant, ws66i) -> MockConfigEnt
|
||||
return config_entry
|
||||
|
||||
|
||||
async def _call_media_player_service(hass, name, data):
|
||||
async def _call_media_player_service(
|
||||
hass: HomeAssistant, name: str, data: dict[str, Any]
|
||||
) -> None:
|
||||
await hass.services.async_call(
|
||||
MEDIA_PLAYER_DOMAIN, name, service_data=data, blocking=True
|
||||
)
|
||||
|
Loading…
x
Reference in New Issue
Block a user