mirror of
https://github.com/home-assistant/core.git
synced 2025-07-22 04:37:06 +00:00
Add missing hass type hint in component tests (s) (#124272)
This commit is contained in:
parent
d961e20b15
commit
f66b539027
@ -6,11 +6,12 @@ components. Instead call the service directly.
|
|||||||
|
|
||||||
from homeassistant.components.scene import DOMAIN
|
from homeassistant.components.scene import DOMAIN
|
||||||
from homeassistant.const import ATTR_ENTITY_ID, ENTITY_MATCH_ALL, SERVICE_TURN_ON
|
from homeassistant.const import ATTR_ENTITY_ID, ENTITY_MATCH_ALL, SERVICE_TURN_ON
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.loader import bind_hass
|
from homeassistant.loader import bind_hass
|
||||||
|
|
||||||
|
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def activate(hass, entity_id=ENTITY_MATCH_ALL):
|
def activate(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
|
||||||
"""Activate a scene."""
|
"""Activate a scene."""
|
||||||
data = {}
|
data = {}
|
||||||
|
|
||||||
|
@ -222,7 +222,7 @@ async def test_restore_state_does_not_restore_unavailable(
|
|||||||
assert hass.states.get("scene.test").state == STATE_UNKNOWN
|
assert hass.states.get("scene.test").state == STATE_UNKNOWN
|
||||||
|
|
||||||
|
|
||||||
async def activate(hass, entity_id=ENTITY_MATCH_ALL):
|
async def activate(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
|
||||||
"""Activate a scene."""
|
"""Activate a scene."""
|
||||||
data = {}
|
data = {}
|
||||||
|
|
||||||
@ -241,7 +241,9 @@ async def test_services_registered(hass: HomeAssistant) -> None:
|
|||||||
assert hass.services.has_service("scene", "apply")
|
assert hass.services.has_service("scene", "apply")
|
||||||
|
|
||||||
|
|
||||||
async def setup_lights(hass, entities):
|
async def setup_lights(
|
||||||
|
hass: HomeAssistant, entities: list[MockLight]
|
||||||
|
) -> tuple[MockLight, MockLight]:
|
||||||
"""Set up the light component."""
|
"""Set up the light component."""
|
||||||
assert await async_setup_component(
|
assert await async_setup_component(
|
||||||
hass, light.DOMAIN, {light.DOMAIN: {"platform": "test"}}
|
hass, light.DOMAIN, {light.DOMAIN: {"platform": "test"}}
|
||||||
@ -261,7 +263,7 @@ async def setup_lights(hass, entities):
|
|||||||
return light_1, light_2
|
return light_1, light_2
|
||||||
|
|
||||||
|
|
||||||
async def turn_off_lights(hass, entity_ids):
|
async def turn_off_lights(hass: HomeAssistant, entity_ids: list[str]) -> None:
|
||||||
"""Turn lights off."""
|
"""Turn lights off."""
|
||||||
await hass.services.async_call(
|
await hass.services.async_call(
|
||||||
"light",
|
"light",
|
||||||
|
@ -5104,7 +5104,9 @@ async def async_record_meter_state(
|
|||||||
return states
|
return states
|
||||||
|
|
||||||
|
|
||||||
async def async_record_states_partially_unavailable(hass, zero, entity_id, attributes):
|
async def async_record_states_partially_unavailable(
|
||||||
|
hass: HomeAssistant, zero: datetime, entity_id: str, attributes: dict[str, Any]
|
||||||
|
) -> tuple[datetime, dict[str, list[State]]]:
|
||||||
"""Record some test states.
|
"""Record some test states.
|
||||||
|
|
||||||
We inject a bunch of state updates temperature sensors.
|
We inject a bunch of state updates temperature sensors.
|
||||||
|
@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
import json
|
import json
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from smart_meter_texas.const import (
|
from smart_meter_texas.const import (
|
||||||
@ -23,6 +24,7 @@ from homeassistant.core import HomeAssistant
|
|||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
|
||||||
from tests.common import MockConfigEntry, load_fixture
|
from tests.common import MockConfigEntry, load_fixture
|
||||||
|
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||||
|
|
||||||
TEST_ENTITY_ID = "sensor.electric_meter_123456789"
|
TEST_ENTITY_ID = "sensor.electric_meter_123456789"
|
||||||
|
|
||||||
@ -33,14 +35,23 @@ def load_smt_fixture(name):
|
|||||||
return json.loads(json_fixture)
|
return json.loads(json_fixture)
|
||||||
|
|
||||||
|
|
||||||
async def setup_integration(hass, config_entry, aioclient_mock, **kwargs):
|
async def setup_integration(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
config_entry: MockConfigEntry,
|
||||||
|
aioclient_mock: AiohttpClientMocker,
|
||||||
|
**kwargs: Any,
|
||||||
|
) -> None:
|
||||||
"""Initialize the Smart Meter Texas integration for testing."""
|
"""Initialize the Smart Meter Texas integration for testing."""
|
||||||
mock_connection(aioclient_mock, **kwargs)
|
mock_connection(aioclient_mock, **kwargs)
|
||||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
|
||||||
async def refresh_data(hass, config_entry, aioclient_mock):
|
async def refresh_data(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
config_entry: MockConfigEntry,
|
||||||
|
aioclient_mock: AiohttpClientMocker,
|
||||||
|
) -> None:
|
||||||
"""Request a DataUpdateCoordinator refresh."""
|
"""Request a DataUpdateCoordinator refresh."""
|
||||||
mock_connection(aioclient_mock)
|
mock_connection(aioclient_mock)
|
||||||
await async_setup_component(hass, HA_DOMAIN, {})
|
await async_setup_component(hass, HA_DOMAIN, {})
|
||||||
|
@ -1,6 +1,9 @@
|
|||||||
"""Tests for the SmartThings component init module."""
|
"""Tests for the SmartThings component init module."""
|
||||||
|
|
||||||
|
from collections.abc import Callable, Coroutine
|
||||||
|
from datetime import datetime, timedelta
|
||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
|
from typing import Any
|
||||||
from unittest.mock import Mock, patch
|
from unittest.mock import Mock, patch
|
||||||
from uuid import uuid4
|
from uuid import uuid4
|
||||||
|
|
||||||
@ -419,7 +422,11 @@ async def test_broker_regenerates_token(hass: HomeAssistant, config_entry) -> No
|
|||||||
stored_action = None
|
stored_action = None
|
||||||
config_entry.add_to_hass(hass)
|
config_entry.add_to_hass(hass)
|
||||||
|
|
||||||
def async_track_time_interval(hass, action, interval):
|
def async_track_time_interval(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
action: Callable[[datetime], Coroutine[Any, Any, None] | None],
|
||||||
|
interval: timedelta,
|
||||||
|
) -> None:
|
||||||
nonlocal stored_action
|
nonlocal stored_action
|
||||||
stored_action = action
|
stored_action = action
|
||||||
|
|
||||||
|
@ -45,7 +45,7 @@ async def test_form(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
|
|||||||
assert len(mock_setup_entry.mock_calls) == 1
|
assert len(mock_setup_entry.mock_calls) == 1
|
||||||
|
|
||||||
|
|
||||||
def init_config_flow(hass):
|
def init_config_flow(hass: HomeAssistant) -> config_flow.SolarLogConfigFlow:
|
||||||
"""Init a configuration flow."""
|
"""Init a configuration flow."""
|
||||||
flow = config_flow.SolarLogConfigFlow()
|
flow = config_flow.SolarLogConfigFlow()
|
||||||
flow.hass = hass
|
flow.hass = hass
|
||||||
|
@ -6,7 +6,12 @@ from unittest.mock import patch
|
|||||||
|
|
||||||
from homeassistant.components import ssdp
|
from homeassistant.components import ssdp
|
||||||
from homeassistant.components.songpal.const import CONF_ENDPOINT, DOMAIN
|
from homeassistant.components.songpal.const import CONF_ENDPOINT, DOMAIN
|
||||||
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_SSDP, SOURCE_USER
|
from homeassistant.config_entries import (
|
||||||
|
SOURCE_IMPORT,
|
||||||
|
SOURCE_SSDP,
|
||||||
|
SOURCE_USER,
|
||||||
|
ConfigFlowResult,
|
||||||
|
)
|
||||||
from homeassistant.const import CONF_HOST, CONF_NAME
|
from homeassistant.const import CONF_HOST, CONF_NAME
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.data_entry_flow import FlowResultType
|
from homeassistant.data_entry_flow import FlowResultType
|
||||||
@ -42,7 +47,7 @@ SSDP_DATA = ssdp.SsdpServiceInfo(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def _flow_next(hass, flow_id):
|
def _flow_next(hass: HomeAssistant, flow_id: str) -> ConfigFlowResult:
|
||||||
return next(
|
return next(
|
||||||
flow
|
flow
|
||||||
for flow in hass.config_entries.flow.async_progress()
|
for flow in hass.config_entries.flow.async_progress()
|
||||||
@ -143,7 +148,7 @@ async def test_flow_import_without_name(hass: HomeAssistant) -> None:
|
|||||||
mocked_device.get_interface_information.assert_called_once()
|
mocked_device.get_interface_information.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
def _create_mock_config_entry(hass):
|
def _create_mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
|
||||||
MockConfigEntry(
|
MockConfigEntry(
|
||||||
domain=DOMAIN,
|
domain=DOMAIN,
|
||||||
unique_id="uuid:0000",
|
unique_id="uuid:0000",
|
||||||
|
@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Any
|
||||||
from unittest.mock import AsyncMock, MagicMock, call, patch
|
from unittest.mock import AsyncMock, MagicMock, call, patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
@ -54,12 +55,12 @@ SUPPORT_SONGPAL = (
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def _get_attributes(hass):
|
def _get_attributes(hass: HomeAssistant) -> dict[str, Any]:
|
||||||
state = hass.states.get(ENTITY_ID)
|
state = hass.states.get(ENTITY_ID)
|
||||||
return state.as_dict()["attributes"]
|
return state.as_dict()["attributes"]
|
||||||
|
|
||||||
|
|
||||||
async def _call(hass, service, **argv):
|
async def _call(hass: HomeAssistant, service: str, **argv: Any) -> None:
|
||||||
await hass.services.async_call(
|
await hass.services.async_call(
|
||||||
media_player.DOMAIN,
|
media_player.DOMAIN,
|
||||||
service,
|
service,
|
||||||
|
@ -295,7 +295,13 @@ def silent_ssdp_scanner() -> Generator[None]:
|
|||||||
def discover_fixture(soco):
|
def discover_fixture(soco):
|
||||||
"""Create a mock soco discover fixture."""
|
"""Create a mock soco discover fixture."""
|
||||||
|
|
||||||
def do_callback(hass, callback, *args, **kwargs):
|
def do_callback(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
callback: Callable[
|
||||||
|
[ssdp.SsdpServiceInfo, ssdp.SsdpChange], Coroutine[Any, Any, None] | None
|
||||||
|
],
|
||||||
|
match_dict: dict[str, str] | None = None,
|
||||||
|
) -> MagicMock:
|
||||||
callback(
|
callback(
|
||||||
ssdp.SsdpServiceInfo(
|
ssdp.SsdpServiceInfo(
|
||||||
ssdp_location=f"http://{soco.ip_address}/",
|
ssdp_location=f"http://{soco.ip_address}/",
|
||||||
|
@ -283,7 +283,12 @@ class MockPyAv:
|
|||||||
return self.container
|
return self.container
|
||||||
|
|
||||||
|
|
||||||
def run_worker(hass, stream, stream_source, stream_settings=None):
|
def run_worker(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
stream: Stream,
|
||||||
|
stream_source: str,
|
||||||
|
stream_settings: StreamSettings | None = None,
|
||||||
|
) -> None:
|
||||||
"""Run the stream worker under test."""
|
"""Run the stream worker under test."""
|
||||||
stream_state = StreamState(hass, stream.outputs, stream._diagnostics)
|
stream_state = StreamState(hass, stream.outputs, stream._diagnostics)
|
||||||
stream_worker(
|
stream_worker(
|
||||||
@ -296,7 +301,12 @@ def run_worker(hass, stream, stream_source, stream_settings=None):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
async def async_decode_stream(hass, packets, py_av=None, stream_settings=None):
|
async def async_decode_stream(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
packets: PacketSequence,
|
||||||
|
py_av: MockPyAv | None = None,
|
||||||
|
stream_settings: StreamSettings | None = None,
|
||||||
|
) -> FakePyAvBuffer:
|
||||||
"""Start a stream worker that decodes incoming stream packets into output segments."""
|
"""Start a stream worker that decodes incoming stream packets into output segments."""
|
||||||
stream = Stream(
|
stream = Stream(
|
||||||
hass,
|
hass,
|
||||||
|
@ -15,28 +15,31 @@ from homeassistant.const import (
|
|||||||
STATE_OFF,
|
STATE_OFF,
|
||||||
STATE_ON,
|
STATE_ON,
|
||||||
)
|
)
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.loader import bind_hass
|
from homeassistant.loader import bind_hass
|
||||||
|
|
||||||
|
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def turn_on(hass, entity_id=ENTITY_MATCH_ALL):
|
def turn_on(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
|
||||||
"""Turn all or specified switch on."""
|
"""Turn all or specified switch on."""
|
||||||
hass.add_job(async_turn_on, hass, entity_id)
|
hass.add_job(async_turn_on, hass, entity_id)
|
||||||
|
|
||||||
|
|
||||||
async def async_turn_on(hass, entity_id=ENTITY_MATCH_ALL):
|
async def async_turn_on(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
|
||||||
"""Turn all or specified switch on."""
|
"""Turn all or specified switch on."""
|
||||||
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
|
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
|
||||||
await hass.services.async_call(DOMAIN, SERVICE_TURN_ON, data, blocking=True)
|
await hass.services.async_call(DOMAIN, SERVICE_TURN_ON, data, blocking=True)
|
||||||
|
|
||||||
|
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def turn_off(hass, entity_id=ENTITY_MATCH_ALL):
|
def turn_off(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
|
||||||
"""Turn all or specified switch off."""
|
"""Turn all or specified switch off."""
|
||||||
hass.add_job(async_turn_off, hass, entity_id)
|
hass.add_job(async_turn_off, hass, entity_id)
|
||||||
|
|
||||||
|
|
||||||
async def async_turn_off(hass, entity_id=ENTITY_MATCH_ALL):
|
async def async_turn_off(
|
||||||
|
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
|
||||||
|
) -> None:
|
||||||
"""Turn all or specified switch off."""
|
"""Turn all or specified switch off."""
|
||||||
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
|
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
|
||||||
await hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, data, blocking=True)
|
await hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, data, blocking=True)
|
||||||
|
@ -1,5 +1,6 @@
|
|||||||
"""Tests for the system health component init."""
|
"""Tests for the system health component init."""
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
from unittest.mock import AsyncMock, Mock, patch
|
from unittest.mock import AsyncMock, Mock, patch
|
||||||
|
|
||||||
from aiohttp.client_exceptions import ClientError
|
from aiohttp.client_exceptions import ClientError
|
||||||
@ -14,7 +15,9 @@ from tests.test_util.aiohttp import AiohttpClientMocker
|
|||||||
from tests.typing import WebSocketGenerator
|
from tests.typing import WebSocketGenerator
|
||||||
|
|
||||||
|
|
||||||
async def gather_system_health_info(hass, hass_ws_client):
|
async def gather_system_health_info(
|
||||||
|
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
|
||||||
|
) -> dict[str, Any]:
|
||||||
"""Gather all info."""
|
"""Gather all info."""
|
||||||
client = await hass_ws_client(hass)
|
client = await hass_ws_client(hass)
|
||||||
|
|
||||||
@ -72,7 +75,7 @@ async def test_info_endpoint_register_callback(
|
|||||||
) -> None:
|
) -> None:
|
||||||
"""Test that the info endpoint allows registering callbacks."""
|
"""Test that the info endpoint allows registering callbacks."""
|
||||||
|
|
||||||
async def mock_info(hass):
|
async def mock_info(hass: HomeAssistant) -> dict[str, Any]:
|
||||||
return {"storage": "YAML"}
|
return {"storage": "YAML"}
|
||||||
|
|
||||||
async_register_info(hass, "lovelace", mock_info)
|
async_register_info(hass, "lovelace", mock_info)
|
||||||
@ -92,7 +95,7 @@ async def test_info_endpoint_register_callback_timeout(
|
|||||||
) -> None:
|
) -> None:
|
||||||
"""Test that the info endpoint timing out."""
|
"""Test that the info endpoint timing out."""
|
||||||
|
|
||||||
async def mock_info(hass):
|
async def mock_info(hass: HomeAssistant) -> dict[str, Any]:
|
||||||
raise TimeoutError
|
raise TimeoutError
|
||||||
|
|
||||||
async_register_info(hass, "lovelace", mock_info)
|
async_register_info(hass, "lovelace", mock_info)
|
||||||
@ -109,7 +112,7 @@ async def test_info_endpoint_register_callback_exc(
|
|||||||
) -> None:
|
) -> None:
|
||||||
"""Test that the info endpoint requires auth."""
|
"""Test that the info endpoint requires auth."""
|
||||||
|
|
||||||
async def mock_info(hass):
|
async def mock_info(hass: HomeAssistant) -> dict[str, Any]:
|
||||||
raise Exception("TEST ERROR") # noqa: TRY002
|
raise Exception("TEST ERROR") # noqa: TRY002
|
||||||
|
|
||||||
async_register_info(hass, "lovelace", mock_info)
|
async_register_info(hass, "lovelace", mock_info)
|
||||||
|
@ -371,7 +371,9 @@ def get_frame(path: str, previous_frame: MagicMock | None) -> MagicMock:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
async def async_log_error_from_test_path(hass, path, watcher):
|
async def async_log_error_from_test_path(
|
||||||
|
hass: HomeAssistant, path: str, watcher: WatchLogErrorHandler
|
||||||
|
) -> None:
|
||||||
"""Log error while mocking the path."""
|
"""Log error while mocking the path."""
|
||||||
call_path = "internal_path.py"
|
call_path = "internal_path.py"
|
||||||
main_frame = get_frame("main_path/main.py", None)
|
main_frame = get_frame("main_path/main.py", None)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user